Example 31: HITL Park-and-Correlate

What It Is

HITL Park-and-Correlate lets an application pause a DAG at a human boundary and resume later with a correlated response. The Dispatcher parks a support flow at park-for-operator, captures checkpoint state, and resumes from the parked cursor after the operator answers.

The DAG owns the pause point. The application owns persistence, correlation, and the UI or transport that collects the external response.

How It Works

The parking node writes a correlation key, routes to parked, and leaves the state lifecycle at awaiting-input. The caller captures a checkpoint from result.parked, persists it under the correlation key, and later restores the state. On resume, the same placement runs again; this time the operator response is already on state, so the node routes ready and the DAG continues to send-response.

This is not a callback hidden in a node. The parked result contains the cursor and correlation key the host needs to persist the pause and re-enter the graph.

Diagrams, Examples, and Outputs

DAG registration and diagram

The graph shows the parked placement and resume path. The Dispatcher is the in-browser owner for this principle: routine messages complete automatically, while escalations park at park-for-operator until the operator supplies a response.

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")))

A human-in-the-loop support flow that parks mid-execution and resumes after an operator response.

• The engine transitions the lifecycle to awaiting-input (non-terminal).
• result.parked carries the correlationKey, cursor, and dagName.
• Checkpoint.capture() works identically on a parked result (cursor is set).
• dispatcher.resume() re-enters at the parked placement with the operator response applied.

Run

npm run docs:dev

Open The Dispatcher, enable HUMAN GATE, send a customer message, then answer it in the Operator pane.

What It Lets You Do

HITL Park-and-Correlate lets a DAG pause for an external actor without turning the workflow into callback code. Use it when a customer, operator, approval queue, or compliance system must answer before the DAG can continue.

The engine returns a parked result with a correlation key and cursor; the application stores that record and resumes when the external response arrives.

Key concepts

Concept	Code
Write correlationKey	state.setMetadata('correlationKey', key)
Route to park	return RoutedBatch.create('parked', Batch.from(parked))
Detect parked result	result.parked !== null
Extract cursor	result.parked.cursor
Capture checkpoint	Checkpoint.capture('urn:noocodec:dag:hitl', result)
Resume with response	dispatcher.resume(dagName, state, cursor)

See HITL Park-and-Correlate guide for the full design rationale and API reference.

Code Samples

The browser run trigger starts the support DAG and captures parked results:

async function ask(): Promise<void> {
  if (isRunning.value || customerQuery.value.trim().length === 0 || activeBackend.value === null) return;

  const queryText = customerQuery.value.trim();
  customerQuery.value = '';
  isRunning.value = true;
  terminalVariant.value = 'pending';
  parked.value = null;
  trace.value = [];
  // Seed the stream with THIS run's customer message so it shows immediately and
  // the Operator pane quotes the message actually being escalated. Each run uses
  // a fresh DispatcherState (empty conversation until send-response/decline
  // appends), so without this the pane would surface a prior run's stale turn.
  conversation.value = [{ 'role': 'customer', 'text': queryText, 'ts': Date.now() }];
  logger.clear();
  logger.note(`run start — message: "${queryText}" · humanMode: ${String(humanMode.value)}`);

  await dagGraph.value?.reset();

  const services = buildServices();
  const setup           = new SetupNode();
  const classifyMessage = new ClassifyMessageNode(services);
  const aiCompose       = new AiComposeNode(services);
  const parkForOperator = new ParkForOperatorNode();
  const sendResponse    = new SendResponseNode();
  const decline         = new DeclineNode();

  const state = new DispatcherState();
  state.message           = queryText;
  state.humanMode         = humanMode.value;
  state.classificationMode = classificationMode.value;
  state.language          = visitorLanguage;

  const dispatcher = new DispatcherBrowserObserver(logger, services);
  dispatcher.registerBundle({
    'nodes': [setup, classifyMessage, aiCompose, parkForOperator, sendResponse, decline],
    'dags':  [supportDispatcherDAG],
  });

  activeAbortController = new AbortController();
  try {
    await dispatcher.execute('support-dispatcher', state, { 'signal': activeAbortController.signal });
  } catch (error) {
    logger.error(
      LogFault.create()
        .component('dispatcher-runner')
        .operation('execute')
        .status('failed')
        .name(error instanceof Error ? error.constructor.name : 'Error')
        .message(`execute error: ${error instanceof Error ? error.message : String(error)}`)
        .context({})
        .build(),
    );
  } finally {
    await dispatcher.destroy();
    activeAbortController = null;
    isRunning.value = false;
  }
}

The browser resume trigger restores the checkpoint, writes the operator response, and resumes from the parked cursor:

async function sendOperatorResponse(): Promise<void> {
  // Use the non-reactive store: parked.value is just the cursor signal.
  const pe = parkedExecution;
  if (pe === null || operatorInput.value.trim().length === 0) return;

  const responseText = operatorInput.value.trim();
  operatorInput.value = '';
  isRunning.value = true;
  trace.value = [...trace.value, {
    'node': 'operator',
    'ts': Date.now(),
    'variant': 'start',
  }];
  logger.note(`operator response captured — resuming from cursor: ${pe.cursor}`);

  const services = buildServices();
  const setup           = new SetupNode();
  const classifyMessage = new ClassifyMessageNode(services);
  const aiCompose       = new AiComposeNode(services);
  const parkForOperator = new ParkForOperatorNode();
  const sendResponse    = new SendResponseNode();
  const decline         = new DeclineNode();
  await dagGraph.value?.reset();

  let restored: { state: DispatcherState; dagName: string; cursor: string } | null = null;
  try {
    // pe.result was stored outside Vue reactivity so it retains class identity.
    const ckpt = await Checkpoint.capture(pe.dagName, pe.result);
    restored = ckpt.restoreState(
      CheckpointRestoreAdapter.wrap((snap) => DispatcherState.restore(snap)),
    );
  } catch (err) {
    logger.error(
      LogFault.create()
        .component('dispatcher-runner')
        .operation('checkpoint-restore')
        .status('failed')
        .name(err instanceof Error ? err.constructor.name : 'Error')
        .message(`checkpoint restore failed: ${err instanceof Error ? err.message : String(err)}`)
        .context({})
        .build(),
    );
    isRunning.value = false;
    return;
  }

  restored.state.response = responseText;

  const dispatcher = new DispatcherBrowserObserver(logger, services);
  dispatcher.registerBundle({
    'nodes': [setup, classifyMessage, aiCompose, parkForOperator, sendResponse, decline],
    'dags':  [supportDispatcherDAG],
  });

  activeAbortController = new AbortController();
  try {
    await dispatcher.resume(restored.dagName, restored.state, restored.cursor, {
      'signal': activeAbortController.signal,
    });
  } catch (error) {
    logger.error(
      LogFault.create()
        .component('dispatcher-runner')
        .operation('resume')
        .status('failed')
        .name(error instanceof Error ? error.constructor.name : 'Error')
        .message(`resume error: ${error instanceof Error ? error.message : String(error)}`)
        .context({})
        .build(),
    );
  } finally {
    await dispatcher.destroy();
    activeAbortController = null;
    isRunning.value = false;
    parkedExecution = null;
    parked.value = null;
    if (leftActiveKey.value === 'operator') leftActiveKey.value = 'customer';
  }
}

The DAG definition contains the parking placement and the ready path:

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();

Details for Nerds

Flow summary

classify-message → park-for-operator ──parked──▶ [awaiting-input]
                                  ◀──resume──── (operator sets response)
                                  ──ready──────▶ send-response → end

• A node routes to the reserved 'parked' output to pause execution.
• result.parked carries dagName, cursor, and correlationKey.
• The checkpoint captures the parked state so a later process can restore and resume.
• The same parking placement runs on resume; state determines whether it parks again or routes ready.

Related Concepts

• Guide: HITL Park-and-Correlate - design rationale, parked result fields, and resume lifecycle
• The Dispatcher - in-browser runnable support escalation demo
• Reference: Checkpoint - checkpoint capture and restore APIs