alkdev/alktty
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: address code review #001 findings (M1, M2, L2, L4, L5, N1-N3, N5)

Browse Source 
- M1: TtySession now handles the negotiation-rejection error frame.
  from_halves peeks the first response byte (ADR-052 §5 disambiguation)
  and returns NegotiationRejected on a 0x00-prefixed error frame;
  ChunkReader gains peek_stream_type/read_chunk_after_peek.
- M2: wait() now surfaces MalformedExitChunk instead of collapsing it
  to NoExitChunk. The exit watch channel carries a cloneable
  ExitOutcome enum; MalformedExitChunk carries a String.
- L2: add EmittingBackend + recv_stdout_and_stderr_route_backend_data
  test covering the consumer read-pump stdout/stderr routing.
- L4: MockBackend/MockControl/MockStdinSink are now #[cfg(test)]
  pub(crate), removing them from the public API.
- L5: cargo fmt (the BAST drift test was unformatted).
- N1: fix all 9 rustdoc intra-doc links.
- N2: fix stale doc paths (crates/tty/ and docs/research/).
- N3: amend AGENTS.md §14 to accurately describe the local module's
  libc::kill unsafe blocks.
- N5: consolidate nanos_seed into tests/common/mod.rs.

Verification: cargo test (84), cargo test --all-features (107),
clippy clean (native + wasm), fmt clean, doc clean, wasm check clean.
Coverage: session.rs 79.71% -> 87.43%, total 90.74% -> 91.47%.
This commit is contained in:
deepseek-v4-pro
2026-08-17 12:07:44 +00:00
parent 8ff7ba27f3
commit 99441530ab
11 changed files with 382 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/session.rs
+307 -25
View File
@@ -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<Option<mpsc::Receiver<Bytes>>>,
/// The exit code, resolved by the read pump when it observes the
/// `Exit` control chunk. `wait()` awaits this. `Option<Result>`
/// 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<Option<Result<i32, TtySessionError>>>,
/// The exit outcome, resolved by the read pump when it observes the
/// `Exit` control chunk. `wait()` awaits this. `Option<ExitOutcome>`
/// 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<Option<ExitOutcome>>,
/// 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::<Bytes>(64);
let (stderr_tx, stderr_rx) = mpsc::channel::<Bytes>(64);
let (exit_tx, exit_rx) =
tokio::sync::watch::channel::<Option<Result<i32, TtySessionError>>>(None);
let (exit_tx, exit_rx) = tokio::sync::watch::channel::<Option<ExitOutcome>>(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<i32, TtySessionError> {
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<R>(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<R>(
read: 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<Result<i32, TtySessionError>>>,
exit_tx: tokio::sync::watch::Sender<Option<ExitOutcome>>,
) 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<R>(
}
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<R>(
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<R>(
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<Bytes>,
stderr: Vec<Bytes>,
exit_code: i32,
}
#[async_trait::async_trait]
impl TtyBackend for EmittingBackend {
async fn allocate(
&self,
_params: &crate::backend::TtyParams,
) -> Result<crate::backend::TtyHandle, crate::backend::TtyError> {
use crate::backend::{TtyControlHandle, TtyHandle};
use tokio_stream::wrappers::ReceiverStream;
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) =
tokio::sync::oneshot::channel::<Result<i32, crate::backend::TtyError>>();
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<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 AsyncWrite + Send + Unpin> = Box::new(tokio::io::sink());
let control = Some(TtyControlHandle::new(Arc::new(
crate::backend::MockControl::default(),
)));
let exit_code: crate::backend::BoxFuture<Result<i32, crate::backend::TtyError>> =
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<Bytes> = 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<Bytes> = 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<Bytes> = 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;
}
}