Testenv.

pyproject.toml

@@ -134,6 +134,7 @@ commands =
     codespell {posargs: --quiet-level=2} src/
     python3 -m bandit -r src/reflow --severity-level medium --confidence-level high
 [testenv:test]
+extras = pretty
 setenv =
     VIRTUALENV_SYSTEM_SITE_PACKAGES=true
 description = "Run the unit tests."

src/reflow/_dag_render.py

@@ -19,10 +19,7 @@
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
-
-if TYPE_CHECKING:
-    from .workflow import Workflow
+from .workflow import Workflow
 
 FORMATS = ("text", "mermaid", "dot", "phart")
 
@@ -94,29 +91,30 @@ def render_dot(wf: Workflow) -> str:
 def render_phart(wf: Workflow, *, use_ascii: bool = False) -> str:
     """Pretty ASCII/Unicode DAG via the optional phart + networkx extra.
 
-    Array tasks are decorated with angle brackets ``<<name>>``; singletons
-    with square brackets ``[name]``.  Raises ImportError if the extra is
-    not installed; the caller should catch this and fall back to text.
+    Array tasks are suffixed ``[array]`` in their label so they stand out
+    from singleton tasks.  Raises ImportError if the extra is not installed;
+    the caller should catch this and fall back to text.
+
+    The marker is baked into the node label rather than using phart's
+    ``custom_decorators``/``NodeStyle.CUSTOM``, because those are only
+    available in newer phart releases and their behaviour varies across
+    versions.  A plain label suffix renders consistently everywhere.
     """
     import networkx as nx  # noqa: PLC0415 - optional dependency
-    from phart import ASCIIRenderer, NodeStyle  # noqa: PLC0415
+    from phart import ASCIIRenderer  # noqa: PLC0415
 
     array = _array_tasks(wf)
+
+    def label(name: str) -> str:
+        return f"{name} [array]" if name in array else name
+
     g = nx.DiGraph()
     # Add all nodes so isolated tasks still render.
     for tname in wf._topological_order():
-        g.add_node(tname)
-    g.add_edges_from(_edges(wf))
-
-    decorators = {
-        name: (("<<", ">>") if name in array else ("[", "]")) for name in g.nodes
-    }
-    renderer = ASCIIRenderer(
-        g,
-        node_style=NodeStyle.CUSTOM,
-        custom_decorators=decorators,
-        use_ascii=use_ascii,
-    )
+        g.add_node(label(tname))
+    g.add_edges_from((label(dep), label(t)) for dep, t in _edges(wf))
+
+    renderer = ASCIIRenderer(g, use_ascii=use_ascii)
     result: str = renderer.render()
     return result.rstrip("\n")
 

tests/test_cli.py

@@ -891,9 +891,9 @@ def test_dag_format_phart_when_available(
             "convert_source",
         ):
             assert name in out
-        # Array tasks are decorated with angle brackets, singletons with []
-        assert "<<download_source>>" in out
-        assert "[gather_sources]" in out
+        # Array tasks carry the [array] suffix; singletons do not.
+        assert "download_source [array]" in out
+        assert "gather_sources [array]" not in out
 
     def test_dag_format_phart_ascii_flag(
         self, capsys: pytest.CaptureFixture[str]

tests/test_dag_render.py

@@ -71,7 +71,7 @@ def test_render_phart_via_dispatcher(self) -> None:
         pytest.importorskip("networkx")
         out = _dag_render.render(_make_wf(), "phart")
         assert "gather_sources" in out
-        assert "<<download_source>>" in out
+        assert "download_source [array]" in out
 
     def test_render_phart_via_dispatcher_ascii(self) -> None:
         pytest.importorskip("phart")
@@ -169,12 +169,16 @@ def test_renders_all_nodes(self) -> None:
         ):
             assert name in out
 
-    def test_array_and_singleton_decorators(self) -> None:
+    def test_array_and_singleton_labels(self) -> None:
         pytest.importorskip("phart")
         pytest.importorskip("networkx")
         out = _dag_render.render_phart(_make_wf())
-        assert "<<download_source>>" in out
-        assert "[gather_sources]" in out
+        # Array tasks carry the [array] suffix; singletons do not.
+        assert "download_source [array]" in out
+        assert "convert_source [array]" in out
+        # A singleton name appears without the suffix
+        assert "gather_sources" in out
+        assert "gather_sources [array]" not in out
 
     def test_ascii_mode_no_unicode(self) -> None:
         pytest.importorskip("phart")