Dagonizer
Dagonizer

Example 07: Retry Flow ​

What It Is ​

Retry Flow is The Archivist showing its work when a model call, parser, or quality gate needs another attempt. A node routes a retry output that loops back through the DAG, bounded by state, or routes salvage to deterministic recovery.

The important part is visibility. Retry is not a hidden while loop in the dispatcher and not an exception swallowed inside a node; it is a named edge in the graph that appears in JSON-LD, Mermaid, traces, and checkpoints.

How It Works ​

When a node fails its own deadline or an LLM call throws, it makes a flow decision rather than swallowing the failure: it routes a retry output that loops back in the DAG, bounded by a counter on the state, or routes salvage to a deterministic recovery node once the budget is spent. The node never fabricates a result to take the happy path, and there is no hidden retry loop inside the dispatcher.

The same shape appears in two places:

  1. Self-loop retry. Each agent node (extract-query, decide-tools, rank-candidates, the composers) routes retry to itself and salvage to a recovery node.
  2. Two-node retry loop. validate-response routes retry back to compose-response when a draft fails the quality gate, bounded by the same compose budget.

Both are loop edges in the topology. The dispatcher always sees a named output; nothing throws.

Diagrams, Examples, and Outputs ​

DAG registration and diagram ​

In The Archivist retry is a flow shape. The compose-retry-loop sub-DAG renders the retry edge, salvage path, and final success path as normal JSON-LD routes.

compose-retry-loop

4 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:compose-retry-loop",
  "@type": "DAG",
  "name": "compose-retry-loop",
  "version": "1.1",
  "entrypoints": {
    "main": "urn:noocodec:dag:compose-retry-loop/node/compose-response"
  },
  "nodes": [
    {
      "@id": "urn:noocodec:dag:compose-retry-loop/node/compose-response",
      "@type": "SingleNode",
      "name": "compose-response",
      "node": "urn:noocodec:node:compose-response",
      "outputs": {
        "drafted": "urn:noocodec:dag:compose-retry-loop/node/validate-response",
        "retry": "urn:noocodec:dag:compose-retry-loop/node/compose-response",
        "salvage": "urn:noocodec:dag:compose-retry-loop/node/compose-salvage"
      }
    },
    {
      "@id": "urn:noocodec:dag:compose-retry-loop/node/compose-salvage",
      "@type": "SingleNode",
      "name": "compose-salvage",
      "node": "urn:noocodec:node:compose-salvage",
      "outputs": {
        "done": "urn:noocodec:dag:compose-retry-loop/node/composed"
      }
    },
    {
      "@id": "urn:noocodec:dag:compose-retry-loop/node/validate-response",
      "@type": "SingleNode",
      "name": "validate-response",
      "node": "urn:noocodec:node:validate-response",
      "outputs": {
        "approved": "urn:noocodec:dag:compose-retry-loop/node/composed",
        "retry": "urn:noocodec:dag:compose-retry-loop/node/compose-response",
        "exhausted": "urn:noocodec:dag:compose-retry-loop/node/composed"
      }
    },
    {
      "@id": "urn:noocodec:dag:compose-retry-loop/node/composed",
      "@type": "TerminalNode",
      "name": "composed",
      "outcome": "completed"
    }
  ]
}
Mermaid generated from the same DAG
Mermaid source
%%{init: {"flowchart":{"nodeSpacing":92,"rankSpacing":104,"padding":28}}}%%
flowchart TB
  %% compose-retry-loop (v1.1)
  entry_main(["main"])
  entry_main --> urn_noocodec_dag_compose-retry-loop/node/compose-response
  urn_noocodec_dag_compose-retry-loop/node/compose-response["compose-response"]
  urn_noocodec_dag_compose-retry-loop/node/compose-response -->|drafted| urn_noocodec_dag_compose-retry-loop/node/validate-response
  urn_noocodec_dag_compose-retry-loop/node/compose-response -->|retry| urn_noocodec_dag_compose-retry-loop/node/compose-response
  urn_noocodec_dag_compose-retry-loop/node/compose-response -->|salvage| urn_noocodec_dag_compose-retry-loop/node/compose-salvage
  urn_noocodec_dag_compose-retry-loop/node/compose-salvage["compose-salvage"]
  urn_noocodec_dag_compose-retry-loop/node/compose-salvage -->|done| urn_noocodec_dag_compose-retry-loop/node/composed
  urn_noocodec_dag_compose-retry-loop/node/validate-response["validate-response"]
  urn_noocodec_dag_compose-retry-loop/node/validate-response -->|approved| urn_noocodec_dag_compose-retry-loop/node/composed
  urn_noocodec_dag_compose-retry-loop/node/validate-response -->|retry| urn_noocodec_dag_compose-retry-loop/node/compose-response
  urn_noocodec_dag_compose-retry-loop/node/validate-response -->|exhausted| urn_noocodec_dag_compose-retry-loop/node/composed
  urn_noocodec_dag_compose-retry-loop/node/composed((("composed")))

Run ​

bash
npx tsx examples/the-archivist/runArchivist.ts

What It Lets You Do ​

Retry flow lets applications represent recovery as visible graph topology instead of hidden exception handling. Use it when a model call, parser, or validation step can make another attempt, but the DAG must still show the loop, budget, and salvage path explicitly.

This is useful for model-backed products because retry policy becomes debuggable. An application author can read the graph and see the recovery story, then trace exactly which placement retried, how many attempts remain, and where salvage rejoins the flow.

Code Samples ​

The retrying node ​

extract-query arms its own deadline and asks the conceptual-root retry budget which way to route. withinRetryBudget(context.nodeName, RETRY_BUDGET) records the attempt and returns whether another remains:

ts
export class ExtractQueryNode extends MonadicNode<ArchivistState, 'success' | 'retry' | 'salvage'> {
  private readonly services: ArchivistServices;
  readonly name = 'extract-query';
  readonly '@id' = 'urn:noocodec:node:extract-query';
  constructor(services: ArchivistServices) {
    super();
    this.services = services;
  }
  readonly outputs = ['success', 'retry', 'salvage'] as const;
  override get outputSchema(): Record<'success' | 'retry' | 'salvage', SchemaObjectType> {
    return {
      'success': { 'type': 'object' },
      'retry':   { 'type': 'object' },
      'salvage': { 'type': 'object' },
    };
  }

  override async execute(batch: Batch<ArchivistState>, context: NodeContextType) {
    const successItems: ItemType<ArchivistState>[] = [];
    const retryItems: ItemType<ArchivistState>[] = [];
    const salvageItems: ItemType<ArchivistState>[] = [];

    for (const item of batch) {
      const { state } = item;
      const signal = Signal.compose({
        'deadlineMs': this.services.nodeTimeouts[context.nodeName] ?? NODE_TIMEOUT_MS,
        'signal':     context.signal,
      });
      try {
        const terms = await this.services.llm.extractTerms(state.query, signal);
        if (terms.length === 0) {
          if (state.withinRetryBudget(context.nodeName, RETRY_BUDGET)) {
            const result = NodeOutput.create('retry');
            for (const error of result.errors) state.collectError(error);
            retryItems.push(item);
          } else {
            state.clearAttempts(context.nodeName);
            const result = NodeOutput.create('salvage');
            for (const error of result.errors) state.collectError(error);
            salvageItems.push(item);
          }
          continue;
        }
        state.terms = terms;
        state.clearAttempts(context.nodeName);
        const result = NodeOutput.create('success');
        for (const error of result.errors) state.collectError(error);
        successItems.push(item);
      } catch (err) {
        // External cancellation / run deadline propagates unchanged.
        if (context.signal.aborted) throw err;
        // Node-local timeout or LLM failure -> retry budget decides the flow.
        if (state.withinRetryBudget(context.nodeName, RETRY_BUDGET)) {
          const result = NodeOutput.create('retry');
          for (const error of result.errors) state.collectError(error);
          retryItems.push(item);
        } else {
          state.clearAttempts(context.nodeName);
          const result = NodeOutput.create('salvage');
          for (const error of result.errors) state.collectError(error);
          salvageItems.push(item);
        }
      }
    }

    const routes: Array<readonly ['success' | 'retry' | 'salvage', Batch<ArchivistState>]> = [];
    if (successItems.length > 0) routes.push(['success', Batch.from(successItems)]);
    if (retryItems.length > 0) routes.push(['retry', Batch.from(retryItems)]);
    if (salvageItems.length > 0) routes.push(['salvage', Batch.from(salvageItems)]);
    return RoutedBatch.create(routes);
  }
}

Closing the loop in the DAG ​

The retry output is a self-edge; salvage routes to a recovery node that performs the deterministic recovery step and rejoins the happy path:

ts
.node(placement('extract-query'), extractQuery, {
  'success': placement('decide-tools'),
  'retry':   placement('extract-query'),          // flow-shape retry loop (self-edge)
  'salvage': placement('extract-query-salvage'),  // recovery route
}, display('extract-query'))
.node(placement('extract-query-salvage'), extractQuerySalvage, {
  'done': placement('decide-tools'),              // deterministic recovery rejoins the happy path
}, display('extract-query-salvage'))

Two-node retry loop ​

ComposeRetryLoopDAG is the same shape spread across two nodes (compose, validate, and a retry edge back to compose), built from plain .node() routes and bounded by state.retriesFor('compose'):

ts
/**
 * ComposeRetryLoopDAG: reusable compose / validate / retry loop.
 *
 * Internal flow:
 *
 *   compose-response
 *     └─ drafted ──► validate-response
 *          ├─ approved  ──► END (success) ─► parent: respond-to-visitor
 *          ├─ retry     ──► compose-response   (bounded by the retry budget on state (retriesFor('compose')))
 *          └─ exhausted ──► END (success) ─► parent: respond-to-visitor
 *
 * Outputs:
 *   success: draft composed (approved or best-effort); parent routes to
 *             the shared respond-to-visitor terminal.
 *   error:   clone-state errors accumulated (propagated by the parent
 *             ScatterNode to the parent's 'error' branch)
 *
 * Convergence policy: this sub-DAG does NOT contain respondToVisitor. It is a
 * pure compose/validate unit that produces state.draft and exits. The
 * single shared respond-to-visitor placement lives at the parent DAG level
 * so that every converging branch strikes exactly one terminal node per run.
 *
 * Molecular import pattern:
 *   import { composeRetryLoopDAG } from './embedded-dags/ComposeRetryLoopDAG.ts';
 *   const nodes = ArchivistNodes.build(services);
 *   dispatcher.registerBundle({ nodes: nodes.composeRetryLoopNodes, dags: [composeRetryLoopDAG] });
 *
 * The sub-DAG operates on the parent's state directly (no projection / gather
 * needed); it reads `state.shortlist` / `state.intent` / `state.priorContext`
 * and writes `state.draft` / `state.approved`, which the parent DAG already
 * manages. Every intent branch funnels through this one composed loop rather
 * than each branch owning its own compose→validate chain.
 */

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

import { DAGBuilder, DAGIdentity, PlaceholderNode } from '@studnicky/dagonizer';
import type { DAGType } from '@studnicky/dagonizer';

const COMPOSE_RETRY_LOOP_DAG_IRI = 'urn:noocodec:dag:compose-retry-loop';
const placement = (placementIdentifier: string): string => DAGIdentity.placementId(COMPOSE_RETRY_LOOP_DAG_IRI, placementIdentifier);
const display = <T extends string>(name: T): { name: T } => ({ name });

const composeResponse        = new PlaceholderNode<ArchivistState, 'drafted' | 'retry' | 'salvage'>('urn:noocodec:node:compose-response', ['drafted', 'retry', 'salvage']);
const composeResponseSalvage = new PlaceholderNode<ArchivistState, 'done'>('urn:noocodec:node:compose-salvage', ['done']);
const validateResponse       = new PlaceholderNode<ArchivistState, 'approved' | 'retry' | 'exhausted'>('urn:noocodec:node:validate-response', ['approved', 'retry', 'exhausted']);

export const composeRetryLoopDAG: DAGType = new DAGBuilder(COMPOSE_RETRY_LOOP_DAG_IRI, '1.1', display('compose-retry-loop'))

      // ── 1. compose-response ──────────────────────────────────────────────────
      // Writes state.draft via an intent-specific compose method. A transient LLM
      // failure routes 'retry' (loops back, bounded by the shared 'compose' budget)
      // or 'salvage' once spent; no in-node RetryPolicy. 'drafted' goes on to the
      // quality gate, which adds its own retry edge for low-quality drafts.
      .node(placement('compose-response'), composeResponse, {
        'drafted': placement('validate-response'),
        'retry':   placement('compose-response'),
        'salvage': placement('compose-salvage'),
      }, display('compose-response'))
      .node(placement('compose-salvage'), composeResponseSalvage, {
        'done': placement('composed'),
      }, display('compose-salvage'))

      // ── 2. validate-response ─────────────────────────────────────────────────
      // Quality gate: length, citations, tone. On 'retry', routes back to
      // compose (bounded by MAX_COMPOSE_ATTEMPTS via state.retriesFor('compose')).
      // 'approved' and 'exhausted' both exit via the canonical `composed`
      // TerminalNode (completed), so the parent EmbeddedDAGNode resolves 'success'
      // and routes to respond-to-visitor.
      .node(placement('validate-response'), validateResponse, {
        'approved':  placement('composed'),
        'retry':     placement('compose-response'),
        'exhausted': placement('composed'),
      }, display('validate-response'))

      // ── 3. composed ──────────────────────────────────────────────────────────
      // Canonical TerminalNode(completed): the single explicit exit of the compose
      // loop. No bare null end-of-flow routes.
      .terminal(placement('composed'), { outcome: 'completed', name: 'composed' })

      .build();

Details for Nerds ​

  • The retry budget on the conceptual root. state.recordAttempt(key), state.retriesFor(key), state.withinRetryBudget(key, max), and state.clearAttempts(key) live on NodeStateBase, keyed by context.nodeName, and ride along in the snapshot; a budget survives checkpoint/resume.
  • Retry is a loop edge. retry routes back to the same placement (a self-edge) or, for the compose loop, from validate-response to compose-response. The bound lives in state; the loop lives in the DAG. No special loop placement type, no acyclic constraint.
  • Salvage is a recovery route, not a fabrication. When the budget is spent the node routes salvage to a dedicated node: extract-query-salvage splits the query on whitespace, decide-tools-salvage emits a minimal tool plan. Recovery stays out of the failing node's catch.
  • External cancellation is not a retry. When context.signal is already aborted the node re-throws, so the engine records the run as cancelled rather than looping.

For per-operation retry with backoff (a flaky network call inside a tool or adapter), see RetryPolicy in the Retry guide.

See this in action in the Archivist live demo.

Watched over by the Order of Dagon.

MIT License, © 2026 Andrew Studnicky