diff --git a/poethepoet/app.py b/poethepoet/app.py
index df9c36079..055e80502 100644
--- a/poethepoet/app.py
+++ b/poethepoet/app.py
@@ -7,6 +7,7 @@
 from .exceptions import ExecutionError, PoeException
 from .task import PoeTask
 from .task.args import PoeTaskArgs
+from .task.graph import TaskExecutionGraph
 from .ui import PoeUi
 
 
@@ -47,12 +48,16 @@ def __call__(self, cli_args: Sequence[str]) -> int:
             self.print_help()
             return 0
 
-        if not self.resolve_task():
+        if not self.resolve_task(" ".join(cli_args)):
             return 1
 
-        return self.run_task() or 0
+        assert self.task
+        if self.task.has_deps():
+            return self.run_task_graph() or 0
+        else:
+            return self.run_task() or 0
 
-    def resolve_task(self) -> bool:
+    def resolve_task(self, invocation: Sequence[str]) -> bool:
         task = self.ui["task"]
         if not task:
             self.print_help(info="No task specified.")
@@ -71,22 +76,23 @@ def resolve_task(self) -> bool:
             )
             return False
 
-        self.task = PoeTask.from_config(task_name, config=self.config, ui=self.ui)
+        self.task = PoeTask.from_config(
+            task_name, config=self.config, ui=self.ui, invocation=tuple(invocation)
+        )
         return True
 
-    def run_task(self) -> Optional[int]:
+    def run_task(self, context: Optional[RunContext] = None) -> Optional[int]:
         _, *extra_args = self.ui["task"]
+        if context is None:
+            context = RunContext(
+                config=self.config,
+                env=os.environ,
+                dry=self.ui["dry_run"],
+                poe_active=os.environ.get("POE_ACTIVE"),
+            )
         try:
             assert self.task
-            return self.task.run(
-                context=RunContext(
-                    config=self.config,
-                    env=os.environ,
-                    dry=self.ui["dry_run"],
-                    poe_active=os.environ.get("POE_ACTIVE"),
-                ),
-                extra_args=extra_args,
-            )
+            return self.task.run(context=context, extra_args=extra_args)
         except PoeException as error:
             self.print_help(error=error)
             return 1
@@ -94,6 +100,40 @@ def run_task(self) -> Optional[int]:
             self.ui.print_error(error=error)
             return 1
 
+    def run_task_graph(self, context: Optional[RunContext] = None) -> Optional[int]:
+        assert self.task
+        graph = TaskExecutionGraph(self.task, self.config)
+        plan = graph.get_execution_plan()
+
+        context = RunContext(
+            config=self.config,
+            env=os.environ,
+            dry=self.ui["dry_run"],
+            poe_active=os.environ.get("POE_ACTIVE"),
+        )
+
+        for stage in plan:
+            for task in stage:
+                if task == self.task:
+                    # The final sink task gets special treatment
+                    return self.run_task(context)
+
+                try:
+                    task_result = task.run(
+                        context=context, extra_args=task.invocation[1:]
+                    )
+                    if task_result:
+                        raise ExecutionError(
+                            f"Task graph aborted after failed task {task.name!r}"
+                        )
+                except PoeException as error:
+                    self.print_help(error=error)
+                    return 1
+                except ExecutionError as error:
+                    self.ui.print_error(error=error)
+                    return 1
+        return 0
+
     def print_help(
         self,
         info: Optional[str] = None,
diff --git a/poethepoet/context.py b/poethepoet/context.py
index 0cc06e8fc..4077faaec 100644
--- a/poethepoet/context.py
+++ b/poethepoet/context.py
@@ -4,6 +4,7 @@
     Dict,
     MutableMapping,
     Optional,
+    Tuple,
     TYPE_CHECKING,
 )
 from .executor import PoeExecutor
@@ -20,6 +21,7 @@ class RunContext:
     project_dir: Path
     multistage: bool = False
     exec_cache: Dict[str, Any]
+    captured_stdout: Dict[Tuple[str, ...], str]
 
     def __init__(
         self,
@@ -34,19 +36,27 @@ def __init__(
         self.dry = dry
         self.poe_active = poe_active
         self.exec_cache = {}
+        self.captured_stdout = {}
+
+    @property
+    def executor_type(self) -> Optional[str]:
+        return self.config.executor["type"]
 
     def get_env(self, env: MutableMapping[str, str]) -> Dict[str, str]:
         return {**self.env, **env}
 
     def get_executor(
         self,
+        invocation: Tuple[str, ...],
         env: MutableMapping[str, str],
-        task_executor: Optional[Dict[str, str]] = None,
+        task_options: Dict[str, Any],
     ) -> PoeExecutor:
         return PoeExecutor.get(
+            invocation=invocation,
             context=self,
             env=self.get_env(env),
             working_dir=self.project_dir,
             dry=self.dry,
-            executor_config=task_executor,
+            executor_config=task_options.get("executor"),
+            capture_stdout=task_options.get("capture_stdout", False),
         )
diff --git a/poethepoet/exceptions.py b/poethepoet/exceptions.py
index b301b158f..4c66b605a 100644
--- a/poethepoet/exceptions.py
+++ b/poethepoet/exceptions.py
@@ -5,6 +5,10 @@ def __init__(self, msg, *args):
         self.args = (msg, *args)
 
 
+class CyclicDependencyError(PoeException):
+    pass
+
+
 class ExecutionError(RuntimeError):
     def __init__(self, msg, *args):
         self.msg = msg
diff --git a/poethepoet/executor/base.py b/poethepoet/executor/base.py
index b11a345e0..d71101b97 100644
--- a/poethepoet/executor/base.py
+++ b/poethepoet/executor/base.py
@@ -1,7 +1,17 @@
 import signal
 from subprocess import Popen, PIPE
 import sys
-from typing import Any, Dict, MutableMapping, Optional, Sequence, Type, TYPE_CHECKING
+from typing import (
+    Any,
+    Dict,
+    MutableMapping,
+    Optional,
+    Sequence,
+    Tuple,
+    Type,
+    TYPE_CHECKING,
+    Union,
+)
 from ..exceptions import PoeException
 from ..virtualenv import Virtualenv
 
@@ -9,7 +19,7 @@
     from pathlib import Path
     from ..context import RunContext
 
-# TODO: maybe invert the control so the executor is given a task to run
+# TODO: maybe invert the control so the executor is given a task to run?
 
 
 class MetaPoeExecutor(type):
@@ -38,32 +48,38 @@ class PoeExecutor(metaclass=MetaPoeExecutor):
 
     def __init__(
         self,
+        invocation: Tuple[str, ...],
         context: "RunContext",
         options: MutableMapping[str, str],
         env: MutableMapping[str, str],
         working_dir: Optional["Path"] = None,
         dry: bool = False,
+        capture_stdout: Union[str, bool] = False,
     ):
+        self.invocation = invocation
         self.context = context
         self.options = options
         self.working_dir = working_dir
         self.env = env
         self.dry = dry
+        self.capture_stdout = capture_stdout
 
     @classmethod
     def get(
         cls,
+        invocation: Tuple[str, ...],
         context: "RunContext",
         env: MutableMapping[str, str],
         working_dir: Optional["Path"] = None,
         dry: bool = False,
         executor_config: Optional[Dict[str, str]] = None,
+        capture_stdout: Union[str, bool] = False,
     ) -> "PoeExecutor":
         """"""
         # use task specific executor config or fallback to global
         options = executor_config or context.config.executor
         return cls._resolve_implementation(context, executor_config)(
-            context, options, env, working_dir, dry
+            invocation, context, options, env, working_dir, dry, capture_stdout
         )
 
     @classmethod
@@ -75,13 +91,14 @@ def _resolve_implementation(
         by making some reasonable assumptions based on visible features of the
         environment
         """
+        config_executor_type = context.executor_type
         if executor_config:
             if executor_config["type"] not in cls.__executor_types:
                 raise PoeException(
                     f"Cannot instantiate unknown executor {executor_config['type']!r}"
                 )
             return cls.__executor_types[executor_config["type"]]
-        elif context.config.executor["type"] == "auto":
+        elif config_executor_type == "auto":
             if "poetry" in context.config.project["tool"]:
                 # Looks like this is a poetry project!
                 return cls.__executor_types["poetry"]
@@ -92,12 +109,11 @@ def _resolve_implementation(
             # Fallback to not using any particular environment
             return cls.__executor_types["simple"]
         else:
-            if context.config.executor["type"] not in cls.__executor_types:
+            if config_executor_type not in cls.__executor_types:
                 raise PoeException(
-                    f"Cannot instantiate unknown executor"
-                    + repr(context.config.executor["type"])
+                    f"Cannot instantiate unknown executor" + repr(config_executor_type)
                 )
-            return cls.__executor_types[context.config.executor["type"]]
+            return cls.__executor_types[config_executor_type]
 
     def execute(self, cmd: Sequence[str], input: Optional[bytes] = None,) -> int:
         raise NotImplementedError
@@ -116,6 +132,11 @@ def _exec_via_subproc(
         popen_kwargs["env"] = self.env if env is None else env
         if input is not None:
             popen_kwargs["stdin"] = PIPE
+        if self.capture_stdout:
+            if isinstance(self.capture_stdout, str):
+                popen_kwargs["stdout"] = open(self.capture_stdout, "wb")
+            else:
+                popen_kwargs["stdout"] = PIPE
         if self.working_dir is not None:
             popen_kwargs["cwd"] = self.working_dir
 
@@ -131,7 +152,10 @@ def handle_signal(signum, _frame):
         old_signal_handler = signal.signal(signal.SIGINT, handle_signal)
 
         # send data to the subprocess and wait for it to finish
-        proc.communicate(input)
+        (captured_stdout, _) = proc.communicate(input)
+
+        if self.capture_stdout == True:
+            self.context.captured_stdout[self.invocation] = captured_stdout.decode()
 
         # restore signal handler
         signal.signal(signal.SIGINT, old_signal_handler)
diff --git a/poethepoet/helpers.py b/poethepoet/helpers.py
new file mode 100644
index 000000000..a10ffb7e6
--- /dev/null
+++ b/poethepoet/helpers.py
@@ -0,0 +1,5 @@
+import re
+
+
+def is_valid_env_var(var_name: str) -> bool:
+    return bool(re.match("[a-zA-Z_][a-zA-Z0-9_]*", var_name))
diff --git a/poethepoet/task/base.py b/poethepoet/task/base.py
index d8403c988..ae55dafe5 100644
--- a/poethepoet/task/base.py
+++ b/poethepoet/task/base.py
@@ -1,12 +1,15 @@
 import re
+import shlex
 import sys
 from typing import (
     Any,
     Dict,
+    Iterator,
     List,
     MutableMapping,
     Optional,
     Sequence,
+    Set,
     Tuple,
     Type,
     TYPE_CHECKING,
@@ -14,6 +17,7 @@
 )
 from .args import PoeTaskArgs
 from ..exceptions import PoeException
+from ..helpers import is_valid_env_var
 
 if TYPE_CHECKING:
     from ..context import RunContext
@@ -63,9 +67,12 @@ class PoeTask(metaclass=MetaPoeTask):
     __content_type__: Type = str
     __base_options: Dict[str, Union[Type, Tuple[Type, ...]]] = {
         "args": (dict, list),
+        "capture_stdout": (str),
+        "deps": list,
         "env": dict,
         "executor": dict,
         "help": str,
+        "uses": dict,
     }
     __task_types: Dict[str, Type["PoeTask"]] = {}
 
@@ -76,20 +83,40 @@ def __init__(
         options: Dict[str, Any],
         ui: "PoeUi",
         config: "PoeConfig",
+        invocation: Tuple[str, ...],
+        capture_stdout: bool = False,
     ):
         self.name = name
         self.content = content.strip() if isinstance(content, str) else content
-        self.options = options
+        if capture_stdout:
+            self.options = dict(options, capture_stdout=True)
+        else:
+            self.options = options
         self._ui = ui
         self._config = config
         self._is_windows = sys.platform == "win32"
+        self.invocation = invocation
 
     @classmethod
-    def from_config(cls, task_name: str, config: "PoeConfig", ui: "PoeUi") -> "PoeTask":
+    def from_config(
+        cls,
+        task_name: str,
+        config: "PoeConfig",
+        ui: "PoeUi",
+        invocation: Tuple[str, ...],
+        capture_stdout: Optional[bool] = None,
+    ) -> "PoeTask":
         task_def = config.tasks.get(task_name)
         if not task_def:
             raise PoeException(f"Cannot instantiate unknown task {task_name!r}")
-        return cls.from_def(task_def, task_name, config, ui)
+        return cls.from_def(
+            task_def,
+            task_name,
+            config,
+            ui,
+            invocation=invocation,
+            capture_stdout=capture_stdout,
+        )
 
     @classmethod
     def from_def(
@@ -98,53 +125,90 @@ def from_def(
         task_name: str,
         config: "PoeConfig",
         ui: "PoeUi",
+        invocation: Tuple[str, ...],
         array_item: Union[bool, str] = False,
+        capture_stdout: Optional[bool] = None,
     ) -> "PoeTask":
-        if array_item:
-            if isinstance(task_def, str):
-                task_type = (
+        task_type = cls.resolve_task_type(task_def, config, array_item)
+        if task_type is None:
+            # Something is wrong with this task_def
+            raise cls.Error(cls.validate_def(task_name, task_def, config))
+
+        options: Dict[str, Any] = {}
+        if capture_stdout is not None:
+            # Override config because we want to specifically capture the stdout of this
+            # task for internal use
+            options["capture_stdout"] = capture_stdout
+
+        if isinstance(task_def, (str, list)):
+            return cls.__task_types[task_type](
+                name=task_name,
+                content=task_def,
+                options=options,
+                ui=ui,
+                config=config,
+                invocation=invocation,
+            )
+
+        assert isinstance(task_def, dict)
+        options = dict(task_def, **options)
+        content = options.pop(task_type)
+        return cls.__task_types[task_type](
+            name=task_name,
+            content=content,
+            options=options,
+            ui=ui,
+            config=config,
+            invocation=invocation,
+        )
+
+    @classmethod
+    def resolve_task_type(
+        cls,
+        task_def: TaskDef,
+        config: "PoeConfig",
+        array_item: Union[bool, str] = False,
+    ) -> Optional[str]:
+        if isinstance(task_def, str):
+            if array_item:
+                return (
                     array_item
                     if isinstance(array_item, str)
                     else config.default_array_item_task_type
                 )
-                return cls.__task_types[task_type](
-                    name=task_name, content=task_def, options={}, ui=ui, config=config
-                )
-        else:
-            if isinstance(task_def, str):
-                return cls.__task_types[config.default_task_type](
-                    name=task_name, content=task_def, options={}, ui=ui, config=config
-                )
-        if isinstance(task_def, list):
-            return cls.__task_types[config.default_array_task_type](
-                name=task_name, content=task_def, options={}, ui=ui, config=config
-            )
+            else:
+                return config.default_task_type
 
-        assert isinstance(task_def, dict)
-        task_type_keys = set(task_def.keys()).intersection(cls.__task_types)
-        if len(task_type_keys) == 1:
-            task_type_key = next(iter(task_type_keys))
-            options = dict(task_def)
-            content = options.pop(task_type_key)
-            return cls.__task_types[task_type_key](
-                name=task_name, content=content, options=options, ui=ui, config=config
-            )
+        elif isinstance(task_def, list):
+            return config.default_array_task_type
 
-        # Something is wrong with this task_def
-        raise cls.Error(cls.validate_def(task_name, task_def, config))
+        elif isinstance(task_def, dict):
+            task_type_keys = set(task_def.keys()).intersection(cls.__task_types)
+            if len(task_type_keys) == 1:
+                return next(iter(task_type_keys))
+
+        return None
 
     def run(
         self,
         context: "RunContext",
-        extra_args: Sequence[str],
+        extra_args: Sequence[str] = tuple(),
         env: Optional[MutableMapping[str, str]] = None,
     ) -> int:
         """
         Run this task
         """
+
+        # Get env vars from glboal options
         env = dict(env or {}, **self._config.global_env)
+
+        # Get env vars from task options
         if self.options.get("env"):
-            env = dict(env, **self.options["env"])
+            env.update(self.options["env"])
+
+        # Get env vars from dependencies
+        env.update(self.get_dep_values(context))
+
         return self._handle_run(context, extra_args, env)
 
     def parse_named_args(self, extra_args: Sequence[str]) -> Optional[Dict[str, str]]:
@@ -160,10 +224,50 @@ def _handle_run(
         env: MutableMapping[str, str],
     ) -> int:
         """
-        _handle_run must be implemented by a subclass and return a single executor result.
+        _handle_run must be implemented by a subclass and return a single executor
+        result.
         """
         raise NotImplementedError
 
+    def iter_upstream_tasks(self) -> Iterator[Tuple[str, "PoeTask"]]:
+        for task_ref in self.options.get("deps", tuple()):
+            yield ("", self._instantiate_dep(task_ref, capture_stdout=False))
+        for key, task_ref in self.options.get("uses", {}).items():
+            yield (key, self._instantiate_dep(task_ref, capture_stdout=True))
+
+    def get_upstream_invocations(self) -> Set[Tuple[str, ...]]:
+        """
+        Get identifiers (i.e. invocation tuples) for all upstream tasks
+        """
+        result = set()
+        for task_ref in self.options.get("deps", {}):
+            result.add(tuple(shlex.split(task_ref)))
+        for task_ref in self.options.get("uses", {}).values():
+            result.add(tuple(shlex.split(task_ref)))
+        return result
+
+    def get_dep_values(self, context: "RunContext") -> Dict[str, str]:
+        """
+        Get env vars from upstream tasks declared via the uses option
+        """
+        return {
+            var: context.captured_stdout[tuple(shlex.split(dep))]
+            for var, dep in self.options.get("uses", {}).items()
+        }
+
+    def has_deps(self) -> bool:
+        return bool(self.options.get("deps", False) or self.options.get("uses", False))
+
+    def _instantiate_dep(self, task_ref: str, capture_stdout: bool) -> "PoeTask":
+        invocation = tuple(shlex.split(task_ref))
+        return self.from_config(
+            invocation[0],
+            config=self._config,
+            ui=self._ui,
+            invocation=invocation,
+            capture_stdout=capture_stdout,
+        )
+
     @staticmethod
     def _resolve_envvars(
         content: str, context: "RunContext", env: MutableMapping[str, str]
@@ -209,7 +313,6 @@ def validate_def(
         """
         Check the given task name and definition for validity and return a message
         describing the first encountered issue if any.
-        If raize is True then the issue is raised as an exception.
         """
         if not (task_name[0].isalpha() or task_name[0] == "_"):
             return (
@@ -259,14 +362,39 @@ def validate_def(
                         if task_type_issue:
                             return task_type_issue
 
+            if "args" in task_def:
+                return PoeTaskArgs.validate_def(task_name, task_def["args"])
+
             if "\n" in task_def.get("help", ""):
                 return (
                     f"Invalid task: {task_name!r}. Help messages cannot contain "
                     "line breaks"
                 )
 
-            if "args" in task_def:
-                return PoeTaskArgs.validate_def(task_name, task_def["args"])
+            all_task_names = set(config.tasks)
+
+            if "deps" in task_def:
+                for dep in task_def["deps"]:
+                    dep_task_name = dep.split(" ", 1)[0]
+                    if dep_task_name not in all_task_names:
+                        return (
+                            f"Invalid task: {task_name!r}. deps options contains "
+                            f"reference to unknown task: {dep_task_name!r}"
+                        )
+
+            if "uses" in task_def:
+                for key, dep in task_def["uses"].items():
+                    if not is_valid_env_var(key):
+                        return (
+                            f"Invalid task: {task_name!r} uses options contains invalid"
+                            f" key: {key!r}"
+                        )
+                    dep_task_name = dep.split(" ", 1)[0]
+                    if dep_task_name not in all_task_names:
+                        return (
+                            f"Invalid task: {task_name!r}. uses options contains "
+                            f"reference to unknown task: {dep_task_name!r}"
+                        )
 
         return None
 
@@ -275,7 +403,7 @@ def is_task_type(
         cls, task_def_key: str, content_type: Optional[Type] = None
     ) -> bool:
         """
-        Checks whether the given key identified a known task type.
+        Checks whether the given key identifies a known task type.
         Optionally also check whether the given content_type matches the type of content
         for this tasks type.
         """
diff --git a/poethepoet/task/cmd.py b/poethepoet/task/cmd.py
index 62d1f1fc1..a41256f90 100644
--- a/poethepoet/task/cmd.py
+++ b/poethepoet/task/cmd.py
@@ -42,7 +42,7 @@ def _handle_run(
         else:
             cmd = (*self._resolve_args(context, env), *extra_args)
         self._print_action(" ".join(cmd), context.dry)
-        return context.get_executor(env, self.options.get("executor")).execute(cmd)
+        return context.get_executor(self.invocation, env, self.options).execute(cmd)
 
     def _add_named_args_to_env(
         self, extra_args: Sequence[str], env: MutableMapping[str, str]
diff --git a/poethepoet/task/graph.py b/poethepoet/task/graph.py
new file mode 100644
index 000000000..463ffd889
--- /dev/null
+++ b/poethepoet/task/graph.py
@@ -0,0 +1,132 @@
+from typing import Dict, Set, List, Tuple
+from ..config import PoeConfig
+from ..exceptions import CyclicDependencyError
+from .base import PoeTask
+
+
+class TaskExecutionNode:
+    task: PoeTask
+    direct_dependants: List["TaskExecutionNode"]
+    direct_dependencies: Set[Tuple[str, ...]]
+    path_dependants: Tuple[str, ...]
+    capture_stdout: bool
+
+    def __init__(
+        self,
+        task: PoeTask,
+        direct_dependants: List["TaskExecutionNode"],
+        path_dependants: Tuple[str, ...],
+        capture_stdout: bool = False,
+    ):
+        self.task = task
+        self.direct_dependants = direct_dependants
+        self.direct_dependencies = task.get_upstream_invocations()
+        self.path_dependants = (task.name, *path_dependants)
+        self.capture_stdout = capture_stdout
+
+    def is_source(self):
+        return not self.task.has_deps()
+
+    @property
+    def identifier(self) -> Tuple[str, ...]:
+        return self.task.invocation
+
+
+class TaskExecutionGraph:
+    """
+    A directed-acyclic execution graph of tasks, with a single sink node, and any number
+    of source nodes. Non-source nodes may have multiple upstream nodes, and non-sink
+    nodes may have multiple downstream nodes.
+
+    A task/node may appear twice in the graph, if one instance has captured output, and
+    one does not. Nodes are deduplicated to enforce this.
+    """
+
+    config: PoeConfig
+    sources: List[TaskExecutionNode]
+    sink: TaskExecutionNode
+    captured_tasks: Dict[str, TaskExecutionNode]
+    uncaptured_tasks: Dict[str, TaskExecutionNode]
+
+    def __init__(
+        self, sink_task: PoeTask, config: PoeConfig,
+    ):
+        self.config = config
+        self.sink = TaskExecutionNode(sink_task, [], tuple())
+        self.sources = []
+        self.captured_tasks = {}
+        self.uncaptured_tasks = {}
+
+        # Build graph
+        self._resolve_node_deps(self.sink)
+
+    def get_execution_plan(self) -> List[List[PoeTask]]:
+        """
+        Derive an execution plan from the DAG in terms of stages consisting of tasks
+        that could theoretically be parallelized.
+        """
+        # TODO: if we parallelize tasks then this should be modified to support lazy
+        #       scheduling
+
+        stages = [self.sources]
+        visited = set(source.identifier for source in self.sources)
+
+        while True:
+            next_stage = []
+            for node in stages[-1]:
+                for dep_node in node.direct_dependants:
+                    if not dep_node.direct_dependencies.issubset(visited):
+                        # Some dependencies of dep_node have not been visited, so skip
+                        # them for now
+                        continue
+                    next_stage.append(dep_node)
+            if next_stage:
+                stages.append(next_stage)
+                visited.update(node.identifier for node in next_stage)
+            else:
+                break
+
+        return [[node.task for node in stage] for stage in stages]
+
+    def _resolve_node_deps(self, node: TaskExecutionNode):
+        """
+        Build a DAG of tasks by depth-first traversal of the dependency tree starting
+        from the sink node.
+        """
+        for key, task in node.task.iter_upstream_tasks():
+            if task.name in node.path_dependants:
+                raise CyclicDependencyError(
+                    f"Encountered cyclic task dependency at {task.name}"
+                )
+
+            # a non empty key indicates output is captured
+            capture_stdout = bool(key)
+
+            # Check if a node already exists for this task
+            if capture_stdout:
+                if task.name in self.captured_tasks:
+                    # reuse instance of task with captured output
+                    self.captured_tasks[task.name].direct_dependants.append(node)
+                    continue
+            elif task.name in self.uncaptured_tasks:
+                # reuse instance of task with uncaptured output
+                self.uncaptured_tasks[task.name].direct_dependants.append(node)
+                continue
+
+            # This task has not been encountered before via another path
+            new_node = TaskExecutionNode(
+                task, [node], node.path_dependants, capture_stdout
+            )
+
+            # Keep track of this task/node so it can be found by other dependants
+            if capture_stdout:
+                self.captured_tasks[task.name] = new_node
+            else:
+                self.uncaptured_tasks[task.name] = new_node
+
+            if new_node.is_source():
+                # Track this node as having no dependencies
+                self.sources.append(new_node)
+            else:
+                # Recurse immediately for DFS
+                self._resolve_node_deps(new_node)
diff --git a/poethepoet/task/ref.py b/poethepoet/task/ref.py
index 5fcc9c992..e171e3b6e 100644
--- a/poethepoet/task/ref.py
+++ b/poethepoet/task/ref.py
@@ -35,7 +35,9 @@ def _handle_run(
         """
         Lookup and delegate to the referenced task
         """
-        task = self.from_config(self.content, self._config, ui=self._ui)
+        task = self.from_config(
+            self.content, self._config, ui=self._ui, invocation=(self.content,)
+        )
         return task.run(context=context, extra_args=extra_args, env=env,)
 
     @classmethod
diff --git a/poethepoet/task/script.py b/poethepoet/task/script.py
index 8f094569f..9ba786bdc 100644
--- a/poethepoet/task/script.py
+++ b/poethepoet/task/script.py
@@ -59,7 +59,7 @@ def _handle_run(
             f"import_module('{target_module}').{target_callable}({call_params or ''})",
         )
         self._print_action(" ".join(argv), context.dry)
-        return context.get_executor(env, self.options.get("executor")).execute(cmd)
+        return context.get_executor(self.invocation, env, self.options).execute(cmd)
 
     @classmethod
     def _parse_content(
diff --git a/poethepoet/task/sequence.py b/poethepoet/task/sequence.py
index 1de03f963..8a1335c3f 100644
--- a/poethepoet/task/sequence.py
+++ b/poethepoet/task/sequence.py
@@ -5,6 +5,7 @@
     MutableMapping,
     Optional,
     Sequence,
+    Tuple,
     Type,
     TYPE_CHECKING,
     Union,
@@ -37,17 +38,23 @@ def __init__(
         options: Dict[str, Any],
         ui: "PoeUi",
         config: "PoeConfig",
+        invocation: Tuple[str, ...],
+        capture_stdout: bool = False,
     ):
-        super().__init__(name, content, options, ui, config)
+        assert capture_stdout == False
+        super().__init__(name, content, options, ui, config, invocation)
+
         self.subtasks = [
             self.from_def(
                 task_def=item,
-                task_name=item if isinstance(item, str) else f"{name}[{index}]",
+                task_name=task_name,
                 config=config,
+                invocation=(task_name,),
                 ui=ui,
                 array_item=self.options.get("default_item_type", True),
             )
             for index, item in enumerate(self.content)
+            for task_name in (item if isinstance(item, str) else f"{name}[{index}]",)
         ]
 
     def _handle_run(
diff --git a/poethepoet/task/shell.py b/poethepoet/task/shell.py
index ce03581d4..1f110ec40 100644
--- a/poethepoet/task/shell.py
+++ b/poethepoet/task/shell.py
@@ -38,7 +38,7 @@ def _handle_run(
             shell = [os.environ.get("SHELL", shutil.which("bash") or "/bin/bash")]
 
         self._print_action(self.content, context.dry)
-        return context.get_executor(env, self.options.get("executor")).execute(
+        return context.get_executor(self.invocation, env, self.options).execute(
             shell, input=self.content.encode()
         )
 
diff --git a/pyproject.toml b/pyproject.toml
index bb44df3f8..5617f88f1 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -51,6 +51,10 @@ check = ["check-docs", "style", "types", "lint", "test"]
 # poeception
 poe = { script = "poethepoet:main", help = "Execute poe programmatically as a poe task" }
 
+_task_a = { cmd = "echo world", deps = { _ = ["style"]}}
+dag = { cmd = "echo hello $SUBJECT!", deps = { SUBJECT = "_task_a", _ = ["style", "check-docs"]}}
+
+
 [tool.tox]
 legacy_tox_ini = """
 [tox]