Workflow DAGs

JarvisCore's WorkflowBuilder lets you compose multi-agent workflows as Directed Acyclic Graphs (DAGs). Each step is assigned to a specific agent role, steps declare their dependencies, and the framework executes them in topological order: parallelising independent steps automatically.

This guide covers the full WorkflowBuilder API, the result reference syntax for chaining step outputs, Redis-backed persistence, and production patterns.

---

Core Concepts

A workflow is a named, executable DAG composed of steps. Each step has:

• A unique step_id (the key you use to reference its output downstream)
• An agent role to dispatch the task to
• A task description (natural language)
• An optional depends_on list of step IDs that must succeed first

Steps without dependencies are eligible to run immediately. When all dependencies of a step have succeeded, that step becomes eligible. The framework runs all eligible steps concurrently in each round.

---

Building a Workflow

from jarviscore.orchestration.workflow_builder import WorkflowBuilder

wf = (
    WorkflowBuilder()
    .step("research",  "researcher", "Gather market data on EV adoption in Europe for Q1 2026.")
    .step("analyse",   "analyst",    "Analyse findings: {research.result}", depends_on=["research"])
    .step("draft",     "writer",     "Draft executive report from: {analyse.result}", depends_on=["analyse"])
    .step("review",    "reviewer",   "QA the draft for accuracy and tone.", depends_on=["draft"])
    .build(title="EV Market Report", team="market-intelligence")
)

WorkflowBuilder.step

builder.step(
    step_id="research",
    agent="researcher",
    task="Gather market data on topic X.",
    depends_on=["prior_step_id"],  # optional
)

Parameter	Type	Required	Description
step_id	str	Yes	Unique step identifier. Used for {step_id.result} references in downstream tasks.
agent	str	Yes	Role of the agent to dispatch this step to (must match the role attribute on the agent class).
task	str	Yes	Natural language task description. May contain {step_id.result} placeholders.
depends_on	list[str]	No	List of step_id values that must succeed before this step runs.

[!IMPORTANT] Steps must be declared in topological order. If step B depends on step A, you must call .step("A", ...) before .step("B", ..., depends_on=["A"]). Declaring a dependency on an undeclared step raises ValueError immediately at build time.

WorkflowBuilder.build

workflow = builder.build(title="My Pipeline", team="my-team")

Returns a Workflow object ready for registration and execution. Raises ValueError if no steps have been added.

---

Result Reference Syntax

Downstream steps can reference the output of upstream steps using {step_id.result} placeholders in the task description:

.step("analyse", "analyst", "Analyse these findings: {research.result}", depends_on=["research"])

At execution time, the placeholder is replaced with the actual output of the research step before the task is dispatched to the agent. If the referenced step's output is a dict, the framework extracts the output or result key. The substituted value is capped at 500 characters.

---

Registering and Executing

register

Persists the DAG to Redis. This enables dashboard visibility and cross-node coordination.

workflow_id = await wf.register(redis_store)

If redis_store is None, the workflow runs in-memory only. The workflow_id is still returned and can be used for local tracking.

execute

Runs the workflow by dispatching steps to agents via the Mesh.

results = await wf.execute(mesh, redis_store=redis_store, timeout_per_step=300)

Parameter	Type	Default	Description
mesh	Mesh	Required	The active JarvisCore Mesh instance
redis_store	RedisContextStore	None	When set, step status is written to Redis in real time
timeout_per_step	int	300	Seconds per step before it is marked timeout

Returns a list of step result dicts:

[
    {"step_id": "research", "agent": "researcher", "status": "success", "output": {...}, "error": None, "elapsed_ms": 4820},
    {"step_id": "analyse",  "agent": "analyst",    "status": "success", "output": {...}, "error": None, "elapsed_ms": 3100},
    {"step_id": "draft",    "agent": "writer",     "status": "success", "output": {...}, "error": None, "elapsed_ms": 6450},
    {"step_id": "review",   "agent": "reviewer",   "status": "success", "output": {...}, "error": None, "elapsed_ms": 2200},
]

Possible status values: success, failure, timeout.

A failed step does not prevent its independent sibling steps from running. Steps that depend on a failed step are blocked.

---

Full Example: Research-to-Report Pipeline

agents/orchestrator.py

import asyncio
from jarviscore import CustomAgent, Mesh
from jarviscore.orchestration.workflow_builder import WorkflowBuilder


class OrchestratorAgent(CustomAgent):
    name = "Orchestrator"
    role = "orchestrator"
    description = "Composes and executes research-to-report workflows."

    async def on_peer_request(self, msg) -> dict:
        mesh = self._mesh  # Mesh injects itself at add() time

        task = msg.data.get("task", "")

        # 1. Compose the DAG
        wf = (
            WorkflowBuilder()
            .step(
                "intel",
                "researcher",
                f"Research the following and return a structured summary: {task}",
            )
            .step(
                "analysis",
                "analyst",
                "Analyse the research and identify three key insights: {intel.result}",
                depends_on=["intel"],
            )
            .step(
                "report",
                "writer",
                "Write a 500-word executive report from these insights: {analysis.result}",
                depends_on=["analysis"],
            )
            .build(title=f"Research pipeline: {task[:60]}", team="research-ops")
        )

        # 2. Register with Redis (enables dashboard visibility)
        workflow_id = await wf.register(self._redis_store)
        self._logger.info("Workflow %s registered", workflow_id)

        # 3. Execute
        results = await wf.execute(
            mesh,
            redis_store=self._redis_store,
            timeout_per_step=600,
        )

        # 4. Return summary
        final_step = next((r for r in results if r["step_id"] == "report"), None)
        return {
            "workflow_id": workflow_id,
            "steps": len(results),
            "successes": sum(1 for r in results if r["status"] == "success"),
            "report": final_step["output"] if final_step else None,
        }

---

Parallel Steps

Steps without mutual dependencies are dispatched concurrently. Use this to parallelise independent work:

wf = (
    WorkflowBuilder()
    # Research and competitive scan run in parallel
    .step("market_research", "researcher", "Gather market size data for EMEA.")
    .step("competitor_scan", "researcher", "Identify top 5 competitors and their positioning.")
    # Synthesis waits for both
    .step(
        "synthesis",
        "analyst",
        "Synthesise market data and competitive landscape: {market_research.result} + {competitor_scan.result}",
        depends_on=["market_research", "competitor_scan"],
    )
    .build(title="EMEA Market Analysis")
)

The market_research and competitor_scan steps are dispatched simultaneously. synthesis begins only after both complete successfully.

---

Dynamic Fan-Out: mesh.fanout()

A workflow declares its steps upfront. That works when you know the step list while writing the code. Sometimes you only learn how many items there are at runtime: search results, a list of files, a board of symbols. For that case, use mesh.fanout(). It runs one task template over each item, a few at a time, and collects the results.

result = await mesh.fanout(
    "board-scan",
    agent="analyst",                       # role or capability
    items=symbols,                         # your runtime list
    task=lambda s: f"Deep-read {s} and return a thesis JSON.",
    context=lambda s: {"symbol": s},       # per-item context (or a static dict)
    concurrency=5,                         # how many run at once
    budget=20,                             # optional cap on items attempted
    on_error="collect",                    # or "fail_fast"
    timeout=120,                           # optional per-item seconds
)

theses = [r["output"] for r in result.succeeded]
errors = result.failed                     # per-item errors
brief  = await result.summarize(llm, "Synthesize the board read.")

Each result dict has this shape:

{"status": "success", "output": {...}, "item": "AAPL", "step_id": "board-scan:0000-aapl"}

The agent's returned value is under output. item is the input that produced this result, and step_id is its unique namespaced identity.

What you can rely on:

• Every result carries its item and a unique step_id. Items that run at the same time cannot mix up each other's results.
• result.results is always in the same order as your items list.
• One failed item does not stop the others. In collect mode (the default) failures land in result.failed with their error messages. In fail_fast mode the first failure cancels the items that have not started yet, and finished results are kept.
• Nothing is dropped silently. Items skipped by budget are listed in result.skipped. When summarize() has to shorten an item's output for the LLM, it marks the cut and includes every failure in the evidence.

---

Accessing redis_store from an Agent

self._redis_store is the RedisContextStore instance injected by the Mesh at setup time. It is None if Redis is not configured. Pass it to register() and execute():

workflow_id = await wf.register(self._redis_store)
results = await wf.execute(mesh, redis_store=self._redis_store)

When self._redis_store is None, the workflow executes in-memory and step statuses are not written to Redis. This is fine for development.

---

Error Handling

When a step fails, execute() records the error and continues with remaining steps that are unblocked:

results = await wf.execute(mesh)
failures = [r for r in results if r["status"] != "success"]
for f in failures:
    logger.error("Step %s failed: %s", f["step_id"], f["error"])

Steps that depend on a failed step are never executed. Check for blocked steps by comparing the number of results to the number of steps in the workflow:

if len(results) < len(wf.steps):
    blocked = [s.step_id for s in wf.steps if s.step_id not in {r["step_id"] for r in results}]
    logger.warning("Blocked steps: %s", blocked)