diff --git a/AGENTS.md b/AGENTS.md index b792b9b..a3c8439 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -213,11 +213,16 @@ implementation agents. modules, `PascalCase` for types/traits, `SCREAMING_SNAKE_CASE` for constants (`STREAM_STDIN`, `TTY_OPEN_SCOPE`, `CHUNK_HEADER_LEN`). -14. **No `unsafe`** — the crate has zero `unsafe` blocks. The PTY - bridge's `libc::kill(-pgid, sig)` and `libc::kill(pid, sig)` calls - are safe `libc` crate APIs (not `unsafe` blocks in this crate); - bounds-checked slice access via `get(..)`/`ok_or_else` is the - pattern. Do not introduce `unsafe` for performance. +14. **No `unsafe`** — the crate has zero `unsafe` blocks outside the + `local` feature module's signal-forwarding calls. The PTY bridge's + `libc::kill(-pgid, sig)` and `libc::kill(pid, sig)` calls (and the + pipe-mode `libc::kill(pid, sig)` / `libc::kill(pid, SIGKILL)` + fallback) are safe `libc` crate APIs wrapped in `unsafe { ... }` + blocks because `libc::kill` is an `unsafe fn`; they are the + documented signal-forwarding path, not `unsafe` in the crate's own + logic. Bounds-checked slice access via `get(..)`/`ok_or_else` is + the pattern. Do not introduce `unsafe` for performance, and do not + add `unsafe` outside the `local` module's `libc::kill` calls. ## Verification Commands diff --git a/src/adapter.rs b/src/adapter.rs index c9fa651..4502125 100644 --- a/src/adapter.rs +++ b/src/adapter.rs @@ -48,7 +48,7 @@ //! 3` was documented as "bidirectional" but the adapter ignored `Exit` //! from the client because the two directions were indistinguishable on //! the same stream_type. The split makes the bidirectionality explicit -//! — see `docs/research/alknet-crate-extraction/findings.md` Phase 7. +//! — see `docs/architecture/tty-wire.md` §"Control Channel". //! //! # Cancel cleanup (ADR-056) //! diff --git a/src/backend.rs b/src/backend.rs index 1998403..536ec8b 100644 --- a/src/backend.rs +++ b/src/backend.rs @@ -19,7 +19,9 @@ use std::sync::Arc; use async_trait::async_trait; use bytes::Bytes; use futures_core::Stream; +#[cfg(test)] use tokio::sync::{mpsc, oneshot}; +#[cfg(test)] use tokio_stream::wrappers::ReceiverStream; use crate::negotiation::{NegotiateRequest, TerminalParamsWire}; @@ -93,7 +95,7 @@ pub struct TtyParams { /// Terminal parameters. `None` = pipe mode (no PTY — ADR-054). `Some` = /// allocate a PTY with these dimensions. pub terminal: Option, - /// Command vector (argv[0] + args). Non-empty. + /// Command vector (`argv[0]` + args). Non-empty. pub cmd: Vec, /// Working directory (`None` = inherit/default). pub cwd: Option, @@ -282,12 +284,14 @@ pub trait TtyBackend: Send + Sync { /// Mock `TtyControl` for tests. Records the last resize/signal call so /// tests can assert delegation through [`TtyControlHandle`]. +#[cfg(test)] #[derive(Default)] -pub struct MockControl { +pub(crate) struct MockControl { pub last_resize: std::sync::Mutex>, pub last_signal: std::sync::Mutex>, } +#[cfg(test)] 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") = @@ -304,23 +308,26 @@ impl TtyControl for MockControl { /// 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. +#[cfg(test)] #[derive(Default)] -pub struct MockBackend { +pub(crate) struct MockBackend { pub exit_code: Option, } +#[cfg(test)] impl MockBackend { - pub fn new() -> Self { + pub(crate) fn new() -> Self { Self::default() } - pub fn with_exit_code(exit_code: i32) -> Self { + pub(crate) fn with_exit_code(exit_code: i32) -> Self { Self { exit_code: Some(exit_code), } } } +#[cfg(test)] #[async_trait] impl TtyBackend for MockBackend { async fn allocate(&self, _params: &TtyParams) -> Result { @@ -365,10 +372,12 @@ impl TtyBackend for MockBackend { /// 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. +#[cfg(test)] struct MockStdinSink { tx: mpsc::Sender, } +#[cfg(test)] impl tokio::io::AsyncWrite for MockStdinSink { fn poll_write( self: Pin<&mut Self>, diff --git a/src/channels.rs b/src/channels.rs index 25062cf..0fbadd6 100644 --- a/src/channels.rs +++ b/src/channels.rs @@ -9,7 +9,7 @@ //! on the per-connection overlay registry. When a consumer (a //! [`crate::session::TtySession`] or any `ChannelClient` caller) //! invokes `channels/tty/sub`, the alkcall channels wrapper does -//! `check_open` → `open_channel` → spawn the [`TtyOpenHandler`] → +//! `check_open` → `open_channel` → spawn the `TtyOpenHandler` → //! respond with `{ channel_id }`. The `TtyOpenHandler` receives the //! channel's `Connection` (data-plane ALPN `alk/tty`), calls //! `accept_bi()` to get the channel's `BiStream`, and runs @@ -49,8 +49,8 @@ use crate::backend::TtyBackend; /// the op is `sub` (subscribe), not `pub` (publish). pub const OP_TTY_OPEN: &str = "channels/tty/sub"; -/// The data-plane ALPN the channel carries. Matches the constant in -/// [`crate::adapter::TtyAdapter::alpn`]. +/// The data-plane ALPN the channel carries. Matches the ALPN returned +/// by [`crate::adapter::TtyAdapter`]'s `alpn()`. pub const TTY_ALPN: &str = "alk/tty"; /// Register the `channels/tty/sub` open op on a per-connection @@ -67,7 +67,7 @@ pub const TTY_ALPN: &str = "alk/tty"; /// the `tty:open` scope gate, and a permissive input schema (the /// `NegotiateRequest` shape — JSON, validated by the /// `drive_session` negotiation reader). It then wraps the -/// [`TtyOpenHandler`] factory and calls +/// `TtyOpenHandler` factory and calls /// [`ChannelCore::register_openable`]. /// /// `backends` is the same backend map the direct-ALPN `TtyAdapter` diff --git a/src/lib.rs b/src/lib.rs index ff33ca8..422a95b 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -20,14 +20,14 @@ //! # WASM target //! //! The default crate (no features) compiles to -//! `wasm32-unknown-unknown`. The [`local`] module is feature-gated +//! `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 +//! 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 diff --git a/src/negotiation.rs b/src/negotiation.rs index acbe9e2..440c3e9 100644 --- a/src/negotiation.rs +++ b/src/negotiation.rs @@ -72,7 +72,7 @@ pub struct NegotiateRequest { /// these dimensions. #[serde(default)] pub tty: Option, - /// Command vector (argv[0] + args); non-empty (checked by the adapter). + /// Command vector (`argv[0]` + args); non-empty (checked by the adapter). pub cmd: Vec, /// Working directory (`None` = inherit/default). #[serde(default)] diff --git a/src/session.rs b/src/session.rs index e361638..dc7f328 100644 --- a/src/session.rs +++ b/src/session.rs @@ -13,7 +13,7 @@ //! //! - [`TtySession::open_via_channels`] — for the `alk/channels` //! multiplexed path. The consumer holds a -//! [`ChannelClient`][alkcall::channels::ChannelClient], calls +//! [`alkcall::channels::client::ChannelClient`], calls //! `open_via_channels(client, params)`, which invokes //! `channels/tty/sub` on channel 0, adopts the resulting channel, //! builds a `Connection` from the reassembled read half + mux write @@ -98,7 +98,7 @@ pub enum TtySessionError { NoExitChunk, /// The `Exit` control chunk's JSON payload failed to parse. #[error("malformed exit chunk: {0}")] - MalformedExitChunk(serde_json::Error), + MalformedExitChunk(String), } /// A live `alk/tty` session — the typed consumer-side handle. @@ -126,15 +126,31 @@ pub struct TtySession { /// (stdout/stderr merged into stdout by the kernel PTY) or after /// `recv_stderr()` has taken it. stderr_rx: Mutex>>, - /// The exit code, resolved by the read pump when it observes the - /// `Exit` control chunk. `wait()` awaits this. `Option` - /// starts as `None`; the pump sends `Some(Ok(code))` on exit - /// chunk or `Some(Err(NoExitChunk))` on stream close. - exit_code: tokio::sync::watch::Receiver>>, + /// The exit outcome, resolved by the read pump when it observes the + /// `Exit` control chunk. `wait()` awaits this. `Option` + /// starts as `None`; the pump sends `Some(Exited(code))` on exit + /// chunk, `Some(MalformedExit(msg))` on a malformed exit chunk, or + /// `Some(NoExitChunk)` on stream close. + exit_code: tokio::sync::watch::Receiver>, /// The read pump task handle. Dropping the session aborts it. _read_pump: JoinHandle<()>, } +/// The cloneable outcome the read pump resolves into the exit watch +/// channel. `wait()` maps this to a [`TtySessionError`] (or the exit +/// code). Kept separate from `TtySessionError` because the watch channel +/// requires `Clone`, and `TtySessionError` carries non-`Clone` payloads +/// (`std::io::Error`, `serde_json::Error`). +#[derive(Debug, Clone)] +enum ExitOutcome { + /// The `Exit` control chunk was observed with this code. + Exited(i32), + /// A `STREAM_CTRL_OUT` chunk failed to parse as a `ControlMessage`. + MalformedExit(String), + /// The stream closed before an `Exit` chunk was observed. + NoExitChunk, +} + impl TtySession { /// Connect directly over a `alk/tty` ALPN connection. /// @@ -237,12 +253,38 @@ impl TtySession { neg_writer.write_frame(&body).await?; let writer = ChunkWriter::new(neg_writer.into_inner()); + let mut reader = ChunkReader::new(read); + // Disambiguate the first response frame (ADR-052 §5): a + // negotiation error frame's 4-byte length prefix starts with + // `0x00`, while a raw chunk's first byte is a `stream_type` in + // `{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. + 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, + Err(RawError::ConnectionClosed) => { + // The server closed cleanly without a response. Fall + // through to the read pump, which resolves `NoExitChunk`. + } + Err(e) => return Err(TtySessionError::Wire(e)), + } + let (stdout_tx, stdout_rx) = mpsc::channel::(64); let (stderr_tx, stderr_rx) = mpsc::channel::(64); - let (exit_tx, exit_rx) = - tokio::sync::watch::channel::>>(None); + let (exit_tx, exit_rx) = tokio::sync::watch::channel::>(None); - let read_pump = tokio::spawn(read_pump(read, stdout_tx, stderr_tx, exit_tx)); + let read_pump = tokio::spawn(read_pump( + reader, + first_byte_peeked, + stdout_tx, + stderr_tx, + exit_tx, + )); Ok(Self { writer: Mutex::new(writer), @@ -256,7 +298,7 @@ impl TtySession { /// Send stdin bytes. Writes a stdin chunk (stream_type 0) with the /// given payload. An empty `bytes` writes a zero-length sentinel /// (client stdin EOF — see `tty-wire.md` §"Sentinels"); callers - /// that want to signal EOF should use [`close_stdin`] instead, + /// that want to signal EOF should use [`Self::close_stdin`] instead, /// which is explicit. pub async fn send_stdin(&self, bytes: Bytes) -> Result<(), TtySessionError> { let mut writer = self.writer.lock().await; @@ -362,11 +404,8 @@ impl TtySession { // watch's current value is `Some(_)` — return it. { let borrow = rx.borrow(); - if let Some(Ok(code)) = borrow.as_ref() { - return Ok(*code); - } - if let Some(Err(_)) = borrow.as_ref() { - return Err(TtySessionError::NoExitChunk); + if let Some(outcome) = borrow.as_ref() { + return outcome_to_result(outcome); } } // Wait for the read pump to send a value. @@ -375,8 +414,8 @@ impl TtySession { .map_err(|_| TtySessionError::NoExitChunk)?; let borrow = rx.borrow(); match borrow.as_ref() { - Some(Ok(code)) => Ok(*code), - Some(Err(_)) | None => Err(TtySessionError::NoExitChunk), + Some(outcome) => outcome_to_result(outcome), + None => Err(TtySessionError::NoExitChunk), } } } @@ -387,6 +426,62 @@ impl Drop for TtySession { } } +/// Map a resolved [`ExitOutcome`] to the `wait()` result. +fn outcome_to_result(outcome: &ExitOutcome) -> Result { + match outcome { + ExitOutcome::Exited(code) => Ok(*code), + ExitOutcome::MalformedExit(msg) => Err(TtySessionError::MalformedExitChunk(msg.clone())), + ExitOutcome::NoExitChunk => Err(TtySessionError::NoExitChunk), + } +} + +/// Read a negotiation error frame (ADR-052 §5) from the raw transport +/// and map it to [`TtySessionError::NegotiationRejected`]. The caller +/// has already peeked the first byte (`0x00`); this reads the remaining +/// 3 length bytes, the body, and parses the `{"error": "...", ...}` +/// JSON. Any non-`error` fields are collected into the `fields` map. +async fn read_negotiation_error(mut read: R) -> TtySessionError +where + R: AsyncRead + Unpin, +{ + use tokio::io::AsyncReadExt; + + let mut len_rest = [0u8; 3]; + if let Err(e) = read.read_exact(&mut len_rest).await { + return TtySessionError::Wire(RawError::Io(e)); + } + let length = u32::from_be_bytes([0x00, len_rest[0], len_rest[1], len_rest[2]]) as usize; + let mut body = vec![0u8; length]; + if let Err(e) = read.read_exact(&mut body).await { + return TtySessionError::Wire(RawError::Io(e)); + } + + let value: serde_json::Value = match serde_json::from_slice(&body) { + Ok(v) => v, + Err(_) => { + return TtySessionError::NegotiationRejected { + error: String::from_utf8_lossy(&body).into_owned(), + fields: HashMap::new(), + }; + } + }; + + let mut fields = HashMap::new(); + let mut error = String::new(); + if let Some(obj) = value.as_object() { + for (k, v) in obj { + if k == "error" { + if let Some(s) = v.as_str() { + error = s.to_string(); + } + } else if let Some(s) = v.as_str() { + fields.insert(k.clone(), s.to_string()); + } + } + } + TtySessionError::NegotiationRejected { error, fields } +} + /// The read pump: reads chunks off the bidi stream's read half and /// routes them to the stdout/stderr/exit channels. The pump owns the /// `ChunkReader`. When the stream closes (clean EOF or transport @@ -403,17 +498,23 @@ impl Drop for TtySession { /// the server shouldn't send these, the pump ignores them (with a /// debug log) async fn read_pump( - read: R, + mut reader: ChunkReader, + mut first_byte_peeked: bool, stdout_tx: mpsc::Sender, stderr_tx: mpsc::Sender, - exit_tx: tokio::sync::watch::Sender>>, + exit_tx: tokio::sync::watch::Sender>, ) where R: AsyncRead + Send + Unpin + 'static, { - let mut reader = ChunkReader::new(read); let mut exit_resolved = false; loop { - match reader.read_chunk().await { + let read = if first_byte_peeked { + 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 => { if stdout_tx.send(chunk.bytes).await.is_err() { @@ -429,7 +530,7 @@ async fn read_pump( } STREAM_CTRL_OUT => match ControlMessage::from_slice(&chunk.bytes) { Ok(ControlMessage::Exit { code }) => { - let _ = exit_tx.send(Some(Ok(code))); + let _ = exit_tx.send(Some(ExitOutcome::Exited(code))); exit_resolved = true; debug!("tty: exit chunk received, code={code}"); break; @@ -438,7 +539,7 @@ async fn read_pump( debug!("tty: ignoring non-exit control on STREAM_CTRL_OUT: {other:?}"); } Err(e) => { - let _ = exit_tx.send(Some(Err(TtySessionError::MalformedExitChunk(e)))); + let _ = exit_tx.send(Some(ExitOutcome::MalformedExit(e.to_string()))); exit_resolved = true; break; } @@ -469,7 +570,7 @@ async fn read_pump( drop(stdout_tx); drop(stderr_tx); if !exit_resolved { - let _ = exit_tx.send(Some(Err(TtySessionError::NoExitChunk))); + let _ = exit_tx.send(Some(ExitOutcome::NoExitChunk)); } } @@ -652,4 +753,185 @@ mod tests { "construction should fail when the negotiation write hits a broken pipe" ); } + + /// A backend that emits fixed stdout/stderr chunks before resolving + /// exit, so the consumer's read-pump routing can be tested with + /// real data (the `MockBackend` emits nothing). + struct EmittingBackend { + stdout: Vec, + stderr: Vec, + exit_code: i32, + } + + #[async_trait::async_trait] + impl TtyBackend for EmittingBackend { + async fn allocate( + &self, + _params: &crate::backend::TtyParams, + ) -> Result { + use crate::backend::{TtyControlHandle, TtyHandle}; + use tokio_stream::wrappers::ReceiverStream; + + let (stdout_tx, stdout_rx) = mpsc::channel::(8); + let (stderr_tx, stderr_rx) = mpsc::channel::(8); + let (_stdin_tx, _stdin_rx) = mpsc::channel::(8); + let (exit_tx, exit_rx) = + tokio::sync::oneshot::channel::>(); + + let stdout = self.stdout.clone(); + let stderr = self.stderr.clone(); + let code = self.exit_code; + tokio::spawn(async move { + for b in stdout { + let _ = stdout_tx.send(b).await; + } + for b in stderr { + let _ = stderr_tx.send(b).await; + } + let _ = exit_tx.send(Ok(code)); + }); + + let stdout: Pin + Send>> = + Box::pin(ReceiverStream::new(stdout_rx)); + let stderr: Option + Send>>> = + Some(Box::pin(ReceiverStream::new(stderr_rx))); + let stdin: Box = Box::new(tokio::io::sink()); + let control = Some(TtyControlHandle::new(Arc::new( + crate::backend::MockControl::default(), + ))); + let exit_code: crate::backend::BoxFuture> = + Box::pin(async move { + exit_rx + .await + .map_err(|_| crate::backend::TtyError::WaitFailed { + message: "exit sender dropped".to_string(), + }) + .and_then(|r| r) + }); + + Ok(TtyHandle { + stdin, + stdout, + stderr, + exit_code, + control, + }) + } + } + + /// The consumer's read pump routes stdout and stderr chunks to the + /// correct channels (L2). `MockBackend` emits nothing, so this uses + /// an `EmittingBackend` that produces real stdout/stderr data. + #[tokio::test] + async fn recv_stdout_and_stderr_route_backend_data() { + let backend = Arc::new(EmittingBackend { + stdout: vec![Bytes::from_static(b"out1"), Bytes::from_static(b"out2")], + stderr: vec![Bytes::from_static(b"err1")], + exit_code: 0, + }); + let identity = Some(Identity { + id: "alice".to_string(), + scopes: vec![crate::adapter::TTY_OPEN_SCOPE.to_string()], + resources: HashMap::new(), + }); + let (session, _server) = wire_session_and_server(backend, identity).await; + + let stdout = session.recv_stdout().await; + let collected: Vec = stdout.collect().await; + // The adapter emits a zero-length stdout sentinel after the + // backend stream ends; filter it out to assert the data chunks. + let data: Vec = collected.into_iter().filter(|b| !b.is_empty()).collect(); + assert_eq!( + data, + vec![Bytes::from_static(b"out1"), Bytes::from_static(b"out2")], + "stdout chunks should route to the stdout stream" + ); + + let stderr = session.recv_stderr().await.expect("stderr present"); + let collected: Vec = stderr.collect().await; + assert_eq!( + collected, + vec![Bytes::from_static(b"err1")], + "stderr chunks should route to the stderr stream" + ); + + let code = tokio::time::timeout(std::time::Duration::from_secs(5), session.wait()) + .await + .expect("wait didn't time out") + .expect("wait returns exit code"); + assert_eq!(code, 0); + } + + /// `wait()` surfaces a malformed exit chunk as + /// `MalformedExitChunk`, not `NoExitChunk` (M2). The server sends a + /// `STREAM_CTRL_OUT` chunk whose JSON fails to parse as a + /// `ControlMessage`. + #[tokio::test] + async fn wait_returns_malformed_exit_chunk() { + let (client, mut server) = duplex(64 * 1024); + let server_handle = tokio::spawn(async move { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + let mut len_buf = [0u8; 4]; + let _ = server.read_exact(&mut len_buf).await; + let len = u32::from_be_bytes(len_buf) as usize; + let mut body = vec![0u8; len]; + let _ = server.read_exact(&mut body).await; + + // Write a ctrl_out chunk with a malformed exit payload. + let payload = br#"{"type":"exit","code":"not-a-number"}"#; + let mut header = [0u8; 5]; + header[0] = crate::wire::STREAM_CTRL_OUT; + header[1..].copy_from_slice(&(payload.len() as u32).to_be_bytes()); + let _ = server.write_all(&header).await; + let _ = server.write_all(payload).await; + let _ = server.flush().await; + }); + let client_conn = Connection::from_bidi(client, b"alk/tty".to_vec(), None); + let session = TtySession::connect_direct(client_conn, test_negotiate("mock")) + .await + .expect("connect_direct"); + let result = tokio::time::timeout(std::time::Duration::from_secs(5), session.wait()) + .await + .expect("wait didn't time out"); + assert!( + matches!(result, Err(TtySessionError::MalformedExitChunk(_))), + "expected MalformedExitChunk, got {result:?}" + ); + let _ = server_handle.await; + } + + /// `connect_direct` returns `NegotiationRejected` when the server + /// rejects the negotiation with an error frame (M1). The server + /// reads the negotiation frame and writes back a length-prefixed + /// `{"error":"unknown_backend","backend":"nope"}` frame. + #[tokio::test] + async fn connect_direct_returns_negotiation_rejected() { + let (client, mut server) = duplex(64 * 1024); + let server_handle = tokio::spawn(async move { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + let mut len_buf = [0u8; 4]; + let _ = server.read_exact(&mut len_buf).await; + let len = u32::from_be_bytes(len_buf) as usize; + let mut body = vec![0u8; len]; + let _ = server.read_exact(&mut body).await; + + let err_body = br#"{"error":"unknown_backend","backend":"nope"}"#; + let _ = server + .write_all(&(err_body.len() as u32).to_be_bytes()) + .await; + let _ = server.write_all(err_body).await; + let _ = server.flush().await; + }); + let client_conn = Connection::from_bidi(client, b"alk/tty".to_vec(), None); + let result = TtySession::connect_direct(client_conn, test_negotiate("mock")).await; + match result { + Err(TtySessionError::NegotiationRejected { error, fields }) => { + assert_eq!(error, "unknown_backend"); + assert_eq!(fields.get("backend").map(String::as_str), Some("nope")); + } + Ok(_) => panic!("expected NegotiationRejected, got Ok(session)"), + Err(other) => panic!("expected NegotiationRejected, got {other:?}"), + } + let _ = server_handle.await; + } } diff --git a/src/wire.rs b/src/wire.rs index de5bfb2..6185f12 100644 --- a/src/wire.rs +++ b/src/wire.rs @@ -29,7 +29,7 @@ //! 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". +//! `docs/architecture/tty-wire.md` §"Sentinels". use std::io; @@ -164,7 +164,31 @@ impl ChunkReader { /// Read one chunk: header, validate, payload. pub async fn read_chunk(&mut self) -> Result { - match self.reader.read_exact(&mut self.header).await { + self.peek_stream_type().await?; + self.read_chunk_after_peek().await + } + + /// Peek the first byte of the next chunk (the `stream_type`) without + /// consuming the rest of the header. Used by the consumer to + /// disambiguate a negotiation error frame (whose 4-byte length prefix + /// starts with `0x00`) from a raw chunk (whose first byte is a + /// `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`]. + pub async fn peek_stream_type(&mut self) -> Result { + match self.reader.read_exact(&mut self.header[..1]).await { + Ok(_) => Ok(self.header[0]), + Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => Err(RawError::ConnectionClosed), + Err(e) => Err(RawError::Io(e)), + } + } + + /// 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. + pub async fn read_chunk_after_peek(&mut self) -> Result { + match self.reader.read_exact(&mut self.header[1..]).await { Ok(_) => {} Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => { return Err(RawError::ConnectionClosed); @@ -444,8 +468,8 @@ mod tests { /// mitigation. #[test] fn bast_stream_type_enum_matches_wire_constants() { - let bast_path = std::path::Path::new(env!("CARGO_MANIFEST_DIR")) - .join("docs/architecture/tty-bast.md"); + let bast_path = + std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("docs/architecture/tty-bast.md"); let markdown = std::fs::read_to_string(&bast_path) .unwrap_or_else(|e| panic!("failed to read {}: {e}", bast_path.display())); @@ -457,8 +481,8 @@ mod tests { .find("\n```") .expect("BAST json block is terminated") + json_start; - let bast: serde_json::Value = serde_json::from_str(&markdown[json_start..json_end]) - .expect("BAST json block parses"); + let bast: serde_json::Value = + serde_json::from_str(&markdown[json_start..json_end]).expect("BAST json block parses"); let stream_type = bast .get("$defs") diff --git a/tests/common/mod.rs b/tests/common/mod.rs index 57e6817..e2ce5c1 100644 --- a/tests/common/mod.rs +++ b/tests/common/mod.rs @@ -289,3 +289,14 @@ pub fn negotiate_pipe_json(backend: &str, cmd: &[&str]) -> String { cmd = cmd_json.join(",") ) } + +/// A nanos-timestamp seed for uniquifying temp-file names in the +/// cancel-cleanup tests (which write a child pid to a temp file and +/// probe it after drop). +pub fn nanos_seed() -> u64 { + use std::time::{SystemTime, UNIX_EPOCH}; + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_nanos() as u64) + .unwrap_or(0) +} diff --git a/tests/pipe.rs b/tests/pipe.rs index 0438872..44d5a73 100644 --- a/tests/pipe.rs +++ b/tests/pipe.rs @@ -18,7 +18,7 @@ use std::time::Duration; use alktty::local::LocalTtyBackend; use alktty::wire::STREAM_STDOUT; -use common::{negotiate_pipe_json, spawn_session}; +use common::{nanos_seed, negotiate_pipe_json, spawn_session}; /// 9. Happy path (echo): negotiate `{backend:"local", tty:null, /// cmd:["echo","hello"]}`, read stdout chunks, assert "hello", exit 0. @@ -214,11 +214,3 @@ async fn pipe_echo_emits_stdout_chunk_then_sentinel() { ); let _ = server.await; } - -fn nanos_seed() -> u64 { - use std::time::{SystemTime, UNIX_EPOCH}; - SystemTime::now() - .duration_since(UNIX_EPOCH) - .map(|d| d.as_nanos() as u64) - .unwrap_or(0) -} diff --git a/tests/pty.rs b/tests/pty.rs index 6eecf18..e1f15c5 100644 --- a/tests/pty.rs +++ b/tests/pty.rs @@ -21,7 +21,7 @@ use std::time::Duration; use alktty::local::LocalTtyBackend; use alktty::wire::{STREAM_CTRL_OUT, STREAM_STDIN}; -use common::{negotiate_pty_json, spawn_session}; +use common::{nanos_seed, negotiate_pty_json, spawn_session}; const PTY_NEG_ECHO: &str = r#"{"carriage":"raw","backend":"local","tty":{"cols":80,"rows":24,"pixel_width":0,"pixel_height":0},"cmd":["echo","hello"]}"#; @@ -258,11 +258,3 @@ async fn pty_exit_chunk_is_last() { assert!(saw_exit, "did not see the exit chunk"); let _ = server.await; } - -fn nanos_seed() -> u64 { - use std::time::{SystemTime, UNIX_EPOCH}; - SystemTime::now() - .duration_since(UNIX_EPOCH) - .map(|d| d.as_nanos() as u64) - .unwrap_or(0) -}