Pipelines

The pipeline tool for defining, executing, and managing multi-agent execution graphs

What it does: Lets agents define, run, save, and monitor multi-agent execution graphs (DAGs). Each node is a sub-agent task; independent nodes run in parallel, dependent nodes wait their turn.

Who it’s for: Anyone who wants to coordinate more than one specialist agent on a single problem — research teams, code review rotations, debate / vote setups, multi-step content workflows, gated trade execution.

For the conceptual background — node types, barrier modes, context modes, scheduling, and the architecture of the graph coordinator — see Execution Graphs. This page is the developer reference for the pipeline tool itself.

From natural language to graph

When a user describes a multi-agent workflow in chat, the agent translates it into a pipeline call. For example:

You: Have four analysts research NVDA in parallel, then write a final
     buy/hold/sell recommendation that incorporates all four reports.

The agent calls pipeline action: execute with a 5-node graph. Four research nodes have no depends_on (they run in parallel); the final recommendation node depends on all four (barrier_mode: all):

tool: pipeline
action: execute
label: "NVDA multi-analyst research"
nodes:
  - node_id: fundamentals
    task: "Research NVDA fundamentals: revenue, margins, guidance."
    agent: equity-analyst
  - node_id: technicals
    task: "Analyze NVDA technicals: trend, support/resistance, volume."
    agent: ta-analyst
  - node_id: macro
    task: "Assess NVDA's macro context: AI cycle, capex, geopolitics."
    agent: macro-analyst
  - node_id: sentiment
    task: "Sample NVDA news/social sentiment from the past 7 days."
    agent: sentiment-analyst
  - node_id: recommend
    depends_on: [fundamentals, technicals, macro, sentiment]
    barrier_mode: all
    task: |
      Synthesize a final BUY/HOLD/SELL recommendation for NVDA.
      Fundamentals: {{fundamentals.result}}
      Technicals:   {{technicals.result}}
      Macro:        {{macro.result}}
      Sentiment:    {{sentiment.result}}
    agent: lead-pm

The graph coordinator runs the four analysts concurrently, blocks the recommendation node until all four return, then injects each upstream result into the prompt via {{nodeId.result}}. You can monitor the run with pipeline action: status graph_id: <id> and grab final outputs with pipeline action: outputs graph_id: <id>.

Prerequisite

Agent-to-agent communication must be enabled in your configuration:

security:
  agentToAgent:
    enabled: true

Without this setting, the pipeline tool will not be available.

Actions

The pipeline tool supports 10 actions. The default action is execute if no action is specified.

define

Validates a graph structure without executing it. Returns the topological execution order and any warnings. Use this to check a graph before running it.

tool: pipeline
action: define
nodes:
  - node_id: search
    task: "Search for TypeScript best practices"
  - node_id: summarize
    task: "Summarize the search results: {{search.result}}"
    depends_on: [search]
label: "TS Best Practices"

execute

Starts a graph and returns a graphId for monitoring. Root nodes (no dependencies) begin immediately.

tool: pipeline
action: execute
nodes:
  - node_id: search
    task: "Search for TypeScript best practices"
  - node_id: summarize
    task: "Summarize the findings: {{search.result}}"
    depends_on: [search]
label: "TS Best Practices"

status

Check the status of a specific graph or list recent graph executions.

# Check a specific graph
tool: pipeline
action: status
graph_id: "abc-123-def"

# List recent graphs from the last 30 minutes
tool: pipeline
action: status
recent_minutes: 30

cancel

Stop a running graph. All running nodes are terminated and remaining nodes are skipped. Requires user confirmation (_confirmed: true) because it is a destructive action gated by the action classifier.

tool: pipeline
action: cancel
graph_id: "abc-123-def"
_confirmed: true

save

Persist a graph definition for reuse. Edges are automatically derived from depends_on if not explicitly provided.

tool: pipeline
action: save
label: "Research Pipeline"
nodes:
  - node_id: search
    task: "Search for ${TOPIC}"
  - node_id: analyze
    task: "Analyze findings: {{search.result}}"
    depends_on: [search]
  - node_id: write
    task: "Write a report based on: {{analyze.result}}"
    depends_on: [analyze]

load

Retrieve a previously saved graph definition by its ID.

tool: pipeline
action: load
id: "research-pipeline-id"

list

Browse saved graph definitions with optional pagination.

tool: pipeline
action: list
limit: 10
offset: 0

delete

Remove a saved graph definition. Requires user confirmation because it is a destructive action.

tool: pipeline
action: delete
id: "research-pipeline-id"
_confirmed: true

outputs

Retrieve node output values from a completed or running graph. Uses memory-first retrieval with disk fallback for expired graphs. Each node output is truncated at 12,000 characters.

tool: pipeline
action: outputs
graph_id: "abc-123-def"

Response:

{
  "graphId": "abc-123-def",
  "outputs": {
    "search": "Found 5 sources on quantum computing...",
    "analyze": "Key trends: 1) ..., 2) ..., 3) ...",
    "write": null
  },
  "source": "memory"
}

Nodes that have not completed yet return null. The source field indicates whether outputs were retrieved from the in-memory coordinator ("memory") or from persisted output files on disk ("disk").

from_intent

Synthesize a multi-agent graph from a one-line intent — a canonical pattern plus a few agent/task names — and run it, without hand-writing the node list. The synthesizer deterministically expands one of four canonical patterns and dispatches the result through the same execute path, so all graph validation and sub-agent governance apply automatically.

tool: pipeline
action: from_intent
pattern: debate          # research-fanout | debate | vote | map-reduce
agents: ["bull", "bear"] # debate: [PRO, CON]; vote: voter pool; map-reduce: mapper pool
tasks: ["should we ship v2?"]  # fills the TOPIC/TASK slot (tasks[0])

Patterns:

Pattern	Shape	Slot inputs
research-fanout	N researchers → synthesize	tasks[0] (the topic)
debate	pro vs con → moderator	agents: [PRO, CON] (2 required) + tasks[0]
vote	N voters → aggregate	tasks[0] (+ optional agents naming the pool)
map-reduce	N mappers → reduce	tasks[0] (+ optional agents naming the pool)

An invalid intent (for example debate with fewer than two agents) is rejected before any graph runs. from_intent is gated by the orchestration.authoring.intentAction config flag (default true — on out of the box); set it false to make the daemon refuse the call. Each successful synthesis emits a graph:synthesized_from_intent audit event.

from_intent produces template-shaped graphs — plain agent nodes ({ nodeId, task, dependsOn }) wired in the named fan-out → fan-in shape. It does not emit the typed orchestration drivers: a synthesized debate is a one-shot pro/con fan-in into a moderator node, not the multi-round type_id: debate driver (with type_config.rounds / synthesizer). For the typed multi-round drivers, author the graph with define/execute and per-node type_id + type_config (see Node configuration below). from_intent accordingly takes no rounds parameter.

Node configuration

Each node in the nodes array accepts the following parameters:

Parameter	Type	Required	Description
node_id	string	Yes	Unique identifier within the graph
task	string	Yes	Task description for the sub-agent. Use {{nodeId.result}} to inline upstream output. Use ${VAR} for user-provided inputs.
depends_on	string[]	No	Node IDs that must complete before this node runs
agent	string	No	Agent ID to run this node (defaults to caller’s agent)
model	string	No	Model override for this node
timeout_ms	integer	No	Per-node timeout in milliseconds (default: 300000)
max_steps	integer	No	Maximum agentic steps for the sub-agent
barrier_mode	string	No	all (default), majority, or best-effort
retries	integer	No	Number of automatic retries on failure (0-3, default: 0). Exponential backoff delays: 1s, 2s, 4s.
type_id	string	No	Built-in node type: agent, debate, vote, refine, collaborate, approval-gate, or map-reduce. If omitted, the node runs as a regular single-agent task.
type_config	object	No	Type-specific configuration. Required when type_id is set. See Node Types for config schemas per type.
context_mode	string	No	Upstream output verbosity: full (default, complete outputs), summary (500 chars + shared dir reference), none (no automatic injection — use {{nodeId.result}} for explicit data).
mcp_servers	string[]	No	MCP server names whose tools should be pre-discovered for this node. Example: ["yfinance"] pre-seeds the yfinance tools so the sub-agent can skip discover_tools calls.

Graph settings

Top-level parameters that apply to the entire graph:

Parameter	Type	Required	Description
action	string	No	Pipeline action (default: execute)
nodes	array	For define/execute/save	Array of node definitions
label	string	No	Human-readable graph label
on_failure	string	No	fail-fast (default) or continue
timeout_ms	integer	No	Graph-level timeout in milliseconds (default: 1500000 / 25 min). Automatically raised to the DAG’s critical-path makespan (topological depth × concurrency waves × per-node timeout) when set too low, so later phases — e.g. a debate or final-synthesis node — are not starved by earlier parallel ones.
graph_id	string	For status/cancel/outputs	ID of a running/completed graph
id	string	For save/load/delete	Named graph ID for persistence
budget	object	No	{ max_tokens, max_cost } — resource limits
variables	object	No	Key-value pairs for ${VARIABLE_NAME} substitution in task text at execute time
edges	array	No	Explicit edge definitions (auto-derived if omitted)
settings	object	No	Additional graph settings
limit	integer	No	Max entries for list action
offset	integer	No	Pagination offset for list action
recent_minutes	integer	No	Filter status listing to last N minutes
_confirmed	boolean	No	Required for cancel/delete after user approval

Edges

Edges define structural connections between nodes. They are auto-derived from depends_on if not explicitly provided:

edges:
  - source: search
    target: analyze
  - from: analyze    # "from" is an alias for "source"
    to: write        # "to" is an alias for "target"

Each edge can have an id (auto-generated as source->target if omitted), and source/target (or from/to aliases).

Data flow

Nodes receive upstream results through {{nodeId.result}} template interpolation directly in the task text. The nodeId must appear in the node’s depends_on array.

nodes:
  - node_id: search
    task: "Search for information about quantum computing"
  - node_id: analyze
    task: "Analyze the search results: {{search.result}}"
    depends_on: [search]

In this example, search is the upstream node ID referenced in depends_on, and {{search.result}} is replaced with the search node’s output at runtime.

Use ${VARIABLE_NAME} for user-provided inputs that the web dashboard prompts for before execution. Use {{nodeId.result}} to inline upstream node outputs. These are two different interpolation syntaxes — ${VAR} is resolved at execute time from the variables parameter, while {{nodeId.result}} is resolved at runtime when the upstream node completes.

Important constraints:

• Template references ({{X.result}}) must use a node ID that appears in the same node’s depends_on array
• Outputs exceeding 12,000 characters are automatically truncated
• If an upstream node did not complete (failed or skipped), the template is replaced with [unavailable: node "nodeId" did not complete]
• Additionally, upstream outputs are automatically injected as context for downstream nodes. Set context_mode: "none" to disable automatic injection and rely solely on {{nodeId.result}} templates.

Shared data folder

Each graph execution creates a temporary directory at ~/.comis/graph-runs/{graphId}/. All nodes can read and write files here, making it ideal for exchanging large payloads between nodes.

- node_id: generate
  task: "Generate a CSV report and save it to the shared pipeline folder"
- node_id: analyze
  task: "Read the CSV report from the shared pipeline folder and summarize it"
  depends_on: [generate]

The shared folder path is automatically provided to each sub-agent.

Error handling

Failure policies

Policy	When a node fails
fail-fast	All downstream dependents are immediately skipped. The graph completes once remaining running nodes finish.
continue	Only dependents whose barrier can never be satisfied are skipped. Other nodes continue running if their barrier may still be met.

Barrier evaluation

With fail-fast, any failed dependency cascades to skip all downstream nodes. With continue, the barrier mode determines whether a downstream node runs or is skipped:

• all: skipped if any dependency failed
• majority: skipped only if more than half the dependencies failed
• best-effort: skipped only if every dependency failed

Node status lifecycle

pending -> ready -> running -> completed
                           \-> failed
                \----------\-> skipped

A node starts as pending, becomes ready when its dependencies are satisfied (or immediately if it has none), transitions to running when spawned, and ends as completed, failed, or skipped.

Budget and timeout cancellation

If the graph exceeds its budget (max_tokens or max_cost) or timeout, all running nodes are killed and remaining nodes are skipped. The graph:completed event includes a cancelReason field (timeout, budget, or manual).

Examples

Research pipeline

A sequential chain where each step builds on the previous:

tool: pipeline
action: execute
nodes:
  - node_id: search
    task: "Search the web for the latest developments in renewable energy storage"
  - node_id: analyze
    task: "Analyze the search results and identify the top 3 trends: {{search.result}}"
    depends_on: [search]
  - node_id: write
    task: "Write a 1000-word article about renewable energy storage trends based on: {{analyze.result}}"
    depends_on: [analyze]
label: "Renewable Energy Research"

Code review pipeline

Parallel analysis steps that feed into a final decision:

tool: pipeline
action: execute
nodes:
  - node_id: lint
    task: "Run the linter on the repository and report all findings"
  - node_id: test
    task: "Run the test suite and report results including any failures"
  - node_id: review
    task: "Review the code changes in the latest commit for security and performance issues"
  - node_id: merge
    task: "Based on lint results, test results, and code review, decide whether to approve the merge: lint={{lint.result}}, tests={{test.result}}, review={{review.result}}"
    depends_on: [lint, test, review]
    barrier_mode: all
label: "Code Review"
on_failure: continue

Content pipeline with budget limits

A production pipeline with cost controls:

tool: pipeline
action: execute
nodes:
  - node_id: outline
    task: "Create a detailed outline for a blog post about ${TOPIC}"
  - node_id: draft
    task: "Write the full blog post based on this outline: {{outline.result}}"
    depends_on: [outline]
    model: "claude-sonnet-4-5-20250929"
  - node_id: edit
    task: "Edit and proofread the draft for clarity and grammar: {{draft.result}}"
    depends_on: [draft]
  - node_id: seo
    task: "Generate SEO metadata (title, description, keywords) for: {{draft.result}}"
    depends_on: [draft]
label: "Blog Content Pipeline"
budget:
  max_tokens: 100000
  max_cost: 2.00
timeout_ms: 300000
variables:
  TOPIC: "AI-powered code review tools"

Debate pipeline

A pipeline with adversarial debate for balanced analysis:

tool: pipeline
action: execute
nodes:
  - node_id: research
    task: "Research the pros and cons of microservices vs monolith architecture"
    retries: 2
  - node_id: evaluate
    task: "Is microservices the right architecture for a 10-person startup? Consider: {{research.result}}"
    depends_on: [research]
    type_id: debate
    type_config:
      agents: [optimist-architect, pragmatic-architect]
      rounds: 3
      synthesizer: lead-architect
  - node_id: recommend
    task: "Write a final recommendation based on the debate outcome: {{evaluate.result}}"
    depends_on: [evaluate]
    context_mode: none
label: "Architecture Decision"

Approval-gated pipeline

A pipeline that pauses for human approval before executing a high-stakes action:

tool: pipeline
action: execute
nodes:
  - node_id: analyze
    task: "Analyze the proposed trade for ${TICKER}"
    agent: ta-analyst
  - node_id: approval
    depends_on: [analyze]
    task: "Review analysis before proceeding"
    type_id: approval-gate
    type_config:
      message: "Analysis complete for ${TICKER}. Approve to proceed with execution?"
      timeout_minutes: 60
  - node_id: execute
    task: "Execute the approved trade plan: {{analyze.result}}"
    depends_on: [approval]
    agent: ta-trader
label: "Gated Trade Execution"
variables:
  TICKER: "NVDA"

Vote pipeline

A pipeline where multiple agents vote independently:

tool: pipeline
action: execute
nodes:
  - node_id: research
    task: "Research ${TICKER} fundamentals and technicals"
    agent: ta-researcher
  - node_id: vote
    depends_on: [research]
    task: "Based on the research, should we buy ${TICKER}?"
    type_id: vote
    type_config:
      voters: [analyst-1, analyst-2, analyst-3, analyst-4]
      verdict_format: "BUY/HOLD/SELL"
label: "Analyst Vote"
variables:
  TICKER: "AAPL"

Agent (typed) pipeline

A pipeline that explicitly sets type_id: agent with driver-level configuration for model and step limits:

tool: pipeline
action: execute
nodes:
  - node_id: gather
    task: "Collect the latest financial data for ${TICKER}"
    agent: data-collector
  - node_id: analyze
    depends_on: [gather]
    task: "Perform deep technical analysis on: {{gather.result}}"
    type_id: agent
    type_config:
      agent: senior-analyst
      model: "claude-sonnet-4-5-20250929"
      max_steps: 15
  - node_id: report
    depends_on: [analyze]
    task: "Write an executive summary from: {{analyze.result}}"
label: "Typed Agent Analysis"
variables:
  TICKER: "TSLA"

Setting type_id: agent explicitly is optional — a node without type_id runs as a single-agent task by default. Use the explicit form when you need to override the agent’s model or step limit via type_config while using the driver system.

Refine pipeline

A pipeline where multiple reviewers sequentially improve a draft, each building on the previous reviewer’s output:

tool: pipeline
action: execute
nodes:
  - node_id: draft
    task: "Write a first draft of the ${DOC_TYPE} about ${TOPIC}"
    agent: content-writer
  - node_id: refine
    depends_on: [draft]
    task: "Refine this draft through sequential review: {{draft.result}}"
    type_id: refine
    type_config:
      reviewers: [style-editor, fact-checker, copy-editor]
  - node_id: publish
    depends_on: [refine]
    task: "Format the refined document for publication: {{refine.result}}"
label: "Editorial Review Chain"
variables:
  DOC_TYPE: "technical blog post"
  TOPIC: "building reliable AI pipelines"

Collaborate pipeline

A pipeline where agents contribute sequentially, each adding a different perspective to the accumulated work:

tool: pipeline
action: execute
nodes:
  - node_id: brief
    task: "Write a project brief for ${PROJECT}"
  - node_id: collaborate
    depends_on: [brief]
    task: "Develop a comprehensive plan based on: {{brief.result}}"
    type_id: collaborate
    type_config:
      agents: [backend-architect, frontend-lead, security-reviewer]
      rounds: 2
  - node_id: finalize
    depends_on: [collaborate]
    task: "Compile the collaborative output into a final project plan: {{collaborate.result}}"
label: "Cross-Team Collaboration"
variables:
  PROJECT: "user authentication redesign"

Map-reduce pipeline

A pipeline that fans out work to parallel mappers and then reduces their outputs into a single result:

tool: pipeline
action: execute
nodes:
  - node_id: plan
    task: "Break ${TOPIC} into 3 independent research areas"
  - node_id: research
    depends_on: [plan]
    task: "Research the assigned area in depth: {{plan.result}}"
    type_id: map-reduce
    type_config:
      mappers:
        - agent: researcher-1
          task_suffix: "Focus on market trends"
        - agent: researcher-2
          task_suffix: "Focus on technical feasibility"
        - agent: researcher-3
          task_suffix: "Focus on competitive landscape"
      reducer: lead-analyst
      reducer_prompt: "Synthesize all research into a unified report with key findings and recommendations"
  - node_id: summary
    depends_on: [research]
    task: "Write an executive summary from: {{research.result}}"
label: "Parallel Research Analysis"
variables:
  TOPIC: "enterprise AI adoption"

Related

Execution Graphs

Conceptual guide to execution graph architecture

Visual Builder

Design and run graphs from the web dashboard

Tool Policy

Configure pipeline tool access via tool policy

Sessions Guide

Sub-agent sessions for simpler delegation