feat(workflow): support directed acyclic graph (DAG) task execution via run_dag

Enables workflow scripts to execute heterogeneous sub-agent tasks as a
directed acyclic graph (DAG) with dependency barriers and concurrency
throttling:

- Validates graph topology and detects cycles pre-flight using Kahn's algorithm
- Concurrently dispatches unblocked ready tasks across the shared semaphore
- Resolves upstream node outputs into downstream prompt templates ({parent_node})
  or callable prompt handlers
- Automatically prunes downstream dependent tasks when an upstream task fails,
  preventing wasteful LLM calls on invalidated branches
- Adds comprehensive unit test coverage (linear, diamond concurrency,
  cycle detection, callable prompts, fail-fast pruning, and script integration)

Author: connerkup

openhands-tools/openhands/tools/workflow/definition.py

@@ -85,6 +85,13 @@ async def main(wf):
   the previous result. Stages may be sync or async. A stage that raises drops that
   item to `None`. Prefer this over chained `map_agents` calls when per-item stages
   are independent, since `map_agents` fully drains each stage before the next.
+- `await wf.run_dag(nodes, max_concurrency=None)` — run a directed acyclic
+  graph (DAG) of sub-agent tasks with dependency resolution and concurrency
+  control. `nodes` maps node IDs to specs (`prompt`, `depends_on`,
+  `subagent_type`, `description`). Tasks execute as soon as their upstream
+  dependencies finish. Upstream results are interpolated into `{parent_node_id}`
+  prompt placeholders. If an upstream task fails, downstream dependent tasks
+  are pruned automatically.
 - `wf.flatten(values)` — flatten one level of nesting (not recursive)
 
 `subagent_type` must be a sub-agent type registered in the parent application.

openhands-tools/openhands/tools/workflow/impl.py

@@ -241,6 +241,147 @@ async def reduce_agent(
             description=description,
         )
 
+    async def run_dag(
+        self,
+        nodes: dict[str, dict[str, Any]],
+        *,
+        max_concurrency: int | None = None,
+    ) -> dict[str, str]:
+        """Execute a directed acyclic graph (DAG) of sub-agent tasks.
+
+        Tasks are dispatched concurrently as soon as their upstream dependencies
+        complete. If an upstream dependency fails, downstream dependent tasks
+        are pruned without executing to avoid wasting agent calls.
+
+        Args:
+            nodes: Mapping of node ID to specification dict:
+                - ``prompt``: Required prompt string or callable
+                  ``(results) -> str``. When a string, occurrences of
+                  ``{parent_id}`` are substituted with the parent node's result.
+                - ``depends_on``: Optional sequence of parent node IDs
+                  (default: ``[]``).
+                - ``subagent_type``: Optional subagent type
+                  (default: ``"general-purpose"``).
+                - ``description``: Optional human-readable description.
+            max_concurrency: Optional per-call concurrency limit (capped at
+                context limit).
+
+        Returns:
+            Mapping of node ID to string result for completed nodes.
+
+        Raises:
+            ValueError: If ``nodes`` is empty, contains unknown dependencies,
+                contains self-dependencies, or contains cycles.
+            ExceptionGroup: If one or more tasks fail during execution.
+        """
+        if not nodes:
+            raise ValueError("run_dag requires at least one node")
+        if max_concurrency is not None and max_concurrency < 1:
+            raise ValueError("max_concurrency must be at least 1")
+
+        # 1. Structural validation
+        for node_id, spec in nodes.items():
+            if not isinstance(spec, dict):
+                raise ValueError(f"Node '{node_id}' specification must be a dict")
+            deps = spec.get("depends_on", [])
+            for dep in deps:
+                if dep == node_id:
+                    raise ValueError(f"Node '{node_id}' cannot depend on itself")
+                if dep not in nodes:
+                    raise ValueError(
+                        f"Node '{node_id}' depends on unknown node '{dep}'"
+                    )
+
+        # 2. Cycle detection via Kahn's topological sort
+        in_degree = {k: len(nodes[k].get("depends_on", [])) for k in nodes}
+        adj: dict[str, list[str]] = {k: [] for k in nodes}
+        for k, spec in nodes.items():
+            for dep in spec.get("depends_on", []):
+                adj[dep].append(k)
+
+        queue = [k for k, deg in in_degree.items() if deg == 0]
+        visited_count = 0
+        while queue:
+            curr = queue.pop(0)
+            visited_count += 1
+            for neighbor in adj[curr]:
+                in_degree[neighbor] -= 1
+                if in_degree[neighbor] == 0:
+                    queue.append(neighbor)
+
+        if visited_count < len(nodes):
+            unresolved = [k for k, deg in in_degree.items() if deg > 0]
+            raise ValueError(
+                f"DAG contains cycles involving: {', '.join(sorted(unresolved))}"
+            )
+
+        # 3. Frontier execution
+        semaphore = (
+            asyncio.Semaphore(min(max_concurrency, self._max_concurrency))
+            if max_concurrency is not None
+            else self._default_semaphore
+        )
+
+        results: dict[str, str] = {}
+        errors: dict[str, Exception] = {}
+        failed_or_skipped: set[str] = set()
+        events: dict[str, asyncio.Event] = {k: asyncio.Event() for k in nodes}
+
+        async def run_node(node_id: str, spec: dict[str, Any]) -> None:
+            deps = spec.get("depends_on", [])
+            # Wait for all upstream dependencies
+            for dep in deps:
+                await events[dep].wait()
+
+            # Prune if any upstream dependency failed or was skipped
+            failed_deps = [dep for dep in deps if dep in failed_or_skipped]
+            if failed_deps:
+                failed_or_skipped.add(node_id)
+                errors[node_id] = RuntimeError(
+                    f"Node '{node_id}' skipped because dependency "
+                    f"'{failed_deps[0]}' failed"
+                )
+                events[node_id].set()
+                return
+
+            # Render prompt
+            raw_prompt = spec.get("prompt", "")
+            subagent_type = spec.get("subagent_type", "general-purpose")
+            description = spec.get("description")
+
+            if callable(raw_prompt):
+                rendered_prompt = str(raw_prompt(results))
+            elif isinstance(raw_prompt, str):
+                rendered_prompt = raw_prompt
+                for dep in deps:
+                    val = str(results.get(dep, ""))
+                    rendered_prompt = rendered_prompt.replace(f"{{{dep}}}", val)
+            else:
+                rendered_prompt = str(raw_prompt)
+
+            async with semaphore:
+                try:
+                    res = await self._run_agent_task(
+                        prompt=rendered_prompt,
+                        subagent_type=subagent_type,
+                        description=description,
+                    )
+                    results[node_id] = res
+                except Exception as exc:
+                    failed_or_skipped.add(node_id)
+                    errors[node_id] = exc
+                finally:
+                    events[node_id].set()
+
+        await asyncio.gather(*(run_node(k, v) for k, v in nodes.items()))
+
+        if errors:
+            raise ExceptionGroup(
+                "run_dag: one or more nodes failed", list(errors.values())
+            )
+
+        return results
+
     def flatten(self, values: list[Any]) -> list[Any]:
         """Flatten one list level."""
         flattened: list[Any] = []

tests/tools/workflow/test_workflow_tool.py

@@ -528,3 +528,173 @@ def test_format_value_long_string_char_truncated() -> None:
     out = _format_value("q" * (_MAX_REDUCE_INPUT_CHARS + 100))
     assert out.endswith("[truncated workflow intermediate results]")
     assert len(out) <= _MAX_REDUCE_INPUT_CHARS + 60
+
+
+def test_run_dag_basic_linear_dependency() -> None:
+    manager = _FakeTaskManager()
+    ctx = _context(manager)
+    nodes = {
+        "step1": {
+            "prompt": "first step",
+            "subagent_type": "analyst",
+        },
+        "step2": {
+            "prompt": "second step using {step1}",
+            "depends_on": ["step1"],
+            "subagent_type": "writer",
+        },
+    }
+    results = asyncio.run(ctx.run_dag(nodes))
+    assert results["step1"] == "result:first step"
+    assert results["step2"] == "result:second step using result:first step"
+    assert manager.prompts == [
+        "analyst: first step",
+        "writer: second step using result:first step",
+    ]
+
+
+def test_run_dag_diamond_concurrency() -> None:
+    manager = _FakeTaskManager()
+    ctx = _context(manager)
+    nodes = {
+        "root": {"prompt": "init"},
+        "branch_a": {"prompt": "do A after {root}", "depends_on": ["root"]},
+        "branch_b": {"prompt": "do B after {root}", "depends_on": ["root"]},
+        "join": {
+            "prompt": "merge {branch_a} and {branch_b}",
+            "depends_on": ["branch_a", "branch_b"],
+        },
+    }
+    results = asyncio.run(ctx.run_dag(nodes))
+    assert results["root"] == "result:init"
+    assert results["branch_a"] == "result:do A after result:init"
+    assert results["branch_b"] == "result:do B after result:init"
+    assert results["join"] == (
+        "result:merge result:do A after result:init and "
+        "result:do B after result:init"
+    )
+    # Root must execute first, join must execute last
+    assert manager.prompts[0] == "general-purpose: init"
+    assert set(manager.prompts[1:3]) == {
+        "general-purpose: do A after result:init",
+        "general-purpose: do B after result:init",
+    }
+    assert manager.prompts[3] == (
+        "general-purpose: merge result:do A after result:init and "
+        "result:do B after result:init"
+    )
+
+
+def test_run_dag_callable_prompt() -> None:
+    manager = _FakeTaskManager()
+    ctx = _context(manager)
+    nodes = {
+        "prep": {"prompt": "setup data"},
+        "process": {
+            "prompt": lambda res: f"custom {res['prep'].upper()}",
+            "depends_on": ["prep"],
+        },
+    }
+    results = asyncio.run(ctx.run_dag(nodes))
+    assert results["process"] == "result:custom RESULT:SETUP DATA"
+
+
+def test_run_dag_cycle_detection() -> None:
+    ctx = _context(_FakeTaskManager())
+    nodes = {
+        "a": {"prompt": "a", "depends_on": ["b"]},
+        "b": {"prompt": "b", "depends_on": ["a"]},
+    }
+    with pytest.raises(ValueError, match="DAG contains cycles"):
+        asyncio.run(ctx.run_dag(nodes))
+
+
+def test_run_dag_self_dependency_raises() -> None:
+    ctx = _context(_FakeTaskManager())
+    nodes = {
+        "a": {"prompt": "a", "depends_on": ["a"]},
+    }
+    with pytest.raises(ValueError, match="cannot depend on itself"):
+        asyncio.run(ctx.run_dag(nodes))
+
+
+def test_run_dag_unknown_dependency_raises() -> None:
+    ctx = _context(_FakeTaskManager())
+    nodes = {
+        "a": {"prompt": "a", "depends_on": ["nonexistent"]},
+    }
+    with pytest.raises(ValueError, match="depends on unknown node"):
+        asyncio.run(ctx.run_dag(nodes))
+
+
+def test_run_dag_empty_nodes_raises() -> None:
+    ctx = _context(_FakeTaskManager())
+    with pytest.raises(ValueError, match="requires at least one node"):
+        asyncio.run(ctx.run_dag({}))
+
+
+def test_run_dag_invalid_spec_raises() -> None:
+    ctx = _context(_FakeTaskManager())
+    with pytest.raises(ValueError, match="specification must be a dict"):
+        asyncio.run(ctx.run_dag({"bad": "not-a-dict"}))  # type: ignore[arg-type]
+
+
+def test_run_dag_prunes_downstream_on_failure() -> None:
+    class FailingTaskManager(_FakeTaskManager):
+        def start_task(
+            self,
+            prompt: str,
+            subagent_type: str = "default",
+            resume: str | None = None,
+            description: str | None = None,
+            conversation: LocalConversation | None = None,
+        ) -> _FakeTask:
+            self.prompts.append(f"{subagent_type}: {prompt}")
+            if "fail" in prompt:
+                return _FakeTask(error="upstream task failed")
+            return _FakeTask(result=f"result:{prompt}")
+
+    manager = FailingTaskManager()
+    ctx = _context(manager)
+    nodes = {
+        "good": {"prompt": "run independent good"},
+        "failing": {"prompt": "run failing task"},
+        "child_of_failing": {
+            "prompt": "should be skipped {failing}",
+            "depends_on": ["failing"],
+        },
+    }
+
+    with pytest.raises(ExceptionGroup) as exc_info:
+        asyncio.run(ctx.run_dag(nodes))
+
+    assert "run_dag" in str(exc_info.value)
+    # The failing node raised, and child_of_failing was pruned
+    error_messages = [str(e) for e in exc_info.value.exceptions]
+    assert any("upstream task failed" in msg for msg in error_messages)
+    assert any(
+        "skipped because dependency 'failing' failed" in msg
+        for msg in error_messages
+    )
+    # child_of_failing was never dispatched to manager
+    dispatched_prompts = set(manager.prompts)
+    assert "general-purpose: run independent good" in dispatched_prompts
+    assert "general-purpose: run failing task" in dispatched_prompts
+    assert not any("should be skipped" in p for p in dispatched_prompts)
+
+
+def test_run_dag_reachable_from_script() -> None:
+    manager = _FakeTaskManager()
+    ctx = _context(manager)
+    script = """
+async def main(wf):
+    nodes = {
+        "spec": {"prompt": "design API"},
+        "impl": {"prompt": "build from {spec}", "depends_on": ["spec"]},
+    }
+    results = await wf.run_dag(nodes)
+    return results["impl"]
+"""
+    result = execute_workflow_script(script, ctx)
+    assert result == "result:build from result:design API"
+