feat(protocols): add NOW execution client

Cargo.toml

@@ -16,7 +16,7 @@ keywords = ["rdp", "remote-desktop", "network", "client", "protocol"]
 categories = ["network-programming"]
 
 [workspace.dependencies]
-now-proto-pdu = { version = "0.1", path = "protocols/rust/now-proto-pdu" }
+now-proto-pdu = { version = "0.4", path = "protocols/rust/now-proto-pdu" }
 now-proto-fuzzing = { version = "0.1", path = "protocols/rust/now-proto-fuzzing" }
 
 [profile.test.package.proptest]

protocols/rust/now-client/Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "now-client"
+version = "0.1.0"
+readme = "README.md"
+description = "High-level Tokio client for NOW execution channels"
+edition.workspace = true
+license.workspace = true
+homepage.workspace = true
+repository.workspace = true
+authors.workspace = true
+keywords.workspace = true
+categories.workspace = true
+
+[lints]
+workspace = true
+
+[dependencies]
+now-proto-pdu = { workspace = true, features = ["std"] }
+thiserror = "2"
+tokio = { version = "1.52", features = ["io-util", "macros", "rt", "sync", "time"] }

protocols/rust/now-client/README.md

@@ -0,0 +1,20 @@
+# NOW client
+
+`now-client` is a high-level, transport-agnostic Rust client for the NOW execution
+channel. It accepts a caller-provided Tokio `AsyncRead + AsyncWrite` byte stream;
+consumers create DVCs, pipes, sockets, and replacement clients after reconnects.
+
+Connect with `NowClient::connect`, then query negotiated `NowCapabilities` from the
+returned handle. `process`, `batch`, `win_ps`, and `pwsh` submit tracked executions;
+their `_detached` counterparts submit detached executions. `run` is submission-only.
+The client defensively negotiates capabilities, applies bounded frame/command/event
+queues, and permits **one tracked execution at a time per stream**. Tracked operations
+expose raw `Vec<u8>` stdout/stderr chunks, stdin forwarding, normal cancellation, and
+terminal status.
+
+Run and detached requests are submission-only. Gateway may emit an immediate
+Started/Data/Result sequence for Run; the client records and discards those matching
+frames so they cannot affect the next tracked operation. This recent-session quarantine
+is bounded and evicts its oldest entry; evicted Run traffic remains harmless because
+session IDs are never reused. This crate deliberately does not provide Abort, Shell
+submission, DVC/pipe setup, reconnect policy, or retained operation output.

protocols/rust/now-client/src/capabilities.rs

@@ -0,0 +1,94 @@
+use core::time::Duration;
+
+use now_proto_pdu::{NowChannelCapsetMsg, NowExecCapsetFlags, NowProtoVersion};
+
+use crate::NowClientError;
+
+/// Capabilities mutually supported by the client and its connected peer.
+///
+/// The value is always calculated as an intersection with the local advertised capset,
+/// even if the peer echoes unsupported flags.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct NowCapabilities {
+    capset: NowChannelCapsetMsg,
+}
+
+impl NowCapabilities {
+    pub(crate) fn negotiate(
+        requested: &NowChannelCapsetMsg,
+        peer: &NowChannelCapsetMsg,
+    ) -> Result<Self, NowClientError> {
+        if requested.version().major != peer.version().major {
+            return Err(NowClientError::IncompatibleVersion {
+                client: requested.version(),
+                peer: peer.version(),
+            });
+        }
+
+        Ok(Self {
+            capset: requested.downgrade(peer),
+        })
+    }
+
+    /// Returns the negotiated capability-set PDU.
+    pub fn capset(&self) -> &NowChannelCapsetMsg {
+        &self.capset
+    }
+
+    /// Returns the negotiated NOW protocol version.
+    pub fn version(&self) -> NowProtoVersion {
+        self.capset.version()
+    }
+
+    /// Returns the negotiated heartbeat interval, if either peer requested one.
+    pub fn heartbeat_interval(&self) -> Option<Duration> {
+        self.capset.heartbeat_interval()
+    }
+
+    /// Returns whether the generic Run style is available.
+    pub fn supports_run(&self) -> bool {
+        self.has(NowExecCapsetFlags::STYLE_RUN)
+    }
+
+    /// Returns whether CreateProcess execution is available.
+    pub fn supports_process(&self) -> bool {
+        self.has(NowExecCapsetFlags::STYLE_PROCESS)
+    }
+
+    /// Returns whether Batch execution is available.
+    pub fn supports_batch(&self) -> bool {
+        self.has(NowExecCapsetFlags::STYLE_BATCH)
+    }
+
+    /// Returns whether Windows PowerShell execution is available.
+    pub fn supports_win_ps(&self) -> bool {
+        self.has(NowExecCapsetFlags::STYLE_WINPS)
+    }
+
+    /// Returns whether PowerShell 7 execution is available.
+    pub fn supports_pwsh(&self) -> bool {
+        self.has(NowExecCapsetFlags::STYLE_PWSH)
+    }
+
+    /// Returns whether tracked I/O redirection is available.
+    pub fn supports_io_redirection(&self) -> bool {
+        self.has(NowExecCapsetFlags::IO_REDIRECTION) && self.at_least(1, 3)
+    }
+
+    /// Returns whether UTF-8 and Unicode-console encoding controls are available.
+    pub fn supports_unicode_console(&self) -> bool {
+        self.has(NowExecCapsetFlags::UNICODE_CONSOLE) && self.version().supports_exec_unicode_console()
+    }
+
+    pub(crate) fn supports_detached(&self) -> bool {
+        self.at_least(1, 4)
+    }
+
+    fn has(&self, capability: NowExecCapsetFlags) -> bool {
+        self.capset.exec_capset().contains(capability)
+    }
+
+    fn at_least(&self, major: u16, minor: u16) -> bool {
+        self.version() >= NowProtoVersion { major, minor }
+    }
+}