feat: update shellflow version to 0.4.7 and enhance output handling in execution runner

Author: Akagi201

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "shellflow"
-version = "0.4.6"
+version = "0.4.7"
 description = "A minimal shell script orchestrator with SSH support"
 readme = "README.md"
 license = "Apache-2.0"

src/shellflow/__init__.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 from .cli import create_parser
-from .config import read_ssh_config
+from .config import parse_server_config, parse_variables, read_ssh_config
 from .constants import (
     EXIT_EXECUTION_FAILURE,
     EXIT_PARSE_FAILURE,
@@ -101,6 +101,8 @@
     "execute_remote",
     "main",
     "parse_script",
+    "parse_server_config",
+    "parse_variables",
     "read_ssh_config",
     "run_script",
 ]

src/shellflow/executor.py

@@ -8,8 +8,12 @@
 import subprocess
 import threading
 import uuid
+from contextlib import suppress
 from pathlib import Path
-from typing import Any, Callable
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
 
 from .constants import TRACE_MARKER
 from .models import Block, BlockExecutor, CommandLog, ExecutionContext, ExecutionResult, SSHConfig
@@ -316,6 +320,7 @@ def _run_remote_subprocess(
     *,
     timeout_seconds: int | None,
     on_command: Callable[[str], None] | None = None,
+    on_output: Callable[[str], None] | None = None,
 ) -> tuple[str, str, int, bool, bool]:
     """Run an SSH subprocess and preserve partial output on timeout or interruption."""
     process = subprocess.Popen(
@@ -330,6 +335,7 @@ def _run_remote_subprocess(
         input_text=remote_script,
         timeout_seconds=timeout_seconds,
         on_command=on_command,
+        on_output=on_output,
     )
 
 
@@ -418,6 +424,7 @@ def execute_local_traced(
     context: ExecutionContext,
     no_input: bool = False,
     on_command: Callable[[str], None] | None = None,
+    on_output: Callable[[str], None] | None = None,
 ) -> ExecutionResult:
     """Execute a local block while preserving per-command trace logs."""
     if not block.commands:
@@ -453,6 +460,7 @@ def execute_local_traced(
             input_text=input_text,
             timeout_seconds=block.timeout_seconds,
             on_command=on_command,
+            on_output=on_output,
         )
         cleaned_stderr = _strip_trace_markers(stderr).strip()
         command_logs = _parse_grouped_trace_output(
@@ -505,6 +513,7 @@ def execute_remote(
     no_input: bool = False,
     servers: dict[str, dict[str, str]] | None = None,
     on_command: Callable[[str], None] | None = None,
+    on_output: Callable[[str], None] | None = None,
 ) -> ExecutionResult:
     """Execute a remote block via SSH.
 
@@ -580,6 +589,7 @@ def execute_remote(
             remote_script,
             timeout_seconds=block.timeout_seconds,
             on_command=on_command,
+            on_output=on_output,
         )
         cleaned_stdout = _strip_trace_markers(stdout)
         cleaned_stderr = _strip_trace_markers(stderr)
@@ -779,6 +789,7 @@ def _run_traced_subprocess(
     input_text: str | None,
     timeout_seconds: int | None,
     on_command: Callable[[str], None] | None = None,
+    on_output: Callable[[str], None] | None = None,
 ) -> tuple[str, str, int, bool, bool]:
     """Read traced stdout/stderr streams while surfacing live command markers."""
     stdout_chunks: list[str] = []
@@ -789,25 +800,16 @@ def _run_traced_subprocess(
     if stdout_stream is None or stderr_stream is None:
         raise subprocess.SubprocessError("traced subprocess did not expose stdout/stderr pipes")
 
-    def read_stdout() -> None:
-        try:
-            for line in iter(stdout_stream.readline, ""):
-                stdout_chunks.append(line)
-                command = _extract_trace_command_line(line)
-                if command is not None and on_command is not None:
-                    on_command(command)
-        finally:
-            stdout_stream.close()
-
-    def read_stderr() -> None:
-        try:
-            for line in iter(stderr_stream.readline, ""):
-                stderr_chunks.append(line)
-        finally:
-            stderr_stream.close()
-
-    stdout_thread = threading.Thread(target=read_stdout, daemon=True)
-    stderr_thread = threading.Thread(target=read_stderr, daemon=True)
+    stdout_thread = threading.Thread(
+        target=_read_traced_stdout_stream,
+        args=(stdout_stream, stdout_chunks, on_command, on_output),
+        daemon=True,
+    )
+    stderr_thread = threading.Thread(
+        target=_read_traced_output_stream,
+        args=(stderr_stream, stderr_chunks, on_output),
+        daemon=True,
+    )
     stdout_thread.start()
     stderr_thread.start()
 
@@ -816,14 +818,7 @@ def read_stderr() -> None:
     stdin_stream = process.stdin
 
     try:
-        if stdin_stream is not None:
-            if input_text is not None:
-                try:
-                    stdin_stream.write(input_text)
-                except BrokenPipeError:
-                    pass
-            stdin_stream.close()
-
+        _write_process_input(stdin_stream, input_text)
         process.wait(timeout=timeout_seconds)
     except subprocess.TimeoutExpired:
         timed_out = True
@@ -853,6 +848,50 @@ def read_stderr() -> None:
     )
 
 
+def _read_traced_stdout_stream(
+    stdout_stream: Any,
+    stdout_chunks: list[str],
+    on_command: Callable[[str], None] | None,
+    on_output: Callable[[str], None] | None,
+) -> None:
+    """Read traced stdout, separating command markers from regular output."""
+    try:
+        for line in iter(stdout_stream.readline, ""):
+            stdout_chunks.append(line)
+            command = _extract_trace_command_line(line)
+            if command is not None and on_command is not None:
+                on_command(command)
+            elif command is None and on_output is not None:
+                on_output(line)
+    finally:
+        stdout_stream.close()
+
+
+def _read_traced_output_stream(
+    stream: Any,
+    chunks: list[str],
+    on_output: Callable[[str], None] | None,
+) -> None:
+    """Read a traced process output stream and forward raw chunks."""
+    try:
+        for line in iter(stream.readline, ""):
+            chunks.append(line)
+            if on_output is not None:
+                on_output(line)
+    finally:
+        stream.close()
+
+
+def _write_process_input(stdin_stream: Any, input_text: str | None) -> None:
+    """Write the traced script to stdin, tolerating early process exits."""
+    if stdin_stream is None:
+        return
+    if input_text is not None:
+        with suppress(BrokenPipeError):
+            stdin_stream.write(input_text)
+    stdin_stream.close()
+
+
 def _parse_grouped_trace_output(
     combined_output: str,
     *,

src/shellflow/runner.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import threading
 import time
 import uuid
 from concurrent.futures import ThreadPoolExecutor, as_completed
@@ -809,6 +810,44 @@ def _print_live_command_status(command: str, command_index: int, total_commands:
     print(f"> [{command_index}/{total_commands}] {command}", flush=True)
 
 
+class _LiveCommandOutputPrinter:
+    """Print each traced command's buffered output before the next command starts."""
+
+    def __init__(self, total_commands: int, output_tail_lines: int) -> None:
+        self.total_commands = total_commands
+        self.output_tail_lines = output_tail_lines
+        self.command_count = 0
+        self._output_chunks: list[str] = []
+        self._lock = threading.Lock()
+
+    def start_command(self, command: str) -> None:
+        """Start a command section, flushing the previous command's output first."""
+        with self._lock:
+            if self.command_count:
+                self._flush_locked()
+            self.command_count += 1
+            _print_live_command_status(command, self.command_count, self.total_commands)
+
+    def append_output(self, chunk: str) -> None:
+        """Buffer output for the active traced command."""
+        if not chunk:
+            return
+        with self._lock:
+            self._output_chunks.append(chunk)
+
+    def finish(self) -> None:
+        """Flush the final command's output after the traced process exits."""
+        with self._lock:
+            if self.command_count:
+                self._flush_locked()
+
+    def _flush_locked(self) -> None:
+        output = "".join(self._output_chunks).strip()
+        self._output_chunks = []
+        if output:
+            print(_truncate_output_lines(output, self.output_tail_lines), flush=True)
+
+
 def _group_blocks_for_parallel_execution(blocks: list[Block]) -> list[list[Block]]:
     """Group blocks for parallel execution based on parallel annotations.
 
@@ -921,41 +960,41 @@ def _execute_remote_block_sequential(
     DIM = ANSI_DIM
     RESET = ANSI_RESET
 
-    live_command_index = 0
-
-    def on_command(command: str) -> None:
-        nonlocal live_command_index
-        live_command_index += 1
-        _print_live_command_status(command, live_command_index, len(commands_to_execute))
+    live_printer = _LiveCommandOutputPrinter(len(commands_to_execute), output_tail_lines) if verbose else None
 
     result = executor_module.execute_remote(
         block,
         context,
         ssh_config=None,
         no_input=no_input,
         servers=servers,
-        on_command=on_command if verbose else None,
+        on_command=live_printer.start_command if live_printer else None,
+        on_output=live_printer.append_output if live_printer else None,
     )
 
     if verbose:
+        if live_printer:
+            live_printer.finish()
+
         # Assign actual command names to parsed logs
         if result.command_logs:
             for i, cl in enumerate(result.command_logs):
                 if cl.command == "<remote-command>" and i < len(commands_to_execute):
                     cl.command = commands_to_execute[i]
 
-        # Print each command with its output
-        if result.command_logs:
-            _print_command_logs(result.command_logs, output_tail_lines)
-            if result.output and not any(command_log.output for command_log in result.command_logs):
-                print(_truncate_output_lines(result.output, output_tail_lines))
-        else:
-            # Fallback: no command logs, just show commands
-            for cmd in commands_to_execute:
-                print(f"{DIM}$ {cmd}{RESET}")
-            if result.output:
-                truncated = _truncate_output_lines(result.output, output_tail_lines)
-                print(truncated)
+        # Print grouped logs only when no live trace markers were observed.
+        if not (live_printer and live_printer.command_count):
+            if result.command_logs:
+                _print_command_logs(result.command_logs, output_tail_lines)
+                if result.output and not any(command_log.output for command_log in result.command_logs):
+                    print(_truncate_output_lines(result.output, output_tail_lines))
+            else:
+                # Fallback: no command logs, just show commands
+                for cmd in commands_to_execute:
+                    print(f"{DIM}$ {cmd}{RESET}")
+                if result.output:
+                    truncated = _truncate_output_lines(result.output, output_tail_lines)
+                    print(truncated)
 
     context.last_output = result.output
     context.success = result.success
@@ -1006,27 +1045,27 @@ def _execute_local_block_sequential(
     RED = ANSI_RED
     RESET = ANSI_RESET
 
-    live_command_index = 0
-
-    def on_command(command: str) -> None:
-        nonlocal live_command_index
-        live_command_index += 1
-        _print_live_command_status(command, live_command_index, len(commands_to_execute))
+    live_printer = _LiveCommandOutputPrinter(len(commands_to_execute), output_tail_lines) if verbose else None
 
     result = execute_local_traced(
         block,
         context,
         no_input=no_input,
-        on_command=on_command if verbose else None,
+        on_command=live_printer.start_command if live_printer else None,
+        on_output=live_printer.append_output if live_printer else None,
     )
 
     if verbose:
-        if result.command_logs:
-            _print_command_logs(result.command_logs, output_tail_lines)
-            if result.output and not any(command_log.output for command_log in result.command_logs):
+        if live_printer:
+            live_printer.finish()
+
+        if not (live_printer and live_printer.command_count):
+            if result.command_logs:
+                _print_command_logs(result.command_logs, output_tail_lines)
+                if result.output and not any(command_log.output for command_log in result.command_logs):
+                    print(_truncate_output_lines(result.output, output_tail_lines))
+            elif result.output:
                 print(_truncate_output_lines(result.output, output_tail_lines))
-        elif result.output:
-            print(_truncate_output_lines(result.output, output_tail_lines))
 
     # Print exit code if failed
     if not result.success and verbose: