feat: add dotflow viz — terminal pipeline visualization command

dotflow/cli/commands/__init__.py

@@ -4,5 +4,12 @@
 from dotflow.cli.commands.log import LogCommand
 from dotflow.cli.commands.schedule import ScheduleCommand
 from dotflow.cli.commands.start import StartCommand
+from dotflow.cli.commands.viz import VizCommand
 
-__all__ = ["InitCommand", "LogCommand", "ScheduleCommand", "StartCommand"]
+__all__ = [
+    "InitCommand",
+    "LogCommand",
+    "ScheduleCommand",
+    "StartCommand",
+    "VizCommand",
+]

dotflow/cli/commands/viz.py

@@ -0,0 +1,41 @@
+"""Command viz module"""
+
+from dotflow.cli.command import Command
+from dotflow.core.module import Module
+from dotflow.utils.visualizer import visualize
+
+
+class VizCommand(Command):
+    def setup(self):
+        step = self.params.step
+        mode = self.params.mode
+        fmt = self.params.format
+
+        # Load the workflow step (same module-loading pattern as StartCommand)
+        loaded = Module(value=step)
+
+        # The step may be a DotFlow instance, a TaskBuilder, or a bare list of Tasks.
+        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
+        """
+        # DotFlow wraps a TaskBuilder under .task
+        if hasattr(obj, "task") and hasattr(obj.task, "queue"):
+            return list(obj.task.queue)
+
+        # TaskBuilder exposes .queue directly
+        if hasattr(obj, "queue"):
+            return list(obj.queue)
+
+        # Plain list
+        if isinstance(obj, list):
+            return obj
+
+        return []

dotflow/cli/setup.py

@@ -8,6 +8,7 @@
     LogCommand,
     ScheduleCommand,
     StartCommand,
+    VizCommand,
 )
 from dotflow.core.exception import (
     MESSAGE_UNKNOWN_ERROR,
@@ -39,6 +40,7 @@ def __init__(self, parser):
         self.setup_logs()
         self.setup_start()
         self.setup_schedule()
+        self.setup_viz()
         self.command()
 
     def setup_init(self):
@@ -145,6 +147,42 @@ def setup_schedule(self):
 
         self.cmd_schedule.set_defaults(exec=ScheduleCommand)
 
+    def setup_viz(self):
+        self.cmd_viz = self.subparsers.add_parser(
+            "viz",
+            help="Visualize a workflow pipeline in the terminal",
+        )
+        self.cmd_viz = self.cmd_viz.add_argument_group(
+            "Usage: dotflow viz [OPTIONS]"
+        )
+
+        self.cmd_viz.add_argument(
+            "-s",
+            "--step",
+            required=True,
+            help="Dotted path to a DotFlow instance, TaskBuilder, or task list",
+        )
+        self.cmd_viz.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_viz.add_argument(
+            "--format",
+            default="terminal",
+            choices=["terminal", "mermaid"],
+            help="Output format: terminal (default) or mermaid",
+        )
+
+        self.cmd_viz.set_defaults(exec=VizCommand)
+
     def setup_logs(self):
         self.cmd_logs = self.subparsers.add_parser("logs", help="Logs")
         self.cmd_logs = self.cmd_logs.add_argument_group(