Retain node progress and support execution result files

Author: rsgalloway

README.md

@@ -46,6 +46,7 @@ subfork create graph.json --name "My new graph"
 subfork interface GRAPH_ID
 subfork publish GRAPH_ID --version v1
 subfork execute GRAPH_ID --version v1 --inputs inputs.json
+subfork execute GRAPH_ID -o results.json
 ```
 
 Export writes a graph's draft definition, which can be passed directly to create.
@@ -54,6 +55,8 @@ Descriptions, tags, and published versions are not included in the export.
 JSON input files must contain an object; use `-` to read from stdin.
 Create makes a public draft, publish creates an immutable version, and execute
 waits for completion and prints the graph outputs. Runs consume account quota.
+Use `-o results.json` or `--out results.json` to write result JSON to a new file
+instead of stdout. Progress still goes to stderr. Existing files are not overwritten.
 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
@@ -62,7 +65,8 @@ 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.
+Finished nodes remain on separate stderr lines with their final status before
+the JSON results appear. Redirected progress uses plain text without animation.
 
 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

@@ -23,7 +23,14 @@ def execution_status(graph_name: str) -> Iterator[Callable[[Dict[str, Any]], Non
     label = "Graph " + graph_name
     status = "Running"
     titles: Dict[str, str] = {}
+    reported: Dict[str, str] = {}
+    name_width = len(label)
     stopped = threading.Event()
+    completed_marker = "✓"
+    try:
+        completed_marker.encode(stream.encoding or "ascii")
+    except UnicodeError:
+        completed_marker = "+"
     frames = "⠋⠙⠹⠸⠼⠴⠦⠧⠇⠏"
     try:
         frames.encode(stream.encoding or "ascii")
@@ -32,7 +39,7 @@ def execution_status(graph_name: str) -> Iterator[Callable[[Dict[str, Any]], Non
 
     def update(snapshot: Dict[str, Any]) -> None:
         """Replace the display state with active node titles from a snapshot."""
-        nonlocal label, status
+        nonlocal label, status, name_width
         definition = snapshot.get("definition") or {}
         for node in definition.get("nodes", []):
             titles[node["node_instance_id"]] = node.get("title") or node["node_instance_id"]
@@ -41,6 +48,43 @@ def update(snapshot: Dict[str, Any]) -> None:
             titles.get(key, key) for key, node in nodes.items() if node.get("status") == "running"
         ]
         with lock:
+            name_width = max(
+                [name_width]
+                + [len("Node " + title) for title in titles.values()]
+                + [len("Node " + key) for key in nodes if key not in titles]
+            )
+            for key, node in nodes.items():
+                outcome = node.get("status")
+                if outcome not in {
+                    "completed",
+                    "failed",
+                    "canceled",
+                    "cancelled",
+                    "skipped",
+                    "outcome_unknown",
+                }:
+                    continue
+                if reported.get(key) == outcome:
+                    continue
+                reported[key] = outcome
+                state = "Canceled" if outcome == "cancelled" else outcome.replace("_", " ").title()
+                name = "Node " + titles.get(key, key)
+                if terminal:
+                    stream.write(
+                        "\r\033[2K"
+                        + format_line(
+                            completed_marker if outcome == "completed" else "-", name, state
+                        )
+                        + "\n"
+                    )
+                else:
+                    stream.write(
+                        format_line(
+                            completed_marker if outcome == "completed" else "-", name, state
+                        )
+                        + "\n"
+                    )
+                stream.flush()
             label = (
                 ("Node " if len(active) == 1 else "Nodes ") + ", ".join(active)
                 if active
@@ -52,25 +96,30 @@ def update(snapshot: Dict[str, Any]) -> None:
                 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)
+    def format_line(frame: str, name: str, state: str) -> str:
+        """Format a width-limited progress row without splitting color escapes."""
+        name = "".join(char if char.isprintable() else " " for char in name)
+        state = "".join(char if char.isprintable() else " " for char in state)
         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)]
+        # Reserve space for the longest status so different outcomes align too.
+        status_column = min(name_width + 14, max(7, width - len("Outcome Unknown")))
+        name = name[: max(0, status_column - 7)]
+        dots = "." * max(3, status_column - len(name) - 4)
         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()
+            return plain[:width]
+        if not terminal or os.environ.get("NO_COLOR"):
+            return plain
+        color = "32" if state in {"Running", "Completed"} else "31" if state == "Failed" else "33"
+        colored_state = "\033[" + color + "m" + state + "\033[0m"
+        marker_color = "32" if state == "Completed" else "33"
+        return "\033[{}m{}\033[0m {} {} {}".format(marker_color, frame, name, dots, colored_state)
+
+    def render(index: int) -> None:
+        """Draw the active row without interleaving retained completion lines."""
+        with lock:
+            stream.write("\r\033[2K" + format_line(frames[index % len(frames)], label, status))
+            stream.flush()
 
     def animate() -> None:
         """Refresh the spinner independently of network requests and polling."""
@@ -122,6 +171,9 @@ def build_parser() -> argparse.ArgumentParser:
             child.add_argument("--comment", default="")
         elif command == "execute":
             child.add_argument("--version", default="draft")
+            child.add_argument(
+                "-o", "--out", help="Write result JSON to a new file instead of stdout"
+            )
             child.add_argument(
                 "--no-wait", action="store_true", help="Return submission status immediately"
             )
@@ -205,13 +257,14 @@ def graph_command(client: Subfork, args: argparse.Namespace) -> Any:
     ):
         raise ValueError("Wait timeout and poll interval must be positive finite numbers.")
     result = graphs.execute(args.graph_id, version=args.version, inputs=inputs)
-    if not args.no_wait and result.get("status") not in {
+    terminal_statuses = {
         "completed",
         "failed",
         "canceled",
         "cancelled",
         "outcome_unknown",
-    }:
+    }
+    if not args.no_wait:
         execution_id = result.get("id") or result.get("execution_id")
         if not isinstance(execution_id, str) or not execution_id:
             raise ValueError(
@@ -225,12 +278,13 @@ def graph_command(client: Subfork, args: argparse.Namespace) -> Any:
             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,
-            )
+            if result.get("status") not in terminal_statuses:
+                result = client.executions.wait(
+                    execution_id,
+                    timeout=args.wait_timeout,
+                    poll_interval=args.poll_interval,
+                    on_update=update,
+                )
     return result
 
 
@@ -242,6 +296,9 @@ def main(argv: Optional[Sequence[str]] = None) -> int:
     """
     args = build_parser().parse_args(argv)
     try:
+        result_file = getattr(args, "out", None)
+        if result_file is not None and Path(result_file).exists():
+            raise ValueError("Output file already exists; choose a new path.")
         with Subfork(base_url=args.base_url, timeout=args.timeout) as client:
             result = graph_command(client, args)
         failed = args.command == "execute" and result.get("status") in {
@@ -261,11 +318,13 @@ def main(argv: Optional[Sequence[str]] = None) -> int:
                 display = result.get("outputs", {})
         rendered = json.dumps(display, indent=2, ensure_ascii=False) + "\n"
         output = getattr(args, "output", "-")
-        if output == "-":
+        if result_file is None and output == "-":
             sys.stdout.write(rendered)
         else:
-            # Exclusive creation avoids silently overwriting a local definition.
-            with Path(output).open("x", encoding="utf-8", newline="\n") as stream:
+            # Exclusive creation avoids silently overwriting an existing file.
+            with Path(result_file if result_file is not None else output).open(
+                "x", encoding="utf-8", newline="\n"
+            ) as stream:
                 stream.write(rendered)
         if failed:
             print(

tests/test_cli.py

@@ -2,6 +2,7 @@
 
 import io
 import json
+import re
 from pathlib import Path
 from typing import Any
 
@@ -260,3 +261,91 @@ def write(self, value: str) -> int:
     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())
+
+
+@pytest.mark.parametrize("terminal", [False, True])
+def test_retained_node_outcomes(
+    monkeypatch: pytest.MonkeyPatch, capsys: Any, terminal: bool
+) -> None:
+    """Retain each terminal node once, including nodes that finish between polls."""
+
+    class ProgressStream(io.StringIO):
+        """Capture either terminal or redirected progress."""
+
+        def isatty(self) -> bool:
+            """Select the requested output mode."""
+            return terminal
+
+    stream = ProgressStream()
+    monkeypatch.setattr("sys.stderr", stream)
+    monkeypatch.setenv("TERM", "xterm")
+    monkeypatch.setenv("NO_COLOR", "1")
+    snapshot = {
+        "status": "failed",
+        "definition": {
+            "nodes": [
+                {"node_instance_id": "a", "title": "Fetch"},
+                {"node_instance_id": "b", "title": "Parse"},
+                {"node_instance_id": "c", "title": "Save"},
+            ]
+        },
+        "node_executions": {
+            "a": {"status": "completed"},
+            "b": {"status": "failed"},
+            "c": {"status": "canceled"},
+        },
+    }
+    with cli.execution_status("Pipeline") as update:
+        update(snapshot)
+        update(snapshot)
+    output = stream.getvalue()
+    for name, outcome in (("Fetch", "Completed"), ("Parse", "Failed"), ("Save", "Canceled")):
+        assert len(re.findall(r"Node " + name + r" \.+ " + outcome + r"\n", output)) == 1
+    assert capsys.readouterr().out == ""
+
+
+def test_node_status_columns_align(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Pad different node names so terminal outcomes start in the same column."""
+    stream = io.StringIO()
+    monkeypatch.setattr("sys.stderr", stream)
+    with cli.execution_status("Demo") as update:
+        update(
+            {
+                "node_executions": {
+                    "A": {"status": "completed"},
+                    "Longer node name": {"status": "failed"},
+                    "Mid": {"status": "canceled"},
+                }
+            }
+        )
+    lines = stream.getvalue().splitlines()[1:]
+    columns = [line.index(state) for line, state in zip(lines, ["Completed", "Failed", "Canceled"])]
+    assert len(columns) == 3
+    assert len(set(columns)) == 1
+
+
+@pytest.mark.parametrize("flag", ["-o", "--out"])
+@pytest.mark.parametrize("mode", [None, "--raw", "--no-wait"])
+def test_execute_result_file(
+    requests: list, tmp_path: Path, capsys: Any, flag: str, mode: Any
+) -> None:
+    """Route results exclusively to a file and reject overwrites before submission."""
+    output = tmp_path / "result.json"
+    arguments = ["execute", "g_test", flag, str(output)]
+    if mode:
+        arguments.append(mode)
+    assert cli.main(arguments) == 0
+    value = json.loads(output.read_text())
+    assert value == (
+        {"id": "test", "status": "completed"}
+        if mode == "--raw"
+        else {"execution_id": "test", "status": "completed"} if mode == "--no-wait" else {}
+    )
+    assert capsys.readouterr().out == ""
+    count = len(requests)
+    assert cli.main(arguments) == 1
+    assert len(requests) == count
+    assert json.loads(output.read_text()) == value
+    captured = capsys.readouterr()
+    assert captured.out == ""
+    assert "already exists" in captured.err