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
//!   [`crate::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};

#[cfg(unix)]
use std::os::unix::process::CommandExt;

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 [`crate::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` (private).
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
// ===========================================================================

const MAX_STDOUT_BYTES: u64 = 16 * 1024 * 1024;
const MAX_STDERR_BYTES: u64 = 256 * 1024;

/// 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 [`crate::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 command = Command::new(&self.config.program);
        command
            .args(&self.config.args)
            .stdin(Stdio::piped())
            .stdout(Stdio::piped())
            .stderr(Stdio::piped());
        #[cfg(unix)]
        {
            command.process_group(0);
        }
        let mut child = command.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 s = stdout;
            let result = read_to_end_limited(&mut s, MAX_STDOUT_BYTES, "stdout");
            let _ = stdout_tx.send(result);
        });

        let (stderr_tx, stderr_rx) = mpsc::channel::<Vec<u8>>();
        thread::spawn(move || {
            let mut s = stderr;
            let buf = read_to_end_truncated(&mut s, MAX_STDERR_BYTES).unwrap_or_default();
            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) => {
                terminate_child_tree(&mut child);
                return Err(SubprocessInvokerError::Timeout);
            }
            Err(WaitError::Io(e)) => {
                terminate_child_tree(&mut child);
                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.
        //
        // Exception: EPIPE on stdin almost always means the child
        // already closed its read end — i.e. it has exited. The exit
        // code, not the BrokenPipe, is the truthful diagnosis:
        //   * non-zero exit -> surface the exit code; the BrokenPipe is
        //     its downstream symptom, and surfacing it would mask the
        //     real failure (and make the error scheduling-dependent,
        //     since a slower runner observes EPIPE before the writer
        //     finishes);
        //   * zero exit -> the child may have read the prefix it needed,
        //     written a valid response, and exited cleanly without
        //     draining the rest of stdin (realistic once the payload
        //     exceeds the OS pipe buffer). In that case the BrokenPipe
        //     is benign and we must NOT discard the captured response.
        //     We defer the decision: parse stdout first, return the
        //     response when present, and fall back to surfacing the
        //     BrokenPipe write error only if no response was produced.
        let mut deferred_pipe_write_err: Option<std::io::Error> = None;
        match writer_rx.recv_timeout(join_timeout) {
            Ok(Ok(())) => {}
            Ok(Err(e)) => {
                if e.kind() == std::io::ErrorKind::BrokenPipe {
                    if !exit_status.success() {
                        let stderr_bytes = stderr_rx.recv_timeout(join_timeout).unwrap_or_default();
                        let stderr_text = String::from_utf8_lossy(&stderr_bytes).into_owned();
                        return Err(SubprocessInvokerError::NonZeroExit {
                            code: exit_status.code(),
                            stderr: stderr_text,
                        });
                    }
                    // Zero exit + BrokenPipe: defer, decide after parse.
                    deferred_pipe_write_err = Some(e);
                } else {
                    return Err(SubprocessInvokerError::WriteRequest(e));
                }
            }
            Err(mpsc::RecvTimeoutError::Timeout) => {
                terminate_child_tree(&mut child);
                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) => {
                terminate_child_tree(&mut child);
                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 = match stderr_rx.recv_timeout(join_timeout) {
            Ok(buf) => buf,
            Err(mpsc::RecvTimeoutError::Timeout) => {
                terminate_child_tree(&mut child);
                Vec::new()
            }
            Err(mpsc::RecvTimeoutError::Disconnected) => Vec::new(),
        };
        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 = match serde_json::from_slice(&stdout_bytes) {
            Ok(response) => response,
            Err(parse_err) => {
                // A deferred benign-pipe write error only matters when
                // the child produced no usable response: prefer the
                // truthful transport diagnosis (the request was not
                // delivered intact) over a misleading parse error.
                if let Some(write_err) = deferred_pipe_write_err {
                    return Err(SubprocessInvokerError::WriteRequest(write_err));
                }
                return Err(SubprocessInvokerError::ParseResponse(parse_err));
            }
        };
        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)
    }
}

fn read_to_end_limited<R: Read>(
    reader: &mut R,
    max_bytes: u64,
    stream_name: &'static str,
) -> std::io::Result<Vec<u8>> {
    let mut buf = Vec::new();
    reader.take(max_bytes + 1).read_to_end(&mut buf)?;
    if buf.len() as u64 > max_bytes {
        return Err(std::io::Error::other(format!(
            "subprocess {stream_name} exceeded {max_bytes} bytes"
        )));
    }
    Ok(buf)
}

fn read_to_end_truncated<R: Read>(reader: &mut R, max_bytes: u64) -> std::io::Result<Vec<u8>> {
    let mut buf = Vec::new();
    reader.take(max_bytes + 1).read_to_end(&mut buf)?;
    if buf.len() as u64 > max_bytes {
        buf.truncate(max_bytes as usize);
        buf.extend_from_slice(b"\n[stderr truncated]\n");
    }
    Ok(buf)
}

fn terminate_child_tree(child: &mut Child) {
    #[cfg(unix)]
    terminate_process_group(child.id());
    let _ = child.kill();
    let _ = child.wait();
}

#[cfg(unix)]
fn terminate_process_group(child_pid: u32) {
    let pgid = format!("-{child_pid}");
    let _ = Command::new("kill")
        .args(["-TERM", "--", &pgid])
        .stdout(Stdio::null())
        .stderr(Stdio::null())
        .status();
    thread::sleep(Duration::from_millis(20));
    let _ = Command::new("kill")
        .args(["-KILL", "--", &pgid])
        .stdout(Stdio::null())
        .stderr(Stdio::null())
        .status();
}

// ===========================================================================
// 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::*;
    use std::io::Cursor;

    #[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 read_to_end_limited_allows_exact_limit_and_rejects_overflow() {
        let mut exact = Cursor::new(b"abcd".to_vec());
        assert_eq!(
            read_to_end_limited(&mut exact, 4, "stdout").unwrap(),
            b"abcd"
        );

        let mut over = Cursor::new(b"abcde".to_vec());
        let err = read_to_end_limited(&mut over, 4, "stdout").unwrap_err();
        assert!(
            err.to_string()
                .contains("subprocess stdout exceeded 4 bytes")
        );
    }

    #[test]
    fn read_to_end_truncated_marks_only_over_limit_stderr() {
        let mut exact = Cursor::new(b"abcd".to_vec());
        assert_eq!(read_to_end_truncated(&mut exact, 4).unwrap(), b"abcd");

        let mut over = Cursor::new(b"abcde".to_vec());
        let mut expected = b"abcd".to_vec();
        expected.extend_from_slice(b"\n[stderr truncated]\n");
        assert_eq!(read_to_end_truncated(&mut over, 4).unwrap(), expected);
    }

    #[test]
    fn remaining_reports_future_duration_and_saturates_past_deadline() {
        let future = Instant::now() + Duration::from_secs(60);
        assert!(remaining(future) > Duration::from_secs(59));

        let past = Instant::now() - Duration::from_millis(1);
        assert_eq!(remaining(past), Duration::ZERO);
    }

    #[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));
    }

    #[cfg(unix)]
    #[test]
    fn zero_exit_child_that_ignores_stdin_yields_its_response_despite_epipe() {
        use crate::{
            AdapterManifest, AdapterRole, IntegrationMode, LifecycleEventKind,
            ManifestContextPressure, ManifestReceipts, ReceiptStatus, SCHEMA_VERSION, SupportState,
        };
        use std::collections::BTreeMap;

        // A client that emits a valid CallbackResponse on stdout and
        // exits zero WITHOUT draining stdin. With a payload larger than
        // the OS pipe buffer, the parent's stdin write hits EPIPE after
        // the child has already exited cleanly. The invoker must still
        // parse the captured response rather than reporting a transport
        // write error.
        let response = CallbackResponse::ok(ReceiptStatus::Delivered);
        let response_json = serde_json::to_string(&response).expect("serialize response");
        // `printf` the response, then exit 0; stdin is never read.
        let script = format!("printf '%s' '{response_json}'; exit 0");

        let cfg = SubprocessInvokerConfig::new("/bin/sh", Duration::from_secs(5))
            .arg("-c")
            .arg(script);
        let invoker = SubprocessCallbackInvoker::new(cfg);

        let manifest = AdapterManifest {
            contract_version: SCHEMA_VERSION.to_string(),
            adapter_id: "fake-adapter".into(),
            adapter_version: "0.0.1".into(),
            display_name: "Fake".into(),
            role: AdapterRole::PrimaryWorker,
            integration_modes: vec![IntegrationMode::NativeHook],
            lifecycle_events: BTreeMap::new(),
            placement: BTreeMap::new(),
            context_pressure: ManifestContextPressure {
                support: SupportState::Native,
                evidence: None,
            },
            receipts: ManifestReceipts {
                native: false,
                lifeloop_synthesized: true,
                receipt_ledger: SupportState::Unavailable,
            },
            session_identity: None,
            session_rename: None,
            renewal: None,
            approval_surface: None,
            failure_modes: Vec::new(),
            telemetry_sources: Vec::new(),
        };
        let plan = RoutingPlan {
            event: LifecycleEventKind::SessionStarting,
            event_id: "evt_1".into(),
            invocation_id: "inv_1".into(),
            adapter: manifest,
            integration_mode: IntegrationMode::NativeHook,
            harness_session_id: None,
            harness_run_id: None,
            harness_task_id: None,
            frame_context: None,
            payload_refs: Vec::new(),
            capability_snapshot_ref: None,
            sequence: None,
            idempotency_key: None,
            metadata: serde_json::Map::new(),
        };

        // A payload body comfortably larger than any common pipe
        // buffer (64 KiB on Linux, 16 KiB on macOS) so the unread
        // stdin write provokes EPIPE.
        let big_body = "x".repeat(1024 * 1024);
        let payload = PayloadEnvelope {
            schema_version: SCHEMA_VERSION.to_string(),
            payload_id: "p_big".into(),
            client_id: "test-client".into(),
            payload_kind: "instruction_frame".into(),
            format: "client-defined".into(),
            content_encoding: "utf8".into(),
            body: Some(big_body),
            body_ref: None,
            byte_size: 1024 * 1024,
            content_digest: None,
            acceptable_placements: Vec::new(),
            idempotency_key: None,
            expires_at_epoch_s: None,
            metadata: serde_json::Map::new(),
        };

        let got = invoker
            .invoke(&plan, &[payload])
            .expect("zero-exit response must be returned despite stdin EPIPE");
        assert_eq!(got.status, ReceiptStatus::Delivered);
    }
}