phase 3: port alknet-tty-local behind the local feature

Folds the alknet-tty-local crate into alktty as a feature-gated local
submodule (the single-crate + local-feature design from ADR-054; the
cyclic-dep workaround the mono-repo needed doesn't apply to one crate).

src/local/mod.rs:
- Module root, re-exports LocalTtyBackend, declares backend/pipe/pty
- Doc-commented as feature-gated + non-wasm by design (portable-pty +
  tokio::process need a real OS; enabling local on a wasm target is a
  build error by design)

src/local/backend.rs:
- LocalTtyBackend: implements crate::backend::TtyBackend
- allocate() dispatches on params.terminal: Some -> pty::allocate_pty,
  None -> pipe::allocate_pipe (ADR-054)
- 7 tests: PTY/pipe dispatch, empty-cmd rejection (both modes),
  resource_id returns None, new() constructs

src/local/pty.rs:
- portable_pty + 3 std threads (reader/writer/waiter) feeding tokio
  mpsc/oneshot (REQ-TTY-01 blocking->async bridge)
- PtyControl: resize + REQ-TTY-02 process-group signal forwarding
  (libc::kill(-pgid, sig) + kill(pid, sig) fallback + ChildKiller::kill
  for unknown names)
- LocalExitFuture: ADR-056 kill-on-Drop guard (ChildKiller on cancel,
  Option::take disarms on resolve)
- StdinSink: AsyncWrite over mpc::Sender<StdinCmd> with in-flight
  reserve+send parking for full-channel backpressure
- 7 tests incl. process-group reach (bash -c sleep), cancel-cleanup,
  unknown-signal fallback

src/local/pipe.rs:
- tokio::process::Command + tokio_util::io::ReaderStream
- PipeControl: no-op resize, libc::kill(pid, sig) with SIGKILL
  fallback for unknown names (pid-only, no process group - documented
  limitation of the runner case)
- PipeExitFuture: in-place Child::wait() poll (avoids self-referential
  borrow) + ADR-056 kill-on-Drop guard (start_kill on cancel)
- BytesStream: wraps ReaderStream, strips io::Error to EOF
- 9 tests incl. separate stderr, SIGTERM=-15, SIGKILL=-9 fallback,
  cancel-cleanup pid probe (kill(pid,0) returns ESRCH)

Import migration: alknet_tty::backend::{...} -> crate::backend::{...},
alknet_tty::control::signal_from_name -> crate::control::signal_from_name.

Cargo.toml: no changes needed (portable-pty + tokio-util optional, libc
under cfg(unix), and the local feature wiring tokio/process +
tokio/rt-multi-thread were already in the scaffold per Phase 0).

Also fixes pre-existing rustfmt drift in adapter.rs/channels.rs/
session.rs/lib.rs left by the phase 2 commit (cargo fmt --check without
--features local reported 28 diffs; cargo fmt does not accept
--features, so the earlier 'clean' check was a false negative — the
check errored on the unknown flag and grepped an empty stdout). Lesson:
run cargo fmt --check with no feature flags; cargo fmt doesn't gate on
features.

Verification:
- cargo test --features local -> 103/103 pass (was 80 at phase 2 end;
  +23 new tests across the 3 local modules: 7 backend, 7 pty, 9 pipe)
- cargo test (no features) -> 80/80 pass (local module not compiled)
- cargo clippy --features local --all-targets -> clean
- cargo clippy --all-targets (no features) -> clean
- cargo check --target wasm32-unknown-unknown -> clean (default crate
  stays wasm-clean; local is feature-gated and non-wasm by design)
- cargo fmt --check -> clean

src/adapter.rs

@@ -76,8 +76,7 @@ use tracing::{debug, warn};
 use crate::backend::{TtyBackend, TtyHandle};
 use crate::control::ControlMessage;
 use crate::negotiation::{
-    error_response_bytes, NegotiateRequest, NegotiationError, NegotiationReader,
+    error_response_bytes, NegotiateRequest, NegotiationError, NegotiationReader, NegotiationWriter,
-    NegotiationWriter,
 };
 use crate::wire::{Chunk, ChunkReader, ChunkWriter, RawError, STREAM_CTRL_IN, STREAM_STDIN};
 

src/channels.rs

@@ -182,7 +182,8 @@ fn make_tty_open_handler(
     ownership: Option<Arc<dyn OwnershipProvider>>,
     identity: Option<alkcall::core::auth::Identity>,
 ) -> OpenHandler {
-    Arc::new(move |input: Value, channel_conn: Connection, auth: AuthContext| {
+    Arc::new(
+        move |input: Value, channel_conn: Connection, auth: AuthContext| {
             let backends = Arc::clone(&backends);
             let ownership = ownership.clone();
             let identity = identity.clone().or_else(|| auth.identity.clone());
@@ -199,16 +200,17 @@ fn make_tty_open_handler(
                 let (client_read, client_write) = tokio::io::split(stream);
                 drive_session(client_write, client_read, backends, ownership, identity).await;
             })
-    })
+        },
+    )
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
     use crate::backend::{MockBackend, TtyError};
+    use alkcall::channels::client::ChannelClient;
     use alkcall::channels::operations::ChannelCore;
     use alkcall::channels::policy::default_policy;
-    use alkcall::channels::client::ChannelClient;
     use alkcall::core::auth::Identity;
     use alkcall::core::types::Connection as CoreConnection;
     use alkcall::registry::registration::OperationRegistry;
@@ -230,19 +232,14 @@ mod tests {
         ownership: Option<Arc<dyn OwnershipProvider>>,
         identity: Option<Identity>,
     ) -> ChannelClient {
-        use alkcall::channels::adapter::{
+        use alkcall::channels::adapter::{ChannelsAdapter, InstallChannelZero};
-            ChannelsAdapter, InstallChannelZero,
-        };
         use alkcall::core::auth::IdentityProvider;
         use alkcall::protocol::connection::split_single_stream;
         use alkcall::protocol::dispatch::Dispatcher;
 
         struct NoopIdProvider;
         impl IdentityProvider for NoopIdProvider {
-            fn resolve_from_fingerprint(
+            fn resolve_from_fingerprint(&self, _: &str) -> Option<alkcall::core::auth::Identity> {
-                &self,
-                _: &str,
-            ) -> Option<alkcall::core::auth::Identity> {
                 None
             }
             fn resolve_from_token(
@@ -257,8 +254,7 @@ mod tests {
         let policy_for_hook = Arc::clone(&policy);
         let identity_for_conn = identity.clone();
 
-        let install_hook: InstallChannelZero =
+        let install_hook: InstallChannelZero = Arc::new(move |manager, channel0_conn, auth| {
-            Arc::new(move |manager, channel0_conn, auth| {
             let backends = Arc::clone(&backends);
             let ownership = ownership.clone();
             let _identity = identity.clone();
@@ -306,16 +302,8 @@ mod tests {
         });
 
         let (client_end, server_end) = duplex(64 * 1024);
-        let client_conn = CoreConnection::from_bidi(
+        let client_conn = CoreConnection::from_bidi(client_end, b"alk/channels".to_vec(), None);
-            client_end,
+        let server_conn = CoreConnection::from_bidi(server_end, b"alk/channels".to_vec(), None);
-            b"alk/channels".to_vec(),
-            None,
-        );
-        let server_conn = CoreConnection::from_bidi(
-            server_end,
-            b"alk/channels".to_vec(),
-            None,
-        );
         // Set the identity on the server connection so the call
         // dispatch sees it (the ACL check runs against the
         // connection's identity, not the `install_channel_zero`
@@ -327,10 +315,8 @@ mod tests {
         let adapter = ChannelsAdapter::new(install_hook, Arc::new(NoCap));
         let auth = AuthContext::anonymous(b"alk/channels");
         let _server_handle = tokio::spawn(async move {
-            let _ = alkcall::core::types::ProtocolHandler::handle(
+            let _ =
-                &adapter, server_conn, &auth,
+                alkcall::core::types::ProtocolHandler::handle(&adapter, server_conn, &auth).await;
-            )
-            .await;
         });
 
         ChannelClient::from_connection(client_conn)
@@ -376,8 +362,7 @@ mod tests {
             .get("required")
             .and_then(|v| v.as_array())
             .expect("required array");
-        let required_names: Vec<&str> =
+        let required_names: Vec<&str> = required.iter().filter_map(|v| v.as_str()).collect();
-            required.iter().filter_map(|v| v.as_str()).collect();
         assert!(required_names.contains(&"carriage"));
         assert!(required_names.contains(&"backend"));
         assert!(required_names.contains(&"cmd"));
@@ -393,8 +378,7 @@ mod tests {
         let manager = alkcall::channels::manager::ChannelManager::with_defaults(handle, None);
         let core = ChannelCore::new(manager, default_policy());
 
-        let backends: Arc<HashMap<String, Arc<dyn TtyBackend>>> =
+        let backends: Arc<HashMap<String, Arc<dyn TtyBackend>>> = Arc::new(HashMap::new());
-            Arc::new(HashMap::new());
         let mut registry = OperationRegistry::new();
         register_openable(
             &core,
@@ -415,10 +399,7 @@ mod tests {
     #[tokio::test]
     async fn end_to_end_open_op_returns_channel_id() {
         let mut backends: HashMap<String, Arc<dyn TtyBackend>> = HashMap::new();
-        backends.insert(
+        backends.insert("mock".to_string(), Arc::new(MockBackend::with_exit_code(0)));
-            "mock".to_string(),
-            Arc::new(MockBackend::with_exit_code(0)),
-        );
         let backends = Arc::new(backends);
 
         let identity = Identity {
@@ -459,10 +440,7 @@ mod tests {
     #[tokio::test]
     async fn end_to_end_open_op_denies_without_tty_open_scope() {
         let mut backends: HashMap<String, Arc<dyn TtyBackend>> = HashMap::new();
-        backends.insert(
+        backends.insert("mock".to_string(), Arc::new(MockBackend::with_exit_code(0)));
-            "mock".to_string(),
-            Arc::new(MockBackend::with_exit_code(0)),
-        );
         let backends = Arc::new(backends);
 
         let identity = Identity {
@@ -488,7 +466,9 @@ mod tests {
 
         let err = response.result.expect_err("open op should be denied");
         assert!(
-            err.message.contains("scope") || err.code.contains("FORBIDDEN") || err.code.contains("AUTH"),
+            err.message.contains("scope")
+                || err.code.contains("FORBIDDEN")
+                || err.code.contains("AUTH"),
             "error should mention scope/forbidden/auth, got code={} message={}",
             err.code,
             err.message

src/local/backend.rs

@@ -0,0 +1,179 @@
+//! `LocalTtyBackend`: the [`crate::backend::TtyBackend`] implementation
+//! for local processes.
+//!
+//! `allocate()` branches on `TtyParams::terminal`: `Some(TerminalParams)`
+//! selects PTY mode (this crate's `pty` module); `None` selects pipe mode
+//! (the runner case, this crate's `pipe` module). One backend serves both
+//! (ADR-054).
+
+use crate::backend::{TtyBackend, TtyError, TtyHandle, TtyParams};
+use async_trait::async_trait;
+
+/// Local TTY backend: implements [`crate::backend::TtyBackend`] for local
+/// processes.
+///
+/// `allocate()` branches on `TtyParams::terminal`: `Some(TerminalParams)`
+/// selects PTY mode (this crate's `pty` module); `None` selects pipe mode
+/// (the runner case, this crate's `pipe` module). One backend serves both
+/// (ADR-054).
+///
+/// Takes no constructor dependencies — unlike `DockerTtyBackend` (wraps a
+/// `bollard::Docker` client) or `SshTtyBackend` (wraps an SSH session), the
+/// `portable_pty` system is process-global. The assembly layer constructs one
+/// `LocalTtyBackend` and registers it as `"local"`.
+pub struct LocalTtyBackend;
+
+impl LocalTtyBackend {
+    /// Construct a new `LocalTtyBackend`. The backend is dependency-free;
+    /// `portable_pty`'s native system is process-global.
+    pub fn new() -> Self {
+        Self
+    }
+}
+
+impl Default for LocalTtyBackend {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+#[async_trait]
+impl TtyBackend for LocalTtyBackend {
+    /// Allocate a terminal/process session, branching on
+    /// `params.terminal`.
+    ///
+    /// `Some(TerminalParams)` dispatches to [`crate::local::pty::allocate_pty`]
+    /// (PTY mode, real terminal semantics — resize, process-group signal
+    /// forwarding, merged stdout/stderr). `None` dispatches to
+    /// [`crate::local::pipe::allocate_pipe`] (pipe mode, the runner case —
+    /// separate stdout/stderr, no-op resize, pid-only signal).
+    ///
+    /// `backend_params` is ignored: the local backend has no
+    /// backend-specific selector fields. An empty `cmd` returns
+    /// [`TtyError::AllocFailed`] (the adapter already checks this at
+    /// negotiation, but the backend fails gracefully if called directly).
+    async fn allocate(&self, params: &TtyParams) -> Result<TtyHandle, TtyError> {
+        if params.cmd.is_empty() {
+            return Err(TtyError::AllocFailed {
+                message: "cmd must be non-empty".to_string(),
+            });
+        }
+        match &params.terminal {
+            Some(terminal) => crate::local::pty::allocate_pty(
+                terminal.clone(),
+                params.cmd.clone(),
+                params.cwd.clone(),
+                params.env.clone(),
+            ),
+            None => crate::local::pipe::allocate_pipe(
+                params.cmd.clone(),
+                params.cwd.clone(),
+                params.env.clone(),
+            ),
+        }
+    }
+
+    /// The local backend creates its own resource (a process), so there is
+    /// no pre-existing resource for the ownership check. Returns `None`.
+    fn resource_id(&self, _params: &TtyParams) -> Option<(&'static str, String)> {
+        None
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::collections::HashMap;
+
+    fn term() -> crate::backend::TerminalParams {
+        crate::backend::TerminalParams {
+            term: None,
+            cols: 80,
+            rows: 24,
+            pixel_width: 0,
+            pixel_height: 0,
+            modes: serde_json::Value::Null,
+        }
+    }
+
+    fn params(terminal: Option<crate::backend::TerminalParams>, cmd: Vec<String>) -> TtyParams {
+        TtyParams {
+            terminal,
+            cmd,
+            cwd: None,
+            env: HashMap::new(),
+            backend_params: serde_json::Map::new(),
+        }
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    async fn pty_dispatch_yields_no_stderr() {
+        let backend = LocalTtyBackend::new();
+        let handle = backend
+            .allocate(&params(
+                Some(term()),
+                vec!["echo".to_string(), "hi".to_string()],
+            ))
+            .await
+            .expect("allocate");
+        assert!(handle.stderr.is_none(), "PTY backends merge stdout/stderr");
+        assert!(handle.control.is_some(), "PTY backends provide control");
+        let _ = handle.exit_code.await;
+    }
+
+    #[tokio::test]
+    async fn pipe_dispatch_yields_some_stderr() {
+        let backend = LocalTtyBackend::new();
+        let handle = backend
+            .allocate(&params(None, vec!["echo".to_string(), "hi".to_string()]))
+            .await
+            .expect("allocate");
+        assert!(
+            handle.stderr.is_some(),
+            "pipe backends have separate stderr"
+        );
+        assert!(handle.control.is_some(), "pipe backends provide control");
+        let _ = handle.exit_code.await;
+    }
+
+    #[tokio::test]
+    async fn empty_cmd_returns_alloc_failed() {
+        let backend = LocalTtyBackend::new();
+        let result = backend.allocate(&params(None, vec![])).await;
+        assert!(
+            matches!(result, Err(TtyError::AllocFailed { .. })),
+            "expected AllocFailed for empty cmd"
+        );
+    }
+
+    #[tokio::test]
+    async fn empty_cmd_with_terminal_returns_alloc_failed() {
+        let backend = LocalTtyBackend::new();
+        let result = backend.allocate(&params(Some(term()), vec![])).await;
+        assert!(
+            matches!(result, Err(TtyError::AllocFailed { .. })),
+            "expected AllocFailed for empty cmd even with terminal set"
+        );
+    }
+
+    #[test]
+    fn resource_id_returns_none() {
+        let backend = LocalTtyBackend::new();
+        let params = params(None, vec!["true".to_string()]);
+        assert!(backend.resource_id(&params).is_none());
+    }
+
+    #[test]
+    fn resource_id_returns_none_with_terminal() {
+        let backend = LocalTtyBackend::new();
+        let params = params(Some(term()), vec!["true".to_string()]);
+        assert!(backend.resource_id(&params).is_none());
+    }
+
+    #[test]
+    fn new_constructs() {
+        let backend = LocalTtyBackend::new();
+        let params = params(None, vec!["true".to_string()]);
+        assert!(backend.resource_id(&params).is_none());
+    }
+}

src/local/mod.rs

@@ -0,0 +1,20 @@
+//! Local TTY backend for alktty: `LocalTtyBackend` implements
+//! [`crate::backend::TtyBackend`] via `portable_pty` (PTY mode, terminal
+//! semantics) and `tokio::process::Command` (pipe mode, the runner case).
+//!
+//! The blocking→async bridge for PTY mode uses three dedicated std
+//! threads feeding tokio mpsc/oneshot channels (REQ-TTY-01). Signal
+//! forwarding targets the foreground process group (REQ-TTY-02).
+//!
+//! This module is feature-gated behind the `local` feature and is
+//! non-wasm by design (`portable_pty` + `tokio::process` need a real
+//! OS). The default crate (no features) compiles to
+//! `wasm32-unknown-unknown`; enabling `local` on a wasm target is a
+//! build error by design — the local-process backend runs on a real
+//! OS (Linux, macOS, Windows), never in a sandbox.
+
+pub mod backend;
+pub mod pipe;
+pub mod pty;
+
+pub use backend::LocalTtyBackend;

src/local/pipe.rs

@@ -0,0 +1,469 @@
+//! Pipe mode for `LocalTtyBackend`: `tokio::process`-backed runner sessions.
+//!
+//! Spawns the command with `Stdio::piped()` for stdin/stdout/stderr, exposes
+//! tokio-native `AsyncRead`/`AsyncWrite` (no std-thread bridge needed),
+//! `PipeControl` (no-op resize, `libc::kill(pid, sig)` signal), and the
+//! ADR-056 kill guard on the `exit_code` future.
+
+use std::collections::HashMap;
+use std::future::Future;
+use std::path::PathBuf;
+use std::pin::Pin;
+use std::process::ExitStatus;
+use std::sync::Arc;
+use std::task::{Context, Poll};
+
+use crate::backend::{BoxFuture, TtyControl, TtyControlHandle, TtyError, TtyHandle};
+use crate::control::signal_from_name;
+use bytes::Bytes;
+use futures_core::Stream;
+use tokio::process::{Child, ChildStderr, ChildStdin, ChildStdout, Command};
+use tokio_util::io::ReaderStream;
+
+/// Allocate a pipe-mode session: spawn `cmd` with `Stdio::piped()` for
+/// stdin/stdout/stderr and return a `TtyHandle` whose `stderr` is `Some`
+/// (separate streams). The `exit_code` future carries the ADR-056 kill
+/// guard (see [`PipeExitFuture`]). `control` is a [`PipeControl`]
+/// (no-op resize, `libc::kill(pid, sig)` signal).
+pub fn allocate_pipe(
+    cmd: Vec<String>,
+    cwd: Option<PathBuf>,
+    env: HashMap<String, String>,
+) -> Result<TtyHandle, TtyError> {
+    if cmd.is_empty() {
+        return Err(TtyError::AllocFailed {
+            message: "empty command vector".to_string(),
+        });
+    }
+
+    let mut command = Command::new(&cmd[0]);
+    command.args(&cmd[1..]);
+    command.stdin(std::process::Stdio::piped());
+    command.stdout(std::process::Stdio::piped());
+    command.stderr(std::process::Stdio::piped());
+    command.kill_on_drop(true);
+    if let Some(cwd) = cwd {
+        command.current_dir(cwd);
+    }
+    for (k, v) in env {
+        command.env(k, v);
+    }
+
+    let mut child = command.spawn().map_err(|e| TtyError::AllocFailed {
+        message: format!("spawn failed: {e}"),
+    })?;
+
+    let pid = child.id();
+
+    let stdin: Box<dyn tokio::io::AsyncWrite + Send + Unpin> = child
+        .stdin
+        .take()
+        .map(|s: ChildStdin| Box::new(s) as Box<dyn tokio::io::AsyncWrite + Send + Unpin>)
+        .ok_or_else(|| TtyError::AllocFailed {
+            message: "child stdin not piped".to_string(),
+        })?;
+
+    let stdout: ChildStdout = child.stdout.take().ok_or_else(|| TtyError::AllocFailed {
+        message: "child stdout not piped".to_string(),
+    })?;
+    let stderr: ChildStderr = child.stderr.take().ok_or_else(|| TtyError::AllocFailed {
+        message: "child stderr not piped".to_string(),
+    })?;
+
+    let stdout: Pin<Box<dyn Stream<Item = Bytes> + Send>> =
+        Box::pin(BytesStream::wrap(ReaderStream::new(stdout)));
+    let stderr: Pin<Box<dyn Stream<Item = Bytes> + Send>> =
+        Box::pin(BytesStream::wrap(ReaderStream::new(stderr)));
+
+    let control = Some(TtyControlHandle::new(Arc::new(PipeControl::new(pid))));
+
+    let exit_code: BoxFuture<Result<i32, TtyError>> = Box::pin(PipeExitFuture::new(child));
+
+    Ok(TtyHandle {
+        stdin,
+        stdout,
+        stderr: Some(stderr),
+        exit_code,
+        control,
+    })
+}
+
+/// Adapter wrapping `tokio_util::io::ReaderStream<R>` (which yields
+/// `Result<Bytes, io::Error>`) into a `Stream<Item = Bytes>`, dropping the
+/// `io::Error` info by ending the stream on read error. The adapter pump
+/// treats a stream end as EOF; a read error is indistinguishable from EOF
+/// at the wire level (the `exit_code` future surfaces the failure if the
+/// child died abnormally).
+struct BytesStream<S> {
+    inner: S,
+}
+
+impl<S> BytesStream<S> {
+    fn wrap(inner: S) -> Self {
+        Self { inner }
+    }
+}
+
+impl<S, E> Stream for BytesStream<S>
+where
+    S: Stream<Item = Result<Bytes, E>> + Unpin,
+{
+    type Item = Bytes;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        match Pin::new(&mut self.inner).poll_next(cx) {
+            Poll::Pending => Poll::Pending,
+            Poll::Ready(None) => Poll::Ready(None),
+            Poll::Ready(Some(Ok(bytes))) => Poll::Ready(Some(bytes)),
+            // Read error: end the stream (EOF to the adapter).
+            Poll::Ready(Some(Err(_))) => Poll::Ready(None),
+        }
+    }
+}
+
+/// Control handle for pipe mode: no-op resize and `libc::kill(pid, sig)`
+/// signal forwarding (the runner case — ADR-054).
+///
+/// # Signal scope — no process group (documented limitation)
+///
+/// Pipe mode has no session leader / controlling tty, so `signal()` targets
+/// the direct child pid only via `libc::kill(pid, sig)`. If the child has
+/// spawned its own children, they will NOT receive the signal — this is a
+/// known limitation of the runner case. A runner that needs process-group
+/// signal delivery uses the PTY case, not the pipe case. See
+/// `tty-local.md` §"Pipe Mode".
+pub struct PipeControl {
+    pid: Option<u32>,
+}
+
+impl PipeControl {
+    fn new(pid: Option<u32>) -> Self {
+        Self { pid }
+    }
+
+    #[cfg(test)]
+    pub(crate) fn pid(&self) -> Option<u32> {
+        self.pid
+    }
+}
+
+impl TtyControl for PipeControl {
+    fn resize(&self, _cols: u16, _rows: u16, _pixel_width: u16, _pixel_height: u16) {
+        // No-op — no PTY in pipe mode.
+    }
+
+    fn signal(&self, name: &str) {
+        // Targets the direct child pid only (no process group — see the
+        // type-level doc comment on `PipeControl`).
+        #[cfg(unix)]
+        {
+            if let Some(pid) = self.pid {
+                let pid = pid as libc::pid_t;
+                match signal_from_name(name) {
+                    Some(sig) => {
+                        // SAFETY: libc::kill on a known pid with a valid
+                        // signal number is the documented signal-forwarding
+                        // path. Best-effort: ignore errors (child may be gone).
+                        let _ = unsafe { libc::kill(pid, sig) };
+                    }
+                    None => {
+                        // Unknown name → SIGKILL. On Unix, `libc::kill(pid,
+                        // SIGKILL)` is exactly what
+                        // `tokio::process::Child::start_kill()` does
+                        // internally; the `Child` handle lives in the
+                        // `PipeExitFuture` and isn't accessible here, so we
+                        // kill by pid instead.
+                        let _ = unsafe { libc::kill(pid, libc::SIGKILL) };
+                    }
+                }
+            } else {
+                tracing::warn!(name = name, "PipeControl::signal: no pid; cannot signal");
+            }
+        }
+        #[cfg(not(unix))]
+        {
+            // Non-Unix: no `libc::kill` available, and the `Child` handle
+            // lives in the `PipeExitFuture` (not accessible from here).
+            // Dropping the `TtyHandle` triggers the ADR-056 kill guard.
+            let _ = name;
+            tracing::warn!(
+                name = name,
+                "PipeControl::signal: non-Unix; drop TtyHandle to kill"
+            );
+        }
+    }
+}
+
+/// The `exit_code` future for pipe mode: wraps `tokio::process::Child::wait`
+/// plus the ADR-056 kill guard.
+///
+/// # ADR-056 — kill-on-Drop contract
+///
+/// On cancel (the future is dropped without being driven to completion),
+/// `Drop` calls `Child::start_kill()` (SIGKILL); the child exits and the
+/// inner `wait` future reaps it. On the happy path, `poll` resolves the
+/// `wait` future, then **disarms** the guard by taking the `Child` out of
+/// `self.child` — so the subsequent `Drop` is a no-op. The guard is the
+/// spec-compliant mechanism; `kill_on_drop(true)` on the `Command` is a
+/// defense-in-depth backstop, not the primary path (the `Child` is moved
+/// into the future and the future's `Drop` is the cancel path, not the
+/// `Child`'s).
+///
+/// # Implementation note
+///
+/// `tokio::process::Child::wait()` borrows `&mut self`, so the `wait` future
+/// cannot be held alongside the `Child` in the same struct without a
+/// self-referential borrow. Instead, the `Child` lives in an `Option<Child>`
+/// and is polled in place: on each `poll`, we `as_mut()` the `Option`,
+/// construct a fresh `Child::wait()` borrow-future, pin it locally, and drive
+/// it. The borrow ends when the local future is dropped at the end of the
+/// `poll` call, so the `Child` remains free to be `take`n (disarmed or
+/// killed) outside the borrow. This is the same pattern the standard
+/// library's `tokio::process` examples use for "wait with a kill guard."
+struct PipeExitFuture {
+    child: Option<Child>,
+}
+
+impl PipeExitFuture {
+    fn new(child: Child) -> Self {
+        Self { child: Some(child) }
+    }
+}
+
+impl Future for PipeExitFuture {
+    type Output = Result<i32, TtyError>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        // Poll the wait future in an inner scope so the borrow of
+        // `self.child` ends before we `take()` the child to disarm/kill.
+        enum Step {
+            Pending,
+            Exited(ExitStatus),
+            Failed(String),
+            Taken,
+        }
+        let step = match self.child.as_mut() {
+            None => Step::Taken,
+            Some(child) => {
+                let wait = child.wait();
+                tokio::pin!(wait);
+                match Future::poll(Pin::new(&mut wait), cx) {
+                    Poll::Pending => Step::Pending,
+                    Poll::Ready(Ok(status)) => Step::Exited(status),
+                    Poll::Ready(Err(e)) => Step::Failed(format!("wait failed: {e}")),
+                }
+            }
+        };
+        // The borrow of `self.child` ended with `step`; now we can `take()`.
+        match step {
+            Step::Pending => Poll::Pending,
+            Step::Exited(status) => {
+                self.child.take();
+                Poll::Ready(Ok(exit_code_from(status)))
+            }
+            Step::Failed(msg) => {
+                self.child.take();
+                Poll::Ready(Err(TtyError::WaitFailed { message: msg }))
+            }
+            Step::Taken => Poll::Ready(Err(TtyError::WaitFailed {
+                message: "child already taken".to_string(),
+            })),
+        }
+    }
+}
+
+impl Drop for PipeExitFuture {
+    fn drop(&mut self) {
+        if let Some(mut child) = self.child.take() {
+            // Best-effort kill; the child may already be exiting.
+            let _ = child.start_kill();
+        }
+    }
+}
+
+/// Map an `ExitStatus` to the adapter's `i32` exit code: `code()` on normal
+/// exit; on Unix, the negative signal number on signal-terminated exit
+/// (matches `std` convention and ADR-055 §4). Falls back to `-1` (the
+/// adapter's "could not determine" sentinel — ADR-055 §4).
+fn exit_code_from(status: ExitStatus) -> i32 {
+    if let Some(code) = status.code() {
+        return code;
+    }
+    #[cfg(unix)]
+    {
+        use std::os::unix::process::ExitStatusExt;
+        if let Some(sig) = status.signal() {
+            return -sig;
+        }
+    }
+    -1
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tokio::io::AsyncWriteExt;
+    use tokio_stream::StreamExt;
+
+    fn sh_cmd(argv: &[&str]) -> Vec<String> {
+        argv.iter().map(|s| s.to_string()).collect()
+    }
+
+    /// Drain a `Stream<Item = Bytes>` into a `Vec<u8>`.
+    async fn drain(mut s: Pin<Box<dyn Stream<Item = Bytes> + Send>>) -> Vec<u8> {
+        let mut out = Vec::new();
+        while let Some(chunk) = s.next().await {
+            out.extend_from_slice(&chunk);
+        }
+        out
+    }
+
+    /// Swap a `Pin<Box<dyn Stream<Item = Bytes> + Send>>` out for an empty
+    /// stream, returning the original. (`TtyHandle.stdout` is not `Option`,
+    /// so `.take()` isn't available; this is the equivalent.)
+    fn swap_stdout(
+        slot: &mut Pin<Box<dyn Stream<Item = Bytes> + Send>>,
+    ) -> Pin<Box<dyn Stream<Item = Bytes> + Send>> {
+        std::mem::replace(slot, Box::pin(tokio_stream::empty()))
+    }
+
+    #[tokio::test]
+    async fn happy_path_echo() {
+        let mut handle =
+            allocate_pipe(sh_cmd(&["echo", "hello"]), None, HashMap::new()).expect("allocate");
+        let stderr = handle.stderr.take().expect("stderr present");
+        let stdout = swap_stdout(&mut handle.stdout);
+        let out = drain(stdout).await;
+        let err = drain(stderr).await;
+        assert!(err.is_empty(), "stderr empty");
+        let code = handle.exit_code.await.expect("exit");
+        assert_eq!(code, 0);
+        assert_eq!(String::from_utf8_lossy(&out), "hello\n");
+    }
+
+    #[tokio::test]
+    async fn stdin_round_trip_cat() {
+        let mut handle = allocate_pipe(sh_cmd(&["cat"]), None, HashMap::new()).expect("allocate");
+        handle
+            .stdin
+            .write_all(b"round-trip\n")
+            .await
+            .expect("write stdin");
+        // Close stdin so `cat` sees EOF and exits. Replace with a sink
+        // (`TtyHandle.stdin` is not `Option`; this drops the child stdin).
+        handle.stdin = Box::new(tokio::io::sink());
+        let stdout = swap_stdout(&mut handle.stdout);
+        let out = drain(stdout).await;
+        let code = handle.exit_code.await.expect("exit");
+        assert_eq!(code, 0);
+        assert_eq!(String::from_utf8_lossy(&out), "round-trip\n");
+    }
+
+    #[tokio::test]
+    async fn separate_stderr() {
+        let mut handle = allocate_pipe(sh_cmd(&["sh", "-c", "echo err >&2"]), None, HashMap::new())
+            .expect("allocate");
+        let stderr = handle.stderr.take().expect("stderr present");
+        let stdout = swap_stdout(&mut handle.stdout);
+        let out = drain(stdout).await;
+        assert!(out.is_empty(), "stdout empty");
+        let err = drain(stderr).await;
+        let code = handle.exit_code.await.expect("exit");
+        assert_eq!(code, 0);
+        assert_eq!(String::from_utf8_lossy(&err), "err\n");
+    }
+
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn signal_term_kills_child() {
+        let handle =
+            allocate_pipe(sh_cmd(&["sleep", "60"]), None, HashMap::new()).expect("allocate");
+        let control = handle.control.clone().expect("control present");
+        control.signal("TERM");
+        let code = handle.exit_code.await.expect("exit");
+        assert_eq!(code, -15, "SIGTERM = -15");
+    }
+
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn cancel_cleanup_drops_kill_child() {
+        // Use a temp file to capture the child's pid, then drop the handle
+        // (which drops exit_code → the ADR-056 guard sends SIGKILL) and
+        // assert the process no longer exists via `kill(pid, 0)`.
+        let pid_file = std::env::temp_dir().join(format!(
+            "alktty_pipe_cancel_pid_{}_{}.txt",
+            std::process::id(),
+            rand_seed()
+        ));
+        let cmd = sh_cmd(&[
+            "sh",
+            "-c",
+            &format!("echo $$ > '{}'; exec sleep 60", pid_file.display()),
+        ]);
+        let handle = allocate_pipe(cmd, None, HashMap::new()).expect("allocate");
+        // Wait for the shell to write its pid.
+        for _ in 0..100 {
+            if pid_file.exists() {
+                break;
+            }
+            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+        }
+        let pid_str = std::fs::read_to_string(&pid_file).expect("pid file written");
+        let pid: i32 = pid_str.trim().parse().expect("pid parses");
+        let _ = std::fs::remove_file(&pid_file);
+        // Drop the handle without awaiting exit_code — the ADR-056 guard
+        // must kill the child.
+        drop(handle);
+        // Give the kernel a moment to deliver SIGKILL and reap.
+        tokio::time::sleep(std::time::Duration::from_millis(200)).await;
+        // kill(pid, 0) returns ESRCH (no such process) when the child is gone.
+        let alive = unsafe { libc::kill(pid, 0) } == 0;
+        assert!(!alive, "child (pid={pid}) should be killed after drop");
+    }
+
+    #[tokio::test]
+    async fn resize_is_noop() {
+        let handle = allocate_pipe(sh_cmd(&["true"]), None, HashMap::new()).expect("allocate");
+        let control = handle.control.as_ref().expect("control present");
+        control.resize(120, 40, 800, 600);
+        control.resize(0, 0, 0, 0);
+    }
+
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn unknown_signal_falls_back_to_sigkill() {
+        let handle =
+            allocate_pipe(sh_cmd(&["sleep", "60"]), None, HashMap::new()).expect("allocate");
+        let control = handle.control.clone().expect("control present");
+        // Unknown name → SIGKILL fallback (equivalent to start_kill()).
+        control.signal("NOSUCH");
+        let code = handle.exit_code.await.expect("exit");
+        assert_eq!(code, -9, "SIGKILL = -9");
+    }
+
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn pipe_control_pid_recorded() {
+        let ctrl = PipeControl::new(Some(42));
+        assert_eq!(ctrl.pid(), Some(42));
+        let ctrl_none = PipeControl::new(None);
+        assert_eq!(ctrl_none.pid(), None);
+    }
+
+    #[test]
+    fn empty_command_returns_alloc_failed() {
+        let result = allocate_pipe(vec![], None, HashMap::new());
+        assert!(
+            matches!(result, Err(TtyError::AllocFailed { .. })),
+            "expected AllocFailed"
+        );
+    }
+
+    fn rand_seed() -> u64 {
+        use std::time::{SystemTime, UNIX_EPOCH};
+        SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .map(|d| d.as_nanos() as u64)
+            .unwrap_or(0)
+    }
+}

src/local/pty.rs

@@ -0,0 +1,705 @@
+//! PTY mode for `LocalTtyBackend`: `portable_pty`-backed terminal sessions.
+//!
+//! Implements the blocking→async bridge (REQ-TTY-01) via three dedicated
+//! std threads (reader, writer, waiter) feeding tokio mpsc/oneshot channels,
+//! `PtyControl` for resize/signal, REQ-TTY-02 process-group signal
+//! forwarding (`libc::kill(-pgid, sig)`), and the ADR-056 kill guard on the
+//! `exit_code` future.
+//!
+//! # REQ-TTY-01 — blocking→async bridge
+//!
+//! `portable_pty` exposes a blocking `std::io` API
+//! (`MasterPty::try_clone_reader()`, `take_writer()`, `Child::wait()`).
+//! The three-thread bridge is the reference pattern for any blocking
+//! backend: dedicated std threads feed tokio mpsc/oneshot channels, and
+//! the async-facing `TtyHandle` fields are wrappers over those channels.
+//!
+//! # REQ-TTY-02 — signal targets the process group
+//!
+//! `portable_pty` spawns the child as a session leader
+//! (`CommandBuilder::set_controlling_tty(true)`, the default), so the
+//! child's pid *is* its process-group id. `signal()` calls
+//! `libc::kill(-pgid, sig)` (the negative pid) to reach the whole group,
+//! with a `kill(pid, sig)` fallback if the group signal fails (e.g. the
+//! child already exited). Unknown signal names fall back to
+//! `ChildKiller::kill` (SIGHUP). See `tty-local.md` §"REQ-TTY-02".
+//!
+//! # ADR-056 — kill-on-Drop guard
+//!
+//! `LocalExitFuture` wraps the `oneshot::Receiver<i32>` from the waiter
+//! thread alongside a `portable_pty::ChildKiller`. On cancel (the future
+//! is dropped without being driven to completion), `Drop` calls
+//! `ChildKiller::kill()` (SIGHUP) — best-effort, the child may already be
+//! exiting. On the happy path (the future resolves), `poll` disarms the
+//! guard (`Option::take()`), so the subsequent `Drop` is a no-op. The
+//! waiter thread reaps the killed child via its blocking `wait()`, so
+//! there is no zombie. See `tty-local.md` §"Cancel-Cleanup (ADR-056)".
+
+use std::collections::HashMap;
+use std::future::Future;
+use std::io::{Read, Write};
+use std::path::PathBuf;
+use std::pin::Pin;
+use std::sync::Arc;
+use std::sync::Mutex;
+use std::task::{Context, Poll};
+use std::thread;
+
+use crate::backend::{
+    BoxFuture, TerminalParams, TtyControl, TtyControlHandle, TtyError, TtyHandle,
+};
+use crate::control::signal_from_name;
+use bytes::Bytes;
+use futures_core::Stream;
+use portable_pty::{native_pty_system, ChildKiller, CommandBuilder, MasterPty, PtySize};
+use tokio::sync::{mpsc, oneshot};
+use tokio_stream::wrappers::ReceiverStream;
+use tracing::{debug, warn};
+
+/// Channel command for the writer thread: bytes to write, or EOF.
+///
+/// `Bytes` writes the bytes to the master writer and flushes. `Eof` drops
+/// the writer (sends EOF to the slave's stdin) and exits the writer thread.
+pub enum StdinCmd {
+    /// Write these bytes to the master writer (write + flush).
+    Bytes(Vec<u8>),
+    /// Close the master writer (EOF to the slave's stdin). The writer thread
+    /// drops the writer and exits.
+    Eof,
+}
+
+/// Control handle for a live PTY: resize + signal forwarding. Cheap to
+/// clone (all fields are `Arc`-backed) so the adapter can hand a clone to
+/// the spawned control-chunk dispatcher.
+#[derive(Clone)]
+pub struct PtyControl {
+    master: Arc<Mutex<Box<dyn MasterPty + Send>>>,
+    killer: Arc<Mutex<Box<dyn ChildKiller + Send + Sync>>>,
+    pid: Option<u32>,
+}
+
+impl PtyControl {
+    /// Construct from the shared master, the cloned killer, and the child's
+    /// pid (used for process-group signal targeting).
+    pub fn new(
+        master: Arc<Mutex<Box<dyn MasterPty + Send>>>,
+        killer: Arc<Mutex<Box<dyn ChildKiller + Send + Sync>>>,
+        pid: Option<u32>,
+    ) -> Self {
+        Self {
+            master,
+            killer,
+            pid,
+        }
+    }
+}
+
+impl TtyControl for PtyControl {
+    /// Resize the PTY. Safe to call from the async pump —
+    /// `MasterPty::resize` is non-blocking (it issues an `ioctl`).
+    fn resize(&self, cols: u16, rows: u16, pixel_width: u16, pixel_height: u16) {
+        let size = PtySize {
+            cols,
+            rows,
+            pixel_width,
+            pixel_height,
+        };
+        let master = self.master.lock().expect("master mutex poisoned");
+        if let Err(e) = master.resize(size) {
+            warn!("pty resize failed: {e}");
+        }
+    }
+
+    /// Forward a signal by name to the child's process group (REQ-TTY-02).
+    ///
+    /// On Unix: maps the name to a `libc` signal number via
+    /// `crate::control::signal_from_name`, then calls `kill(-pgid, sig)` (the
+    /// negative pid reaches the whole process group — the child is a
+    /// session leader because `set_controlling_tty(true)`, the default).
+    /// Falls back to `kill(pid, sig)` if the group signal fails (e.g. the
+    /// child already exited). Unknown names fall back to
+    /// `ChildKiller::kill` (SIGHUP).
+    ///
+    /// On non-Unix: `ChildKiller::kill` (SIGHUP) directly.
+    fn signal(&self, name: &str) {
+        #[cfg(unix)]
+        {
+            if let Some(pid) = self.pid {
+                if let Some(sig) = signal_from_name(name) {
+                    let pgid = pid as i32;
+                    let r = unsafe { libc::kill(-pgid, sig) };
+                    if r == 0 {
+                        return;
+                    }
+                    let err = std::io::Error::last_os_error();
+                    let r2 = unsafe { libc::kill(pgid, sig) };
+                    if r2 == 0 {
+                        return;
+                    }
+                    warn!(
+                        "pty signal `{name}` (group {pgid}) failed: {err}; \
+                         direct kill also failed: {}",
+                        std::io::Error::last_os_error()
+                    );
+                    return;
+                }
+            }
+            // Unknown name or no pid: fall back to ChildKiller (SIGHUP).
+            let mut killer = self.killer.lock().expect("killer mutex poisoned");
+            if let Err(e) = killer.kill() {
+                warn!("pty fallback ChildKiller::kill failed: {e}");
+            }
+        }
+
+        #[cfg(not(unix))]
+        {
+            let _ = name;
+            let mut killer = self.killer.lock().expect("killer mutex poisoned");
+            if let Err(e) = killer.kill() {
+                warn!("pty ChildKiller::kill failed: {e}");
+            }
+        }
+    }
+}
+
+/// ADR-056 kill guard wrapping the waiter-thread oneshot + a
+/// `portable_pty::ChildKiller`.
+///
+/// `poll` delegates to the oneshot receiver (resolves on natural exit). On
+/// `Ready`, the killer is taken (`Option::take()`) — disarmed — so the
+/// subsequent `Drop` is a no-op. On cancel (the future is dropped before
+/// resolving), `Drop` calls `ChildKiller::kill()` (SIGHUP) — best-effort;
+/// the child may already be exiting. The waiter thread's blocking `wait()`
+/// reaps the killed child, so there is no zombie. The contract is "kill on
+/// cancel; no-op on resolve."
+pub struct LocalExitFuture {
+    rx: oneshot::Receiver<i32>,
+    killer: Option<Box<dyn ChildKiller + Send + Sync>>,
+}
+
+impl LocalExitFuture {
+    fn new(rx: oneshot::Receiver<i32>, killer: Box<dyn ChildKiller + Send + Sync>) -> Self {
+        Self {
+            rx,
+            killer: Some(killer),
+        }
+    }
+}
+
+impl Future for LocalExitFuture {
+    type Output = Result<i32, TtyError>;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        match Pin::new(&mut self.rx).poll(cx) {
+            Poll::Pending => Poll::Pending,
+            Poll::Ready(Ok(code)) => {
+                // Disarm the kill guard — the child exited naturally.
+                self.killer.take();
+                Poll::Ready(Ok(code))
+            }
+            Poll::Ready(Err(_)) => {
+                // The waiter thread's oneshot sender was dropped (wait()
+                // failed). Disarm to avoid killing an already-reaped child.
+                self.killer.take();
+                Poll::Ready(Err(TtyError::WaitFailed {
+                    message: "waiter thread exited without sending exit code".to_string(),
+                }))
+            }
+        }
+    }
+}
+
+impl Drop for LocalExitFuture {
+    fn drop(&mut self) {
+        if let Some(killer) = self.killer.take() {
+            // ADR-056: kill on cancel. Best-effort — the child may already
+            // be exiting. SIGHUP is what portable_pty sends on Unix.
+            let mut killer = killer;
+            if let Err(e) = killer.kill() {
+                debug!("LocalExitFuture drop: ChildKiller::kill failed: {e}");
+            }
+        }
+    }
+}
+
+/// `AsyncWrite` adapter over an `mpsc::Sender<StdinCmd>`. `poll_write` sends
+/// `StdinCmd::Bytes(buf.to_vec())`; `poll_flush` is a no-op (the writer
+/// thread flushes); `poll_close` sends `StdinCmd::Eof`.
+///
+/// When the channel is full, `poll_write` parks in an in-flight send
+/// future stored on the struct (so a re-poll resumes the same send rather
+/// than starting a new one — `reserve()` is not `Unpin`).
+struct StdinSink {
+    tx: mpsc::Sender<StdinCmd>,
+    /// In-flight `reserve()` + send, captured as a boxed future. `None`
+    /// when no write is pending.
+    inflight: Option<InflightSend>,
+    /// Bytes for the in-flight write (returned as the write count on
+    /// completion).
+    inflight_len: usize,
+    close_sent: bool,
+}
+
+/// Boxed future for an in-flight stdin `reserve + send`. The permit borrows
+/// the sender, so the future owns a cloned sender and the bytes to send.
+type InflightSend = Pin<Box<dyn Future<Output = Result<(), mpsc::error::SendError<()>>> + Send>>;
+
+impl StdinSink {
+    fn new(tx: mpsc::Sender<StdinCmd>) -> Self {
+        Self {
+            tx,
+            inflight: None,
+            inflight_len: 0,
+            close_sent: false,
+        }
+    }
+}
+
+impl tokio::io::AsyncWrite for StdinSink {
+    fn poll_write(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<Result<usize, std::io::Error>> {
+        // Drain any in-flight write first.
+        if let Some(fut) = self.inflight.as_mut() {
+            match fut.as_mut().poll(cx) {
+                Poll::Pending => return Poll::Pending,
+                Poll::Ready(Ok(())) => {
+                    let n = self.inflight_len;
+                    self.inflight = None;
+                    self.inflight_len = 0;
+                    return Poll::Ready(Ok(n));
+                }
+                Poll::Ready(Err(_)) => {
+                    self.inflight = None;
+                    self.inflight_len = 0;
+                    return Poll::Ready(Err(std::io::Error::new(
+                        std::io::ErrorKind::BrokenPipe,
+                        "stdin channel closed",
+                    )));
+                }
+            }
+        }
+        // Fast path: try to send without parking.
+        match self.tx.try_send(StdinCmd::Bytes(buf.to_vec())) {
+            Ok(()) => Poll::Ready(Ok(buf.len())),
+            Err(mpsc::error::TrySendError::Full(_)) => {
+                // Park a `reserve + send` future.
+                let tx = self.tx.clone();
+                let bytes = buf.to_vec();
+                let len = bytes.len();
+                self.inflight_len = len;
+                self.inflight = Some(Box::pin(async move {
+                    let permit = tx.reserve().await?;
+                    permit.send(StdinCmd::Bytes(bytes));
+                    Ok(())
+                }));
+                // Recurse via a re-poll so the parked future is polled now.
+                self.poll_write(cx, buf)
+            }
+            Err(mpsc::error::TrySendError::Closed(_)) => Poll::Ready(Err(std::io::Error::new(
+                std::io::ErrorKind::BrokenPipe,
+                "stdin channel closed",
+            ))),
+        }
+    }
+
+    fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), std::io::Error>> {
+        Poll::Ready(Ok(()))
+    }
+
+    fn poll_shutdown(
+        mut self: Pin<&mut Self>,
+        _cx: &mut Context<'_>,
+    ) -> Poll<Result<(), std::io::Error>> {
+        if self.close_sent {
+            return Poll::Ready(Ok(()));
+        }
+        match self.tx.try_send(StdinCmd::Eof) {
+            Ok(()) => {
+                self.close_sent = true;
+                Poll::Ready(Ok(()))
+            }
+            Err(mpsc::error::TrySendError::Full(_)) => Poll::Pending,
+            Err(mpsc::error::TrySendError::Closed(_)) => {
+                self.close_sent = true;
+                Poll::Ready(Ok(()))
+            }
+        }
+    }
+}
+
+/// Allocate a local PTY, spawn `cmd` into it, and return the async-facing
+/// `TtyHandle`.
+///
+/// Spawns the child as a session leader with a controlling tty
+/// (`CommandBuilder::set_controlling_tty(true)` — the default; REQ-TTY-02
+/// depends on it). Wires the three-thread bridge (reader/writer/waiter),
+/// the `PtyControl`, and the ADR-056 `LocalExitFuture`.
+pub fn allocate_pty(
+    terminal: TerminalParams,
+    cmd: Vec<String>,
+    cwd: Option<PathBuf>,
+    env: HashMap<String, String>,
+) -> Result<TtyHandle, TtyError> {
+    if cmd.is_empty() {
+        return Err(TtyError::AllocFailed {
+            message: "cmd must be non-empty".to_string(),
+        });
+    }
+
+    let pty_system = native_pty_system();
+    let size = PtySize {
+        cols: terminal.cols,
+        rows: terminal.rows,
+        pixel_width: terminal.pixel_width,
+        pixel_height: terminal.pixel_height,
+    };
+    let pair = pty_system
+        .openpty(size)
+        .map_err(|e| TtyError::AllocFailed {
+            message: format!("openpty: {e}"),
+        })?;
+
+    let mut builder = CommandBuilder::new(&cmd[0]);
+    for arg in &cmd[1..] {
+        builder.arg(arg);
+    }
+    if let Some(cwd) = cwd {
+        builder.cwd(cwd);
+    }
+    for (k, v) in env {
+        builder.env(k, v);
+    }
+    // Session leader + controlling tty — the default. REQ-TTY-02:
+    // `kill(-pgid, sig)` reaches the whole group only when the child is a
+    // session leader, which requires a controlling tty.
+    builder.set_controlling_tty(true);
+
+    // Spawn the child on the slave side, then drop the slave so that when
+    // the master writer closes, the child sees EOF on its stdin.
+    let mut child = pair
+        .slave
+        .spawn_command(builder)
+        .map_err(|e| TtyError::AllocFailed {
+            message: format!("spawn_command: {e}"),
+        })?;
+    drop(pair.slave);
+
+    let pid = child.process_id();
+    // Two killer views: one for the ADR-056 kill guard (LocalExitFuture's
+    // Drop), one for the signal path's fallback kill (PtyControl). Both
+    // reference the same underlying pid/handle via clone_killer.
+    let killer = child.clone_killer();
+    let killer_for_control = killer.clone_killer();
+
+    let master: Arc<Mutex<Box<dyn MasterPty + Send>>> = Arc::new(Mutex::new(pair.master));
+
+    // --- Reader thread: blocking reads from the master reader → mpsc ---
+    let reader_master = master.clone();
+    let (stdout_tx, stdout_rx) = mpsc::channel::<Bytes>(64);
+    thread::Builder::new()
+        .name("pty-reader".into())
+        .spawn(move || {
+            let reader = {
+                let m = reader_master.lock().expect("master mutex poisoned");
+                match m.try_clone_reader() {
+                    Ok(r) => r,
+                    Err(e) => {
+                        warn!("pty-reader: try_clone_reader failed: {e}");
+                        let _ = stdout_tx.blocking_send(Bytes::new());
+                        return;
+                    }
+                }
+            };
+            let mut reader = reader;
+            let mut buf = vec![0u8; 8192];
+            loop {
+                match reader.read(&mut buf) {
+                    Ok(0) => break,
+                    Ok(n) => {
+                        let chunk = Bytes::copy_from_slice(&buf[..n]);
+                        if stdout_tx.blocking_send(chunk).is_err() {
+                            break;
+                        }
+                    }
+                    Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue,
+                    Err(e) => {
+                        warn!("pty-reader: read error: {e}");
+                        break;
+                    }
+                }
+            }
+            // Zero-length sentinel: signals the pump the stream drained.
+            let _ = stdout_tx.blocking_send(Bytes::new());
+            debug!("pty-reader thread done");
+        })
+        .expect("spawn pty-reader");
+
+    // --- Writer thread: drain mpsc<StdinCmd> → blocking writes ---
+    let writer_master = master.clone();
+    let (stdin_tx, mut stdin_rx) = mpsc::channel::<StdinCmd>(64);
+    thread::Builder::new()
+        .name("pty-writer".into())
+        .spawn(move || {
+            let writer = {
+                let m = writer_master.lock().expect("master mutex poisoned");
+                match m.take_writer() {
+                    Ok(w) => w,
+                    Err(e) => {
+                        warn!("pty-writer: take_writer failed: {e}");
+                        return;
+                    }
+                }
+            };
+            let mut writer = writer;
+            while let Some(cmd) = stdin_rx.blocking_recv() {
+                match cmd {
+                    StdinCmd::Bytes(bytes) => {
+                        if let Err(e) = writer.write_all(&bytes) {
+                            warn!("pty-writer: write_all failed: {e}");
+                            break;
+                        }
+                        if let Err(e) = writer.flush() {
+                            warn!("pty-writer: flush failed: {e}");
+                            break;
+                        }
+                    }
+                    StdinCmd::Eof => {
+                        drop(writer);
+                        break;
+                    }
+                }
+            }
+            debug!("pty-writer thread done");
+        })
+        .expect("spawn pty-writer");
+
+    // --- Waiter thread: blocking Child::wait() → oneshot<i32> ---
+    let (exit_tx, exit_rx) = oneshot::channel::<i32>();
+    thread::Builder::new()
+        .name("pty-waiter".into())
+        .spawn(move || {
+            let status = match child.wait() {
+                Ok(s) => s,
+                Err(e) => {
+                    warn!("pty-waiter: wait failed: {e}");
+                    let _ = exit_tx.send(-1);
+                    return;
+                }
+            };
+            let code = status.exit_code() as i32;
+            debug!(exit_code = code, "pty-waiter: child reaped");
+            let _ = exit_tx.send(code);
+        })
+        .expect("spawn pty-waiter");
+
+    // --- Assemble the TtyHandle ---
+    let stdout: Pin<Box<dyn Stream<Item = Bytes> + Send>> =
+        Box::pin(ReceiverStream::new(stdout_rx));
+    let stdin: Box<dyn tokio::io::AsyncWrite + Send + Unpin> = Box::new(StdinSink::new(stdin_tx));
+    let exit_code: BoxFuture<Result<i32, TtyError>> =
+        Box::pin(LocalExitFuture::new(exit_rx, killer));
+    let control = Some(TtyControlHandle::new(Arc::new(PtyControl::new(
+        master,
+        Arc::new(Mutex::new(killer_for_control)),
+        pid,
+    ))));
+
+    Ok(TtyHandle {
+        stdin,
+        stdout,
+        stderr: None,
+        exit_code,
+        control,
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tokio::io::AsyncWriteExt;
+    use tokio_stream::StreamExt;
+
+    fn term() -> TerminalParams {
+        TerminalParams {
+            term: None,
+            cols: 80,
+            rows: 24,
+            pixel_width: 0,
+            pixel_height: 0,
+            modes: serde_json::Value::Null,
+        }
+    }
+
+    fn env_default() -> HashMap<String, String> {
+        let mut env = HashMap::new();
+        env.insert("TERM".to_string(), "dumb".to_string());
+        env
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    async fn happy_path_echo_exits_zero() {
+        let handle = allocate_pty(
+            term(),
+            vec!["echo".to_string(), "hello".to_string()],
+            None,
+            env_default(),
+        )
+        .expect("allocate");
+        let mut stdout = handle.stdout;
+        let mut collected = Vec::new();
+        while let Some(chunk) = stdout.next().await {
+            if chunk.is_empty() {
+                break;
+            }
+            collected.extend_from_slice(&chunk);
+        }
+        let code = handle.exit_code.await.expect("exit_code");
+        assert_eq!(code, 0);
+        let s = String::from_utf8_lossy(&collected);
+        assert!(s.contains("hello"), "stdout should contain hello: {s:?}");
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    async fn stdin_round_trip_cat() {
+        let handle =
+            allocate_pty(term(), vec!["cat".to_string()], None, env_default()).expect("allocate");
+        let mut stdin = handle.stdin;
+        let mut stdout = handle.stdout;
+
+        // PTY may echo input; we still expect to see our bytes somewhere
+        // in the output. Write, then close, then drain.
+        stdin.write_all(b"ping\n").await.expect("write");
+        stdin.shutdown().await.expect("shutdown (eof)");
+
+        let mut collected = Vec::new();
+        while let Some(chunk) = stdout.next().await {
+            if chunk.is_empty() {
+                break;
+            }
+            collected.extend_from_slice(&chunk);
+        }
+        let code = handle.exit_code.await.expect("exit_code");
+        let s = String::from_utf8_lossy(&collected);
+        assert!(s.contains("ping"), "stdout should contain ping: {s:?}");
+        assert_eq!(code, 0);
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    async fn resize_does_not_error() {
+        let handle = allocate_pty(
+            term(),
+            vec!["sleep".to_string(), "1".to_string()],
+            None,
+            env_default(),
+        )
+        .expect("allocate");
+        let control = handle.control.as_ref().expect("control");
+        control.resize(120, 40, 0, 0);
+        let _ = handle.exit_code.await.expect("exit_code");
+    }
+
+    #[cfg(unix)]
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    async fn signal_int_kills_child() {
+        let handle = allocate_pty(
+            term(),
+            vec!["sleep".to_string(), "60".to_string()],
+            None,
+            env_default(),
+        )
+        .expect("allocate");
+        let control = handle.control.as_ref().expect("control");
+        // Give the child a moment to actually exec sleep.
+        tokio::time::sleep(std::time::Duration::from_millis(150)).await;
+        control.signal("INT");
+        let code = tokio::time::timeout(std::time::Duration::from_secs(5), handle.exit_code)
+            .await
+            .expect("exit timed out")
+            .expect("exit_code");
+        assert_ne!(
+            code, 0,
+            "signal-terminated child should report non-zero exit: {code}"
+        );
+    }
+
+    #[cfg(unix)]
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    async fn signal_reaches_process_group_child() {
+        // bash -c "sleep 60" — sleep is a child of bash. The group signal
+        // must reach sleep too (REQ-TTY-02). bash exits when its child does.
+        let handle = allocate_pty(
+            term(),
+            vec!["bash".to_string(), "-c".to_string(), "sleep 60".to_string()],
+            None,
+            env_default(),
+        )
+        .expect("allocate");
+        let control = handle.control.as_ref().expect("control");
+        tokio::time::sleep(std::time::Duration::from_millis(200)).await;
+        control.signal("INT");
+        let code = tokio::time::timeout(std::time::Duration::from_secs(5), handle.exit_code)
+            .await
+            .expect("exit timed out")
+            .expect("exit_code");
+        assert_ne!(code, 0, "process group should have been killed: {code}");
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    async fn cancel_cleanup_kills_child_on_drop() {
+        // ADR-056: dropping the TtyHandle (and thus the exit_code future)
+        // without awaiting it MUST kill the child. The waiter thread reaps
+        // the killed child (no zombie). We assert the kill happened by
+        // spawning a second session that completes promptly — i.e., the
+        // dropped session's child does not outlive a short grace period
+        // (if it did, the SIGHUP from Drop would not have fired).
+        let handle = allocate_pty(
+            term(),
+            vec!["sleep".to_string(), "60".to_string()],
+            None,
+            env_default(),
+        )
+        .expect("allocate");
+        drop(handle);
+        // Drop fires LocalExitFuture::Drop → ChildKiller::kill (SIGHUP).
+        // The waiter thread reaps the child. Give it a moment.
+        tokio::time::sleep(std::time::Duration::from_millis(300)).await;
+
+        // Probe: a new session should be allocatable and complete cleanly.
+        let probe = allocate_pty(
+            term(),
+            vec!["echo".to_string(), "ok".to_string()],
+            None,
+            env_default(),
+        )
+        .expect("allocate");
+        let code = tokio::time::timeout(std::time::Duration::from_secs(5), probe.exit_code)
+            .await
+            .expect("probe timed out")
+            .expect("probe exit_code");
+        assert_eq!(code, 0);
+    }
+
+    #[cfg(unix)]
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    async fn unknown_signal_falls_back_to_child_killer() {
+        let handle = allocate_pty(
+            term(),
+            vec!["sleep".to_string(), "60".to_string()],
+            None,
+            env_default(),
+        )
+        .expect("allocate");
+        let control = handle.control.as_ref().expect("control");
+        // "NOSUCH" is not a known signal name → falls back to
+        // ChildKiller::kill (SIGHUP). The child should die.
+        control.signal("NOSUCH");
+        let code = tokio::time::timeout(std::time::Duration::from_secs(5), handle.exit_code)
+            .await
+            .expect("exit timed out")
+            .expect("exit_code");
+        assert_ne!(code, 0, "fallback kill should terminate the child: {code}");
+    }
+}

src/session.rs

@@ -87,7 +87,10 @@ pub enum TtySessionError {
     /// the stream is a length-prefixed JSON `{"error":"..."}` rather
     /// than a raw chunk).
     #[error("negotiation rejected: {error}")]
-    NegotiationRejected { error: String, fields: HashMap<String, String> },
+    NegotiationRejected {
+        error: String,
+        fields: HashMap<String, String>,
+    },
     /// The session ended (server closed the stream) before an `Exit`
     /// control chunk arrived. `wait()` returns this when the
     /// stdout/stderr pumps drain and no exit chunk was observed.
@@ -174,17 +177,19 @@ impl TtySession {
         params: serde_json::Value,
     ) -> Result<Self, TtySessionError> {
         let (channel_id, send, recv) = client
-            .open_channel(crate::channels::OP_TTY_OPEN, params.clone(), crate::channels::TTY_ALPN)
+            .open_channel(
+                crate::channels::OP_TTY_OPEN,
+                params.clone(),
+                crate::channels::TTY_ALPN,
+            )
             .await
             .map_err(TtySessionError::ChannelsOpen)?;
         debug!("tty: opened channel {channel_id} via channels");
 
         let remote_addr = client.manager().remote_addr();
         let source = alkcall::channels::source::channel_source(recv, send, remote_addr);
-        let channel_conn = Connection::from_source(
+        let channel_conn =
-            source,
+            Connection::from_source(source, crate::channels::TTY_ALPN.as_bytes().to_vec());
-            crate::channels::TTY_ALPN.as_bytes().to_vec(),
-        );
 
         let negotiate: NegotiateRequest =
             serde_json::from_value(params).map_err(TtySessionError::NegotiationSerialize)?;
@@ -237,12 +242,7 @@ impl TtySession {
         let (exit_tx, exit_rx) =
             tokio::sync::watch::channel::<Option<Result<i32, TtySessionError>>>(None);
 
-        let read_pump = tokio::spawn(read_pump(
+        let read_pump = tokio::spawn(read_pump(read, stdout_tx, stderr_tx, exit_tx));
-            read,
-            stdout_tx,
-            stderr_tx,
-            exit_tx,
-        ));
 
         Ok(Self {
             writer: Mutex::new(writer),
@@ -327,10 +327,9 @@ impl TtySession {
     pub async fn recv_stdout(&self) -> Pin<Box<dyn Stream<Item = Bytes> + Send>> {
         let mut guard = self.stdout_rx.lock().await;
         if let Some(rx) = guard.take() {
-            return Box::pin(futures::stream::unfold(
+            return Box::pin(futures::stream::unfold(rx, |mut rx| async move {
-                rx,
+                rx.recv().await.map(|bytes| (bytes, rx))
-                |mut rx| async move { rx.recv().await.map(|bytes| (bytes, rx)) },
+            }));
-            ));
         }
         // Already taken — return an empty stream.
         Box::pin(futures::stream::empty())
@@ -343,10 +342,9 @@ impl TtySession {
     pub async fn recv_stderr(&self) -> Option<Pin<Box<dyn Stream<Item = Bytes> + Send>>> {
         let mut guard = self.stderr_rx.lock().await;
         let rx = guard.take()?;
-        Some(Box::pin(futures::stream::unfold(
+        Some(Box::pin(futures::stream::unfold(rx, |mut rx| async move {
-            rx,
+            rx.recv().await.map(|bytes| (bytes, rx))
-            |mut rx| async move { rx.recv().await.map(|bytes| (bytes, rx)) },
+        })))
-        )))
     }
 
     /// Await the `Exit` control chunk and return the process exit
@@ -516,18 +514,11 @@ mod tests {
         let (client, server) = duplex(64 * 1024);
         let (server_read, server_write) = tokio::io::split(server);
         let server_task = tokio::spawn(async move {
-            crate::adapter::drive_session(
+            crate::adapter::drive_session(server_write, server_read, backends, None, identity)
-                server_write,
-                server_read,
-                backends,
-                None,
-                identity,
-            )
                 .await;
         });
 
-        let client_conn =
+        let client_conn = Connection::from_bidi(client, b"alk/tty".to_vec(), None);
-            Connection::from_bidi(client, b"alk/tty".to_vec(), None);
         let session = TtySession::connect_direct(client_conn, test_negotiate("mock"))
             .await
             .expect("connect_direct");
@@ -547,10 +538,7 @@ mod tests {
             resources: HashMap::new(),
         });
         let (session, _server) = wire_session_and_server(backend, identity).await;
-        let code = tokio::time::timeout(
+        let code = tokio::time::timeout(std::time::Duration::from_secs(5), session.wait())
-            std::time::Duration::from_secs(5),
-            session.wait(),
-        )
             .await
             .expect("wait didn't time out")
             .expect("wait returns exit code");
@@ -574,15 +562,8 @@ mod tests {
             .send_stdin(Bytes::from_static(b"hello"))
             .await
             .expect("send_stdin");
-        session
+        session.close_stdin().await.expect("close_stdin");
-            .close_stdin()
+        let _ = tokio::time::timeout(std::time::Duration::from_secs(5), session.wait()).await;
-            .await
-            .expect("close_stdin");
-        let _ = tokio::time::timeout(
-            std::time::Duration::from_secs(5),
-            session.wait(),
-        )
-        .await;
     }
 
     #[tokio::test]
@@ -600,11 +581,7 @@ mod tests {
         let (session, _server) = wire_session_and_server(backend, identity).await;
         session.resize(80, 24, 0, 0).await.expect("resize");
         session.signal("INT").await.expect("signal");
-        let _ = tokio::time::timeout(
+        let _ = tokio::time::timeout(std::time::Duration::from_secs(5), session.wait()).await;
-            std::time::Duration::from_secs(5),
-            session.wait(),
-        )
-        .await;
     }
 
     #[tokio::test]
@@ -650,15 +627,11 @@ mod tests {
             let _ = server.read_exact(&mut body).await;
             // Drop `server` — the client's read pump hits EOF.
         });
-        let client_conn =
+        let client_conn = Connection::from_bidi(client, b"alk/tty".to_vec(), None);
-            Connection::from_bidi(client, b"alk/tty".to_vec(), None);
         let session = TtySession::connect_direct(client_conn, test_negotiate("mock"))
             .await
             .expect("connect_direct");
-        let result = tokio::time::timeout(
+        let result = tokio::time::timeout(std::time::Duration::from_secs(5), session.wait())
-            std::time::Duration::from_secs(5),
-            session.wait(),
-        )
             .await
             .expect("wait didn't time out");
         assert!(matches!(result, Err(TtySessionError::NoExitChunk)));
@@ -672,8 +645,7 @@ mod tests {
     async fn connect_direct_errors_when_stream_is_broken() {
         let (client, server) = duplex(64);
         drop(server);
-        let client_conn =
+        let client_conn = Connection::from_bidi(client, b"alk/tty".to_vec(), None);
-            Connection::from_bidi(client, b"alk/tty".to_vec(), None);
         let result = TtySession::connect_direct(client_conn, test_negotiate("mock")).await;
         assert!(
             result.is_err(),