diff --git a/dotflow/cli/commands/__init__.py b/dotflow/cli/commands/__init__.py
index 0f66e5e1..c04b3a48 100644
--- a/dotflow/cli/commands/__init__.py
+++ b/dotflow/cli/commands/__init__.py
@@ -2,6 +2,7 @@
 
 from dotflow.cli.commands.cloud import CloudGenerateCommand, CloudListCommand
 from dotflow.cli.commands.deploy import DeployCommand
+from dotflow.cli.commands.flow import FlowCommand
 from dotflow.cli.commands.init import InitCommand
 from dotflow.cli.commands.log import LogCommand
 from dotflow.cli.commands.schedule import ScheduleCommand
@@ -11,6 +12,7 @@
     "CloudGenerateCommand",
     "CloudListCommand",
     "DeployCommand",
+    "FlowCommand",
     "InitCommand",
     "LogCommand",
     "ScheduleCommand",
diff --git a/dotflow/cli/commands/flow.py b/dotflow/cli/commands/flow.py
new file mode 100644
index 00000000..f0e288d0
--- /dev/null
+++ b/dotflow/cli/commands/flow.py
@@ -0,0 +1,35 @@
+"""Command flow module"""
+
+from dotflow.cli.command import Command
+from dotflow.core.module import Module
+from dotflow.utils.visualizer import visualize
+
+
+class FlowCommand(Command):
+    def setup(self):
+        step = self.params.step
+        mode = self.params.mode
+        fmt = self.params.format
+
+        loaded = Module(value=step)
+        tasks = self._extract_tasks(loaded)
+        visualize(tasks=tasks, mode=mode, fmt=fmt)
+
+    @staticmethod
+    def _extract_tasks(obj) -> list:
+        """
+        Accept any of the three common ways a user might point us at tasks:
+          1. A DotFlow instance          → obj.task.queue
+          2. A TaskBuilder instance      → obj.queue
+          3. A plain list[Task]          → obj
+        """
+        if hasattr(obj, "task") and hasattr(obj.task, "queue"):
+            return list(obj.task.queue)
+
+        if hasattr(obj, "queue"):
+            return list(obj.queue)
+
+        if isinstance(obj, list):
+            return obj
+
+        return []
diff --git a/dotflow/cli/setup.py b/dotflow/cli/setup.py
index 0ea48fe8..356bda8f 100644
--- a/dotflow/cli/setup.py
+++ b/dotflow/cli/setup.py
@@ -7,6 +7,7 @@
     CloudGenerateCommand,
     CloudListCommand,
     DeployCommand,
+    FlowCommand,
     InitCommand,
     LogCommand,
     ScheduleCommand,
@@ -44,6 +45,7 @@ def __init__(self, parser):
         self.setup_schedule()
         self.setup_cloud()
         self.setup_deploy()
+        self.setup_flow()
         self.command()
 
     def setup_init(self):
@@ -210,6 +212,42 @@ def setup_deploy(self):
         )
         self.cmd_deploy.set_defaults(exec=DeployCommand)
 
+    def setup_flow(self):
+        self.cmd_flow = self.subparsers.add_parser(
+            "flow",
+            help="Visualize a workflow pipeline in the terminal",
+        )
+        self.cmd_flow = self.cmd_flow.add_argument_group(
+            "Usage: dotflow flow [OPTIONS]"
+        )
+
+        self.cmd_flow.add_argument(
+            "-s",
+            "--step",
+            required=True,
+            help="Dotted path to a DotFlow instance, TaskBuilder, or task list",
+        )
+        self.cmd_flow.add_argument(
+            "-m",
+            "--mode",
+            default=TypeExecution.SEQUENTIAL,
+            choices=[
+                TypeExecution.SEQUENTIAL,
+                TypeExecution.BACKGROUND,
+                TypeExecution.PARALLEL,
+                "sequential_group",
+            ],
+            help="Execution mode to visualize (default: sequential)",
+        )
+        self.cmd_flow.add_argument(
+            "--format",
+            default="terminal",
+            choices=["terminal", "mermaid"],
+            help="Output format: terminal (default) or mermaid",
+        )
+
+        self.cmd_flow.set_defaults(exec=FlowCommand)
+
     def setup_logs(self):
         self.cmd_logs = self.subparsers.add_parser("logs", help="Logs")
         self.cmd_logs = self.cmd_logs.add_argument_group(
diff --git a/dotflow/utils/visualizer.py b/dotflow/utils/visualizer.py
new file mode 100644
index 00000000..40da837d
--- /dev/null
+++ b/dotflow/utils/visualizer.py
@@ -0,0 +1,321 @@
+"""Workflow visualizer — terminal and Mermaid rendering."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from rich.console import Console
+from rich.text import Text
+
+if TYPE_CHECKING:
+    from dotflow.core.task import Task
+
+console = Console()
+
+# ── Box-drawing constants ─────────────────────────────────────────────────────
+
+_TL = "┌"
+_TR = "┐"
+_BL = "└"
+_BR = "┘"
+_H = "─"
+_V = "│"
+_ARR = "──▶"
+_FORK = "┬"
+_JOIN_TOP = "┐"
+_JOIN_BOT = "┘"
+_BRANCH_TOP = "┌"
+_BRANCH_BOT = "└"
+
+
+def _task_name(task: Task) -> str:
+    step = task.step
+
+    func = getattr(step, "func", None)
+    if func is not None:
+        return getattr(func, "__name__", None) or type(func).__name__
+
+    # @action(retry=N) wraps the original function in a closure.
+    # We must introspect __closure__ to recover the original function name
+    # because the wrapper doesn't preserve __name__.
+    if callable(step) and getattr(step, "__closure__", None):
+        free_vars = step.__code__.co_freevars
+        cells = step.__closure__
+        if "args" in free_vars:
+            args_cell = cells[free_vars.index("args")].cell_contents
+            if args_cell:
+                return (
+                    getattr(args_cell[0], "__name__", None)
+                    or type(args_cell[0]).__name__
+                )
+
+    return getattr(step, "__name__", None) or type(step).__name__
+
+
+def _action_instance(task: Task):
+    """Return the Action instance for a task regardless of decoration style."""
+    step = task.step
+    if hasattr(step, "retry"):
+        return step
+    if callable(step) and getattr(step, "__closure__", None):
+        free_vars = step.__code__.co_freevars
+        if "self" in free_vars:
+            return step.__closure__[free_vars.index("self")].cell_contents
+    return None
+
+
+def _task_config(task: Task) -> list[str]:
+    """Return notable config lines for the box body (retry, timeout)."""
+    lines = []
+    action = _action_instance(task)
+    if action is None:
+        return lines
+    if getattr(action, "retry", 1) != 1:
+        lines.append(f"retry: {action.retry}")
+    if getattr(action, "timeout", 0):
+        lines.append(f"timeout: {action.timeout}s")
+    return lines
+
+
+def _task_status(task: Task) -> str | None:
+    """Return a status line if the task has already been executed."""
+    from dotflow.core.types.status import TypeStatus
+
+    status = getattr(task, "_status", None)
+    if status is None or status == TypeStatus.NOT_STARTED:
+        return None
+
+    symbol = TypeStatus.get_symbol(status) or ""
+    duration = getattr(task, "_duration", None)
+    dur_str = f" {duration:.2f}s" if duration is not None else ""
+    errors = getattr(task, "_errors", None) or []
+    err_str = ""
+    if errors:
+        last = errors[-1]
+        exc = getattr(last, "exception", None)
+        if exc:
+            err_str = f"\n{exc}"
+    return f"{symbol}{dur_str}{err_str}".strip()
+
+
+# ── Box builder ───────────────────────────────────────────────────────────────
+
+
+def _build_box(task: Task, width: int = 16) -> list[str]:
+    """Return a list of strings representing a single task box."""
+    name = _task_name(task)
+    config_lines = _task_config(task)
+    status_line = _task_status(task)
+
+    inner = width - 2
+
+    def pad(text: str) -> str:
+        text = text[:inner]
+        return f"{_V} {text:<{inner - 1}}{_V}"
+
+    top = _TL + _H * (width - 2) + _TR
+    bot = _BL + _H * (width - 2) + _BR
+
+    rows = [top, pad(name)]
+
+    if config_lines:
+        rows.append(pad(""))
+        for line in config_lines:
+            rows.append(pad(line))
+
+    if status_line:
+        for line in status_line.splitlines():
+            rows.append(pad(line))
+
+    rows.append(bot)
+    return rows
+
+
+# ── Sequential renderer ───────────────────────────────────────────────────────
+
+
+def _render_sequential(tasks: list[Task]) -> str:
+    if not tasks:
+        return "(no tasks)"
+
+    box_w = 18
+    connector = f" {_ARR} "
+    boxes = [_build_box(t, width=box_w) for t in tasks]
+    height = max(len(b) for b in boxes)
+
+    # Pad all boxes to same height (insert blank lines before bottom border)
+    padded = []
+    for box in boxes:
+        if len(box) < height:
+            inner_w = box_w - 2
+            filler = f"{_V} {' ' * (inner_w - 1)}{_V}"
+            box = box[:-1] + [filler] * (height - len(box)) + [box[-1]]
+        padded.append(box)
+
+    mid = 1
+    lines = []
+    for row in range(height):
+        parts = []
+        for i, box in enumerate(padded):
+            parts.append(box[row])
+            if i < len(padded) - 1:
+                parts.append(connector if row == mid else " " * len(connector))
+        lines.append("".join(parts))
+
+    return "\n".join(lines)
+
+
+# ── Parallel renderer ─────────────────────────────────────────────────────────
+
+
+def _render_parallel(tasks: list[Task]) -> str:
+    """
+    Render parallel tasks as a fork/join diagram:
+
+        ┌──────────┐     ┌────────────┐
+        │ task_a   │─┬──▶│ parallel_1 │──┐
+        └──────────┘ │   └────────────┘  │  ┌──────────┐
+                     │   ┌────────────┐  ├─▶│  join?   │
+                     └──▶│ parallel_2 │──┘  └──────────┘
+                         └────────────┘
+    For dotflow parallel mode all tasks run concurrently so we just
+    show them stacked with a leading/trailing bracket.
+    """
+    if not tasks:
+        return "(no tasks)"
+
+    box_w = 18
+    boxes = [_build_box(t, width=box_w) for t in tasks]
+    box_h = max(len(b) for b in boxes)
+
+    padded = []
+    for box in boxes:
+        if len(box) < box_h:
+            inner_w = box_w - 2
+            filler = f"{_V} {' ' * (inner_w - 1)}{_V}"
+            box = box[:-1] + [filler] * (box_h - len(box)) + [box[-1]]
+        padded.append(box)
+
+    indent = "  "
+
+    lines = []
+    for t_idx, box in enumerate(padded):
+        for b_row, b_line in enumerate(box):
+            if len(tasks) == 1:
+                bracket = "  "
+            elif t_idx == 0 and b_row == box_h // 2:
+                bracket = "┌─"
+            elif t_idx == len(tasks) - 1 and b_row == box_h // 2:
+                bracket = "└─"
+            elif b_row == box_h // 2:
+                bracket = "├─"
+            elif (
+                (t_idx == 0 and b_row > box_h // 2)
+                or (t_idx == len(tasks) - 1 and b_row < box_h // 2)
+                or t_idx > 0
+            ):
+                bracket = "│ "
+            else:
+                bracket = "  "
+
+            lines.append(f"{indent}{bracket}{b_line}")
+
+        if t_idx < len(tasks) - 1:
+            lines.append(f"{indent}│ ")
+
+    return "\n".join(lines)
+
+
+# ── Group renderer ────────────────────────────────────────────────────────────
+
+
+def _render_groups(groups: dict[str, list[Task]]) -> str:
+    """Render sequential groups — each group runs in its own process."""
+    sections = []
+    for group_name, tasks in groups.items():
+        header = f"  ── group: {group_name} ──"
+        body = _render_sequential(tasks)
+        indented = "\n".join("  " + line for line in body.splitlines())
+        sections.append(f"{header}\n{indented}")
+    return "\n\n".join(sections)
+
+
+# ── Mermaid export ────────────────────────────────────────────────────────────
+
+
+def _render_mermaid(tasks: list[Task], mode: str) -> str:
+    lines = ["graph LR"]
+    names = [_task_name(t) for t in tasks]
+    # Use positional suffixes as node IDs to prevent Mermaid collapsing
+    # duplicate function names into a single node (self-loop).
+    node_ids = [f"{name}_{i}" for i, name in enumerate(names)]
+
+    if mode == "parallel":
+        lines.append("  START:::hidden")
+        lines.append("  END:::hidden")
+        for node_id, name in zip(node_ids, names):
+            lines.append(f'  START --> {node_id}["{name}"]')
+            lines.append(f'  {node_id}["{name}"] --> END')
+        lines.append("  classDef hidden display:none")
+    else:
+        for i in range(len(node_ids) - 1):
+            lines.append(
+                f'  {node_ids[i]}["{names[i]}"] --> {node_ids[i + 1]}["{names[i + 1]}"]'
+            )
+
+    return "\n".join(lines)
+
+
+# ── Public API ────────────────────────────────────────────────────────────────
+
+
+def visualize(
+    tasks: list[Task],
+    mode: str = "sequential",
+    fmt: str = "terminal",
+) -> None:
+    """
+    Render a workflow pipeline to the terminal or as Mermaid markup.
+
+    Args:
+        tasks: The task list from a DotFlow / TaskBuilder instance.
+        mode:  Execution mode string — 'sequential', 'parallel',
+               'background', or 'sequential_group'.
+        fmt:   Output format — 'terminal' (default) or 'mermaid'.
+    """
+    from dotflow.core.workflow import grouper
+
+    if fmt == "mermaid":
+        console.print(_render_mermaid(tasks, mode), highlight=False)
+        return
+
+    # ── Terminal output ───────────────────────────────────────────────────────
+    has_groups = len(grouper(tasks)) > 1
+
+    if mode == "parallel":
+        diagram = _render_parallel(tasks)
+        mode_label = "parallel"
+    elif mode == "background":
+        diagram = _render_sequential(tasks)
+        mode_label = "background"
+    elif has_groups or mode == "sequential_group":
+        diagram = _render_groups(grouper(tasks))
+        mode_label = "sequential_group"
+    else:
+        diagram = _render_sequential(tasks)
+        mode_label = "sequential"
+
+    task_count = len(tasks)
+    header = Text()
+    header.append("dotflow viz", style="bold cyan")
+    header.append(
+        f"  ·  {task_count} task{'s' if task_count != 1 else ''}", style="dim"
+    )
+    header.append("  ·  mode: ", style="dim")
+    header.append(mode_label, style="bold yellow")
+
+    console.print()
+    console.print(header)
+    console.print()
+    console.print(diagram)
+    console.print()
diff --git a/tests/utils/test_visualizer.py b/tests/utils/test_visualizer.py
new file mode 100644
index 00000000..af478fa5
--- /dev/null
+++ b/tests/utils/test_visualizer.py
@@ -0,0 +1,149 @@
+"""Tests for the workflow visualizer."""
+
+import unittest
+
+from dotflow import DotFlow, action
+from dotflow.utils.visualizer import (
+    _build_box,
+    _render_mermaid,
+    _render_parallel,
+    _render_sequential,
+    _task_name,
+)
+
+
+def _make_tasks(*steps):
+    wf = DotFlow()
+    for step in steps:
+        wf.task.add(step)
+    return list(wf.task.queue)
+
+
+@action
+def step_a():
+    pass
+
+
+@action(retry=3)
+def step_b():
+    pass
+
+
+@action(timeout=60)
+def step_c():
+    pass
+
+
+class TestTaskName(unittest.TestCase):
+    def test_direct_decorator(self):
+        tasks = _make_tasks(step_a)
+        self.assertEqual(_task_name(tasks[0]), "step_a")
+
+    def test_parametrised_decorator(self):
+        tasks = _make_tasks(step_b)
+        self.assertEqual(_task_name(tasks[0]), "step_b")
+
+    def test_timeout_decorator(self):
+        tasks = _make_tasks(step_c)
+        self.assertEqual(_task_name(tasks[0]), "step_c")
+
+
+class TestBuildBox(unittest.TestCase):
+    def test_box_has_top_and_bottom_border(self):
+        tasks = _make_tasks(step_a)
+        box = _build_box(tasks[0], width=18)
+        self.assertTrue(box[0].startswith("┌"))
+        self.assertTrue(box[-1].startswith("└"))
+
+    def test_box_contains_task_name(self):
+        tasks = _make_tasks(step_a)
+        box = _build_box(tasks[0], width=18)
+        full = "\n".join(box)
+        self.assertIn("step_a", full)
+
+    def test_box_shows_retry(self):
+        tasks = _make_tasks(step_b)
+        box = _build_box(tasks[0], width=18)
+        full = "\n".join(box)
+        self.assertIn("retry: 3", full)
+
+    def test_box_shows_timeout(self):
+        tasks = _make_tasks(step_c)
+        box = _build_box(tasks[0], width=18)
+        full = "\n".join(box)
+        self.assertIn("timeout: 60s", full)
+
+
+class TestRenderSequential(unittest.TestCase):
+    def test_all_task_names_present(self):
+        tasks = _make_tasks(step_a, step_b, step_c)
+        output = _render_sequential(tasks)
+        self.assertIn("step_a", output)
+        self.assertIn("step_b", output)
+        self.assertIn("step_c", output)
+
+    def test_arrow_connector_present(self):
+        tasks = _make_tasks(step_a, step_b)
+        output = _render_sequential(tasks)
+        self.assertIn("──▶", output)
+
+    def test_single_task_no_arrow(self):
+        tasks = _make_tasks(step_a)
+        output = _render_sequential(tasks)
+        self.assertNotIn("──▶", output)
+
+    def test_empty_returns_fallback(self):
+        output = _render_sequential([])
+        self.assertEqual(output, "(no tasks)")
+
+
+class TestRenderParallel(unittest.TestCase):
+    def test_all_names_present(self):
+        tasks = _make_tasks(step_a, step_b, step_c)
+        output = _render_parallel(tasks)
+        self.assertIn("step_a", output)
+        self.assertIn("step_b", output)
+        self.assertIn("step_c", output)
+
+    def test_bracket_chars_present(self):
+        tasks = _make_tasks(step_a, step_b)
+        output = _render_parallel(tasks)
+        self.assertIn("┌", output)
+        self.assertIn("└", output)
+
+    def test_empty_returns_fallback(self):
+        output = _render_parallel([])
+        self.assertEqual(output, "(no tasks)")
+
+
+class TestRenderMermaid(unittest.TestCase):
+    def test_sequential_chain(self):
+        tasks = _make_tasks(step_a, step_b, step_c)
+        output = _render_mermaid(tasks, mode="sequential")
+        self.assertIn("graph LR", output)
+        # Node IDs include positional suffix; display labels are the original names
+        self.assertIn('"step_a"', output)
+        self.assertIn('"step_b"', output)
+        self.assertIn('"step_c"', output)
+        # There should be two arrows: a→b and b→c
+        self.assertEqual(output.count("-->"), 2)
+
+    def test_parallel_uses_start_end(self):
+        tasks = _make_tasks(step_a, step_b)
+        output = _render_mermaid(tasks, mode="parallel")
+        self.assertIn("START -->", output)
+        self.assertIn("--> END", output)
+        self.assertIn('"step_a"', output)
+        self.assertIn('"step_b"', output)
+
+    def test_duplicate_names_produce_distinct_nodes(self):
+        # Two tasks wrapping the same function must not collapse into a self-loop
+        tasks = _make_tasks(step_a, step_a)
+        output = _render_mermaid(tasks, mode="sequential")
+        arrow_lines = [
+            line.strip() for line in output.splitlines() if "-->" in line
+        ]
+        self.assertEqual(len(arrow_lines), 1)
+        parts = arrow_lines[0].split("-->")
+        # The left and right node IDs must differ
+        self.assertNotEqual(parts[0].strip(), parts[1].strip())