test(tty): end-to-end integration tests for LocalTtyBackend + drive_session

Add 18 integration tests in crates/alknet-tty-local/tests/ exercising
the full stack (LocalTtyBackend + TtyAdapter::drive_session over
tokio::io::duplex, real commands), validating the two crates work
together through the TtyBackend trait seam.

Tests live in alknet-tty-local (not alknet-tty) per the feature-gate
deviation: the cyclic dependency alknet-tty → alknet-tty-local →
alknet-tty prevents alknet-tty from depending on alknet-tty-local, so
the integration tests live in the crate that depends on both. Unix-only
tests (signal, process-group) use #[cfg(unix)].

PTY mode (8): happy path echo, interactive cat round-trip, resize,
SIGINT signal, process-group signal (REQ-TTY-02), stdin EOF sentinel,
cancel cleanup (ADR-056), exit-chunk-is-last (ADR-055).

Pipe mode (6): happy path echo, separate stderr, SIGTERM signal,
cancel cleanup, resize no-op, stdout chunk + sentinel.

Negotiation errors (5): unknown_backend, malformed_negotiation (bad
JSON, carriage != raw, empty cmd), allocate_failed (nonexistent
binary).

A shared test harness (tests/common/mod.rs) provides the client-side
wire protocol helpers: write_negotiation, write_chunk, write_control,
try_read_chunk, read_chunk_timeout, read_error_frame (asserts the
0x00 first-byte framing-disambiguation invariant), read_until_exit,
assert_no_more_chunks, close_write_half, plus spawn_session which
wires drive_session over a duplex pair with a test identity carrying
tty:open.

Verification: cargo test -p alknet-tty-local (42 tests pass), cargo
clippy -p alknet-tty-local -- -D warnings (clean), cargo fmt --check
-p alknet-tty-local (clean).

Cargo.lock

@@ -151,6 +151,7 @@ dependencies = [
 name = "alknet-tty-local"
 version = "0.1.0"
 dependencies = [
+ "alknet-core",
  "alknet-tty",
  "async-trait",
  "bytes",

crates/alknet-tty-local/Cargo.toml

@@ -25,7 +25,10 @@ thiserror = "2"
 libc = "0.2"
 
 [dev-dependencies]
+alknet-core = { path = "../alknet-core" }
+alknet-tty = { path = "../alknet-tty" }
 tokio = { version = "1", features = ["full", "test-util", "macros"] }
 tokio-stream = "0.1"
 serde_json = "1"
-tempfile = "3"
+tempfile = "3"
+bytes = "1"

crates/alknet-tty-local/tests/common/mod.rs

@@ -0,0 +1,291 @@
+//! Test harness: client-side wire protocol helpers + `drive_session`
+//! stand-in over a `tokio::io::duplex` pair.
+//!
+//! Mirrors the `MockBackend` test harness in
+//! `crates/alknet-tty/src/adapter.rs`, but drives a real
+//! `LocalTtyBackend` instead of a mock. The test acts as the client:
+//! writes the negotiation frame, sends stdin/control chunks, reads
+//! stdout/stderr/control chunks, asserts the exit chunk and stream
+//! close (ADR-052, ADR-055, ADR-056).
+//!
+//! `allow(dead_code)` is needed because each test binary uses a
+//! different subset of the helpers; clippy would otherwise flag the
+//! unused ones (the helpers are part of the shared client-side wire
+//! protocol toolkit the task requires).
+
+#![allow(dead_code)]
+
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use alknet_core::auth::Identity;
+use alknet_tty::adapter::drive_session;
+use alknet_tty::backend::TtyBackend;
+use alknet_tty::wire::{ChunkReader, STREAM_CONTROL, STREAM_STDERR, STREAM_STDOUT};
+use bytes::Bytes;
+use tokio::io::duplex;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+/// The scope the test identity carries (matches `TtyAdapter::TTY_OPEN_SCOPE`).
+pub const TTY_OPEN_SCOPE: &str = "tty:open";
+
+/// A client-side negotiator: writes the JSON negotiation frame and
+/// provides helpers for sending stdin chunks and control messages,
+/// and for reading chunks/error frames the server writes back.
+pub struct ClientSide {
+    pub write: tokio::io::WriteHalf<tokio::io::DuplexStream>,
+    pub read: tokio::io::ReadHalf<tokio::io::DuplexStream>,
+}
+
+impl ClientSide {
+    /// Write a length-prefixed JSON negotiation frame (the Phase 1
+    /// carriage). The `body` should already be a complete JSON
+    /// `NegotiateRequest` object.
+    pub async fn write_negotiation(&mut self, body: &str) {
+        let len = body.len() as u32;
+        self.write.write_all(&len.to_be_bytes()).await.unwrap();
+        self.write.write_all(body.as_bytes()).await.unwrap();
+        self.write.flush().await.unwrap();
+    }
+
+    /// Write a length-prefixed raw bytes negotiation frame (for
+    /// malformed-JSON tests).
+    pub async fn write_negotiation_bytes(&mut self, body: &[u8]) {
+        let len = body.len() as u32;
+        self.write.write_all(&len.to_be_bytes()).await.unwrap();
+        self.write.write_all(body).await.unwrap();
+        self.write.flush().await.unwrap();
+    }
+
+    /// Write a raw chunk: `[stream_type: u8][len: u32 be][payload]`.
+    pub async fn write_chunk(&mut self, stream_type: u8, payload: &[u8]) {
+        let mut header = [0u8; 5];
+        header[0] = stream_type;
+        let len = payload.len() as u32;
+        header[1..].copy_from_slice(&len.to_be_bytes());
+        self.write.write_all(&header).await.unwrap();
+        if !payload.is_empty() {
+            self.write.write_all(payload).await.unwrap();
+        }
+        self.write.flush().await.unwrap();
+    }
+
+    /// Write a control chunk (stream_type 3) carrying a serialized
+    /// `ControlMessage` JSON payload.
+    pub async fn write_control(&mut self, json: &[u8]) {
+        self.write_chunk(STREAM_CONTROL, json).await;
+    }
+
+    /// Read one raw chunk from the server. Returns the `stream_type`
+    /// and the payload bytes, or `None` when the server closed the
+    /// stream cleanly (EOF).
+    pub async fn try_read_chunk(&mut self) -> Option<(u8, Bytes)> {
+        let mut reader = ChunkReader::new(&mut self.read);
+        match reader.read_chunk().await {
+            Ok(chunk) => Some((chunk.stream_type, chunk.bytes)),
+            Err(_) => None,
+        }
+    }
+
+    /// Read one chunk, panicking on read errors (used when the test
+    /// expects a chunk, not a close).
+    pub async fn read_chunk(&mut self) -> (u8, Bytes) {
+        self.try_read_chunk()
+            .await
+            .expect("expected a chunk, got stream close")
+    }
+
+    /// Read one chunk with a timeout. Returns `None` on timeout or
+    /// stream close.
+    pub async fn read_chunk_timeout(
+        &mut self,
+        timeout: std::time::Duration,
+    ) -> Option<(u8, Bytes)> {
+        tokio::time::timeout(timeout, self.try_read_chunk())
+            .await
+            .unwrap_or_default()
+    }
+
+    /// Read a length-prefixed JSON error frame (Phase 1 framing, used
+    /// for negotiation errors). The first byte of the 4-byte length
+    /// prefix MUST be `0x00` (the framing-disambiguation invariant —
+    /// ADR-052 §5); this method asserts that. Returns the parsed JSON.
+    pub async fn read_error_frame(&mut self) -> serde_json::Value {
+        let mut first = [0u8; 1];
+        self.read.read_exact(&mut first).await.unwrap();
+        assert_eq!(
+            first[0], 0x00,
+            "error frame length prefix high byte must be 0x00 (ADR-052 §5)"
+        );
+        let mut len_rest = [0u8; 3];
+        self.read.read_exact(&mut len_rest).await.unwrap();
+        let len = u32::from_be_bytes([first[0], len_rest[0], len_rest[1], len_rest[2]]) as usize;
+        let mut body = vec![0u8; len];
+        self.read.read_exact(&mut body).await.unwrap();
+        serde_json::from_slice(&body).unwrap()
+    }
+
+    /// Read chunks until the exit control chunk is observed, then
+    /// return the accumulated stdout bytes, stderr bytes, and the
+    /// exit code. Returns `None` if the stream closes before an exit
+    /// chunk arrives.
+    pub async fn read_until_exit(&mut self) -> Option<(Vec<u8>, Vec<u8>, i32)> {
+        let mut stdout = Vec::new();
+        let mut stderr = Vec::new();
+        loop {
+            let (st, bytes) = self.try_read_chunk().await?;
+            match st {
+                STREAM_STDOUT => {
+                    if !bytes.is_empty() {
+                        stdout.extend_from_slice(&bytes);
+                    }
+                }
+                STREAM_STDERR => {
+                    if !bytes.is_empty() {
+                        stderr.extend_from_slice(&bytes);
+                    }
+                }
+                STREAM_CONTROL => {
+                    let v: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
+                    if v["type"] == "exit" {
+                        return Some((stdout, stderr, v["code"].as_i64().unwrap() as i32));
+                    }
+                }
+                other => panic!("unexpected stream_type {other} from server"),
+            }
+        }
+    }
+
+    /// Same as [`read_until_exit`](Self::read_until_exit) but with a
+    /// timeout on each individual chunk read.
+    pub async fn read_until_exit_timeout(
+        &mut self,
+        per_chunk_timeout: std::time::Duration,
+    ) -> Option<(Vec<u8>, Vec<u8>, i32)> {
+        let mut stdout = Vec::new();
+        let mut stderr = Vec::new();
+        loop {
+            let (st, bytes) = self.read_chunk_timeout(per_chunk_timeout).await?;
+            match st {
+                STREAM_STDOUT => {
+                    if !bytes.is_empty() {
+                        stdout.extend_from_slice(&bytes);
+                    }
+                }
+                STREAM_STDERR => {
+                    if !bytes.is_empty() {
+                        stderr.extend_from_slice(&bytes);
+                    }
+                }
+                STREAM_CONTROL => {
+                    let v: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
+                    if v["type"] == "exit" {
+                        return Some((stdout, stderr, v["code"].as_i64().unwrap() as i32));
+                    }
+                }
+                other => panic!("unexpected stream_type {other} from server"),
+            }
+        }
+    }
+
+    /// Read chunks until the exit chunk is observed, then assert
+    /// the stream closes cleanly and NO further chunks arrive
+    /// (ADR-055 exit-chunk-is-last invariant). Returns the stdout,
+    /// stderr, and exit code.
+    pub async fn read_until_exit_and_close(&mut self) -> (Vec<u8>, Vec<u8>, i32) {
+        let (stdout, stderr, code) = self
+            .read_until_exit()
+            .await
+            .expect("expected exit chunk before stream close");
+        self.assert_no_more_chunks().await;
+        (stdout, stderr, code)
+    }
+
+    /// After the exit chunk, assert the server closes the stream
+    /// without sending any further chunks. Panics if a chunk arrives.
+    pub async fn assert_no_more_chunks(&mut self) {
+        match self
+            .read_chunk_timeout(std::time::Duration::from_millis(500))
+            .await
+        {
+            None => {}
+            Some((st, bytes)) => panic!(
+                "chunk arrived after exit chunk: stream_type={st}, bytes={bytes:?} (ADR-055)"
+            ),
+        }
+    }
+
+    /// Close the client write half cleanly. The server's read half sees
+    /// a clean EOF on its `ChunkReader`, which the adapter treats as a
+    /// client-cancel of the input direction (it signals EOF to the
+    /// backend's stdin and stops the input pump). For pipe-mode
+    /// backends, this is the path that actually closes the child's
+    /// stdin pipe (tokio's `ChildStdin::poll_shutdown` is a no-op on
+    /// Unix; dropping the `ChildStdin` is what closes the pipe, and
+    /// the input pump drops it on ConnectionClosed).
+    pub async fn close_write_half(&mut self) {
+        let _ = self.write.shutdown().await;
+    }
+}
+
+/// Build a `ClientSide` + spawned `drive_session` task over a
+/// `tokio::io::duplex` pair. The single `backend` is registered under
+/// the given key. The test identity carries `tty:open`, so the
+/// adapter's scope-gate passes.
+pub fn spawn_session(
+    backend_key: &str,
+    backend: Arc<dyn TtyBackend>,
+) -> (ClientSide, tokio::task::JoinHandle<()>) {
+    let mut backends: HashMap<String, Arc<dyn TtyBackend>> = HashMap::new();
+    backends.insert(backend_key.to_string(), backend);
+    spawn_session_with_backends(backends)
+}
+
+/// Like [`spawn_session`] but accepts the full backend map (for
+/// `unknown_backend` tests).
+pub fn spawn_session_with_backends(
+    backends: HashMap<String, Arc<dyn TtyBackend>>,
+) -> (ClientSide, tokio::task::JoinHandle<()>) {
+    let (client, server) = duplex(8 * 1024);
+    let (client_read, client_write) = tokio::io::split(client);
+    let client_side = ClientSide {
+        write: client_write,
+        read: client_read,
+    };
+    let backends = Arc::new(backends);
+    let identity = Some(test_identity());
+    let handle = tokio::spawn(async move {
+        let (server_read, server_write) = tokio::io::split(server);
+        drive_session(server_write, server_read, backends, None, identity).await;
+    });
+    (client_side, handle)
+}
+
+/// A test identity carrying the `tty:open` scope, so the adapter's
+/// scope-gate at negotiation passes.
+pub fn test_identity() -> Identity {
+    Identity {
+        id: "test-user".to_string(),
+        scopes: vec![TTY_OPEN_SCOPE.to_string()],
+        resources: HashMap::new(),
+    }
+}
+
+/// Build a negotiation frame JSON string for a PTY-mode session.
+pub fn negotiate_pty_json(backend: &str, cmd: &[&str]) -> String {
+    let cmd_json: Vec<String> = cmd.iter().map(|c| format!("\"{}\"", c)).collect();
+    format!(
+        r#"{{"carriage":"raw","backend":"{backend}","tty":{{"cols":80,"rows":24,"pixel_width":0,"pixel_height":0}},"cmd":[{cmd}]}}"#,
+        cmd = cmd_json.join(",")
+    )
+}
+
+/// Build a negotiation frame JSON string for a pipe-mode session
+/// (`tty: null`).
+pub fn negotiate_pipe_json(backend: &str, cmd: &[&str]) -> String {
+    let cmd_json: Vec<String> = cmd.iter().map(|c| format!("\"{}\"", c)).collect();
+    format!(
+        r#"{{"carriage":"raw","backend":"{backend}","tty":null,"cmd":[{cmd}]}}"#,
+        cmd = cmd_json.join(",")
+    )
+}

crates/alknet-tty-local/tests/negotiation.rs

@@ -0,0 +1,114 @@
+//! Negotiation error tests — scenarios 14–18 of
+//! `tasks/tty/integration-test.md`.
+//!
+//!   14. unknown_backend: error response, first byte `0x00`
+//!   15. malformed_negotiation (bad JSON): error response
+//!   16. malformed_negotiation (carriage != raw): error response
+//!   17. malformed_negotiation (empty cmd): error response
+//!   18. allocate_failed (nonexistent binary): error response
+//!
+//! The test identity carries the `tty:open` scope, so the
+//! `unknown_backend` test reaches the backend lookup; the scope-gate
+//! is exercised in the adapter's unit tests. These tests assert the
+//! server sends a length-prefixed JSON error frame (Phase 1 framing,
+//! ADR-052) and closes the stream without entering raw mode.
+
+mod common;
+
+use std::sync::Arc;
+
+use alknet_tty_local::LocalTtyBackend;
+use common::{negotiate_pipe_json, spawn_session};
+
+/// 14. unknown_backend: negotiate `{backend:"kubernetes",...}`, assert
+/// the error response `{"error":"unknown_backend","backend":"kubernetes"}`,
+/// the first byte of the error frame is `0x00`, and the stream closes.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn unknown_backend_error() {
+    let backend = Arc::new(LocalTtyBackend::new());
+    let (mut client, server) = spawn_session("local", backend);
+    client
+        .write_negotiation(
+            r#"{"carriage":"raw","backend":"kubernetes","tty":null,"cmd":["echo","hi"]}"#,
+        )
+        .await;
+
+    let err = client.read_error_frame().await;
+    assert_eq!(err["error"], "unknown_backend");
+    assert_eq!(err["backend"], "kubernetes");
+
+    client.assert_no_more_chunks().await;
+    let _ = server.await;
+}
+
+/// 15. malformed_negotiation (bad JSON): write garbage bytes as the
+/// negotiation frame, assert `malformed_negotiation` error response.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn malformed_negotiation_bad_json() {
+    let backend = Arc::new(LocalTtyBackend::new());
+    let (mut client, server) = spawn_session("local", backend);
+    client.write_negotiation("not valid json").await;
+
+    let err = client.read_error_frame().await;
+    assert_eq!(err["error"], "malformed_negotiation");
+
+    client.assert_no_more_chunks().await;
+    let _ = server.await;
+}
+
+/// 16. malformed_negotiation (carriage != raw): negotiate
+/// `{carriage:"json",...}`, assert `malformed_negotiation`.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn malformed_negotiation_carriage_not_raw() {
+    let backend = Arc::new(LocalTtyBackend::new());
+    let (mut client, server) = spawn_session("local", backend);
+    client
+        .write_negotiation(
+            r#"{"carriage":"json","backend":"local","tty":null,"cmd":["echo","hi"]}"#,
+        )
+        .await;
+
+    let err = client.read_error_frame().await;
+    assert_eq!(err["error"], "malformed_negotiation");
+
+    client.assert_no_more_chunks().await;
+    let _ = server.await;
+}
+
+/// 17. malformed_negotiation (empty cmd): negotiate `{cmd:[]}`, assert
+/// `malformed_negotiation`.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn malformed_negotiation_empty_cmd() {
+    let backend = Arc::new(LocalTtyBackend::new());
+    let (mut client, server) = spawn_session("local", backend);
+    client
+        .write_negotiation(r#"{"carriage":"raw","backend":"local","tty":null,"cmd":[]}"#)
+        .await;
+
+    let err = client.read_error_frame().await;
+    assert_eq!(err["error"], "malformed_negotiation");
+
+    client.assert_no_more_chunks().await;
+    let _ = server.await;
+}
+
+/// 18. allocate_failed (nonexistent binary): negotiate with a
+/// nonexistent command path, assert `allocate_failed` (the spawn
+/// fails). The adapter sends the error response in negotiation
+/// framing and closes.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn allocate_failed_nonexistent_binary() {
+    let backend = Arc::new(LocalTtyBackend::new());
+    let (mut client, server) = spawn_session("local", backend);
+    client
+        .write_negotiation(
+            negotiate_pipe_json("local", &["/nonexistent/binary/that/does/not/exist"]).as_str(),
+        )
+        .await;
+
+    let err = client.read_error_frame().await;
+    assert_eq!(err["error"], "allocate_failed");
+
+    client.assert_no_more_chunks().await;
+    let _ = server.await;
+}

crates/alknet-tty-local/tests/pipe.rs

@@ -0,0 +1,222 @@
+//! End-to-end integration tests for pipe mode (`terminal: None`) —
+//! `LocalTtyBackend` + `TtyAdapter::drive_session` over a
+//! `tokio::io::duplex` transport stand-in, running real commands.
+//!
+//! Covers scenarios 9–13 of `tasks/tty/integration-test.md`:
+//!   9.  Happy path (echo): stdout "hello", stderr empty, exit 0
+//!   10. Separate stderr: stdout "out", stderr "err", exit 0
+//!   11. Signal (SIGTERM, Unix): exit signal-terminated
+//!   12. Cancel cleanup: drop duplex → child killed
+//!   13. Resize no-op: control chunk accepted, no error
+
+mod common;
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use alknet_tty::wire::STREAM_STDOUT;
+use alknet_tty_local::LocalTtyBackend;
+use common::{negotiate_pipe_json, spawn_session};
+
+/// 9. Happy path (echo): negotiate `{backend:"local", tty:null,
+/// cmd:["echo","hello"]}`, read stdout chunks, assert "hello", exit 0.
+/// Assert stderr is empty.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn pipe_happy_path_echo() {
+    let backend = Arc::new(LocalTtyBackend::new());
+    let (mut client, server) = spawn_session("local", backend);
+    client
+        .write_negotiation(negotiate_pipe_json("local", &["echo", "hello"]).as_str())
+        .await;
+
+    let (stdout, stderr, code) = client
+        .read_until_exit()
+        .await
+        .expect("expected exit chunk before stream close");
+    let out = String::from_utf8_lossy(&stdout);
+    assert!(
+        out.contains("hello"),
+        "stdout should contain 'hello'; got: {out:?}"
+    );
+    assert!(stderr.is_empty(), "stderr should be empty; got: {stderr:?}");
+    assert_eq!(code, 0, "echo should exit 0");
+    client.assert_no_more_chunks().await;
+    let _ = server.await;
+}
+
+/// 10. Separate stderr: negotiate `cmd:["sh","-c","echo out; echo err >&2"]`,
+/// assert stdout stream receives "out", stderr stream receives "err"
+/// (as stderr chunks, stream_type 2), exit 0.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn pipe_separate_stderr() {
+    let backend = Arc::new(LocalTtyBackend::new());
+    let (mut client, server) = spawn_session("local", backend);
+    client
+        .write_negotiation(
+            negotiate_pipe_json("local", &["sh", "-c", "echo out; echo err >&2"]).as_str(),
+        )
+        .await;
+
+    let (stdout, stderr, code) = client
+        .read_until_exit()
+        .await
+        .expect("expected exit chunk before stream close");
+    let out = String::from_utf8_lossy(&stdout);
+    let err = String::from_utf8_lossy(&stderr);
+    assert!(
+        out.contains("out"),
+        "stdout should contain 'out'; got: {out:?}"
+    );
+    assert!(
+        err.contains("err"),
+        "stderr should contain 'err'; got: {err:?}"
+    );
+    assert_eq!(code, 0, "sh should exit 0");
+    let _ = server.await;
+}
+
+/// 11. Signal (SIGTERM, Unix): negotiate `cmd:["sleep","60"]`, send
+/// `signal:"TERM"`, await exit, assert signal-terminated.
+#[cfg(unix)]
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn pipe_signal_sigterm_kills_child() {
+    let backend = Arc::new(LocalTtyBackend::new());
+    let (mut client, server) = spawn_session("local", backend);
+    client
+        .write_negotiation(negotiate_pipe_json("local", &["sleep", "60"]).as_str())
+        .await;
+
+    tokio::time::sleep(Duration::from_millis(150)).await;
+
+    client
+        .write_control(br#"{"type":"signal","name":"TERM"}"#)
+        .await;
+
+    let (_out, _err, code) = client
+        .read_until_exit_timeout(Duration::from_secs(5))
+        .await
+        .expect("expected exit chunk after SIGTERM");
+    assert_ne!(
+        code, 0,
+        "child killed by SIGTERM should report non-zero exit; got {code}"
+    );
+    let _ = server.await;
+}
+
+/// 12. Cancel cleanup (ADR-056): negotiate `sleep 60`, drop the duplex
+/// mid-session, assert the child is killed (no orphan). The child
+/// writes its pid to a temp file so we can probe it after the drop.
+#[cfg(unix)]
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn pipe_cancel_cleanup_kills_child_no_orphan() {
+    let pid_file = std::env::temp_dir().join(format!(
+        "alknet_pipe_cancel_pid_{}_{}.txt",
+        std::process::id(),
+        nanos_seed()
+    ));
+    let cmd = format!("echo $$ > '{}'; exec sleep 60", pid_file.display());
+    let backend = Arc::new(LocalTtyBackend::new());
+    let (mut client, server) = spawn_session("local", backend);
+    client
+        .write_negotiation(negotiate_pipe_json("local", &["sh", "-c", cmd.as_str()]).as_str())
+        .await;
+
+    for _ in 0..200 {
+        if pid_file.exists() {
+            break;
+        }
+        tokio::time::sleep(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);
+
+    tokio::time::sleep(Duration::from_millis(150)).await;
+
+    drop(client);
+    server.abort();
+    let _ = server.await;
+
+    let mut alive = true;
+    for _ in 0..100 {
+        let r = unsafe { libc::kill(pid, 0) };
+        if r != 0 && std::io::Error::last_os_error().raw_os_error() == Some(libc::ESRCH) {
+            alive = false;
+            break;
+        }
+        tokio::time::sleep(Duration::from_millis(20)).await;
+    }
+    assert!(!alive, "child (pid={pid}) should be killed after cancel");
+}
+
+/// 13. Resize no-op: negotiate `cmd:["cat"]`, send a `resize` control
+/// chunk, assert no error (PipeControl::resize is a no-op). Close the
+/// write half to signal end-of-input (the adapter's input pump drops
+/// the `ChildStdin` on `ConnectionClosed`, which closes the pipe —
+/// tokio's `ChildStdin::poll_shutdown` is a no-op on Unix). Await exit.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn pipe_resize_noop() {
+    let backend = Arc::new(LocalTtyBackend::new());
+    let (mut client, server) = spawn_session("local", backend);
+    client
+        .write_negotiation(negotiate_pipe_json("local", &["cat"]).as_str())
+        .await;
+
+    tokio::time::sleep(Duration::from_millis(150)).await;
+
+    client
+        .write_control(br#"{"type":"resize","cols":120,"rows":40}"#)
+        .await;
+
+    client.write_control(br#"{"type":"eof"}"#).await;
+    client.close_write_half().await;
+
+    let (_out, _err, code) = client
+        .read_until_exit_timeout(Duration::from_secs(5))
+        .await
+        .expect("expected exit chunk after resize + eof");
+    assert_eq!(
+        code, 0,
+        "cat should exit 0 after no-op resize + write-half close"
+    );
+
+    let _ = server.await;
+}
+
+/// Sanity: an `echo` in pipe mode should produce at least one
+/// stdout chunk with non-empty bytes (the adapter emits a
+/// zero-length stdout sentinel after the backend stream ends).
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn pipe_echo_emits_stdout_chunk_then_sentinel() {
+    let backend = Arc::new(LocalTtyBackend::new());
+    let (mut client, server) = spawn_session("local", backend);
+    client
+        .write_negotiation(negotiate_pipe_json("local", &["echo", "hi"]).as_str())
+        .await;
+
+    let mut saw_nonempty_stdout = false;
+    while let Some((st, bytes)) = client.read_chunk_timeout(Duration::from_secs(5)).await {
+        if st == STREAM_STDOUT && !bytes.is_empty() {
+            saw_nonempty_stdout = true;
+        }
+        if st == alknet_tty::wire::STREAM_CONTROL {
+            let v: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
+            if v["type"] == "exit" {
+                break;
+            }
+        }
+    }
+    assert!(
+        saw_nonempty_stdout,
+        "expected at least one non-empty stdout chunk"
+    );
+    let _ = server.await;
+}
+
+fn nanos_seed() -> u64 {
+    use std::time::{SystemTime, UNIX_EPOCH};
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map(|d| d.as_nanos() as u64)
+        .unwrap_or(0)
+}

crates/alknet-tty-local/tests/pty.rs

@@ -0,0 +1,266 @@
+//! End-to-end integration tests for PTY mode (`terminal: Some`) —
+//! `LocalTtyBackend` + `TtyAdapter::drive_session` over a
+//! `tokio::io::duplex` transport stand-in, running real commands.
+//!
+//! Covers scenarios 1–8 of `tasks/tty/integration-test.md`:
+//!   1. Happy path (echo): stdout contains "hello", exit 0, exit chunk last
+//!   2. Interactive (cat): stdin round-trips, eof → exit 0
+//!   3. Resize: control chunk accepted, no error
+//!   4. Signal (SIGINT, Unix): exit code signal-terminated, child reaped
+//!   5. Process-group signal (Unix): bash -c "sleep 60", INT reaches sleep child
+//!   6. Stdin EOF (zero-length chunk): backend stdin closes, exit chunk sent
+//!   7. Cancel cleanup (ADR-056): drop duplex → child killed, no orphan
+//!   8. Exit-chunk-is-last: no stdout chunk after exit chunk
+
+mod common;
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use alknet_tty::wire::{STREAM_CONTROL, STREAM_STDIN};
+use alknet_tty_local::LocalTtyBackend;
+use common::{negotiate_pty_json, spawn_session};
+
+const PTY_NEG_ECHO: &str = r#"{"carriage":"raw","backend":"local","tty":{"cols":80,"rows":24,"pixel_width":0,"pixel_height":0},"cmd":["echo","hello"]}"#;
+
+/// 1. Happy path (echo): stdout contains "hello", exit 0, exit chunk is
+/// the last chunk before stream close (ADR-055).
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn pty_happy_path_echo() {
+    let backend = Arc::new(LocalTtyBackend::new());
+    let (mut client, server) = spawn_session("local", backend);
+    client.write_negotiation(PTY_NEG_ECHO).await;
+
+    let (stdout, _stderr, code) = client
+        .read_until_exit()
+        .await
+        .expect("expected exit chunk before stream close");
+    let s = String::from_utf8_lossy(&stdout);
+    assert!(
+        s.contains("hello"),
+        "stdout should contain 'hello'; got: {s:?}"
+    );
+    assert_eq!(code, 0, "echo should exit 0");
+    client.assert_no_more_chunks().await;
+    let _ = server.await;
+}
+
+/// 2. Interactive (cat): write stdin chunks, then send `eof` and
+/// drain stdout. The adapter's drainer writes chunks to the client
+/// after the exit chunk resolves, so the test sends all input first
+/// (closing cat's stdin with `eof`), then reads the echoed stdout
+/// and the exit chunk together. Asserts the echoed "ping" appears in
+/// stdout and cat exits 0.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn pty_interactive_cat_round_trip() {
+    let backend = Arc::new(LocalTtyBackend::new());
+    let (mut client, server) = spawn_session("local", backend);
+    client
+        .write_negotiation(negotiate_pty_json("local", &["cat"]).as_str())
+        .await;
+
+    tokio::time::sleep(Duration::from_millis(200)).await;
+
+    client.write_chunk(STREAM_STDIN, b"ping\n").await;
+    client.write_control(br#"{"type":"eof"}"#).await;
+
+    let (stdout, _stderr, code) = client
+        .read_until_exit()
+        .await
+        .expect("expected exit chunk before stream close");
+    let s = String::from_utf8_lossy(&stdout);
+    assert!(
+        s.contains("ping"),
+        "stdin did not round-trip via the PTY echo; got: {s:?}"
+    );
+    assert_eq!(code, 0, "cat should exit 0 on eof");
+    let _ = server.await;
+}
+
+/// 3. Resize: send a `resize` control chunk mid-session; assert no
+/// error (the PTY resizes). Send `eof`, await exit.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn pty_resize_no_error() {
+    let backend = Arc::new(LocalTtyBackend::new());
+    let (mut client, server) = spawn_session("local", backend);
+    client
+        .write_negotiation(negotiate_pty_json("local", &["cat"]).as_str())
+        .await;
+
+    tokio::time::sleep(Duration::from_millis(150)).await;
+
+    client
+        .write_control(br#"{"type":"resize","cols":120,"rows":40}"#)
+        .await;
+
+    client.write_control(br#"{"type":"eof"}"#).await;
+
+    let (_out, _err, code) = client
+        .read_until_exit_timeout(Duration::from_secs(5))
+        .await
+        .expect("expected exit chunk");
+    assert_eq!(code, 0, "cat should exit 0 after resize + eof");
+    let _ = server.await;
+}
+
+/// 4. Signal (SIGINT, Unix): negotiate `sleep 60`, send `signal:"INT"`,
+/// await the exit chunk. Assert exit code is signal-terminated (non-zero,
+/// negative on Unix). Assert the child is reaped (no zombie).
+#[cfg(unix)]
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn pty_signal_sigint_kills_child() {
+    let backend = Arc::new(LocalTtyBackend::new());
+    let (mut client, server) = spawn_session("local", backend);
+    client
+        .write_negotiation(negotiate_pty_json("local", &["sleep", "60"]).as_str())
+        .await;
+
+    tokio::time::sleep(Duration::from_millis(200)).await;
+
+    client
+        .write_control(br#"{"type":"signal","name":"INT"}"#)
+        .await;
+
+    let (_out, _err, code) = client
+        .read_until_exit_timeout(Duration::from_secs(5))
+        .await
+        .expect("expected exit chunk after signal");
+    assert_ne!(
+        code, 0,
+        "child killed by SIGINT should report non-zero exit; got {code}"
+    );
+    let _ = server.await;
+}
+
+/// 5. Process-group signal (Unix): negotiate `bash -c "sleep 60"`,
+/// send `signal:"INT"`, assert the `sleep` child also receives the
+/// signal (the process group is targeted — REQ-TTY-02).
+#[cfg(unix)]
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn pty_signal_reaches_process_group_child() {
+    let backend = Arc::new(LocalTtyBackend::new());
+    let (mut client, server) = spawn_session("local", backend);
+    client
+        .write_negotiation(negotiate_pty_json("local", &["bash", "-c", "sleep 60"]).as_str())
+        .await;
+
+    tokio::time::sleep(Duration::from_millis(250)).await;
+
+    client
+        .write_control(br#"{"type":"signal","name":"INT"}"#)
+        .await;
+
+    let (_out, _err, code) = client
+        .read_until_exit_timeout(Duration::from_secs(5))
+        .await
+        .expect("expected exit chunk after group signal");
+    assert_ne!(
+        code, 0,
+        "process group should have been killed (REQ-TTY-02); got {code}"
+    );
+    let _ = server.await;
+}
+
+/// 6. Stdin EOF (zero-length chunk): negotiate `cat`, send a
+/// zero-length stdin chunk (the sentinel), assert the backend's stdin
+/// closes, stdout drains, exit chunk is sent.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn pty_stdin_eof_zero_length_chunk() {
+    let backend = Arc::new(LocalTtyBackend::new());
+    let (mut client, server) = spawn_session("local", backend);
+    client
+        .write_negotiation(negotiate_pty_json("local", &["cat"]).as_str())
+        .await;
+
+    tokio::time::sleep(Duration::from_millis(150)).await;
+
+    client.write_chunk(STREAM_STDIN, b"").await;
+
+    let (_out, _err, code) = client
+        .read_until_exit_timeout(Duration::from_secs(5))
+        .await
+        .expect("expected exit chunk after zero-length stdin sentinel");
+    assert_eq!(code, 0, "cat should exit 0 on stdin EOF");
+    let _ = server.await;
+}
+
+/// 7. Cancel cleanup (ADR-056): negotiate `sleep 60`, drop the duplex
+/// (simulating connection drop) mid-session, assert the child is
+/// killed (no orphan). The child writes its pid to a temp file so we
+/// can probe it after the drop.
+#[cfg(unix)]
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn pty_cancel_cleanup_kills_child_no_orphan() {
+    let pid_file = std::env::temp_dir().join(format!(
+        "alknet_pty_cancel_pid_{}_{}.txt",
+        std::process::id(),
+        nanos_seed()
+    ));
+    let cmd = format!("echo $$ > '{}'; exec sleep 60", pid_file.display());
+    let backend = Arc::new(LocalTtyBackend::new());
+    let (mut client, server) = spawn_session("local", backend);
+    client
+        .write_negotiation(negotiate_pty_json("local", &["bash", "-c", cmd.as_str()]).as_str())
+        .await;
+
+    for _ in 0..200 {
+        if pid_file.exists() {
+            break;
+        }
+        tokio::time::sleep(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);
+
+    tokio::time::sleep(Duration::from_millis(150)).await;
+
+    drop(client);
+    server.abort();
+    let _ = server.await;
+
+    let mut alive = true;
+    for _ in 0..100 {
+        let r = unsafe { libc::kill(pid, 0) };
+        if r != 0 && std::io::Error::last_os_error().raw_os_error() == Some(libc::ESRCH) {
+            alive = false;
+            break;
+        }
+        tokio::time::sleep(Duration::from_millis(20)).await;
+    }
+    assert!(!alive, "child (pid={pid}) should be killed after cancel");
+}
+
+/// 8. Exit-chunk-is-last: in the happy path, assert no stdout chunk
+/// arrives after the exit chunk. Read all chunks, find the exit chunk,
+/// assert it is the last chunk before stream close.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn pty_exit_chunk_is_last() {
+    let backend = Arc::new(LocalTtyBackend::new());
+    let (mut client, server) = spawn_session("local", backend);
+    client.write_negotiation(PTY_NEG_ECHO).await;
+
+    let mut saw_exit = false;
+    while let Some((st, bytes)) = client.read_chunk_timeout(Duration::from_secs(5)).await {
+        if saw_exit {
+            panic!("chunk arrived after exit: stream_type={st}, bytes={bytes:?} (ADR-055)");
+        }
+        if st == STREAM_CONTROL {
+            let v: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
+            if v["type"] == "exit" {
+                assert_eq!(v["code"], 0);
+                saw_exit = true;
+            }
+        }
+    }
+    assert!(saw_exit, "did not see the exit chunk");
+    let _ = server.await;
+}
+
+fn nanos_seed() -> u64 {
+    use std::time::{SystemTime, UNIX_EPOCH};
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map(|d| d.as_nanos() as u64)
+        .unwrap_or(0)
+}