src/speech_text_data_aligners/runtime/process.py

# License: Apache-2.0 License
# Created by: Patrick Lumbantobing, VertoX-AI
# Copyright (c) 2026 VertoX-AI. All rights reserved.
#
# This work is licensed under the Apache-2.0 License.
# To view a copy of this license, visit
# https://www.apache.org/licenses/LICENSE-2.0

"""Run native commands through an injectable, timeout-aware, shell-free boundary."""

from __future__ import annotations

import os
import subprocess
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Protocol

from speech_text_data_aligners.core.errors import (
    BackendExecutionError,
    BackendUnavailableError,
)


@dataclass(frozen=True, slots=True)
class Command:
    """Describe a subprocess invocation without shell interpretation.

    Parameters
    ----------
        argv: Non-empty executable and argument tuple.
        cwd: Optional working directory.
        environment: Environment entries layered over the current process.
        timeout_s: Optional positive timeout in seconds.
        stdin_text: Optional UTF-8 text supplied on standard input.
        check: Raise a typed error when the process exits non-zero.

    """

    argv: tuple[str, ...]
    cwd: Path | None = None
    environment: tuple[tuple[str, str], ...] = ()
    timeout_s: float | None = None
    stdin_text: str | None = None
    check: bool = True


@dataclass(frozen=True, slots=True)
class CommandResult:
    """Capture one completed process's text streams, status, and elapsed time.

    Parameters
    ----------
        argv: Exact argument vector invoked.
        returncode: Native process exit status.
        stdout: Captured decoded standard output.
        stderr: Captured decoded standard error.
        duration_s: Non-negative monotonic elapsed duration.

    """

    argv: tuple[str, ...]
    returncode: int
    stdout: str
    stderr: str
    duration_s: float


class ProcessRunner(Protocol):
    """Abstract native process execution for injectable backend testing."""

    def run(self, command: Command) -> CommandResult:
        """Execute one command and return captured results.

        Parameters
        ----------
            command: Shell-free command description.

        Returns
        -------
            Captured successful or explicitly unchecked process result.

        Raises
        ------
            BackendUnavailableError: If the executable cannot be started.
            BackendExecutionError: If execution times out or a checked command fails.

        """
        ...


class SubprocessRunner:
    """Execute commands via :mod:`subprocess` with no shell or inherited stdin."""

    def run(self, command: Command) -> CommandResult:
        """Execute one validated command and translate expected OS/process failures.

        Parameters
        ----------
            command: Shell-free command description.

        Returns
        -------
            Captured result, including non-zero status when ``check`` is false.

        Raises
        ------
            BackendUnavailableError: If the executable is missing or inaccessible.
            BackendExecutionError: If the command is invalid, times out, or exits
                non-zero while ``check`` is true.

        """
        if not command.argv or not command.argv[0]:
            raise BackendExecutionError("command argument vector cannot be empty")
        if command.timeout_s is not None and command.timeout_s <= 0:
            raise BackendExecutionError("command timeout must be positive")
        environment = os.environ.copy()
        environment.update(command.environment)
        started = time.monotonic()
        try:
            completed = subprocess.run(
                command.argv,
                cwd=command.cwd,
                env=environment,
                input=command.stdin_text,
                capture_output=True,
                text=True,
                encoding="utf-8",
                errors="replace",
                timeout=command.timeout_s,
                check=False,
                shell=False,
            )
        except FileNotFoundError as error:
            raise BackendUnavailableError(
                f"backend executable is unavailable: {command.argv[0]}"
            ) from error
        except OSError as error:
            raise BackendUnavailableError(
                f"backend executable could not be started: {command.argv[0]}"
            ) from error
        except subprocess.TimeoutExpired as error:
            raise BackendExecutionError(
                f"backend command timed out after {command.timeout_s} seconds",
                retryable=True,
            ) from error
        duration = time.monotonic() - started
        result = CommandResult(
            argv=command.argv,
            returncode=completed.returncode,
            stdout=completed.stdout,
            stderr=completed.stderr,
            duration_s=duration,
        )
        if command.check and result.returncode != 0:
            raise BackendExecutionError(
                f"backend command exited with status {result.returncode}",
                context=(
                    ("returncode", result.returncode),
                    ("stdout", result.stdout[-4000:]),
                    ("stderr", result.stderr[-4000:]),
                ),
            )
        return result