lifeloop/router/

subprocess.rs

//! Subprocess-backed callback invocation (issue #8).
//!
//! Fills the [`CallbackInvoker`] seam declared in `src/router/seams.rs`
//! (issue #7) for the **process-boundary** transport. This module is
//! the load-bearing proof that Lifeloop can drive an external client
//! command without taking a Rust dependency on it: the only contract
//! between Lifeloop and the client is JSON over stdio.
//!
//! # Boundary
//!
//! Owns:
//! * [`SubprocessInvokerConfig`] — configurable command/args/timeout
//!   used to spawn the external client per invocation;
//! * [`SubprocessCallbackInvoker`] — concrete [`CallbackInvoker`]
//!   implementation that spawns a fresh child for each event, pipes a
//!   [`CallbackRequest`] JSON document to its stdin, reads a
//!   [`CallbackResponse`] JSON document from its stdout, and validates
//!   it;
//! * [`SubprocessInvokerError`] — typed error variants carrying enough
//!   detail for [`super::LifeloopFailureMapper`] to map onto the
//!   shared [`crate::FailureClass`] vocabulary;
//! * [`failure_class_for_subprocess_error`] / `From<&SubprocessInvokerError>
//!   for FailureClass` — deterministic mapping into the existing
//!   failure-class vocabulary. No new variants are introduced.
//!
//! Does **not** own:
//! * receipt synthesis (that is [`super::receipts`]);
//! * wire schema (the request/response types come from the lifeloop
//!   crate root and are unchanged by this module).
//!
//! # Receipt-emitted guard
//!
//! `receipt.emitted` is a notification event and must not be invoked
//! downstream. The subprocess invoker rejects such plans **before** it
//! spawns a child — same rule as the in-memory path, enforced via
//! [`super::validate_receipt_eligible`]. The rejection surfaces as
//! [`SubprocessInvokerError::ReceiptEmittedRejected`] which maps to
//! [`crate::FailureClass::InvalidRequest`].

use std::ffi::OsString;
use std::io::{Read, Write};
use std::path::PathBuf;
use std::process::{Child, Command, Stdio};
use std::sync::mpsc;
use std::thread;
use std::time::{Duration, Instant};

use crate::{CallbackResponse, DispatchEnvelope, FailureClass, PayloadEnvelope, ValidationError};

use super::failure_mapping::{TransportError, validate_receipt_eligible};
use super::plan::RoutingPlan;
use super::seams::CallbackInvoker;
use super::validation::RouteError;

// ===========================================================================
// SubprocessInvokerConfig
// ===========================================================================

/// Configuration for [`SubprocessCallbackInvoker`].
///
/// `program` is the external client command (an absolute path to the
/// client's callback binary). `args` are appended on every spawn.
/// `timeout` bounds the total round trip — write request, read
/// response, child exit. A child still running at deadline is killed
/// and the invocation fails with [`SubprocessInvokerError::Timeout`].
#[derive(Debug, Clone)]
pub struct SubprocessInvokerConfig {
    program: PathBuf,
    args: Vec<OsString>,
    timeout: Duration,
}

impl SubprocessInvokerConfig {
    /// Construct a config from an explicit program path and timeout.
    /// The default timeout is intentionally not provided — the caller
    /// must pick a value that matches its lifecycle SLOs.
    pub fn new(program: impl Into<PathBuf>, timeout: Duration) -> Self {
        Self {
            program: program.into(),
            args: Vec::new(),
            timeout,
        }
    }

    /// Append a single argument. Returns `self` for chaining.
    pub fn arg(mut self, arg: impl Into<OsString>) -> Self {
        self.args.push(arg.into());
        self
    }

    /// Append a sequence of arguments. Returns `self` for chaining.
    pub fn args<I, A>(mut self, args: I) -> Self
    where
        I: IntoIterator<Item = A>,
        A: Into<OsString>,
    {
        self.args.extend(args.into_iter().map(Into::into));
        self
    }

    pub fn program(&self) -> &PathBuf {
        &self.program
    }

    pub fn timeout(&self) -> Duration {
        self.timeout
    }
}

// ===========================================================================
// SubprocessInvokerError
// ===========================================================================

/// Failure variants surfaced by [`SubprocessCallbackInvoker::invoke`].
///
/// Every variant maps deterministically onto a single
/// [`FailureClass`] from the shared vocabulary — no new failure
/// classes are introduced by the subprocess transport.
#[derive(Debug)]
pub enum SubprocessInvokerError {
    /// Plan rejected before spawn because the event is
    /// `receipt.emitted` (a notification event that must not produce
    /// downstream invocations). Maps to
    /// [`FailureClass::InvalidRequest`].
    ReceiptEmittedRejected(RouteError),
    /// Failed to spawn the child (e.g. binary not found, permission
    /// denied). Maps to [`FailureClass::TransportError`].
    Spawn(std::io::Error),
    /// Failed to write the request JSON to the child's stdin. Maps to
    /// [`FailureClass::TransportError`].
    WriteRequest(std::io::Error),
    /// Failed to serialize the [`CallbackRequest`] before writing.
    /// Treated as an internal-class failure — the bug is on the
    /// Lifeloop side, not the transport.
    SerializeRequest(serde_json::Error),
    /// Failed to read the child's stdout. Maps to
    /// [`FailureClass::TransportError`].
    ReadResponse(std::io::Error),
    /// Child wrote bytes that did not parse as JSON. Maps to
    /// [`FailureClass::InvalidRequest`] — the response is malformed
    /// at the wire layer.
    ParseResponse(serde_json::Error),
    /// JSON parsed cleanly but the response failed
    /// [`CallbackResponse::validate`]. Maps to
    /// [`FailureClass::InvalidRequest`].
    InvalidResponse(ValidationError),
    /// Child exited non-zero. Maps to
    /// [`FailureClass::TransportError`] — the transport returned a
    /// failure signal we cannot interpret further.
    NonZeroExit { code: Option<i32>, stderr: String },
    /// Round trip exceeded the configured timeout; child was killed.
    /// Maps to [`FailureClass::Timeout`].
    Timeout,
}

impl std::fmt::Display for SubprocessInvokerError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::ReceiptEmittedRejected(e) => {
                write!(f, "subprocess invoker refused receipt.emitted plan: {e}")
            }
            Self::Spawn(e) => write!(f, "failed to spawn callback subprocess: {e}"),
            Self::WriteRequest(e) => write!(f, "failed to write request to subprocess stdin: {e}"),
            Self::SerializeRequest(e) => write!(f, "failed to serialize CallbackRequest: {e}"),
            Self::ReadResponse(e) => write!(f, "failed to read subprocess stdout: {e}"),
            Self::ParseResponse(e) => write!(
                f,
                "subprocess stdout was not a valid JSON CallbackResponse: {e}"
            ),
            Self::InvalidResponse(e) => write!(
                f,
                "subprocess returned a CallbackResponse that failed validation: {e}"
            ),
            Self::NonZeroExit { code, stderr } => match code {
                Some(c) => write!(f, "callback subprocess exited with code {c}: {stderr}"),
                None => write!(f, "callback subprocess terminated by signal: {stderr}"),
            },
            Self::Timeout => f.write_str("callback subprocess exceeded configured timeout"),
        }
    }
}

impl std::error::Error for SubprocessInvokerError {}

/// Map a [`SubprocessInvokerError`] onto the shared
/// [`FailureClass`] vocabulary.
///
/// Pure function: same variant always maps to the same class so a
/// receipt ledger replays consistently. Stays aligned with the
/// in-process [`TransportError`] mapping in
/// [`super::failure_mapping`].
pub fn failure_class_for_subprocess_error(err: &SubprocessInvokerError) -> FailureClass {
    use SubprocessInvokerError as E;
    match err {
        E::ReceiptEmittedRejected(_) => FailureClass::InvalidRequest,
        E::Spawn(_) | E::WriteRequest(_) | E::ReadResponse(_) | E::NonZeroExit { .. } => {
            FailureClass::TransportError
        }
        E::SerializeRequest(_) => FailureClass::InternalError,
        E::ParseResponse(_) | E::InvalidResponse(_) => FailureClass::InvalidRequest,
        E::Timeout => FailureClass::Timeout,
    }
}

impl From<&SubprocessInvokerError> for FailureClass {
    fn from(err: &SubprocessInvokerError) -> Self {
        failure_class_for_subprocess_error(err)
    }
}

/// Convert a [`SubprocessInvokerError`] into the shared
/// [`TransportError`] enum so callers that already speak in
/// [`super::LifeloopFailureMapper::map_transport_error`] terms can
/// route subprocess failures through the same path. `None` for
/// variants that are not transport-class (validation, parse,
/// receipt-emitted rejection).
pub fn transport_error_for(err: &SubprocessInvokerError) -> Option<TransportError> {
    use SubprocessInvokerError as E;
    match err {
        E::Spawn(e) | E::WriteRequest(e) | E::ReadResponse(e) => {
            Some(TransportError::Io(e.to_string()))
        }
        E::NonZeroExit { code, stderr } => Some(TransportError::Io(match code {
            Some(c) => format!("exit code {c}: {stderr}"),
            None => format!("terminated by signal: {stderr}"),
        })),
        E::Timeout => Some(TransportError::Timeout),
        E::SerializeRequest(e) => Some(TransportError::Internal(e.to_string())),
        E::ReceiptEmittedRejected(_) | E::ParseResponse(_) | E::InvalidResponse(_) => None,
    }
}

// ===========================================================================
// SubprocessCallbackInvoker
// ===========================================================================

/// Subprocess-backed [`CallbackInvoker`].
///
/// Spawns a fresh child per invocation. The protocol is intentionally
/// minimal: stdin receives a single JSON-serialized
/// [`DispatchEnvelope`] (carrying the [`crate::CallbackRequest`] and
/// any opaque [`PayloadEnvelope`] bodies); stdout returns a single
/// JSON-serialized [`CallbackResponse`]. The child must exit zero on
/// success. Anything outside that contract surfaces as a typed
/// [`SubprocessInvokerError`].
///
/// # Payload bodies
///
/// Payload bodies flow through the subprocess channel inside the
/// [`DispatchEnvelope`] (issue #22). Bodies are transported verbatim:
/// Lifeloop never parses `body` or dereferences `body_ref`. Clients
/// receive the same envelopes a caller passed to
/// [`CallbackInvoker::invoke`].
#[derive(Debug, Clone)]
pub struct SubprocessCallbackInvoker {
    config: SubprocessInvokerConfig,
}

impl SubprocessCallbackInvoker {
    pub fn new(config: SubprocessInvokerConfig) -> Self {
        Self { config }
    }

    pub fn config(&self) -> &SubprocessInvokerConfig {
        &self.config
    }

    /// Internal: build the [`CallbackRequest`], spawn, write, read,
    /// validate. Factored out so [`CallbackInvoker::invoke`] stays a
    /// thin adapter.
    fn invoke_inner(
        &self,
        plan: &RoutingPlan,
        payloads: &[PayloadEnvelope],
    ) -> Result<CallbackResponse, SubprocessInvokerError> {
        // Pre-spawn guard: receipt.emitted must never produce a
        // downstream invocation. Same rule as the receipt emitter,
        // enforced before any IO so a misuse cannot leak side
        // effects (spawned process, stderr, exit-code observability).
        validate_receipt_eligible(plan).map_err(SubprocessInvokerError::ReceiptEmittedRejected)?;

        let request = super::callbacks::synthesize_request(plan);
        let envelope = DispatchEnvelope::new(request, payloads.to_vec());
        let request_bytes =
            serde_json::to_vec(&envelope).map_err(SubprocessInvokerError::SerializeRequest)?;

        let mut child = Command::new(&self.config.program)
            .args(&self.config.args)
            .stdin(Stdio::piped())
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .spawn()
            .map_err(SubprocessInvokerError::Spawn)?;

        // Hand stdin/stdout/stderr to helper threads so the main
        // thread can honor the timeout independently of how the
        // child consumes input. Each thread reports its result over
        // an mpsc channel so the main thread can bound every join by
        // the configured deadline — a direct `JoinHandle::join()`
        // would block indefinitely if a descendant of the child
        // inherited a pipe handle and held it open after the direct
        // child exited.
        let stdin = child.stdin.take().expect("stdin piped");
        let stdout = child.stdout.take().expect("stdout piped");
        let stderr = child.stderr.take().expect("stderr piped");

        let (writer_tx, writer_rx) = mpsc::channel::<std::io::Result<()>>();
        thread::spawn({
            let bytes = request_bytes;
            move || {
                let mut stdin = stdin;
                let result = stdin.write_all(&bytes).and_then(|()| stdin.flush());
                let _ = writer_tx.send(result);
                // stdin dropped on scope exit, signaling EOF.
            }
        });

        let (stdout_tx, stdout_rx) = mpsc::channel::<std::io::Result<Vec<u8>>>();
        thread::spawn(move || {
            let mut buf = Vec::new();
            let mut s = stdout;
            let result = s.read_to_end(&mut buf).map(|_| buf);
            let _ = stdout_tx.send(result);
        });

        let (stderr_tx, stderr_rx) = mpsc::channel::<Vec<u8>>();
        thread::spawn(move || {
            let mut buf = Vec::new();
            let mut s = stderr;
            let _ = s.read_to_end(&mut buf);
            let _ = stderr_tx.send(buf);
        });

        let deadline = Instant::now() + self.config.timeout;
        let exit_status = match wait_with_deadline(&mut child, deadline) {
            Ok(status) => status,
            Err(WaitError::Timeout) => {
                let _ = child.kill();
                let _ = child.wait();
                return Err(SubprocessInvokerError::Timeout);
            }
            Err(WaitError::Io(e)) => {
                let _ = child.kill();
                let _ = child.wait();
                return Err(SubprocessInvokerError::ReadResponse(e));
            }
        };

        // Reap helper threads via channels with deadline-bounded
        // recvs. After child exit the OS closes the child-side pipe
        // ends, so on a well-behaved client all three threads finish
        // immediately. If a descendant inherited stdout/stderr and is
        // still holding a writer end, the read threads block — and
        // we treat that as a transport timeout rather than hanging
        // the invoker. The configured `timeout` therefore bounds the
        // *total* round trip (write + child run + drain), not just
        // child exit.
        //
        // A small grace window covers the common case where the
        // child has just exited and the threads have not yet noticed
        // EOF; without it, a deadline that already elapsed would
        // spuriously time out a successful run.
        let grace = Duration::from_millis(100);
        let join_timeout = remaining(deadline).max(grace);

        // Writer first: a stdin write failure means the request was
        // not delivered intact, so we MUST surface it even when the
        // child later exits zero. A misbehaving client that ignores
        // stdin and fabricates a response is exactly the false-
        // success path the transport contract forbids.
        match writer_rx.recv_timeout(join_timeout) {
            Ok(Ok(())) => {}
            Ok(Err(e)) => return Err(SubprocessInvokerError::WriteRequest(e)),
            Err(mpsc::RecvTimeoutError::Timeout) => return Err(SubprocessInvokerError::Timeout),
            Err(mpsc::RecvTimeoutError::Disconnected) => {
                return Err(SubprocessInvokerError::WriteRequest(std::io::Error::other(
                    "writer thread disconnected before reporting result",
                )));
            }
        }

        let stdout_bytes = match stdout_rx.recv_timeout(join_timeout) {
            Ok(Ok(buf)) => buf,
            Ok(Err(e)) => return Err(SubprocessInvokerError::ReadResponse(e)),
            Err(mpsc::RecvTimeoutError::Timeout) => return Err(SubprocessInvokerError::Timeout),
            Err(mpsc::RecvTimeoutError::Disconnected) => {
                return Err(SubprocessInvokerError::ReadResponse(std::io::Error::other(
                    "stdout reader thread disconnected before reporting result",
                )));
            }
        };

        let stderr_bytes = stderr_rx.recv_timeout(join_timeout).unwrap_or_default();
        let stderr_text = String::from_utf8_lossy(&stderr_bytes).into_owned();

        if !exit_status.success() {
            return Err(SubprocessInvokerError::NonZeroExit {
                code: exit_status.code(),
                stderr: stderr_text,
            });
        }

        let response: CallbackResponse =
            serde_json::from_slice(&stdout_bytes).map_err(SubprocessInvokerError::ParseResponse)?;
        response
            .validate()
            .map_err(SubprocessInvokerError::InvalidResponse)?;
        Ok(response)
    }
}

impl CallbackInvoker for SubprocessCallbackInvoker {
    type Error = SubprocessInvokerError;

    fn invoke(
        &self,
        plan: &RoutingPlan,
        payloads: &[PayloadEnvelope],
    ) -> Result<CallbackResponse, Self::Error> {
        self.invoke_inner(plan, payloads)
    }
}

// ===========================================================================
// wait_with_deadline
// ===========================================================================

enum WaitError {
    Timeout,
    Io(std::io::Error),
}

/// Poll-based deadline wait. `std::process::Child` has no portable
/// timed-wait; we poll `try_wait` with bounded sleeps. The poll
/// interval is short relative to typical lifecycle SLOs (10ms) and
/// scales with the remaining deadline so a 5-second timeout does not
/// burn CPU.
/// Time remaining until `deadline`. Saturates at zero so callers can
/// use it as a `recv_timeout` argument without underflowing.
fn remaining(deadline: Instant) -> Duration {
    deadline.saturating_duration_since(Instant::now())
}

fn wait_with_deadline(
    child: &mut Child,
    deadline: Instant,
) -> Result<std::process::ExitStatus, WaitError> {
    let mut interval = Duration::from_millis(2);
    let cap = Duration::from_millis(50);
    loop {
        match child.try_wait() {
            Ok(Some(status)) => return Ok(status),
            Ok(None) => {}
            Err(e) => return Err(WaitError::Io(e)),
        }
        let now = Instant::now();
        if now >= deadline {
            // One last try_wait to avoid losing a race where the
            // child exited between the previous check and now.
            match child.try_wait() {
                Ok(Some(status)) => return Ok(status),
                Ok(None) => return Err(WaitError::Timeout),
                Err(e) => return Err(WaitError::Io(e)),
            }
        }
        let remaining = deadline.saturating_duration_since(now);
        thread::sleep(interval.min(remaining));
        if interval < cap {
            interval = (interval * 2).min(cap);
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn failure_class_mapping_is_deterministic() {
        let cases: Vec<(SubprocessInvokerError, FailureClass)> = vec![
            (
                SubprocessInvokerError::ReceiptEmittedRejected(RouteError::InvalidEventEnvelope {
                    detail: "x".into(),
                }),
                FailureClass::InvalidRequest,
            ),
            (
                SubprocessInvokerError::Spawn(std::io::Error::other("nope")),
                FailureClass::TransportError,
            ),
            (
                SubprocessInvokerError::WriteRequest(std::io::Error::other("epipe")),
                FailureClass::TransportError,
            ),
            (
                SubprocessInvokerError::ReadResponse(std::io::Error::other("eof")),
                FailureClass::TransportError,
            ),
            (
                SubprocessInvokerError::NonZeroExit {
                    code: Some(1),
                    stderr: "bang".into(),
                },
                FailureClass::TransportError,
            ),
            (SubprocessInvokerError::Timeout, FailureClass::Timeout),
            (
                SubprocessInvokerError::ParseResponse(
                    serde_json::from_str::<serde_json::Value>("not json").unwrap_err(),
                ),
                FailureClass::InvalidRequest,
            ),
        ];
        for (err, expected) in cases {
            let fc = failure_class_for_subprocess_error(&err);
            assert_eq!(fc, expected, "subprocess err -> failure class: {err}");
            let via_from: FailureClass = (&err).into();
            assert_eq!(via_from, fc);
        }
    }

    #[test]
    fn transport_error_for_distinguishes_retryable_shapes() {
        assert!(matches!(
            transport_error_for(&SubprocessInvokerError::Timeout),
            Some(TransportError::Timeout)
        ));
        assert!(matches!(
            transport_error_for(&SubprocessInvokerError::Spawn(std::io::Error::other("x"))),
            Some(TransportError::Io(_))
        ));
        assert!(
            transport_error_for(&SubprocessInvokerError::ReceiptEmittedRejected(
                RouteError::InvalidEventEnvelope { detail: "x".into() }
            ))
            .is_none()
        );
    }

    #[test]
    fn config_is_chainable() {
        let cfg = SubprocessInvokerConfig::new("/bin/cat", Duration::from_secs(1))
            .arg("-")
            .args(["--flag"]);
        assert_eq!(cfg.program(), &PathBuf::from("/bin/cat"));
        assert_eq!(cfg.timeout(), Duration::from_secs(1));
    }
}