feat: implement worker lifecycle and sandbox

src/process/lifecycle.rs

@@ -0,0 +1,616 @@
+//! One-attempt synchronous worker lifecycle and process-group cancellation.
+
+use std::{
+    io,
+    os::unix::process::CommandExt as _,
+    process::{Command, ExitStatus as ProcessExitStatus, Stdio},
+    thread,
+    time::{Duration, Instant, SystemTime, UNIX_EPOCH},
+};
+
+use rustix::process::{Pid, Signal, kill_process_group, test_kill_process_group};
+use thiserror::Error;
+
+use crate::{
+    AppError, ErrorCode,
+    domain::PositiveU64,
+    protocol::{AdapterRequest, AdapterResponse, AdapterResult, InvocationFiles},
+    sandbox::WorkerCommand,
+};
+
+use super::{
+    capture::CapturedStream,
+    diagnostics::{
+        DiagnosticBundle, InvocationDiagnosticMetadata, TerminationOutcome, TerminationReason,
+    },
+};
+
+/// Fixed time after SIGTERM before forced process-group termination.
+pub const TERMINATION_GRACE: Duration = Duration::from_secs(10);
+const POLL_INTERVAL: Duration = Duration::from_millis(10);
+const WATCHDOG_OVERHEAD_SECONDS: u64 = 120;
+
+/// Pollable caller cancellation state.
+///
+/// A value of one requests graceful interruption; two or more immediately
+/// escalates an active grace period to SIGKILL.
+pub trait Cancellation {
+    /// Number of interrupt requests observed so far.
+    fn interrupt_count(&self) -> u32;
+}
+
+/// Production default when no frontend cancellation adapter is installed.
+#[derive(Clone, Copy, Debug, Default)]
+pub struct NeverCancel;
+
+impl Cancellation for NeverCancel {
+    fn interrupt_count(&self) -> u32 {
+        0
+    }
+}
+
+/// Enabled native phases used to derive, never independently configure, a watchdog.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub struct NativePhaseTimeouts {
+    /// Auto-analysis phase, absent for a compatible cached Analysis.
+    pub analysis_seconds: Option<PositiveU64>,
+    /// Targeted decompilation phase, absent for non-decompilation Queries.
+    pub decompile_seconds: Option<PositiveU64>,
+}
+
+impl NativePhaseTimeouts {
+    /// Sum of enabled phase bounds plus the fixed 120-second harness overhead.
+    pub fn watchdog(self) -> Result<Duration, WatchdogError> {
+        let seconds = self
+            .analysis_seconds
+            .map_or(0, PositiveU64::get)
+            .checked_add(self.decompile_seconds.map_or(0, PositiveU64::get))
+            .and_then(|value| value.checked_add(WATCHDOG_OVERHEAD_SECONDS))
+            .ok_or(WatchdogError::Overflow)?;
+        Ok(Duration::from_secs(seconds))
+    }
+}
+
+/// Derived watchdog construction failure.
+#[derive(Clone, Copy, Debug, Error, Eq, PartialEq)]
+pub enum WatchdogError {
+    /// Selected phase bounds cannot be represented after addition.
+    #[error("derived child watchdog overflows the supported duration")]
+    Overflow,
+    /// Protocol provenance did not echo the automatically derived value.
+    #[error("request child watchdog does not match enabled native phases")]
+    RequestMismatch,
+}
+
+/// Inputs for exactly one worker attempt.
+pub struct Invocation<'a> {
+    /// Strict request written before launch.
+    request: &'a AdapterRequest,
+    /// Existing private file-protocol directory.
+    files: &'a InvocationFiles,
+    /// Fully resolved direct or sandboxed command.
+    command: &'a WorkerCommand,
+    /// Automatically derived hard watchdog.
+    watchdog: Duration,
+}
+
+impl<'a> Invocation<'a> {
+    /// Creates an Invocation only when limit provenance matches the derived watchdog.
+    pub fn new(
+        request: &'a AdapterRequest,
+        files: &'a InvocationFiles,
+        command: &'a WorkerCommand,
+        phases: NativePhaseTimeouts,
+    ) -> Result<Self, WatchdogError> {
+        let watchdog = phases.watchdog()?;
+        if request.limits.child_watchdog_seconds.get() != watchdog.as_secs() {
+            return Err(WatchdogError::RequestMismatch);
+        }
+        Ok(Self {
+            request,
+            files,
+            command,
+            watchdog,
+        })
+    }
+
+    /// Replaces the production watchdog only in debug builds for fast lifecycle tests.
+    #[cfg(debug_assertions)]
+    #[must_use]
+    pub fn with_test_watchdog(mut self, watchdog: Duration) -> Self {
+        self.watchdog = watchdog;
+        self
+    }
+}
+
+/// A complete successful transport attempt.
+#[derive(Debug)]
+pub struct CompletedInvocation {
+    /// Strict validated adapter response.
+    pub response: AdapterResponse,
+    /// Captured streams, normally dropped unless a success warning references them.
+    pub diagnostics: DiagnosticBundle,
+}
+
+impl CompletedInvocation {
+    /// Typed adapter errors always require durable diagnostic publication.
+    #[must_use]
+    pub const fn requires_diagnostic_publication(&self) -> bool {
+        matches!(&self.response.result, AdapterResult::Error { .. })
+    }
+}
+
+/// Failure category with stable public error mapping.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum InvocationFailureKind {
+    /// Request file construction failed before launch.
+    RequestProtocol,
+    /// OS process creation failed.
+    Spawn,
+    /// Derived watchdog expired.
+    WatchdogTimeout,
+    /// Caller interruption won the result race.
+    Interrupted,
+    /// Child exited unsuccessfully.
+    ChildExit,
+    /// Response file or typed response validation failed.
+    ResponseProtocol,
+    /// Concurrent stream drain failed.
+    DiagnosticDrain,
+}
+
+impl InvocationFailureKind {
+    /// Stable public code used by command orchestration.
+    #[must_use]
+    pub const fn error_code(self) -> ErrorCode {
+        match self {
+            Self::WatchdogTimeout => ErrorCode::Timeout,
+            Self::Interrupted => ErrorCode::Interrupted,
+            Self::RequestProtocol | Self::ResponseProtocol => ErrorCode::ProtocolViolation,
+            Self::Spawn | Self::ChildExit | Self::DiagnosticDrain => ErrorCode::Internal,
+        }
+    }
+}
+
+/// Failed transport attempt with retained diagnostic material.
+#[derive(Debug, Error)]
+#[error("{message}")]
+pub struct InvocationFailure {
+    /// Stable lifecycle classification.
+    pub kind: InvocationFailureKind,
+    /// Concise internal context.
+    pub message: String,
+    /// Captured material to publish atomically.
+    pub diagnostics: Box<DiagnosticBundle>,
+}
+
+impl InvocationFailure {
+    /// Converts lifecycle classification to the public error taxonomy.
+    #[must_use]
+    pub fn app_error(&self) -> AppError {
+        AppError::new(self.kind.error_code(), self.message.clone(), false)
+    }
+}
+
+/// Runs one and only one worker process attempt.
+pub fn run(
+    invocation: Invocation<'_>,
+    cancellation: &impl Cancellation,
+) -> Result<CompletedInvocation, InvocationFailure> {
+    let started_unix_millis = unix_millis();
+    if let Err(error) = invocation.files.write_request(invocation.request) {
+        return Err(failure_without_child(
+            invocation.request,
+            started_unix_millis,
+            InvocationFailureKind::RequestProtocol,
+            TerminationReason::ProtocolViolation,
+            error.to_string(),
+        ));
+    }
+
+    let mut command = Command::new(&invocation.command.program);
+    command
+        .args(&invocation.command.arguments)
+        .stdin(Stdio::null())
+        .stdout(Stdio::piped())
+        .stderr(Stdio::piped())
+        .process_group(0);
+    let mut child = match command.spawn() {
+        Ok(child) => child,
+        Err(error) => {
+            return Err(failure_without_child(
+                invocation.request,
+                started_unix_millis,
+                InvocationFailureKind::Spawn,
+                TerminationReason::SpawnFailure,
+                format!("worker launch failed: {error}"),
+            ));
+        }
+    };
+
+    let stdout = child.stdout.take();
+    let stderr = child.stderr.take();
+    let stdout_thread = stdout.map(|pipe| thread::spawn(move || CapturedStream::drain(pipe)));
+    let stderr_thread = stderr.map(|pipe| thread::spawn(move || CapturedStream::drain(pipe)));
+    let process_group = Pid::from_raw(i32::try_from(child.id()).unwrap_or(i32::MAX));
+    let deadline = Instant::now().checked_add(invocation.watchdog);
+
+    let mut cancellation_reason = None;
+    let status = loop {
+        if cancellation.interrupt_count() > 0 {
+            cancellation_reason = Some(TerminationReason::Interrupted);
+            break None;
+        }
+        if deadline.is_none_or(|value| Instant::now() >= value) {
+            cancellation_reason = Some(TerminationReason::WatchdogTimeout);
+            break None;
+        }
+        match child.try_wait() {
+            Ok(Some(status)) => break Some(status),
+            Ok(None) => thread::sleep(POLL_INTERVAL),
+            Err(_error) => {
+                cancellation_reason = Some(TerminationReason::ChildExit);
+                break None;
+            }
+        }
+    };
+
+    while cancellation_reason.is_none()
+        && !drains_finished(stdout_thread.as_ref(), stderr_thread.as_ref())
+    {
+        if cancellation.interrupt_count() > 0 {
+            cancellation_reason = Some(TerminationReason::Interrupted);
+        } else if deadline.is_none_or(|value| Instant::now() >= value) {
+            cancellation_reason = Some(TerminationReason::WatchdogTimeout);
+        } else {
+            thread::sleep(POLL_INTERVAL);
+        }
+    }
+
+    let (status, termination_outcome) = if let Some(reason) = cancellation_reason {
+        let (status, outcome) = terminate_group(
+            &mut child,
+            process_group,
+            cancellation,
+            stdout_thread.as_ref(),
+            stderr_thread.as_ref(),
+        );
+        let failure_kind = if reason == TerminationReason::Interrupted {
+            InvocationFailureKind::Interrupted
+        } else if reason == TerminationReason::WatchdogTimeout {
+            InvocationFailureKind::WatchdogTimeout
+        } else {
+            InvocationFailureKind::ChildExit
+        };
+        let (stdout, stderr, drain_error) = join_captures(stdout_thread, stderr_thread);
+        let diagnostics = diagnostic_bundle(
+            invocation.request,
+            started_unix_millis,
+            reason,
+            outcome,
+            status.as_ref(),
+            stdout,
+            stderr,
+        );
+        return Err(InvocationFailure {
+            kind: failure_kind,
+            message: drain_error.unwrap_or_else(|| match reason {
+                TerminationReason::Interrupted => "worker invocation was interrupted".to_owned(),
+                TerminationReason::WatchdogTimeout => {
+                    "worker exceeded the derived watchdog".to_owned()
+                }
+                _ => "worker status could not be observed".to_owned(),
+            }),
+            diagnostics,
+        });
+    } else {
+        (status, TerminationOutcome::None)
+    };
+
+    let (stdout, stderr, drain_error) = join_captures(stdout_thread, stderr_thread);
+    if let Some(message) = drain_error {
+        return Err(InvocationFailure {
+            kind: InvocationFailureKind::DiagnosticDrain,
+            message,
+            diagnostics: diagnostic_bundle(
+                invocation.request,
+                started_unix_millis,
+                TerminationReason::ProtocolViolation,
+                termination_outcome,
+                status.as_ref(),
+                stdout,
+                stderr,
+            ),
+        });
+    }
+    let Some(status) = status else {
+        return Err(InvocationFailure {
+            kind: InvocationFailureKind::ChildExit,
+            message: "worker exited without an observable status".to_owned(),
+            diagnostics: diagnostic_bundle(
+                invocation.request,
+                started_unix_millis,
+                TerminationReason::ChildExit,
+                termination_outcome,
+                None,
+                stdout,
+                stderr,
+            ),
+        });
+    };
+    if cancellation.interrupt_count() > 0 {
+        return Err(InvocationFailure {
+            kind: InvocationFailureKind::Interrupted,
+            message: "worker invocation was interrupted".to_owned(),
+            diagnostics: diagnostic_bundle(
+                invocation.request,
+                started_unix_millis,
+                TerminationReason::Interrupted,
+                TerminationOutcome::Graceful,
+                Some(&status),
+                stdout,
+                stderr,
+            ),
+        });
+    }
+    if !status.success() {
+        return Err(InvocationFailure {
+            kind: InvocationFailureKind::ChildExit,
+            message: format!("worker exited unsuccessfully: {status}"),
+            diagnostics: diagnostic_bundle(
+                invocation.request,
+                started_unix_millis,
+                TerminationReason::ChildExit,
+                termination_outcome,
+                Some(&status),
+                stdout,
+                stderr,
+            ),
+        });
+    }
+    if process_group.is_some_and(group_is_alive) {
+        let outcome = terminate_lingering_group(process_group, cancellation);
+        return Err(InvocationFailure {
+            kind: InvocationFailureKind::ChildExit,
+            message: "worker leader exited while its process group remained alive".to_owned(),
+            diagnostics: diagnostic_bundle(
+                invocation.request,
+                started_unix_millis,
+                TerminationReason::ChildExit,
+                outcome,
+                Some(&status),
+                stdout,
+                stderr,
+            ),
+        });
+    }
+
+    match invocation.files.read_response(invocation.request) {
+        Ok(response) => {
+            let interrupted = cancellation.interrupt_count() > 0;
+            let adapter_error = matches!(&response.result, AdapterResult::Error { .. });
+            let diagnostics = diagnostic_bundle(
+                invocation.request,
+                started_unix_millis,
+                if interrupted {
+                    TerminationReason::Interrupted
+                } else if adapter_error {
+                    TerminationReason::AdapterError
+                } else {
+                    TerminationReason::Success
+                },
+                termination_outcome,
+                Some(&status),
+                stdout,
+                stderr,
+            );
+            if interrupted {
+                Err(InvocationFailure {
+                    kind: InvocationFailureKind::Interrupted,
+                    message: "worker invocation was interrupted".to_owned(),
+                    diagnostics,
+                })
+            } else {
+                Ok(CompletedInvocation {
+                    response,
+                    diagnostics: *diagnostics,
+                })
+            }
+        }
+        Err(error) => Err(InvocationFailure {
+            kind: InvocationFailureKind::ResponseProtocol,
+            message: format!("worker response violated the file protocol: {error}"),
+            diagnostics: diagnostic_bundle(
+                invocation.request,
+                started_unix_millis,
+                TerminationReason::ProtocolViolation,
+                termination_outcome,
+                Some(&status),
+                stdout,
+                stderr,
+            ),
+        }),
+    }
+}
+
+fn terminate_group(
+    child: &mut std::process::Child,
+    process_group: Option<Pid>,
+    cancellation: &impl Cancellation,
+    stdout: Option<&DrainThread>,
+    stderr: Option<&DrainThread>,
+) -> (Option<ProcessExitStatus>, TerminationOutcome) {
+    signal_group(process_group, Signal::TERM);
+    let deadline = Instant::now() + TERMINATION_GRACE;
+    let mut status = None;
+    loop {
+        if cancellation.interrupt_count() > 1 || Instant::now() >= deadline {
+            signal_group(process_group, Signal::KILL);
+            return (
+                status.or_else(|| child.wait().ok()),
+                TerminationOutcome::Forced,
+            );
+        }
+        if status.is_none() {
+            status = match child.try_wait() {
+                Ok(value) => value,
+                Err(_) => child.wait().ok(),
+            };
+        }
+        if status.is_some()
+            && drains_finished(stdout, stderr)
+            && process_group.is_none_or(|group| !group_is_alive(group))
+        {
+            return (status, TerminationOutcome::Graceful);
+        }
+        thread::sleep(POLL_INTERVAL);
+    }
+}
+
+fn terminate_lingering_group(
+    process_group: Option<Pid>,
+    cancellation: &impl Cancellation,
+) -> TerminationOutcome {
+    signal_group(process_group, Signal::TERM);
+    let deadline = Instant::now() + TERMINATION_GRACE;
+    while process_group.is_some_and(group_is_alive) {
+        if cancellation.interrupt_count() > 1 || Instant::now() >= deadline {
+            signal_group(process_group, Signal::KILL);
+            return TerminationOutcome::Forced;
+        }
+        thread::sleep(POLL_INTERVAL);
+    }
+    TerminationOutcome::Graceful
+}
+
+fn group_is_alive(process_group: Pid) -> bool {
+    test_kill_process_group(process_group).is_ok()
+}
+
+fn signal_group(process_group: Option<Pid>, signal: Signal) {
+    if let Some(process_group) = process_group {
+        let _ignored = kill_process_group(process_group, signal);
+    }
+}
+
+type DrainThread = thread::JoinHandle<io::Result<CapturedStream>>;
+
+fn drains_finished(stdout: Option<&DrainThread>, stderr: Option<&DrainThread>) -> bool {
+    stdout.is_none_or(thread::JoinHandle::is_finished)
+        && stderr.is_none_or(thread::JoinHandle::is_finished)
+}
+
+fn join_captures(
+    stdout: Option<DrainThread>,
+    stderr: Option<DrainThread>,
+) -> (CapturedStream, CapturedStream, Option<String>) {
+    let (stdout, stdout_error) = join_capture(stdout, "stdout");
+    let (stderr, stderr_error) = join_capture(stderr, "stderr");
+    (stdout, stderr, stdout_error.or(stderr_error))
+}
+
+fn join_capture(thread: Option<DrainThread>, name: &str) -> (CapturedStream, Option<String>) {
+    let Some(thread) = thread else {
+        return (
+            CapturedStream::empty(),
+            Some(format!("worker {name} pipe was unavailable")),
+        );
+    };
+    match thread.join() {
+        Ok(Ok(capture)) => (capture, None),
+        Ok(Err(error)) => (
+            CapturedStream::empty(),
+            Some(format!("worker {name} drain failed: {error}")),
+        ),
+        Err(_) => (
+            CapturedStream::empty(),
+            Some(format!("worker {name} drain thread failed")),
+        ),
+    }
+}
+
+fn failure_without_child(
+    request: &AdapterRequest,
+    started_unix_millis: u128,
+    kind: InvocationFailureKind,
+    reason: TerminationReason,
+    message: String,
+) -> InvocationFailure {
+    InvocationFailure {
+        kind,
+        message,
+        diagnostics: diagnostic_bundle(
+            request,
+            started_unix_millis,
+            reason,
+            TerminationOutcome::None,
+            None,
+            CapturedStream::empty(),
+            CapturedStream::empty(),
+        ),
+    }
+}
+
+fn diagnostic_bundle(
+    request: &AdapterRequest,
+    started_unix_millis: u128,
+    termination_reason: TerminationReason,
+    termination_outcome: TerminationOutcome,
+    status: Option<&ProcessExitStatus>,
+    stdout: CapturedStream,
+    stderr: CapturedStream,
+) -> Box<DiagnosticBundle> {
+    Box::new(DiagnosticBundle::new(
+        InvocationDiagnosticMetadata {
+            invocation_id: request.invocation_id.clone(),
+            operation: request.operation,
+            started_unix_millis,
+            finished_unix_millis: unix_millis(),
+            termination_reason,
+            termination_outcome,
+            child_exit_code: status.and_then(ProcessExitStatus::code),
+        },
+        stdout,
+        stderr,
+    ))
+}
+
+fn unix_millis() -> u128 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .unwrap_or(Duration::ZERO)
+        .as_millis()
+}
+
+#[cfg(test)]
+#[allow(clippy::expect_used)]
+mod tests {
+    use std::time::Duration;
+
+    use super::NativePhaseTimeouts;
+    use crate::domain::PositiveU64;
+
+    #[test]
+    fn derives_watchdog_from_enabled_native_phases() {
+        let analysis = PositiveU64::try_from(600).expect("positive");
+        let decompile = PositiveU64::try_from(60).expect("positive");
+        assert_eq!(
+            NativePhaseTimeouts {
+                analysis_seconds: Some(analysis),
+                decompile_seconds: Some(decompile)
+            }
+            .watchdog()
+            .expect("duration"),
+            Duration::from_secs(780),
+        );
+        assert_eq!(
+            NativePhaseTimeouts {
+                analysis_seconds: None,
+                decompile_seconds: None
+            }
+            .watchdog()
+            .expect("duration"),
+            Duration::from_secs(120),
+        );
+    }
+}