HITL Park-and-Correlate

What It Is

Human-in-the-loop flows need to pause execution, free the worker, and resume later when an external decision arrives. Dagonizer models that as park-and-correlate: a node routes to the reserved parked output, the result carries a correlation key and cursor, and the host persists a checkpoint until a webhook, operator action, or approval response arrives.

This is not engine suspension. It is a controlled early exit with enough state to resume the same DAG from the parked placement.

How It Works

A parking node writes correlation metadata and routes to the reserved parked output. The dispatcher returns an ExecutionResult with parked details and a checkpointable cursor. The caller persists the checkpoint, waits for the external decision, restores state, writes the response, and calls resume at the parked cursor.

Full lifecycle

// 1. Initial execute — parks at urn:noocodec:dag:support-dispatcher/node/park-for-operator
const supportDispatcherDagIri = 'urn:noocodec:dag:support-dispatcher';
const firstResult = await dispatcher.execute(supportDispatcherDagIri, initialState);
// firstResult.state.lifecycle.variant === 'awaiting-input'
// firstResult.parked.correlationKey  starts with 'escalation:'
// firstResult.parked.cursor          === 'urn:noocodec:dag:support-dispatcher/node/park-for-operator'

// 2. Capture checkpoint (persist: DB, queue, etc.)
const ckpt = await Checkpoint.capture(supportDispatcherDagIri, firstResult);
await db.save(firstResult.parked.correlationKey, ckpt.toJson());

// --- Operator writes the response out of band ---

// 3. On webhook/callback: restore and apply decision
const raw = await db.load(correlationKey);
const recalled = Checkpoint.load(JSON.parse(raw));
const { state, dagName, cursor } = recalled.restoreState(
  CheckpointRestoreAdapter.wrap((snap) => MyState.restore(snap)),
);
state.response = 'I can help with that order.'; // inject the operator response

// 4. Resume — re-enters at the parked placement
const finalResult = await dispatcher.resume(dagName, state, cursor);
// finalResult.state.lifecycle.variant === 'completed'
// finalResult.parked                  === null

Lifecycle state

The 'awaiting-input' lifecycle variant is not terminal. isTerminal() returns false for it; isParked() returns true. The scheduler resets the lifecycle on resume() before calling markRunning(), exactly as it does for crash-recovery resumes from terminal states.

The correlationKey field on the lifecycle state (state.lifecycle.correlationKey) reflects the key stored in the awaiting-input state object. It is null on all other variants.

Diagrams, Examples, and Outputs

The Dispatcher demo contains the support escalation flow that parks for an operator response, persists a checkpoint, and resumes once the simulated operator decision is written back into state.

support-dispatcher HITL DAG

7 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:support-dispatcher",
  "@type": "DAG",
  "name": "dag:support-dispatcher",
  "version": "1",
  "entrypoints": {
    "main": "urn:noocodec:dag:support-dispatcher/node/classify-message"
  },
  "nodes": [
    {
      "@id": "urn:noocodec:dag:support-dispatcher/node/setup",
      "@type": "PhaseNode",
      "name": "dag:support-dispatcher/node/setup",
      "node": "urn:noocodec:node:dispatcher-setup",
      "phase": "pre"
    },
    {
      "@id": "urn:noocodec:dag:support-dispatcher/node/classify-message",
      "@type": "SingleNode",
      "name": "dag:support-dispatcher/node/classify-message",
      "node": "urn:noocodec:node:classify-message",
      "outputs": {
        "routine": "urn:noocodec:dag:support-dispatcher/node/ai-compose",
        "escalate": "urn:noocodec:dag:support-dispatcher/node/park-for-operator",
        "off-topic": "urn:noocodec:dag:support-dispatcher/node/decline"
      }
    },
    {
      "@id": "urn:noocodec:dag:support-dispatcher/node/ai-compose",
      "@type": "SingleNode",
      "name": "dag:support-dispatcher/node/ai-compose",
      "node": "urn:noocodec:node:ai-compose",
      "outputs": {
        "drafted": "urn:noocodec:dag:support-dispatcher/node/send-response"
      }
    },
    {
      "@id": "urn:noocodec:dag:support-dispatcher/node/park-for-operator",
      "@type": "SingleNode",
      "name": "dag:support-dispatcher/node/park-for-operator",
      "node": "urn:noocodec:node:park-for-operator",
      "outputs": {
        "parked": "urn:noocodec:dag:support-dispatcher/node/end",
        "ready": "urn:noocodec:dag:support-dispatcher/node/send-response"
      }
    },
    {
      "@id": "urn:noocodec:dag:support-dispatcher/node/send-response",
      "@type": "SingleNode",
      "name": "dag:support-dispatcher/node/send-response",
      "node": "urn:noocodec:node:send-response",
      "outputs": {
        "sent": "urn:noocodec:dag:support-dispatcher/node/end"
      }
    },
    {
      "@id": "urn:noocodec:dag:support-dispatcher/node/decline",
      "@type": "SingleNode",
      "name": "dag:support-dispatcher/node/decline",
      "node": "urn:noocodec:node:decline",
      "outputs": {
        "declined": "urn:noocodec:dag:support-dispatcher/node/end"
      }
    },
    {
      "@id": "urn:noocodec:dag:support-dispatcher/node/end",
      "@type": "TerminalNode",
      "name": "dag:support-dispatcher/node/end",
      "outcome": "completed"
    }
  ]
}

Mermaid generated from the same DAG

Mermaid source

%%{init: {"flowchart":{"nodeSpacing":92,"rankSpacing":104,"padding":28}}}%%
flowchart TB
  %% dag:support-dispatcher (v1)
  entry_main(["main"])
  entry_main --> urn_noocodec_dag_support-dispatcher/node/classify-message
  urn_noocodec_dag_support-dispatcher/node/setup(["dag:support-dispatcher/node/setup (pre)"])
  urn_noocodec_dag_support-dispatcher/node/classify-message["dag:support-dispatcher/node/classify-message"]
  urn_noocodec_dag_support-dispatcher/node/classify-message -->|routine| urn_noocodec_dag_support-dispatcher/node/ai-compose
  urn_noocodec_dag_support-dispatcher/node/classify-message -->|escalate| urn_noocodec_dag_support-dispatcher/node/park-for-operator
  urn_noocodec_dag_support-dispatcher/node/classify-message -->|off-topic| urn_noocodec_dag_support-dispatcher/node/decline
  urn_noocodec_dag_support-dispatcher/node/ai-compose["dag:support-dispatcher/node/ai-compose"]
  urn_noocodec_dag_support-dispatcher/node/ai-compose -->|drafted| urn_noocodec_dag_support-dispatcher/node/send-response
  urn_noocodec_dag_support-dispatcher/node/park-for-operator["dag:support-dispatcher/node/park-for-operator"]
  urn_noocodec_dag_support-dispatcher/node/park-for-operator -->|parked| urn_noocodec_dag_support-dispatcher/node/end
  urn_noocodec_dag_support-dispatcher/node/park-for-operator -->|ready| urn_noocodec_dag_support-dispatcher/node/send-response
  urn_noocodec_dag_support-dispatcher/node/send-response["dag:support-dispatcher/node/send-response"]
  urn_noocodec_dag_support-dispatcher/node/send-response -->|sent| urn_noocodec_dag_support-dispatcher/node/end
  urn_noocodec_dag_support-dispatcher/node/decline["dag:support-dispatcher/node/decline"]
  urn_noocodec_dag_support-dispatcher/node/decline -->|declined| urn_noocodec_dag_support-dispatcher/node/end
  urn_noocodec_dag_support-dispatcher/node/end((("dag:support-dispatcher/node/end")))

• Example 31: HITL Park-and-Correlate - Dispatcher browser demo showing execute, park, checkpoint, and resume
• The Dispatcher - in-browser runnable support escalation flow
• Reference: Checkpoint - checkpoint capture and restore APIs

What It Lets You Do

Use when

Use HITL park-and-correlate when a DAG must wait for an external actor: an operator, reviewer, customer, approval system, webhook, or compliance gate. The engine should free the worker, return a parked result, and resume later from a correlation key and cursor.

Code Samples

The snippets below show the parking node, the support dispatcher topology, and the checkpoint/resume lifecycle around a parked cursor.

Details for Nerds

Design: park-and-correlate vs. engine-suspend

The primitive does not suspend the engine. Node processes, workers, and containers remain free; the flow simply terminates early with a well-defined exit state. The caller persists the checkpoint and re-invokes when the decision arrives. This matches how serverless and queue-based architectures work: a function call parks, stores its state in a database, and resumes when a new invocation arrives.

The three artifacts the engine surfaces on a parked result:

Field	Type	Meaning
result.parked.correlationKey	string	Opaque key set by the node in state metadata; use it to correlate a webhook/callback with the parked run
result.parked.cursor	string	Placement IRI to pass to dispatcher.resume()
result.parked.dagName	string	DAG IRI/CURIE string to pass to dispatcher.resume()
result.cursor	string | null	Same as parked.cursor; present for Checkpoint.capture()
result.state.lifecycle.variant	'awaiting-input'	Non-terminal lifecycle variant; the run can resume

Authoring a parking node

A node parks by:

1. Writing a correlationKey to state metadata before routing.
2. Routing to the reserved 'parked' output.

The engine intercepts the 'parked' output before normal downstream execution continues. The Dispatcher is the canonical runnable example: ParkForOperatorNode parks the support flow when no operator response exists, then routes 'ready' after the restored state contains the human reply.

/**
 * ParkForOperatorNode: HITL suspension point.
 *
 * On FIRST call (state.response is empty):
 *   - Generates a correlationKey.
 *   - Calls state.park(correlationKey) to transition lifecycle → awaiting-input.
 *   - Routes 'parked' (intercepted by the engine; flow suspends).
 *
 * On RESUME call (state.response is non-empty, set by the human operator):
 *   - Routes 'ready' → send-response continues the flow normally.
 *
 * The 'parked' output is not wired in the DAG — the engine intercepts it
 * before routing and surfaces `result.parked` with the correlationKey and cursor.
 */

import { Batch, MonadicNode, NodeOutput } from '@studnicky/dagonizer';
import type { ItemType, NodeContextType, NodeOutputType, RoutedBatchType, SchemaObjectType } from '@studnicky/dagonizer';

import type { DispatcherState } from '../DispatcherState.ts';

export class ParkForOperatorNode extends MonadicNode<DispatcherState, 'parked' | 'ready'> {
  readonly name = 'park-for-operator';
  readonly '@id' = 'urn:noocodec:node:park-for-operator';
  readonly outputs = ['parked', 'ready'] as const;

  override get outputSchema(): Record<'parked' | 'ready', SchemaObjectType> {
    return {
      'parked': { 'type': 'object' },
      'ready':  { 'type': 'object' },
    };
  }

  override async execute(
    batch: Batch<DispatcherState>,
    _context: NodeContextType,
  ): Promise<RoutedBatchType<'parked' | 'ready', DispatcherState>> {
    const acc = new Map<'parked' | 'ready', ItemType<DispatcherState>[]>();

    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<'parked' | 'ready', Batch<DispatcherState>>();
    for (const [output, items] of acc) {
      routed.set(output, Batch.from(items));
    }
    return routed;
  }

  private routeItem(state: DispatcherState): NodeOutputType<'parked' | 'ready'> {
    if (state.response.length > 0) {
      // Operator has filled in the response; resume the flow.
      return NodeOutput.create('ready');
    }

    // First call: park and await human input.
    const correlationKey = 'escalation:' + Date.now().toString();
    state.park(correlationKey);
    return NodeOutput.create('parked');
  }
}

The DAG placement still declares the escalation branch as normal topology. The important part is that the parking node's output union includes 'parked'; when that output appears, the engine surfaces result.parked with the correlation key and cursor.

const setup           = new PlaceholderNode<DispatcherState, 'ready'>('urn:noocodec:node:dispatcher-setup', ['ready']);
const classifyMessage = new PlaceholderNode<DispatcherState, 'routine' | 'escalate' | 'off-topic'>('urn:noocodec:node:classify-message', ['routine', 'escalate', 'off-topic']);
const aiCompose       = new PlaceholderNode<DispatcherState, 'drafted'>('urn:noocodec:node:ai-compose', ['drafted']);
const parkForOperator = new PlaceholderNode<DispatcherState, 'parked' | 'ready'>('urn:noocodec:node:park-for-operator', ['parked', 'ready']);
const sendResponse    = new PlaceholderNode<DispatcherState, 'sent'>('urn:noocodec:node:send-response', ['sent']);
const decline         = new PlaceholderNode<DispatcherState, 'declined'>('urn:noocodec:node:decline', ['declined']);

const supportDispatcherDagIri = 'urn:noocodec:dag:support-dispatcher' as const;
const placement = (placementIdentifier: string): string =>
  DAGIdentity.placementId(supportDispatcherDagIri, placementIdentifier);

export const supportDispatcherDAG: DAGType = new DAGBuilder(supportDispatcherDagIri, '1')
  // Pre-phase: stamps runId before the entrypoint runs.
  .phase(placement('setup'), 'pre', setup)

  // Entrypoint: classify the inbound message.
  .node(placement('classify-message'), classifyMessage, {
    'routine':   placement('ai-compose'),
    'escalate':  placement('park-for-operator'),
    'off-topic': placement('decline'),
  })

  // Routine branch: AI composes a canned reply -> send -> done.
  .node(placement('ai-compose'), aiCompose, {
    'drafted': placement('send-response'),
  })

  // Escalation branch: HITL suspension point.
  .node(placement('park-for-operator'), parkForOperator, {
    'parked': placement('end'),
    'ready':  placement('send-response'),
  })

  // Shared convergence: both routine and escalated paths flow through send-response.
  .node(placement('send-response'), sendResponse, {
    'sent': placement('end'),
  })

  // Off-topic branch: decline and close.
  .node(placement('decline'), decline, {
    'declined': placement('end'),
  })

  .terminal(placement('end'), { 'outcome': 'completed' })

  .build();

The ParkedType entity

type ParkedType = {
  correlationKey: string;  // opaque, set by the node
  cursor: string;          // placement IRI; pass to dispatcher.resume()
  dagName: string;         // DAG IRI/CURIE; pass to dispatcher.resume()
};

result.parked is null when the flow ran to a terminal without parking. Check result.parked !== null before reading its fields.

NodeStateBase helpers

NodeStateBase exposes two conveniences:

// Transition to awaiting-input and store the correlationKey in metadata.
// Equivalent to what ApproveNode.execute above does by hand.
state.park('approval:req-001');

// True iff lifecycle.variant === 'awaiting-input'
state.parked; // boolean

The engine calls state.park(correlationKey) automatically when it detects a 'parked' output, so nodes do not need to call it manually — just route to 'parked' and write the key to metadata.

Related Concepts

• Example 31: HITL Park-and-Correlate - Dispatcher browser demo showing execute, park, checkpoint, and resume
• The Dispatcher - in-browser runnable support escalation flow
• Reference: Checkpoint - checkpoint capture and restore APIs
• Reference: Lifecycle - lifecycle states including awaiting-input