feat: native process manager (#47)

Enable swactor to spawn and manage native processes using ssh.


Signed-off-by: Zachery Aaron Shores-Chmielewski <zacheryasc@gmail.com>

Cargo.toml

@@ -8,6 +8,7 @@ members = [
     "crates/dashboard",
     "crates/distribution",
     "crates/std",
+    "crates/process",
     "crates/datastore",
     "crates/shared-types",
     "crates/swactor-node",

crates/process/Cargo.toml

@@ -0,0 +1,21 @@
+[package]
+name = "swactor-process"
+version = "0.1.0"
+edition = "2024"
+
+[features]
+default = []
+ssh = ["dep:russh", "dep:russh-keys", "dep:tokio", "dep:async-trait"]
+
+[dependencies]
+swactor = { path = "../..", default-features = false, features = ["no_random"] }
+crossbeam-queue = "0.3.12"
+libc = "0.2"
+russh = { version = "0.46", optional = true }
+russh-keys = { version = "0.46", optional = true }
+tokio = { version = "1", features = ["rt", "time", "sync"], optional = true }
+async-trait = { version = "0.1", optional = true }
+
+[dev-dependencies]
+proptest = "1"
+proptest-state-machine = "0.3"

crates/process/src/action.rs

@@ -0,0 +1,53 @@
+use std::time::Duration;
+
+use crate::types::{ExitStatus, ProcessError, ProcessSpec, PtySize, Signal};
+use swactor::actor::ActorAddress;
+
+/// Which output stream produced data.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum OutputStream {
+    Stdout,
+    Stderr,
+}
+
+/// Actions emitted by ProcessSession for the driver or actor layer to execute.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum ProcessAction {
+    // --- Driver commands ---
+    /// Spawn the process described by the spec.
+    SpawnProcess { spec: ProcessSpec },
+    /// Write bytes to the process's stdin.
+    WriteStdin { data: Vec<u8> },
+    /// Send a signal to the process.
+    SendSignal { signal: Signal },
+    /// Resize the process's PTY.
+    ResizePty { size: PtySize },
+    /// Close the process's stdin pipe.
+    CloseStdin,
+    /// Schedule a kill timeout that fires KillTimeout after the given duration.
+    ScheduleKillTimeout { duration: Duration },
+
+    // --- Subscriber notifications ---
+    /// Notify subscribers that the process started.
+    NotifyStarted { subscribers: Vec<ActorAddress> },
+    /// Notify subscribers of output.
+    NotifyOutput {
+        subscribers: Vec<ActorAddress>,
+        data: Vec<u8>,
+        stream: OutputStream,
+    },
+    /// Notify subscribers that the process exited.
+    NotifyExited {
+        subscribers: Vec<ActorAddress>,
+        status: ExitStatus,
+    },
+    /// Notify subscribers of an error.
+    NotifyError {
+        subscribers: Vec<ActorAddress>,
+        error: ProcessError,
+    },
+
+    // --- Lifecycle ---
+    /// The session is done; the owning actor should stop itself.
+    SelfTerminate,
+}

crates/process/src/actor.rs

@@ -0,0 +1,161 @@
+use std::sync::{Arc, OnceLock};
+
+use swactor::actor::{ActorAddress, ActorInterface, Ctx};
+
+use crate::action::ProcessAction;
+use crate::driver::ProcessDriver;
+use crate::event::ProcessEvent;
+use crate::message::{ProcessCommand, ProcessNotification};
+use crate::session::ProcessSession;
+use crate::waker::ProcessWaker;
+
+/// Actor wrapper around a `ProcessSession` and its driver.
+///
+/// Generic over `D: ProcessDriver` so that tests can use `MockDriver` or
+/// `TestDriver` while production uses `LocalDriver`.
+pub struct ProcessActor<D: ProcessDriver> {
+    session: ProcessSession,
+    driver: D,
+    self_addr: Option<ActorAddress>,
+    /// Actions from `ProcessSession::new()`, executed in `on_start`.
+    deferred_actions: Option<Vec<ProcessAction>>,
+    /// Shared slot for the waker — filled after the actor address is known.
+    pub waker_slot: Arc<OnceLock<ProcessWaker>>,
+}
+
+impl<D: ProcessDriver> ProcessActor<D> {
+    pub fn new(
+        session: ProcessSession,
+        driver: D,
+        initial_actions: Vec<ProcessAction>,
+        waker_slot: Arc<OnceLock<ProcessWaker>>,
+    ) -> Self {
+        Self {
+            session,
+            driver,
+            self_addr: None,
+            deferred_actions: Some(initial_actions),
+            waker_slot,
+        }
+    }
+
+    /// Drain events from the driver, apply each to the session, and dispatch
+    /// all resulting actions.
+    fn drain_and_dispatch(&mut self, ctx: &Ctx) {
+        let events = self.driver.poll();
+        for event in events {
+            let actions = self.session.apply(event);
+            self.dispatch_actions(ctx, actions);
+        }
+    }
+
+    /// Execute actions produced by the session state machine.
+    fn dispatch_actions(&mut self, ctx: &Ctx, actions: Vec<ProcessAction>) {
+        let self_addr = self.self_addr.expect("self_addr not set");
+        for action in actions {
+            match action {
+                // Driver commands — forward to the driver
+                ProcessAction::SpawnProcess { .. }
+                | ProcessAction::WriteStdin { .. }
+                | ProcessAction::SendSignal { .. }
+                | ProcessAction::ResizePty { .. }
+                | ProcessAction::CloseStdin
+                | ProcessAction::ScheduleKillTimeout { .. } => {
+                    self.driver.execute(action);
+                }
+
+                // Subscriber notifications — send to each subscriber
+                ProcessAction::NotifyStarted { subscribers } => {
+                    let notif = ProcessNotification::Started { process: self_addr };
+                    for sub in subscribers {
+                        let _ = ctx.send(sub, notif.clone());
+                    }
+                }
+                ProcessAction::NotifyOutput {
+                    subscribers,
+                    data,
+                    stream,
+                } => {
+                    let notif = ProcessNotification::Output {
+                        process: self_addr,
+                        data,
+                        stream,
+                    };
+                    for sub in subscribers {
+                        let _ = ctx.send(sub, notif.clone());
+                    }
+                }
+                ProcessAction::NotifyExited {
+                    subscribers,
+                    status,
+                } => {
+                    let notif = ProcessNotification::Exited {
+                        process: self_addr,
+                        status,
+                    };
+                    for sub in subscribers {
+                        let _ = ctx.send(sub, notif.clone());
+                    }
+                }
+                ProcessAction::NotifyError {
+                    subscribers,
+                    error,
+                } => {
+                    let notif = ProcessNotification::Error {
+                        process: self_addr,
+                        error,
+                    };
+                    for sub in subscribers {
+                        let _ = ctx.send(sub, notif.clone());
+                    }
+                }
+
+                // Lifecycle
+                ProcessAction::SelfTerminate => {
+                    ctx.stop_self();
+                }
+            }
+        }
+    }
+
+    /// Map a `ProcessCommand` to the corresponding `ProcessEvent`.
+    fn command_to_event(cmd: ProcessCommand) -> Option<ProcessEvent> {
+        match cmd {
+            ProcessCommand::WriteStdin { data } => Some(ProcessEvent::WriteStdin { data }),
+            ProcessCommand::SendSignal { signal } => Some(ProcessEvent::SendSignal { signal }),
+            ProcessCommand::ResizePty { size } => Some(ProcessEvent::ResizePty { size }),
+            ProcessCommand::CloseStdin => Some(ProcessEvent::CloseStdin),
+            ProcessCommand::Close => Some(ProcessEvent::CloseRequested),
+            ProcessCommand::Subscribe { address } => Some(ProcessEvent::Subscribe { address }),
+            ProcessCommand::Unsubscribe { address } => {
+                Some(ProcessEvent::Unsubscribe { address })
+            }
+            ProcessCommand::PollTick => None, // handled by drain
+        }
+    }
+}
+
+impl<D: ProcessDriver + 'static> ActorInterface for ProcessActor<D> {
+    type Incoming = ProcessCommand;
+    type Response = ();
+
+    fn on_start(&mut self, ctx: &Ctx) {
+        self.self_addr = Some(ctx.self_addr());
+        if let Some(actions) = self.deferred_actions.take() {
+            self.dispatch_actions(ctx, actions);
+        }
+    }
+
+    fn handle(&mut self, ctx: &Ctx, msg: ProcessCommand) {
+        // Process the incoming command first — this ensures Subscribe
+        // registers before drain dispatches notifications, and keeps
+        // user commands (Close, WriteStdin) responsive.
+        if let Some(event) = Self::command_to_event(msg) {
+            let actions = self.session.apply(event);
+            self.dispatch_actions(ctx, actions);
+        }
+
+        // Then drain pending I/O events from background threads.
+        self.drain_and_dispatch(ctx);
+    }
+}

crates/process/src/driver.rs

@@ -0,0 +1,14 @@
+use crate::action::ProcessAction;
+use crate::event::ProcessEvent;
+
+/// Abstraction over the mechanism that actually runs a process.
+///
+/// Implementations translate `ProcessAction` commands into real I/O (or mock I/O)
+/// and produce `ProcessEvent`s by polling for state changes.
+pub trait ProcessDriver: Send {
+    /// Execute an action (spawn, write stdin, send signal, etc.).
+    fn execute(&mut self, action: ProcessAction);
+
+    /// Poll for new events from the underlying process.
+    fn poll(&mut self) -> Vec<ProcessEvent>;
+}

crates/process/src/event.rs

@@ -0,0 +1,71 @@
+use crate::types::{ExitStatus, PtySize, Signal};
+use swactor::actor::ActorAddress;
+
+/// Events that can be applied to a ProcessSession.
+///
+/// Some events come from the driver (Started, SpawnFailed, OutputReceived, etc.),
+/// others come from the owning actor (WriteStdin, SendSignal, Subscribe, etc.).
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum ProcessEvent {
+    // --- Driver-sourced events ---
+    /// The process spawned successfully.
+    Started,
+    /// The kill timeout fired (process didn't exit after SIGTERM).
+    KillTimeout,
+    /// The process failed to spawn.
+    SpawnFailed { reason: String },
+    /// Output received on stdout or stderr.
+    OutputReceived { data: Vec<u8>, is_stderr: bool },
+    /// The process exited.
+    Exited { status: ExitStatus },
+    /// Connection to the process was lost unexpectedly.
+    ConnectionLost { reason: String },
+
+    // --- Driver acknowledgement events ---
+    /// Stdin bytes were successfully written.
+    StdinWritten { byte_count: usize },
+    /// A signal was delivered.
+    SignalSent,
+    /// The PTY was resized.
+    PtyResized,
+
+    // --- Actor-sourced events ---
+    /// Write data to the process's stdin.
+    WriteStdin { data: Vec<u8> },
+    /// Send a signal to the process.
+    SendSignal { signal: Signal },
+    /// Resize the process's PTY.
+    ResizePty { size: PtySize },
+    /// Close the process's stdin.
+    CloseStdin,
+    /// Request a graceful close of the process.
+    CloseRequested,
+    /// Subscribe an actor to process notifications.
+    Subscribe { address: ActorAddress },
+    /// Unsubscribe an actor from process notifications.
+    Unsubscribe { address: ActorAddress },
+}
+
+impl ProcessEvent {
+    /// Human-readable name for error messages.
+    pub fn name(&self) -> &'static str {
+        match self {
+            Self::Started => "Started",
+            Self::KillTimeout => "KillTimeout",
+            Self::SpawnFailed { .. } => "SpawnFailed",
+            Self::OutputReceived { .. } => "OutputReceived",
+            Self::Exited { .. } => "Exited",
+            Self::ConnectionLost { .. } => "ConnectionLost",
+            Self::StdinWritten { .. } => "StdinWritten",
+            Self::SignalSent => "SignalSent",
+            Self::PtyResized => "PtyResized",
+            Self::WriteStdin { .. } => "WriteStdin",
+            Self::SendSignal { .. } => "SendSignal",
+            Self::ResizePty { .. } => "ResizePty",
+            Self::CloseStdin => "CloseStdin",
+            Self::CloseRequested => "CloseRequested",
+            Self::Subscribe { .. } => "Subscribe",
+            Self::Unsubscribe { .. } => "Unsubscribe",
+        }
+    }
+}

crates/process/src/lib.rs

@@ -0,0 +1,35 @@
+pub mod action;
+pub mod actor;
+pub mod driver;
+pub mod event;
+pub mod local;
+pub mod message;
+pub mod mock;
+pub mod queue;
+pub mod session;
+pub mod spawn;
+pub mod subscriber;
+pub mod types;
+pub mod waker;
+
+#[cfg(feature = "ssh")]
+pub mod ssh;
+
+pub use action::{OutputStream, ProcessAction};
+pub use actor::ProcessActor;
+pub use driver::ProcessDriver;
+pub use event::ProcessEvent;
+pub use local::LocalDriver;
+pub use message::{ProcessCommand, ProcessNotification};
+pub use mock::MockDriver;
+pub use queue::EventQueue;
+pub use session::{ProcessSession, ProcessState};
+pub use spawn::{spawn_local_process, spawn_process};
+pub use subscriber::SubscriberSet;
+pub use types::{ExitStatus, FlowControl, ProcessError, ProcessMode, ProcessSpec, PtySize, Signal};
+pub use waker::ProcessWaker;
+
+#[cfg(feature = "ssh")]
+pub use ssh::{SshConfig, SshDriver};
+#[cfg(feature = "ssh")]
+pub use spawn::spawn_ssh_process;

crates/process/src/local/mod.rs

@@ -0,0 +1,182 @@
+mod pipes;
+mod signal;
+mod wait;
+
+use std::io::Write;
+use std::process::{Child, ChildStdin, Command, Stdio};
+use std::sync::{Arc, OnceLock};
+use std::thread::{self, JoinHandle};
+
+use crate::action::ProcessAction;
+use crate::driver::ProcessDriver;
+use crate::event::ProcessEvent;
+use crate::queue::EventQueue;
+use crate::types::ProcessSpec;
+use crate::waker::ProcessWaker;
+
+/// A `ProcessDriver` that spawns real OS subprocesses via `std::process::Command`.
+///
+/// Background threads read stdout/stderr and wait for process exit,
+/// pushing events into a shared `EventQueue`. The actor polls via `poll()`.
+pub struct LocalDriver {
+    queue: EventQueue,
+    waker_slot: Arc<OnceLock<ProcessWaker>>,
+    child: Option<Child>,
+    stdin: Option<ChildStdin>,
+    _reader_threads: Vec<JoinHandle<()>>,
+    _wait_thread: Option<JoinHandle<()>>,
+}
+
+impl LocalDriver {
+    pub fn new(queue: EventQueue, waker_slot: Arc<OnceLock<ProcessWaker>>) -> Self {
+        Self {
+            queue,
+            waker_slot,
+            child: None,
+            stdin: None,
+            _reader_threads: Vec::new(),
+            _wait_thread: None,
+        }
+    }
+
+    fn spawn_process(&mut self, spec: &ProcessSpec) {
+        let mut cmd = Command::new(&spec.command);
+        cmd.args(&spec.args);
+        for (k, v) in &spec.env {
+            cmd.env(k, v);
+        }
+        if let Some(ref dir) = spec.working_dir {
+            cmd.current_dir(dir);
+        }
+        cmd.stdin(Stdio::piped())
+            .stdout(Stdio::piped())
+            .stderr(Stdio::piped());
+
+        match cmd.spawn() {
+            Ok(mut child) => {
+                let pid = child.id();
+
+                // Take the stdin handle
+                self.stdin = child.stdin.take();
+
+                // Spawn stdout reader thread
+                if let Some(stdout) = child.stdout.take() {
+                    let queue = self.queue.clone();
+                    let waker = self.waker_slot.clone();
+                    self._reader_threads.push(
+                        thread::Builder::new()
+                            .name(format!("proc-{}-stdout", pid))
+                            .spawn(move || pipes::read_pipe(stdout, false, queue, waker))
+                            .expect("failed to spawn stdout reader"),
+                    );
+                }
+
+                // Spawn stderr reader thread
+                if let Some(stderr) = child.stderr.take() {
+                    let queue = self.queue.clone();
+                    let waker = self.waker_slot.clone();
+                    self._reader_threads.push(
+                        thread::Builder::new()
+                            .name(format!("proc-{}-stderr", pid))
+                            .spawn(move || pipes::read_pipe(stderr, true, queue, waker))
+                            .expect("failed to spawn stderr reader"),
+                    );
+                }
+
+                // Spawn wait thread
+                let queue = self.queue.clone();
+                let waker = self.waker_slot.clone();
+                self._wait_thread = Some(
+                    thread::Builder::new()
+                        .name(format!("proc-{}-wait", pid))
+                        .spawn(move || wait::wait_for_exit(pid, queue, waker))
+                        .expect("failed to spawn wait thread"),
+                );
+
+                self.child = Some(child);
+                self.queue.push(ProcessEvent::Started);
+            }
+            Err(e) => {
+                self.queue.push(ProcessEvent::SpawnFailed {
+                    reason: e.to_string(),
+                });
+            }
+        }
+    }
+}
+
+impl ProcessDriver for LocalDriver {
+    fn execute(&mut self, action: ProcessAction) {
+        match action {
+            ProcessAction::SpawnProcess { spec } => {
+                self.spawn_process(&spec);
+            }
+            ProcessAction::WriteStdin { data } => {
+                if let Some(ref mut stdin) = self.stdin {
+                    match stdin.write_all(&data) {
+                        Ok(()) => {
+                            self.queue.push(ProcessEvent::StdinWritten {
+                                byte_count: data.len(),
+                            });
+                        }
+                        Err(e) => {
+                            self.queue.push(ProcessEvent::ConnectionLost {
+                                reason: format!("stdin write failed: {}", e),
+                            });
+                        }
+                    }
+                }
+            }
+            ProcessAction::SendSignal { signal } => {
+                if let Some(ref child) = self.child {
+                    let pid = child.id();
+                    match signal::send_signal(pid, signal) {
+                        Ok(()) => {
+                            self.queue.push(ProcessEvent::SignalSent);
+                        }
+                        Err(reason) => {
+                            self.queue.push(ProcessEvent::ConnectionLost { reason });
+                        }
+                    }
+                }
+            }
+            ProcessAction::ResizePty { .. } => {
+                // No-op for Phase 1 (pipes only, no PTY support)
+                self.queue.push(ProcessEvent::PtyResized);
+            }
+            ProcessAction::CloseStdin => {
+                // Drop the stdin handle to close the pipe
+                self.stdin.take();
+            }
+            ProcessAction::ScheduleKillTimeout { duration } => {
+                let queue = self.queue.clone();
+                let waker = self.waker_slot.clone();
+                thread::spawn(move || {
+                    thread::sleep(duration);
+                    queue.push(ProcessEvent::KillTimeout);
+                    if let Some(w) = waker.get() {
+                        w.wake();
+                    }
+                });
+            }
+            // Notification actions are not driver commands
+            _ => {}
+        }
+    }
+
+    fn poll(&mut self) -> Vec<ProcessEvent> {
+        self.queue.drain()
+    }
+}
+
+impl Drop for LocalDriver {
+    fn drop(&mut self) {
+        // Close stdin to let the process know we're done
+        self.stdin.take();
+        // Kill the process if still alive
+        if let Some(ref mut child) = self.child {
+            let _ = child.kill();
+            let _ = child.wait();
+        }
+    }
+}

crates/process/src/local/pipes.rs

@@ -0,0 +1,34 @@
+use std::io::Read;
+use std::sync::{Arc, OnceLock};
+
+use crate::event::ProcessEvent;
+use crate::queue::EventQueue;
+use crate::waker::ProcessWaker;
+
+/// Read from a pipe in a loop, pushing events to the queue and waking the actor.
+///
+/// Runs in a background thread. Exits when the pipe reaches EOF or errors.
+pub(crate) fn read_pipe(
+    mut pipe: impl Read + Send + 'static,
+    is_stderr: bool,
+    queue: EventQueue,
+    waker: Arc<OnceLock<ProcessWaker>>,
+) {
+    let mut buf = [0u8; 8192];
+    loop {
+        match pipe.read(&mut buf) {
+            Ok(0) => break, // EOF
+            Ok(n) => {
+                queue.push(ProcessEvent::OutputReceived {
+                    data: buf[..n].to_vec(),
+                    is_stderr,
+                });
+                if let Some(w) = waker.get() {
+                    w.wake();
+                }
+            }
+            Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue,
+            Err(_) => break,
+        }
+    }
+}

crates/process/src/local/signal.rs

@@ -0,0 +1,30 @@
+use crate::types::Signal;
+
+/// Map a `Signal` enum variant to the corresponding libc signal constant.
+pub(crate) fn signal_to_libc(signal: Signal) -> libc::c_int {
+    match signal {
+        Signal::Terminate => libc::SIGTERM,
+        Signal::Kill => libc::SIGKILL,
+        Signal::Hangup => libc::SIGHUP,
+        Signal::Interrupt => libc::SIGINT,
+        Signal::Other(n) => n,
+    }
+}
+
+/// Send a signal to a process by PID. Returns `Ok(())` on success.
+pub(crate) fn send_signal(pid: u32, signal: Signal) -> Result<(), String> {
+    let sig = signal_to_libc(signal);
+    // Safety: kill() is safe to call with any pid/signal combo;
+    // it returns -1 on error which we check.
+    let ret = unsafe { libc::kill(pid as libc::pid_t, sig) };
+    if ret == 0 {
+        Ok(())
+    } else {
+        Err(format!(
+            "kill({}, {}) failed: {}",
+            pid,
+            sig,
+            std::io::Error::last_os_error()
+        ))
+    }
+}

crates/process/src/local/wait.rs

@@ -0,0 +1,41 @@
+use std::sync::{Arc, OnceLock};
+
+use crate::event::ProcessEvent;
+use crate::queue::EventQueue;
+use crate::types::ExitStatus;
+use crate::waker::ProcessWaker;
+
+/// Wait for a child process to exit, then push the appropriate event.
+///
+/// Runs in a background thread. Uses `libc::waitpid` for accurate exit status.
+pub(crate) fn wait_for_exit(
+    pid: u32,
+    queue: EventQueue,
+    waker: Arc<OnceLock<ProcessWaker>>,
+) {
+    let mut status: libc::c_int = 0;
+    let ret = unsafe { libc::waitpid(pid as libc::pid_t, &mut status, 0) };
+
+    let exit_status = if ret < 0 {
+        ExitStatus::Unknown
+    } else {
+        decode_wait_status(status)
+    };
+
+    queue.push(ProcessEvent::Exited {
+        status: exit_status,
+    });
+    if let Some(w) = waker.get() {
+        w.wake();
+    }
+}
+
+fn decode_wait_status(status: libc::c_int) -> ExitStatus {
+    if libc::WIFEXITED(status) {
+        ExitStatus::Code(libc::WEXITSTATUS(status))
+    } else if libc::WIFSIGNALED(status) {
+        ExitStatus::Signal(libc::WTERMSIG(status))
+    } else {
+        ExitStatus::Unknown
+    }
+}

crates/process/src/message.rs

@@ -0,0 +1,49 @@
+use swactor::actor::ActorAddress;
+
+use crate::action::OutputStream;
+use crate::types::{ExitStatus, ProcessError, PtySize, Signal};
+
+/// Commands sent to a process actor.
+#[derive(Debug, Clone)]
+pub enum ProcessCommand {
+    /// Write data to the process's stdin.
+    WriteStdin { data: Vec<u8> },
+    /// Send a signal to the process.
+    SendSignal { signal: Signal },
+    /// Resize the process's PTY.
+    ResizePty { size: PtySize },
+    /// Close the process's stdin pipe.
+    CloseStdin,
+    /// Request a graceful close of the process.
+    Close,
+    /// Subscribe to process notifications.
+    Subscribe { address: ActorAddress },
+    /// Unsubscribe from process notifications.
+    Unsubscribe { address: ActorAddress },
+    /// Internal: sent by the waker to trigger event draining.
+    #[doc(hidden)]
+    PollTick,
+}
+
+/// Notifications sent from a process actor to subscribers.
+#[derive(Debug, Clone)]
+pub enum ProcessNotification {
+    /// The process started successfully.
+    Started { process: ActorAddress },
+    /// Output was received from the process.
+    Output {
+        process: ActorAddress,
+        data: Vec<u8>,
+        stream: OutputStream,
+    },
+    /// The process exited.
+    Exited {
+        process: ActorAddress,
+        status: ExitStatus,
+    },
+    /// An error occurred.
+    Error {
+        process: ActorAddress,
+        error: ProcessError,
+    },
+}

crates/process/src/mock.rs

@@ -0,0 +1,61 @@
+use std::collections::VecDeque;
+
+use crate::action::ProcessAction;
+use crate::driver::ProcessDriver;
+use crate::event::ProcessEvent;
+
+/// A test-oriented driver that records executed actions and lets you inject events.
+pub struct MockDriver {
+    pending_events: VecDeque<ProcessEvent>,
+    executed_actions: Vec<ProcessAction>,
+}
+
+impl MockDriver {
+    pub fn new() -> Self {
+        Self {
+            pending_events: VecDeque::new(),
+            executed_actions: Vec::new(),
+        }
+    }
+
+    /// Queue a single event to be returned by the next `poll()`.
+    pub fn inject(&mut self, event: ProcessEvent) {
+        self.pending_events.push_back(event);
+    }
+
+    /// Queue multiple events to be returned by subsequent `poll()` calls.
+    pub fn inject_many(&mut self, events: impl IntoIterator<Item = ProcessEvent>) {
+        self.pending_events.extend(events);
+    }
+
+    /// View all actions that have been executed so far.
+    pub fn executed_actions(&self) -> &[ProcessAction] {
+        &self.executed_actions
+    }
+
+    /// Take all executed actions, clearing the internal log.
+    pub fn take_executed_actions(&mut self) -> Vec<ProcessAction> {
+        std::mem::take(&mut self.executed_actions)
+    }
+
+    /// Number of events waiting to be polled.
+    pub fn pending_event_count(&self) -> usize {
+        self.pending_events.len()
+    }
+}
+
+impl Default for MockDriver {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl ProcessDriver for MockDriver {
+    fn execute(&mut self, action: ProcessAction) {
+        self.executed_actions.push(action);
+    }
+
+    fn poll(&mut self) -> Vec<ProcessEvent> {
+        self.pending_events.drain(..).collect()
+    }
+}

crates/process/src/queue.rs

@@ -0,0 +1,42 @@
+use std::sync::Arc;
+
+use crossbeam_queue::SegQueue;
+
+use crate::event::ProcessEvent;
+
+/// Thread-safe queue for buffering process events from I/O threads.
+///
+/// Cloneable via inner `Arc` — I/O threads push events, the driver's
+/// `poll()` drains them.
+#[derive(Clone)]
+pub struct EventQueue {
+    inner: Arc<SegQueue<ProcessEvent>>,
+}
+
+impl EventQueue {
+    pub fn new() -> Self {
+        Self {
+            inner: Arc::new(SegQueue::new()),
+        }
+    }
+
+    /// Push an event (called from I/O threads).
+    pub fn push(&self, event: ProcessEvent) {
+        self.inner.push(event);
+    }
+
+    /// Drain all pending events (called from driver's `poll()`).
+    pub fn drain(&self) -> Vec<ProcessEvent> {
+        let mut events = Vec::new();
+        while let Some(event) = self.inner.pop() {
+            events.push(event);
+        }
+        events
+    }
+}
+
+impl Default for EventQueue {
+    fn default() -> Self {
+        Self::new()
+    }
+}

crates/process/src/session.rs

@@ -0,0 +1,363 @@
+use std::collections::VecDeque;
+
+use crate::action::{OutputStream, ProcessAction};
+use crate::event::ProcessEvent;
+use crate::subscriber::SubscriberSet;
+use crate::types::{ExitStatus, FlowControl, ProcessError, ProcessMode, ProcessSpec, Signal};
+
+/// The lifecycle states of a process session.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ProcessState {
+    Starting,
+    Running,
+    Stopping,
+    Exited,
+}
+
+impl ProcessState {
+    pub fn name(&self) -> &'static str {
+        match self {
+            Self::Starting => "Starting",
+            Self::Running => "Running",
+            Self::Stopping => "Stopping",
+            Self::Exited => "Exited",
+        }
+    }
+}
+
+/// Pure-logic state machine for managing a process lifecycle.
+///
+/// Created via `new()` which returns the session plus initial actions (SpawnProcess).
+/// Drive it forward by calling `apply(event)` which returns actions to execute.
+pub struct ProcessSession {
+    spec: ProcessSpec,
+    state: ProcessState,
+    subscribers: SubscriberSet,
+    flow: FlowControl,
+    exit_status: Option<ExitStatus>,
+    stdin_closed: bool,
+    close_requested_before_start: bool,
+    stdin_buffer: VecDeque<Vec<u8>>,
+    stdin_buffer_bytes: usize,
+}
+
+impl ProcessSession {
+    /// Create a new session. Returns the session and the initial actions to execute
+    /// (always a single `SpawnProcess` action).
+    pub fn new(spec: ProcessSpec) -> (Self, Vec<ProcessAction>) {
+        let actions = vec![ProcessAction::SpawnProcess { spec: spec.clone() }];
+        let session = Self {
+            spec,
+            state: ProcessState::Starting,
+            subscribers: SubscriberSet::new(),
+            flow: FlowControl::default(),
+            exit_status: None,
+            stdin_closed: false,
+            close_requested_before_start: false,
+            stdin_buffer: VecDeque::new(),
+            stdin_buffer_bytes: 0,
+        };
+        (session, actions)
+    }
+
+    /// Apply an event and return the resulting actions.
+    pub fn apply(&mut self, event: ProcessEvent) -> Vec<ProcessAction> {
+        // Subscribe/Unsubscribe handled in all states
+        match &event {
+            ProcessEvent::Subscribe { address } => {
+                self.subscribers.add(*address);
+                return vec![];
+            }
+            ProcessEvent::Unsubscribe { address } => {
+                self.subscribers.remove(address);
+                return vec![];
+            }
+            _ => {}
+        }
+
+        // Driver acks — silently consumed in all states
+        match &event {
+            ProcessEvent::StdinWritten { byte_count } => {
+                self.flow.pending_stdin_bytes =
+                    self.flow.pending_stdin_bytes.saturating_sub(*byte_count);
+                return self.drain_stdin_buffer();
+            }
+            ProcessEvent::SignalSent | ProcessEvent::PtyResized => {
+                return vec![];
+            }
+            _ => {}
+        }
+
+        // KillTimeout — handled in all states before per-state dispatch
+        if matches!(event, ProcessEvent::KillTimeout) {
+            return if self.state == ProcessState::Stopping {
+                vec![ProcessAction::SendSignal { signal: Signal::Kill }]
+            } else {
+                vec![]
+            };
+        }
+
+        // Dispatch to per-state handler
+        match self.state {
+            ProcessState::Starting => self.handle_starting(event),
+            ProcessState::Running => self.handle_running(event),
+            ProcessState::Stopping => self.handle_stopping(event),
+            ProcessState::Exited => self.handle_exited(event),
+        }
+    }
+
+    // --- Per-state handlers ---
+
+    fn handle_starting(&mut self, event: ProcessEvent) -> Vec<ProcessAction> {
+        match event {
+            ProcessEvent::Started => {
+                self.state = ProcessState::Running;
+                let mut actions = vec![ProcessAction::NotifyStarted {
+                    subscribers: self.subscribers.snapshot(),
+                }];
+                // If close was requested before the process started, transition to Stopping
+                if self.close_requested_before_start {
+                    self.state = ProcessState::Stopping;
+                    actions.push(ProcessAction::SendSignal {
+                        signal: Signal::Terminate,
+                    });
+                    if let Some(duration) = self.spec.kill_timeout {
+                        actions.push(ProcessAction::ScheduleKillTimeout { duration });
+                    }
+                }
+                actions
+            }
+            ProcessEvent::SpawnFailed { reason } => {
+                self.state = ProcessState::Exited;
+                vec![
+                    ProcessAction::NotifyError {
+                        subscribers: self.subscribers.snapshot(),
+                        error: ProcessError::SpawnFailed { reason },
+                    },
+                    ProcessAction::SelfTerminate,
+                ]
+            }
+            ProcessEvent::CloseRequested => {
+                self.close_requested_before_start = true;
+                vec![]
+            }
+            _ => self.invalid_state_error(&event),
+        }
+    }
+
+    fn handle_running(&mut self, event: ProcessEvent) -> Vec<ProcessAction> {
+        match event {
+            ProcessEvent::OutputReceived { data, is_stderr } => {
+                let stream = if is_stderr {
+                    OutputStream::Stderr
+                } else {
+                    OutputStream::Stdout
+                };
+                vec![ProcessAction::NotifyOutput {
+                    subscribers: self.subscribers.snapshot(),
+                    data,
+                    stream,
+                }]
+            }
+            ProcessEvent::Exited { status } => {
+                self.enter_exited(status)
+            }
+            ProcessEvent::ConnectionLost { reason } => {
+                self.state = ProcessState::Exited;
+                self.exit_status = Some(ExitStatus::Unknown);
+                self.clear_stdin_buffer();
+                vec![
+                    ProcessAction::NotifyError {
+                        subscribers: self.subscribers.snapshot(),
+                        error: ProcessError::ConnectionLost { reason },
+                    },
+                    ProcessAction::SelfTerminate,
+                ]
+            }
+            ProcessEvent::WriteStdin { data } => {
+                if self.stdin_closed {
+                    return self.notify_error(ProcessError::InvalidState {
+                        attempted: "WriteStdin",
+                        current_state: "Running (stdin closed)",
+                    });
+                }
+                // Backpressure: buffer if over limit
+                if let Some(limit) = self.spec.stdin_buffer_limit {
+                    if self.flow.pending_stdin_bytes >= limit {
+                        self.stdin_buffer_bytes += data.len();
+                        self.stdin_buffer.push_back(data);
+                        return vec![];
+                    }
+                }
+                self.flow.pending_stdin_bytes += data.len();
+                vec![ProcessAction::WriteStdin { data }]
+            }
+            ProcessEvent::SendSignal { signal } => {
+                vec![ProcessAction::SendSignal { signal }]
+            }
+            ProcessEvent::ResizePty { size } => {
+                vec![ProcessAction::ResizePty { size }]
+            }
+            ProcessEvent::CloseStdin => {
+                if self.stdin_closed {
+                    return vec![];
+                }
+                self.stdin_closed = true;
+                self.clear_stdin_buffer();
+                vec![ProcessAction::CloseStdin]
+            }
+            ProcessEvent::CloseRequested => {
+                self.state = ProcessState::Stopping;
+                self.clear_stdin_buffer();
+                let mut actions = vec![ProcessAction::SendSignal {
+                    signal: Signal::Terminate,
+                }];
+                if let Some(duration) = self.spec.kill_timeout {
+                    actions.push(ProcessAction::ScheduleKillTimeout { duration });
+                }
+                actions
+            }
+            _ => self.invalid_state_error(&event),
+        }
+    }
+
+    fn handle_stopping(&mut self, event: ProcessEvent) -> Vec<ProcessAction> {
+        match event {
+            ProcessEvent::OutputReceived { data, is_stderr } => {
+                let stream = if is_stderr {
+                    OutputStream::Stderr
+                } else {
+                    OutputStream::Stdout
+                };
+                vec![ProcessAction::NotifyOutput {
+                    subscribers: self.subscribers.snapshot(),
+                    data,
+                    stream,
+                }]
+            }
+            ProcessEvent::Exited { status } => {
+                self.enter_exited(status)
+            }
+            ProcessEvent::ConnectionLost { reason } => {
+                self.state = ProcessState::Exited;
+                self.exit_status = Some(ExitStatus::Unknown);
+                vec![
+                    ProcessAction::NotifyError {
+                        subscribers: self.subscribers.snapshot(),
+                        error: ProcessError::ConnectionLost { reason },
+                    },
+                    ProcessAction::SelfTerminate,
+                ]
+            }
+            ProcessEvent::SendSignal { signal } => {
+                // Escalation (e.g., Kill after Terminate) is allowed in Stopping
+                vec![ProcessAction::SendSignal { signal }]
+            }
+            ProcessEvent::CloseStdin => {
+                if self.stdin_closed {
+                    return vec![];
+                }
+                self.stdin_closed = true;
+                vec![ProcessAction::CloseStdin]
+            }
+            ProcessEvent::CloseRequested => {
+                // Already stopping, no-op
+                vec![]
+            }
+            _ => self.invalid_state_error(&event),
+        }
+    }
+
+    fn handle_exited(&mut self, event: ProcessEvent) -> Vec<ProcessAction> {
+        // Everything in Exited is invalid — produce an error.
+        // (Acks and Subscribe/Unsubscribe are already handled before dispatch.)
+        self.invalid_state_error(&event)
+    }
+
+    // --- Helpers ---
+
+    fn enter_exited(&mut self, status: ExitStatus) -> Vec<ProcessAction> {
+        self.state = ProcessState::Exited;
+        self.exit_status = Some(status);
+        self.clear_stdin_buffer();
+        vec![
+            ProcessAction::NotifyExited {
+                subscribers: self.subscribers.snapshot(),
+                status,
+            },
+            ProcessAction::SelfTerminate,
+        ]
+    }
+
+    fn clear_stdin_buffer(&mut self) {
+        self.stdin_buffer.clear();
+        self.stdin_buffer_bytes = 0;
+    }
+
+    fn drain_stdin_buffer(&mut self) -> Vec<ProcessAction> {
+        let limit = match self.spec.stdin_buffer_limit {
+            Some(limit) => limit,
+            None => return vec![],
+        };
+        let mut actions = Vec::new();
+        while self.flow.pending_stdin_bytes < limit {
+            match self.stdin_buffer.pop_front() {
+                Some(data) => {
+                    self.stdin_buffer_bytes -= data.len();
+                    self.flow.pending_stdin_bytes += data.len();
+                    actions.push(ProcessAction::WriteStdin { data });
+                }
+                None => break,
+            }
+        }
+        actions
+    }
+
+    fn invalid_state_error(&self, event: &ProcessEvent) -> Vec<ProcessAction> {
+        self.notify_error(ProcessError::InvalidState {
+            attempted: event.name(),
+            current_state: self.state.name(),
+        })
+    }
+
+    fn notify_error(&self, error: ProcessError) -> Vec<ProcessAction> {
+        vec![ProcessAction::NotifyError {
+            subscribers: self.subscribers.snapshot(),
+            error,
+        }]
+    }
+
+    // --- Query methods ---
+
+    pub fn state(&self) -> ProcessState {
+        self.state
+    }
+
+    pub fn spec(&self) -> &ProcessSpec {
+        &self.spec
+    }
+
+    pub fn mode(&self) -> ProcessMode {
+        self.spec.mode
+    }
+
+    pub fn exit_status(&self) -> Option<ExitStatus> {
+        self.exit_status
+    }
+
+    pub fn flow_control(&self) -> &FlowControl {
+        &self.flow
+    }
+
+    pub fn subscriber_count(&self) -> usize {
+        self.subscribers.count()
+    }
+
+    pub fn stdin_closed(&self) -> bool {
+        self.stdin_closed
+    }
+
+    pub fn stdin_buffer_bytes(&self) -> usize {
+        self.stdin_buffer_bytes
+    }
+}

crates/process/src/spawn.rs

@@ -0,0 +1,89 @@
+use std::sync::{Arc, OnceLock};
+
+use swactor::actor::{ActorAddress, Ctx};
+use swactor::runtime::ExternalSender;
+use swactor::Error;
+
+use crate::actor::ProcessActor;
+use crate::driver::ProcessDriver;
+use crate::local::LocalDriver;
+use crate::message::ProcessCommand;
+use crate::queue::EventQueue;
+use crate::session::ProcessSession;
+use crate::types::ProcessSpec;
+use crate::waker::ProcessWaker;
+
+/// Spawn a process actor using the real `LocalDriver` (OS subprocess).
+///
+/// Creates a `ProcessActor<LocalDriver>`, spawns it in the runtime, and
+/// wires up the waker so that I/O thread events automatically wake the actor.
+///
+/// Returns the actor's address. Send `ProcessCommand` messages to control it.
+pub fn spawn_local_process(
+    ctx: &Ctx,
+    sender: &ExternalSender,
+    spec: ProcessSpec,
+) -> Result<ActorAddress, Error> {
+    let waker_slot = Arc::new(OnceLock::new());
+    let queue = EventQueue::new();
+    let driver = LocalDriver::new(queue, waker_slot.clone());
+    spawn_process_inner(ctx, sender, spec, driver, waker_slot)
+}
+
+/// Spawn a process actor with a custom driver.
+///
+/// Useful for testing with `MockDriver` or other custom drivers while
+/// still getting the full actor integration (waker, lifecycle, etc.).
+pub fn spawn_process<D: ProcessDriver + 'static>(
+    ctx: &Ctx,
+    sender: &ExternalSender,
+    spec: ProcessSpec,
+    driver: D,
+    waker_slot: Arc<OnceLock<ProcessWaker>>,
+) -> Result<ActorAddress, Error> {
+    spawn_process_inner(ctx, sender, spec, driver, waker_slot)
+}
+
+/// Spawn a process actor using the `SshDriver` (remote host via SSH).
+///
+/// Requires a tokio runtime handle (e.g. from `IrohDriver::tokio_handle()`)
+/// and SSH connection config.
+#[cfg(feature = "ssh")]
+pub fn spawn_ssh_process(
+    ctx: &Ctx,
+    sender: &ExternalSender,
+    spec: ProcessSpec,
+    tokio_handle: tokio::runtime::Handle,
+    ssh_config: crate::ssh::SshConfig,
+) -> Result<ActorAddress, Error> {
+    let waker_slot = Arc::new(OnceLock::new());
+    let queue = EventQueue::new();
+    let driver = crate::ssh::SshDriver::new(queue, waker_slot.clone(), tokio_handle, ssh_config);
+    spawn_process_inner(ctx, sender, spec, driver, waker_slot)
+}
+
+fn spawn_process_inner<D: ProcessDriver + 'static>(
+    ctx: &Ctx,
+    sender: &ExternalSender,
+    spec: ProcessSpec,
+    driver: D,
+    waker_slot: Arc<OnceLock<ProcessWaker>>,
+) -> Result<ActorAddress, Error> {
+    let (session, initial_actions) = ProcessSession::new(spec);
+    let actor = ProcessActor::new(session, driver, initial_actions, waker_slot.clone());
+    let addr = ctx.spawn(actor)?;
+
+    // Now that we have the address, fill the waker
+    let sender = sender.clone();
+    let waker = ProcessWaker::new(move || {
+        let _ = sender.send_to(addr, ProcessCommand::PollTick);
+    });
+    waker_slot
+        .set(waker.clone())
+        .expect("waker slot already set");
+
+    // Flush any events from the startup race window
+    waker.wake();
+
+    Ok(addr)
+}

crates/process/src/ssh/config.rs

@@ -0,0 +1,32 @@
+use std::path::PathBuf;
+
+/// Configuration for connecting to a remote host over SSH.
+pub struct SshConfig {
+    pub host: String,
+    pub port: u16,
+    pub username: String,
+    pub key_file: PathBuf,
+    pub key_passphrase: Option<String>,
+}
+
+impl SshConfig {
+    pub fn new(host: impl Into<String>, username: impl Into<String>, key_file: PathBuf) -> Self {
+        Self {
+            host: host.into(),
+            port: 22,
+            username: username.into(),
+            key_file,
+            key_passphrase: None,
+        }
+    }
+
+    pub fn with_port(mut self, port: u16) -> Self {
+        self.port = port;
+        self
+    }
+
+    pub fn with_passphrase(mut self, passphrase: impl Into<String>) -> Self {
+        self.key_passphrase = Some(passphrase.into());
+        self
+    }
+}

crates/process/src/ssh/handler.rs

@@ -0,0 +1,17 @@
+use russh::client;
+use russh_keys::key::PublicKey;
+
+/// Minimal SSH client handler that accepts all host keys.
+pub(super) struct SshHandler;
+
+#[async_trait::async_trait]
+impl client::Handler for SshHandler {
+    type Error = russh::Error;
+
+    async fn check_server_key(
+        &mut self,
+        _server_public_key: &PublicKey,
+    ) -> Result<bool, Self::Error> {
+        Ok(true)
+    }
+}

crates/process/src/ssh/mod.rs

@@ -0,0 +1,121 @@
+pub mod config;
+mod handler;
+mod task;
+
+use std::sync::{Arc, OnceLock};
+
+use tokio::sync::mpsc;
+
+use crate::action::ProcessAction;
+use crate::driver::ProcessDriver;
+use crate::event::ProcessEvent;
+use crate::queue::EventQueue;
+use crate::waker::ProcessWaker;
+
+pub use config::SshConfig;
+use task::SshCommand;
+
+/// A `ProcessDriver` that runs processes on remote hosts over SSH.
+///
+/// Commands are sent via a tokio mpsc channel to a background async task
+/// that manages the SSH connection. Events flow back through the shared
+/// `EventQueue` + `ProcessWaker` (same pattern as `LocalDriver`).
+pub struct SshDriver {
+    queue: EventQueue,
+    waker_slot: Arc<OnceLock<ProcessWaker>>,
+    command_tx: Option<mpsc::UnboundedSender<SshCommand>>,
+    command_rx: Option<mpsc::UnboundedReceiver<SshCommand>>,
+    tokio_handle: tokio::runtime::Handle,
+    ssh_config: SshConfig,
+    task_handle: Option<tokio::task::JoinHandle<()>>,
+}
+
+impl SshDriver {
+    pub fn new(
+        queue: EventQueue,
+        waker_slot: Arc<OnceLock<ProcessWaker>>,
+        tokio_handle: tokio::runtime::Handle,
+        ssh_config: SshConfig,
+    ) -> Self {
+        let (tx, rx) = mpsc::unbounded_channel();
+        Self {
+            queue,
+            waker_slot,
+            command_tx: Some(tx),
+            command_rx: Some(rx),
+            tokio_handle,
+            ssh_config,
+            task_handle: None,
+        }
+    }
+}
+
+impl ProcessDriver for SshDriver {
+    fn execute(&mut self, action: ProcessAction) {
+        match action {
+            ProcessAction::SpawnProcess { spec } => {
+                let Some(rx) = self.command_rx.take() else {
+                    return;
+                };
+                let queue = self.queue.clone();
+                let waker_slot = self.waker_slot.clone();
+                // Move the ssh_config out — we only need it once for connection
+                let config = SshConfig {
+                    host: self.ssh_config.host.clone(),
+                    port: self.ssh_config.port,
+                    username: self.ssh_config.username.clone(),
+                    key_file: self.ssh_config.key_file.clone(),
+                    key_passphrase: self.ssh_config.key_passphrase.clone(),
+                };
+                self.task_handle = Some(self.tokio_handle.spawn(
+                    task::run_ssh_session(config, spec, queue, waker_slot, rx),
+                ));
+            }
+            ProcessAction::WriteStdin { data } => {
+                if let Some(ref tx) = self.command_tx {
+                    let _ = tx.send(SshCommand::WriteStdin(data));
+                }
+            }
+            ProcessAction::SendSignal { signal } => {
+                if let Some(ref tx) = self.command_tx {
+                    let _ = tx.send(SshCommand::SendSignal(signal));
+                }
+            }
+            ProcessAction::ResizePty { size } => {
+                if let Some(ref tx) = self.command_tx {
+                    let _ = tx.send(SshCommand::ResizePty {
+                        cols: size.cols,
+                        rows: size.rows,
+                    });
+                }
+            }
+            ProcessAction::CloseStdin => {
+                if let Some(ref tx) = self.command_tx {
+                    let _ = tx.send(SshCommand::CloseStdin);
+                }
+            }
+            ProcessAction::ScheduleKillTimeout { duration } => {
+                if let Some(ref tx) = self.command_tx {
+                    let _ = tx.send(SshCommand::ScheduleKillTimeout(duration));
+                }
+            }
+            // Notification actions are not driver commands
+            _ => {}
+        }
+    }
+
+    fn poll(&mut self) -> Vec<ProcessEvent> {
+        self.queue.drain()
+    }
+}
+
+impl Drop for SshDriver {
+    fn drop(&mut self) {
+        // Drop sender to signal the task to shut down
+        self.command_tx.take();
+        // Abort the background task if still running
+        if let Some(handle) = self.task_handle.take() {
+            handle.abort();
+        }
+    }
+}

crates/process/src/ssh/task.rs

@@ -0,0 +1,317 @@
+use std::sync::{Arc, OnceLock};
+use std::time::Duration;
+
+use russh::{ChannelMsg, Sig};
+use tokio::sync::mpsc;
+use tokio::time::{Instant, sleep_until};
+
+use crate::event::ProcessEvent;
+use crate::queue::EventQueue;
+use crate::types::{ExitStatus, ProcessMode, ProcessSpec, Signal};
+use crate::waker::ProcessWaker;
+
+use super::config::SshConfig;
+use super::handler::SshHandler;
+
+/// Commands sent from the SshDriver to the background task.
+pub(super) enum SshCommand {
+    WriteStdin(Vec<u8>),
+    SendSignal(Signal),
+    ResizePty { cols: u16, rows: u16 },
+    CloseStdin,
+    ScheduleKillTimeout(Duration),
+}
+
+/// Run the full SSH session lifecycle.
+///
+/// Three phases: connect+auth, channel setup, event loop.
+/// All events are pushed to `queue` and the waker is fired.
+pub(super) async fn run_ssh_session(
+    config: SshConfig,
+    spec: ProcessSpec,
+    queue: EventQueue,
+    waker_slot: Arc<OnceLock<ProcessWaker>>,
+    mut command_rx: mpsc::UnboundedReceiver<SshCommand>,
+) {
+    let has_pty = spec.mode == ProcessMode::Interactive;
+
+    // --- Phase 1: Connect + Auth ---
+    let session = match connect_and_auth(&config).await {
+        Ok(session) => session,
+        Err(e) => {
+            push_and_wake(&queue, &waker_slot, ProcessEvent::SpawnFailed {
+                reason: format!("SSH connection failed: {e}"),
+            });
+            return;
+        }
+    };
+
+    // --- Phase 2: Channel setup ---
+    let channel = match setup_channel(&session, &spec, has_pty).await {
+        Ok(ch) => ch,
+        Err(e) => {
+            push_and_wake(&queue, &waker_slot, ProcessEvent::SpawnFailed {
+                reason: format!("SSH channel setup failed: {e}"),
+            });
+            return;
+        }
+    };
+
+    push_and_wake(&queue, &waker_slot, ProcessEvent::Started);
+
+    // --- Phase 3: Event loop ---
+    run_event_loop(channel, &mut command_rx, &queue, &waker_slot, has_pty).await;
+}
+
+async fn connect_and_auth(
+    config: &SshConfig,
+) -> Result<russh::client::Handle<SshHandler>, Box<dyn std::error::Error + Send + Sync>> {
+    let ssh_config = russh::client::Config {
+        keepalive_interval: Some(Duration::from_secs(30)),
+        keepalive_max: 3,
+        ..Default::default()
+    };
+
+    let mut session = russh::client::connect(
+        Arc::new(ssh_config),
+        (config.host.as_str(), config.port),
+        SshHandler,
+    )
+    .await?;
+
+    let key = russh_keys::load_secret_key(
+        &config.key_file,
+        config.key_passphrase.as_deref(),
+    )?;
+
+    let authenticated = session
+        .authenticate_publickey(&config.username, Arc::new(key))
+        .await?;
+
+    if !authenticated {
+        return Err("authentication rejected by server".into());
+    }
+
+    Ok(session)
+}
+
+async fn setup_channel(
+    session: &russh::client::Handle<SshHandler>,
+    spec: &ProcessSpec,
+    has_pty: bool,
+) -> Result<russh::Channel<russh::client::Msg>, Box<dyn std::error::Error + Send + Sync>> {
+    let channel = session.channel_open_session().await?;
+
+    if has_pty {
+        let (cols, rows) = spec
+            .initial_pty_size
+            .map(|s| (s.cols as u32, s.rows as u32))
+            .unwrap_or((80, 24));
+        channel
+            .request_pty(true, "xterm-256color", cols, rows, 0, 0, &[])
+            .await?;
+    }
+
+    // Best-effort env vars (many SSH servers restrict SetEnv)
+    for (key, val) in &spec.env {
+        let _ = channel.set_env(true, key, val).await;
+    }
+
+    if has_pty {
+        channel.request_shell(true).await?;
+    } else {
+        let cmd = build_remote_command(spec);
+        channel.exec(true, cmd).await?;
+    }
+
+    Ok(channel)
+}
+
+async fn run_event_loop(
+    mut channel: russh::Channel<russh::client::Msg>,
+    command_rx: &mut mpsc::UnboundedReceiver<SshCommand>,
+    queue: &EventQueue,
+    waker_slot: &Arc<OnceLock<ProcessWaker>>,
+    has_pty: bool,
+) {
+    let mut pending_exit: Option<ExitStatus> = None;
+    let mut kill_deadline: Option<Instant> = None;
+
+    loop {
+        // Build the kill-timeout future
+        let kill_sleep = async {
+            match kill_deadline {
+                Some(deadline) => sleep_until(deadline).await,
+                None => std::future::pending().await,
+            }
+        };
+
+        tokio::select! {
+            msg = channel.wait() => {
+                match msg {
+                    Some(ChannelMsg::Data { data }) => {
+                        push_and_wake(queue, waker_slot, ProcessEvent::OutputReceived {
+                            data: data.to_vec(),
+                            is_stderr: false,
+                        });
+                    }
+                    Some(ChannelMsg::ExtendedData { data, ext: 1 }) => {
+                        push_and_wake(queue, waker_slot, ProcessEvent::OutputReceived {
+                            data: data.to_vec(),
+                            is_stderr: true,
+                        });
+                    }
+                    Some(ChannelMsg::ExtendedData { .. }) => {
+                        // Ignore non-stderr extended data
+                    }
+                    Some(ChannelMsg::ExitStatus { exit_status }) => {
+                        pending_exit = Some(ExitStatus::Code(exit_status as i32));
+                    }
+                    Some(ChannelMsg::ExitSignal { signal_name, .. }) => {
+                        pending_exit = Some(ExitStatus::Signal(
+                            signal_name_to_code(&signal_name),
+                        ));
+                    }
+                    Some(ChannelMsg::Eof) | Some(ChannelMsg::Close) | None => {
+                        let status = pending_exit.take().unwrap_or(ExitStatus::Unknown);
+                        push_and_wake(queue, waker_slot, ProcessEvent::Exited { status });
+                        break;
+                    }
+                    _ => {}
+                }
+            }
+
+            cmd = command_rx.recv() => {
+                match cmd {
+                    Some(SshCommand::WriteStdin(data)) => {
+                        let len = data.len();
+                        match channel.data(&data[..]).await {
+                            Ok(()) => {
+                                push_and_wake(queue, waker_slot, ProcessEvent::StdinWritten {
+                                    byte_count: len,
+                                });
+                            }
+                            Err(e) => {
+                                push_and_wake(queue, waker_slot, ProcessEvent::ConnectionLost {
+                                    reason: format!("stdin write failed: {e}"),
+                                });
+                            }
+                        }
+                    }
+                    Some(SshCommand::SendSignal(Signal::Kill)) => {
+                        let _ = channel.close().await;
+                        push_and_wake(queue, waker_slot, ProcessEvent::SignalSent);
+                    }
+                    Some(SshCommand::SendSignal(signal)) => {
+                        let sig = signal_to_russh(signal);
+                        let _ = channel.signal(sig).await;
+                        push_and_wake(queue, waker_slot, ProcessEvent::SignalSent);
+                    }
+                    Some(SshCommand::ResizePty { cols, rows }) => {
+                        if has_pty {
+                            let _ = channel.window_change(
+                                cols as u32, rows as u32, 0, 0,
+                            ).await;
+                        }
+                        push_and_wake(queue, waker_slot, ProcessEvent::PtyResized);
+                    }
+                    Some(SshCommand::CloseStdin) => {
+                        let _ = channel.eof().await;
+                    }
+                    Some(SshCommand::ScheduleKillTimeout(dur)) => {
+                        kill_deadline = Some(Instant::now() + dur);
+                    }
+                    None => {
+                        // Sender dropped — close channel
+                        let _ = channel.close().await;
+                        break;
+                    }
+                }
+            }
+
+            _ = kill_sleep => {
+                push_and_wake(queue, waker_slot, ProcessEvent::KillTimeout);
+                kill_deadline = None;
+            }
+        }
+    }
+}
+
+/// Push an event and wake the actor.
+fn push_and_wake(
+    queue: &EventQueue,
+    waker_slot: &Arc<OnceLock<ProcessWaker>>,
+    event: ProcessEvent,
+) {
+    queue.push(event);
+    if let Some(w) = waker_slot.get() {
+        w.wake();
+    }
+}
+
+/// Build a remote exec command string from a ProcessSpec.
+///
+/// Produces: `cd '<dir>' && KEY='VAL' ... <cmd> <args>`
+fn build_remote_command(spec: &ProcessSpec) -> String {
+    let mut parts = Vec::new();
+
+    if let Some(ref dir) = spec.working_dir {
+        parts.push(format!("cd {}", shell_escape(dir)));
+    }
+
+    for (key, val) in &spec.env {
+        parts.push(format!("{}={}", key, shell_escape(val)));
+    }
+
+    let mut cmd = shell_escape(&spec.command);
+    for arg in &spec.args {
+        cmd.push(' ');
+        cmd.push_str(&shell_escape(arg));
+    }
+    parts.push(cmd);
+
+    if parts.len() > 1 && spec.working_dir.is_some() {
+        // Join with && so cd failure aborts
+        let cd_part = parts.remove(0);
+        format!("{} && {}", cd_part, parts.join(" "))
+    } else {
+        parts.join(" ")
+    }
+}
+
+/// POSIX single-quote escaping: wrap in single quotes, escape embedded quotes.
+fn shell_escape(s: &str) -> String {
+    if s.is_empty() {
+        return "''".to_string();
+    }
+    // If the string is simple (alphanumeric + safe chars), no quoting needed
+    if s.chars().all(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.' | '/' | ':' | ',' | '+' | '=')) {
+        return s.to_string();
+    }
+    // Single-quote the string, replacing ' with '\''
+    format!("'{}'", s.replace('\'', "'\\''"))
+}
+
+fn signal_to_russh(signal: Signal) -> Sig {
+    match signal {
+        Signal::Terminate => Sig::TERM,
+        Signal::Kill => Sig::KILL,
+        Signal::Hangup => Sig::HUP,
+        Signal::Interrupt => Sig::INT,
+        Signal::Other(_) => Sig::TERM, // Best effort fallback
+    }
+}
+
+fn signal_name_to_code(sig: &Sig) -> i32 {
+    match sig {
+        Sig::HUP => 1,
+        Sig::INT => 2,
+        Sig::QUIT => 3,
+        Sig::ABRT => 6,
+        Sig::KILL => 9,
+        Sig::ALRM => 14,
+        Sig::TERM => 15,
+        Sig::USR1 => 10,
+        _ => 15, // Default to SIGTERM code
+    }
+}

crates/process/src/subscriber.rs

@@ -0,0 +1,41 @@
+use swactor::actor::ActorAddress;
+
+/// A deduplicated collection of subscriber addresses.
+#[derive(Debug, Clone)]
+pub struct SubscriberSet {
+    inner: Vec<ActorAddress>,
+}
+
+impl SubscriberSet {
+    pub fn new() -> Self {
+        Self { inner: Vec::new() }
+    }
+
+    /// Add an address. No-op if already present.
+    pub fn add(&mut self, address: ActorAddress) {
+        if !self.inner.contains(&address) {
+            self.inner.push(address);
+        }
+    }
+
+    /// Remove an address. No-op if not present.
+    pub fn remove(&mut self, address: &ActorAddress) {
+        self.inner.retain(|a| a != address);
+    }
+
+    /// Snapshot of current subscribers.
+    pub fn snapshot(&self) -> Vec<ActorAddress> {
+        self.inner.clone()
+    }
+
+    /// Number of subscribers.
+    pub fn count(&self) -> usize {
+        self.inner.len()
+    }
+}
+
+impl Default for SubscriberSet {
+    fn default() -> Self {
+        Self::new()
+    }
+}

crates/process/src/types.rs

@@ -0,0 +1,71 @@
+use std::collections::HashMap;
+use std::time::Duration;
+
+/// Describes how to spawn a process.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ProcessSpec {
+    pub command: String,
+    pub args: Vec<String>,
+    pub env: HashMap<String, String>,
+    pub working_dir: Option<String>,
+    pub mode: ProcessMode,
+    pub initial_pty_size: Option<PtySize>,
+    /// If set, escalate to SIGKILL after this duration if the process hasn't exited
+    /// after SIGTERM. None = no escalation.
+    pub kill_timeout: Option<Duration>,
+    /// If set, buffer stdin writes when pending bytes exceed this limit.
+    /// None = unlimited (current behavior).
+    pub stdin_buffer_limit: Option<usize>,
+}
+
+/// Whether the process is interactive (PTY) or automated (pipes).
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ProcessMode {
+    Interactive,
+    Automated,
+}
+
+/// Dimensions of a pseudo-terminal.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct PtySize {
+    pub cols: u16,
+    pub rows: u16,
+}
+
+/// How a process exited.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ExitStatus {
+    Code(i32),
+    Signal(i32),
+    Unknown,
+}
+
+/// Signals that can be sent to a process.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Signal {
+    Terminate,
+    Kill,
+    Hangup,
+    Interrupt,
+    Other(i32),
+}
+
+/// Errors produced by the process session.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum ProcessError {
+    SpawnFailed { reason: String },
+    ConnectionLost { reason: String },
+    InvalidState { attempted: &'static str, current_state: &'static str },
+}
+
+/// Passive tracking of stdin backpressure.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct FlowControl {
+    pub pending_stdin_bytes: usize,
+}
+
+impl Default for FlowControl {
+    fn default() -> Self {
+        Self { pending_stdin_bytes: 0 }
+    }
+}

crates/process/src/waker.rs

@@ -0,0 +1,25 @@
+use std::sync::Arc;
+
+/// A handle that I/O threads use to wake the owning actor.
+///
+/// Constructed with a closure that sends a `ProcessCommand::PollTick`
+/// to the actor via `ExternalSender`. Thread-safe and cloneable.
+#[derive(Clone)]
+pub struct ProcessWaker(Arc<dyn Fn() + Send + Sync>);
+
+impl std::fmt::Debug for ProcessWaker {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("ProcessWaker").finish_non_exhaustive()
+    }
+}
+
+impl ProcessWaker {
+    pub fn new(f: impl Fn() + Send + Sync + 'static) -> Self {
+        Self(Arc::new(f))
+    }
+
+    /// Wake the owning actor so it drains pending events.
+    pub fn wake(&self) {
+        (self.0)();
+    }
+}

crates/process/tests/actor_scenarios.rs

@@ -0,0 +1,514 @@
+//! Layer 3 — Actor integration tests.
+//!
+//! Uses a TestDriver backed by a shared EventQueue so tests can inject
+//! events and observe actions without real OS processes.
+
+use std::collections::HashMap;
+use std::sync::{Arc, Mutex, OnceLock};
+
+use swactor::actor::{ActorAddress, ActorInterface, Ctx};
+use swactor::runtime::{Inbox, Runtime, RuntimeConfig};
+
+use swactor_process::*;
+
+// ── TestDriver ──────────────────────────────────────────────────────────────
+
+/// Shared harness for injecting events and inspecting driver actions.
+#[derive(Clone)]
+struct TestHarness {
+    queue: EventQueue,
+    actions: Arc<Mutex<Vec<ProcessAction>>>,
+}
+
+impl TestHarness {
+    fn new() -> Self {
+        Self {
+            queue: EventQueue::new(),
+            actions: Arc::new(Mutex::new(Vec::new())),
+        }
+    }
+
+    fn inject(&self, event: ProcessEvent) {
+        self.queue.push(event);
+    }
+
+    fn take_actions(&self) -> Vec<ProcessAction> {
+        std::mem::take(&mut self.actions.lock().unwrap())
+    }
+}
+
+/// A ProcessDriver that records actions and drains from a shared queue.
+struct TestDriver {
+    queue: EventQueue,
+    actions: Arc<Mutex<Vec<ProcessAction>>>,
+}
+
+impl TestDriver {
+    fn from_harness(harness: &TestHarness) -> Self {
+        Self {
+            queue: harness.queue.clone(),
+            actions: harness.actions.clone(),
+        }
+    }
+}
+
+impl ProcessDriver for TestDriver {
+    fn execute(&mut self, action: ProcessAction) {
+        self.actions.lock().unwrap().push(action);
+    }
+
+    fn poll(&mut self) -> Vec<ProcessEvent> {
+        self.queue.drain()
+    }
+}
+
+// ── Helpers ─────────────────────────────────────────────────────────────────
+
+fn automated_spec() -> ProcessSpec {
+    ProcessSpec {
+        command: "echo".into(),
+        args: vec!["hello".into()],
+        env: HashMap::new(),
+        working_dir: None,
+        mode: ProcessMode::Automated,
+        initial_pty_size: None,
+        kill_timeout: None,
+        stdin_buffer_limit: None,
+    }
+}
+
+fn setup() -> (Runtime, ExternalSender) {
+    let rt = Runtime::new(RuntimeConfig::default());
+    let sender = rt.create_sender();
+    (rt, sender)
+}
+
+/// Helper: tick until we receive N messages, returning them.
+fn tick_collect<M: swactor::actor::Message>(
+    rt: &Runtime,
+    inbox: &Inbox<M>,
+    n: usize,
+    max_ticks: usize,
+) -> Vec<M> {
+    let mut msgs = Vec::new();
+    for _ in 0..max_ticks {
+        rt.tick();
+        while let Some(m) = inbox.try_recv() {
+            msgs.push(m);
+            if msgs.len() >= n {
+                return msgs;
+            }
+        }
+    }
+    msgs
+}
+
+use swactor::runtime::ExternalSender;
+
+// ── Spawner actor ───────────────────────────────────────────────────────────
+// We can't call ctx.spawn from outside a handle(), so we use a small "spawner"
+// actor that spawns the process actor and reports its address.
+
+#[derive(Clone)]
+struct SpawnRequest {
+    spec: ProcessSpec,
+    harness: TestHarness,
+    reply_to: ActorAddress,
+    sender: ExternalSender,
+}
+
+#[derive(Clone, Debug)]
+struct SpawnedAddr(ActorAddress);
+
+struct SpawnerActor;
+
+impl ActorInterface for SpawnerActor {
+    type Incoming = SpawnRequest;
+    type Response = SpawnedAddr;
+
+    fn handle(&mut self, ctx: &Ctx, msg: SpawnRequest) {
+        let waker_slot = Arc::new(OnceLock::new());
+        let driver = TestDriver::from_harness(&msg.harness);
+        let addr = spawn_process(ctx, &msg.sender, msg.spec, driver, waker_slot)
+            .expect("spawn_process failed");
+        let _ = ctx.send(msg.reply_to, SpawnedAddr(addr));
+    }
+}
+
+// ── Tests ───────────────────────────────────────────────────────────────────
+
+#[test]
+fn happy_path_spawn_output_exit_notifies_subscriber() {
+    let (rt, sender) = setup();
+    let harness = TestHarness::new();
+    let notif_inbox = rt.new_inbox::<ProcessNotification>().unwrap();
+
+    // Spawn the spawner actor
+    let spawner_addr = rt.spawn(SpawnerActor).unwrap();
+    let reply_inbox = rt.new_inbox::<SpawnedAddr>().unwrap();
+    rt.tick();
+
+    // Ask spawner to create a process actor
+    rt.send_to(
+        spawner_addr,
+        SpawnRequest {
+            spec: automated_spec(),
+            harness: harness.clone(),
+            reply_to: *reply_inbox.addr(),
+            sender: sender.clone(),
+        },
+    )
+    .unwrap();
+
+    // Tick to process spawn request
+    for _ in 0..5 {
+        rt.tick();
+    }
+
+    let spawned = reply_inbox.try_recv().expect("should get spawned addr");
+    let proc_addr = spawned.0;
+
+    // Verify SpawnProcess action was sent to driver
+    let actions = harness.take_actions();
+    assert!(
+        actions.iter().any(|a| matches!(a, ProcessAction::SpawnProcess { .. })),
+        "driver should receive SpawnProcess, got: {:?}",
+        actions
+    );
+
+    // Subscribe to notifications
+    rt.send_to(proc_addr, ProcessCommand::Subscribe { address: *notif_inbox.addr() })
+        .unwrap();
+    rt.tick();
+
+    // Inject Started event from "driver"
+    harness.inject(ProcessEvent::Started);
+    // Send PollTick to trigger drain
+    rt.send_to(proc_addr, ProcessCommand::PollTick).unwrap();
+    for _ in 0..3 {
+        rt.tick();
+    }
+
+    let msgs = tick_collect(&rt, &notif_inbox, 1, 10);
+    assert!(
+        msgs.iter().any(|m| matches!(m, ProcessNotification::Started { .. })),
+        "subscriber should get Started notification, got: {:?}",
+        msgs
+    );
+
+    // Inject output
+    harness.inject(ProcessEvent::OutputReceived {
+        data: b"hello\n".to_vec(),
+        is_stderr: false,
+    });
+    rt.send_to(proc_addr, ProcessCommand::PollTick).unwrap();
+    let msgs = tick_collect(&rt, &notif_inbox, 1, 10);
+    assert!(
+        msgs.iter().any(|m| matches!(m, ProcessNotification::Output { .. })),
+        "subscriber should get Output notification"
+    );
+
+    // Inject exit
+    harness.inject(ProcessEvent::Exited {
+        status: ExitStatus::Code(0),
+    });
+    rt.send_to(proc_addr, ProcessCommand::PollTick).unwrap();
+    let msgs = tick_collect(&rt, &notif_inbox, 1, 10);
+    assert!(
+        msgs.iter().any(|m| matches!(
+            m,
+            ProcessNotification::Exited { status: ExitStatus::Code(0), .. }
+        )),
+        "subscriber should get Exited(0) notification"
+    );
+}
+
+#[test]
+fn polltick_drains_queued_events() {
+    let (rt, sender) = setup();
+    let harness = TestHarness::new();
+    let notif_inbox = rt.new_inbox::<ProcessNotification>().unwrap();
+
+    let spawner_addr = rt.spawn(SpawnerActor).unwrap();
+    let reply_inbox = rt.new_inbox::<SpawnedAddr>().unwrap();
+    rt.tick();
+
+    rt.send_to(
+        spawner_addr,
+        SpawnRequest {
+            spec: automated_spec(),
+            harness: harness.clone(),
+            reply_to: *reply_inbox.addr(),
+            sender: sender.clone(),
+        },
+    )
+    .unwrap();
+    for _ in 0..5 {
+        rt.tick();
+    }
+
+    let proc_addr = reply_inbox.try_recv().unwrap().0;
+
+    // Subscribe
+    rt.send_to(proc_addr, ProcessCommand::Subscribe { address: *notif_inbox.addr() })
+        .unwrap();
+    rt.tick();
+
+    // Queue multiple events before sending PollTick
+    harness.inject(ProcessEvent::Started);
+    harness.inject(ProcessEvent::OutputReceived {
+        data: b"line1\n".to_vec(),
+        is_stderr: false,
+    });
+    harness.inject(ProcessEvent::OutputReceived {
+        data: b"line2\n".to_vec(),
+        is_stderr: false,
+    });
+
+    // Single PollTick should drain all
+    rt.send_to(proc_addr, ProcessCommand::PollTick).unwrap();
+    let msgs = tick_collect(&rt, &notif_inbox, 3, 20);
+
+    assert_eq!(msgs.len(), 3, "all three events should produce notifications");
+    assert!(matches!(msgs[0], ProcessNotification::Started { .. }));
+    assert!(matches!(msgs[1], ProcessNotification::Output { .. }));
+    assert!(matches!(msgs[2], ProcessNotification::Output { .. }));
+}
+
+#[test]
+fn close_command_triggers_graceful_shutdown() {
+    let (rt, sender) = setup();
+    let harness = TestHarness::new();
+    let notif_inbox = rt.new_inbox::<ProcessNotification>().unwrap();
+
+    let spawner_addr = rt.spawn(SpawnerActor).unwrap();
+    let reply_inbox = rt.new_inbox::<SpawnedAddr>().unwrap();
+    rt.tick();
+
+    rt.send_to(
+        spawner_addr,
+        SpawnRequest {
+            spec: automated_spec(),
+            harness: harness.clone(),
+            reply_to: *reply_inbox.addr(),
+            sender: sender.clone(),
+        },
+    )
+    .unwrap();
+    for _ in 0..5 {
+        rt.tick();
+    }
+    let proc_addr = reply_inbox.try_recv().unwrap().0;
+
+    // Subscribe and get to Running state
+    rt.send_to(proc_addr, ProcessCommand::Subscribe { address: *notif_inbox.addr() })
+        .unwrap();
+    rt.tick();
+    harness.inject(ProcessEvent::Started);
+    rt.send_to(proc_addr, ProcessCommand::PollTick).unwrap();
+    let _ = tick_collect::<ProcessNotification>(&rt, &notif_inbox, 1, 10);
+
+    // Send Close
+    harness.take_actions(); // clear previous actions
+    rt.send_to(proc_addr, ProcessCommand::Close).unwrap();
+    for _ in 0..5 {
+        rt.tick();
+    }
+
+    let actions = harness.take_actions();
+    assert!(
+        actions.iter().any(|a| matches!(
+            a,
+            ProcessAction::SendSignal { signal: Signal::Terminate }
+        )),
+        "Close should trigger SIGTERM, got: {:?}",
+        actions
+    );
+}
+
+#[test]
+fn write_stdin_and_signal_forwarded_to_driver() {
+    let (rt, sender) = setup();
+    let harness = TestHarness::new();
+
+    let spawner_addr = rt.spawn(SpawnerActor).unwrap();
+    let reply_inbox = rt.new_inbox::<SpawnedAddr>().unwrap();
+    rt.tick();
+
+    rt.send_to(
+        spawner_addr,
+        SpawnRequest {
+            spec: automated_spec(),
+            harness: harness.clone(),
+            reply_to: *reply_inbox.addr(),
+            sender: sender.clone(),
+        },
+    )
+    .unwrap();
+    for _ in 0..5 {
+        rt.tick();
+    }
+    let proc_addr = reply_inbox.try_recv().unwrap().0;
+
+    // Get to Running
+    harness.inject(ProcessEvent::Started);
+    rt.send_to(proc_addr, ProcessCommand::PollTick).unwrap();
+    for _ in 0..5 {
+        rt.tick();
+    }
+
+    harness.take_actions(); // clear SpawnProcess action
+
+    // Write stdin
+    rt.send_to(
+        proc_addr,
+        ProcessCommand::WriteStdin {
+            data: b"input\n".to_vec(),
+        },
+    )
+    .unwrap();
+    for _ in 0..3 {
+        rt.tick();
+    }
+
+    let actions = harness.take_actions();
+    assert!(
+        actions.iter().any(|a| matches!(a, ProcessAction::WriteStdin { .. })),
+        "WriteStdin should be forwarded to driver, got: {:?}",
+        actions
+    );
+
+    // Send signal
+    rt.send_to(
+        proc_addr,
+        ProcessCommand::SendSignal {
+            signal: Signal::Interrupt,
+        },
+    )
+    .unwrap();
+    for _ in 0..3 {
+        rt.tick();
+    }
+
+    let actions = harness.take_actions();
+    assert!(
+        actions.iter().any(|a| matches!(
+            a,
+            ProcessAction::SendSignal { signal: Signal::Interrupt }
+        )),
+        "SendSignal should be forwarded to driver, got: {:?}",
+        actions
+    );
+}
+
+#[test]
+fn spawn_failure_notifies_error_and_stops_actor() {
+    let (rt, sender) = setup();
+    let harness = TestHarness::new();
+    let notif_inbox = rt.new_inbox::<ProcessNotification>().unwrap();
+
+    let spawner_addr = rt.spawn(SpawnerActor).unwrap();
+    let reply_inbox = rt.new_inbox::<SpawnedAddr>().unwrap();
+    rt.tick();
+
+    rt.send_to(
+        spawner_addr,
+        SpawnRequest {
+            spec: automated_spec(),
+            harness: harness.clone(),
+            reply_to: *reply_inbox.addr(),
+            sender: sender.clone(),
+        },
+    )
+    .unwrap();
+    for _ in 0..5 {
+        rt.tick();
+    }
+    let proc_addr = reply_inbox.try_recv().unwrap().0;
+
+    // Subscribe
+    rt.send_to(proc_addr, ProcessCommand::Subscribe { address: *notif_inbox.addr() })
+        .unwrap();
+    rt.tick();
+
+    // Inject spawn failure
+    harness.inject(ProcessEvent::SpawnFailed {
+        reason: "command not found".into(),
+    });
+    rt.send_to(proc_addr, ProcessCommand::PollTick).unwrap();
+
+    let msgs = tick_collect(&rt, &notif_inbox, 1, 20);
+    assert!(
+        msgs.iter().any(|m| matches!(m, ProcessNotification::Error { .. })),
+        "subscriber should get Error notification on spawn failure"
+    );
+
+    // Actor should have stopped — sending further messages should fail or be ignored
+    // (the address may still be in the map briefly, but the actor won't process)
+    for _ in 0..10 {
+        rt.tick();
+    }
+}
+
+#[test]
+fn subscribe_and_unsubscribe_routing() {
+    let (rt, sender) = setup();
+    let harness = TestHarness::new();
+    let inbox_a = rt.new_inbox::<ProcessNotification>().unwrap();
+    let inbox_b = rt.new_inbox::<ProcessNotification>().unwrap();
+
+    let spawner_addr = rt.spawn(SpawnerActor).unwrap();
+    let reply_inbox = rt.new_inbox::<SpawnedAddr>().unwrap();
+    rt.tick();
+
+    rt.send_to(
+        spawner_addr,
+        SpawnRequest {
+            spec: automated_spec(),
+            harness: harness.clone(),
+            reply_to: *reply_inbox.addr(),
+            sender: sender.clone(),
+        },
+    )
+    .unwrap();
+    for _ in 0..5 {
+        rt.tick();
+    }
+    let proc_addr = reply_inbox.try_recv().unwrap().0;
+
+    // Subscribe both
+    rt.send_to(proc_addr, ProcessCommand::Subscribe { address: *inbox_a.addr() })
+        .unwrap();
+    rt.send_to(proc_addr, ProcessCommand::Subscribe { address: *inbox_b.addr() })
+        .unwrap();
+    rt.tick();
+
+    // Get to Running
+    harness.inject(ProcessEvent::Started);
+    rt.send_to(proc_addr, ProcessCommand::PollTick).unwrap();
+    for _ in 0..5 {
+        rt.tick();
+    }
+
+    // Both should have received Started
+    assert!(inbox_a.try_recv().is_some(), "inbox_a should get Started");
+    assert!(inbox_b.try_recv().is_some(), "inbox_b should get Started");
+
+    // Unsubscribe inbox_b
+    rt.send_to(proc_addr, ProcessCommand::Unsubscribe { address: *inbox_b.addr() })
+        .unwrap();
+    rt.tick();
+
+    // Inject output — only inbox_a should receive it
+    harness.inject(ProcessEvent::OutputReceived {
+        data: b"data".to_vec(),
+        is_stderr: false,
+    });
+    rt.send_to(proc_addr, ProcessCommand::PollTick).unwrap();
+    for _ in 0..5 {
+        rt.tick();
+    }
+
+    assert!(inbox_a.try_recv().is_some(), "inbox_a should get Output");
+    assert!(inbox_b.try_recv().is_none(), "inbox_b should NOT get Output after unsubscribe");
+}

crates/process/tests/e2e_process.rs

@@ -0,0 +1,189 @@
+//! End-to-end tests: full Runtime + ExternalSender + ProcessActor<LocalDriver>.
+//!
+//! Spawns real OS processes through the actor system and verifies the
+//! complete notification flow.
+
+use std::collections::HashMap;
+
+use swactor::actor::{ActorAddress, ActorInterface, Ctx};
+use swactor::runtime::{ExternalSender, Inbox, Runtime, RuntimeConfig};
+
+use swactor_process::*;
+
+fn automated_spec(cmd: &str, args: &[&str]) -> ProcessSpec {
+    ProcessSpec {
+        command: cmd.into(),
+        args: args.iter().map(|s| s.to_string()).collect(),
+        env: HashMap::new(),
+        working_dir: None,
+        mode: ProcessMode::Automated,
+        initial_pty_size: None,
+        kill_timeout: None,
+        stdin_buffer_limit: None,
+    }
+}
+
+/// Tick and collect up to `n` notifications, with a max tick budget.
+fn tick_collect(
+    rt: &Runtime,
+    inbox: &Inbox<ProcessNotification>,
+    n: usize,
+    max_ticks: usize,
+) -> Vec<ProcessNotification> {
+    let mut msgs = Vec::new();
+    for _ in 0..max_ticks {
+        rt.tick();
+        // Small sleep to let I/O threads produce events
+        std::thread::sleep(std::time::Duration::from_millis(5));
+        while let Some(m) = inbox.try_recv() {
+            msgs.push(m);
+            if msgs.len() >= n {
+                return msgs;
+            }
+        }
+    }
+    msgs
+}
+
+// ── Spawner actor (needed because spawn_local_process requires &Ctx) ────────
+
+#[derive(Clone)]
+struct E2eSpawnRequest {
+    spec: ProcessSpec,
+    subscriber: ActorAddress,
+    reply_to: ActorAddress,
+    sender: ExternalSender,
+}
+
+#[derive(Clone, Debug)]
+struct E2eSpawned(ActorAddress);
+
+struct E2eSpawnerActor;
+
+impl ActorInterface for E2eSpawnerActor {
+    type Incoming = E2eSpawnRequest;
+    type Response = E2eSpawned;
+
+    fn handle(&mut self, ctx: &Ctx, msg: E2eSpawnRequest) {
+        let addr = spawn_local_process(ctx, &msg.sender, msg.spec)
+            .expect("spawn_local_process failed");
+        // Subscribe the notification inbox
+        let _ = ctx.send(addr, ProcessCommand::Subscribe { address: msg.subscriber });
+        let _ = ctx.send(msg.reply_to, E2eSpawned(addr));
+    }
+}
+
+// ── Tests ───────────────────────────────────────────────────────────────────
+
+#[test]
+fn echo_hello_full_lifecycle() {
+    let rt = Runtime::new(RuntimeConfig::default());
+    let sender = rt.create_sender();
+    let notif_inbox = rt.new_inbox::<ProcessNotification>().unwrap();
+
+    let spawner_addr = rt.spawn(E2eSpawnerActor).unwrap();
+    let reply_inbox = rt.new_inbox::<E2eSpawned>().unwrap();
+    rt.tick();
+
+    rt.send_to(
+        spawner_addr,
+        E2eSpawnRequest {
+            spec: automated_spec("echo", &["hello"]),
+            subscriber: *notif_inbox.addr(),
+            reply_to: *reply_inbox.addr(),
+            sender: sender.clone(),
+        },
+    )
+    .unwrap();
+
+    // Tick enough for the spawner to process + the process actor to start
+    for _ in 0..10 {
+        rt.tick();
+        std::thread::sleep(std::time::Duration::from_millis(5));
+    }
+
+    let spawned = reply_inbox.try_recv().expect("should get spawned address");
+    let _proc_addr = spawned.0;
+
+    // Collect notifications: Started, Output("hello\n"), Exited(0)
+    let msgs = tick_collect(&rt, &notif_inbox, 3, 200);
+
+    let has_started = msgs.iter().any(|m| matches!(m, ProcessNotification::Started { .. }));
+    let has_output = msgs.iter().any(|m| {
+        if let ProcessNotification::Output { data, .. } = m {
+            String::from_utf8_lossy(data).contains("hello")
+        } else {
+            false
+        }
+    });
+    let has_exited = msgs.iter().any(|m| {
+        matches!(
+            m,
+            ProcessNotification::Exited {
+                status: ExitStatus::Code(0),
+                ..
+            }
+        )
+    });
+
+    assert!(has_started, "should receive Started notification, got: {:?}", msgs);
+    assert!(has_output, "should receive Output with 'hello', got: {:?}", msgs);
+    assert!(has_exited, "should receive Exited(0) notification, got: {:?}", msgs);
+
+    // Verify ordering: Started before Output before Exited
+    let started_idx = msgs
+        .iter()
+        .position(|m| matches!(m, ProcessNotification::Started { .. }))
+        .unwrap();
+    let output_idx = msgs
+        .iter()
+        .position(|m| matches!(m, ProcessNotification::Output { .. }))
+        .unwrap();
+    let exited_idx = msgs
+        .iter()
+        .position(|m| matches!(m, ProcessNotification::Exited { .. }))
+        .unwrap();
+
+    assert!(
+        started_idx < output_idx,
+        "Started should come before Output"
+    );
+    assert!(
+        output_idx < exited_idx,
+        "Output should come before Exited"
+    );
+}
+
+#[test]
+fn bad_command_reports_error_e2e() {
+    let rt = Runtime::new(RuntimeConfig::default());
+    let sender = rt.create_sender();
+    let notif_inbox = rt.new_inbox::<ProcessNotification>().unwrap();
+
+    let spawner_addr = rt.spawn(E2eSpawnerActor).unwrap();
+    let reply_inbox = rt.new_inbox::<E2eSpawned>().unwrap();
+    rt.tick();
+
+    rt.send_to(
+        spawner_addr,
+        E2eSpawnRequest {
+            spec: automated_spec("/nonexistent/binary/xyz", &[]),
+            subscriber: *notif_inbox.addr(),
+            reply_to: *reply_inbox.addr(),
+            sender: sender.clone(),
+        },
+    )
+    .unwrap();
+
+    for _ in 0..10 {
+        rt.tick();
+        std::thread::sleep(std::time::Duration::from_millis(5));
+    }
+
+    let msgs = tick_collect(&rt, &notif_inbox, 1, 200);
+    assert!(
+        msgs.iter().any(|m| matches!(m, ProcessNotification::Error { .. })),
+        "should receive Error notification for bad command, got: {:?}",
+        msgs
+    );
+}

crates/process/tests/local_driver.rs

@@ -0,0 +1,324 @@
+//! Layer 4 — LocalDriver integration tests.
+//!
+//! Real OS processes, no actor layer. Tests LocalDriver in isolation.
+
+use std::collections::HashMap;
+use std::sync::{Arc, OnceLock};
+use std::thread;
+use std::time::Duration;
+
+use swactor_process::*;
+
+fn automated_spec(cmd: &str, args: &[&str]) -> ProcessSpec {
+    ProcessSpec {
+        command: cmd.into(),
+        args: args.iter().map(|s| s.to_string()).collect(),
+        env: HashMap::new(),
+        working_dir: None,
+        mode: ProcessMode::Automated,
+        initial_pty_size: None,
+        kill_timeout: None,
+        stdin_buffer_limit: None,
+    }
+}
+
+/// Poll the driver until `pred` matches at least one collected event, or timeout.
+fn poll_until_match(
+    driver: &mut LocalDriver,
+    timeout: Duration,
+    pred: impl Fn(&ProcessEvent) -> bool,
+) -> Vec<ProcessEvent> {
+    let start = std::time::Instant::now();
+    let mut all_events = Vec::new();
+    loop {
+        let events = driver.poll();
+        if events.is_empty() {
+            if start.elapsed() >= timeout {
+                break;
+            }
+            thread::sleep(Duration::from_millis(10));
+        }
+        all_events.extend(events);
+        if all_events.iter().any(&pred) {
+            break;
+        }
+    }
+    all_events
+}
+
+fn has_event(events: &[ProcessEvent], pred: impl Fn(&ProcessEvent) -> bool) -> bool {
+    events.iter().any(pred)
+}
+
+#[test]
+fn echo_produces_started_output_and_exit_zero() {
+    let queue = EventQueue::new();
+    let waker_slot = Arc::new(OnceLock::new());
+    let mut driver = LocalDriver::new(queue, waker_slot);
+
+    let spec = automated_spec("echo", &["hello"]);
+    driver.execute(ProcessAction::SpawnProcess { spec });
+
+    // Wait for Exited (which means Started + output + exit are all in)
+    let events = poll_until_match(&mut driver, Duration::from_secs(5), |e| {
+        matches!(e, ProcessEvent::Exited { .. })
+    });
+
+    assert!(
+        has_event(&events, |e| matches!(e, ProcessEvent::Started)),
+        "should have Started event, got: {:?}",
+        events
+    );
+    assert!(
+        has_event(&events, |e| matches!(e, ProcessEvent::OutputReceived { is_stderr: false, .. })),
+        "should have stdout OutputReceived"
+    );
+
+    // Check the output contains "hello"
+    let output: Vec<u8> = events
+        .iter()
+        .filter_map(|e| match e {
+            ProcessEvent::OutputReceived {
+                data, is_stderr: false,
+            } => Some(data.clone()),
+            _ => None,
+        })
+        .flatten()
+        .collect();
+    let output_str = String::from_utf8_lossy(&output);
+    assert!(
+        output_str.contains("hello"),
+        "output should contain 'hello', got: {:?}",
+        output_str
+    );
+
+    assert!(
+        has_event(&events, |e| matches!(
+            e,
+            ProcessEvent::Exited { status: ExitStatus::Code(0) }
+        )),
+        "should have Exited(0)"
+    );
+}
+
+#[test]
+fn cat_stdin_echo_and_close() {
+    let queue = EventQueue::new();
+    let waker_slot = Arc::new(OnceLock::new());
+    let mut driver = LocalDriver::new(queue, waker_slot);
+
+    let spec = automated_spec("cat", &[]);
+    driver.execute(ProcessAction::SpawnProcess { spec });
+
+    // Wait for Started
+    let events = poll_until_match(&mut driver, Duration::from_secs(5), |e| {
+        matches!(e, ProcessEvent::Started)
+    });
+    assert!(has_event(&events, |e| matches!(e, ProcessEvent::Started)));
+
+    // Write to stdin
+    driver.execute(ProcessAction::WriteStdin {
+        data: b"ping\n".to_vec(),
+    });
+
+    // Wait until we see actual output (not just the StdinWritten ack)
+    let events = poll_until_match(&mut driver, Duration::from_secs(5), |e| {
+        matches!(e, ProcessEvent::OutputReceived { .. })
+    });
+    let output: Vec<u8> = events
+        .iter()
+        .filter_map(|e| match e {
+            ProcessEvent::OutputReceived { data, .. } => Some(data.clone()),
+            _ => None,
+        })
+        .flatten()
+        .collect();
+    let output_str = String::from_utf8_lossy(&output);
+    assert!(
+        output_str.contains("ping"),
+        "cat should echo back 'ping', got: {:?}",
+        output_str
+    );
+
+    // Close stdin — cat should exit
+    driver.execute(ProcessAction::CloseStdin);
+    let events = poll_until_match(&mut driver, Duration::from_secs(5), |e| {
+        matches!(e, ProcessEvent::Exited { .. })
+    });
+    assert!(
+        has_event(&events, |e| matches!(
+            e,
+            ProcessEvent::Exited { status: ExitStatus::Code(0) }
+        )),
+        "cat should exit cleanly after stdin close, got: {:?}",
+        events
+    );
+}
+
+#[test]
+fn signal_terminates_long_running_process() {
+    let queue = EventQueue::new();
+    let waker_slot = Arc::new(OnceLock::new());
+    let mut driver = LocalDriver::new(queue, waker_slot);
+
+    let spec = automated_spec("sleep", &["60"]);
+    driver.execute(ProcessAction::SpawnProcess { spec });
+
+    // Wait for Started
+    let events = poll_until_match(&mut driver, Duration::from_secs(5), |e| {
+        matches!(e, ProcessEvent::Started)
+    });
+    assert!(has_event(&events, |e| matches!(e, ProcessEvent::Started)));
+
+    // Send SIGTERM
+    driver.execute(ProcessAction::SendSignal {
+        signal: Signal::Terminate,
+    });
+
+    // Wait for Exited (may also see SignalSent ack first)
+    let events = poll_until_match(&mut driver, Duration::from_secs(5), |e| {
+        matches!(e, ProcessEvent::Exited { .. })
+    });
+    assert!(
+        has_event(&events, |e| matches!(
+            e,
+            ProcessEvent::Exited { status: ExitStatus::Signal(_) }
+        )),
+        "sleep should exit with signal status after SIGTERM, got: {:?}",
+        events
+    );
+}
+
+#[test]
+fn bad_command_produces_spawn_failed() {
+    let queue = EventQueue::new();
+    let waker_slot = Arc::new(OnceLock::new());
+    let mut driver = LocalDriver::new(queue, waker_slot);
+
+    let spec = automated_spec("/nonexistent/binary/that/does/not/exist", &[]);
+    driver.execute(ProcessAction::SpawnProcess { spec });
+
+    let events = poll_until_match(&mut driver, Duration::from_secs(5), |e| {
+        matches!(e, ProcessEvent::SpawnFailed { .. })
+    });
+    assert!(
+        has_event(&events, |e| matches!(e, ProcessEvent::SpawnFailed { .. })),
+        "nonexistent binary should produce SpawnFailed, got: {:?}",
+        events
+    );
+}
+
+#[test]
+fn large_output_no_data_loss() {
+    let queue = EventQueue::new();
+    let waker_slot = Arc::new(OnceLock::new());
+    let mut driver = LocalDriver::new(queue, waker_slot);
+
+    // Generate a large amount of output: seq 1 10000
+    let spec = automated_spec("seq", &["1", "10000"]);
+    driver.execute(ProcessAction::SpawnProcess { spec });
+
+    // Collect all events until exit
+    let events = poll_until_match(&mut driver, Duration::from_secs(10), |e| {
+        matches!(e, ProcessEvent::Exited { .. })
+    });
+
+    // Gather all output
+    let output: Vec<u8> = events
+        .iter()
+        .filter_map(|e| match e {
+            ProcessEvent::OutputReceived { data, .. } => Some(data.clone()),
+            _ => None,
+        })
+        .flatten()
+        .collect();
+
+    let output_str = String::from_utf8_lossy(&output);
+    // seq 1 10000 should end with "10000\n"
+    assert!(
+        output_str.contains("10000"),
+        "large output should contain '10000'"
+    );
+    // Check that it starts with "1\n"
+    assert!(
+        output_str.starts_with("1\n"),
+        "large output should start with '1\\n'"
+    );
+
+    assert!(
+        has_event(&events, |e| matches!(
+            e,
+            ProcessEvent::Exited { status: ExitStatus::Code(0) }
+        )),
+        "seq should exit cleanly"
+    );
+}
+
+#[test]
+fn kill_timeout_escalates_to_sigkill() {
+    let queue = EventQueue::new();
+    let waker_slot = Arc::new(OnceLock::new());
+    let mut driver = LocalDriver::new(queue, waker_slot);
+
+    // Spawn a process that traps SIGTERM. Use exec to replace the shell so
+    // SIGTERM goes directly to the perl process (avoids shell vs child races).
+    let spec = automated_spec(
+        "perl",
+        &["-e", "$SIG{TERM} = 'IGNORE'; sleep 300"],
+    );
+    driver.execute(ProcessAction::SpawnProcess { spec });
+
+    // Wait for Started
+    let events = poll_until_match(&mut driver, Duration::from_secs(5), |e| {
+        matches!(e, ProcessEvent::Started)
+    });
+    assert!(has_event(&events, |e| matches!(e, ProcessEvent::Started)));
+
+    // Give the process a moment to set up the trap
+    thread::sleep(Duration::from_millis(100));
+
+    // Send SIGTERM (the process ignores it)
+    driver.execute(ProcessAction::SendSignal { signal: Signal::Terminate });
+    poll_until_match(&mut driver, Duration::from_secs(1), |e| {
+        matches!(e, ProcessEvent::SignalSent)
+    });
+
+    // Verify the process is still alive after a short wait (SIGTERM was ignored)
+    thread::sleep(Duration::from_millis(200));
+    let events = driver.poll();
+    assert!(
+        !has_event(&events, |e| matches!(e, ProcessEvent::Exited { .. })),
+        "process should still be alive after SIGTERM (trap should ignore it)"
+    );
+
+    // Schedule a short kill timeout
+    driver.execute(ProcessAction::ScheduleKillTimeout {
+        duration: Duration::from_millis(200),
+    });
+
+    // Wait for KillTimeout event
+    let events = poll_until_match(&mut driver, Duration::from_secs(3), |e| {
+        matches!(e, ProcessEvent::KillTimeout)
+    });
+    assert!(
+        has_event(&events, |e| matches!(e, ProcessEvent::KillTimeout)),
+        "should receive KillTimeout, got: {:?}",
+        events
+    );
+
+    // Now send SIGKILL
+    driver.execute(ProcessAction::SendSignal { signal: Signal::Kill });
+
+    // Wait for exit
+    let events = poll_until_match(&mut driver, Duration::from_secs(5), |e| {
+        matches!(e, ProcessEvent::Exited { .. })
+    });
+    assert!(
+        has_event(&events, |e| matches!(
+            e,
+            ProcessEvent::Exited { status: ExitStatus::Signal(_) }
+        )),
+        "process should exit with signal after SIGKILL, got: {:?}",
+        events
+    );
+}

crates/process/tests/proptest_session.rs

@@ -0,0 +1,195 @@
+use std::collections::HashMap;
+
+use proptest::prelude::*;
+use swactor::actor::ActorAddress;
+use swactor_process::*;
+
+fn automated_spec() -> ProcessSpec {
+    ProcessSpec {
+        command: "test".into(),
+        args: vec![],
+        env: HashMap::new(),
+        working_dir: None,
+        mode: ProcessMode::Automated,
+        initial_pty_size: None,
+        kill_timeout: None,
+        stdin_buffer_limit: None,
+    }
+}
+
+fn addr(n: u8) -> ActorAddress {
+    let mut bytes = [0u8; 32];
+    bytes[0] = n;
+    ActorAddress(bytes)
+}
+
+fn arb_signal() -> impl Strategy<Value = Signal> {
+    prop_oneof![
+        Just(Signal::Terminate),
+        Just(Signal::Kill),
+        Just(Signal::Hangup),
+        Just(Signal::Interrupt),
+        (0..32i32).prop_map(Signal::Other),
+    ]
+}
+
+fn arb_event() -> impl Strategy<Value = ProcessEvent> {
+    prop_oneof![
+        Just(ProcessEvent::Started),
+        Just(ProcessEvent::KillTimeout),
+        ".*".prop_map(|reason| ProcessEvent::SpawnFailed { reason }),
+        proptest::collection::vec(any::<u8>(), 0..64)
+            .prop_map(|data| ProcessEvent::OutputReceived { data, is_stderr: false }),
+        proptest::collection::vec(any::<u8>(), 0..64)
+            .prop_map(|data| ProcessEvent::OutputReceived { data, is_stderr: true }),
+        prop_oneof![
+            any::<i32>().prop_map(ExitStatus::Code),
+            any::<i32>().prop_map(ExitStatus::Signal),
+            Just(ExitStatus::Unknown),
+        ]
+        .prop_map(|status| ProcessEvent::Exited { status }),
+        ".*".prop_map(|reason| ProcessEvent::ConnectionLost { reason }),
+        (0..5usize).prop_map(|n| ProcessEvent::StdinWritten { byte_count: n * 10 }),
+        Just(ProcessEvent::SignalSent),
+        Just(ProcessEvent::PtyResized),
+        proptest::collection::vec(any::<u8>(), 0..64)
+            .prop_map(|data| ProcessEvent::WriteStdin { data }),
+        arb_signal().prop_map(|signal| ProcessEvent::SendSignal { signal }),
+        Just(ProcessEvent::ResizePty {
+            size: PtySize { cols: 80, rows: 24 },
+        }),
+        Just(ProcessEvent::CloseStdin),
+        Just(ProcessEvent::CloseRequested),
+        (0..4u8).prop_map(|n| ProcessEvent::Subscribe { address: addr(n) }),
+        (0..4u8).prop_map(|n| ProcessEvent::Unsubscribe { address: addr(n) }),
+    ]
+}
+
+// ──────────────────────────────────────────────
+// 1. No panics for arbitrary event sequences
+// ──────────────────────────────────────────────
+
+proptest! {
+    #[test]
+    fn no_panics_on_arbitrary_events(events in proptest::collection::vec(arb_event(), 0..50)) {
+        let (mut session, _) = ProcessSession::new(automated_spec());
+        for event in events {
+            let _ = session.apply(event);
+        }
+    }
+}
+
+// ──────────────────────────────────────────────
+// 2. Exited is terminal
+// ──────────────────────────────────────────────
+
+proptest! {
+    #[test]
+    fn exited_is_terminal(events in proptest::collection::vec(arb_event(), 0..50)) {
+        let (mut session, _) = ProcessSession::new(automated_spec());
+        let mut reached_exited = false;
+
+        for event in events {
+            let _ = session.apply(event);
+            if session.state() == ProcessState::Exited {
+                reached_exited = true;
+            }
+            if reached_exited {
+                prop_assert_eq!(session.state(), ProcessState::Exited);
+            }
+        }
+    }
+}
+
+// ──────────────────────────────────────────────
+// 3. SelfTerminate always last action when entering Exited
+// ──────────────────────────────────────────────
+
+proptest! {
+    #[test]
+    fn self_terminate_is_last_when_entering_exited(events in proptest::collection::vec(arb_event(), 0..50)) {
+        let (mut session, _) = ProcessSession::new(automated_spec());
+        let mut was_exited = false;
+
+        for event in events {
+            let prev_state = session.state();
+            let actions = session.apply(event);
+
+            // If we just transitioned into Exited
+            if session.state() == ProcessState::Exited && !was_exited && prev_state != ProcessState::Exited {
+                prop_assert!(
+                    matches!(actions.last(), Some(ProcessAction::SelfTerminate)),
+                    "SelfTerminate must be last action when entering Exited, got: {:?}", actions
+                );
+            }
+
+            if session.state() == ProcessState::Exited {
+                was_exited = true;
+            }
+        }
+    }
+}
+
+// ──────────────────────────────────────────────
+// 4. Subscriber count matches add/remove operations
+// ──────────────────────────────────────────────
+
+proptest! {
+    #[test]
+    fn subscriber_count_is_consistent(
+        ops in proptest::collection::vec(
+            prop_oneof![
+                (0..8u8).prop_map(|n| (true, n)),
+                (0..8u8).prop_map(|n| (false, n)),
+            ],
+            0..30
+        )
+    ) {
+        let (mut session, _) = ProcessSession::new(automated_spec());
+        let mut expected: Vec<u8> = Vec::new();
+
+        for (is_add, n) in ops {
+            if is_add {
+                session.apply(ProcessEvent::Subscribe { address: addr(n) });
+                if !expected.contains(&n) {
+                    expected.push(n);
+                }
+            } else {
+                session.apply(ProcessEvent::Unsubscribe { address: addr(n) });
+                expected.retain(|&x| x != n);
+            }
+            prop_assert_eq!(session.subscriber_count(), expected.len());
+        }
+    }
+}
+
+// ──────────────────────────────────────────────
+// 5. State monotonicity (never goes backward)
+// ──────────────────────────────────────────────
+
+fn state_ordinal(s: ProcessState) -> u8 {
+    match s {
+        ProcessState::Starting => 0,
+        ProcessState::Running => 1,
+        ProcessState::Stopping => 2,
+        ProcessState::Exited => 3,
+    }
+}
+
+proptest! {
+    #[test]
+    fn state_never_goes_backward(events in proptest::collection::vec(arb_event(), 0..50)) {
+        let (mut session, _) = ProcessSession::new(automated_spec());
+        let mut max_ordinal = state_ordinal(session.state());
+
+        for event in events {
+            let _ = session.apply(event);
+            let current = state_ordinal(session.state());
+            prop_assert!(
+                current >= max_ordinal,
+                "State went backward: ordinal {} -> {}", max_ordinal, current
+            );
+            max_ordinal = current;
+        }
+    }
+}

crates/process/tests/session_scenarios.rs

@@ -0,0 +1,698 @@
+use std::collections::HashMap;
+use std::time::Duration;
+
+use swactor::actor::ActorAddress;
+use swactor_process::*;
+
+fn automated_spec() -> ProcessSpec {
+    ProcessSpec {
+        command: "echo".into(),
+        args: vec!["hello".into()],
+        env: HashMap::new(),
+        working_dir: None,
+        mode: ProcessMode::Automated,
+        initial_pty_size: None,
+        kill_timeout: None,
+        stdin_buffer_limit: None,
+    }
+}
+
+fn spec_with_kill_timeout(timeout: Duration) -> ProcessSpec {
+    ProcessSpec {
+        kill_timeout: Some(timeout),
+        ..automated_spec()
+    }
+}
+
+fn spec_with_stdin_limit(limit: usize) -> ProcessSpec {
+    ProcessSpec {
+        stdin_buffer_limit: Some(limit),
+        ..automated_spec()
+    }
+}
+
+fn interactive_spec() -> ProcessSpec {
+    ProcessSpec {
+        command: "/bin/bash".into(),
+        args: vec![],
+        env: HashMap::new(),
+        working_dir: None,
+        mode: ProcessMode::Interactive,
+        initial_pty_size: Some(PtySize { cols: 80, rows: 24 }),
+        kill_timeout: None,
+        stdin_buffer_limit: None,
+    }
+}
+
+fn addr(n: u8) -> ActorAddress {
+    let mut bytes = [0u8; 32];
+    bytes[0] = n;
+    ActorAddress(bytes)
+}
+
+/// Verify that a SelfTerminate is present and is the last action.
+fn assert_self_terminate_is_last(actions: &[ProcessAction]) {
+    assert!(
+        matches!(actions.last(), Some(ProcessAction::SelfTerminate)),
+        "SelfTerminate must be the last action, got: {actions:?}"
+    );
+}
+
+// ──────────────────────────────────────────────
+// 1. Happy path — automated process
+// ──────────────────────────────────────────────
+
+#[test]
+fn automated_process_runs_produces_output_and_exits_cleanly() {
+    let (mut session, init) = ProcessSession::new(automated_spec());
+    assert_eq!(session.state(), ProcessState::Starting);
+    assert!(matches!(&init[0], ProcessAction::SpawnProcess { .. }));
+
+    // Process starts
+    let actions = session.apply(ProcessEvent::Started);
+    assert_eq!(session.state(), ProcessState::Running);
+    assert!(matches!(&actions[0], ProcessAction::NotifyStarted { .. }));
+
+    // Some output arrives
+    let actions = session.apply(ProcessEvent::OutputReceived {
+        data: b"hello\n".to_vec(),
+        is_stderr: false,
+    });
+    assert!(matches!(
+        &actions[0],
+        ProcessAction::NotifyOutput { stream: OutputStream::Stdout, .. }
+    ));
+
+    // More output on stderr
+    let actions = session.apply(ProcessEvent::OutputReceived {
+        data: b"warn\n".to_vec(),
+        is_stderr: true,
+    });
+    assert!(matches!(
+        &actions[0],
+        ProcessAction::NotifyOutput { stream: OutputStream::Stderr, .. }
+    ));
+
+    // Process exits
+    let actions = session.apply(ProcessEvent::Exited {
+        status: ExitStatus::Code(0),
+    });
+    assert_eq!(session.state(), ProcessState::Exited);
+    assert_eq!(session.exit_status(), Some(ExitStatus::Code(0)));
+    assert_self_terminate_is_last(&actions);
+}
+
+// ──────────────────────────────────────────────
+// 2. Interactive process with subscriber lifecycle
+// ──────────────────────────────────────────────
+
+#[test]
+fn interactive_session_manages_subscribers_correctly() {
+    let (mut session, _) = ProcessSession::new(interactive_spec());
+
+    // Add two subscribers before start
+    session.apply(ProcessEvent::Subscribe { address: addr(1) });
+    session.apply(ProcessEvent::Subscribe { address: addr(2) });
+    assert_eq!(session.subscriber_count(), 2);
+
+    // Duplicate add is a no-op
+    session.apply(ProcessEvent::Subscribe { address: addr(1) });
+    assert_eq!(session.subscriber_count(), 2);
+
+    // Start — both subscribers notified
+    let actions = session.apply(ProcessEvent::Started);
+    match &actions[0] {
+        ProcessAction::NotifyStarted { subscribers } => {
+            assert_eq!(subscribers.len(), 2);
+        }
+        other => panic!("expected NotifyStarted, got {other:?}"),
+    }
+
+    // Remove one subscriber
+    session.apply(ProcessEvent::Unsubscribe { address: addr(1) });
+    assert_eq!(session.subscriber_count(), 1);
+
+    // Output only goes to remaining subscriber
+    let actions = session.apply(ProcessEvent::OutputReceived {
+        data: b"data".to_vec(),
+        is_stderr: false,
+    });
+    match &actions[0] {
+        ProcessAction::NotifyOutput { subscribers, .. } => {
+            assert_eq!(subscribers, &vec![addr(2)]);
+        }
+        other => panic!("expected NotifyOutput, got {other:?}"),
+    }
+
+    // Exit
+    let actions = session.apply(ProcessEvent::Exited {
+        status: ExitStatus::Code(0),
+    });
+    match &actions[0] {
+        ProcessAction::NotifyExited { subscribers, .. } => {
+            assert_eq!(subscribers, &vec![addr(2)]);
+        }
+        other => panic!("expected NotifyExited, got {other:?}"),
+    }
+    assert_self_terminate_is_last(&actions);
+}
+
+// ──────────────────────────────────────────────
+// 3. Spawn failure
+// ──────────────────────────────────────────────
+
+#[test]
+fn spawn_failure_notifies_and_self_terminates() {
+    let (mut session, _) = ProcessSession::new(automated_spec());
+    session.apply(ProcessEvent::Subscribe { address: addr(1) });
+
+    let actions = session.apply(ProcessEvent::SpawnFailed {
+        reason: "command not found".into(),
+    });
+    assert_eq!(session.state(), ProcessState::Exited);
+    assert!(matches!(
+        &actions[0],
+        ProcessAction::NotifyError {
+            error: ProcessError::SpawnFailed { .. },
+            ..
+        }
+    ));
+    assert_self_terminate_is_last(&actions);
+}
+
+// ──────────────────────────────────────────────
+// 4. Connection loss mid-run
+// ──────────────────────────────────────────────
+
+#[test]
+fn connection_loss_during_running_transitions_to_exited() {
+    let (mut session, _) = ProcessSession::new(automated_spec());
+    session.apply(ProcessEvent::Started);
+
+    let actions = session.apply(ProcessEvent::ConnectionLost {
+        reason: "pipe broken".into(),
+    });
+    assert_eq!(session.state(), ProcessState::Exited);
+    assert_eq!(session.exit_status(), Some(ExitStatus::Unknown));
+    assert!(matches!(
+        &actions[0],
+        ProcessAction::NotifyError {
+            error: ProcessError::ConnectionLost { .. },
+            ..
+        }
+    ));
+    assert_self_terminate_is_last(&actions);
+}
+
+// ──────────────────────────────────────────────
+// 5. Close requested before start
+// ──────────────────────────────────────────────
+
+#[test]
+fn close_before_start_sends_signal_on_belated_start() {
+    let (mut session, _) = ProcessSession::new(automated_spec());
+
+    // Close requested while still Starting
+    let actions = session.apply(ProcessEvent::CloseRequested);
+    assert!(actions.is_empty());
+    assert_eq!(session.state(), ProcessState::Starting);
+
+    // Process starts belatedly — should immediately get SIGTERM
+    let actions = session.apply(ProcessEvent::Started);
+    assert_eq!(session.state(), ProcessState::Stopping);
+    assert!(matches!(&actions[0], ProcessAction::NotifyStarted { .. }));
+    assert!(matches!(
+        &actions[1],
+        ProcessAction::SendSignal { signal: Signal::Terminate }
+    ));
+}
+
+// ──────────────────────────────────────────────
+// 6. Invalid operations produce errors, not panics
+// ──────────────────────────────────────────────
+
+#[test]
+fn invalid_event_in_starting_produces_error() {
+    let (mut session, _) = ProcessSession::new(automated_spec());
+
+    let actions = session.apply(ProcessEvent::WriteStdin {
+        data: b"hi".to_vec(),
+    });
+    assert!(matches!(
+        &actions[0],
+        ProcessAction::NotifyError {
+            error: ProcessError::InvalidState { attempted: "WriteStdin", current_state: "Starting" },
+            ..
+        }
+    ));
+    // State unchanged
+    assert_eq!(session.state(), ProcessState::Starting);
+}
+
+#[test]
+fn invalid_event_in_exited_produces_error() {
+    let (mut session, _) = ProcessSession::new(automated_spec());
+    session.apply(ProcessEvent::SpawnFailed {
+        reason: "no".into(),
+    });
+    assert_eq!(session.state(), ProcessState::Exited);
+
+    let actions = session.apply(ProcessEvent::WriteStdin {
+        data: b"hi".to_vec(),
+    });
+    assert!(matches!(
+        &actions[0],
+        ProcessAction::NotifyError {
+            error: ProcessError::InvalidState { attempted: "WriteStdin", current_state: "Exited" },
+            ..
+        }
+    ));
+}
+
+// ──────────────────────────────────────────────
+// 7. Stdin closed then write → error
+// ──────────────────────────────────────────────
+
+#[test]
+fn write_after_stdin_closed_produces_error() {
+    let (mut session, _) = ProcessSession::new(automated_spec());
+    session.apply(ProcessEvent::Started);
+
+    let actions = session.apply(ProcessEvent::CloseStdin);
+    assert!(matches!(&actions[0], ProcessAction::CloseStdin));
+    assert!(session.stdin_closed());
+
+    // Duplicate close is a no-op
+    let actions = session.apply(ProcessEvent::CloseStdin);
+    assert!(actions.is_empty());
+
+    // Write after close → error
+    let actions = session.apply(ProcessEvent::WriteStdin {
+        data: b"too late".to_vec(),
+    });
+    assert!(matches!(
+        &actions[0],
+        ProcessAction::NotifyError {
+            error: ProcessError::InvalidState { .. },
+            ..
+        }
+    ));
+}
+
+// ──────────────────────────────────────────────
+// 8. MockDriver round-trip (driver + session tick loop)
+// ──────────────────────────────────────────────
+
+#[test]
+fn mock_driver_round_trip() {
+    let (mut session, init_actions) = ProcessSession::new(automated_spec());
+    let mut driver = MockDriver::new();
+
+    // Execute initial actions (SpawnProcess)
+    for action in init_actions {
+        driver.execute(action);
+    }
+    assert!(matches!(
+        &driver.executed_actions()[0],
+        ProcessAction::SpawnProcess { .. }
+    ));
+
+    // Simulate: driver produces Started
+    driver.inject(ProcessEvent::Started);
+
+    // Tick loop: poll → apply → execute
+    let events = driver.poll();
+    for event in events {
+        let actions = session.apply(event);
+        for action in actions {
+            driver.execute(action);
+        }
+    }
+    assert_eq!(session.state(), ProcessState::Running);
+
+    // Simulate output and exit
+    driver.inject(ProcessEvent::OutputReceived {
+        data: b"done".to_vec(),
+        is_stderr: false,
+    });
+    driver.inject(ProcessEvent::Exited {
+        status: ExitStatus::Code(0),
+    });
+
+    let events = driver.poll();
+    for event in events {
+        let actions = session.apply(event);
+        for action in actions {
+            driver.execute(action);
+        }
+    }
+
+    assert_eq!(session.state(), ProcessState::Exited);
+
+    // Verify the driver saw the expected sequence
+    let all_actions = driver.take_executed_actions();
+    assert!(matches!(&all_actions[0], ProcessAction::SpawnProcess { .. }));
+    assert!(matches!(&all_actions[1], ProcessAction::NotifyStarted { .. }));
+    assert!(matches!(&all_actions[2], ProcessAction::NotifyOutput { .. }));
+    assert!(matches!(&all_actions[3], ProcessAction::NotifyExited { .. }));
+    assert!(matches!(&all_actions[4], ProcessAction::SelfTerminate));
+}
+
+// ──────────────────────────────────────────────
+// 9. Signal escalation in Stopping
+// ──────────────────────────────────────────────
+
+#[test]
+fn signal_escalation_allowed_in_stopping() {
+    let (mut session, _) = ProcessSession::new(automated_spec());
+    session.apply(ProcessEvent::Started);
+    session.apply(ProcessEvent::CloseRequested);
+    assert_eq!(session.state(), ProcessState::Stopping);
+
+    // Escalate to Kill
+    let actions = session.apply(ProcessEvent::SendSignal {
+        signal: Signal::Kill,
+    });
+    assert!(matches!(
+        &actions[0],
+        ProcessAction::SendSignal { signal: Signal::Kill }
+    ));
+
+    // Can still receive output while stopping
+    let actions = session.apply(ProcessEvent::OutputReceived {
+        data: b"final".to_vec(),
+        is_stderr: false,
+    });
+    assert!(matches!(&actions[0], ProcessAction::NotifyOutput { .. }));
+
+    // Finally exits
+    let actions = session.apply(ProcessEvent::Exited {
+        status: ExitStatus::Signal(9),
+    });
+    assert_eq!(session.exit_status(), Some(ExitStatus::Signal(9)));
+    assert_self_terminate_is_last(&actions);
+}
+
+// ──────────────────────────────────────────────
+// 10. Late acks in Exited silently consumed
+// ──────────────────────────────────────────────
+
+#[test]
+fn late_acks_in_exited_are_silently_consumed() {
+    let (mut session, _) = ProcessSession::new(automated_spec());
+    session.apply(ProcessEvent::Started);
+    session.apply(ProcessEvent::Exited {
+        status: ExitStatus::Code(0),
+    });
+    assert_eq!(session.state(), ProcessState::Exited);
+
+    // Acks should produce no actions, no errors
+    assert!(session.apply(ProcessEvent::StdinWritten { byte_count: 10 }).is_empty());
+    assert!(session.apply(ProcessEvent::SignalSent).is_empty());
+    assert!(session.apply(ProcessEvent::PtyResized).is_empty());
+
+    // Subscribe/Unsubscribe also still works in Exited
+    assert!(session.apply(ProcessEvent::Subscribe { address: addr(1) }).is_empty());
+    assert_eq!(session.subscriber_count(), 1);
+    assert!(session.apply(ProcessEvent::Unsubscribe { address: addr(1) }).is_empty());
+    assert_eq!(session.subscriber_count(), 0);
+}
+
+// ──────────────────────────────────────────────
+// Flow control tracking
+// ──────────────────────────────────────────────
+
+#[test]
+fn flow_control_tracks_pending_stdin_bytes() {
+    let (mut session, _) = ProcessSession::new(automated_spec());
+    session.apply(ProcessEvent::Started);
+
+    session.apply(ProcessEvent::WriteStdin {
+        data: vec![0u8; 100],
+    });
+    assert_eq!(session.flow_control().pending_stdin_bytes, 100);
+
+    session.apply(ProcessEvent::WriteStdin {
+        data: vec![0u8; 50],
+    });
+    assert_eq!(session.flow_control().pending_stdin_bytes, 150);
+
+    session.apply(ProcessEvent::StdinWritten { byte_count: 80 });
+    assert_eq!(session.flow_control().pending_stdin_bytes, 70);
+
+    // Ack more than pending → saturates at 0
+    session.apply(ProcessEvent::StdinWritten { byte_count: 200 });
+    assert_eq!(session.flow_control().pending_stdin_bytes, 0);
+}
+
+// ──────────────────────────────────────────────
+// CloseStdin in Stopping
+// ──────────────────────────────────────────────
+
+#[test]
+fn close_stdin_allowed_in_stopping() {
+    let (mut session, _) = ProcessSession::new(automated_spec());
+    session.apply(ProcessEvent::Started);
+    session.apply(ProcessEvent::CloseRequested);
+    assert_eq!(session.state(), ProcessState::Stopping);
+
+    let actions = session.apply(ProcessEvent::CloseStdin);
+    assert!(matches!(&actions[0], ProcessAction::CloseStdin));
+    assert!(session.stdin_closed());
+}
+
+// ──────────────────────────────────────────────
+// Connection loss in Stopping
+// ──────────────────────────────────────────────
+
+#[test]
+fn connection_loss_in_stopping_transitions_to_exited() {
+    let (mut session, _) = ProcessSession::new(automated_spec());
+    session.apply(ProcessEvent::Started);
+    session.apply(ProcessEvent::CloseRequested);
+    assert_eq!(session.state(), ProcessState::Stopping);
+
+    let actions = session.apply(ProcessEvent::ConnectionLost {
+        reason: "gone".into(),
+    });
+    assert_eq!(session.state(), ProcessState::Exited);
+    assert_self_terminate_is_last(&actions);
+}
+
+// ──────────────────────────────────────────────
+// Redundant CloseRequested in Stopping is no-op
+// ──────────────────────────────────────────────
+
+#[test]
+fn duplicate_close_requested_in_stopping_is_noop() {
+    let (mut session, _) = ProcessSession::new(automated_spec());
+    session.apply(ProcessEvent::Started);
+    session.apply(ProcessEvent::CloseRequested);
+    assert_eq!(session.state(), ProcessState::Stopping);
+
+    let actions = session.apply(ProcessEvent::CloseRequested);
+    assert!(actions.is_empty());
+    assert_eq!(session.state(), ProcessState::Stopping);
+}
+
+// ──────────────────────────────────────────────
+// Kill timeout — A1–A6
+// ──────────────────────────────────────────────
+
+#[test]
+fn close_requested_with_kill_timeout_schedules_timer() {
+    let (mut session, _) = ProcessSession::new(spec_with_kill_timeout(Duration::from_secs(5)));
+    session.apply(ProcessEvent::Started);
+
+    let actions = session.apply(ProcessEvent::CloseRequested);
+    assert_eq!(session.state(), ProcessState::Stopping);
+    assert!(matches!(
+        &actions[0],
+        ProcessAction::SendSignal { signal: Signal::Terminate }
+    ));
+    assert!(matches!(
+        &actions[1],
+        ProcessAction::ScheduleKillTimeout { duration } if *duration == Duration::from_secs(5)
+    ));
+}
+
+#[test]
+fn close_before_start_with_kill_timeout_schedules_timer_on_belated_start() {
+    let (mut session, _) = ProcessSession::new(spec_with_kill_timeout(Duration::from_secs(3)));
+    session.apply(ProcessEvent::CloseRequested);
+
+    let actions = session.apply(ProcessEvent::Started);
+    assert_eq!(session.state(), ProcessState::Stopping);
+    assert!(matches!(&actions[0], ProcessAction::NotifyStarted { .. }));
+    assert!(matches!(
+        &actions[1],
+        ProcessAction::SendSignal { signal: Signal::Terminate }
+    ));
+    assert!(matches!(
+        &actions[2],
+        ProcessAction::ScheduleKillTimeout { duration } if *duration == Duration::from_secs(3)
+    ));
+}
+
+#[test]
+fn kill_timeout_in_stopping_sends_sigkill() {
+    let (mut session, _) = ProcessSession::new(spec_with_kill_timeout(Duration::from_secs(5)));
+    session.apply(ProcessEvent::Started);
+    session.apply(ProcessEvent::CloseRequested);
+    assert_eq!(session.state(), ProcessState::Stopping);
+
+    let actions = session.apply(ProcessEvent::KillTimeout);
+    assert!(matches!(
+        &actions[0],
+        ProcessAction::SendSignal { signal: Signal::Kill }
+    ));
+    assert_eq!(session.state(), ProcessState::Stopping);
+}
+
+#[test]
+fn kill_timeout_silently_consumed_outside_stopping() {
+    // Starting
+    let (mut session, _) = ProcessSession::new(automated_spec());
+    assert!(session.apply(ProcessEvent::KillTimeout).is_empty());
+    assert_eq!(session.state(), ProcessState::Starting);
+
+    // Running
+    session.apply(ProcessEvent::Started);
+    assert!(session.apply(ProcessEvent::KillTimeout).is_empty());
+    assert_eq!(session.state(), ProcessState::Running);
+
+    // Exited
+    session.apply(ProcessEvent::Exited { status: ExitStatus::Code(0) });
+    assert!(session.apply(ProcessEvent::KillTimeout).is_empty());
+    assert_eq!(session.state(), ProcessState::Exited);
+}
+
+#[test]
+fn close_requested_without_kill_timeout_no_schedule_action() {
+    let (mut session, _) = ProcessSession::new(automated_spec());
+    session.apply(ProcessEvent::Started);
+
+    let actions = session.apply(ProcessEvent::CloseRequested);
+    assert_eq!(actions.len(), 1);
+    assert!(matches!(
+        &actions[0],
+        ProcessAction::SendSignal { signal: Signal::Terminate }
+    ));
+}
+
+#[test]
+fn kill_timeout_full_escalation_to_sigkill_then_exit() {
+    let (mut session, _) = ProcessSession::new(spec_with_kill_timeout(Duration::from_secs(1)));
+    session.apply(ProcessEvent::Started);
+
+    // CloseRequested → SIGTERM + schedule
+    let actions = session.apply(ProcessEvent::CloseRequested);
+    assert_eq!(session.state(), ProcessState::Stopping);
+    assert!(matches!(&actions[0], ProcessAction::SendSignal { signal: Signal::Terminate }));
+    assert!(matches!(&actions[1], ProcessAction::ScheduleKillTimeout { .. }));
+
+    // KillTimeout fires → SIGKILL
+    let actions = session.apply(ProcessEvent::KillTimeout);
+    assert!(matches!(&actions[0], ProcessAction::SendSignal { signal: Signal::Kill }));
+
+    // Process finally exits via signal 9
+    let actions = session.apply(ProcessEvent::Exited { status: ExitStatus::Signal(9) });
+    assert_eq!(session.state(), ProcessState::Exited);
+    assert_eq!(session.exit_status(), Some(ExitStatus::Signal(9)));
+    assert_self_terminate_is_last(&actions);
+}
+
+// ──────────────────────────────────────────────
+// Backpressure — B1–B5
+// ──────────────────────────────────────────────
+
+#[test]
+fn backpressure_buffers_when_over_limit() {
+    let (mut session, _) = ProcessSession::new(spec_with_stdin_limit(100));
+    session.apply(ProcessEvent::Started);
+
+    // First write (50 bytes) — under limit, passes through
+    let actions = session.apply(ProcessEvent::WriteStdin { data: vec![1u8; 50] });
+    assert_eq!(actions.len(), 1);
+    assert!(matches!(&actions[0], ProcessAction::WriteStdin { .. }));
+    assert_eq!(session.flow_control().pending_stdin_bytes, 50);
+
+    // Second write (60 bytes) — still under limit (50 < 100), passes through
+    let actions = session.apply(ProcessEvent::WriteStdin { data: vec![2u8; 60] });
+    assert_eq!(actions.len(), 1);
+    assert_eq!(session.flow_control().pending_stdin_bytes, 110);
+
+    // Third write (30 bytes) — now at 110 >= 100, buffered
+    let actions = session.apply(ProcessEvent::WriteStdin { data: vec![3u8; 30] });
+    assert!(actions.is_empty());
+    assert_eq!(session.stdin_buffer_bytes(), 30);
+    // pending_stdin_bytes unchanged (buffered data not counted as pending)
+    assert_eq!(session.flow_control().pending_stdin_bytes, 110);
+}
+
+#[test]
+fn stdin_written_ack_drains_buffer() {
+    let (mut session, _) = ProcessSession::new(spec_with_stdin_limit(100));
+    session.apply(ProcessEvent::Started);
+
+    // Fill up: 100 bytes pending
+    session.apply(ProcessEvent::WriteStdin { data: vec![1u8; 100] });
+    assert_eq!(session.flow_control().pending_stdin_bytes, 100);
+
+    // Buffer two chunks
+    session.apply(ProcessEvent::WriteStdin { data: vec![2u8; 40] });
+    session.apply(ProcessEvent::WriteStdin { data: vec![3u8; 30] });
+    assert_eq!(session.stdin_buffer_bytes(), 70);
+
+    // Ack 80 bytes → pending drops to 20, buffer should drain in FIFO order
+    let actions = session.apply(ProcessEvent::StdinWritten { byte_count: 80 });
+    // pending was 100, now 20. Drain first chunk (40 bytes) → pending = 60.
+    // 60 < 100, drain second chunk (30 bytes) → pending = 90.
+    // 90 < 100, buffer empty.
+    assert_eq!(actions.len(), 2);
+    assert!(matches!(&actions[0], ProcessAction::WriteStdin { data } if data.len() == 40));
+    assert!(matches!(&actions[1], ProcessAction::WriteStdin { data } if data.len() == 30));
+    assert_eq!(session.flow_control().pending_stdin_bytes, 90);
+    assert_eq!(session.stdin_buffer_bytes(), 0);
+}
+
+#[test]
+fn close_requested_clears_stdin_buffer() {
+    let (mut session, _) = ProcessSession::new(spec_with_stdin_limit(50));
+    session.apply(ProcessEvent::Started);
+
+    session.apply(ProcessEvent::WriteStdin { data: vec![1u8; 60] });
+    session.apply(ProcessEvent::WriteStdin { data: vec![2u8; 30] });
+    assert_eq!(session.stdin_buffer_bytes(), 30);
+
+    session.apply(ProcessEvent::CloseRequested);
+    assert_eq!(session.stdin_buffer_bytes(), 0);
+}
+
+#[test]
+fn no_backpressure_when_limit_is_none() {
+    let (mut session, _) = ProcessSession::new(automated_spec());
+    session.apply(ProcessEvent::Started);
+
+    // All writes pass through regardless of pending bytes
+    for _ in 0..10 {
+        let actions = session.apply(ProcessEvent::WriteStdin { data: vec![0u8; 1000] });
+        assert_eq!(actions.len(), 1);
+        assert!(matches!(&actions[0], ProcessAction::WriteStdin { .. }));
+    }
+    assert_eq!(session.flow_control().pending_stdin_bytes, 10_000);
+    assert_eq!(session.stdin_buffer_bytes(), 0);
+}
+
+#[test]
+fn exit_clears_stdin_buffer() {
+    let (mut session, _) = ProcessSession::new(spec_with_stdin_limit(50));
+    session.apply(ProcessEvent::Started);
+
+    session.apply(ProcessEvent::WriteStdin { data: vec![1u8; 60] });
+    session.apply(ProcessEvent::WriteStdin { data: vec![2u8; 30] });
+    assert_eq!(session.stdin_buffer_bytes(), 30);
+
+    session.apply(ProcessEvent::Exited { status: ExitStatus::Code(0) });
+    assert_eq!(session.stdin_buffer_bytes(), 0);
+}

docs/development_history/PROCESS_RUNNER.md

@@ -0,0 +1,434 @@
+# Process Runner Design: Async Process Management in Swactor
+
+## Context
+
+Swactor is a synchronous, tick-based actor framework (Erlang-inspired). Actors must return quickly from `handle()` — blocking stalls the entire worker thread. There is no built-in async I/O.
+
+The goal: let actors manage long-lived async "processes" — OS subprocesses and SSH shells — with full lifecycle control. Must support both interactive use (live shell, bidirectional real-time I/O) and automated execution (run commands, stream output, report exit).
+
+Constraints from discussion:
+- Backends: SSH + local processes (two backends, not more)
+- Scale: Architecture should support thousands; first implementation handles tens
+- This is a standalone new feature — not related to or derived from the CI runner system
+
+---
+
+## Architecture: State Machine + Driver + Process-as-Actor
+
+### Data Flow (full picture)
+
+```
+OS process stdout/stderr
+       │ (background thread reads pipe)
+       ▼
+  EventQueue (Arc<SegQueue>) — shared lock-free buffer
+       │ (background thread calls ProcessWaker → ExternalSender → PollTick)
+       ▼
+  Actor handle(PollTick)
+       │ calls driver.poll() which drains EventQueue
+       ▼
+  Vec<ProcessEvent>
+       │
+       ▼
+  session.apply(event) → Vec<ProcessAction>
+       │
+       ├─ Driver commands → driver.execute(action) → OS I/O
+       ├─ Notifications → ctx.send(subscriber, ProcessNotification)
+       └─ SelfTerminate → ctx.stop_self()
+```
+
+### The Layers
+
+| Layer | Purpose | Status |
+|-------|---------|--------|
+| 1 — ProcessSession | Pure-logic state machine | **Implemented** |
+| 2 — ProcessDriver trait + MockDriver | Driver abstraction + test double | **Implemented** |
+| 3 — Process Actor + ExternalSender | Swactor integration, waker, event queue | **Implemented** |
+| 4 — LocalDriver | `std::process::Command` + pipe I/O + signal | **Implemented** |
+| 5 — SshDriver | SSH library + channel I/O | Not started |
+
+---
+
+## Implemented: Layers 1 + 2 (Pure Logic)
+
+Crate: `crates/process/` (`swactor-process`)
+
+### Layer 1 — ProcessSession (State Machine)
+
+The core state machine. Pure logic, no I/O, fully deterministic.
+
+**States:** `Starting` → `Running` → `Stopping` → `Exited`
+
+State transitions are monotonic — the state never goes backward. `Exited` is terminal.
+
+**Construction:**
+
+```rust
+let (session, initial_actions) = ProcessSession::new(spec);
+// initial_actions == [SpawnProcess { spec }]
+// session.state() == Starting
+```
+
+**Event loop:**
+
+```rust
+let actions = session.apply(event);
+for action in actions {
+    match action {
+        ProcessAction::SpawnProcess { .. } |
+        ProcessAction::WriteStdin { .. } |
+        ProcessAction::SendSignal { .. } |
+        ProcessAction::ResizePty { .. } |
+        ProcessAction::CloseStdin |
+        ProcessAction::ScheduleKillTimeout { .. } => driver.execute(action),
+
+        ProcessAction::NotifyStarted { subscribers } |
+        ProcessAction::NotifyOutput { subscribers, .. } |
+        ProcessAction::NotifyExited { subscribers, .. } |
+        ProcessAction::NotifyError { subscribers, .. } => { /* send to subscribers */ }
+
+        ProcessAction::SelfTerminate => { /* actor stops itself */ }
+    }
+}
+```
+
+**Key invariants (all verified by property-based tests):**
+- Invalid events produce `NotifyError` actions — never panic
+- `SelfTerminate` is always the last action when entering `Exited`
+- State monotonicity: Starting ≤ Running ≤ Stopping ≤ Exited
+- Subscriber count always matches add/remove operations
+- No panics for arbitrary event sequences
+
+**Event handling by state:**
+
+| Event | Starting | Running | Stopping | Exited |
+|-------|----------|---------|----------|--------|
+| Started | → Running (+ NotifyStarted) | error | error | error |
+| SpawnFailed | → Exited (+ NotifyError + SelfTerminate) | error | error | error |
+| OutputReceived | error | NotifyOutput | NotifyOutput | error |
+| Exited | error | → Exited (+ NotifyExited + SelfTerminate) | → Exited (+ NotifyExited + SelfTerminate) | error |
+| ConnectionLost | error | → Exited (+ NotifyError + SelfTerminate) | → Exited (+ NotifyError + SelfTerminate) | error |
+| WriteStdin | error | WriteStdin (or buffer/error) | error | error |
+| SendSignal | error | SendSignal | SendSignal (escalation) | error |
+| ResizePty | error | ResizePty | error | error |
+| CloseStdin | error | CloseStdin (+ clear buffer) | CloseStdin (+ set flag) | error |
+| CloseRequested | set deferred flag | → Stopping (+ SendSignal Terminate [+ ScheduleKillTimeout]) | no-op | error |
+| KillTimeout | silent | silent | SendSignal Kill | silent |
+| Subscribe | add subscriber | add subscriber | add subscriber | add subscriber |
+| Unsubscribe | remove subscriber | remove subscriber | remove subscriber | remove subscriber |
+| StdinWritten | update flow | update flow + drain buffer | update flow | update flow |
+| SignalSent | silent | silent | silent | silent |
+| PtyResized | silent | silent | silent | silent |
+
+**Special behaviors:**
+- **Close-before-start:** If `CloseRequested` arrives in `Starting`, a flag is set. When `Started` arrives, the session transitions through Running straight to Stopping and emits `SendSignal(Terminate)` (plus `ScheduleKillTimeout` if configured).
+- **Kill timeout:** When `spec.kill_timeout` is `Some(duration)`, entering `Stopping` emits `ScheduleKillTimeout { duration }` alongside `SendSignal(Terminate)`. If the process hasn't exited when the timeout fires, the `KillTimeout` event triggers `SendSignal(Kill)`. `KillTimeout` in non-Stopping states is silently consumed (harmless late arrival after the process already exited).
+- **Backpressure:** When `spec.stdin_buffer_limit` is `Some(limit)` and `pending_stdin_bytes >= limit`, `WriteStdin` events are buffered in a `VecDeque` instead of emitting actions. When `StdinWritten` acks reduce `pending_stdin_bytes` below the limit, buffered writes drain in FIFO order. The buffer is cleared on `CloseRequested`, `CloseStdin`, `ConnectionLost`, and `Exited`. When `stdin_buffer_limit` is `None`, all writes pass through immediately (original behavior).
+- **FlowControl:** `pending_stdin_bytes` is incremented on `WriteStdin` emission, decremented on `StdinWritten` receipt (saturating).
+- **Stdin closed:** Once `CloseStdin` is applied, further `WriteStdin` events produce `InvalidState` errors. Duplicate `CloseStdin` is a no-op. Closing stdin also clears any buffered writes.
+- **Late acks in Exited:** `StdinWritten`, `SignalSent`, `PtyResized`, and `KillTimeout` are silently consumed in all states (including Exited) — they never produce errors.
+
+### Types
+
+**ProcessSpec** — describes how to spawn a process:
+- `command: String`, `args: Vec<String>`, `env: HashMap<String, String>`
+- `working_dir: Option<String>`, `mode: ProcessMode`, `initial_pty_size: Option<PtySize>`
+- `kill_timeout: Option<Duration>` — escalate SIGTERM → SIGKILL after this duration (None = no escalation)
+- `stdin_buffer_limit: Option<usize>` — buffer stdin writes when pending bytes exceed limit (None = unlimited)
+
+**ProcessMode** — `Interactive` | `Automated` (Copy)
+
+**ExitStatus** — `Code(i32)` | `Signal(i32)` | `Unknown` (Copy)
+
+**Signal** — `Terminate` | `Kill` | `Hangup` | `Interrupt` | `Other(i32)` (Copy)
+
+**ProcessError** — `SpawnFailed { reason }` | `ConnectionLost { reason }` | `InvalidState { attempted, current_state }`
+
+**OutputStream** — `Stdout` | `Stderr` (Copy)
+
+**SubscriberSet** — deduplicated `Vec<ActorAddress>` with linear-scan dedup. Methods: `add()`, `remove()`, `snapshot()`, `count()`.
+
+### Layer 2 — ProcessDriver Trait + MockDriver
+
+```rust
+pub trait ProcessDriver: Send {
+    fn execute(&mut self, action: ProcessAction);
+    fn poll(&mut self) -> Vec<ProcessEvent>;
+}
+```
+
+**MockDriver** — test-oriented implementation:
+- `inject(event)` / `inject_many(events)` — queue events for `poll()`
+- `executed_actions()` — view recorded actions
+- `take_executed_actions()` — take + clear recorded actions
+- `pending_event_count()` — number of queued events
+- `poll()` drains all pending events, `execute()` records actions
+
+---
+
+## Implemented: Layers 3 + 4 (Actor Integration + Local OS Processes)
+
+### ExternalSender (swactor core primitive)
+
+A `Clone + Send + Sync` handle for injecting messages into actor mailboxes from any thread. Lives in the `swactor` crate (because `Envelope` and `AddressMap` are `pub(crate)`).
+
+```rust
+// Create from a runtime
+let sender = runtime.create_sender();
+
+// Use from any thread (including I/O background threads)
+sender.send_to(actor_addr, MyMessage { ... })?;
+```
+
+**Implementation:** Clones of the runtime's `Arc<AddressMap>`, per-worker `Sender<Envelope>` channels, and `Arc<Vec<OnceLock<Thread>>>` for worker thread unparking. The `send_to` method looks up the actor's worker, pushes an envelope, and unparks the worker thread.
+
+**Changes to swactor core:**
+- `src/channel.rs` — Added `Clone` for `Sender<T>` (clones the inner `Arc`)
+- `src/runtime.rs` — Changed `worker_threads` from `Vec<OnceLock<Thread>>` to `Arc<Vec<OnceLock<Thread>>>`, added `ExternalSender` struct and `Runtime::create_sender()` factory
+
+### Layer 3 — Process Actor
+
+**`ProcessActor<D: ProcessDriver>`** — generic actor implementing `ActorInterface` with `Incoming = ProcessCommand`.
+
+**Message types:**
+
+```rust
+pub enum ProcessCommand {
+    WriteStdin { data: Vec<u8> },
+    SendSignal { signal: Signal },
+    ResizePty { size: PtySize },
+    CloseStdin,
+    Close,
+    Subscribe { address: ActorAddress },
+    Unsubscribe { address: ActorAddress },
+    PollTick,  // internal: sent by waker from I/O threads
+}
+
+pub enum ProcessNotification {
+    Started { process: ActorAddress },
+    Output { process: ActorAddress, data: Vec<u8>, stream: OutputStream },
+    Exited { process: ActorAddress, status: ExitStatus },
+    Error { process: ActorAddress, error: ProcessError },
+}
+```
+
+**Handle ordering:** Commands are processed first, then I/O events are drained. This ensures `Subscribe` registers the subscriber before `Started` (or other buffered events) get dispatched. `PollTick` has no command effect — it just triggers the drain.
+
+**Event queue (`EventQueue`):** Thin wrapper around `Arc<SegQueue<ProcessEvent>>`. I/O threads push events; `driver.poll()` drains them.
+
+**Waker (`ProcessWaker`):** `Arc<dyn Fn() + Send + Sync>` — constructed with a closure that sends `PollTick` via `ExternalSender`. I/O threads call `waker.wake()` after pushing events.
+
+**Factory functions:**
+
+```rust
+// Spawn with real OS subprocess
+let addr = spawn_local_process(ctx, &sender, spec)?;
+
+// Spawn with custom driver (for testing)
+let addr = spawn_process(ctx, &sender, spec, driver, waker_slot)?;
+```
+
+The factory creates the driver, session, and actor, spawns it, then fills the waker slot with a closure that sends `PollTick` to the actor's address.
+
+### Layer 4 — LocalDriver
+
+Real OS process management via `std::process::Command` with piped I/O.
+
+**Components:**
+
+| File | Purpose |
+|------|---------|
+| `local/mod.rs` | `LocalDriver` struct, `ProcessDriver` impl, process spawning |
+| `local/pipes.rs` | Background thread reading stdout/stderr pipes (8KB buffer) |
+| `local/signal.rs` | `Signal` → libc constant mapping, `kill()` wrapper |
+| `local/wait.rs` | Background `waitpid()` thread with WIFEXITED/WIFSIGNALED decoding |
+
+**Thread structure per process:**
+- 1 stdout reader thread
+- 1 stderr reader thread
+- 1 waitpid thread
+
+Each thread pushes events to the shared `EventQueue` and calls `waker.wake()`.
+
+**Drop behavior:** Closes stdin, kills the process, waits for exit.
+
+**PTY support:** Not yet implemented — `ResizePty` is a no-op that returns a `PtyResized` ack. Pipe-based I/O only in this phase.
+
+---
+
+## File Structure
+
+```
+swactor (root crate):
+  src/
+    channel.rs       — + Clone for Sender<T>
+    runtime.rs       — + ExternalSender, create_sender(), Arc<worker_threads>
+
+crates/process/ (swactor-process):
+  Cargo.toml         — + crossbeam-queue, libc deps
+  src/
+    lib.rs           — module declarations + re-exports
+    types.rs         — ProcessSpec, ProcessMode, ExitStatus, Signal, PtySize, etc.
+    event.rs         — ProcessEvent enum
+    action.rs        — ProcessAction enum + OutputStream
+    subscriber.rs    — SubscriberSet
+    session.rs       — ProcessSession state machine
+    driver.rs        — ProcessDriver trait
+    mock.rs          — MockDriver
+    queue.rs         — EventQueue (Arc<SegQueue>)
+    waker.rs         — ProcessWaker (Arc<dyn Fn>)
+    message.rs       — ProcessCommand, ProcessNotification
+    actor.rs         — ProcessActor<D> impl ActorInterface
+    spawn.rs         — spawn_local_process(), spawn_process() factory functions
+    local/
+      mod.rs         — LocalDriver struct + ProcessDriver impl
+      pipes.rs       — Pipe reader background threads
+      signal.rs      — OS signal delivery
+      wait.rs        — waitpid background thread
+  tests/
+    session_scenarios.rs  — 26 session state machine scenario tests
+    proptest_session.rs   — 5 property-based session tests (KillTimeout included in arb_event)
+    actor_scenarios.rs    — 6 actor integration tests (TestDriver)
+    local_driver.rs       — 6 LocalDriver integration tests (real processes)
+    e2e_process.rs        — 2 end-to-end tests (Runtime + LocalDriver + real processes)
+```
+
+---
+
+## Test Coverage
+
+### Layers 1 + 2 — Session + MockDriver (31 tests)
+
+**Scenario tests** (26 tests in `tests/session_scenarios.rs`):
+1. Happy path automated: new → Started → OutputReceived×N → Exited(0)
+2. Interactive session with subscriber lifecycle (add/remove, verify notification membership)
+3. Spawn failure → error notification + SelfTerminate
+4. Connection loss mid-run → Exited with Unknown status
+5. Close before start → deferred SIGTERM on belated start
+6. Invalid event in Starting → NotifyError (no panic)
+7. Invalid event in Exited → NotifyError (no panic)
+8. Stdin closed then write → NotifyError
+9. MockDriver round-trip (driver + session in simulated tick loop)
+10. Signal escalation in Stopping (Kill after Terminate)
+11. Late acks in Exited silently consumed
+12. Flow control tracks pending stdin bytes (including saturating subtract)
+13. CloseStdin allowed in Stopping
+14. Connection loss in Stopping → Exited
+15. Duplicate CloseRequested in Stopping → no-op
+16. CloseRequested with kill_timeout emits both SendSignal{Terminate} and ScheduleKillTimeout
+17. Close-before-start with kill_timeout schedules timer on belated start
+18. KillTimeout in Stopping → SendSignal{Kill}, state stays Stopping
+19. KillTimeout silently consumed in Starting, Running, Exited
+20. CloseRequested without kill_timeout emits no ScheduleKillTimeout
+21. Full escalation flow: CloseRequested → KillTimeout → Exited{Signal(9)}
+22. Backpressure buffers writes when pending bytes exceed limit
+23. StdinWritten ack drains buffered chunks in FIFO order
+24. CloseRequested clears stdin buffer
+25. No backpressure when limit is None (all writes pass through)
+26. Exited clears stdin buffer
+
+**Property-based tests** (5 tests in `tests/proptest_session.rs`):
+1. No panics for arbitrary event sequences (up to 50 events, including KillTimeout)
+2. Exited is terminal (state never leaves Exited)
+3. SelfTerminate always last action when entering Exited
+4. Subscriber count matches add/remove operations
+5. State monotonicity (state ordinal never decreases)
+
+### Layer 3 — Actor Integration (6 tests)
+
+Tests in `tests/actor_scenarios.rs` using a `TestDriver` (shared `EventQueue` + recorded actions):
+
+1. **Happy path** — spawn → Started → Output → Exited → subscriber gets all notifications → actor stops
+2. **PollTick drains queued events** — three events buffered, single PollTick delivers all three notifications
+3. **Close triggers graceful shutdown** — Close command produces SIGTERM via driver
+4. **WriteStdin/SendSignal forwarded** — commands reach the driver as actions
+5. **Spawn failure** — error notification sent to subscriber, actor self-terminates
+6. **Subscribe/Unsubscribe routing** — two subscribers, unsubscribe one, only remaining gets subsequent notifications
+
+### Layer 4 — LocalDriver Integration (6 tests)
+
+Tests in `tests/local_driver.rs` using real OS processes, no actor layer:
+
+1. **`echo hello`** — Started + OutputReceived("hello\n") + Exited(0)
+2. **`cat` stdin echo** — write "ping\n" → read "ping\n" back → close stdin → Exited(0)
+3. **`sleep 60` + SIGTERM** — Started → send Terminate → Exited(Signal)
+4. **Bad command** → SpawnFailed
+5. **`seq 1 10000`** — large output integrity (no data loss, correct start/end)
+6. **Kill timeout escalation** — spawn SIGTERM-ignoring process, ScheduleKillTimeout fires KillTimeout, SIGKILL terminates it
+
+### End-to-End (2 tests)
+
+Tests in `tests/e2e_process.rs` — full stack (Runtime + ExternalSender + ProcessActor + LocalDriver + real process):
+
+1. **`echo hello` lifecycle** — spawn, subscribe, verify Started → Output("hello") → Exited(0) in order
+2. **Bad command** — spawn nonexistent binary, verify Error notification arrives
+
+---
+
+## Design Decisions Made
+
+1. **ExternalSender over WorkerExtension:** The I/O → actor bridge is a general-purpose swactor core primitive, not process-specific. Any crate can use `ExternalSender` to inject messages from background threads.
+
+2. **Handle ordering (command first, then drain):** Processing the incoming command before draining I/O events ensures that `Subscribe` registers the subscriber before buffered events (like `Started`) are dispatched. This avoids a race where early lifecycle events are sent to an empty subscriber list.
+
+3. **ProcessActor is generic over `D: ProcessDriver`:** Enables testing with `TestDriver` while production uses `LocalDriver`. No trait object overhead.
+
+4. **Thread-per-pipe model:** Each LocalDriver spawns 3 threads (stdout reader, stderr reader, waitpid). Simple, debuggable, correct for Phase 1 (tens of processes).
+
+5. **EventQueue is lock-free:** Uses `crossbeam_queue::SegQueue` — no contention between I/O writer threads and the actor's poll draining.
+
+6. **Waker uses OnceLock:** The waker slot (`Arc<OnceLock<ProcessWaker>>`) is filled after the actor address is known. I/O threads that call `waker.get()` before it's set simply skip the wake — events accumulate in the EventQueue and are drained on the next message.
+
+---
+
+## Next Steps
+
+### Near-term
+
+1. **PTY support for Interactive mode** — The `LocalDriver` currently uses pipes only. Interactive mode needs PTY allocation (via raw libc: `openpty()` → `fork()` → `setsid()` + `ioctl(TIOCSCTTY)` + `dup2` + `execvp`), `SIGWINCH` for resize, and merged stdout/stderr on a single PTY master FD. The `ResizePty` action is already wired through as a no-op.
+
+2. **Output buffering policies** — Subscribers currently receive every raw byte chunk. Add optional line-buffering or size-buffering in the session layer for consumers that want complete lines.
+
+### Layer 5 — SshDriver
+
+SSH-based process management. Same `ProcessDriver` trait, different backend.
+
+**Open decisions:**
+- **SSH library:** `russh` (pure Rust, async — needs tokio bridge) vs. `ssh2` (libssh2 bindings, synchronous — fits the thread model naturally)
+- **Authentication:** Password, key file, agent forwarding, or pluggable credential provider
+- **Connection multiplexing:** One SSH connection per process actor, or connection pool with multiple channels
+- **Health monitoring:** Heartbeat/keepalive to detect connection drops → `ConnectionLost` events
+
+### Scaling Path
+
+The architecture isolates scaling concerns in the driver layer:
+
+- **Phase 1 (tens):** Each driver spawns OS threads for I/O. Simple, debuggable. ← **current**
+- **Phase 2 (hundreds):** Shared thread pool for driver I/O. Replace per-process threads with a pool that multiplexes reads across processes.
+- **Phase 3 (thousands):** Async internals (tokio tasks for I/O). State machine and actor layers unchanged — only `ProcessDriver` implementations change.
+
+---
+
+## Alternative Approaches Considered
+
+### WorkerExtension Approach
+
+Managing processes as a per-worker extension (like TimerWheel). Rejected because:
+- Ties processes to specific workers, complicating supervision
+- Processes can't benefit from the actor model's naming, grouping, and monitoring
+- The API would be less intuitive than "send a message to the process"
+- Tick-bound latency is problematic for interactive use
+
+### Pure Bridge Actor Approach
+
+A single centralized bridge actor owning all processes (like IrohDriver). Rejected as the primary design because:
+- Doesn't give individual processes actor identity — can't supervise, name, or monitor them independently
+- Centralizes failure — the bridge dying kills all processes
+- However, this pattern does appear inside the recommended approach: the driver layer within each process actor is essentially a tiny bridge
+
+### Pure Process-as-Actor (without state machine)
+
+Just actors with embedded I/O logic, no state machine separation. Rejected because:
+- Untestable without real processes or SSH connections
+- Can't simulate
+- Backend-specific logic (SSH vs. local) interleaved with lifecycle logic

src/channel.rs

@@ -55,6 +55,14 @@ pub(crate) struct Sender<T> {
     queue: Arc<HybridChannel<T>>,
 }
 
+impl<T> Clone for Sender<T> {
+    fn clone(&self) -> Self {
+        Self {
+            queue: self.queue.clone(),
+        }
+    }
+}
+
 impl<T> Sender<T> {
     pub fn send(&self, value: T) {
         self.queue.push(value)

src/runtime.rs

@@ -110,7 +110,7 @@ pub struct Runtime {
     /// Workers available for tick(). run() drains this and moves workers to threads.
     tick_workers: RefCell<Vec<Worker>>,
     /// Thread handles for waking parked workers. Set by workers on startup via OnceLock.
-    worker_threads: Vec<OnceLock<Thread>>,
+    worker_threads: Arc<Vec<OnceLock<Thread>>>,
     created_at: Instant,
     #[cfg(feature = "transport")]
     codec_registry: Option<Arc<crate::transport::CodecRegistry>>,
@@ -146,6 +146,50 @@ impl RuntimeAddress {
     }
 }
 
+/// A cloneable, `Send + Sync` handle for injecting messages into actor mailboxes
+/// from any thread — including non-actor I/O threads.
+///
+/// Created via [`Runtime::create_sender`]. The primary use case is bridging
+/// background I/O (e.g., pipe readers, network listeners) with the tick-based
+/// actor system.
+pub struct ExternalSender {
+    address_map: Arc<AddressMap>,
+    transfer_txs: Vec<Sender<Envelope>>,
+    worker_threads: Arc<Vec<OnceLock<Thread>>>,
+}
+
+impl Clone for ExternalSender {
+    fn clone(&self) -> Self {
+        Self {
+            address_map: self.address_map.clone(),
+            transfer_txs: self.transfer_txs.clone(),
+            worker_threads: self.worker_threads.clone(),
+        }
+    }
+}
+
+// Safety: All fields are Send+Sync (Arc<AddressMap> uses RwLock,
+// Sender<Envelope> wraps Arc<HybridChannel>, Thread is Send+Sync).
+unsafe impl Send for ExternalSender {}
+unsafe impl Sync for ExternalSender {}
+
+impl ExternalSender {
+    /// Send a typed message to an actor address, waking the owning worker thread.
+    ///
+    /// Returns `Err` if the address is not found in the runtime's address map.
+    pub fn send_to<M: Message>(&self, addr: ActorAddress, msg: M) -> Result<(), Error> {
+        match self.address_map.lookup(&addr) {
+            Some(wid) => {
+                self.transfer_txs[wid.as_usize()]
+                    .send(Envelope::new(addr, Box::new(msg)));
+                notify_worker(&self.worker_threads, wid.as_usize());
+                Ok(())
+            }
+            None => Err(Error::from("Address not found")),
+        }
+    }
+}
+
 impl Runtime {
     /// Builds a new `Runtime` struct, but does not yet run anything. If multithreaded, call
     /// `run()`, if single threaded, needs to be driven by calls to the `tick()` method.
@@ -188,8 +232,8 @@ impl Runtime {
 
         let placement = Placement::new(num_workers, worker_stats.clone());
 
-        let worker_threads: Vec<OnceLock<Thread>> =
-            (0..num_workers).map(|_| OnceLock::new()).collect();
+        let worker_threads: Arc<Vec<OnceLock<Thread>>> =
+            Arc::new((0..num_workers).map(|_| OnceLock::new()).collect());
 
         let rt = Self {
             config,
@@ -317,6 +361,18 @@ impl Runtime {
         })
     }
 
+    /// Create an [`ExternalSender`] handle for injecting messages from any thread.
+    ///
+    /// The returned handle is `Clone + Send + Sync` and can be moved into
+    /// background I/O threads to bridge external events into the actor system.
+    pub fn create_sender(&self) -> ExternalSender {
+        ExternalSender {
+            address_map: self.address_map.clone(),
+            transfer_txs: self.transfer_txs.iter().map(|tx| tx.clone()).collect(),
+            worker_threads: self.worker_threads.clone(),
+        }
+    }
+
     fn make_tick_context(&self) -> TickContext<'_> {
         TickContext {
             address_map: &self.address_map,
@@ -438,7 +494,7 @@ impl Runtime {
 
         self.is_running.store(false, Ordering::Release);
         // Wake all parked workers so they see the shutdown flag immediately
-        for thread in &self.worker_threads {
+        for thread in self.worker_threads.iter() {
             if let Some(t) = thread.get() {
                 t.unpark();
             }