feat: crate-root re-exports, wire write validation, peek guard; README + CHANGELOG

Review #003 Session 2 (P12 + P6 + P7 + P4 + P16):

- P12: primary types re-exported at the crate root (TtyAdapter,
  TtySession, TtyBackend/TtyHandle/TtyParams, Chunk codec, wire
  constants, ControlMessage, negotiation types, channels helpers;
  LocalTtyBackend under `local`). signal_from_name re-export is
  #[cfg(unix)]-gated to match its definition (wasm check caught it).
- P6: ChunkWriter validates before writing — stream_type > 4 or
  payload > MAX_CHUNK_LEN fails locally (RawError) instead of
  corrupting the peer's framing; length validated as u64 before the
  u32 cast so oversized payloads cannot wrap past the check.
  Empty-payload shape unified via a single write_validated helper.
- P7: ChunkReader tracks peeked state — read_chunk() after
  peek_stream_type() completes the peeked chunk instead of consuming
  a second header byte; second peek is idempotent;
  read_chunk_after_peek debug-asserts a peek. Session read pump drops
  its manual first_byte_peeked threading.
- P4: README.md (alkcall structure; quick-start examples per role
  using the new root paths) + readme = "README.md" in [package].
- P16: CHANGELOG.md with the full [0.1.0] entry and tag link.

Verification: cargo test 104 lib (default) / 147 (--all-features);
clippy all-targets + wasm32 -D warnings; fmt; wasm check; doc 0
warnings; publish dry-run 44 files OK.

CHANGELOG.md

@@ -0,0 +1,124 @@
+# Changelog
+
+All notable changes to this crate are documented here. The format is
+based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and
+this crate adheres to [Semantic Versioning](https://semver.org/).
+
+## [0.1.0] - 2026-09-05
+
+Initial crates.io release: the `alk/tty` terminal-session protocol —
+producer/consumer protocol crate on top of alkcall channels, ported and
+consolidated from the `alknet-tty` + `alknet-tty-local` crates of the
+alknet mono-repo.
+
+### Added
+
+- **Wire format (ADR-001).** Two-carriage protocol: a 4-byte
+  big-endian length-prefixed JSON negotiation frame (`NegotiateRequest`
+  — `carriage`/`backend`/`tty`/`cmd`/`cwd`/`env` plus opaque
+  `serde(flatten)` backend params), then raw chunks
+  (`[stream_type: u8][length: u32 be][payload]`) with five stream types
+  (`STREAM_STDIN`=0, `STREAM_STDOUT`=1, `STREAM_STDERR`=2,
+  `STREAM_CTRL_IN`=3, `STREAM_CTRL_OUT`=4). Zero-length data chunks are
+  sentinels (zero-length stdin = client EOF; zero-length stdout =
+  server drained); control chunks are never zero-length. Payloads are
+  capped at 16 MiB (`MAX_CHUNK_LEN`) — this keeps the length prefix's
+  high byte `0x00`, which is the disambiguation invariant between an
+  error frame and a raw chunk (ADR-001 §5). The framing is
+  self-contained (ADR-006): not reused from alkcall's `EventEnvelope`
+  framing. The control channel is split into `STREAM_CTRL_IN`/`STREAM_CTRL_OUT`
+  so it is genuinely bidirectional on the wire (the Phase 7 amendment
+  inside ADR-001). TTY always uses its own 5-byte chunk format inside
+  channels — the channels layer carries it transparently in the channel
+  payload (ADR-008, reversing ADR-007). The machine-readable contract
+  is the BAST document (`docs/architecture/tty-bast.md`); a unit test
+  guards the BAST's `StreamType` enum against drift from the wire
+  constants.
+- **`TtyBackend` trait (ADR-002).** The inversion point between the
+  wire-format adapter and backend crates. `allocate(&TtyParams) ->
+  TtyHandle` (stdin `AsyncWrite`, stdout/stderr `Stream<Item = Bytes>`,
+  `exit_code: BoxFuture`, `TtyControlHandle` for resize/signal);
+  `resource_id(&TtyParams) -> Option<(kind, id)>` feeds the ADR-050
+  ownership check. `TtyError` is `#[non_exhaustive]`. The trait shape
+  is a one-way door once backends exist.
+- **Local backend behind the `local` feature (ADR-003).** The
+  `alknet-tty-local` crate folded in as a feature-gated module:
+  PTY mode via `portable_pty` (real terminal semantics — resize,
+  process-group signal forwarding, merged stdout/stderr) and pipe mode
+  via `tokio::process::Command` (the runner case — separate
+  stdout/stderr, no-op resize, pid-only signal), selected by
+  `TtyParams::terminal`. The blocking→async bridge uses three dedicated
+  std threads feeding tokio mpsc/oneshot channels. Non-wasm by design;
+  the default crate (no features) stays `wasm32-unknown-unknown`-clean.
+- **`TtyAdapter` producer (direct `alk/tty` ALPN).** A
+  `ProtocolHandler` holding a `HashMap<String, Arc<dyn TtyBackend>>`
+  keyed by the negotiation frame's `backend` string. Accepts a
+  connection, loops `accept_bi`, and dispatches each bidi stream to a
+  session. Per-stream flow (ADR-001/002/005): parse + validate the
+  negotiation frame (`carriage == "raw"`, non-empty `cmd`), scope-gate
+  via `tty:open`, optional `OwnershipProvider` resource check,
+  `backend.allocate()`, then three concurrent pumps (stdin→backend,
+  stdout/stderr→client, exit→exit chunk) with the **exit-chunk-is-last**
+  invariant — `{"type":"exit","code":N}` (`-1` on wait failure, ADR-004;
+  negative = signal-terminated) is enqueued only after both stream
+  pumps complete and `exit_code` resolves. A concurrent drainer task
+  owns the client write half, so a burst of >64 chunks from the backend
+  cannot deadlock the session (the drainer starts before the pumps
+  join; the single FIFO preserves exit-chunk-last). On session cancel
+  the `TtyHandle` drops without driving `exit_code` to completion,
+  which triggers the backend's kill-on-`Drop` guard — the adapter has
+  no separate kill path (the session-cancel contract, ADR-005: dropping
+  the future kills the session target).
+- **`channels` integration (ADR-008, ADR-009).** `register_openable` +
+  `tty_open_spec` register the `channels/tty/sub` open op on a
+  per-connection `OperationRegistry` (per ADR-047's per-connection
+  amendment): the registry's `AccessControl` carries the `tty:open`
+  scope gate, the op's input schema validates the shared
+  `NegotiateRequest` fields (`carriage`/`backend`/`cmd` required;
+  alkcall 0.4 enforces at dispatch), and the `OpenHandler` spawns the
+  session driver pre-negotiated — **the open op's params ARE the
+  negotiation** (ADR-009), no second frame on the channel stream.
+  Access control is enforced by the registry before the handler runs;
+  the handler validates the backend exists and drives the session.
+- **`TtySession` consumer.** The typed client handle with two
+  constructors — `connect_direct` (direct `alk/tty` connection:
+  writes the negotiation frame, then typed methods) and
+  `open_via_channels` (opens `channels/tty/sub` on a `ChannelClient`,
+  adopts the channel, starts in raw-chunk mode). Typed methods:
+  `send_stdin`/`close_stdin` (stdin chunks / EOF sentinel),
+  `recv_stdout`/`recv_stderr` (`Stream<Item = Bytes>`), `resize`,
+  `signal`, and `wait` (awaits the `Exit` control chunk; watch-based,
+  no lost wakeup). Both constructors disambiguate the first response
+  frame (`0x00` prefix = negotiation error frame → `NegotiationRejected`
+  with the server's error code; raw chunk = session proceeds).
+  `TtySessionError` is `#[non_exhaustive]`. Dropping the session aborts
+  the read pump and closes the write half.
+- **Control messages.** `ControlMessage` (`Resize`/`Signal`/`Eof` on
+  `ctrl_in`; `Exit` on `ctrl_out`) — JSON, tagged by `"type"`, unknown
+  types ignored by policy. `signal_from_name` maps the common signal
+  set (`HUP`/`INT`/`QUIT`/`TERM`/`KILL`/`USR1`/`USR2`/`TSTP`/`CONT`)
+  to libc numbers (unix).
+- **Crate-root re-exports.** The primary types (`TtyAdapter`,
+  `TtySession`, `TtyBackend`, `TtyHandle`, `TtyParams`, `Chunk`,
+  `NegotiateRequest`, `ControlMessage`, the wire constants, and
+  `LocalTtyBackend` under `local`) are re-exported at the crate root;
+  the module paths remain the full surface.
+- **Wire write-path validation.** All `ChunkWriter` methods validate
+  before touching the transport — a `stream_type` > 4 or a payload over
+  `MAX_CHUNK_LEN` fails locally instead of writing a header the peer
+  cannot frame (the length is validated before the `u32` cast, so an
+  oversized payload cannot truncate past the check and corrupt the
+  peer's framing). Empty-payload writes are one shape: `length == 0`
+  with no payload bytes.
+- **Peek-safe `ChunkReader`.** The reader tracks peeked state:
+  `read_chunk()` after `peek_stream_type()` completes the peeked chunk
+  instead of consuming a second header byte (which silently
+  desynchronized framing), and a second peek returns the peeked byte
+  without reading. `read_chunk_after_peek` remains the explicit form.
+- **Backpressure regression test.** >64-chunk sessions are driven
+  through the real adapter with a per-read timeout so a deadlock
+  regression fails the test instead of hanging it, plus exact-boundary
+  (`MAX_CHUNK_LEN`) round-trip and server-sends-client-direction-
+  stream-types discard tests.
+
+[0.1.0]: https://git.alk.dev/alkdev/alktty/releases/tag/v0.1.0

Cargo.toml

@@ -6,6 +6,7 @@ rust-version = "1.85"
 license = "MIT OR Apache-2.0"
 description = "Terminal session protocol: wire format, TtyBackend trait, TtyAdapter, and typed consumer client. Producer/consumer protocol crate on top of alkcall channels."
 repository = "https://git.alk.dev/alkdev/alktty"
+readme = "README.md"
 keywords = ["tty", "terminal", "pty", "channels", "alkcall"]
 categories = ["network-programming", "asynchronous"]
 exclude = [".opencode/", "docs/reviews/", "docs/research/", "docs/sdd_process.md", "Cargo.lock"]

README.md

@@ -0,0 +1,145 @@
+# alktty
+
+Terminal session protocol for the `alk/tty` ALPN: a JSON negotiation
+frame followed by raw chunks (`[stream_type: u8][length: u32
+be][payload]`), a backend-agnostic `TtyBackend` trait, a
+backend-agnostic producer adapter, and a typed consumer client.
+
+A producer/consumer protocol crate on top of
+[alkcall](https://crates.io/crates/alkcall) channels — no networking, no
+transport dependencies. Downstream crates (a `alknet-docker` or
+`alknet-ssh` backend, a sandboxed TS/Python adapter) compose on top of
+it: implement [`TtyBackend`](https://docs.rs/alktty/latest/alktty/trait.TtyBackend.html)
+for your execution environment, and the adapter handles the wire.
+
+## Quick start
+
+### Producer — register backends and run the adapter
+
+```rust
+use std::collections::HashMap;
+use std::sync::Arc;
+
+// `LocalTtyBackend` needs the `local` feature; swap in any `TtyBackend`
+// implementation (docker, ssh, ...) compiled for your target.
+let mut backends: HashMap<String, Arc<dyn alktty::TtyBackend>> = HashMap::new();
+backends.insert("local".into(), Arc::new(alktty::LocalTtyBackend::new()));
+
+let tty_adapter = alktty::TtyAdapter::new(backends);
+// tty_adapter implements alkcall::core::ProtocolHandler on the `alk/tty`
+// ALPN — hand it to your transport accept loop:
+//   tty_adapter.handle(connection, &auth).await
+```
+
+### Producer — via `alk/channels` instead of a dedicated ALPN
+
+```rust
+use std::collections::HashMap;
+use std::sync::Arc;
+
+let backends: HashMap<String, Arc<dyn alktty::TtyBackend>> =
+    std::iter::once(("local".to_string(), Arc::new(alktty::LocalTtyBackend::new()) as _))
+        .collect();
+
+// In your channels acceptor's `install_channel_zero` hook (per-connection):
+//   alktty::register_openable(&channel_core, Arc::new(backends), None, &mut registry, auth)
+// Consumers open `channels/tty/sub`; the open op's params ARE the
+// negotiation (ADR-009) — no second frame on the channel stream.
+```
+
+### Consumer — open a session over the direct ALPN
+
+```rust
+use alkcall::core::Connection;
+use alktty::{NegotiateRequest, TtySession};
+
+let connection: Connection = /* dial the transport, negotiate `alk/tty` */;
+let session = TtySession::connect_direct(
+    connection,
+    NegotiateRequest {
+        carriage: "raw".into(),
+        backend: "local".into(),
+        tty: None,
+        cmd: vec!["bash".into()],
+        cwd: None,
+        env: Default::default(),
+        backend_params: Default::default(),
+    },
+).await?;
+
+session.resize(120, 40, 0, 0).await?;
+session.send_stdin("ls -l\n".into()).await?;
+session.close_stdin().await?;
+
+let mut stdout = session.recv_stdout().await;
+while let Some(bytes) = futures::StreamExt::next(&mut stdout).await {
+    if bytes.is_empty() { break; } // zero-length stdout sentinel = drained
+    // ... render bytes ...
+}
+let code = session.wait().await?;
+```
+
+### Consumer — open a session via `alk/channels`
+
+```rust
+use alkcall::channels::client::ChannelClient;
+use alktty::TtySession;
+
+let client: ChannelClient = /* from_connection on an `alk/channels` connection */;
+let session = TtySession::open_via_channels(
+    &client,
+    serde_json::json!({
+        "carriage": "raw",
+        "backend": "local",
+        "cmd": ["bash"],
+        "tty": {"term": "xterm-256color", "cols": 120, "rows": 40,
+                "pixel_width": 0, "pixel_height": 0}
+    }),
+).await?;
+```
+
+## Architecture
+
+Two roles, symmetric per the alk convention (avoid "server"/"client":
+both sides of a connection can be either):
+
+| Role | Direct `alk/tty` | Via `alk/channels` |
+|------|------------------|--------------------|
+| **Producer** (runs backends, emits terminal bytes) | [`TtyAdapter`](https://docs.rs/alktty/latest/alktty/adapter/struct.TtyAdapter.html) as a `ProtocolHandler` — negotiation frame → access control → backend allocation → three-pump session driver | [`register_openable`](https://docs.rs/alktty/latest/alktty/channels/fn.register_openable.html) — the registry's ACL gates the open op; the handler spawns the same session driver pre-negotiated |
+| **Consumer** (types input, renders output) | [`TtySession::connect_direct`](https://docs.rs/alktty/latest/alktty/struct.TtySession.html#method.connect_direct) — writes the negotiation frame, exposes typed stdin/stdout/stderr/control/exit methods | [`TtySession::open_via_channels`](https://docs.rs/alktty/latest/alktty/struct.TtySession.html#method.open_via_channels) — the open op's params are the negotiation |
+
+One connection hosts many sessions (one per bidi stream on the direct
+path, one per channel on the channels path); sessions are independent.
+
+The wire format (ADR-001) is two carriages: a 4-byte
+length-prefixed JSON negotiation frame, then raw chunks —
+`[stream_type: u8][length: u32 be][payload]` with five stream types
+(`STREAM_STDIN`=0 … `STREAM_CTRL_OUT`=4; see
+[`alktty::wire`](https://docs.rs/alktty/latest/alktty/wire/index.html)).
+Zero-length data chunks are sentinels (stdin EOF from the client,
+stdout drained from the server); the exit code rides as the final
+`ctrl_out` control chunk (ADR-005). The codec is
+[`alktty::wire`](https://docs.rs/alktty/latest/alktty/wire/index.html)'s
+`ChunkReader`/`ChunkWriter`; the machine-readable contract is the BAST
+document at `docs/architecture/tty-bast.md`.
+
+### WASM target
+
+The default crate (no features) compiles to `wasm32-unknown-unknown` —
+this is what makes the downstream TS/Python adapter story work: a
+wasm-compiled alktty is the protocol layer for a sandboxed adapter. The
+`local` feature (the [`LocalTtyBackend`](https://docs.rs/alktty/latest/alktty/struct.LocalTtyBackend.html)
+reference implementation over `portable_pty` + `tokio::process`) needs a
+real OS and is non-wasm by design.
+
+## Documentation
+
+- [Architecture docs](docs/architecture/README.md) — the authoritative
+  spec: ADRs (001–009), the wire format, the backend trait contract,
+  and the local backend's cancel-cleanup semantics.
+- [API docs](https://docs.rs/alktty) — full crate documentation on
+  docs.rs.
+
+## License
+
+MIT OR Apache-2.0

src/lib.rs

@@ -40,11 +40,25 @@
 //! let mut backends = std::collections::HashMap::new();
 //! backends.insert(
 //!     "local".into(),
-//!     std::sync::Arc::new(alktty::local::LocalTtyBackend::new())
+//!     std::sync::Arc::new(alktty::LocalTtyBackend::new())
-//!         as std::sync::Arc<dyn alktty::backend::TtyBackend>,
+//!         as std::sync::Arc<dyn alktty::TtyBackend>,
 //! );
-//! let tty_adapter = alktty::adapter::TtyAdapter::new(backends);
+//! let tty_adapter = alktty::TtyAdapter::new(backends);
 //! ```
+//!
+//! # Crate-root surface
+//!
+//! The primary types are re-exported at the crate root —
+//! [`TtyAdapter`]/[`drive_session`] (producer), [`TtySession`]
+//! (consumer), [`TtyBackend`]/[`TtyHandle`]/[`TtyParams`] (backend
+//! authors), plus the wire codec ([`Chunk`], [`ChunkReader`],
+//! [`ChunkWriter`]), the wire constants ([`STREAM_STDOUT`],
+//! [`MAX_CHUNK_LEN`], ...), [`ControlMessage`], the negotiation types
+//! [`NegotiateRequest`], ...), the channels registration helpers
+//! ([`register_openable`]), and `local::LocalTtyBackend` under the
+//! `local` feature. The module paths below remain the full public surface —
+//! the root re-exports are the ergonomic short paths for the common
+//! imports.
 
 pub mod adapter;
 pub mod backend;
@@ -59,3 +73,25 @@ pub mod local;
 
 #[cfg(test)]
 pub(crate) mod testing;
+
+pub use adapter::{drive_session, drive_session_pre_negotiated, TtyAdapter, TTY_OPEN_SCOPE};
+pub use backend::{
+    BoxFuture, TerminalParams, TtyBackend, TtyControl, TtyControlHandle, TtyError, TtyHandle,
+    TtyParams,
+};
+pub use channels::{register_openable, tty_open_spec, OP_TTY_OPEN, TTY_ALPN};
+#[cfg(unix)]
+pub use control::signal_from_name;
+pub use control::ControlMessage;
+pub use negotiation::{
+    error_response_bytes, NegotiateRequest, NegotiationError, NegotiationReader, NegotiationWriter,
+    TerminalParamsWire,
+};
+pub use session::{TtySession, TtySessionError};
+pub use wire::{
+    Chunk, ChunkReader, ChunkWriter, RawError, CHUNK_HEADER_LEN, MAX_CHUNK_LEN, STREAM_CTRL_IN,
+    STREAM_CTRL_OUT, STREAM_STDERR, STREAM_STDIN, STREAM_STDOUT,
+};
+
+#[cfg(feature = "local")]
+pub use local::LocalTtyBackend;

src/session.rs

@@ -292,13 +292,13 @@ impl TtySession {
         // `{1, 2, 4}` (the server never sends `0` or `3`). If the server
         // rejected the negotiation, read the error frame and return
         // `NegotiationRejected`; otherwise hand the peeked reader to the
-        // read pump.
+        // read pump — `ChunkReader` tracks the peeked byte, so the pump's
-        let mut first_byte_peeked = false;
+        // `read_chunk()` completes that first chunk.
         match reader.peek_stream_type().await {
             Ok(0x00) => {
                 return Err(read_negotiation_error(reader.into_inner()).await);
             }
-            Ok(_) => first_byte_peeked = true,
+            Ok(_) => {}
             Err(RawError::ConnectionClosed) => {
                 // The server closed cleanly without a response. Fall
                 // through to the read pump, which resolves `NoExitChunk`.
@@ -306,7 +306,7 @@ impl TtySession {
             Err(e) => return Err(TtySessionError::Wire(e)),
         }
 
-        Self::start_pump(writer, reader, first_byte_peeked)
+        Self::start_pump(writer, reader)
     }
 
     /// Core inner for the channels path (ADR-009): the negotiation
@@ -323,12 +323,11 @@ impl TtySession {
     {
         let writer = ChunkWriter::new(Box::new(write) as Box<dyn AsyncWrite + Send + Unpin>);
         let mut reader = ChunkReader::new(read);
-        let mut first_byte_peeked = false;
         match reader.peek_stream_type().await {
             Ok(0x00) => {
                 return Err(read_negotiation_error(reader.into_inner()).await);
             }
-            Ok(_) => first_byte_peeked = true,
+            Ok(_) => {}
             Err(RawError::ConnectionClosed) => {
                 // The server closed cleanly without a response. Fall
                 // through to the read pump, which resolves `NoExitChunk`.
@@ -336,15 +335,17 @@ impl TtySession {
             Err(e) => return Err(TtySessionError::Wire(e)),
         }
 
-        Self::start_pump(writer, reader, first_byte_peeked)
+        Self::start_pump(writer, reader)
     }
 
     /// Wire up the exit watch + stdout/stderr channels and spawn the
-    /// read pump. Shared by `from_halves` and `from_halves_raw`.
+    /// read pump. Shared by `from_halves` and `from_halves_raw`. The
+    /// reader may arrive with a peeked first byte (the negotiation
+    /// disambiguation peek); `ChunkReader` tracks that state, so the
+    /// pump reads every chunk — including the first — via `read_chunk`.
     fn start_pump<R>(
         writer: ChunkWriter<Box<dyn AsyncWrite + Send + Unpin>>,
         reader: ChunkReader<R>,
-        first_byte_peeked: bool,
     ) -> Result<Self, TtySessionError>
     where
         R: AsyncRead + Send + Unpin + 'static,
@@ -353,13 +354,7 @@ impl TtySession {
         let (stderr_tx, stderr_rx) = mpsc::channel::<Bytes>(64);
         let (exit_tx, exit_rx) = tokio::sync::watch::channel::<Option<ExitOutcome>>(None);
 
-        let read_pump = tokio::spawn(read_pump(
+        let read_pump = tokio::spawn(read_pump(reader, stdout_tx, stderr_tx, exit_tx));
-            reader,
-            first_byte_peeked,
-            stdout_tx,
-            stderr_tx,
-            exit_tx,
-        ));
 
         Ok(Self {
             writer: Mutex::new(writer),
@@ -574,7 +569,6 @@ where
 ///   debug log)
 async fn read_pump<R>(
     mut reader: ChunkReader<R>,
-    mut first_byte_peeked: bool,
     stdout_tx: mpsc::Sender<Bytes>,
     stderr_tx: mpsc::Sender<Bytes>,
     exit_tx: tokio::sync::watch::Sender<Option<ExitOutcome>>,
@@ -583,12 +577,7 @@ async fn read_pump<R>(
 {
     let mut exit_resolved = false;
     loop {
-        let read = if first_byte_peeked {
+        let read = reader.read_chunk().await;
-            first_byte_peeked = false;
-            reader.read_chunk_after_peek().await
-        } else {
-            reader.read_chunk().await
-        };
         match read {
             Ok(chunk) => match chunk.stream_type {
                 crate::wire::STREAM_STDOUT => {

src/wire.rs

@@ -143,9 +143,19 @@ impl Chunk {
 /// 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.
+///
+/// The reader tracks peeked state: after
+/// [`ChunkReader::peek_stream_type`], [`ChunkReader::read_chunk`]
+/// completes the already-peeked chunk (the peeked byte is the header's
+/// first `stream_type` byte) instead of consuming a second header byte,
+/// and a second `peek_stream_type` returns the peeked byte without
+/// reading. Both call orders are safe; [`ChunkReader::read_chunk_after_peek`]
+/// remains the explicit completion form (`debug_assert!`s that a peek
+/// happened).
 pub struct ChunkReader<R: AsyncRead + Unpin> {
     reader: R,
     header: [u8; CHUNK_HEADER_LEN],
+    peeked: bool,
 }
 
 impl<R: AsyncRead + Unpin> ChunkReader<R> {
@@ -154,6 +164,7 @@ impl<R: AsyncRead + Unpin> ChunkReader<R> {
         Self {
             reader,
             header: [0u8; CHUNK_HEADER_LEN],
+            peeked: false,
         }
     }
 
@@ -163,8 +174,15 @@ impl<R: AsyncRead + Unpin> ChunkReader<R> {
     }
 
     /// Read one chunk: header, validate, payload.
+    ///
+    /// Composes with [`ChunkReader::peek_stream_type`]: if a byte was
+    /// already peeked, this completes that chunk (the peeked byte is
+    /// consumed as the header's `stream_type`); otherwise it reads the
+    /// full 5-byte header. Either order yields exactly one chunk.
     pub async fn read_chunk(&mut self) -> Result<Chunk, RawError> {
-        self.peek_stream_type().await?;
+        if !self.peeked {
+            self.peek_stream_type().await?;
+        }
         self.read_chunk_after_peek().await
     }
 
@@ -175,10 +193,20 @@ impl<R: AsyncRead + Unpin> ChunkReader<R> {
     /// `stream_type` in `{1, 2, 4}` — the server never sends `0` or `3`).
     /// After peeking, the caller either reads the error frame itself (via
     /// [`ChunkReader::into_inner`]) or completes the chunk with
-    /// [`ChunkReader::read_chunk_after_peek`].
+    /// [`ChunkReader::read_chunk_after_peek`] or
+    /// [`ChunkReader::read_chunk`].
+    ///
+    /// Idempotent: if a byte was already peeked, returns it without
+    /// reading another.
     pub async fn peek_stream_type(&mut self) -> Result<u8, RawError> {
+        if self.peeked {
+            return Ok(self.header[0]);
+        }
         match self.reader.read_exact(&mut self.header[..1]).await {
-            Ok(_) => Ok(self.header[0]),
+            Ok(_) => {
+                self.peeked = true;
+                Ok(self.header[0])
+            }
             Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => Err(RawError::ConnectionClosed),
             Err(e) => Err(RawError::Io(e)),
         }
@@ -187,7 +215,15 @@ impl<R: AsyncRead + Unpin> ChunkReader<R> {
     /// Complete reading a chunk whose first byte was already read via
     /// [`ChunkReader::peek_stream_type`]. Reads the remaining 4 header
     /// bytes, validates, and reads the payload.
+    ///
+    /// Must be called after a peek (debug-asserted); for peek-agnostic
+    /// reading use [`ChunkReader::read_chunk`], which handles both orders.
     pub async fn read_chunk_after_peek(&mut self) -> Result<Chunk, RawError> {
+        debug_assert!(
+            self.peeked,
+            "read_chunk_after_peek called without peek_stream_type; use read_chunk"
+        );
+        self.peeked = false;
         match self.reader.read_exact(&mut self.header[1..]).await {
             Ok(_) => {}
             Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => {
@@ -197,19 +233,13 @@ impl<R: AsyncRead + Unpin> ChunkReader<R> {
         }
 
         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 {
+        validate_header(stream_type, length as u64)?;
-            return Err(RawError::ChunkTooLarge(length));
-        }
 
         let mut buf = vec![0u8; length as usize];
         if length > 0 {
@@ -229,6 +259,25 @@ impl<R: AsyncRead + Unpin> ChunkReader<R> {
     }
 }
 
+/// Shared header invariant for both codec directions: a `stream_type` in
+/// `{0..=4}` and a payload length ≤ [`MAX_CHUNK_LEN`] (which keeps the
+/// length prefix's high byte `0x00` — the negotiation-disambiguation
+/// invariant, ADR-052 §5). The read path validates peer input (length is
+/// already a `u32` from the header); the write paths validate caller
+/// input *before* the length is cast to `u32` — validating the truncated
+/// value would let a ≥2³² payload wrap past the check and corrupt the
+/// peer's framing, which is exactly the bug this validation exists to
+/// catch.
+fn validate_header(stream_type: u8, length: u64) -> Result<(), RawError> {
+    if stream_type > 4 {
+        return Err(RawError::InvalidStreamType(stream_type));
+    }
+    if length > MAX_CHUNK_LEN as u64 {
+        return Err(RawError::ChunkTooLarge(length as u32));
+    }
+    Ok(())
+}
+
 /// Writes raw chunks to an [`AsyncWrite`] transport.
 ///
 /// [`ChunkWriter::write_chunk`] writes the 5-byte header then the payload
@@ -236,6 +285,14 @@ impl<R: AsyncRead + Unpin> ChunkReader<R> {
 /// [`ChunkWriter::write_ctrl_in_json`], and
 /// [`ChunkWriter::write_ctrl_out_json`] are convenience helpers for the
 /// most common write paths.
+///
+/// Every write validates before touching the transport: a `stream_type`
+/// above 4 or a payload longer than [`MAX_CHUNK_LEN`] returns
+/// [`RawError::InvalidStreamType`] / [`RawError::ChunkTooLarge`] instead
+/// of writing a header the peer cannot frame (a truncated `u32` length
+/// would corrupt the stream). An empty payload writes the same header as
+/// a zero-length payload — every helper takes `&[u8]`-shaped input and
+/// emits `length == 0` with no payload bytes.
 pub struct ChunkWriter<W: AsyncWrite + Unpin> {
     writer: W,
 }
@@ -251,56 +308,48 @@ impl<W: AsyncWrite + Unpin> ChunkWriter<W> {
         self.writer
     }
 
-    /// Write a chunk: header + payload (if non-empty) + flush.
+    /// Write a chunk: header + payload (if non-empty) + flush. The
+    /// chunk's `stream_type` and payload length are validated first —
+    /// an invalid chunk fails locally with [`RawError::InvalidStreamType`]
+    /// or [`RawError::ChunkTooLarge`] instead of corrupting the peer's
+    /// framing.
     pub async fn write_chunk(&mut self, chunk: &Chunk) -> Result<(), RawError> {
-        let mut header = [0u8; CHUNK_HEADER_LEN];
+        validate_header(chunk.stream_type, chunk.bytes.len() as u64)?;
-        header[0] = chunk.stream_type;
+        self.write_validated(chunk.stream_type, &chunk.bytes).await
-        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.
+    /// An empty slice writes the zero-length EOF sentinel.
     pub async fn write_stdin(&mut self, bytes: &[u8]) -> Result<(), RawError> {
-        let mut header = [0u8; CHUNK_HEADER_LEN];
+        validate_header(STREAM_STDIN, bytes.len() as u64)?;
-        header[0] = STREAM_STDIN;
+        self.write_validated(STREAM_STDIN, bytes).await
-        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];
+        validate_header(STREAM_CTRL_IN, json.len() as u64)?;
-        header[0] = STREAM_CTRL_IN;
+        self.write_validated(STREAM_CTRL_IN, json).await
-        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> {
+        validate_header(STREAM_CTRL_OUT, json.len() as u64)?;
+        self.write_validated(STREAM_CTRL_OUT, json).await
+    }
+
+    /// Single write shape shared by all four write paths: header, then
+    /// payload if non-empty (length 0 ≡ no payload bytes on the wire),
+    /// then flush. Callers have validated the header already.
+    async fn write_validated(&mut self, stream_type: u8, payload: &[u8]) -> Result<(), RawError> {
         let mut header = [0u8; CHUNK_HEADER_LEN];
-        header[0] = STREAM_CTRL_OUT;
+        header[0] = stream_type;
-        let len = json.len() as u32;
+        header[1..].copy_from_slice(&(payload.len() as u32).to_be_bytes());
-        header[1..].copy_from_slice(&len.to_be_bytes());
         self.writer.write_all(&header).await?;
-        self.writer.write_all(json).await?;
+        if !payload.is_empty() {
+            self.writer.write_all(payload).await?;
+        }
         self.writer.flush().await?;
         Ok(())
     }
@@ -409,6 +458,149 @@ mod tests {
         assert!(matches!(err, RawError::InvalidStreamType(5)));
     }
 
+    /// Write-side validation: a `Chunk` with stream_type > 4 is rejected
+    /// locally (nothing written, transport untouched) instead of
+    /// producing a header the peer cannot frame.
+    #[tokio::test]
+    async fn write_chunk_rejects_invalid_stream_type() {
+        let (mut a, mut b) = duplex(8 * 1024);
+        {
+            let mut writer = ChunkWriter::new(&mut a);
+            let chunk = Chunk {
+                stream_type: 5,
+                bytes: Bytes::from_static(b"nope"),
+            };
+            let err = writer.write_chunk(&chunk).await.unwrap_err();
+            assert!(matches!(err, RawError::InvalidStreamType(5)));
+        }
+        a.shutdown().await.unwrap();
+
+        let mut reader = ChunkReader::new(&mut b);
+        assert!(matches!(
+            reader.peek_stream_type().await.unwrap_err(),
+            RawError::ConnectionClosed
+        ));
+    }
+
+    /// Write-side validation: a payload over `MAX_CHUNK_LEN` fails with
+    /// `ChunkTooLarge` before any header is written — previously the
+    /// `as u32` cast would silently write a truncated length and
+    /// desynchronize the peer.
+    #[tokio::test]
+    async fn write_chunk_rejects_oversized_payload() {
+        let (mut a, mut b) = duplex(8 * 1024);
+        {
+            let mut writer = ChunkWriter::new(&mut a);
+            let payload = vec![0u8; MAX_CHUNK_LEN as usize + 1];
+            let chunk = Chunk {
+                stream_type: STREAM_STDOUT,
+                bytes: Bytes::from(payload),
+            };
+            let err = writer.write_chunk(&chunk).await.unwrap_err();
+            assert!(
+                matches!(err, RawError::ChunkTooLarge(v) if v as u64 == MAX_CHUNK_LEN as u64 + 1)
+            );
+        }
+        a.shutdown().await.unwrap();
+
+        let mut reader = ChunkReader::new(&mut b);
+        assert!(matches!(
+            reader.peek_stream_type().await.unwrap_err(),
+            RawError::ConnectionClosed
+        ));
+    }
+
+    /// Write-side validation on the convenience helpers: same invariants
+    /// as `write_chunk` (nothing is written on a rejected write).
+    #[tokio::test]
+    async fn write_helpers_reject_oversized_payload() {
+        let (mut a, mut b) = duplex(8 * 1024);
+        {
+            let mut writer = ChunkWriter::new(&mut a);
+            let payload = vec![0u8; MAX_CHUNK_LEN as usize + 1];
+
+            let err = writer.write_stdin(&payload).await.unwrap_err();
+            assert!(matches!(err, RawError::ChunkTooLarge(_)));
+            let err = writer.write_ctrl_in_json(&payload).await.unwrap_err();
+            assert!(matches!(err, RawError::ChunkTooLarge(_)));
+            let err = writer.write_ctrl_out_json(&payload).await.unwrap_err();
+            assert!(matches!(err, RawError::ChunkTooLarge(_)));
+        }
+        a.shutdown().await.unwrap();
+
+        let mut reader = ChunkReader::new(&mut b);
+        assert!(matches!(
+            reader.peek_stream_type().await.unwrap_err(),
+            RawError::ConnectionClosed
+        ));
+    }
+
+    /// Write-side validation accepts the exact boundary: a
+    /// `MAX_CHUNK_LEN` payload writes and round-trips.
+    #[tokio::test]
+    async fn write_stdin_accepts_exact_max_len() {
+        let (a, b) = duplex(MAX_CHUNK_LEN as usize + CHUNK_HEADER_LEN + 64);
+        let mut writer = ChunkWriter::new(a);
+        let mut reader = ChunkReader::new(b);
+
+        let payload = vec![0x5Au8; MAX_CHUNK_LEN as usize];
+        writer
+            .write_stdin(&payload)
+            .await
+            .expect("boundary write accepted");
+        let read = reader.read_chunk().await.expect("boundary chunk readable");
+        assert_eq!(read.stream_type, STREAM_STDIN);
+        assert_eq!(read.bytes.as_ref(), payload.as_slice());
+    }
+
+    /// P7 regression: after `peek_stream_type`, plain `read_chunk` must
+    /// complete the already-peeked chunk — not consume a second header
+    /// byte and desynchronize the stream. The old shape silently read
+    /// the peeked byte's `length` bytes from the *payload* as a header.
+    #[tokio::test]
+    async fn read_chunk_after_peek_completes_the_peeked_chunk() {
+        let (mut a, mut b) = duplex(8 * 1024);
+        a.write_all(&[STREAM_STDOUT, 0, 0, 0, 5]).await.unwrap();
+        a.write_all(b"hello").await.unwrap();
+        a.write_all(&[STREAM_STDERR, 0, 0, 0, 2]).await.unwrap();
+        a.write_all(b"ok").await.unwrap();
+        a.flush().await.unwrap();
+
+        let mut reader = ChunkReader::new(&mut b);
+        let peeked = reader.peek_stream_type().await.unwrap();
+        assert_eq!(peeked, STREAM_STDOUT);
+
+        let first = reader
+            .read_chunk()
+            .await
+            .expect("first chunk via read_chunk");
+        assert_eq!(first.stream_type, STREAM_STDOUT);
+        assert_eq!(first.bytes.as_ref(), b"hello");
+
+        let second = reader.read_chunk().await.expect("second chunk");
+        assert_eq!(second.stream_type, STREAM_STDERR);
+        assert_eq!(second.bytes.as_ref(), b"ok");
+    }
+
+    /// `peek_stream_type` is idempotent: a second peek returns the
+    /// peeked byte without reading another byte from the transport.
+    #[tokio::test]
+    async fn peek_stream_type_is_idempotent() {
+        let (mut a, mut b) = duplex(8 * 1024);
+        a.write_all(&[STREAM_STDOUT, 0, 0, 0, 5]).await.unwrap();
+        a.write_all(b"hello").await.unwrap();
+        a.flush().await.unwrap();
+
+        let mut reader = ChunkReader::new(&mut b);
+        let first = reader.peek_stream_type().await.unwrap();
+        let second = reader.peek_stream_type().await.unwrap();
+        assert_eq!(first, second);
+
+        let chunk = reader.read_chunk().await.expect("chunk completes");
+        assert_eq!(chunk.stream_type, STREAM_STDOUT);
+        assert_eq!(chunk.bytes.as_ref(), b"hello");
+    }
+
     #[tokio::test]
     async fn chunk_too_large() {
         let (mut a, mut b) = duplex(8 * 1024);