Example 14: Gather Strategies

What It Is

Gather Strategies decide how producer results become parent state. The Cartographer uses them as application logic: the top-level fold-insights gather folds stream records into bounded aggregates, and the geo-weighted-fusion gather fuses embedded geo resolver outputs into a selected location.

This page is about merge policy, not fan-out. Scatter runs clone work; first-class gather placements decide what the parent sees after each clone, after the whole scatter, or at an explicit graph-visible barrier.

How It Works

A scatter produces clone outcomes. A first-class gather strategy reads completed clone records through the dispatcher accessor and writes the parent fields named by the gather placement. The reducer is a separate decision: gather mutates parent state, then the route map chooses the next placement (all-success, partial, all-error, or empty).

A first-class GatherNode is a normal DAG placement. Producers route into it by placement IRI; the gather node lists the producer placement IRIs it waits for, applies its strategy, then routes on success, error, or empty. In the Cartographer geo resolver, six embedded resolver DAGs project state.candidate into gather records; the geo-weighted-fusion gather node is the explicit barrier that turns those records into state.resolvedGeo and state.geoContext.

Keeping gather separate from outcome reduction matters for application code. A scatter can fold partial data into state and still route partial, all-error, or empty based on the outcome set.

Diagrams, Examples, and Outputs

DAG registration and diagram

The in-browser Cartographer owns the real gather examples. Its top-level process-stream scatter routes to fold-insights, which uses insights-fold to update bounded insight state; its geo-source-resolve sub-DAG makes fusion graph-visible with a first-class geo-weighted-fusion gather node.

Cartographer insights-fold gather DAG

5 placements

DAG JSON-LD registered with the dispatcher

{
  "@context": {
    "@version": 1.1,
    "name": {
      "@id": "https://noocodec.dev/ontology/dag/name"
    },
    "version": {
      "@id": "https://noocodec.dev/ontology/dag/version"
    },
    "entrypoints": {
      "@id": "https://noocodec.dev/ontology/dag/entrypoints",
      "@container": "@index"
    },
    "nodes": {
      "@id": "https://noocodec.dev/ontology/dag/nodes",
      "@container": "@set"
    },
    "outputs": {
      "@id": "https://noocodec.dev/ontology/dag/outputs"
    },
    "node": {
      "@id": "https://noocodec.dev/ontology/dag/node"
    },
    "dag": {
      "@id": "https://noocodec.dev/ontology/dag/dag"
    },
    "body": {
      "@id": "https://noocodec.dev/ontology/dag/body"
    },
    "source": {
      "@id": "https://noocodec.dev/ontology/dag/source"
    },
    "sources": {
      "@id": "https://noocodec.dev/ontology/dag/sources",
      "@container": "@index"
    },
    "itemKey": {
      "@id": "https://noocodec.dev/ontology/dag/itemKey"
    },
    "execution": {
      "@id": "https://noocodec.dev/ontology/dag/execution"
    },
    "concurrency": {
      "@id": "https://noocodec.dev/ontology/dag/concurrency"
    },
    "throttle": {
      "@id": "https://noocodec.dev/ontology/dag/throttle"
    },
    "reservoir": {
      "@id": "https://noocodec.dev/ontology/dag/reservoir"
    },
    "gather": {
      "@id": "https://noocodec.dev/ontology/dag/gather"
    },
    "dagReference": {
      "@id": "https://noocodec.dev/ontology/dag/dagReference",
      "@type": "@id"
    },
    "DagReference": {
      "@id": "https://noocodec.dev/ontology/dag/DagReference"
    },
    "from": {
      "@id": "https://noocodec.dev/ontology/dag/from"
    },
    "path": {
      "@id": "https://noocodec.dev/ontology/dag/path"
    },
    "candidates": {
      "@id": "https://noocodec.dev/ontology/dag/candidates",
      "@container": "@set"
    },
    "candidateDag": {
      "@id": "https://noocodec.dev/ontology/dag/candidateDag",
      "@type": "@id"
    },
    "selectedDag": {
      "@id": "https://noocodec.dev/ontology/dag/selectedDag",
      "@type": "@id"
    },
    "resultField": {
      "@id": "https://noocodec.dev/ontology/dag/resultField"
    },
    "policy": {
      "@id": "https://noocodec.dev/ontology/dag/policy"
    },
    "reducer": {
      "@id": "https://noocodec.dev/ontology/dag/reducer"
    },
    "outcome": {
      "@id": "https://noocodec.dev/ontology/dag/outcome"
    },
    "phase": {
      "@id": "https://noocodec.dev/ontology/dag/phase"
    },
    "stateMapping": {
      "@id": "https://noocodec.dev/ontology/dag/stateMapping"
    },
    "container": {
      "@id": "https://noocodec.dev/ontology/dag/container"
    },
    "DAG": {
      "@id": "https://noocodec.dev/ontology/dag/DAG"
    },
    "Placement": {
      "@id": "https://noocodec.dev/ontology/dag/Placement"
    },
    "SingleNode": {
      "@id": "https://noocodec.dev/ontology/dag/SingleNode"
    },
    "ScatterNode": {
      "@id": "https://noocodec.dev/ontology/dag/ScatterNode"
    },
    "EmbeddedDAGNode": {
      "@id": "https://noocodec.dev/ontology/dag/EmbeddedDAGNode"
    },
    "GatherNode": {
      "@id": "https://noocodec.dev/ontology/dag/GatherNode"
    },
    "TerminalNode": {
      "@id": "https://noocodec.dev/ontology/dag/TerminalNode"
    },
    "PhaseNode": {
      "@id": "https://noocodec.dev/ontology/dag/PhaseNode"
    }
  },
  "@id": "urn:noocodec:dag:cartographer",
  "@type": "DAG",
  "name": "dag:cartographer",
  "version": "1.0",
  "entrypoints": {
    "position-ping": "urn:noocodec:dag:cartographer/node/intake-gather",
    "facility-scan": "urn:noocodec:dag:cartographer/node/intake-gather",
    "sensor-reading": "urn:noocodec:dag:cartographer/node/intake-gather",
    "customs-event": "urn:noocodec:dag:cartographer/node/intake-gather",
    "delivery-confirmation": "urn:noocodec:dag:cartographer/node/intake-gather"
  },
  "nodes": [
    {
      "@id": "urn:noocodec:dag:cartographer/node/intake-gather",
      "@type": "GatherNode",
      "name": "dag:cartographer/node/intake-gather",
      "sources": {
        "urn:noocodec:dag:cartographer/entrypoint/position-ping": {},
        "urn:noocodec:dag:cartographer/entrypoint/facility-scan": {},
        "urn:noocodec:dag:cartographer/entrypoint/sensor-reading": {},
        "urn:noocodec:dag:cartographer/entrypoint/customs-event": {},
        "urn:noocodec:dag:cartographer/entrypoint/delivery-confirmation": {}
      },
      "gather": {
        "strategy": "source-intake"
      },
      "outputs": {
        "success": "urn:noocodec:dag:cartographer/node/process-stream",
        "error": "urn:noocodec:dag:cartographer/node/process-stream",
        "empty": "urn:noocodec:dag:cartographer/node/done"
      }
    },
    {
      "@id": "urn:noocodec:dag:cartographer/node/process-stream",
      "@type": "ScatterNode",
      "name": "dag:cartographer/node/process-stream",
      "source": "sources",
      "body": {
        "dag": "urn:noocodec:dag:stream-event"
      },
      "outputs": {
        "all-success": "urn:noocodec:dag:cartographer/node/fold-insights",
        "partial": "urn:noocodec:dag:cartographer/node/fold-insights",
        "all-error": "urn:noocodec:dag:cartographer/node/fold-insights",
        "empty": "urn:noocodec:dag:cartographer/node/summarize"
      },
      "itemKey": "source-payload",
      "reducer": "aggregate",
      "execution": {
        "mode": "reservoir",
        "concurrency": 16,
        "reservoir": {
          "keyField": "eventType",
          "capacity": 1000
        }
      }
    },
    {
      "@id": "urn:noocodec:dag:cartographer/node/fold-insights",
      "@type": "GatherNode",
      "name": "dag:cartographer/node/fold-insights",
      "sources": {
        "urn:noocodec:dag:cartographer/node/process-stream": {}
      },
      "gather": {
        "strategy": "insights-fold"
      },
      "outputs": {
        "success": "urn:noocodec:dag:cartographer/node/summarize",
        "error": "urn:noocodec:dag:cartographer/node/summarize",
        "empty": "urn:noocodec:dag:cartographer/node/summarize"
      }
    },
    {
      "@id": "urn:noocodec:dag:cartographer/node/summarize",
      "@type": "SingleNode",
      "name": "dag:cartographer/node/summarize",
      "node": "urn:noocodec:node:summarize",
      "outputs": {
        "success": "urn:noocodec:dag:cartographer/node/done"
      }
    },
    {
      "@id": "urn:noocodec:dag:cartographer/node/done",
      "@type": "TerminalNode",
      "name": "dag:cartographer/node/done",
      "outcome": "completed"
    }
  ]
}

Mermaid generated from the same DAG

Mermaid source

%%{init: {"flowchart":{"nodeSpacing":92,"rankSpacing":104,"padding":28}}}%%
flowchart TB
  %% dag:cartographer (v1.0)
  entry_position-ping(["position-ping"])
  entry_position-ping --> urn_noocodec_dag_cartographer/node/intake-gather
  entry_facility-scan(["facility-scan"])
  entry_facility-scan --> urn_noocodec_dag_cartographer/node/intake-gather
  entry_sensor-reading(["sensor-reading"])
  entry_sensor-reading --> urn_noocodec_dag_cartographer/node/intake-gather
  entry_customs-event(["customs-event"])
  entry_customs-event --> urn_noocodec_dag_cartographer/node/intake-gather
  entry_delivery-confirmation(["delivery-confirmation"])
  entry_delivery-confirmation --> urn_noocodec_dag_cartographer/node/intake-gather
  urn_noocodec_dag_cartographer/node/intake-gather{"dag_cartographer/node/intake-gather"}
  urn_noocodec_dag_cartographer/node/intake-gather -->|success| urn_noocodec_dag_cartographer/node/process-stream
  urn_noocodec_dag_cartographer/node/intake-gather -->|error| urn_noocodec_dag_cartographer/node/process-stream
  urn_noocodec_dag_cartographer/node/intake-gather -->|empty| urn_noocodec_dag_cartographer/node/done
  urn_noocodec_dag_cartographer/node/process-stream[/"dag:cartographer/node/process-stream ▣ eventType ×1000"/]
  urn_noocodec_dag_cartographer/node/process-stream -->|all-success| urn_noocodec_dag_cartographer/node/fold-insights
  urn_noocodec_dag_cartographer/node/process-stream -->|partial| urn_noocodec_dag_cartographer/node/fold-insights
  urn_noocodec_dag_cartographer/node/process-stream -->|all-error| urn_noocodec_dag_cartographer/node/fold-insights
  urn_noocodec_dag_cartographer/node/process-stream -->|empty| urn_noocodec_dag_cartographer/node/summarize
  urn_noocodec_dag_cartographer/node/fold-insights{"dag_cartographer/node/fold-insights"}
  urn_noocodec_dag_cartographer/node/fold-insights -->|success| urn_noocodec_dag_cartographer/node/summarize
  urn_noocodec_dag_cartographer/node/fold-insights -->|error| urn_noocodec_dag_cartographer/node/summarize
  urn_noocodec_dag_cartographer/node/fold-insights -->|empty| urn_noocodec_dag_cartographer/node/summarize
  urn_noocodec_dag_cartographer/node/summarize["dag:cartographer/node/summarize"]
  urn_noocodec_dag_cartographer/node/summarize -->|success| urn_noocodec_dag_cartographer/node/done
  urn_noocodec_dag_cartographer/node/done((("dag:cartographer/node/done")))
  classDef reservoir fill:#1e3a5f,stroke:#3b82f6,color:#bfdbfe
  class urn_noocodec_dag_cartographer/node/process-stream reservoir

Both custom strategies are registered and exercised by the Cartographer runnable. insights-fold keeps the browser demo bounded for large streams; geo-weighted-fusion merges parallel geo resolver candidates into one selected resolution.

Run

npm run docs:dev

What It Lets You Do

Gather strategies let producer placements decide how results become parent state. They fit cases where worker bodies stay focused on one item while the parent DAG owns the merge policy: fold every result into a bounded aggregate, pick the best candidate, discard fire-and-forget output, or run a domain-specific merge at an explicit fan-in barrier.

Code Samples

Read the snippets with the diagrams nearby so the TypeScript behavior, JSON-LD graph shape, and runtime output line up as one contract.

/**
 * cartographerDAG: multi-entry source intake gather over raw source payloads.
 *
 * Five canonical entrypoint IRIs target intake-gather directly. The
 * source-intake gather opens those per-type streams and merges them into
 * state.sources.
 * The processing scatter reads that merged stream at concurrency 16, runs
 * stream-event per item, and folds completed
 * clone state through insights-fold. Memory is O(1) regardless of event count.
 *
 * Topology:
 *   5 data-type entrypoints → gather('intake-gather', source-intake)
 *     → scatter('process-stream', 'sources', { dag: 'stream-event' }, concurrency: 16)
 *     → gather('fold-insights', strategy: insights-fold)
 *     → summarize → done
 */
export const cartographerDAG: DAGType = new DAGBuilder(CARTOGRAPHER_DAG_IRI, '1.0')

  .gather(
    CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'intake-gather'),
    CARTOGRAPHER_IRIS.intakeSources(CARTOGRAPHER_DAG_IRI),
    { 'strategy': 'source-intake' },
    {
      'success': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'process-stream'),
      'error':   CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'process-stream'),
      'empty':   CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'done'),
    },
  )

  .scatter(
    CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'process-stream'),
    'sources',
    { 'dag': STREAM_EVENT_DAG_IRI },
    {
      'all-success': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'fold-insights'),
      'partial':     CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'fold-insights'),
      'all-error':   CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'fold-insights'),
      'empty':       CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'summarize'),
    },
    {
      'itemKey':     'source-payload',
      'execution': { 'mode': 'reservoir', 'concurrency': 16, 'reservoir': { 'keyField': 'eventType', 'capacity': 1000 } },
    },
  )
  .gather(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'fold-insights'), {
    [CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'process-stream')]: {},
  }, { 'strategy': 'insights-fold' }, {
    'success': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'summarize'),
    'error':   CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'summarize'),
    'empty':   CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'summarize'),
  })

  // Pass-through in the streaming path (insights-fold already populated
  // state.insights, state.journeys, and state.sampleRecords). Falls back
  // to the records-based fold for non-streaming callers.
  .node(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'summarize'), summarizeInsights, {
    'success': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'done'),
  })

  .terminal(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'done'), { outcome: 'completed' })

  .entrypoints({
    'position-ping':          CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'intake-gather'),
    'facility-scan':          CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'intake-gather'),
    'sensor-reading':         CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'intake-gather'),
    'customs-event':          CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'intake-gather'),
    'delivery-confirmation':  CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_DAG_IRI, 'intake-gather'),
  })

  .build();

/**
 * GeoSourceResolveDAG: validity-gated weighted multi-entry geo-resolution sub-DAG.
 *
 * The parent DAG has one entrypoint per geo modality. Each entrypoint embeds a
 * modality-specific resolver DAG that prepares its descriptor, resolves the
 * candidate when present, and projects `state.candidate` into a gather record.
 * The first-class `geo-weighted-fusion` gather node is the graph-visible barrier
 * that folds candidates by weight into `state.resolvedGeo`, `state.geoContext`,
 * and `state.routing.{geoConfidence,geoModalities}`. When no signal resolves,
 * the gather writes the baseline values directly.
 *
 * Topology:
 *   entrypoints:
 *     coords  ─► resolve-coords-source  ─success/error─┐
 *     address ─► resolve-address-source ─success/error─┤
 *     ip      ─► resolve-ip-source      ─success/error─┤
 *     code    ─► resolve-code-source    ─success/error─┤─► geo-weighted-fusion
 *     phone   ─► resolve-phone-source   ─success/error─┤
 *     locale  ─► resolve-locale-source  ─success/error─┘
 *   geo-weighted-fusion (gather) ─success/error/empty──► resolved
 *   resolved: terminal (completed)
 *
 * DI: `ipGeolocator` and `addressGeocoder` are injected per-call so each
 * dispatcher (main thread, worker thread) owns its own transport instances.
 */

// #region geo-source-resolve-dag
import { resolveCoords } from '../nodes/geo/resolveCoords.ts';
import { resolveLocale } from '../nodes/geo/resolveLocale.ts';
import { resolveCode } from '../nodes/geo/resolveCode.ts';
import { resolvePhone } from '../nodes/geo/resolvePhone.ts';
import { ResolveIpNode } from '../nodes/geo/resolveIp.ts';
import { ResolveAddressNode } from '../nodes/geo/resolveAddress.ts';
import {
  prepareGeoAddress,
  prepareGeoCode,
  prepareGeoCoords,
  prepareGeoIp,
  prepareGeoLocale,
  prepareGeoPhone,
} from '../nodes/geo/prepareGeoSignal.ts';
import { CARTOGRAPHER_IRIS } from '../cartographerIds.ts';
import type { IpGeolocator } from '../contracts/IpGeolocator.ts';
import type { AddressGeocoder } from '../contracts/AddressGeocoder.ts';
import type { CartographerState } from '../CartographerState.ts';

// Side-effect import: registers 'geo-weighted-fusion' at module load so the
// scatter placement can resolve the strategy name at dispatcher construction.
import '../core/GeoWeightedFusionGather.ts';

import type { DispatcherBundleType } from '@studnicky/dagonizer';
import { DAGBuilder } from '@studnicky/dagonizer';

export class GeoSourceResolveDAG {
  private constructor() { /* static-only */ }

  static build(
    ipGeolocator: IpGeolocator,
    addressGeocoder: AddressGeocoder,
  ): DispatcherBundleType<CartographerState> {
    const resolveIp = new ResolveIpNode(ipGeolocator);
    const resolveAddress = new ResolveAddressNode(addressGeocoder);

    const resolveCoordsDag = new DAGBuilder(CARTOGRAPHER_IRIS.dag.geoResolveCoords, '1.0')
      .node(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveCoords, 'prepare-geo-coords'), prepareGeoCoords, {
        'present': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveCoords, 'resolve-coords'),
        'missing': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveCoords, 'done'),
      })
      .node(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveCoords, 'resolve-coords'), resolveCoords, { 'resolved': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveCoords, 'done') })
      .terminal(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveCoords, 'done'), { 'outcome': 'completed' })
      .build();

    const resolveAddressDag = new DAGBuilder(CARTOGRAPHER_IRIS.dag.geoResolveAddress, '1.0')
      .node(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveAddress, 'prepare-geo-address'), prepareGeoAddress, {
        'present': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveAddress, 'resolve-address'),
        'missing': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveAddress, 'done'),
      })
      .node(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveAddress, 'resolve-address'), resolveAddress, { 'resolved': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveAddress, 'done') })
      .terminal(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveAddress, 'done'), { 'outcome': 'completed' })
      .build();

    const resolveIpDag = new DAGBuilder(CARTOGRAPHER_IRIS.dag.geoResolveIp, '1.0')
      .node(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveIp, 'prepare-geo-ip'), prepareGeoIp, {
        'present': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveIp, 'resolve-ip'),
        'missing': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveIp, 'done'),
      })
      .node(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveIp, 'resolve-ip'), resolveIp, { 'resolved': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveIp, 'done') })
      .terminal(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveIp, 'done'), { 'outcome': 'completed' })
      .build();

    const resolveCodeDag = new DAGBuilder(CARTOGRAPHER_IRIS.dag.geoResolveCode, '1.0')
      .node(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveCode, 'prepare-geo-code'), prepareGeoCode, {
        'present': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveCode, 'resolve-code'),
        'missing': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveCode, 'done'),
      })
      .node(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveCode, 'resolve-code'), resolveCode, { 'resolved': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveCode, 'done') })
      .terminal(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveCode, 'done'), { 'outcome': 'completed' })
      .build();

    const resolvePhoneDag = new DAGBuilder(CARTOGRAPHER_IRIS.dag.geoResolvePhone, '1.0')
      .node(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolvePhone, 'prepare-geo-phone'), prepareGeoPhone, {
        'present': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolvePhone, 'resolve-phone'),
        'missing': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolvePhone, 'done'),
      })
      .node(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolvePhone, 'resolve-phone'), resolvePhone, { 'resolved': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolvePhone, 'done') })
      .terminal(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolvePhone, 'done'), { 'outcome': 'completed' })
      .build();

    const resolveLocaleDag = new DAGBuilder(CARTOGRAPHER_IRIS.dag.geoResolveLocale, '1.0')
      .node(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveLocale, 'prepare-geo-locale'), prepareGeoLocale, {
        'present': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveLocale, 'resolve-locale'),
        'missing': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveLocale, 'done'),
      })
      .node(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveLocale, 'resolve-locale'), resolveLocale, { 'resolved': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveLocale, 'done') })
      .terminal(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoResolveLocale, 'done'), { 'outcome': 'completed' })
      .build();

    const dag = new DAGBuilder(CARTOGRAPHER_IRIS.dag.geoSourceResolve, '1.0')

      .embed<CartographerState, CartographerState>(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolve-coords-source'), CARTOGRAPHER_IRIS.dag.geoResolveCoords, {
        'success': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'geo-weighted-fusion'),
        'error':   CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'geo-weighted-fusion'),
      }, {
        'gatherResult': { 'resultField': 'candidate' },
      })

      .embed<CartographerState, CartographerState>(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolve-address-source'), CARTOGRAPHER_IRIS.dag.geoResolveAddress, {
        'success': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'geo-weighted-fusion'),
        'error':   CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'geo-weighted-fusion'),
      }, {
        'gatherResult': { 'resultField': 'candidate' },
      })

      .embed<CartographerState, CartographerState>(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolve-ip-source'), CARTOGRAPHER_IRIS.dag.geoResolveIp, {
        'success': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'geo-weighted-fusion'),
        'error':   CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'geo-weighted-fusion'),
      }, {
        'gatherResult': { 'resultField': 'candidate' },
      })

      .embed<CartographerState, CartographerState>(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolve-code-source'), CARTOGRAPHER_IRIS.dag.geoResolveCode, {
        'success': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'geo-weighted-fusion'),
        'error':   CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'geo-weighted-fusion'),
      }, {
        'gatherResult': { 'resultField': 'candidate' },
      })

      .embed<CartographerState, CartographerState>(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolve-phone-source'), CARTOGRAPHER_IRIS.dag.geoResolvePhone, {
        'success': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'geo-weighted-fusion'),
        'error':   CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'geo-weighted-fusion'),
      }, {
        'gatherResult': { 'resultField': 'candidate' },
      })

      .embed<CartographerState, CartographerState>(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolve-locale-source'), CARTOGRAPHER_IRIS.dag.geoResolveLocale, {
        'success': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'geo-weighted-fusion'),
        'error':   CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'geo-weighted-fusion'),
      }, {
        'gatherResult': { 'resultField': 'candidate' },
      })

      .entrypoints({
        'coords':  CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolve-coords-source'),
        'address': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolve-address-source'),
        'ip':      CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolve-ip-source'),
        'code':    CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolve-code-source'),
        'phone':   CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolve-phone-source'),
        'locale':  CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolve-locale-source'),
      })

      // First-class gather barrier over all embedded resolver producers.
      .gather(
        CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'geo-weighted-fusion'),
        {
          [CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolve-coords-source')]: {},
          [CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolve-address-source')]: {},
          [CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolve-ip-source')]: {},
          [CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolve-code-source')]: {},
          [CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolve-phone-source')]: {},
          [CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolve-locale-source')]: {},
        },
        { 'strategy': 'geo-weighted-fusion' },
        {
          'success': CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolved'),
          'error':   CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolved'),
          'empty':   CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolved'),
        },
      )

      .terminal(CARTOGRAPHER_IRIS.placementIri(CARTOGRAPHER_IRIS.dag.geoSourceResolve, 'resolved'), { 'outcome': 'completed' })

      .build();

    return {
      'nodes': [
        prepareGeoCoords,
        prepareGeoAddress,
        prepareGeoIp,
        prepareGeoCode,
        prepareGeoPhone,
        prepareGeoLocale,
        resolveCoords,
        resolveAddress,
        resolveIp,
        resolveCode,
        resolvePhone,
        resolveLocale,
      ],
      'dags': [
        resolveCoordsDag,
        resolveAddressDag,
        resolveIpDag,
        resolveCodeDag,
        resolvePhoneDag,
        resolveLocaleDag,
        dag,
      ],
    };
  }
}
// #endregion geo-source-resolve-dag

/**
 * GeoWeightedFusionGather: stateless weighted-fusion gather strategy for the
 * geo-source-resolve first-class gather barrier.
 *
 * Each upstream resolver DAG resolves one geo signal and writes the result as
 * `state.candidate` (a GeoResolution). The embedding placement projects that
 * value into `GatherRecord.result`. This strategy:
 *
 *   1. reduce:   Accumulates weight>0 candidates from gather records into
 *                `state.geoCandidates`.
 *
 *   2. finalize: Applies the weighted-fusion rules to produce
 *                `state.resolvedGeo`, `state.geoContext`, and merges
 *                `state.routing.geoConfidence` / `state.routing.geoModalities`.
 *
 * Fusion rules (in priority order):
 *   - WINNER   = highest-weight candidate (ties: first encountered).
 *   - BACK-FILL: region, locality, countryName, locale, timezone —
 *               if winner's field is empty, fill from the next-highest-weight
 *               candidate that has a non-empty value.
 *   - PROVENANCE = contributing candidates' source strings, ordered
 *                  highest-weight first, de-duplicated.
 *   - CONFIDENCE = winner's weight, with composite override: if BOTH a `code`
 *                  and a `locale` candidate resolved and agree on the same ISO-2
 *                  country, confidence = max(winnerWeight, 0.45).
 *   - MODALITIES: 'gps' if any contributing source is 'coords'; 'ip' if any is 'ip'.
 *   - EMPTY case (zero weight>0 candidates): baseline resolvedGeo, confidence 0.
 *
 * Registered as 'geo-weighted-fusion' at module load.
 */

import type { GatherExecutionType } from '@studnicky/dagonizer/contracts';
import { GatherStrategies, GatherStrategy } from '@studnicky/dagonizer/core';
import type { GatherConfigType, NodeStateInterface } from '@studnicky/dagonizer/types';
import type { StateAccessorInterface } from '@studnicky/dagonizer/contracts';

import type { GeoResolution } from '../entities/GeoResolution.ts';
import { SignalWeight } from '../entities/SignalWeight.ts';
import { Continents, Jurisdictions, TimeZoneResolver } from '../services.ts';
import { GeoBaseline } from './GeoBaseline.ts';

// ── Private structural type-guard ─────────────────────────────────────────────

class CandidateRecord {
  private constructor() { /* static-only */ }

  static is(v: unknown): v is GeoResolution {
    if (v === null || typeof v !== 'object') return false;
    return (
      typeof Reflect.get(v, 'source')      === 'string' &&
      typeof Reflect.get(v, 'weight')      === 'number' &&
      typeof Reflect.get(v, 'country')     === 'string' &&
      typeof Reflect.get(v, 'region')      === 'string' &&
      typeof Reflect.get(v, 'locality')    === 'string' &&
      typeof Reflect.get(v, 'countryName') === 'string' &&
      typeof Reflect.get(v, 'locale')      === 'string' &&
      typeof Reflect.get(v, 'timezone')    === 'string' &&
      typeof Reflect.get(v, 'lat')         === 'number' &&
      typeof Reflect.get(v, 'lng')         === 'number' &&
      typeof Reflect.get(v, 'status')      === 'string' &&
      typeof Reflect.get(v, 'secondaryLookupUsed') === 'boolean'
    );
  }
}

class GeoResolutionArray {
  private constructor() { /* static-only */ }

  static is(v: unknown): v is GeoResolution[] {
    return Array.isArray(v);
  }
}

// ── GeoWeightedFusionGather ───────────────────────────────────────────────────

export class GeoWeightedFusionGather extends GatherStrategy {
  readonly name = 'geo-weighted-fusion';
  readonly '@id' = 'urn:noocodec:node:geo-weighted-fusion';

  // ── initial: reset geoCandidates accumulator in parent state ─────────────

  override initial(
    _config: GatherConfigType,
    state: NodeStateInterface,
    accessor: StateAccessorInterface,
  ): void {
    accessor.set(state, 'geoCandidates', []);
  }

  // ── reduce: collect weight>0 candidates from each clone ──────────────────

  override reduce(
    _config: GatherConfigType,
    batch: Parameters<GatherStrategy['reduce']>[1],
    state: NodeStateInterface,
    accessor: StateAccessorInterface,
  ): void {
    for (const item of batch) {
      const record = item.state;
      const rawCandidate = CandidateRecord.is(record.result)
        ? record.result
        : accessor.get(record.cloneState, 'candidate');
      if (!CandidateRecord.is(rawCandidate)) continue;
      if (rawCandidate.weight <= 0) continue;

      // Read-array, push, set-back idiom (matches InsightsFoldGather pattern).
      const rawCandidates = accessor.get(state, 'geoCandidates');
      const candidates: GeoResolution[] = GeoResolutionArray.is(rawCandidates)
        ? rawCandidates.filter(CandidateRecord.is)
        : [];
      candidates.push(rawCandidate);
      accessor.set(state, 'geoCandidates', candidates);
    }
  }

  // ── finalize: apply fusion rules, write resolvedGeo + geoContext ──────────

  override async finalize(
    _config: GatherConfigType,
    execution: GatherExecutionType<NodeStateInterface>,
  ): Promise<void> {
    GeoWeightedFusionGather.mergeCapturedErrors(execution);

    const rawCandidates = execution.accessor.get(execution.state, 'geoCandidates');
    const candidates: GeoResolution[] = GeoResolutionArray.is(rawCandidates)
      ? rawCandidates.filter(CandidateRecord.is)
      : [];

    // Sort descending by weight (highest first; stable sort preserves tie order).
    const sorted = [...candidates].sort((a, b) => b.weight - a.weight);
    const winner = sorted[0];

    if (winner === undefined) {
      // EMPTY case: delegate to the single source of truth for baseline values.
      execution.accessor.set(execution.state, 'resolvedGeo', GeoBaseline.resolvedGeo());
      execution.accessor.set(execution.state, 'geoContext',  GeoBaseline.geoContext());
      return;
    }

    // BACK-FILL: for each back-fillable field, find the next-highest-weight candidate
    // with a non-empty value for that field.
    const region      = GeoWeightedFusionGather.backFill(sorted, 'region',      winner.region);
    const locality    = GeoWeightedFusionGather.backFill(sorted, 'locality',    winner.locality);
    const countryName = GeoWeightedFusionGather.backFill(sorted, 'countryName', winner.countryName);
    const locale      = GeoWeightedFusionGather.backFill(sorted, 'locale',      winner.locale);
    const timezone    = GeoWeightedFusionGather.backFill(sorted, 'timezone',    winner.timezone);

    // PROVENANCE: sources ordered highest-weight first, de-duplicated
    const provenance = GeoWeightedFusionGather.buildProvenance(sorted);

    // MODALITIES
    const modalities = GeoWeightedFusionGather.buildModalities(provenance);

    // CONFIDENCE
    const confidence = GeoWeightedFusionGather.computeConfidence(sorted, winner.weight);

    // JURISDICTION, CONTINENT, TIMEZONE, STATUS
    const country = winner.country;
    const status  = winner.status;

    const jurisdiction = status === 'water'
      ? 'international-waters'
      : country.length > 0
      ? Jurisdictions.forIso2(country).jurisdiction
      : 'baseline';

    const continent = country.length > 0 ? Continents.forIso2(country) : 'Unmapped';

    const lat = winner.lat;
    const lng = winner.lng;
    const effectiveTimezone = timezone.length > 0
      ? timezone
      : (lat !== 0 || lng !== 0 ? TimeZoneResolver.zoneFor(lat, lng) : 'UTC');

    // GRID ZONE: coords source → 'GEOHASH'; else 'API'
    const gridZone = winner.source === 'coords' ? 'GEOHASH' : 'API';

    const effectiveHub = locality.length > 0
      ? locality
      : (country.length > 0 ? country : 'Unknown');

    const geoStatus: 'land' | 'water' | 'coastal' | 'unmapped' =
      status === 'water'
        ? 'water'
        : status === 'coastal'
        ? 'coastal'
        : country.length > 0
        ? 'land'
        : 'unmapped';

    execution.accessor.set(execution.state, 'resolvedGeo', {
      'country':      country,
      'countryName':  countryName.length > 0 ? countryName : country,
      'continent':    continent,
      'region':       region,
      'locality':     locality,
      'locale':       locale,
      'lat':          lat,
      'lng':          lng,
      'status':       status,
      'jurisdiction': jurisdiction,
      'confidence':   confidence,
      'modalities':   modalities,
      'provenance':   provenance,
    });

    execution.accessor.set(execution.state, 'geoContext', {
      'gridZone':    gridZone,
      'country':     country.length > 0 ? country : 'INTL',
      'continent':   continent,
      'countries':   country.length > 0 ? [country] : [],
      'region':      region,
      'hub':         effectiveHub,
      'status':      geoStatus,
      'waterBodies': status === 'water' ? [locality] : [],
      'timezone':    effectiveTimezone,
      'jurisdiction': jurisdiction,
    });

    // Merge geoConfidence and geoModalities into routing (read-spread-write pattern)
    const rawRouting = execution.accessor.get(execution.state, 'routing');
    if (rawRouting !== null && typeof rawRouting === 'object' && !Array.isArray(rawRouting)) {
      execution.accessor.set(execution.state, 'routing', {
        ...Object.fromEntries(Object.entries(rawRouting)),
        'geoConfidence': confidence,
        'geoModalities': provenance,
      });
    }
  }

  // ── Private helpers ───────────────────────────────────────────────────────

  /**
   * Back-fill: if the winner's field is non-empty, return it.
   * Otherwise find the next-highest-weight candidate with a non-empty value.
   */
  private static backFill(
    sorted: GeoResolution[],
    field: 'region' | 'locality' | 'countryName' | 'locale' | 'timezone',
    winnerValue: string,
  ): string {
    if (winnerValue.length > 0) return winnerValue;
    for (const candidate of sorted) {
      const val = candidate[field];
      if (val.length > 0) return val;
    }
    return '';
  }

  /**
   * Build de-duplicated provenance list ordered highest-weight first.
   * Source strings are added in sorted order (already sorted desc by weight).
   */
  private static buildProvenance(sorted: GeoResolution[]): string[] {
    const seen = new Set<string>();
    const result: string[] = [];
    for (const candidate of sorted) {
      if (!seen.has(candidate.source)) {
        seen.add(candidate.source);
        result.push(candidate.source);
      }
    }
    return result;
  }

  /**
   * Build modalities enum: 'gps' if any contributing source is 'coords';
   * 'ip' if any contributing source is 'ip'.
   */
  private static buildModalities(provenance: string[]): Array<'gps' | 'ip'> {
    const result: Array<'gps' | 'ip'> = [];
    if (provenance.includes('coords')) result.push('gps');
    if (provenance.includes('ip'))     result.push('ip');
    return result;
  }

  /**
   * Compute confidence. Applies the composite code+locale override when
   * both are present and agree on ISO-2 country.
   */
  private static computeConfidence(sorted: GeoResolution[], winnerWeight: number): number {
    let codeCandidate:   GeoResolution | undefined;
    let localeCandidate: GeoResolution | undefined;

    for (const candidate of sorted) {
      if (candidate.source === 'code'   && codeCandidate   === undefined) codeCandidate   = candidate;
      if (candidate.source === 'locale' && localeCandidate === undefined) localeCandidate = candidate;
    }

    if (
      codeCandidate   !== undefined &&
      localeCandidate !== undefined &&
      codeCandidate.country.length > 0 &&
      codeCandidate.country === localeCandidate.country
    ) {
      return Math.max(winnerWeight, SignalWeight.COMPOSITE_CODE_LOCALE);
    }

    return winnerWeight;
  }

  private static mergeCapturedErrors(execution: GatherExecutionType<NodeStateInterface>): void {
    const current = execution.accessor.get(execution.state, 'capturedErrors');
    const merged = Array.isArray(current) ? [...current] : [];
    const seen = new Set<string>(merged.map((entry) => JSON.stringify(entry)));

    for (const record of execution.records) {
      const rawErrors = execution.accessor.get(record.cloneState, 'capturedErrors');
      if (!Array.isArray(rawErrors)) continue;
      for (const error of rawErrors) {
        const key = JSON.stringify(error);
        if (seen.has(key)) continue;
        seen.add(key);
        merged.push(error);
      }
    }

    execution.accessor.set(execution.state, 'capturedErrors', merged);
  }
}

// ── Module-load registration ──────────────────────────────────────────────────

GatherStrategies.register(new GeoWeightedFusionGather());
// GatherStrategies.resolve('geo-weighted-fusion') now works in any gather placement.

Details for Nerds

• Gather is placement policy. First-class gather nodes declare a strategy key; worker/body nodes do not own parent merge behavior.
• Streaming fold. insights-fold updates parent aggregates as stream clones complete.
• Explicit fusion barrier. geo-source-resolve starts six embedded resolver DAGs from labeled entrypoints, then routes their projected candidates into a GatherNode named geo-weighted-fusion.
• Weighted selection. geo-weighted-fusion folds multiple resolver candidates into one canonical geo result.
• Same engine surface. Both strategies are JSON-LD strategy keys and registry entries in the runnable demo.

Related Concepts

• Example 04: Scatter Scout - scatter mechanics: source, body DAG, gather placement, reduce
• Example 04B: Scatter Collect - first-class gather: generate-and-select pattern
• Example 15: Incremental gather - incremental vs batch gather timing
• Reference: Core, GatherStrategies