Add animated execution progress with active node names

Author: rsgalloway

README.md

@@ -58,6 +58,11 @@ Use `execute --no-wait` for a submission summary or `execute --raw` for the full
 execution snapshot. Waiting requires read and run scopes; submission alone requires
 run scope. `--wait-timeout` (default 120) and `--poll-interval` (default 2) are in
 seconds. A timeout stops waiting without canceling the remote run.
+A yellow spinner precedes `Graph <name> .......... Running`, with `Running` in
+green. As status snapshots arrive, the line shows the currently running node titles
+(or multiple titles for parallel nodes). Short-lived nodes may finish between polls.
+Set `NO_COLOR` to disable colors. Progress goes to stderr; stdout remains JSON.
+Redirected progress uses a single plain-text line.
 
 Other commands include `get`, `published`, and `versions`. Use `--help` on any
 command. Node discovery and execution monitoring are available through the Python API.

src/subfork/cli.py

@@ -4,13 +4,102 @@
 import json
 import math
 import os
+import shutil
 import sys
+import threading
+from contextlib import contextmanager
 from pathlib import Path
-from typing import Any, Dict, Optional, Sequence
+from typing import Any, Callable, Dict, Iterator, Optional, Sequence
 
 from . import AuthenticationError, ExecutionTimeout, Subfork, SubforkError, __version__
 
 
+@contextmanager
+def execution_status(graph_name: str) -> Iterator[Callable[[Dict[str, Any]], None]]:
+    """Animate the active nodes on terminal stderr while preserving JSON stdout."""
+    stream = sys.stderr
+    terminal = stream.isatty() and os.environ.get("TERM") != "dumb"
+    lock = threading.Lock()
+    label = "Graph " + graph_name
+    status = "Running"
+    titles: Dict[str, str] = {}
+    stopped = threading.Event()
+    frames = "⠋⠙⠹⠸⠼⠴⠦⠧⠇⠏"
+    try:
+        frames.encode(stream.encoding or "ascii")
+    except UnicodeError:
+        frames = "|/-\\"
+
+    def update(snapshot: Dict[str, Any]) -> None:
+        """Replace the display state with active node titles from a snapshot."""
+        nonlocal label, status
+        definition = snapshot.get("definition") or {}
+        for node in definition.get("nodes", []):
+            titles[node["node_instance_id"]] = node.get("title") or node["node_instance_id"]
+        nodes = snapshot.get("node_executions") or {}
+        active = [
+            titles.get(key, key) for key, node in nodes.items() if node.get("status") == "running"
+        ]
+        with lock:
+            label = (
+                ("Node " if len(active) == 1 else "Nodes ") + ", ".join(active)
+                if active
+                else "Graph " + graph_name
+            )
+            status = (
+                "Running"
+                if active
+                else str(snapshot.get("status", "running")).replace("_", " ").title()
+            )
+
+    def render(index: int) -> None:
+        """Draw one width-limited line with a yellow spinner and green run status."""
+        with lock:
+            name = "".join(char if char.isprintable() else " " for char in label)
+            state = "".join(char if char.isprintable() else " " for char in status)
+        width = max(1, shutil.get_terminal_size().columns - 1)
+        name = name[: max(0, width - len(state) - 7)]
+        dots = "." * max(3, min(10, width - len(name) - len(state) - 4))
+        frame = frames[index % len(frames)]
+        plain = "{} {} {} {}".format(frame, name, dots, state)
+        if len(plain) > width:
+            line = plain[:width]
+        elif os.environ.get("NO_COLOR"):
+            line = plain
+        else:
+            colored_state = "\033[32m" + state + "\033[0m" if state == "Running" else state
+            line = "\033[33m{}\033[0m {} {} {}".format(frame, name, dots, colored_state)
+        stream.write("\r\033[2K" + line)
+        stream.flush()
+
+    def animate() -> None:
+        """Refresh the spinner independently of network requests and polling."""
+        index = 1
+        while not stopped.wait(0.1):
+            try:
+                render(index)
+            except (OSError, ValueError):
+                return
+            index += 1
+
+    if not terminal:
+        safe_name = "".join(char if char.isprintable() else " " for char in graph_name)
+        print("Graph {} .......... Running".format(safe_name), file=stream)
+        yield update
+        return
+    worker = threading.Thread(target=animate, name="subfork-spinner", daemon=True)
+    try:
+        render(0)
+        worker.start()
+        yield update
+    finally:
+        stopped.set()
+        if worker.ident is not None:
+            worker.join()
+        stream.write("\r\033[2K")
+        stream.flush()
+
+
 def build_parser() -> argparse.ArgumentParser:
     """Build the public command tree without opening a connection."""
     parser = argparse.ArgumentParser(description=__doc__)
@@ -128,10 +217,20 @@ def graph_command(client: Subfork, args: argparse.Namespace) -> Any:
             raise ValueError(
                 "Submission returned no execution ID; inspect remote state before retrying."
             )
-        print("subfork: waiting for execution {}".format(execution_id), file=sys.stderr)
-        result = client.executions.wait(
-            execution_id, timeout=args.wait_timeout, poll_interval=args.poll_interval
+        snapshot_definition = result.get("definition")
+        graph_name = (
+            snapshot_definition.get("name") if isinstance(snapshot_definition, dict) else None
         )
+        with execution_status(
+            graph_name if isinstance(graph_name, str) else args.graph_id
+        ) as update:
+            update(result)
+            result = client.executions.wait(
+                execution_id,
+                timeout=args.wait_timeout,
+                poll_interval=args.poll_interval,
+                on_update=update,
+            )
     return result
 
 

src/subfork/resources.py

@@ -4,7 +4,7 @@
 
 import math
 import time
-from typing import TYPE_CHECKING, Any, Dict, List, Optional
+from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional
 from urllib.parse import quote
 
 from .errors import ExecutionTimeout
@@ -173,7 +173,12 @@ def cancel(self, execution_id: str) -> Dict[str, Any]:
         return self._client._request("POST", "/executions/" + segment(execution_id) + "/cancel")
 
     def wait(
-        self, execution_id: str, *, timeout: float = 120, poll_interval: float = 2
+        self,
+        execution_id: str,
+        *,
+        timeout: float = 120,
+        poll_interval: float = 2,
+        on_update: Optional[Callable[[Dict[str, Any]], None]] = None,
     ) -> Dict[str, Any]:
         """Poll until a terminal status or the polling deadline.
 
@@ -182,6 +187,8 @@ def wait(
             timeout: Positive deadline in seconds. Remaining time caps HTTP
                 timeouts, which apply per I/O phase, not to the entire request.
             poll_interval: Positive delay in seconds between status requests.
+            on_update: Optional synchronous callback for each retrieved snapshot.
+                Callback exceptions propagate to the caller without canceling the run.
 
         Returns:
             The terminal snapshot, including failed or canceled executions.
@@ -208,6 +215,8 @@ def wait(
                 "/executions/" + segment(execution_id),
                 timeout=min(self._client.timeout, remaining),
             )
+            if on_update is not None:
+                on_update(result)
             if result["status"] in {
                 "completed",
                 "failed",

tests/test_cli.py

@@ -181,3 +181,82 @@ def timeout(*args: Any, **kwargs: Any) -> dict:
             "status": "failed" if mode == "failed" else "running",
         }
     assert len(seen) == (1 if mode == "no-wait" else 2)
+
+
+@pytest.mark.parametrize("interrupted", [False, True])
+@pytest.mark.parametrize("no_color", [False, True])
+def test_status_cleanup(monkeypatch: pytest.MonkeyPatch, interrupted: bool, no_color: bool) -> None:
+    """Show the graph name and restore terminal output on success or interruption."""
+
+    class Terminal(io.StringIO):
+        """Capture writes as an interactive terminal."""
+
+        def isatty(self) -> bool:
+            """Identify this stream as a terminal."""
+            return True
+
+    terminal = Terminal()
+    monkeypatch.setattr("sys.stderr", terminal)
+    monkeypatch.setenv("TERM", "xterm")
+    monkeypatch.setenv("NO_COLOR", "1" if no_color else "")
+    try:
+        with cli.execution_status("URL Extract Pipe"):
+            expected = "Running" if no_color else "\033[32mRunning\033[0m"
+            assert "Graph URL Extract Pipe .......... " + expected in terminal.getvalue()
+            if interrupted:
+                raise KeyboardInterrupt
+    except KeyboardInterrupt:
+        assert interrupted
+    assert terminal.getvalue().endswith("\r\033[2K")
+
+
+def test_status_redirected_stderr(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Keep redirected progress plain and sanitize control characters in names."""
+    stream = io.StringIO()
+    monkeypatch.setattr("sys.stderr", stream)
+    with cli.execution_status("Greeting\nGraph"):
+        pass
+    assert stream.getvalue() == "Graph Greeting Graph .......... Running\n"
+
+
+def test_active_node_progress(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Animate named parallel nodes and clean up the rendering thread."""
+    import threading
+
+    refreshed = threading.Event()
+
+    class Terminal(io.StringIO):
+        """Capture animated frames and signal when node titles appear."""
+
+        def isatty(self) -> bool:
+            """Enable interactive progress rendering."""
+            return True
+
+        def write(self, value: str) -> int:
+            """Signal a node frame without timing-dependent sleeps."""
+            count = super().write(value)
+            if "Nodes Fetch, Parse" in value:
+                refreshed.set()
+            return count
+
+    terminal = Terminal()
+    monkeypatch.setattr("sys.stderr", terminal)
+    monkeypatch.setenv("TERM", "xterm")
+    monkeypatch.delenv("NO_COLOR", raising=False)
+    with cli.execution_status("Pipeline") as update:
+        update(
+            {
+                "status": "running",
+                "definition": {
+                    "nodes": [
+                        {"node_instance_id": "a", "title": "Fetch"},
+                        {"node_instance_id": "b", "title": "Parse"},
+                    ]
+                },
+                "node_executions": {"a": {"status": "running"}, "b": {"status": "running"}},
+            }
+        )
+        assert refreshed.wait(2)
+    assert "\033[33m" in terminal.getvalue()
+    assert "\033[32mRunning\033[0m" in terminal.getvalue()
+    assert not any(thread.name == "subfork-spinner" for thread in threading.enumerate())

tests/test_client.py

@@ -185,3 +185,18 @@ def handle(request: httpx.Request) -> httpx.Response:
         with pytest.raises(ExecutionTimeout):
             client.executions.wait("e_test", timeout=1)
     assert len(seen) == 1 and seen[0].method == "GET"
+
+
+def test_wait_reports_snapshots() -> None:
+    """Deliver both active and terminal snapshots to an optional progress callback."""
+    snapshots = []
+    statuses = iter(["running", "completed"])
+    with Subfork(
+        "synthetic-key",
+        transport=httpx.MockTransport(
+            lambda request: httpx.Response(200, json={"status": next(statuses)})
+        ),
+    ) as client:
+        result = client.executions.wait("e_test", poll_interval=0.001, on_update=snapshots.append)
+    assert [snapshot["status"] for snapshot in snapshots] == ["running", "completed"]
+    assert result == snapshots[-1]