Primitive Nodes

Low-level building blocks for dynamic, composable workflows

Primitive Nodes

Low-level building blocks for composing complex, data-driven workflows.

Overview

Unlike high-level nodes that handle specific tasks, primitives provide simple building blocks that compose into sophisticated patterns:

Primitive Purpose LLM?
EvalNode Context manipulation, loop control No
DynamicAgentNode Agent with runtime config Yes
DynamicCommandNode Shell with runtime config No

Design Philosophy

Primitives over abstractions. Instead of creating a TaskIteratorNode that does everything, we provide simple building blocks:

Combined with dynamic then transitions, these primitives enable any control flow pattern.

EvalNode

Pure context transformation without LLM calls.

Use Cases

Configuration

schema.eval('INCREMENT', {
  // Pure function: state in, partial context out
  update: (state) => ({
    currentIndex: state.context.currentIndex + 1,
    currentTask: state.context.tasks[state.context.currentIndex + 1],
  }),

  // Transition (static or dynamic)
  then: (state) => state.context.currentTask ? 'EXECUTE' : 'DONE',
})

Result Storage

Stores metadata in state.context.lastEvalResult:

interface EvalResult {
  success: boolean;
  updatedKeys: string[];  // Keys that were updated
  duration: number;       // Execution time in ms
}

Examples

Increment Loop Counter

schema.eval('INCREMENT', {
  update: (state) => ({
    currentIndex: state.context.currentIndex + 1,
  }),
  then: 'CHECK_BOUNDS',
}),

Collect Results

schema.eval('COLLECT', {
  update: (state) => ({
    results: [...state.context.results, state.context.lastDynamicAgentResult],
  }),
  then: 'NEXT_ITEM',
}),

Set Current Item from Array

schema.eval('SET_CURRENT', {
  update: (state) => {
    const idx = state.context.currentIndex;
    return {
      currentTask: state.context.tasks[idx] ?? null,
    };
  },
  then: (state) => state.context.currentTask ? 'PROCESS' : 'DONE',
}),

DynamicAgentNode

AI agent with configuration resolved at runtime from workflow state.

Use Cases

Configuration

schema.dynamicAgent('EXECUTE_TASK', {
  // Model: static or dynamic
  model: (state) => state.context.currentTask.model,  // 'haiku' | 'sonnet' | 'opus'

  // Prompt: static or dynamic
  prompt: (state) => state.context.currentTask.prompt,

  // Optional system prompt (separate from user prompt)
  system: 'You are a helpful assistant.',

  // Tools: static or dynamic
  capabilities: (state) => state.context.currentTask.tools ?? [StdlibTool.Read, StdlibTool.Write],

  // Optional settings
  maxTurns: 10,
  temperature: 0,
  maxTokens: 4096,

  then: 'COLLECT_RESULT',
})

Dynamic Type

All configuration values support the Dynamic<T> type:

// Static value
model: AgentModel.Sonnet

// Dynamic value from state
model: (state) => state.context.currentTask.model

Result Storage

Stores result in state.context.lastDynamicAgentResult:

interface DynamicAgentResult {
  success: boolean;
  response: string;
  model: string;          // Actual model ID used
  usage: {
    inputTokens: number;
    outputTokens: number;
  };
  error?: string;
  duration: number;
}

Example: Task with Dynamic Model

schema.dynamicAgent('EXECUTE_TASK', {
  model: (state) => {
    // Use opus for complex tasks, haiku for simple ones
    const task = state.context.currentTask;
    return task.complexity === 'high' ? AgentModel.Opus : AgentModel.Haiku;
  },
  prompt: (state) => state.context.currentTask.prompt,
  capabilities: [StdlibTool.Read, StdlibTool.Write, StdlibTool.Bash],
  then: 'VERIFY',
}),

DynamicCommandNode

Shell command with configuration resolved at runtime.

Use Cases

Configuration

schema.dynamicCommand('RUN_SCRIPT', {
  // Command: static or dynamic (string or array)
  command: (state) => state.context.currentTask.command,

  // Working directory: static or dynamic
  cwd: (state) => state.context.projectDir,

  // Environment variables: static or dynamic
  env: (state) => ({
    NODE_ENV: 'test',
    ...state.context.envOverrides,
  }),

  // Timeout in ms: static or dynamic
  timeout: 60000,

  then: 'CHECK_RESULT',
})

Command Formats

Commands support both string and array forms:

// String: shell execution (supports pipes, redirects)
command: (state) => `bun test ${state.context.module}`

// Array: direct execution (safer for user input)
command: (state) => ['gh', 'pr', 'create', '--title', state.context.title]

Use array form when including user-provided values to prevent injection.

Result Storage

Stores result in state.context.lastDynamicCommandResult:

interface DynamicCommandResult {
  exitCode: number;
  stdout: string;
  stderr: string;
  success: boolean;      // exitCode === 0
  command: string;       // The resolved command
  duration: number;
}

Example: Run Tests for Module

schema.dynamicCommand('RUN_TEST', {
  command: (state) => `bun test ${state.context.currentModule}`,
  cwd: '/path/to/project',
  timeout: 120000,
  then: (state) => {
    const result = state.context.lastDynamicCommandResult;
    return result?.success ? 'NEXT_MODULE' : 'FIX_TEST';
  },
}),

Complete Example: Task Iterator Pattern

This example shows how to iterate over a list of tasks generated by a planning agent, executing each with the specified model.

import { z } from 'zod';
import { defineNodes, defineWorkflow, StdlibTool, AgentModel } from '@sys/graph';
import { LLMNodeRuntime } from '@sys/graph/nodes';

interface Task {
  prompt: string;
  model: 'haiku' | 'sonnet' | 'opus';
}

interface TaskResult {
  success: boolean;
  response: string;
}

interface TaskContext extends Record<string, unknown> {
  request: string;
  tasks: Task[];
  currentIndex: number;
  currentTask: Task | null;
  results: TaskResult[];
  planResult?: { output?: { tasks: Task[] } };
  lastDynamicAgentResult?: { success: boolean; response: string };
  lastCommandResult?: { exitCode: number };
}

// Define the output schema for the planner
const TaskPlanSchema = z.object({
  tasks: z.array(z.object({
    prompt: z.string(),
    model: z.enum(['haiku', 'sonnet', 'opus']),
  })),
});

// Define schema with all node names
const schema = defineNodes<TaskContext>()([
  'PLAN',
  'INIT_ITERATION',
  'NEXT_TASK',
  'EXECUTE',
  'COLLECT_RESULT',
  'FINALIZE',
] as const);

export default defineWorkflow({
  id: 'task-iterator',
  schema,
  initialContext: {
    request: '',
    tasks: [],
    currentIndex: -1,
    currentTask: null,
    results: [],
  },

  nodes: [
    // 1. Generate task plan using LLMNode (direct API call)
    // Note: LLMNode uses runtime class directly, not schema method
    // For schema-based workflows, use an AgentNode or custom implementation
    schema.agent('PLAN', {
      role: 'planner',
      prompt: `You are a task planner. Break down the request into tasks.
Output JSON: { "tasks": [{ "prompt": "...", "model": "haiku"|"sonnet"|"opus" }...] }`,
      capabilities: [StdlibTool.Read],
      then: 'INIT_ITERATION',
    }),

    // 2. Initialize iteration
    schema.eval('INIT_ITERATION', {
      update: (state) => ({
        tasks: state.context.planResult?.output?.tasks ?? [],
        currentIndex: -1,
        results: [],
      }),
      then: 'NEXT_TASK',
    }),

    // 3. Advance to next task
    schema.eval('NEXT_TASK', {
      update: (state) => {
        const nextIndex = state.context.currentIndex + 1;
        const tasks = state.context.tasks;
        return {
          currentIndex: nextIndex,
          currentTask: tasks[nextIndex] ?? null,
        };
      },
      then: (state) => state.context.currentTask ? 'EXECUTE' : 'FINALIZE',
    }),

    // 4. Execute current task with dynamic model
    schema.dynamicAgent('EXECUTE', {
      model: (state) => state.context.currentTask!.model as AgentModel,
      prompt: (state) => state.context.currentTask!.prompt,
      capabilities: [StdlibTool.Read, StdlibTool.Write, StdlibTool.Bash],
      then: 'COLLECT_RESULT',
    }),

    // 5. Collect result and loop back
    schema.eval('COLLECT_RESULT', {
      update: (state) => {
        const agentResult = state.context.lastDynamicAgentResult;
        return {
          results: [
            ...state.context.results,
            {
              success: agentResult?.success ?? false,
              response: agentResult?.response ?? '',
            },
          ],
        };
      },
      then: 'NEXT_TASK',
    }),

    // 6. Final step
    schema.command('FINALIZE', {
      command: 'bun test',
      then: 'END',
    }),
  ],
});

Best Practices

1. Keep EvalNode Functions Pure

// Good: pure function, no side effects
update: (state) => ({
  count: state.context.count + 1,
})

// Bad: side effects
update: (state) => {
  console.log('Processing...');  // Side effect
  state.context.count++;          // Mutation
  return state.context;
}

2. Use Type Guards for Dynamic Values

schema.dynamicAgent('EXECUTE', {
  model: (state) => {
    const task = state.context.currentTask;
    // Guard against null/undefined
    if (!task) return AgentModel.Haiku;
    return task.model;
  },
  prompt: (state) => state.context.currentTask?.prompt ?? 'No task',
  then: 'NEXT',
}),

3. Combine with Dynamic Transitions

// EvalNode sets up state, dynamic then handles flow control
schema.eval('ADVANCE', {
  update: (state) => ({
    index: state.context.index + 1,
    current: state.context.items[state.context.index + 1],
  }),
  then: (state) => {
    if (!state.context.current) return 'DONE';
    if (state.context.current.skip) return 'ADVANCE';  // Skip item
    return 'PROCESS';
  },
}),

4. Use Descriptive Result Keys

// For multiple dynamic agents in a workflow
schema.dynamicAgent('PLAN', {
  ...config,
  resultKey: 'planResult',  // Custom key
  then: 'IMPLEMENT',
}),

schema.dynamicAgent('IMPLEMENT', {
  ...config,
  resultKey: 'implementResult',  // Different key
  then: 'VERIFY',
}),