diff --git a/src/flext_cli/_protocols/_base_parts/flextcliprotocolsbase_part_02.py b/src/flext_cli/_protocols/_base_parts/flextcliprotocolsbase_part_02.py index 7b926137..01c4fa6f 100644 --- a/src/flext_cli/_protocols/_base_parts/flextcliprotocolsbase_part_02.py +++ b/src/flext_cli/_protocols/_base_parts/flextcliprotocolsbase_part_02.py @@ -121,7 +121,12 @@ def pid(self) -> int: @property def stdout(self) -> IO[bytes] | None: - """Combined binary output pipe.""" + """Binary standard-output pipe when requested.""" + ... + + @property + def stderr(self) -> IO[bytes] | None: + """Binary standard-error pipe when requested separately.""" ... def kill(self) -> None: diff --git a/src/flext_cli/_protocols/_base_parts/flextcliprotocolsbase_part_03.py b/src/flext_cli/_protocols/_base_parts/flextcliprotocolsbase_part_03.py index 94be7a5f..2827a90c 100644 --- a/src/flext_cli/_protocols/_base_parts/flextcliprotocolsbase_part_03.py +++ b/src/flext_cli/_protocols/_base_parts/flextcliprotocolsbase_part_03.py @@ -30,6 +30,9 @@ def run( timeout: int | None = None, env: t.StrMapping | None = None, remove_env_keys: t.StrSequence = (), + input_data: str | bytes | None = None, + *, + capture: bool = True, ) -> p.Result[p.Cli.CommandOutput]: """Execute a command and require zero exit status.""" ... @@ -41,6 +44,7 @@ def capture( timeout: int | None = None, env: t.StrMapping | None = None, remove_env_keys: t.StrSequence = (), + input_data: str | bytes | None = None, ) -> p.Result[str]: """Execute a command and return stripped stdout.""" ... @@ -52,7 +56,9 @@ def run_raw( timeout: int | None = None, env: t.StrMapping | None = None, remove_env_keys: t.StrSequence = (), - input_data: bytes | None = None, + input_data: str | bytes | None = None, + *, + capture: bool = True, ) -> p.Result[p.Cli.CommandOutput]: """Execute a command without enforcing zero exit status.""" ... @@ -65,7 +71,7 @@ def run_bytes( timeout: int | None = None, env: t.StrMapping | None = None, remove_env_keys: t.StrSequence = (), - input_data: bytes | None = None, + input_data: str | bytes | None = None, ) -> p.Result[p.Cli.CommandBytesOutput]: """Execute a command and preserve byte-exact output.""" ... @@ -77,10 +83,25 @@ def run_checked( timeout: int | None = None, env: t.StrMapping | None = None, remove_env_keys: t.StrSequence = (), + input_data: str | bytes | None = None, + *, + capture: bool = True, ) -> p.Result[bool]: """Execute a command and return a success flag.""" ... + def run_live( + self, + cmd: t.StrSequence, + cwd: t.Cli.TextPath | None = None, + timeout: int | None = None, + env: t.StrMapping | None = None, + remove_env_keys: t.StrSequence = (), + input_data: str | bytes | None = None, + ) -> p.Result[p.Cli.CommandOutput]: + """Execute a checked command with inherited live output.""" + ... + def run_to_file( self, cmd: t.StrSequence, diff --git a/src/flext_cli/_utilities/_runtime_commands.py b/src/flext_cli/_utilities/_runtime_commands.py index 07980fcc..c8dddb1b 100644 --- a/src/flext_cli/_utilities/_runtime_commands.py +++ b/src/flext_cli/_utilities/_runtime_commands.py @@ -13,8 +13,9 @@ class FlextCliUtilitiesRuntimeCommandsMixin: if TYPE_CHECKING: - @staticmethod + @classmethod def run_raw( + cls, cmd: t.StrSequence, cwd: t.Cli.TextPath | None = None, timeout: int | None = None, diff --git a/src/flext_cli/_utilities/_runtime_process_cleanup.py b/src/flext_cli/_utilities/_runtime_process_cleanup.py index cd2f9cd0..bbf4781e 100644 --- a/src/flext_cli/_utilities/_runtime_process_cleanup.py +++ b/src/flext_cli/_utilities/_runtime_process_cleanup.py @@ -76,11 +76,10 @@ def _reap_and_drain( cls, process: p.Cli.ProcessHandle, waiter: threading.Thread, - pump: threading.Thread, process_done: threading.Event, wake: threading.Event, stop: threading.Event, - source: IO[bytes], + pump_streams: tuple[tuple[threading.Thread, IO[bytes]], ...], cleanup_errors: list[str], job_handle: int, absolute_deadline: float | None, @@ -98,7 +97,8 @@ def _reap_and_drain( waiter.join(cls._remaining(cleanup_deadline)) if waiter.is_alive(): cleanup_errors.append("process deadline expired before root reaping") - cls._drain_output(pump, stop, source, cleanup_errors, cleanup_deadline) + for pump, source in pump_streams: + cls._drain_output(pump, stop, source, cleanup_errors, cleanup_deadline) return return_codes[0] if return_codes else process.poll() @classmethod @@ -157,7 +157,7 @@ def _drain_output( try: source.close() except (OSError, ValueError) as exc: - cleanup_errors.append(f"combined output close error: {exc}") + cleanup_errors.append(f"process output close error: {exc}") pump.join(cls._remaining(cleanup_deadline)) if pump.is_alive(): cleanup_errors.append("process deadline expired before output drain") diff --git a/src/flext_cli/_utilities/_runtime_process_execution.py b/src/flext_cli/_utilities/_runtime_process_execution.py index 268425ee..e3459d0e 100644 --- a/src/flext_cli/_utilities/_runtime_process_execution.py +++ b/src/flext_cli/_utilities/_runtime_process_execution.py @@ -4,30 +4,39 @@ import contextlib import threading +import time from collections.abc import Callable from pathlib import Path from typing import IO, BinaryIO -from flext_cli import c, p, t +from flext_cli import c, p, r, t from flext_cli._utilities._runtime_process_cleanup import ( FlextCliUtilitiesRuntimeProcessCleanupMixin, ) from flext_cli._utilities._runtime_process_outcome import ( FlextCliUtilitiesRuntimeProcessOutcomeMixin, ) +from flext_cli._utilities._runtime_process_output import ( + FlextCliUtilitiesRuntimeProcessOutputMixin, +) from flext_cli._utilities._runtime_process_resources import ( FlextCliUtilitiesRuntimeProcessResourcesMixin, ) from flext_cli._utilities._runtime_process_start import ( FlextCliUtilitiesRuntimeProcessStartMixin, ) +from flext_cli._utilities._runtime_process_timing import ( + FlextCliUtilitiesRuntimeProcessTimingMixin, +) class FlextCliUtilitiesRuntimeProcessExecutionMixin( FlextCliUtilitiesRuntimeProcessCleanupMixin, FlextCliUtilitiesRuntimeProcessOutcomeMixin, + FlextCliUtilitiesRuntimeProcessOutputMixin, FlextCliUtilitiesRuntimeProcessResourcesMixin, FlextCliUtilitiesRuntimeProcessStartMixin, + FlextCliUtilitiesRuntimeProcessTimingMixin, ): """Own one child process and its streaming resources.""" @@ -35,23 +44,35 @@ class FlextCliUtilitiesRuntimeProcessExecutionMixin( def _execute_streamed_process( cls, cmd: t.StrSequence, - output_path: Path, + output_path: Path | None, cwd: t.Cli.TextPath | None, env: dict[str, str] | None, input_data: str | bytes | None, *, + capture_output: bool, live: bool, - absolute_deadline: float | None, - grace_seconds: float, - timeout_exit_code: int, - legacy_timeout: bool, - legacy_timeout_seconds: int | None, - ) -> p.Result[int]: + timeout: int | None, + deadline: p.Cli.ProcessDeadline | None, + ) -> p.Result[p.Cli.CommandBytesOutput]: """Own resources and complete one streamed child lifecycle.""" + started = time.monotonic() + timing_result = cls._resolve_process_timing( + cmd, + timeout, + deadline, + started, + capture_output=capture_output, + has_output_path=output_path is not None, + live=live, + on_main_thread=threading.current_thread() is threading.main_thread(), + ) + if timing_result.failure: + return r[p.Cli.CommandBytesOutput].fail( + timing_result.error or "process deadline resolution failed" + ) + absolute_deadline, grace_seconds, timeout_exit_code = timing_result.unwrap() process: p.Cli.ProcessHandle | None = None waiter: threading.Thread | None = None - pump: threading.Thread | None = None - source: IO[bytes] | None = None durable_log: BinaryIO | None = None job_handle = 0 failures: list[str] = [] @@ -61,6 +82,9 @@ def _execute_streamed_process( forwarded_signals: list[int] = [] received_signals: list[int] = [] return_codes: list[int] = [] + stdout_output = bytearray() + stderr_output = bytearray() + pump_streams: list[tuple[threading.Thread, IO[bytes]]] = [] pump_stop = threading.Event() process_done = threading.Event() wake = threading.Event() @@ -77,9 +101,7 @@ def execute_lifecycle() -> None: final_deadline, \ job_handle, \ process, \ - pump, \ return_code, \ - source, \ timed_out, \ waiter if threading.current_thread() is threading.main_thread(): @@ -92,8 +114,9 @@ def execute_lifecycle() -> None: if received_signals: wake.set() return - output_path.parent.mkdir(parents=True, exist_ok=True) - durable_log = stack.enter_context(output_path.open("wb", buffering=0)) + if output_path is not None: + output_path.parent.mkdir(parents=True, exist_ok=True) + durable_log = stack.enter_context(output_path.open("wb", buffering=0)) stdin_result = cls._prepare_streamed_stdin(stack, input_data) live_result = cls._prepare_live_descriptor(stack, live=live) if stdin_result.failure: @@ -105,32 +128,41 @@ def execute_lifecycle() -> None: elif cls._spawn_deadline_exhausted(absolute_deadline, grace_seconds): failures.append("process deadline exhausted before child spawn") else: - started = cls._start_contained_process( - prepared_cmd, cwd, env, stdin_result.value[0] + combine_output = output_path is not None + pipe_output = combine_output or capture_output + start_result = cls._start_contained_process( + prepared_cmd, + cwd, + env, + stdin_result.value[0], + capture_output=pipe_output, + combine_output=combine_output, ) - if started.failure: - failures.append(started.error or "process start failed") + if start_result.failure: + failures.append(start_result.error or "process start failed") else: - process, job_handle = started.unwrap() - source = process.stdout - if source is None: - failures.append("process stdout is not available") - return - stack.callback(source.close) + owned_process, job_handle = start_result.unwrap() + process = owned_process waiter = cls._start_root_waiter( - process, return_codes, failures, process_done, wake + owned_process, return_codes, failures, process_done, wake ) - pump = cls._start_output_pump( - source, - durable_log, - live_result.value[0], - failures, - live_diagnostics, - pump_stop, - wake, + pump_streams.extend( + cls._start_process_output( + owned_process, + stack, + durable_log, + live_result.value[0], + failures, + live_diagnostics, + pump_stop, + wake, + stdout_output, + stderr_output, + capture_output=capture_output, + ) ) timed_out, final_deadline = cls._monitor_process( - process, + owned_process, process_done, wake, failures, @@ -140,13 +172,12 @@ def execute_lifecycle() -> None: grace_seconds, ) return_code = cls._reap_and_drain( - process, + owned_process, waiter, - pump, process_done, wake, pump_stop, - source, + tuple(pump_streams), cleanup_errors, job_handle, final_deadline, @@ -159,21 +190,14 @@ def execute_lifecycle() -> None: except c.EXC_OS_VALUE as exc: failures.append(f"execution error: {exc}") finally: - if ( - process is not None - and waiter is not None - and pump is not None - and source is not None - and not cleanup_complete - ): + if process is not None and waiter is not None and not cleanup_complete: return_code = cls._reap_and_drain( process, waiter, - pump, process_done, wake, pump_stop, - source, + tuple(pump_streams), cleanup_errors, job_handle, final_deadline, @@ -188,15 +212,16 @@ def execute_lifecycle() -> None: cleanup_errors.append(close_error) cleanup_errors.extend(cls._close_process_resources(stack)) cleanup_errors.extend(cls._restore_forwarding_handlers(restore_handlers)) - return cls._process_exit_result( + return cls._captured_process_result( cmd, return_code, received_signals, (*failures, *cleanup_errors), - nonfatal_diagnostics=tuple(live_diagnostics), + stdout_output, + stderr_output, + max(0.0, time.monotonic() - started), timed_out=timed_out, - legacy_timeout=legacy_timeout, - legacy_timeout_seconds=legacy_timeout_seconds, + timeout_seconds=timeout, timeout_exit_code=timeout_exit_code, ) diff --git a/src/flext_cli/_utilities/_runtime_process_outcome.py b/src/flext_cli/_utilities/_runtime_process_outcome.py index 3399da8f..1012877a 100644 --- a/src/flext_cli/_utilities/_runtime_process_outcome.py +++ b/src/flext_cli/_utilities/_runtime_process_outcome.py @@ -4,7 +4,7 @@ import shlex -from flext_cli import p, r, t +from flext_cli import m, p, r, t class FlextCliUtilitiesRuntimeProcessOutcomeMixin: @@ -44,5 +44,39 @@ def _process_exit_result( return r[int].fail("root process did not expose an exit status") return r[int].ok(primary_exit) + @classmethod + def _captured_process_result( + cls, + cmd: t.StrSequence, + return_code: int | None, + received_signals: list[int], + diagnostics: tuple[str, ...], + stdout_output: bytearray, + stderr_output: bytearray, + duration: float, + *, + timed_out: bool, + timeout_seconds: int | None, + timeout_exit_code: int, + ) -> p.Result[p.Cli.CommandBytesOutput]: + """Attach captured bytes only after the owned process boundary is empty.""" + return cls._process_exit_result( + cmd, + return_code, + received_signals, + diagnostics, + timed_out=timed_out, + legacy_timeout=timeout_seconds is not None, + legacy_timeout_seconds=timeout_seconds, + timeout_exit_code=timeout_exit_code, + ).map( + lambda exit_code: m.Cli.CommandBytesOutput( + stdout=bytes(stdout_output), + stderr=bytes(stderr_output), + exit_code=exit_code, + duration=duration, + ) + ) + __all__: list[str] = ["FlextCliUtilitiesRuntimeProcessOutcomeMixin"] diff --git a/src/flext_cli/_utilities/_runtime_process_output.py b/src/flext_cli/_utilities/_runtime_process_output.py new file mode 100644 index 00000000..daa0ccb7 --- /dev/null +++ b/src/flext_cli/_utilities/_runtime_process_output.py @@ -0,0 +1,80 @@ +"""Output-pipe ownership for the canonical contained process lifecycle.""" + +from __future__ import annotations + +import contextlib +import threading +from typing import IO, BinaryIO + +from flext_cli import p +from flext_cli._utilities._runtime_process_threads import ( + FlextCliUtilitiesRuntimeProcessThreadsMixin, +) + + +class FlextCliUtilitiesRuntimeProcessOutputMixin( + FlextCliUtilitiesRuntimeProcessThreadsMixin +): + """Attach every requested child pipe to exactly one bounded pump.""" + + @classmethod + def _start_process_output( + cls, + process: p.Cli.ProcessHandle, + stack: contextlib.ExitStack, + durable_log: BinaryIO | None, + live_fd: int | None, + failures: list[str], + live_diagnostics: list[str], + stop: threading.Event, + wake: threading.Event, + stdout_output: bytearray, + stderr_output: bytearray, + *, + capture_output: bool, + ) -> tuple[tuple[threading.Thread, IO[bytes]], ...]: + combine_output = durable_log is not None + pipe_output = combine_output or capture_output + pump_streams: list[tuple[threading.Thread, IO[bytes]]] = [] + stdout_source = process.stdout + if pipe_output and stdout_source is None: + failures.append("process stdout is not available") + elif stdout_source is not None: + stack.callback(stdout_source.close) + stdout_pump = cls._start_output_pump( + stdout_source, + durable_log, + stdout_output if capture_output else None, + live_fd, + failures, + live_diagnostics, + stop, + wake, + thread_name=( + "flext-cli-process-output" + if combine_output + else "flext-cli-process-stdout" + ), + ) + pump_streams.append((stdout_pump, stdout_source)) + stderr_source = process.stderr + if capture_output and stderr_source is None: + failures.append("process stderr is not available") + elif stderr_source is not None: + stack.callback(stderr_source.close) + stderr_pump = cls._start_output_pump( + stderr_source, + None, + stderr_output, + None, + failures, + live_diagnostics, + stop, + wake, + thread_name="flext-cli-process-stderr", + ) + pump_streams.append((stderr_pump, stderr_source)) + return tuple(pump_streams) + + +__all__: list[str] = ["FlextCliUtilitiesRuntimeProcessOutputMixin"] diff --git a/src/flext_cli/_utilities/_runtime_process_start.py b/src/flext_cli/_utilities/_runtime_process_start.py index a23951cf..26323dfd 100644 --- a/src/flext_cli/_utilities/_runtime_process_start.py +++ b/src/flext_cli/_utilities/_runtime_process_start.py @@ -20,6 +20,8 @@ def _spawn_streamed_process( env: dict[str, str] | None, stdin_handle: BinaryIO | None, *, + capture_output: bool, + combine_output: bool, creation_flags: int, ) -> p.Cli.ProcessHandle: ... @@ -52,9 +54,18 @@ def _start_contained_process( cwd: t.Cli.TextPath | None, env: dict[str, str] | None, stdin_handle: BinaryIO | None, + *, + capture_output: bool, + combine_output: bool, ) -> p.Result[tuple[p.Cli.ProcessHandle, int]]: process = cls._spawn_streamed_process( - cmd, cwd, env, stdin_handle, creation_flags=cls._streamed_creation_flags() + cmd, + cwd, + env, + stdin_handle, + capture_output=capture_output, + combine_output=combine_output, + creation_flags=cls._streamed_creation_flags(), ) job_result = cls._windows_job_create(process.pid) if job_result.failure: diff --git a/src/flext_cli/_utilities/_runtime_process_stream.py b/src/flext_cli/_utilities/_runtime_process_stream.py index d5684cce..fe1ddcd5 100644 --- a/src/flext_cli/_utilities/_runtime_process_stream.py +++ b/src/flext_cli/_utilities/_runtime_process_stream.py @@ -8,7 +8,7 @@ class FlextCliUtilitiesRuntimeProcessStreamMixin: - """Mirror combined child output to a live descriptor and durable log.""" + """Route child bytes to captured, durable, and live output owners.""" _STREAM_CHUNK_BYTES: ClassVar[int] = 64 * 1024 _STREAM_POLL_SECONDS: ClassVar[float] = 0.01 @@ -17,24 +17,28 @@ class FlextCliUtilitiesRuntimeProcessStreamMixin: def _pump_process_output( cls, source: IO[bytes], - durable_log: BinaryIO, + durable_log: BinaryIO | None, + captured_output: bytearray | None, live_fd: int | None, failures: list[str], live_diagnostics: list[str], stop: threading.Event, wake: threading.Event, ) -> None: - """Persist each chunk before bounded best-effort live mirroring.""" + """Own one child pipe until EOF and preserve each byte exactly once.""" live_available = live_fd is not None try: while not stop.is_set(): chunk = cls._read_process_chunk(source, failures) if chunk is None: return - durable_error = cls._write_durable_chunk(durable_log, chunk) - if durable_error is not None: - failures.append(durable_error) - return + if durable_log is not None: + durable_error = cls._write_durable_chunk(durable_log, chunk) + if durable_error is not None: + failures.append(durable_error) + return + if captured_output is not None: + captured_output.extend(chunk) if live_available and live_fd is not None: live_available = cls._write_live_chunk( live_fd, chunk, stop, live_diagnostics diff --git a/src/flext_cli/_utilities/_runtime_process_threads.py b/src/flext_cli/_utilities/_runtime_process_threads.py index 7add437b..6be569b4 100644 --- a/src/flext_cli/_utilities/_runtime_process_threads.py +++ b/src/flext_cli/_utilities/_runtime_process_threads.py @@ -41,17 +41,29 @@ def _start_root_waiter( def _start_output_pump( cls, source: IO[bytes], - durable_log: BinaryIO, + durable_log: BinaryIO | None, + captured_output: bytearray | None, live_fd: int | None, failures: list[str], live_diagnostics: list[str], stop: threading.Event, wake: threading.Event, + *, + thread_name: str, ) -> threading.Thread: pump = threading.Thread( target=cls._pump_process_output, - args=(source, durable_log, live_fd, failures, live_diagnostics, stop, wake), - name="flext-cli-process-output", + args=( + source, + durable_log, + captured_output, + live_fd, + failures, + live_diagnostics, + stop, + wake, + ), + name=thread_name, daemon=False, ) pump.start() diff --git a/src/flext_cli/_utilities/_runtime_process_timing.py b/src/flext_cli/_utilities/_runtime_process_timing.py new file mode 100644 index 00000000..028e9479 --- /dev/null +++ b/src/flext_cli/_utilities/_runtime_process_timing.py @@ -0,0 +1,68 @@ +"""Deadline normalization for the canonical contained process lifecycle.""" + +from __future__ import annotations + +import shlex + +from flext_cli import p, r, t + + +class FlextCliUtilitiesRuntimeProcessTimingMixin: + """Resolve relative and absolute deadlines through one policy owner.""" + + @staticmethod + def _resolve_process_timing( + cmd: t.StrSequence, + timeout: int | None, + deadline: p.Cli.ProcessDeadline | None, + started: float, + *, + capture_output: bool, + has_output_path: bool, + live: bool, + on_main_thread: bool, + ) -> p.Result[tuple[float | None, float, int]]: + if timeout is not None and deadline is not None: + return r[tuple[float | None, float, int]].fail( + "timeout and deadline are mutually exclusive" + ) + if live and not has_output_path: + return r[tuple[float | None, float, int]].fail( + "live output requires a durable output path" + ) + if capture_output and has_output_path: + return r[tuple[float | None, float, int]].fail( + "captured and durable output are mutually exclusive" + ) + if (live or deadline is not None) and not on_main_thread: + return r[tuple[float | None, float, int]].fail( + "live/deadline process execution requires the main interpreter thread" + ) + absolute_deadline: float | None = None + grace_seconds = 0.0 + timeout_exit_code = 124 + if deadline is not None: + absolute_deadline = deadline.expires_at_monotonic + grace_seconds = deadline.termination_grace_seconds + timeout_exit_code = deadline.timeout_exit_code + elif timeout is not None: + if timeout <= 0: + return r[tuple[float | None, float, int]].fail( + f"timeout {timeout}s: {shlex.join(list(cmd))}" + ) + absolute_deadline = started + timeout + grace_seconds = min(max(timeout * 0.1, 0.05), timeout * 0.5) + if absolute_deadline is not None: + remaining = absolute_deadline - started + if remaining <= 0 or grace_seconds <= 0 or grace_seconds >= remaining: + return r[tuple[float | None, float, int]].fail( + "process deadline must leave a positive grace reserve" + ) + return r[tuple[float | None, float, int]].ok(( + absolute_deadline, + grace_seconds, + timeout_exit_code, + )) + + +__all__: list[str] = ["FlextCliUtilitiesRuntimeProcessTimingMixin"] diff --git a/src/flext_cli/_utilities/_runtime_run_to_file.py b/src/flext_cli/_utilities/_runtime_run_to_file.py index 53d10e89..ccac80ec 100644 --- a/src/flext_cli/_utilities/_runtime_run_to_file.py +++ b/src/flext_cli/_utilities/_runtime_run_to_file.py @@ -2,13 +2,10 @@ from __future__ import annotations -import shlex -import threading -import time from pathlib import Path from typing import TYPE_CHECKING -from flext_cli import p, r, t +from flext_cli import p, t from flext_cli._utilities._runtime_process_execution import ( FlextCliUtilitiesRuntimeProcessExecutionMixin, ) @@ -49,47 +46,17 @@ def run_to_file( An outer caller wall remains responsible for an OS syscall that becomes uninterruptible. """ - if timeout is not None and deadline is not None: - return r[int].fail("timeout and deadline are mutually exclusive") - if (live or deadline is not None) and ( - threading.current_thread() is not threading.main_thread() - ): - return r[int].fail( - "live/deadline process execution requires the main interpreter thread" - ) - started = time.monotonic() - absolute_deadline: float | None = None - grace_seconds = 0.0 - timeout_exit_code = 124 - legacy_timeout = timeout is not None - if deadline is not None: - absolute_deadline = deadline.expires_at_monotonic - grace_seconds = deadline.termination_grace_seconds - timeout_exit_code = deadline.timeout_exit_code - elif timeout is not None: - if timeout <= 0: - return r[int].fail(f"timeout {timeout}s: {shlex.join(list(cmd))}") - absolute_deadline = started + timeout - grace_seconds = min(max(timeout * 0.1, 0.05), timeout * 0.5) - if absolute_deadline is not None: - remaining = absolute_deadline - started - if remaining <= 0 or grace_seconds <= 0 or grace_seconds >= remaining: - return r[int].fail( - "process deadline must leave a positive grace reserve" - ) return cls._execute_streamed_process( cmd, Path(output_file), cwd, cls._resolved_env(env, remove_env_keys), input_data, + capture_output=False, live=live, - absolute_deadline=absolute_deadline, - grace_seconds=grace_seconds, - timeout_exit_code=timeout_exit_code, - legacy_timeout=legacy_timeout, - legacy_timeout_seconds=timeout, - ) + timeout=timeout, + deadline=deadline, + ).map(lambda output: output.exit_code) __all__: list[str] = ["FlextCliUtilitiesRuntimeRunToFileMixin"] diff --git a/src/flext_cli/_utilities/runtime.py b/src/flext_cli/_utilities/runtime.py index e2edd41a..4e6adec8 100644 --- a/src/flext_cli/_utilities/runtime.py +++ b/src/flext_cli/_utilities/runtime.py @@ -5,10 +5,9 @@ import os import shlex import subprocess -import time from typing import BinaryIO, ClassVar, override -from flext_cli import c, m, p, r, t +from flext_cli import m, p, r, t from flext_cli._utilities._runtime_commands import FlextCliUtilitiesRuntimeCommandsMixin from flext_cli._utilities._runtime_run_to_file import ( FlextCliUtilitiesRuntimeRunToFileMixin, @@ -64,6 +63,8 @@ def _spawn_streamed_process( env: dict[str, str] | None, stdin_handle: BinaryIO | None, *, + capture_output: bool, + combine_output: bool, creation_flags: int, ) -> p.Cli.ProcessHandle: """Create the sole raw child owned by the streamed lifecycle.""" @@ -71,8 +72,14 @@ def _spawn_streamed_process( list(cmd), cwd=cwd, stdin=subprocess.DEVNULL if stdin_handle is None else stdin_handle, - stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, + stdout=subprocess.PIPE if capture_output else None, + stderr=( + subprocess.STDOUT + if combine_output + else subprocess.PIPE + if capture_output + else None + ), text=False, bufsize=0, env=env, @@ -90,9 +97,10 @@ def _streamed_creation_flags() -> int: getattr(subprocess, "CREATE_SUSPENDED", 0x00000004) ) - @staticmethod + @classmethod @override def run_raw( + cls, cmd: t.StrSequence, cwd: t.Cli.TextPath | None = None, timeout: int | None = None, @@ -112,46 +120,41 @@ def run_raw( (for long-running makes/rollouts); the returned stdout/stderr are then empty and only the exit code is meaningful. """ - start = time.monotonic() - stdin = ( - input_data.encode("utf-8") if isinstance(input_data, str) else input_data - ) - try: - result = subprocess.run( - list(cmd), - cwd=cwd, - capture_output=capture, - text=False, - check=False, - timeout=timeout, - env=FlextCliUtilitiesRuntime._resolved_env(env, remove_env_keys), - input=stdin, - ) - except subprocess.TimeoutExpired as exc: - return r[p.Cli.CommandOutput].fail( - f"timeout {exc.timeout}s: {shlex.join(list(cmd))}" - ) - except c.EXC_OS_VALUE as exc: - return r[p.Cli.CommandOutput].fail(f"execution error: {exc}") - try: - stdout = (result.stdout or b"").decode("utf-8") - stderr = (result.stderr or b"").decode("utf-8") - except UnicodeDecodeError as exc: - return r[p.Cli.CommandOutput].fail( - f"non-UTF-8 output from {shlex.join(list(cmd))}: {exc}" - ) - duration = max(0.0, time.monotonic() - start) - return r[p.Cli.CommandOutput].ok( - m.Cli.CommandOutput( - stdout=stdout, - stderr=stderr, - exit_code=result.returncode, - duration=duration, + + def decode_output( + output: p.Cli.CommandBytesOutput, + ) -> p.Result[p.Cli.CommandOutput]: + try: + stdout = output.stdout.decode("utf-8") + stderr = output.stderr.decode("utf-8") + except UnicodeDecodeError as exc: + return r[p.Cli.CommandOutput].fail( + f"non-UTF-8 output from {shlex.join(list(cmd))}: {exc}" + ) + return r[p.Cli.CommandOutput].ok( + m.Cli.CommandOutput( + stdout=stdout, + stderr=stderr, + exit_code=output.exit_code, + duration=output.duration, + ) ) - ) - @staticmethod + return cls._execute_streamed_process( + cmd, + None, + cwd, + cls._resolved_env(env, remove_env_keys), + input_data, + capture_output=capture, + live=False, + timeout=timeout, + deadline=None, + ).flat_map(decode_output) + + @classmethod def run_bytes( + cls, cmd: t.StrSequence, cwd: t.Cli.TextPath | None = None, timeout: int | None = None, @@ -160,35 +163,16 @@ def run_bytes( input_data: str | bytes | None = None, ) -> p.Result[p.Cli.CommandBytesOutput]: """Run a command capturing byte-exact stdout/stderr (no text decoding).""" - start = time.monotonic() - stdin = ( - input_data.encode("utf-8") if isinstance(input_data, str) else input_data - ) - try: - result = subprocess.run( - list(cmd), - cwd=cwd, - capture_output=True, - text=False, - check=False, - timeout=timeout, - env=FlextCliUtilitiesRuntime._resolved_env(env, remove_env_keys), - input=stdin, - ) - except subprocess.TimeoutExpired as exc: - return r[p.Cli.CommandBytesOutput].fail( - f"timeout {exc.timeout}s: {shlex.join(list(cmd))}" - ) - except c.EXC_OS_VALUE as exc: - return r[p.Cli.CommandBytesOutput].fail(f"execution error: {exc}") - duration = max(0.0, time.monotonic() - start) - return r[p.Cli.CommandBytesOutput].ok( - m.Cli.CommandBytesOutput( - stdout=result.stdout or b"", - stderr=result.stderr or b"", - exit_code=result.returncode, - duration=duration, - ) + return cls._execute_streamed_process( + cmd, + None, + cwd, + cls._resolved_env(env, remove_env_keys), + input_data, + capture_output=True, + live=False, + timeout=timeout, + deadline=None, ) diff --git a/tests/unit/test_runtime_process_containment.py b/tests/unit/test_runtime_process_containment.py index ffe4d28b..f8016e6d 100644 --- a/tests/unit/test_runtime_process_containment.py +++ b/tests/unit/test_runtime_process_containment.py @@ -8,13 +8,13 @@ import threading import time from collections import UserList -from collections.abc import Iterator +from collections.abc import Callable, Iterator from typing import TYPE_CHECKING, override import pytest from flext_tests import tm -from tests import m, u +from tests import m, p, u if TYPE_CHECKING: from pathlib import Path @@ -30,49 +30,128 @@ def _deadline( ) -def _process_exists(process_id: int) -> bool: - try: - os.kill(process_id, 0) - except OSError: - return False - return True +def _survivor_acknowledged(probe: Path, acknowledgement: Path) -> bool: + """Ask an owned descendant to prove it is still able to execute.""" + probe.touch() + acknowledgement_deadline = time.monotonic() + 0.5 + while not acknowledgement.exists() and time.monotonic() < acknowledgement_deadline: + time.sleep(0.01) + return acknowledgement.exists() + + +def _assert_owned_descendant_stopped( + process_info: Path, probe: Path, acknowledgement: Path +) -> None: + """Prove no owned descendant can execute and clean an observed failure.""" + child_survived = _survivor_acknowledged(probe, acknowledgement) + if child_survived: + os.kill(int(process_info.read_text(encoding="utf-8")), signal.SIGTERM) + tm.that(child_survived, eq=False) + + +def _assert_timeout_empties_descendants[ + Output: (p.Cli.CommandOutput | p.Cli.CommandBytesOutput) +](tmp_path: Path, execute: Callable[[tuple[str, ...]], p.Result[Output]]) -> None: + process_info = tmp_path / "captured-process-info" + survivor_probe = tmp_path / "captured-survivor-probe" + survivor_ack = tmp_path / "captured-survivor-ack" + child = ( + "import os,pathlib,sys,time;" + "info=pathlib.Path(sys.argv[1]);probe=pathlib.Path(sys.argv[2]);" + "ack=pathlib.Path(sys.argv[3]);info.write_text(str(os.getpid()));" + "\nwhile True:\n" + " if probe.exists(): ack.touch()\n" + " time.sleep(.01)" + ) + parent = ( + "import pathlib,subprocess,sys,time;" + f"subprocess.Popen([sys.executable,'-c',{child!r}," + "sys.argv[1],sys.argv[2],sys.argv[3]],stdout=subprocess.DEVNULL," + "stderr=subprocess.DEVNULL);" + "info=pathlib.Path(sys.argv[1]);" + "\nwhile not info.exists():\n time.sleep(.01)\n" + "time.sleep(30)" + ) + + result = execute(( + sys.executable, + "-c", + parent, + str(process_info), + str(survivor_probe), + str(survivor_ack), + )) + + tm.fail(result, has="timeout") + _assert_owned_descendant_stopped(process_info, survivor_probe, survivor_ack) class _InterruptingCommand(UserList[str]): @override def __iter__(self) -> Iterator[str]: - os.kill(os.getpid(), signal.SIGTERM) + signal.raise_signal(signal.SIGTERM) return super().__iter__() class TestsFlextCliRuntimeProcessContainment: """Prove pre-spawn signals, deadline escalation, and empty boundaries.""" - @pytest.mark.skipif(os.name == "nt", reason="POSIX process-group contract") - def test_return_proves_owned_process_group_empty(self, tmp_path: Path) -> None: - boundary_file = tmp_path / "boundary" - nested = "import os,time;os.close(1);os.close(2);time.sleep(30)" + def test_run_raw_timeout_leaves_no_descendant(self, tmp_path: Path) -> None: + """Return only after the captured text runner empties its owned boundary.""" + _assert_timeout_empties_descendants( + tmp_path, lambda command: u.Cli().run_raw(command, timeout=1) + ) + + def test_run_timeout_leaves_no_descendant(self, tmp_path: Path) -> None: + """Return only after the checked text runner empties its owned boundary.""" + _assert_timeout_empties_descendants( + tmp_path, lambda command: u.Cli().run(command, timeout=1) + ) + + def test_run_bytes_timeout_leaves_no_descendant(self, tmp_path: Path) -> None: + """Return only after the byte runner empties its owned boundary.""" + _assert_timeout_empties_descendants( + tmp_path, lambda command: u.Cli().run_bytes(command, timeout=1) + ) + + def test_return_proves_owned_process_boundary_empty(self, tmp_path: Path) -> None: + process_info = tmp_path / "boundary-process-info" + survivor_probe = tmp_path / "boundary-survivor-probe" + survivor_ack = tmp_path / "boundary-survivor-ack" + nested = ( + "import os,pathlib,sys,time;" + "info=pathlib.Path(sys.argv[1]);probe=pathlib.Path(sys.argv[2]);" + "ack=pathlib.Path(sys.argv[3]);info.write_text(str(os.getpid()));" + "\nwhile True:\n" + " if probe.exists(): ack.touch()\n" + " time.sleep(.01)" + ) root = ( - "import os,pathlib,subprocess,sys;" - f"child=subprocess.Popen([sys.executable,'-c',{nested!r}]);" - "pathlib.Path(sys.argv[1]).write_text(f'{os.getpgrp()} {child.pid}')" + "import pathlib,subprocess,sys,time;" + f"subprocess.Popen([sys.executable,'-c',{nested!r}," + "sys.argv[1],sys.argv[2],sys.argv[3]]," + "stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL);" + "info=pathlib.Path(sys.argv[1]);" + "\nwhile not info.exists():\n time.sleep(.01)" ) result = u.Cli().run_to_file( - [sys.executable, "-c", root, str(boundary_file)], + [ + sys.executable, + "-c", + root, + str(process_info), + str(survivor_probe), + str(survivor_ack), + ], tmp_path / "boundary.log", deadline=_deadline(seconds=2.0, grace=0.8, exit_code=94), ) - process_group, _child = ( - int(value) for value in boundary_file.read_text().split() - ) tm.ok(result) tm.that(result.value, eq=0) - with pytest.raises(ProcessLookupError): - os.killpg(process_group, 0) + _assert_owned_descendant_stopped(process_info, survivor_probe, survivor_ack) - @pytest.mark.skipif(os.name == "nt", reason="POSIX operator-signal contract") @pytest.mark.parametrize("signal_number", [signal.SIGINT, signal.SIGTERM]) def test_manual_signal_is_forwarded_and_normalized( self, tmp_path: Path, signal_number: signal.Signals @@ -92,7 +171,7 @@ def signal_when_ready() -> None: while not ready.exists() and time.monotonic() < ready_deadline: time.sleep(0.005) if ready.exists(): - os.kill(os.getpid(), signal_number) + signal.raise_signal(signal_number) else: signaler_errors.append("child did not become ready") @@ -112,7 +191,6 @@ def signal_when_ready() -> None: tm.that(signaler.is_alive(), eq=False) tm.that(time.monotonic() - signal_started, lt=3.0) - @pytest.mark.skipif(os.name == "nt", reason="POSIX operator-signal contract") def test_pre_spawn_signal_is_captured_before_command_materialization( self, tmp_path: Path ) -> None: @@ -158,44 +236,44 @@ def test_deadline_forwards_interrupt_before_forced_cleanup( def test_deadline_kills_recursive_process_tree(self, tmp_path: Path) -> None: output_file = tmp_path / "tree.log" - heartbeat = tmp_path / "heartbeat" process_info = tmp_path / "process-info" + survivor_probe = tmp_path / "survivor-probe" + survivor_ack = tmp_path / "survivor-ack" child = ( "import os,pathlib,signal,sys,time;" "signal.signal(signal.SIGINT,signal.SIG_IGN);" - "path=pathlib.Path(sys.argv[1]);" - "group=getattr(os,'getpgrp',lambda:0)();" - "pathlib.Path(sys.argv[2]).write_text(f'{os.getpid()} {group}');" - "\nwhile True:\n path.write_text(str(time.monotonic()));time.sleep(.02)" + "info=pathlib.Path(sys.argv[1]);probe=pathlib.Path(sys.argv[2]);" + "ack=pathlib.Path(sys.argv[3]);info.write_text(str(os.getpid()));" + "\nwhile True:\n" + " if probe.exists(): ack.touch()\n" + " time.sleep(.01)" ) parent = ( "import signal,subprocess,sys,time;" "signal.signal(signal.SIGINT,signal.SIG_IGN);" f"subprocess.Popen([sys.executable,'-c',{child!r}," - "sys.argv[1],sys.argv[2]]);" + "sys.argv[1],sys.argv[2],sys.argv[3]]);" "time.sleep(30)" ) started = time.monotonic() result = u.Cli().run_to_file( - [sys.executable, "-c", parent, str(heartbeat), str(process_info)], + [ + sys.executable, + "-c", + parent, + str(process_info), + str(survivor_probe), + str(survivor_ack), + ], output_file, deadline=_deadline(seconds=1.5, grace=0.7, exit_code=92), ) tm.ok(result) tm.that(result.value, eq=92) - tm.that(heartbeat.exists(), eq=True) - child_pid, process_group = ( - int(value) for value in process_info.read_text().split() - ) - stopped_value = heartbeat.stat().st_mtime_ns - time.sleep(0.15) - tm.that(heartbeat.stat().st_mtime_ns, eq=stopped_value) - tm.that(_process_exists(child_pid), eq=False) - if os.name != "nt": - with pytest.raises(ProcessLookupError): - os.killpg(process_group, 0) + tm.that(process_info.exists(), eq=True) + _assert_owned_descendant_stopped(process_info, survivor_probe, survivor_ack) tm.that(time.monotonic() - started, lt=2.0)