@graph-compose/runtime

Open-source Temporal runtime for dependency-ordered HTTP workflow DAG execution. Define workflows as JSON graphs of HTTP nodes, execute them on your own Temporal cluster with automatic dependency resolution, JSONata expression templating, and runtime state inspection.

What This Package Does

| Feature | Description | |---------|-------------| | HTTP DAG execution | Execute HTTP nodes in dependency order across your workflow graph | | Expression resolution | JSONata templating in URLs, headers, and bodies via {{ results.x.data.y }}, {{ context.z }} | | Per-node retry & timeout | Configure retry policies and timeouts per node via activityConfig | | Context injection | Workflow-level context variables available to all nodes | | Graph validation | Cycle detection, dependency validation, expression syntax checks, HTTP-only node gate | | Temporal queries | Query execution state and individual node results mid-run | | HTTP activity | Execute HTTP requests as Temporal activities with configurable retries |

What This Package Does Not Do

The runtime is deliberately scoped to HTTP DAG execution. The following features are available on the Graph Compose platform:

• Error boundaries (try/catch around node groups)
• forEach / iterator child workflows
• ADK agent nodes (multi-agent AI orchestration)
• Confirmation / human-in-the-loop nodes
• Streaming execution
• State persistence
• Conditional branching
• Webhook notifications
• Secret resolution ($secret())
• Visual workflow builder and AI assistant
• Managed Temporal infrastructure

If you need to add custom node types, lifecycle hooks, or change how the execution loop works, see @graph-compose/execution-kernel — the lower-level package this runtime is built on.

Installation

npm install @graph-compose/runtime @graph-compose/core

Peer Dependencies

The runtime requires Temporal SDK packages, provided by your worker:

npm install @temporalio/workflow @temporalio/common @temporalio/worker @temporalio/client

Quick Start

1. Define a Workflow

A workflow is a JSON graph of HTTP nodes with explicit dependencies:

import type { WorkflowGraph } from "@graph-compose/core";

const workflow: WorkflowGraph = {
  nodes: [
    {
      id: "fetch_user",
      type: "http",
      dependencies: [],
      http: {
        method: "GET",
        url: "https://api.example.com/users/{{context.userId}}",
      },
    },
    {
      id: "enrich_profile",
      type: "http",
      dependencies: ["fetch_user"],
      http: {
        method: "POST",
        url: "https://api.example.com/enrich",
        headers: { "Content-Type": "application/json" },
        body: {
          name: "{{results.fetch_user.data.name}}",
          email: "{{results.fetch_user.data.email}}",
        },
      },
    },
    {
      id: "notify",
      type: "http",
      dependencies: ["enrich_profile"],
      http: {
        method: "POST",
        url: "https://api.example.com/notify",
        body: {
          message: "Profile enriched for {{context.userId}}",
        },
      },
      activityConfig: {
        retryPolicy: { maximumAttempts: 3, initialInterval: "1 second" },
        startToCloseTimeout: "15 seconds",
      },
    },
  ],
  context: { userId: "user_123" },
};

Nodes execute in dependency order. Nodes with no dependencies on each other run concurrently. The {{ }} expressions are JSONata templates resolved against the current workflow state.

2. Set Up a Temporal Worker

The worker runs the workflow code and activities. You need two imports:

• The workflow — loaded by path so Temporal can bundle it into its deterministic sandbox
• The activities — httpCall (makes HTTP requests) and resolveExpression (evaluates JSONata templates)

import { Worker } from "@temporalio/worker";
import { httpCall, resolveExpression } from "@graph-compose/runtime/activities";

async function run() {
  const worker = await Worker.create({
    workflowsPath: require.resolve("@graph-compose/runtime"),
    activities: { httpCall, resolveExpression },
    taskQueue: "http-worker",
  });

  await worker.run();
}

run().catch(console.error);

3. Start a Workflow

Use the Temporal client to start a workflow and query its state:

import { Client } from "@temporalio/client";
import type { RuntimeWorkflowInput } from "@graph-compose/runtime";

const client = new Client();

const handle = await client.workflow.start("runtimeWorkflow", {
  taskQueue: "http-worker",
  workflowId: "my-workflow-run",
  args: [
    {
      workflowGraph: workflow,
      context: { userId: "user_123" },
    } satisfies RuntimeWorkflowInput,
  ],
});

// Wait for completion
const result = await handle.result();
console.log(result);
// => { context: { userId: "user_123" }, results: { fetch_user: {...}, enrich_profile: {...}, notify: {...} }, workflowId: "my-workflow-run" }

4. Query State Mid-Run

Temporal queries let you inspect execution progress while the workflow is running:

import { WORKFLOW_QUERIES } from "@graph-compose/core";

const executionState = await handle.query(WORKFLOW_QUERIES.EXECUTION_STATE);
console.log("Executed nodes:", executionState.executed);
console.log("Results so far:", executionState.results);

const nodeResult = await handle.query(WORKFLOW_QUERIES.NODE_RESULT, "fetch_user");
console.log("fetch_user data:", nodeResult.data);

const nodeState = await handle.query(WORKFLOW_QUERIES.NODE_STATE, "fetch_user");
console.log("fetch_user state:", nodeState.executionState); // "pending" | "executed" | "executed_and_failed"

Execution Model

The runtime executes workflows in batched dependency order:

WorkflowGraph (JSON)
       │
       ▼
┌─────────────────────────────────────────────┐
│  Validation                                  │
│  - Only HTTP nodes allowed                   │
│  - All dependencies reference existing nodes │
│  - No cycles in the graph                    │
│  - All {{ }} expressions are valid JSONata   │
└──────────────────┬──────────────────────────┘
                   ▼
┌─────────────────────────────────────────────┐
│  Build DAG (directed acyclic graph)          │
└──────────────────┬──────────────────────────┘
                   ▼
┌─────────────────────────────────────────────┐
│  Execution Loop                              │
│                                              │
│  while (nodes remain) {                      │
│    batch = nodes whose deps are all done     │
│    for each node in batch (concurrently):    │
│      1. Resolve {{ }} expressions            │
│      2. Execute HTTP request (activity)      │
│      3. Record result                        │
│  }                                           │
└──────────────────┬──────────────────────────┘
                   ▼
           Return all results

Temporal handles retries, timeouts, and failure recovery at the activity level. Each node can configure its own retry policy and timeouts via activityConfig (see Per-Node Retry & Timeout). If a node's HTTP request fails and exhausts its retry policy, the entire workflow fails.

Expression Syntax

All string fields in a node's HTTP configuration support JSONata expressions wrapped in {{ }}:

// Reference results from completed nodes
"{{results.fetch_user.data.name}}"

// Reference workflow context variables
"{{context.apiKey}}"

// JSONata functions
"{{$uppercase(results.fetch_user.data.name)}}"

// Arithmetic
"{{results.price.data.amount * 1.1}}"

// Conditionals
'{{results.status.data.code = 200 ? "ok" : "error"}}'

// Array values in URLs are comma-joined and encoded
"https://api.example.com/users?ids={{results.list.data.ids}}"

Expressions are evaluated by JSONata against the workflow state:

| Variable | Contents | |----------|----------| | results | Completed node results keyed by node ID. Each result has { data, statusCode, headers }. | | context | Workflow-level context variables passed at workflow start. |

If an expression resolves to undefined, a warning is recorded but execution continues. If an expression has invalid syntax, the workflow fails during validation before any node executes.

Per-Node Retry & Timeout

Each HTTP node can configure its own Temporal activity retry policy and timeouts via activityConfig. Nodes without activityConfig use the default (startToCloseTimeout: "30 seconds", no retries).

{
  id: "flaky_api",
  type: "http",
  dependencies: [],
  http: { method: "GET", url: "https://unreliable-api.com/data" },
  activityConfig: {
    retryPolicy: {
      maximumAttempts: 5,          // retry up to 5 times
      initialInterval: "1 second", // first retry after 1s
      backoffCoefficient: 2,       // double the interval each retry
      maximumInterval: "30 seconds" // cap interval at 30s
    },
    startToCloseTimeout: "60 seconds",   // max time per attempt
    scheduleToCloseTimeout: "5 minutes", // max total time including retries
  },
}

This creates a dedicated Temporal activity proxy for that node with the specified configuration. Duration strings use the ms library format: "1 second", "30 seconds", "5 minutes", etc.

API Reference

Exports from @graph-compose/runtime

runtimeWorkflow(input)

The Temporal workflow function. This is what your worker loads via workflowsPath.

interface RuntimeWorkflowInput {
  workflowGraph: WorkflowGraph;
  context?: Record<string, unknown>;
  workflowInfo?: WorkflowInfo;
}

httpWorkflow is exported as an alias for runtimeWorkflow.

validateWorkflowGraph(workflow)

Validates a workflow graph and throws on the first failure. Checks:

1. HTTP-only nodes — non-HTTP node types are rejected (they require the platform)
2. Dependency targets exist — every dependencies entry references a real node ID
3. No cycles — the dependency graph is acyclic
4. Valid expressions — all {{ }} templates parse as valid JSONata

import { validateWorkflowGraph } from "@graph-compose/runtime";

validateWorkflowGraph(myWorkflow); // throws if invalid

Exports from @graph-compose/runtime/activities

These are Temporal activities registered with your worker.

resolveExpression(input): Promise<ExpressionResolutionOutput>

Resolves JSONata template expressions in a node's URL, headers, and body against the current workflow state.

httpCall(input): Promise<NodeResult>

Executes an HTTP request using axios. Returns { data, statusCode, headers }.

You can replace either activity with your own implementation — just provide functions with matching signatures when creating the worker:

const worker = await Worker.create({
  workflowsPath: require.resolve("@graph-compose/runtime"),
  activities: {
    resolveExpression, // use the default
    httpCall: myCustomHttpCall, // replace with your own
  },
  taskQueue: "http-worker",
});

Extending the Runtime

The runtime is built on @graph-compose/execution-kernel, which provides a pluggable orchestrator with support for custom node types, lifecycle hooks, and structural overrides.

If you need capabilities beyond HTTP nodes, see the kernel's README for:

• NodeHandler — add custom node types (Slack, email, database, etc.)
• ExecutionPlugin — hook into lifecycle events (logging, persistence, metrics)
• Template Methods — override graph building, scheduling, or the execution loop

Related Packages

| Package | Description | |---------|-------------| | @graph-compose/core | TypeScript types, Zod schemas, and validation utilities for workflow graphs | | @graph-compose/execution-kernel | Lower-level execution primitives — use this to build custom orchestrators | | @graph-compose/client | Fluent TypeScript SDK for the Graph Compose managed platform |

Requirements

• Node.js 18+
• A running Temporal server (local or cloud)
• TypeScript 5+ (recommended)

License

This project is dual-licensed:

• AGPL-3.0 for open-source use. See LICENSE for details.
• Commercial License available for organizations that need an alternative to AGPL. Contact the maintainers for details.