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

refactor(tty): split control channel into STREAM_CTRL_IN/STREAM_CTRL_OUT halves (Phase 7)

Browse Source 
The 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 — half-duplex in
disguise. Phase 7 splits it into two halves so the bidirectionality is
literal on the wire.

Changes:
- wire.rs: STREAM_CTRL_IN = 3 (client→server), STREAM_CTRL_OUT = 4
  (server→client); InvalidStreamType bound > 3 → > 4; Chunk::control
  → Chunk::ctrl_in/ctrl_out; ChunkWriter::write_control_json →
  write_ctrl_in_json/write_ctrl_out_json; tests split accordingly.
- control.rs: ControlMessage doc updated with the stream_type column;
  JSON shape unchanged.
- adapter.rs: pump_client_to_backend dispatches on STREAM_CTRL_IN
  (Resize/Signal/Eof; Exit on ctrl_in is a protocol violation,
  ignored); send_exit_chunk emits on STREAM_CTRL_OUT; STREAM_CTRL_OUT
  from the client is a protocol violation, ignored. 3 new tests for
  the direction enforcement; existing tests updated to the new
  stream_types.
- negotiation.rs: framing-disambiguation doc updated (server-sent
  stream_type set is {1, 2, 4}).
- alknet-tty-local/tests: common/mod.rs, pty.rs, pipe.rs updated to
  the new constants.

Specs:
- ADR-052 amended (§4a 'Control channel split (Phase 7 amendment)').
- tty-wire.md + tty-adapter.md updated (last_updated 2026-07-18).

Verification:
- cargo test -p alknet-tty: 65 passed (was 61; +4 new tests).
- cargo test -p alknet-tty-local: 19 passed.
- cargo test --workspace --all-features: 1017 passed, 0 failed.
- cargo clippy --workspace --all-features: clean.
- cargo fmt --all: clean.
This commit is contained in:
glm-5.2 committed 2026-07-18 17:02:59 +00:00
1 parent 859ad35896
commit c2b7055a64
11 files changed
+571 -142
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/alknet-tty-local/tests/common/mod.rs
+9 -6
View File
@@ -21,7 +21,9 @@ use std::sync::Arc;
use alknet_core::auth::Identity;
use alknet_tty::adapter::drive_session;
use alknet_tty::backend::TtyBackend;
use alknet_tty::wire::{ChunkReader, STREAM_CONTROL, STREAM_STDERR, STREAM_STDOUT};
use alknet_tty::wire::{
ChunkReader, STREAM_CTRL_IN, STREAM_CTRL_OUT, STREAM_STDERR, STREAM_STDOUT,
};
use bytes::Bytes;
use tokio::io::duplex;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
@@ -70,10 +72,11 @@ impl ClientSide {
self.write.flush().await.unwrap();
}
/// Write a control chunk (stream_type 3) carrying a serialized
/// `ControlMessage` JSON payload.
/// Write a client→server control chunk (`STREAM_CTRL_IN`, stream_type
/// 3) carrying a serialized `ControlMessage` JSON payload (`Resize`,
/// `Signal`, or `Eof`).
pub async fn write_control(&mut self, json: &[u8]) {
self.write_chunk(STREAM_CONTROL, json).await;
self.write_chunk(STREAM_CTRL_IN, json).await;
}
/// Read one raw chunk from the server. Returns the `stream_type`
@@ -145,7 +148,7 @@ impl ClientSide {
stderr.extend_from_slice(&bytes);
}
}
STREAM_CONTROL => {
STREAM_CTRL_OUT => {
let v: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
if v["type"] == "exit" {
return Some((stdout, stderr, v["code"].as_i64().unwrap() as i32));
@@ -177,7 +180,7 @@ impl ClientSide {
stderr.extend_from_slice(&bytes);
}
}
STREAM_CONTROL => {
STREAM_CTRL_OUT => {
let v: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
if v["type"] == "exit" {
return Some((stdout, stderr, v["code"].as_i64().unwrap() as i32));
crates/alknet-tty-local/tests/pipe.rs
+1 -1
View File
@@ -199,7 +199,7 @@ async fn pipe_echo_emits_stdout_chunk_then_sentinel() {
if st == STREAM_STDOUT && !bytes.is_empty() {
saw_nonempty_stdout = true;
}
if st == alknet_tty::wire::STREAM_CONTROL {
if st == alknet_tty::wire::STREAM_CTRL_OUT {
let v: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
if v["type"] == "exit" {
break;
crates/alknet-tty-local/tests/pty.rs
+2 -2
View File
@@ -17,7 +17,7 @@ mod common;
use std::sync::Arc;
use std::time::Duration;
use alknet_tty::wire::{STREAM_CONTROL, STREAM_STDIN};
use alknet_tty::wire::{STREAM_CTRL_OUT, STREAM_STDIN};
use alknet_tty_local::LocalTtyBackend;
use common::{negotiate_pty_json, spawn_session};
@@ -245,7 +245,7 @@ async fn pty_exit_chunk_is_last() {
if saw_exit {
panic!("chunk arrived after exit: stream_type={st}, bytes={bytes:?} (ADR-055)");
}
if st == STREAM_CONTROL {
if st == STREAM_CTRL_OUT {
let v: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
if v["type"] == "exit" {
assert_eq!(v["code"], 0);
crates/alknet-tty/src/adapter.rs
+147 -23
View File
@@ -22,19 +22,34 @@
//! (stream_type 2) when `TtyHandle.stderr` is `Some`. On backend stdout
//! EOF, emit a zero-length stdout sentinel.
//! - **B. client → backend**: stdin chunks (stream_type 0) →
//! `TtyHandle.stdin`; control chunks (stream_type 3) →
//! `ControlMessage` dispatch (`Resize`, `Signal`, `Eof`; `Exit` is
//! server→client only and ignored). Zero-length stdin chunk or
//! read-half close → EOF to backend stdin.
//! `TtyHandle.stdin`; client→server control chunks (stream_type 3,
//! `STREAM_CTRL_IN`) → `ControlMessage` dispatch (`Resize`, `Signal`,
//! `Eof`). `STREAM_CTRL_OUT` (stream_type 4) from the client is a
//! protocol violation (it's the server→client half) and is ignored;
//! `Exit` arriving on `STREAM_CTRL_IN` is likewise a protocol
//! violation and ignored. Zero-length stdin chunk or read-half close
//! → EOF to backend stdin.
//! - **C. exit → exit chunk**: await `TtyHandle.exit_code`; on resolve,
//! enqueue `{"type":"exit","code":N}` as a control chunk (stream_type
//! 3). On `TtyError` → `{"type":"exit","code":-1}`.
//! enqueue `{"type":"exit","code":N}` as a server→client control
//! chunk (stream_type 4, `STREAM_CTRL_OUT`). On `TtyError` →
//! `{"type":"exit","code":-1}`.
//!
//! The adapter enforces the **exit-chunk-is-last** invariant (ADR-055):
//! it waits for BOTH the stdout/stderr pumps to complete AND `exit_code`
//! to resolve before enqueueing the exit chunk. A drainer task writes
//! chunks to the client in arrival order; the exit chunk is last.
//!
//! # Bidirectional control channel (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 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.
//!
//! # Cancel cleanup (ADR-056)
//!
//! On connection drop or stream reset, the pump tasks are dropped, which
@@ -63,7 +78,7 @@ use crate::control::ControlMessage;
use crate::negotiation::{
error_response_bytes, NegotiateRequest, NegotiationError, NegotiationReader, NegotiationWriter,
};
use crate::wire::{Chunk, ChunkReader, ChunkWriter, RawError, STREAM_CONTROL, STREAM_STDIN};
use crate::wire::{Chunk, ChunkReader, ChunkWriter, RawError, STREAM_CTRL_IN, STREAM_STDIN};
/// The scope required to open a `alknet/tty` session (ADR-050). A two-way-door
/// choice (reversible: a deployment-configured scope, not a wire-format
@@ -378,7 +393,7 @@ async fn send_exit_chunk(writer_tx: &mpsc::Sender<Chunk>, code: i32) {
let exit_msg = ControlMessage::Exit { code };
match exit_msg.to_json() {
Ok(json) => {
let chunk = Chunk::control(json);
let chunk = Chunk::ctrl_out(json);
if writer_tx.send(chunk).await.is_err() {
debug!("tty: writer channel closed before exit chunk");
}
@@ -423,9 +438,25 @@ async fn pump_stderr(
debug!("tty: stderr pump done");
}
/// Pump client chunks → backend: stdin chunks → `TtyHandle.stdin`, control
/// chunks → `ControlMessage` dispatch. On client read-half close or a
/// zero-length stdin chunk, signal EOF to the backend's stdin.
/// Pump client chunks → backend: stdin chunks → `TtyHandle.stdin`,
/// client→server control chunks (`STREAM_CTRL_IN`, stream_type 3) →
/// `ControlMessage` dispatch. On client read-half close or a zero-length
/// stdin chunk, signal EOF to the backend's stdin.
///
/// # Direction enforcement (Phase 7)
///
/// The control channel is split into two halves. This pump reads from
/// the client, so it dispatches only `STREAM_CTRL_IN` (client→server):
///
/// - `Resize` / `Signal` / `Eof` → forward to the backend's control
/// handle (`TtyControlHandle::resize` / `signal` / `stdin.shutdown`).
/// - `Exit` arriving on `STREAM_CTRL_IN` is a protocol violation
/// (`Exit` is server→client only, belongs on `STREAM_CTRL_OUT`); the
/// adapter ignores it. (The previous single `STREAM_CONTROL = 3`
/// couldn't distinguish the two directions, so `Exit` from the client
/// was always ignored — the split makes the rejection explicit.)
/// - `STREAM_CTRL_OUT` (stream_type 4) from the client is a protocol
/// violation (it's the server→client half); the adapter ignores it.
async fn pump_client_to_backend<R>(
client_read: R,
mut stdin: Box<dyn tokio::io::AsyncWrite + Send + Unpin>,
@@ -446,7 +477,7 @@ async fn pump_client_to_backend<R>(
break;
}
}
STREAM_CONTROL => match ControlMessage::from_slice(&chunk.bytes) {
STREAM_CTRL_IN => match ControlMessage::from_slice(&chunk.bytes) {
Ok(ControlMessage::Resize {
cols,
rows,
@@ -467,12 +498,21 @@ async fn pump_client_to_backend<R>(
debug!("tty: client stdin EOF (eof control)");
}
Ok(ControlMessage::Exit { .. }) => {
debug!("tty: ignoring Exit control from client (server→client only)");
debug!(
"tty: ignoring Exit control on STREAM_CTRL_IN \
(server→client only; belongs on STREAM_CTRL_OUT)"
);
}
Err(e) => {
debug!("tty: ignoring unknown control type: {e}");
}
},
crate::wire::STREAM_CTRL_OUT => {
debug!(
"tty: ignoring STREAM_CTRL_OUT (stream_type 4) from client \
(server→client half; client should not write on it)"
);
}
other => {
debug!("tty: ignoring stream_type {other} from client");
}
@@ -868,7 +908,7 @@ mod tests {
assert!(bytes.is_empty());
let (st, bytes) = client.read_chunk().await;
assert_eq!(st, crate::wire::STREAM_CONTROL);
assert_eq!(st, crate::wire::STREAM_CTRL_OUT);
let v: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
assert_eq!(v["type"], "exit");
assert_eq!(v["code"], 0);
@@ -899,7 +939,7 @@ mod tests {
loop {
let (st, bytes) = client.read_chunk().await;
if st == crate::wire::STREAM_CONTROL {
if st == crate::wire::STREAM_CTRL_OUT {
let v: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
if v["type"] == "exit" {
assert_eq!(v["code"], 7);
@@ -959,13 +999,13 @@ mod tests {
client
.write_chunk(
crate::wire::STREAM_CONTROL,
crate::wire::STREAM_CTRL_IN,
br#"{"type":"resize","cols":100,"rows":50}"#,
)
.await;
client
.write_chunk(
crate::wire::STREAM_CONTROL,
crate::wire::STREAM_CTRL_IN,
br#"{"type":"signal","name":"INT"}"#,
)
.await;
@@ -1003,7 +1043,7 @@ mod tests {
client.write_negotiation(TEST_NEG).await;
client
.write_chunk(crate::wire::STREAM_CONTROL, br#"{"type":"unknown"}"#)
.write_chunk(crate::wire::STREAM_CTRL_IN, br#"{"type":"unknown"}"#)
.await;
let stdout_tx = backend.take_stdout_tx().await.expect("stdout tx");
@@ -1025,6 +1065,9 @@ mod tests {
#[tokio::test]
async fn exit_control_from_client_ignored() {
// `Exit` is server→client only (belongs on `STREAM_CTRL_OUT`).
// Sending it on `STREAM_CTRL_IN` (client→server) is a protocol
// violation; the adapter ignores it and keeps pumping stdout.
let (backend, _control, _cancel) = TestBackend::builder().build();
let backends = make_backends(backend.clone());
let (mut client, server) = make_client_and_server();
@@ -1037,7 +1080,7 @@ mod tests {
client.write_negotiation(TEST_NEG).await;
client
.write_chunk(crate::wire::STREAM_CONTROL, br#"{"type":"exit","code":99}"#)
.write_chunk(crate::wire::STREAM_CTRL_IN, br#"{"type":"exit","code":99}"#)
.await;
let stdout_tx = backend.take_stdout_tx().await.expect("stdout tx");
@@ -1057,6 +1100,87 @@ mod tests {
let _ = session.await;
}
#[tokio::test]
async fn ctrl_out_from_client_ignored() {
// `STREAM_CTRL_OUT` (stream_type 4) is the server→client half.
// The client writing on it is a protocol violation; the adapter
// ignores the chunk and keeps pumping stdout (Phase 7).
let (backend, _control, _cancel) = TestBackend::builder().build();
let backends = make_backends(backend.clone());
let (mut client, server) = make_client_and_server();
let identity = identity_with_scope(TTY_OPEN_SCOPE);
let session = tokio::spawn(async move {
drive_session_server(server, backends, None, identity).await;
});
client.write_negotiation(TEST_NEG).await;
// Bogus: a client writing on the server→client control half.
client
.write_chunk(
crate::wire::STREAM_CTRL_OUT,
br#"{"type":"exit","code":99}"#,
)
.await;
let stdout_tx = backend.take_stdout_tx().await.expect("stdout tx");
let _ = backend.take_stderr_tx().await;
stdout_tx
.send(Bytes::from_static(b"after-bogus-ctrl-out"))
.await
.unwrap();
drop(stdout_tx);
let exit_tx = backend.take_exit_tx().await.expect("exit tx");
exit_tx.send(Ok(0)).unwrap();
let (st, bytes) = client.read_chunk().await;
assert_eq!(st, STREAM_STDOUT);
assert_eq!(bytes.as_ref(), b"after-bogus-ctrl-out");
let _ = session.await;
}
#[tokio::test]
async fn exit_chunk_arrives_on_ctrl_out_not_ctrl_in() {
// Verifies the adapter emits `Exit` on `STREAM_CTRL_OUT` (4), not
// `STREAM_CTRL_IN` (3) — the Phase 7 bidirectionality fix. A client
// distinguishing the two halves can route exit vs. control
// without parsing the JSON tag first.
let (backend, _control, _cancel) = TestBackend::builder().build();
let backends = make_backends(backend.clone());
let (mut client, server) = make_client_and_server();
let identity = identity_with_scope(TTY_OPEN_SCOPE);
let session = tokio::spawn(async move {
drive_session_server(server, backends, None, identity).await;
});
client.write_negotiation(TEST_NEG).await;
let stdout_tx = backend.take_stdout_tx().await.expect("stdout tx");
let _ = backend.take_stderr_tx().await;
drop(stdout_tx);
let exit_tx = backend.take_exit_tx().await.expect("exit tx");
exit_tx.send(Ok(42)).unwrap();
let (st, bytes) = client.read_chunk().await;
assert_eq!(st, STREAM_STDOUT);
assert!(bytes.is_empty());
let (st, bytes) = client.read_chunk().await;
assert_eq!(
st,
crate::wire::STREAM_CTRL_OUT,
"exit chunk must arrive on STREAM_CTRL_OUT (4), not STREAM_CTRL_IN (3)"
);
let v: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
assert_eq!(v["type"], "exit");
assert_eq!(v["code"], 42);
let _ = session.await;
}
#[tokio::test]
async fn unknown_backend_error() {
let (backend, _control, _cancel) = TestBackend::builder().build();
@@ -1184,7 +1308,7 @@ mod tests {
loop {
let (st, bytes) = client.read_chunk().await;
if st == crate::wire::STREAM_CONTROL {
if st == crate::wire::STREAM_CTRL_OUT {
let v: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
assert_eq!(v["type"], "exit");
assert_eq!(v["code"], -1);
@@ -1294,7 +1418,7 @@ mod tests {
loop {
let (st, bytes) = client.read_chunk().await;
if st == crate::wire::STREAM_CONTROL {
if st == crate::wire::STREAM_CTRL_OUT {
let v: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
assert_eq!(v["type"], "exit");
assert_eq!(v["code"], 0);
@@ -1346,7 +1470,7 @@ mod tests {
assert_eq!(bytes.as_ref(), b"err");
saw_stderr = true;
}
crate::wire::STREAM_CONTROL => {
crate::wire::STREAM_CTRL_OUT => {
let v: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
assert_eq!(v["type"], "exit");
saw_exit = true;
@@ -1379,7 +1503,7 @@ mod tests {
client.write_chunk(STREAM_STDIN, b"first").await;
client
.write_chunk(crate::wire::STREAM_CONTROL, br#"{"type":"eof"}"#)
.write_chunk(crate::wire::STREAM_CTRL_IN, br#"{"type":"eof"}"#)
.await;
let mut received = Vec::new();
crates/alknet-tty/src/control.rs
+41 -20
View File
@@ -1,4 +1,14 @@
//! Control messages carried in `stream_type 3` chunks (ADR-052).
//! 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
//! `docs/research/alknet-crate-extraction/findings.md` Phase 7.
//!
//! Control chunks carry a JSON payload tagged by `type`. The schema is the
//! POC's `ControlMessage` (`/workspace/alknet-tty-poc/src/control.rs`):
@@ -23,20 +33,30 @@
use serde::{Deserialize, Serialize};
/// A control message riding on `stream_type 3`.
/// 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 | variant | maps to |
/// |----------------|----------|----------------------------------------------------|
/// | client→server | `Resize` | SSH `window-change`, docker exec resize, `ioctl` |
/// | client→server | `Signal` | SSH `signal`, docker exec signal, `kill(-pgid, n)` |
/// | client→server | `Eof` | SSH channel EOF, docker stdin close, `ChildStdin` |
/// | server→client | `Exit` | the completion signal (ADR-055) |
/// | 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).
/// 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
@@ -49,23 +69,24 @@ pub enum ControlMessage {
#[serde(default)]
pixel_height: u16,
},
/// Forward a signal to the child process group (client→server).
/// 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). 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".
/// 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). 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).
/// 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 },
}
crates/alknet-tty/src/negotiation.rs
+4 -2
View File
@@ -21,8 +21,10 @@
//! - 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` in `{1, 2, 3}` —
//! `0` (stdin from server) is invalid, so `0x00` is unambiguous.
//! - 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".
crates/alknet-tty/src/wire.rs
+84 -27
View File
@@ -7,10 +7,20 @@
//! ```
//!
//! `stream_type`:
//! - 0 = stdin (client→server, raw bytes)
//! - 1 = stdout (server→client, raw bytes)
//! - 2 = stderr (server→client, raw bytes)
//! - 3 = control (bidirectional, JSON control message — see [`crate::control`])
//! - 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 `docs/research/alknet-crate-extraction/
//! findings.md` Phase 7.
//!
//! 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
@@ -29,8 +39,11 @@ pub const STREAM_STDIN: u8 = 0;
pub const STREAM_STDOUT: u8 = 1;
/// stderr channel (server→client, raw bytes).
pub const STREAM_STDERR: u8 = 2;
/// control channel (bidirectional, JSON control message).
pub const STREAM_CONTROL: u8 = 3;
/// 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;
@@ -54,7 +67,7 @@ pub enum RawError {
/// The peer closed the stream cleanly (unexpected EOF on header or payload).
#[error("connection closed")]
ConnectionClosed,
/// The chunk header's `stream_type` byte was > 3.
/// 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`.
@@ -66,11 +79,12 @@ pub enum RawError {
/// payload bytes.
///
/// Construct with [`Chunk::stdin`], [`Chunk::stdout`], [`Chunk::stderr`],
/// or [`Chunk::control`] for the four fixed channels.
/// [`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_CONTROL`].
/// [`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,
@@ -101,10 +115,19 @@ impl Chunk {
}
}
/// A control chunk (stream_type 3).
pub fn control(bytes: bytes::Bytes) -> Self {
/// A client→server control chunk (stream_type 3) — `Resize`, `Signal`,
/// or `Eof`.
pub fn ctrl_in(bytes: bytes::Bytes) -> Self {
Self {
stream_type: STREAM_CONTROL,
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,
}
}
@@ -113,7 +136,7 @@ impl Chunk {
/// Reads raw chunks from an [`AsyncRead`] transport.
///
/// [`ChunkReader::read_chunk`] reads the 5-byte header, validates the
/// `stream_type` (≤ 3, else [`RawError::InvalidStreamType`]) and 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
@@ -148,7 +171,7 @@ impl<R: AsyncRead + Unpin> ChunkReader<R> {
}
let stream_type = self.header[0];
if stream_type > 3 {
if stream_type > 4 {
return Err(RawError::InvalidStreamType(stream_type));
}
@@ -183,9 +206,10 @@ impl<R: AsyncRead + Unpin> ChunkReader<R> {
/// 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`] and
/// [`ChunkWriter::write_control_json`] are convenience helpers for the
/// two most common write paths.
/// (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,
}
@@ -229,10 +253,24 @@ impl<W: AsyncWrite + Unpin> ChunkWriter<W> {
Ok(())
}
/// Write a control chunk (stream_type 3) carrying a JSON payload.
pub async fn write_control_json(&mut self, json: &[u8]) -> Result<(), RawError> {
/// 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_CONTROL;
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?;
@@ -281,8 +319,13 @@ mod tests {
}
#[tokio::test]
async fn round_trip_control() {
round_trip(STREAM_CONTROL, br#"{"type":"eof"}"#).await;
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]
@@ -303,27 +346,41 @@ mod tests {
}
#[tokio::test]
async fn round_trip_write_control_json_helper() {
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_control_json(json).await.unwrap();
writer.write_ctrl_in_json(json).await.unwrap();
let read = reader.read_chunk().await.unwrap();
assert_eq!(read.stream_type, STREAM_CONTROL);
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);
a.write_all(&[4u8, 0, 0, 0, 0]).await.unwrap();
// 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(4)));
assert!(matches!(err, RawError::InvalidStreamType(5)));
}
#[tokio::test]