phase 1: port core types from alknet-tty

Port wire.rs, control.rs, negotiation.rs, backend.rs, adapter.rs from
alknet-tty (the protocol half of the alknet mono-repo split) into the
single alktty crate. All 65 unit tests pass; cargo check on
wasm32-unknown-unknown is clean (no features); clippy is clean.

Migration changes:
- adapter.rs imports: alknet_core::{auth, ownership, types} →
  alkcall::core::{auth, ownership, types + Connection/HandlerError/
  ProtocolHandler/StreamError re-exported at the crate root}
- adapter.rs TtyAdapter::alpn(): b"alknet/tty" → b"alk/tty"
- backend.rs BoxFuture type alias: Pin<Box<dyn Future + Send +
  'static>> → futures::future::BoxFuture<'static, T> (matches alkcall
  convention; alktty declares futures = 0.3 directly)
- doc comments: alknet/tty → alk/tty, alknet-tty → alktty, the
  alknet-tty-poc and findings.md cross-references trimmed to local
  docs

Test shape: alkcall's Identity.resources is HashMap<String, Vec<String>>
(was HashMap<String, String> in alknet-core); the tests construct with
StdHashMap::new() and infer the new shape from the struct, so no test
edits were needed. alkcall's OwnershipStore::record is 3-arg (no
action) and OwnershipProvider::owns is 4-arg (with action) — both
already match what the ported code calls.

Cargo.lock committed (matches alkcall/alktype convention; still
excluded from the published package via Cargo.toml's exclude list).

src/backend.rs

@@ -0,0 +1,512 @@
+//! Backend trait and handle shapes: `TtyBackend`, `TtyHandle`, `TtyControl`,
+//! `TtyControlHandle`, `TtyParams`, `TerminalParams`, and `TtyError`.
+//!
+//! This is the inversion point (ADR-053) between the wire-format adapter
+//! (`crate::adapter`) and the backend crates (`alktty`'s own `local`
+//! feature module, future `alknet-docker`, `alknet-ssh`). alktty
+//! defines the trait; the
+//! backends implement it. The trait shape is a one-way door — changing it
+//! after backends exist is a rewrite across crates. The adapter holds a
+//! `HashMap<String, Arc<dyn TtyBackend>>` keyed by the negotiation frame's
+//! `backend` string and pumps the `TtyHandle` fields bidirectionally;
+//! backends produce handles, they do not write to the wire.
+
+use std::collections::HashMap;
+use std::path::PathBuf;
+use std::pin::Pin;
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use bytes::Bytes;
+use futures_core::Stream;
+use tokio::sync::{mpsc, oneshot};
+use tokio_stream::wrappers::ReceiverStream;
+
+use crate::negotiation::{NegotiateRequest, TerminalParamsWire};
+
+/// A boxed, sendable future used for the [`TtyHandle::exit_code`] field.
+///
+/// Equivalent to `Pin<Box<dyn Future + Send>>`; defined as a
+/// `futures::future::BoxFuture<'static, T>` to match the alkcall
+/// convention (alkcall pulls in `futures`, so no new transitive dep).
+/// Functionally identical to the hand-rolled `Pin<Box<dyn Future +
+/// Send + 'static>>` — `futures::future::BoxFuture<'static, T>` is
+/// exactly that alias.
+pub type BoxFuture<T> = futures::future::BoxFuture<'static, T>;
+
+/// The error type for [`TtyBackend::allocate`] and the
+/// [`TtyHandle::exit_code`] future.
+///
+/// `#[non_exhaustive]` so new variants are additive (two-way-door extension
+/// within the one-way trait shape — ADR-053).
+#[non_exhaustive]
+#[derive(Debug, thiserror::Error)]
+pub enum TtyError {
+    /// The PTY couldn't be allocated, the docker exec failed to start, the
+    /// SSH channel request was rejected. Returned by `allocate()`; the
+    /// adapter sends `{"error":"allocate_failed",...}` and closes.
+    #[error("allocate failed: {message}")]
+    AllocFailed { message: String },
+    /// The backend couldn't reap the child / determine the exit code.
+    /// Returned by the `exit_code` future; the adapter sends
+    /// `{"type":"exit","code":-1}` (ADR-055 §4).
+    #[error("wait failed: {message}")]
+    WaitFailed { message: String },
+    /// An I/O error from a backend's stream/handle.
+    #[error("io: {0}")]
+    Io(#[from] std::io::Error),
+    /// A backend-specific error not covered by the above (e.g., a bollard
+    /// API error, a russh protocol error).
+    #[error("backend-specific: {message}")]
+    Backend { message: String },
+}
+
+/// Terminal dimensions and mode hints for a PTY allocation
+/// (`TtyParams::terminal: Some`).
+///
+/// `modes` is reserved (OQ-44); backends MUST ignore its content in v1.
+#[derive(Debug, Clone)]
+pub struct TerminalParams {
+    /// `TERM` environment value (e.g., `"xterm-256color"`); `None` =
+    /// backend default.
+    pub term: Option<String>,
+    pub cols: u16,
+    pub rows: u16,
+    pub pixel_width: u16,
+    pub pixel_height: u16,
+    /// Reserved — OQ-44; backends MUST ignore the content in v1.
+    pub modes: serde_json::Value,
+}
+
+/// The allocation request the adapter passes to [`TtyBackend::allocate`].
+///
+/// `terminal: None` is pipe/runner mode (no PTY, separate stdout/stderr —
+/// ADR-054). `terminal: Some` is PTY mode (stdout/stderr merged into
+/// `stdout` by the kernel PTY, real terminal semantics).
+///
+/// `backend_params` is an opaque JSON object the adapter passes verbatim;
+/// each backend deserializes its own strongly-typed params struct from it.
+/// alktty has zero knowledge of any backend's params shape. See ADR-053
+/// §"Backend params are opaque."
+#[derive(Debug, Clone)]
+pub struct TtyParams {
+    /// Terminal parameters. `None` = pipe mode (no PTY — ADR-054). `Some` =
+    /// allocate a PTY with these dimensions.
+    pub terminal: Option<TerminalParams>,
+    /// Command vector (argv[0] + args). Non-empty.
+    pub cmd: Vec<String>,
+    /// Working directory (`None` = inherit/default).
+    pub cwd: Option<PathBuf>,
+    /// Environment variables (empty = inherit).
+    pub env: HashMap<String, String>,
+    /// Backend-specific selector fields from the negotiation frame,
+    /// unparsed. The adapter passes the JSON object through verbatim; the
+    /// backend deserializes its own strongly-typed params struct from it.
+    /// alktty has zero knowledge of any backend's params shape.
+    pub backend_params: serde_json::Map<String, serde_json::Value>,
+}
+
+impl From<TerminalParamsWire> for TerminalParams {
+    fn from(w: TerminalParamsWire) -> Self {
+        Self {
+            term: w.term,
+            cols: w.cols,
+            rows: w.rows,
+            pixel_width: w.pixel_width,
+            pixel_height: w.pixel_height,
+            modes: w.modes,
+        }
+    }
+}
+
+/// Map a wire negotiation frame to the allocation request the backend
+/// consumes. The adapter calls this after parsing the negotiation
+/// carriage; `backend_params` is passed through verbatim (alktty has
+/// zero knowledge of any backend's params shape — ADR-053). Lives here so
+/// the adapter does not hand-roll the mapping.
+impl From<NegotiateRequest> for TtyParams {
+    fn from(req: NegotiateRequest) -> Self {
+        Self {
+            terminal: req.tty.map(TerminalParams::from),
+            cmd: req.cmd,
+            cwd: req.cwd,
+            env: req.env,
+            backend_params: req.backend_params,
+        }
+    }
+}
+
+/// What a backend's `allocate()` produces. The adapter pumps these fields
+/// bidirectionally against the wire format (ADR-052).
+pub struct TtyHandle {
+    /// Stdin writer — bytes the adapter pumps from client stdin chunks.
+    /// `tokio::io::AsyncWrite` (the tokio flavor, not `futures::io` — they
+    /// are incompatible traits; the tokio stack is the adapter's runtime).
+    pub stdin: Box<dyn tokio::io::AsyncWrite + Send + Unpin>,
+    /// Stdout stream — bytes the adapter pumps to client stdout chunks.
+    /// Ends when the backend's stdout reaches EOF.
+    pub stdout: Pin<Box<dyn Stream<Item = Bytes> + Send>>,
+    /// Stderr stream — `None` for PTY backends (stdout/stderr merged into
+    /// `stdout` by the kernel PTY). `Some` for pipe backends (separate
+    /// streams).
+    pub stderr: Option<Pin<Box<dyn Stream<Item = Bytes> + Send>>>,
+    /// Exit code — a `Future` the adapter awaits. Resolves when the
+    /// process/container/SSH exec exits. The adapter sends the result as
+    /// the `{"type":"exit","code":N}` control chunk (ADR-055) and closes
+    /// the stream. This is a `BoxFuture`, not a method on `TtyHandle`, so
+    /// the adapter can `select` between exit and stream-close without
+    /// coupling to the other fields (REQ-TTY-01).
+    ///
+    /// # ADR-056 — kill-on-Drop contract
+    ///
+    /// Dropping this future without driving it to completion MUST kill the
+    /// session target (the child process, the docker exec, the SSH
+    /// channel's process). The kill is best-effort (a no-op if the target
+    /// already exited) but MUST be attempted even when the target is
+    /// blocked in a state that ignores stdin EOF (a daemon, a process in
+    /// uninterruptible sleep, a container whose process ignores channel
+    /// close). The adapter triggers this by dropping the `TtyHandle` on
+    /// session cancel (connection drop, stream reset, task panic); the
+    /// backend wires the kill into this future's `Drop`-on-cancel guard
+    /// (e.g. the local backend holds a `portable_pty::ChildKiller`; docker
+    /// holds the container id + bollard client; SSH holds the russh
+    /// channel). A backend that returns a bare `oneshot::Receiver<i32>`
+    /// (or any future without a kill-on-`Drop` guard) as `exit_code`
+    /// violates the contract and will orphan processes on cancel. The
+    /// `Drop` MUST be a no-op when the future resolved normally (the
+    /// adapter awaited it to completion). See ADR-056 and `tty-local.md`
+    /// §"Cancel-Cleanup" for the local backend's mechanism.
+    pub exit_code: BoxFuture<Result<i32, TtyError>>,
+    /// Control handle (resize, signal) — `Clone` so the adapter can hand
+    /// it to the spawned control-chunk dispatcher. `None` only when the
+    /// backend genuinely has no control path. See OQ-43.
+    pub control: Option<TtyControlHandle>,
+}
+
+/// Control path for a live terminal session: resize and signal forwarding.
+///
+/// Object-safe (`Send + Sync`, no `Clone` — `Clone` is not object-safe).
+/// The `Clone`-ability lives on the [`TtyControlHandle`] newtype, which
+/// holds the trait object behind an `Arc`. A backend produces its own
+/// control type via `TtyControlHandle::new(Arc::new(MyControl))` without
+/// the adapter knowing the concrete shape (OQ-43).
+pub trait TtyControl: Send + Sync {
+    /// Resize the terminal. Maps to SSH `window-change`, docker exec
+    /// resize, or `ioctl(TIOCSWINSZ)` on a local PTY. No-op for pipe
+    /// backends without a PTY (the adapter still calls it; the backend
+    /// ignores).
+    fn resize(&self, cols: u16, rows: u16, pixel_width: u16, pixel_height: u16);
+
+    /// Forward a signal by name. Best-effort delivery to the foreground
+    /// process group (see `tty-local.md` REQ-TTY-02). Unknown names fall
+    /// back to the backend's default kill.
+    fn signal(&self, name: &str);
+}
+
+/// The `Clone`-able handle to a backend's control path. The
+/// [`TtyControl`] trait is NOT `Clone` (`Clone` is not object-safe —
+/// `fn clone(&self) -> Self` returns `Self`, which forbids `dyn` dispatch);
+/// the `Clone`-ability lives on this concrete newtype, which holds the
+/// trait object behind an `Arc`. The adapter clones the `Arc` to hand a
+/// handle to the spawned control-chunk dispatcher. A backend produces its
+/// own control type via `TtyControlHandle::new(Arc::new(MyControl))`
+/// without the adapter knowing the concrete shape. See OQ-43.
+#[derive(Clone)]
+pub struct TtyControlHandle(Arc<dyn TtyControl + Send + Sync>);
+
+impl TtyControlHandle {
+    /// Wrap a backend's control implementation. The backend typically
+    /// calls `TtyControlHandle::new(Arc::new(MyControl))` inside its
+    /// `allocate()`.
+    pub fn new(control: Arc<dyn TtyControl + Send + Sync>) -> Self {
+        Self(control)
+    }
+
+    /// Resize the terminal. Delegates to the inner [`TtyControl`].
+    pub fn resize(&self, cols: u16, rows: u16, pixel_width: u16, pixel_height: u16) {
+        self.0.resize(cols, rows, pixel_width, pixel_height);
+    }
+
+    /// Forward a signal by name. Delegates to the inner [`TtyControl`].
+    pub fn signal(&self, name: &str) {
+        self.0.signal(name);
+    }
+}
+
+/// The backend inversion point (ADR-053). alktty defines the trait;
+/// the backend crates (alktty's `local` feature module, future
+/// `alknet-docker`, `alknet-ssh`) implement it. The adapter holds a
+/// `HashMap<String, Arc<dyn TtyBackend>>` keyed by the negotiation frame's
+/// `backend` string and dispatches by that key.
+///
+/// # REQ-TTY-01 — backends need not be natively async
+///
+/// The adapter-facing types this trait returns (`AsyncWrite`,
+/// `Stream<Item = Bytes>`, `BoxFuture`, `TtyControl`) are the **adapter's
+/// contract**. A backend may expose blocking handles internally (e.g.
+/// `portable_pty`'s blocking `std::io::{Read, Write}` + `Child::wait()`)
+/// and bridge them to these async-facing types via dedicated std threads
+/// or `tokio::task::spawn_blocking` feeding tokio mpsc/oneshot channels.
+/// This bridging pattern is a **documented, supported implementation
+/// strategy**, not a workaround. The local backend (alktty's `local`
+/// the reference implementation: it spawns reader/writer/waiter threads
+/// that feed `mpsc::Receiver<Bytes>` (stdout), an `AsyncWrite` adapter
+/// over `mpsc::Sender<StdinCmd>` (stdin), and a `oneshot::Receiver<i32>`
+/// wrapped in a kill-guard future (exit). The adapter consumes the bridged
+/// async-facing types and is unaware of the threading.
+///
+/// # ADR-056 — kill-on-Drop contract
+///
+/// The [`TtyHandle::exit_code`] future returned by `allocate()` MUST kill
+/// the session target when dropped without being driven to completion. See
+/// the doc comment on [`TtyHandle::exit_code`] for the full contract.
+#[async_trait]
+pub trait TtyBackend: Send + Sync {
+    /// Allocate a terminal/process session and return the handles the
+    /// adapter pumps. The `backend` field of the negotiation frame
+    /// (ADR-052) selects which registered backend's `allocate` is called.
+    async fn allocate(&self, params: &TtyParams) -> Result<TtyHandle, TtyError>;
+
+    /// The pre-existing resource this session targets, for ownership
+    /// checks (ADR-050). `None` = no pre-existing resource (the session
+    /// creates its own — local process, SSH channel). `Some((kind, id))`
+    /// = the session targets an existing resource the caller must own
+    /// (e.g., `DockerTtyBackend` returns `Some(("container", id))`). The
+    /// adapter calls this at negotiation to gate access; the backend
+    /// extracts the id from its own `backend_params`. Default `None`
+    /// (most backends create their own resource).
+    fn resource_id(&self, _params: &TtyParams) -> Option<(&'static str, String)> {
+        None
+    }
+}
+
+/// Mock `TtyControl` for tests. Records the last resize/signal call so
+/// tests can assert delegation through [`TtyControlHandle`].
+#[derive(Default)]
+pub struct MockControl {
+    pub last_resize: std::sync::Mutex<Option<(u16, u16, u16, u16)>>,
+    pub last_signal: std::sync::Mutex<Option<String>>,
+}
+
+impl TtyControl for MockControl {
+    fn resize(&self, cols: u16, rows: u16, pixel_width: u16, pixel_height: u16) {
+        *self.last_resize.lock().expect("resize mutex poisoned") =
+            Some((cols, rows, pixel_width, pixel_height));
+    }
+
+    fn signal(&self, name: &str) {
+        *self.last_signal.lock().expect("signal mutex poisoned") = Some(name.to_string());
+    }
+}
+
+/// In-memory `TtyBackend` for tests. `allocate()` wires tokio mpsc
+/// channels for stdin/stdout/stderr, a oneshot for `exit_code`, and a
+/// mock [`TtyControl`]. The caller can drive the channels directly or via
+/// the adapter pump. Use [`MockBackend::with_exit_code`] to fix the exit
+/// code the handle resolves to.
+#[derive(Default)]
+pub struct MockBackend {
+    pub exit_code: Option<i32>,
+}
+
+impl MockBackend {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    pub fn with_exit_code(exit_code: i32) -> Self {
+        Self {
+            exit_code: Some(exit_code),
+        }
+    }
+}
+
+#[async_trait]
+impl TtyBackend for MockBackend {
+    async fn allocate(&self, _params: &TtyParams) -> Result<TtyHandle, TtyError> {
+        let (_stdout_tx, stdout_rx) = mpsc::channel::<Bytes>(8);
+        let (_stderr_tx, stderr_rx) = mpsc::channel::<Bytes>(8);
+        let (stdin_tx, _stdin_rx) = mpsc::channel::<Bytes>(8);
+        let (exit_tx, exit_rx) = oneshot::channel::<Result<i32, TtyError>>();
+
+        let code = self.exit_code.unwrap_or(0);
+        tokio::spawn(async move {
+            let _ = exit_tx.send(Ok(code));
+        });
+
+        let stdout: Pin<Box<dyn Stream<Item = Bytes> + Send>> =
+            Box::pin(ReceiverStream::new(stdout_rx));
+        let stderr: Option<Pin<Box<dyn Stream<Item = Bytes> + Send>>> =
+            Some(Box::pin(ReceiverStream::new(stderr_rx)));
+        let stdin: Box<dyn tokio::io::AsyncWrite + Send + Unpin> =
+            Box::new(MockStdinSink { tx: stdin_tx });
+        let control = Some(TtyControlHandle::new(Arc::new(MockControl::default())));
+
+        let exit_code: BoxFuture<Result<i32, TtyError>> = Box::pin(async move {
+            exit_rx
+                .await
+                .map_err(|_| TtyError::WaitFailed {
+                    message: "exit_code sender dropped".to_string(),
+                })
+                .and_then(|r| r)
+        });
+
+        Ok(TtyHandle {
+            stdin,
+            stdout,
+            stderr,
+            exit_code,
+            control,
+        })
+    }
+}
+
+/// `AsyncWrite` adapter over an `mpsc::Sender<Bytes>` — the mock
+/// backend's stdin sink. Copies the buffer into a `Bytes` and best-effort
+/// sends; on a full channel returns `Pending`, on a closed channel returns
+/// a broken-pipe error.
+struct MockStdinSink {
+    tx: mpsc::Sender<Bytes>,
+}
+
+impl tokio::io::AsyncWrite for MockStdinSink {
+    fn poll_write(
+        self: Pin<&mut Self>,
+        _cx: &mut std::task::Context<'_>,
+        buf: &[u8],
+    ) -> std::task::Poll<Result<usize, std::io::Error>> {
+        use std::task::Poll;
+        match self.tx.try_reserve() {
+            Ok(permit) => {
+                permit.send(Bytes::copy_from_slice(buf));
+                Poll::Ready(Ok(buf.len()))
+            }
+            Err(mpsc::error::TrySendError::Full(_)) => Poll::Pending,
+            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 std::task::Context<'_>,
+    ) -> std::task::Poll<Result<(), std::io::Error>> {
+        std::task::Poll::Ready(Ok(()))
+    }
+
+    fn poll_shutdown(
+        self: Pin<&mut Self>,
+        _cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Result<(), std::io::Error>> {
+        std::task::Poll::Ready(Ok(()))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn tty_control_handle_clone_delegates_resize_and_signal() {
+        let control = Arc::new(MockControl::default());
+        let handle = TtyControlHandle::new(control.clone());
+        let handle_clone = handle.clone();
+
+        handle.resize(80, 24, 0, 0);
+        handle_clone.signal("HUP");
+
+        let resize = control.last_resize.lock().expect("resize mutex poisoned");
+        assert_eq!(*resize, Some((80, 24, 0, 0)));
+        let signal = control.last_signal.lock().expect("signal mutex poisoned");
+        assert_eq!(*signal, Some("HUP".to_string()));
+    }
+
+    #[tokio::test]
+    async fn mock_backend_allocates_and_exits() {
+        let backend = MockBackend::with_exit_code(42);
+        let params = TtyParams {
+            terminal: None,
+            cmd: vec!["echo".to_string(), "hi".to_string()],
+            cwd: None,
+            env: HashMap::new(),
+            backend_params: serde_json::Map::new(),
+        };
+        let handle = backend.allocate(&params).await.expect("allocate");
+        assert!(handle.stderr.is_some());
+        assert!(handle.control.is_some());
+        let code = handle.exit_code.await.expect("exit_code");
+        assert_eq!(code, 42);
+    }
+
+    #[tokio::test]
+    async fn mock_backend_resource_id_default_none() {
+        let backend = MockBackend::new();
+        let params = TtyParams {
+            terminal: None,
+            cmd: vec!["true".to_string()],
+            cwd: None,
+            env: HashMap::new(),
+            backend_params: serde_json::Map::new(),
+        };
+        assert!(backend.resource_id(&params).is_none());
+    }
+
+    #[test]
+    fn tty_params_from_negotiate_request_maps_fields() {
+        let req = NegotiateRequest {
+            carriage: "raw".to_string(),
+            backend: "local".to_string(),
+            tty: Some(TerminalParamsWire {
+                term: Some("xterm-256color".to_string()),
+                cols: 80,
+                rows: 24,
+                pixel_width: 0,
+                pixel_height: 0,
+                modes: serde_json::Value::Null,
+            }),
+            cmd: vec!["bash".to_string()],
+            cwd: Some(PathBuf::from("/tmp")),
+            env: HashMap::from([("FOO".to_string(), "bar".to_string())]),
+            backend_params: {
+                let mut m = serde_json::Map::new();
+                m.insert(
+                    "container".to_string(),
+                    serde_json::Value::String("abc".to_string()),
+                );
+                m
+            },
+        };
+        let params = TtyParams::from(req);
+        let term = params.terminal.expect("terminal");
+        assert_eq!(term.term.as_deref(), Some("xterm-256color"));
+        assert_eq!(term.cols, 80);
+        assert_eq!(term.rows, 24);
+        assert_eq!(params.cmd, vec!["bash".to_string()]);
+        assert_eq!(params.cwd.as_deref(), Some(std::path::Path::new("/tmp")));
+        assert_eq!(params.env.get("FOO").map(String::as_str), Some("bar"));
+        assert_eq!(
+            params
+                .backend_params
+                .get("container")
+                .and_then(|v| v.as_str()),
+            Some("abc"),
+        );
+    }
+
+    #[test]
+    fn tty_params_from_negotiate_request_pipe_mode() {
+        let req = NegotiateRequest {
+            carriage: "raw".to_string(),
+            backend: "local".to_string(),
+            tty: None,
+            cmd: vec!["true".to_string()],
+            cwd: None,
+            env: HashMap::new(),
+            backend_params: serde_json::Map::new(),
+        };
+        let params = TtyParams::from(req);
+        assert!(params.terminal.is_none());
+    }
+}

src/control.rs

@@ -0,0 +1,275 @@
+//! Control messages carried in `stream_type 3` (`ctrl_in`) and
+//! `stream_type 4` (`ctrl_out`) chunks (ADR-052, amended Phase 7).
+//!
+//! The control channel is split into two halves so it is genuinely
+//! bidirectional on the wire: `STREAM_CTRL_IN = 3` carries client→server
+//! control (`Resize`, `Signal`, `Eof`); `STREAM_CTRL_OUT = 4` carries
+//! server→client control (`Exit`). The previous single
+//! `STREAM_CONTROL = 3` was documented as "bidirectional" but the adapter
+//! ignored `Exit` from the client because it had no way to distinguish
+//! the two directions on the same stream_type — see the Phase 7 notes
+//! in `docs/architecture/tty-wire.md` (the bidirectionality fix that
+//! split the single `STREAM_CONTROL = 3` into `STREAM_CTRL_IN` (3)
+//! and `STREAM_CTRL_OUT` (4)).
+//!
+//! Control chunks carry a JSON payload tagged by `type`. The schema is the
+//! POC's `ControlMessage`:
+//!
+//! ```json
+//! {"type":"resize","cols":80,"rows":24,"pixel_width":0,"pixel_height":0}
+//! {"type":"signal","name":"INT"}
+//! {"type":"eof"}
+//! {"type":"exit","code":0}
+//! ```
+//!
+//! Control messages are rare (resize on window drag, signal on Ctrl-C), so
+//! serialization cost is negligible versus data chunks. The JSON shape is
+//! consistent with the call protocol's JSON-everything stance and easy to
+//! extend: new types are additive on the `type` tag (ADR-052 §"Control
+//! Channel").
+//!
+//! Unknown `type` values: `from_slice` returns a `serde_json::Error`. The
+//! adapter (task `tty/adapter`) ignores that error per the wire spec's
+//! "unknown types are ignored" policy — this keeps the enum exhaustive and
+//! the policy an adapter-level concern, not a schema-level leak.
+
+use serde::{Deserialize, Serialize};
+
+/// A control message riding on `STREAM_CTRL_IN` (stream_type 3,
+/// client→server) or `STREAM_CTRL_OUT` (stream_type 4, server→client).
+///
+/// Direction and mapping (per `tty-wire.md` §"Control Channel"):
+///
+/// | direction      | stream_type     | variant  | maps to                                            |
+/// |----------------|-----------------|----------|----------------------------------------------------|
+/// | client→server  | `STREAM_CTRL_IN` (3)  | `Resize` | SSH `window-change`, docker exec resize, `ioctl`   |
+/// | client→server  | `STREAM_CTRL_IN` (3)  | `Signal` | SSH `signal`, docker exec signal, `kill(-pgid, n)` |
+/// | client→server  | `STREAM_CTRL_IN` (3)  | `Eof`    | SSH channel EOF, docker stdin close, `ChildStdin`  |
+/// | server→client  | `STREAM_CTRL_OUT` (4) | `Exit`   | the completion signal (ADR-055)                    |
+///
+/// The direction is enforced by the adapter, not by this enum: a `Resize`
+/// arriving on `STREAM_CTRL_OUT` is a protocol violation (the adapter
+/// ignores it), and an `Exit` arriving on `STREAM_CTRL_IN` is likewise a
+/// protocol violation (the adapter ignores it). The split is what makes
+/// the control channel genuinely bidirectional — the previous single
+/// `STREAM_CONTROL = 3` was documented as "bidirectional" but the
+/// adapter had to ignore `Exit` from the client because the two
+/// directions were indistinguishable on the same stream_type.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(tag = "type", rename_all = "snake_case")]
+pub enum ControlMessage {
+    /// Terminal window resize (client→server, `STREAM_CTRL_IN`).
+    ///
+    /// `pixel_width`/`pixel_height` default to 0 (most terminals don't
+    /// report pixel dimensions; SSH's `pty_request` carries them for
+    /// completeness).
+    Resize {
+        cols: u16,
+        rows: u16,
+        #[serde(default)]
+        pixel_width: u16,
+        #[serde(default)]
+        pixel_height: u16,
+    },
+    /// Forward a signal to the child process group (client→server,
+    /// `STREAM_CTRL_IN`).
+    ///
+    /// `name` is an uppercase string from the supported set (see
+    /// [`signal_from_name`]). Unknown names fall back to the backend's
+    /// default kill in the adapter (tty-local.md REQ-TTY-02).
+    Signal { name: String },
+    /// Client stdin is done (client→server, `STREAM_CTRL_IN`). The
+    /// server closes the backend's stdin (`ChildStdin::drop` / PTY writer
+    /// close) but keeps pumping stdout + the exit chunk. See
+    /// `tty-wire.md` §"Stdin Closure".
+    Eof,
+    /// Process exit code (server→client, `STREAM_CTRL_OUT`). The exit
+    /// chunk is the last control chunk before stream close (ADR-055).
+    /// `code` is `i32` matching `std::process::ExitStatus::code()`;
+    /// negative values are signal-terminated (e.g., `-9` for SIGKILL on
+    /// Unix). `-1` is the adapter's best-effort "backend could not
+    /// determine the exit code" sentinel (ADR-055 §4).
+    Exit { code: i32 },
+}
+
+impl ControlMessage {
+    /// Serialize to JSON bytes for the control chunk payload.
+    pub fn to_json(&self) -> serde_json::Result<bytes::Bytes> {
+        serde_json::to_vec(self).map(bytes::Bytes::from)
+    }
+
+    /// Deserialize from a control chunk payload (UTF-8 JSON).
+    ///
+    /// Returns `serde_json::Error` on unknown `type` tags; the adapter
+    /// ignores that error per the wire spec's extensibility policy.
+    pub fn from_slice(b: &[u8]) -> serde_json::Result<Self> {
+        serde_json::from_slice(b)
+    }
+}
+
+/// Map an uppercase signal name to a libc signal number.
+///
+/// Supports the common set a terminal front-end would forward: `HUP`,
+/// `INT`, `QUIT`, `TERM`, `KILL`, `USR1`, `USR2`, `TSTP`, `CONT`
+/// (Ctrl-C → `INT`, Ctrl-\ → `QUIT`, Ctrl-Z → `TSTP`). Unknown names
+/// return `None`; the caller (the local backend) decides whether to
+/// ignore or fall back to the backend's default kill
+/// (`portable_pty`'s `ChildKiller::kill` sends SIGHUP — see
+/// tty-local.md REQ-TTY-02).
+///
+/// Unix-only: the non-Unix path falls back to `ChildKiller::kill`
+/// directly. The `#[cfg(unix)]` gate matches the POC.
+#[cfg(unix)]
+pub fn signal_from_name(name: &str) -> Option<i32> {
+    use libc::*;
+    match name {
+        "HUP" => Some(SIGHUP),
+        "INT" => Some(SIGINT),
+        "QUIT" => Some(SIGQUIT),
+        "TERM" => Some(SIGTERM),
+        "KILL" => Some(SIGKILL),
+        "USR1" => Some(SIGUSR1),
+        "USR2" => Some(SIGUSR2),
+        "TSTP" => Some(SIGTSTP),
+        "CONT" => Some(SIGCONT),
+        _ => None,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn round_trip_resize() {
+        let msg = ControlMessage::Resize {
+            cols: 80,
+            rows: 24,
+            pixel_width: 0,
+            pixel_height: 0,
+        };
+        let bytes = msg.to_json().unwrap();
+        let back = ControlMessage::from_slice(&bytes).unwrap();
+        assert_eq!(msg, back);
+    }
+
+    #[test]
+    fn round_trip_resize_with_pixels() {
+        let msg = ControlMessage::Resize {
+            cols: 120,
+            rows: 40,
+            pixel_width: 800,
+            pixel_height: 600,
+        };
+        let bytes = msg.to_json().unwrap();
+        let back = ControlMessage::from_slice(&bytes).unwrap();
+        assert_eq!(msg, back);
+    }
+
+    #[test]
+    fn round_trip_signal() {
+        let msg = ControlMessage::Signal {
+            name: "INT".to_string(),
+        };
+        let bytes = msg.to_json().unwrap();
+        let back = ControlMessage::from_slice(&bytes).unwrap();
+        assert_eq!(msg, back);
+    }
+
+    #[test]
+    fn round_trip_eof() {
+        let msg = ControlMessage::Eof;
+        let bytes = msg.to_json().unwrap();
+        let back = ControlMessage::from_slice(&bytes).unwrap();
+        assert_eq!(msg, back);
+    }
+
+    #[test]
+    fn round_trip_exit() {
+        let msg = ControlMessage::Exit { code: 42 };
+        let bytes = msg.to_json().unwrap();
+        let back = ControlMessage::from_slice(&bytes).unwrap();
+        assert_eq!(msg, back);
+    }
+
+    #[test]
+    fn to_json_emits_snake_case_type_tag() {
+        let resize = ControlMessage::Resize {
+            cols: 80,
+            rows: 24,
+            pixel_width: 0,
+            pixel_height: 0,
+        };
+        let json: serde_json::Value = serde_json::from_slice(&resize.to_json().unwrap()).unwrap();
+        assert_eq!(json["type"], "resize");
+        assert_eq!(json["cols"], 80);
+        assert_eq!(json["rows"], 24);
+        assert_eq!(json["pixel_width"], 0);
+        assert_eq!(json["pixel_height"], 0);
+
+        let signal = ControlMessage::Signal {
+            name: "INT".to_string(),
+        };
+        let json: serde_json::Value = serde_json::from_slice(&signal.to_json().unwrap()).unwrap();
+        assert_eq!(json["type"], "signal");
+        assert_eq!(json["name"], "INT");
+
+        let eof = ControlMessage::Eof;
+        let json: serde_json::Value = serde_json::from_slice(&eof.to_json().unwrap()).unwrap();
+        assert_eq!(json["type"], "eof");
+
+        let exit = ControlMessage::Exit { code: 0 };
+        let json: serde_json::Value = serde_json::from_slice(&exit.to_json().unwrap()).unwrap();
+        assert_eq!(json["type"], "exit");
+        assert_eq!(json["code"], 0);
+    }
+
+    #[test]
+    fn resize_omits_pixel_defaults_on_deserialize() {
+        let json = br#"{"type":"resize","cols":80,"rows":24}"#;
+        let msg = ControlMessage::from_slice(json).unwrap();
+        match msg {
+            ControlMessage::Resize {
+                cols,
+                rows,
+                pixel_width,
+                pixel_height,
+            } => {
+                assert_eq!(cols, 80);
+                assert_eq!(rows, 24);
+                assert_eq!(pixel_width, 0);
+                assert_eq!(pixel_height, 0);
+            }
+            _ => panic!("expected Resize"),
+        }
+    }
+
+    #[test]
+    fn from_slice_unknown_type_returns_error() {
+        let json = br#"{"type":"unknown"}"#;
+        assert!(ControlMessage::from_slice(json).is_err());
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn signal_from_name_known() {
+        use libc::*;
+        assert_eq!(signal_from_name("HUP"), Some(SIGHUP));
+        assert_eq!(signal_from_name("INT"), Some(SIGINT));
+        assert_eq!(signal_from_name("QUIT"), Some(SIGQUIT));
+        assert_eq!(signal_from_name("TERM"), Some(SIGTERM));
+        assert_eq!(signal_from_name("KILL"), Some(SIGKILL));
+        assert_eq!(signal_from_name("USR1"), Some(SIGUSR1));
+        assert_eq!(signal_from_name("USR2"), Some(SIGUSR2));
+        assert_eq!(signal_from_name("TSTP"), Some(SIGTSTP));
+        assert_eq!(signal_from_name("CONT"), Some(SIGCONT));
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn signal_from_name_unknown() {
+        assert_eq!(signal_from_name("NOPE"), None);
+        assert_eq!(signal_from_name(""), None);
+        assert_eq!(signal_from_name("int"), None);
+    }
+}

src/lib.rs

@@ -0,0 +1,56 @@
+//! alktty: Terminal session protocol for the `alk/tty` ALPN.
+//!
+//! Producer/consumer protocol crate on top of alkcall channels. Two
+//! halves (per alkcall's protocol-crate pattern):
+//!
+//! - **Producer half** — [`adapter::TtyAdapter`] (direct `alk/tty`
+//!   ALPN via `ProtocolHandler`) + [`channels`] (registers the
+//!   `channels/tty/sub` open op via `ChannelCore::register_openable`
+//!   for the `alk/channels` multiplexed path).
+//! - **Consumer half** — [`session::TtySession`] (typed client wrapper
+//!   around the wire protocol, with `connect_direct` and
+//!   `open_via_channels` constructors).
+//!
+//! Two-carriage wire format (ADR-052): a JSON negotiation frame, then
+//! raw chunks (`[stream_type: u8][length: u32 be][payload]`).
+//! Backend-agnostic via the [`backend::TtyBackend`] trait (ADR-053).
+//! Depends on alkcall (ADR-057 — the negotiation framing is
+//! self-contained; alkcall's `EventEnvelope` framing is not reused).
+//!
+//! # WASM target
+//!
+//! The default crate (no features) compiles to
+//! `wasm32-unknown-unknown`. The [`local`] module is feature-gated
+//! and non-wasm by design (`portable-pty` + `tokio::process` need a
+//! real OS). Downstream TS/Python adapters compile the protocol
+//! layer in a sandbox; the local-process backend runs on a real OS.
+//!
+//! # Local backend
+//!
+//! The local backend ([`local::LocalTtyBackend`]) is gated behind the
+//! `local` feature. It implements [`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).
+//!
+//! # Assembly pattern
+//!
+//! ```ignore
+//! let mut backends = std::collections::HashMap::new();
+//! backends.insert(
+//!     "local".into(),
+//!     std::sync::Arc::new(alktty::local::LocalTtyBackend::new())
+//!         as std::sync::Arc<dyn alktty::backend::TtyBackend>,
+//! );
+//! let tty_adapter = alktty::adapter::TtyAdapter::new(backends);
+//! ```
+
+pub mod adapter;
+pub mod backend;
+pub mod control;
+pub mod negotiation;
+pub mod wire;
+
+#[cfg(feature = "local")]
+pub mod local;

src/negotiation.rs

@@ -0,0 +1,526 @@
+//! Negotiation carriage: `NegotiateRequest`, `TerminalParamsWire`,
+//! length-prefixed framing reader/writer, and the error response shape.
+//!
+//! Phase 1 of the `alk/tty` wire protocol (ADR-052). The client opens a
+//! bidi stream and writes a single length-prefixed JSON frame carrying the
+//! terminal parameters, backend selector, command, and environment. After
+//! this frame, the stream switches to raw chunks (task `tty/wire-codec`).
+//!
+//! The framing is self-contained in alktty (ADR-057): a 4-byte
+//! big-endian length prefix + UTF-8 JSON body. The format coincides with
+//! alkcall's `EventEnvelope` framing by convention, not by code reuse
+//! — alktty does not depend on alkcall's internal wire types.
+//!
+//! # Framing disambiguation
+//!
+//! A server-side error response (JSON, length-prefixed) and a successful
+//! allocation's first raw chunk both begin with bytes the client reads
+//! before knowing which framing applies. The disambiguation is by the
+//! first byte:
+//!
+//! - An error frame's 4-byte big-endian length prefix starts with `0x00`
+//!   because error frames MUST be under 16 MiB ([`MAX_CHUNK_LEN`]) so the
+//!   high byte is zero (a wire-format invariant, not an assumption).
+//! - A raw chunk's first byte is a `stream_type`. The server never sends
+//!   `0` (stdin — client→server only) or `3` (`STREAM_CTRL_IN` —
+//!   client→server only), so the server-sent set is `{1, 2, 4}`
+//!   (stdout, stderr, `STREAM_CTRL_OUT`); `0x00` is unambiguous.
+//!
+//! See ADR-052 §5 and `tty-wire.md` §"Constraints".
+
+use std::collections::HashMap;
+use std::io;
+use std::path::PathBuf;
+
+use bytes::Bytes;
+use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
+
+use crate::wire::MAX_CHUNK_LEN;
+
+/// The Phase 1 JSON negotiation frame payload (ADR-052).
+///
+/// The client writes a single length-prefixed frame containing this struct,
+/// then switches to raw chunks. The adapter parses it, dispatches on
+/// `backend`, and passes `backend_params` verbatim to the selected
+/// backend's `allocate()`.
+///
+/// # Validation
+///
+/// This struct only parses. The adapter (task `tty/adapter`) validates:
+/// - `carriage` MUST be `"raw"` (else `malformed_negotiation`).
+/// - `cmd` MUST be non-empty (else `malformed_negotiation`).
+/// - `backend` MUST be a registered backend key (else `unknown_backend`).
+///
+/// Backend-specific params validation is the backend's job (in `allocate()`).
+///
+/// # `serde(flatten)` for backend-specific fields
+///
+/// The negotiation frame's top-level JSON object carries both the shared
+/// fields (`carriage`, `backend`, `tty`, `cmd`, `cwd`, `env`) and
+/// backend-specific fields (e.g., `"container": "abc123"` for docker); the
+/// latter land in `backend_params` via the `serde(flatten)` below. The
+/// shared fields are consumed by name; whatever remains flows into the
+/// `backend_params` map.
+#[derive(Debug, Clone, serde::Deserialize)]
+pub struct NegotiateRequest {
+    /// `"raw"` in v1; any other value → `malformed_negotiation` (checked by
+    /// the adapter, not this parser).
+    pub carriage: String,
+    /// Backend selector key (`"local"`, `"docker"`, `"ssh"`).
+    pub backend: String,
+    /// `None` = pipe mode (no PTY — ADR-054). `Some` = allocate a PTY with
+    /// these dimensions.
+    #[serde(default)]
+    pub tty: Option<TerminalParamsWire>,
+    /// Command vector (argv[0] + args); non-empty (checked by the adapter).
+    pub cmd: Vec<String>,
+    /// Working directory (`None` = inherit/default).
+    #[serde(default)]
+    pub cwd: Option<PathBuf>,
+    /// Environment variables (empty = inherit).
+    #[serde(default)]
+    pub env: HashMap<String, String>,
+    /// Backend-specific selector fields, opaque to alktty. The adapter
+    /// passes this map through verbatim; each backend deserializes its own
+    /// strongly-typed params struct from it. See ADR-053 §"Backend params
+    /// are opaque."
+    ///
+    /// Populated by `serde(flatten)`: any top-level key not matching a named
+    /// field above lands here.
+    #[serde(flatten)]
+    pub backend_params: serde_json::Map<String, serde_json::Value>,
+}
+
+/// Terminal parameters carried in [`NegotiateRequest::tty`] (ADR-052).
+///
+/// Maps to SSH's `pty_request` parameters, to docker's
+/// `CreateExecOptions { tty: true }`, and to `portable_pty::PtySystem::openpty`
+/// for the local backend. The `modes` field is reserved (OQ-44 — default
+/// terminal modes suffice for the current scope); backends MUST ignore its
+/// content in v1.
+#[derive(Debug, Clone, serde::Deserialize)]
+pub struct TerminalParamsWire {
+    /// `TERM` environment value (e.g., `"xterm-256color"`); `None` =
+    /// backend default.
+    #[serde(default)]
+    pub term: Option<String>,
+    /// Terminal columns.
+    pub cols: u16,
+    /// Terminal rows.
+    pub rows: u16,
+    /// Pixel width (most terminals don't report this; defaults to 0).
+    #[serde(default)]
+    pub pixel_width: u16,
+    /// Pixel height (most terminals don't report this; defaults to 0).
+    #[serde(default)]
+    pub pixel_height: u16,
+    /// Reserved — OQ-44; backends MUST ignore the content in v1.
+    #[serde(default)]
+    pub modes: serde_json::Value,
+}
+
+/// Errors from the negotiation framing reader/writer.
+///
+/// `ConnectionClosed` is returned (rather than `Io`) when `read_frame` hits
+/// a clean `UnexpectedEof` reading either the length prefix or the body —
+/// the peer closed the stream cleanly rather than failing the transport.
+#[derive(Debug, thiserror::Error)]
+pub enum NegotiationError {
+    /// Underlying transport I/O error (not a clean EOF).
+    #[error("io: {0}")]
+    Io(#[from] io::Error),
+    /// The peer closed the stream cleanly (unexpected EOF on the length
+    /// prefix or the body).
+    #[error("connection closed")]
+    ConnectionClosed,
+    /// The frame length exceeded [`MAX_CHUNK_LEN`]. A malformed length
+    /// prefix can't trigger an oversized allocation.
+    #[error("frame too large: {0}")]
+    FrameTooLarge(u32),
+    /// JSON parse error on the negotiation frame or error response.
+    #[error("json: {0}")]
+    Json(#[from] serde_json::Error),
+}
+
+/// Reads length-prefixed negotiation frames from an [`AsyncRead`] transport.
+///
+/// [`NegotiationReader::read_frame`] reads a 4-byte big-endian length
+/// prefix, bounds-checks it against [`MAX_CHUNK_LEN`] (so a malformed
+/// prefix can't trigger an oversized allocation), then reads the body. On a
+/// clean `UnexpectedEof` it returns [`NegotiationError::ConnectionClosed`].
+///
+/// After reading the single negotiation frame, call
+/// [`NegotiationReader::into_inner`] to reclaim the underlying stream for
+/// raw-chunk reading (the reader buffers nothing past the frame boundary,
+/// so the stream is clean for [`crate::wire::ChunkReader`]).
+pub struct NegotiationReader<R: AsyncRead + Unpin> {
+    reader: R,
+    len_buf: [u8; 4],
+}
+
+impl<R: AsyncRead + Unpin> NegotiationReader<R> {
+    /// Wrap an [`AsyncRead`] transport in a negotiation frame reader.
+    pub fn new(reader: R) -> Self {
+        Self {
+            reader,
+            len_buf: [0u8; 4],
+        }
+    }
+
+    /// Consume the reader and return the underlying transport. Use this
+    /// after reading the negotiation frame to reclaim the stream for
+    /// raw-chunk reading.
+    pub fn into_inner(self) -> R {
+        self.reader
+    }
+
+    /// Read one length-prefixed frame: 4-byte BE length, bounds-check,
+    /// body.
+    ///
+    /// Returns the raw frame bytes (the caller deserializes JSON). On a
+    /// clean `UnexpectedEof` reading either the length prefix or the body,
+    /// returns [`NegotiationError::ConnectionClosed`]. On a length prefix
+    /// exceeding [`MAX_CHUNK_LEN`], returns
+    /// [`NegotiationError::FrameTooLarge`].
+    pub async fn read_frame(&mut self) -> Result<Bytes, NegotiationError> {
+        match self.reader.read_exact(&mut self.len_buf).await {
+            Ok(_) => {}
+            Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => {
+                return Err(NegotiationError::ConnectionClosed);
+            }
+            Err(e) => return Err(NegotiationError::Io(e)),
+        }
+
+        let length = u32::from_be_bytes(self.len_buf);
+        if length > MAX_CHUNK_LEN {
+            return Err(NegotiationError::FrameTooLarge(length));
+        }
+
+        let mut buf = vec![0u8; length as usize];
+        if length > 0 {
+            match self.reader.read_exact(&mut buf).await {
+                Ok(_) => {}
+                Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => {
+                    return Err(NegotiationError::ConnectionClosed);
+                }
+                Err(e) => return Err(NegotiationError::Io(e)),
+            }
+        }
+
+        Ok(Bytes::from(buf))
+    }
+}
+
+/// Writes length-prefixed negotiation frames to an [`AsyncWrite`] transport.
+///
+/// [`NegotiationWriter::write_frame`] writes a 4-byte big-endian length
+/// prefix followed by the body, then flushes.
+///
+/// For server-side error responses, the body MUST be under 16 MiB
+/// ([`MAX_CHUNK_LEN`]) so the high byte of the length prefix is `0x00` —
+/// this is the wire-format invariant that makes the framing-disambiguation
+/// trick sound (see ADR-052 §5). [`Self::write_frame`] does not enforce
+/// this; callers building error responses with [`error_response_bytes`] are
+/// well within the limit by construction.
+pub struct NegotiationWriter<W: AsyncWrite + Unpin> {
+    writer: W,
+}
+
+impl<W: AsyncWrite + Unpin> NegotiationWriter<W> {
+    /// Wrap an [`AsyncWrite`] transport in a negotiation frame writer.
+    pub fn new(writer: W) -> Self {
+        Self { writer }
+    }
+
+    /// Consume the writer and return the underlying transport.
+    pub fn into_inner(self) -> W {
+        self.writer
+    }
+
+    /// Write one length-prefixed frame: 4-byte BE length + body + flush.
+    pub async fn write_frame(&mut self, body: &[u8]) -> Result<(), NegotiationError> {
+        let len = body.len() as u32;
+        self.writer.write_all(&len.to_be_bytes()).await?;
+        if !body.is_empty() {
+            self.writer.write_all(body).await?;
+        }
+        self.writer.flush().await?;
+        Ok(())
+    }
+}
+
+/// Serialize a negotiation error response to JSON bytes.
+///
+/// Produces `{"error":"<error>","<field>":"<value>",...}` — the
+/// length-prefixed error frame the server sends when it cannot allocate
+/// the session (unknown backend, malformed negotiation, allocate failed).
+/// The caller writes the result via [`NegotiationWriter::write_frame`].
+///
+/// Error frames MUST be under 16 MiB ([`MAX_CHUNK_LEN`]) so the high byte
+/// of the 4-byte length prefix is `0x00` (framing disambiguation — ADR-052
+/// §5). Realistic error responses are tens of bytes; this invariant holds
+/// by construction.
+pub fn error_response_bytes(error: &str, fields: &[(&str, &str)]) -> serde_json::Result<Vec<u8>> {
+    use serde_json::json;
+    let mut map = serde_json::Map::new();
+    map.insert("error".to_string(), json!(error));
+    for (k, v) in fields {
+        map.insert((*k).to_string(), json!(v));
+    }
+    serde_json::to_vec(&map)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    use tokio::io::duplex;
+
+    #[test]
+    fn round_trip_negotiate_request_all_fields() {
+        let json = serde_json::json!({
+            "carriage": "raw",
+            "backend": "local",
+            "tty": {
+                "term": "xterm-256color",
+                "cols": 80,
+                "rows": 24,
+                "pixel_width": 0,
+                "pixel_height": 0,
+                "modes": {}
+            },
+            "cmd": ["/bin/bash", "-l"],
+            "cwd": "/tmp",
+            "env": {"FOO": "bar"},
+            "container": "abc123"
+        });
+        let req: NegotiateRequest = serde_json::from_value(json).expect("parse");
+        assert_eq!(req.carriage, "raw");
+        assert_eq!(req.backend, "local");
+        let tty = req.tty.expect("tty");
+        assert_eq!(tty.term.as_deref(), Some("xterm-256color"));
+        assert_eq!(tty.cols, 80);
+        assert_eq!(tty.rows, 24);
+        assert_eq!(tty.pixel_width, 0);
+        assert_eq!(tty.pixel_height, 0);
+        assert_eq!(tty.modes, serde_json::json!({}));
+        assert_eq!(req.cmd, vec!["/bin/bash".to_string(), "-l".to_string()]);
+        assert_eq!(req.cwd.as_deref(), Some(std::path::Path::new("/tmp")));
+        assert_eq!(req.env.get("FOO").map(String::as_str), Some("bar"));
+        assert_eq!(
+            req.backend_params.get("container").and_then(|v| v.as_str()),
+            Some("abc123"),
+        );
+    }
+
+    #[test]
+    fn serde_flatten_captures_backend_specific_fields_into_backend_params() {
+        let json = serde_json::json!({
+            "carriage": "raw",
+            "backend": "docker",
+            "cmd": ["bash"],
+            "container": "abc123",
+            "image": "ubuntu:22.04",
+            "remove": true
+        });
+        let req: NegotiateRequest = serde_json::from_value(json).expect("parse");
+        assert_eq!(req.backend_params.len(), 3);
+        assert_eq!(
+            req.backend_params.get("container").and_then(|v| v.as_str()),
+            Some("abc123"),
+        );
+        assert_eq!(
+            req.backend_params.get("image").and_then(|v| v.as_str()),
+            Some("ubuntu:22.04"),
+        );
+        assert_eq!(
+            req.backend_params.get("remove").and_then(|v| v.as_bool()),
+            Some(true),
+        );
+    }
+
+    #[test]
+    fn defaults_applied_when_optional_fields_absent() {
+        let json = serde_json::json!({
+            "carriage": "raw",
+            "backend": "local",
+            "cmd": ["true"]
+        });
+        let req: NegotiateRequest = serde_json::from_value(json).expect("parse");
+        assert!(req.tty.is_none());
+        assert!(req.cwd.is_none());
+        assert!(req.env.is_empty());
+        assert!(req.backend_params.is_empty());
+    }
+
+    #[test]
+    fn terminal_params_wire_defaults() {
+        let json = serde_json::json!({"cols": 80, "rows": 24});
+        let tty: TerminalParamsWire = serde_json::from_value(json).expect("parse");
+        assert!(tty.term.is_none());
+        assert_eq!(tty.pixel_width, 0);
+        assert_eq!(tty.pixel_height, 0);
+        assert!(tty.modes.is_null());
+    }
+
+    #[test]
+    fn carriage_not_raw_still_parses_adapter_validates() {
+        let json = serde_json::json!({
+            "carriage": "json",
+            "backend": "local",
+            "cmd": ["bash"]
+        });
+        let req: NegotiateRequest = serde_json::from_value(json).expect("parse");
+        assert_eq!(req.carriage, "json");
+    }
+
+    #[tokio::test]
+    async fn round_trip_frame_reader_writer() {
+        let (mut a, mut b) = duplex(8 * 1024);
+        let body = br#"{"carriage":"raw","backend":"local","cmd":["bash"]}"#;
+        let mut writer = NegotiationWriter::new(&mut a);
+        let mut reader = NegotiationReader::new(&mut b);
+        writer.write_frame(body).await.expect("write");
+        let read = reader.read_frame().await.expect("read");
+        assert_eq!(read.as_ref(), body);
+    }
+
+    #[tokio::test]
+    async fn frame_too_large_on_length_exceeding_max_chunk_len() {
+        let (mut a, mut b) = duplex(8 * 1024);
+        let over = MAX_CHUNK_LEN + 1;
+        a.write_all(&over.to_be_bytes()).await.expect("write len");
+        a.flush().await.expect("flush");
+
+        let mut reader = NegotiationReader::new(&mut b);
+        let err = reader.read_frame().await.unwrap_err();
+        assert!(matches!(err, NegotiationError::FrameTooLarge(v) if v == over));
+    }
+
+    #[tokio::test]
+    async fn connection_closed_on_truncated_length_prefix() {
+        let (mut a, mut b) = duplex(8 * 1024);
+        a.write_all(&[0u8, 0]).await.expect("write partial");
+        a.flush().await.expect("flush");
+        a.shutdown().await.expect("shutdown");
+
+        let mut reader = NegotiationReader::new(&mut b);
+        let err = reader.read_frame().await.unwrap_err();
+        assert!(matches!(err, NegotiationError::ConnectionClosed));
+    }
+
+    #[tokio::test]
+    async fn connection_closed_on_truncated_body() {
+        let (mut a, mut b) = duplex(8 * 1024);
+        a.write_all(&16u32.to_be_bytes()).await.expect("write len");
+        a.write_all(b"short").await.expect("write partial body");
+        a.flush().await.expect("flush");
+        a.shutdown().await.expect("shutdown");
+
+        let mut reader = NegotiationReader::new(&mut b);
+        let err = reader.read_frame().await.unwrap_err();
+        assert!(matches!(err, NegotiationError::ConnectionClosed));
+    }
+
+    #[tokio::test]
+    async fn connection_closed_clean_close_no_bytes() {
+        let (mut a, mut b) = duplex(8 * 1024);
+        a.shutdown().await.expect("shutdown");
+
+        let mut reader = NegotiationReader::new(&mut b);
+        let err = reader.read_frame().await.unwrap_err();
+        assert!(matches!(err, NegotiationError::ConnectionClosed));
+    }
+
+    #[tokio::test]
+    async fn into_inner_reader_reclaims_stream() {
+        let (mut a, mut b) = duplex(8 * 1024);
+        let body = br#"{"carriage":"raw","backend":"local","cmd":["bash"]}"#;
+        let mut writer = NegotiationWriter::new(&mut a);
+        let mut reader = NegotiationReader::new(&mut b);
+        writer.write_frame(body).await.expect("write");
+        let read = reader.read_frame().await.expect("read");
+        assert_eq!(read.as_ref(), body);
+
+        let reclaimed = reader.into_inner();
+        let mut leftover = [0u8; 4];
+        a.write_all(b"tail").await.expect("write leftover");
+        a.flush().await.expect("flush");
+        reclaimed
+            .read_exact(&mut leftover)
+            .await
+            .expect("read leftover");
+        assert_eq!(&leftover, b"tail");
+    }
+
+    #[tokio::test]
+    async fn into_inner_writer_reclaims_stream() {
+        let (mut a, mut b) = duplex(8 * 1024);
+        let mut writer = NegotiationWriter::new(&mut a);
+        writer.write_frame(b"x").await.expect("write");
+        let reclaimed = writer.into_inner();
+        reclaimed.write_all(b"raw").await.expect("write raw");
+        reclaimed.flush().await.expect("flush");
+
+        let mut len = [0u8; 4];
+        b.read_exact(&mut len).await.expect("read len");
+        assert_eq!(u32::from_be_bytes(len), 1);
+        let mut body = [0u8; 1];
+        b.read_exact(&mut body).await.expect("read body");
+        assert_eq!(&body, b"x");
+        let mut tail = [0u8; 3];
+        b.read_exact(&mut tail).await.expect("read tail");
+        assert_eq!(&tail, b"raw");
+    }
+
+    #[test]
+    fn error_response_bytes_shape() {
+        let bytes = error_response_bytes("unknown_backend", &[("backend", "kubernetes")])
+            .expect("serialize");
+        let v: serde_json::Value = serde_json::from_slice(&bytes).expect("parse");
+        assert_eq!(v["error"], "unknown_backend");
+        assert_eq!(v["backend"], "kubernetes");
+    }
+
+    #[test]
+    fn error_response_bytes_no_extra_fields() {
+        let bytes = error_response_bytes("malformed_negotiation", &[("message", "bad")])
+            .expect("serialize");
+        let v: serde_json::Value = serde_json::from_slice(&bytes).expect("parse");
+        assert_eq!(v["error"], "malformed_negotiation");
+        assert_eq!(v["message"], "bad");
+        assert_eq!(v.as_object().map(|m| m.len()), Some(2));
+    }
+
+    #[tokio::test]
+    async fn error_frame_first_byte_is_zero_for_framing_disambiguation() {
+        let (mut a, mut b) = duplex(8 * 1024);
+        let body =
+            error_response_bytes("unknown_backend", &[("backend", "kubernetes")]).expect("ser");
+        let mut writer = NegotiationWriter::new(&mut a);
+        writer.write_frame(&body).await.expect("write");
+
+        let mut first = [0u8; 1];
+        b.read_exact(&mut first).await.expect("read first byte");
+        assert_eq!(first[0], 0x00);
+
+        let mut len_rest = [0u8; 3];
+        b.read_exact(&mut len_rest).await.expect("read len rest");
+        let len = u32::from_be_bytes([first[0], len_rest[0], len_rest[1], len_rest[2]]);
+        let mut buf = vec![0u8; len as usize];
+        b.read_exact(&mut buf).await.expect("read body");
+        let v: serde_json::Value = serde_json::from_slice(&buf).expect("parse");
+        assert_eq!(v["error"], "unknown_backend");
+        assert_eq!(v["backend"], "kubernetes");
+    }
+
+    #[tokio::test]
+    async fn write_frame_empty_body_writes_length_zero() {
+        let (mut a, mut b) = duplex(8 * 1024);
+        let mut writer = NegotiationWriter::new(&mut a);
+        writer.write_frame(b"").await.expect("write");
+        let mut reader = NegotiationReader::new(&mut b);
+        let read = reader.read_frame().await.expect("read");
+        assert!(read.is_empty());
+    }
+}

src/wire.rs

@@ -0,0 +1,435 @@
+//! Raw chunk codec for the `alk/tty` bidi stream (ADR-052, Phase 2
+//! "raw carriage").
+//!
+//! Wire format:
+//! ```text
+//!   [stream_type: u8][length: u32 be][payload bytes]
+//! ```
+//!
+//! `stream_type`:
+//!   - 0 = stdin     (client→server, raw bytes)
+//!   - 1 = stdout    (server→client, raw bytes)
+//!   - 2 = stderr    (server→client, raw bytes)
+//!   - 3 = ctrl_in   (client→server, JSON control message — see [`crate::control`])
+//!   - 4 = ctrl_out  (server→client, JSON control message — see [`crate::control`])
+//!
+//! The control channel is split into two halves so it is genuinely
+//! bidirectional on the wire: `STREAM_CTRL_IN = 3` carries client→server
+//! control (resize, signal, eof); `STREAM_CTRL_OUT = 4` carries
+//! server→client control (exit). The previous single `STREAM_CONTROL = 3`
+//! was documented as "bidirectional" but the adapter ignored `Exit` from
+//! the client because it had no way to distinguish the two directions on
+//! the same stream_type — see the Phase 7 notes in
+//! `docs/architecture/tty-wire.md` (the bidirectionality fix that
+//! split the single `STREAM_CONTROL = 3` into `STREAM_CTRL_IN` (3)
+//! and `STREAM_CTRL_OUT` (4)).
+//!
+//! Zero-length data chunks are sentinels: a zero-length stdin chunk is EOF
+//! from the client; a zero-length stdout chunk is "drained" from the
+//! server. Control chunks are never zero-length (the JSON payload is at
+//! least `{}`). The codec does not special-case sentinels — they are just
+//! chunks with `length == 0`; the adapter interprets them. See
+//! `docs/architecture/crates/tty/tty-wire.md` §"Sentinels".
+
+use std::io;
+
+use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
+
+/// stdin channel (client→server, raw bytes).
+pub const STREAM_STDIN: u8 = 0;
+/// stdout channel (server→client, raw bytes).
+pub const STREAM_STDOUT: u8 = 1;
+/// stderr channel (server→client, raw bytes).
+pub const STREAM_STDERR: u8 = 2;
+/// Control channel, client→server half (JSON control message —
+/// `Resize`, `Signal`, `Eof`).
+pub const STREAM_CTRL_IN: u8 = 3;
+/// Control channel, server→client half (JSON control message — `Exit`).
+pub const STREAM_CTRL_OUT: u8 = 4;
+
+/// Chunk header length in bytes: 1 byte `stream_type` + 4 bytes `length`.
+pub const CHUNK_HEADER_LEN: usize = 5;
+/// Maximum payload length. A larger chunk is a `ChunkTooLarge` protocol
+/// error. Shared with the negotiation module so error frames (which reuse
+/// the 4-byte length-prefix framing) stay under 16 MiB — this keeps the
+/// high byte of the length prefix `0x00`, which is what makes the
+/// framing-disambiguation trick sound (see ADR-052 §5).
+pub const MAX_CHUNK_LEN: u32 = 16 * 1024 * 1024;
+
+/// Errors from the raw chunk codec.
+///
+/// `ConnectionClosed` is returned (rather than `Io`) when `read_chunk`
+/// hits a clean `UnexpectedEof` reading either the header or the payload —
+/// the peer closed the stream cleanly rather than failing the transport.
+#[derive(Debug, thiserror::Error)]
+pub enum RawError {
+    /// Underlying transport I/O error (not a clean EOF).
+    #[error("io: {0}")]
+    Io(#[from] io::Error),
+    /// The peer closed the stream cleanly (unexpected EOF on header or payload).
+    #[error("connection closed")]
+    ConnectionClosed,
+    /// The chunk header's `stream_type` byte was > 4.
+    #[error("invalid chunk header: stream type {0}")]
+    InvalidStreamType(u8),
+    /// The chunk payload length exceeded `MAX_CHUNK_LEN`.
+    #[error("chunk too large: {0}")]
+    ChunkTooLarge(u32),
+}
+
+/// A single raw chunk on the wire: a `stream_type` byte's channel and the
+/// payload bytes.
+///
+/// Construct with [`Chunk::stdin`], [`Chunk::stdout`], [`Chunk::stderr`],
+/// [`Chunk::ctrl_in`], or [`Chunk::ctrl_out`] for the five fixed
+/// channels.
+#[derive(Debug, Clone)]
+pub struct Chunk {
+    /// The channel: one of [`STREAM_STDIN`], [`STREAM_STDOUT`],
+    /// [`STREAM_STDERR`], [`STREAM_CTRL_IN`], [`STREAM_CTRL_OUT`].
+    pub stream_type: u8,
+    /// The payload bytes (raw for data channels, UTF-8 JSON for control).
+    pub bytes: bytes::Bytes,
+}
+
+impl Chunk {
+    /// A stdin chunk (stream_type 0).
+    pub fn stdin(bytes: bytes::Bytes) -> Self {
+        Self {
+            stream_type: STREAM_STDIN,
+            bytes,
+        }
+    }
+
+    /// A stdout chunk (stream_type 1).
+    pub fn stdout(bytes: bytes::Bytes) -> Self {
+        Self {
+            stream_type: STREAM_STDOUT,
+            bytes,
+        }
+    }
+
+    /// A stderr chunk (stream_type 2).
+    pub fn stderr(bytes: bytes::Bytes) -> Self {
+        Self {
+            stream_type: STREAM_STDERR,
+            bytes,
+        }
+    }
+
+    /// A client→server control chunk (stream_type 3) — `Resize`, `Signal`,
+    /// or `Eof`.
+    pub fn ctrl_in(bytes: bytes::Bytes) -> Self {
+        Self {
+            stream_type: STREAM_CTRL_IN,
+            bytes,
+        }
+    }
+
+    /// A server→client control chunk (stream_type 4) — `Exit`.
+    pub fn ctrl_out(bytes: bytes::Bytes) -> Self {
+        Self {
+            stream_type: STREAM_CTRL_OUT,
+            bytes,
+        }
+    }
+}
+
+/// Reads raw chunks from an [`AsyncRead`] transport.
+///
+/// [`ChunkReader::read_chunk`] reads the 5-byte header, validates the
+/// `stream_type` (≤ 4, else [`RawError::InvalidStreamType`]) and the
+/// payload length (≤ [`MAX_CHUNK_LEN`], else [`RawError::ChunkTooLarge`]),
+/// then reads the payload. On a clean `UnexpectedEof` reading either the
+/// header or the payload, it returns [`RawError::ConnectionClosed`] — the
+/// stream ended cleanly, not with a transport error.
+pub struct ChunkReader<R: AsyncRead + Unpin> {
+    reader: R,
+    header: [u8; CHUNK_HEADER_LEN],
+}
+
+impl<R: AsyncRead + Unpin> ChunkReader<R> {
+    /// Wrap an [`AsyncRead`] transport in a chunk reader.
+    pub fn new(reader: R) -> Self {
+        Self {
+            reader,
+            header: [0u8; CHUNK_HEADER_LEN],
+        }
+    }
+
+    /// Consume the reader and return the underlying transport.
+    pub fn into_inner(self) -> R {
+        self.reader
+    }
+
+    /// Read one chunk: header, validate, payload.
+    pub async fn read_chunk(&mut self) -> Result<Chunk, RawError> {
+        match self.reader.read_exact(&mut self.header).await {
+            Ok(_) => {}
+            Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => {
+                return Err(RawError::ConnectionClosed);
+            }
+            Err(e) => return Err(RawError::Io(e)),
+        }
+
+        let stream_type = self.header[0];
+        if stream_type > 4 {
+            return Err(RawError::InvalidStreamType(stream_type));
+        }
+
+        let length = u32::from_be_bytes([
+            self.header[1],
+            self.header[2],
+            self.header[3],
+            self.header[4],
+        ]);
+        if length > MAX_CHUNK_LEN {
+            return Err(RawError::ChunkTooLarge(length));
+        }
+
+        let mut buf = vec![0u8; length as usize];
+        if length > 0 {
+            match self.reader.read_exact(&mut buf).await {
+                Ok(_) => {}
+                Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => {
+                    return Err(RawError::ConnectionClosed);
+                }
+                Err(e) => return Err(RawError::Io(e)),
+            }
+        }
+
+        Ok(Chunk {
+            stream_type,
+            bytes: bytes::Bytes::from(buf),
+        })
+    }
+}
+
+/// Writes raw chunks to an [`AsyncWrite`] transport.
+///
+/// [`ChunkWriter::write_chunk`] writes the 5-byte header then the payload
+/// (if non-empty), then flushes. [`ChunkWriter::write_stdin`],
+/// [`ChunkWriter::write_ctrl_in_json`], and
+/// [`ChunkWriter::write_ctrl_out_json`] are convenience helpers for the
+/// most common write paths.
+pub struct ChunkWriter<W: AsyncWrite + Unpin> {
+    writer: W,
+}
+
+impl<W: AsyncWrite + Unpin> ChunkWriter<W> {
+    /// Wrap an [`AsyncWrite`] transport in a chunk writer.
+    pub fn new(writer: W) -> Self {
+        Self { writer }
+    }
+
+    /// Consume the writer and return the underlying transport.
+    pub fn into_inner(self) -> W {
+        self.writer
+    }
+
+    /// Write a chunk: header + payload (if non-empty) + flush.
+    pub async fn write_chunk(&mut self, chunk: &Chunk) -> Result<(), RawError> {
+        let mut header = [0u8; CHUNK_HEADER_LEN];
+        header[0] = chunk.stream_type;
+        let len = chunk.bytes.len() as u32;
+        header[1..].copy_from_slice(&len.to_be_bytes());
+        self.writer.write_all(&header).await?;
+        if !chunk.bytes.is_empty() {
+            self.writer.write_all(&chunk.bytes).await?;
+        }
+        self.writer.flush().await?;
+        Ok(())
+    }
+
+    /// Write a stdin chunk (stream_type 0) directly from a byte slice.
+    pub async fn write_stdin(&mut self, bytes: &[u8]) -> Result<(), RawError> {
+        let mut header = [0u8; CHUNK_HEADER_LEN];
+        header[0] = STREAM_STDIN;
+        let len = bytes.len() as u32;
+        header[1..].copy_from_slice(&len.to_be_bytes());
+        self.writer.write_all(&header).await?;
+        if !bytes.is_empty() {
+            self.writer.write_all(bytes).await?;
+        }
+        self.writer.flush().await?;
+        Ok(())
+    }
+
+    /// Write a client→server control chunk (stream_type 3) carrying a JSON
+    /// payload (`Resize`, `Signal`, or `Eof`).
+    pub async fn write_ctrl_in_json(&mut self, json: &[u8]) -> Result<(), RawError> {
+        let mut header = [0u8; CHUNK_HEADER_LEN];
+        header[0] = STREAM_CTRL_IN;
+        let len = json.len() as u32;
+        header[1..].copy_from_slice(&len.to_be_bytes());
+        self.writer.write_all(&header).await?;
+        self.writer.write_all(json).await?;
+        self.writer.flush().await?;
+        Ok(())
+    }
+
+    /// Write a server→client control chunk (stream_type 4) carrying a JSON
+    /// payload (`Exit`).
+    pub async fn write_ctrl_out_json(&mut self, json: &[u8]) -> Result<(), RawError> {
+        let mut header = [0u8; CHUNK_HEADER_LEN];
+        header[0] = STREAM_CTRL_OUT;
+        let len = json.len() as u32;
+        header[1..].copy_from_slice(&len.to_be_bytes());
+        self.writer.write_all(&header).await?;
+        self.writer.write_all(json).await?;
+        self.writer.flush().await?;
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    use bytes::Bytes;
+    use tokio::io::{duplex, AsyncWriteExt};
+
+    async fn round_trip(stream_type: u8, payload: &[u8]) {
+        let (mut a, mut b) = duplex(8 * 1024);
+        let mut writer = ChunkWriter::new(&mut a);
+        let mut reader = ChunkReader::new(&mut b);
+
+        let chunk = Chunk {
+            stream_type,
+            bytes: Bytes::copy_from_slice(payload),
+        };
+        writer.write_chunk(&chunk).await.unwrap();
+
+        let read = reader.read_chunk().await.unwrap();
+        assert_eq!(read.stream_type, stream_type);
+        assert_eq!(read.bytes.as_ref(), payload);
+    }
+
+    #[tokio::test]
+    async fn round_trip_stdin() {
+        round_trip(STREAM_STDIN, b"hello stdin").await;
+    }
+
+    #[tokio::test]
+    async fn round_trip_stdout() {
+        round_trip(STREAM_STDOUT, b"hello stdout").await;
+    }
+
+    #[tokio::test]
+    async fn round_trip_stderr() {
+        round_trip(STREAM_STDERR, b"hello stderr").await;
+    }
+
+    #[tokio::test]
+    async fn round_trip_ctrl_in() {
+        round_trip(STREAM_CTRL_IN, br#"{"type":"eof"}"#).await;
+    }
+
+    #[tokio::test]
+    async fn round_trip_ctrl_out() {
+        round_trip(STREAM_CTRL_OUT, br#"{"type":"exit","code":0}"#).await;
+    }
+
+    #[tokio::test]
+    async fn round_trip_empty_payload() {
+        round_trip(STREAM_STDIN, b"").await;
+    }
+
+    #[tokio::test]
+    async fn round_trip_write_stdin_helper() {
+        let (mut a, mut b) = duplex(8 * 1024);
+        let mut writer = ChunkWriter::new(&mut a);
+        let mut reader = ChunkReader::new(&mut b);
+
+        writer.write_stdin(b"piped").await.unwrap();
+        let read = reader.read_chunk().await.unwrap();
+        assert_eq!(read.stream_type, STREAM_STDIN);
+        assert_eq!(read.bytes.as_ref(), b"piped");
+    }
+
+    #[tokio::test]
+    async fn round_trip_write_ctrl_in_json_helper() {
+        let (mut a, mut b) = duplex(8 * 1024);
+        let mut writer = ChunkWriter::new(&mut a);
+        let mut reader = ChunkReader::new(&mut b);
+
+        let json = br#"{"type":"resize","cols":80,"rows":24}"#;
+        writer.write_ctrl_in_json(json).await.unwrap();
+        let read = reader.read_chunk().await.unwrap();
+        assert_eq!(read.stream_type, STREAM_CTRL_IN);
+        assert_eq!(read.bytes.as_ref(), json);
+    }
+
+    #[tokio::test]
+    async fn round_trip_write_ctrl_out_json_helper() {
+        let (mut a, mut b) = duplex(8 * 1024);
+        let mut writer = ChunkWriter::new(&mut a);
+        let mut reader = ChunkReader::new(&mut b);
+
+        let json = br#"{"type":"exit","code":0}"#;
+        writer.write_ctrl_out_json(json).await.unwrap();
+        let read = reader.read_chunk().await.unwrap();
+        assert_eq!(read.stream_type, STREAM_CTRL_OUT);
+        assert_eq!(read.bytes.as_ref(), json);
+    }
+
+    #[tokio::test]
+    async fn invalid_stream_type() {
+        let (mut a, mut b) = duplex(8 * 1024);
+        // 5 is one past the highest valid stream_type (4 = STREAM_CTRL_OUT).
+        a.write_all(&[5u8, 0, 0, 0, 0]).await.unwrap();
+        a.flush().await.unwrap();
+
+        let mut reader = ChunkReader::new(&mut b);
+        let err = reader.read_chunk().await.unwrap_err();
+        assert!(matches!(err, RawError::InvalidStreamType(5)));
+    }
+
+    #[tokio::test]
+    async fn chunk_too_large() {
+        let (mut a, mut b) = duplex(8 * 1024);
+        let over = MAX_CHUNK_LEN + 1;
+        a.write_all(&[0u8]).await.unwrap();
+        a.write_all(&over.to_be_bytes()).await.unwrap();
+        a.flush().await.unwrap();
+
+        let mut reader = ChunkReader::new(&mut b);
+        let err = reader.read_chunk().await.unwrap_err();
+        assert!(matches!(err, RawError::ChunkTooLarge(v) if v == over));
+    }
+
+    #[tokio::test]
+    async fn connection_closed_truncated_header() {
+        let (mut a, mut b) = duplex(8 * 1024);
+        a.write_all(&[0u8, 0]).await.unwrap();
+        a.flush().await.unwrap();
+        a.shutdown().await.unwrap();
+
+        let mut reader = ChunkReader::new(&mut b);
+        let err = reader.read_chunk().await.unwrap_err();
+        assert!(matches!(err, RawError::ConnectionClosed));
+    }
+
+    #[tokio::test]
+    async fn connection_closed_truncated_payload() {
+        let (mut a, mut b) = duplex(8 * 1024);
+        a.write_all(&[0u8, 0, 0, 0, 8]).await.unwrap();
+        a.write_all(b"short").await.unwrap();
+        a.flush().await.unwrap();
+        a.shutdown().await.unwrap();
+
+        let mut reader = ChunkReader::new(&mut b);
+        let err = reader.read_chunk().await.unwrap_err();
+        assert!(matches!(err, RawError::ConnectionClosed));
+    }
+
+    #[tokio::test]
+    async fn connection_closed_clean_close_no_bytes() {
+        let (mut a, mut b) = duplex(8 * 1024);
+        a.shutdown().await.unwrap();
+
+        let mut reader = ChunkReader::new(&mut b);
+        let err = reader.read_chunk().await.unwrap_err();
+        assert!(matches!(err, RawError::ConnectionClosed));
+    }
+}