From 4953ac5988f60abbd9c073a9e463f4280eecfcdf Mon Sep 17 00:00:00 2001 From: leynos Date: Sat, 2 May 2026 02:28:09 +0200 Subject: [PATCH 1/4] Document ACP capability masking modules and session knobs Update the exec subsystem repository layout in `docs/developers-guide.md` to list `acp_helpers.rs` and the three ACP test modules (`protocol_acp_tests.rs`, `protocol_acp_bdd_tests.rs`, `protocol_acp_forwarding_tests.rs`), and refresh the `session.rs` entry to mention the `with_acp_initialize_rewrite_enabled` opt-in. Add `///` doc comments to the `rewrite_acp_initialize` field on `ExecSessionOptions` and `ProtocolSessionOptions`, and to the `with_stdin_forwarding_disabled` builder on `ProtocolSessionOptions`, so the new ACP seam is documented at the field and method level. --- docs/developers-guide.md | 27 ++++++++++++++++++++++---- src/engine/connection/exec/protocol.rs | 3 +++ src/engine/connection/exec/session.rs | 3 +++ 3 files changed, 29 insertions(+), 4 deletions(-) diff --git a/docs/developers-guide.md b/docs/developers-guide.md index eaa5d48f..fbe7bfc3 100644 --- a/docs/developers-guide.md +++ b/docs/developers-guide.md @@ -45,6 +45,12 @@ src/engine/connection/exec/ +-- protocol.rs # Protocol proxy loops, stdin forwarding, | # Stdout Purity Contract, STDIN_BUFFER_CAPACITY, | # STDIN_SETTLE_TIMEOUT, ProtocolProxyIo ++-- acp_helpers.rs # ACP initialize-frame rewriting; forwards the +| # first newline-delimited frame from host stdin, +| # masks terminal/* and fs/* clientCapabilities, +| # and re-serialises the JSON before forwarding +| # to the container; bounded by +| # MAX_FIRST_FRAME_BYTES +-- attached.rs # Attached-mode session, terminal resize, | # SIGWINCH handling, stdin echo forwarding +-- terminal.rs # Terminal size detection (stty), resize helpers, @@ -60,10 +66,11 @@ src/engine/connection/exec/ | # PODBOT_DISABLE_STDIN_FORWARDING_FOR_TESTS=1 +-- session.rs # ExecSessionOptions struct and | # protocol_session_options helper; controls -| # per-call session knobs (e.g. stdin forwarding -| # disable seam for tests; -| # with_protocol_stdin_forwarding_disabled(bool) -| # is compiled only for #[cfg(test)] builds +| # per-call session knobs (stdin forwarding +| # disable seam for tests via +| # with_protocol_stdin_forwarding_disabled(bool); +| # ACP initialize rewriting opt-in via +| # with_acp_initialize_rewrite_enabled(bool)) +-- runtime_helpers.rs # Blocking runtime helpers for synchronous exec | # wrappers; block_on_runtime detects nested | # Tokio contexts and routes to block_in_place @@ -84,6 +91,18 @@ src/engine/connection/exec/ +-- detached_helpers.rs # Detached mode test helpers +-- lifecycle_helpers.rs # Shared lifecycle fixtures +-- protocol_helpers.rs # Protocol-specific test helpers + +-- protocol_acp_tests.rs # Unit and shared BDD fixture helpers + | # for ACP capability masking; + | # RecordingInputWriter, frame + | # builders, run_forwarding harness, + | # and pub(super) BDD fixtures + +-- protocol_acp_bdd_tests.rs # rstest-bdd BDD scenarios for ACP + | # capability masking (masked + | # initialize, malformed pass-through, + | # unblocked pass-through) + +-- protocol_acp_forwarding_tests.rs # Forwarding path tests: rewrite + # enabled/disabled flag, oversized + # frame timeout safety ``` ### 3.1. Execution modes diff --git a/src/engine/connection/exec/protocol.rs b/src/engine/connection/exec/protocol.rs index bdf7e462..6d0fa2db 100644 --- a/src/engine/connection/exec/protocol.rs +++ b/src/engine/connection/exec/protocol.rs @@ -66,6 +66,8 @@ const STDIN_BUFFER_CAPACITY: usize = 65_536; #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] pub(super) struct ProtocolSessionOptions { disable_stdin_forwarding: bool, + /// When `true`, the first ACP `initialize` frame is rewritten to remove + /// `terminal` and `fs` capabilities before being forwarded to the container. rewrite_acp_initialize: bool, } @@ -77,6 +79,7 @@ impl ProtocolSessionOptions { } } + /// Disable or enable protocol stdin forwarding for this session. pub(super) const fn with_stdin_forwarding_disabled(mut self, disable: bool) -> Self { self.disable_stdin_forwarding = disable; self diff --git a/src/engine/connection/exec/session.rs b/src/engine/connection/exec/session.rs index 71e5fa01..0be3195e 100644 --- a/src/engine/connection/exec/session.rs +++ b/src/engine/connection/exec/session.rs @@ -7,6 +7,9 @@ use super::protocol::ProtocolSessionOptions; #[derive(Debug, Clone, Copy, Default)] pub(crate) struct ExecSessionOptions { disable_protocol_stdin_forwarding: bool, + /// When `true`, the first ACP `initialize` frame sent from host stdin is + /// rewritten to remove `terminal` and `fs` capabilities before being + /// forwarded to the container. rewrite_acp_initialize: bool, } From a0cec6d274eabf382294247917bdab0167ce6600 Mon Sep 17 00:00:00 2001 From: leynos Date: Mon, 4 May 2026 02:35:43 +0200 Subject: [PATCH 2/4] Document remaining ACP exec protocol items Add `///` doc comments to every previously undocumented item in `src/engine/connection/exec/protocol.rs`, `src/engine/connection/exec/session.rs`, and `src/engine/connection/exec/acp_helpers.rs` so that docstring coverage across the protocol-mode exec proxy and ACP first-frame helpers clears the 80% threshold. The new docs cover `ProtocolSessionOptions` and its `disable_stdin_forwarding` field, `ProtocolSessionOptions::new`, the `HeldOpenStdin` adapter, `ProtocolProxyIo::with_options`, the `run_protocol_session_async_with_options` entry point, and the `protocol_host_stdin`, `settle_stdin_forwarding_task`, `classify_stdin_forwarding_task_result`, `abort_stdin_forwarding_task`, `forward_host_stdin_to_exec_async`, `run_output_loop_async`, `handle_log_output_chunk`, and `write_output_chunk` helpers. In `session.rs` the `disable_protocol_stdin_forwarding` field and the `protocol_session_options` conversion gain matching docs. In `acp_helpers.rs` the `InitialFrameAction` and `ForwardUnchangedReason` enums, the `next_initial_frame_action`, `log_unmodified_forwarding`, `log_exceeded_maximum_size`, `log_eof_before_newline`, `forward_unmodified_initial_frame`, `forward_masked_initial_frame`, `log_masked_frame_forwarded`, `read_next_bounded_frame_chunk`, and `remove_masked_acp_capabilities` items are now documented. No logic, signatures, tests, or non-doc comments are changed. --- src/engine/connection/exec/acp_helpers.rs | 15 +++++++++++++++ src/engine/connection/exec/protocol.rs | 23 +++++++++++++++++++++++ src/engine/connection/exec/session.rs | 4 ++++ 3 files changed, 42 insertions(+) diff --git a/src/engine/connection/exec/acp_helpers.rs b/src/engine/connection/exec/acp_helpers.rs index 234e0826..9a5012d9 100644 --- a/src/engine/connection/exec/acp_helpers.rs +++ b/src/engine/connection/exec/acp_helpers.rs @@ -33,12 +33,14 @@ pub(super) const ACP_FILE_SYSTEM_CAPABILITY: &str = "fs"; /// which are masked before the frame is forwarded to the container. pub(super) const ACP_TERMINAL_CAPABILITY: &str = "terminal"; +/// Decision produced by each iteration of the first-frame scanning loop. enum InitialFrameAction { Continue, ForwardMasked, ForwardUnchanged(ForwardUnchangedReason), } +/// Reason why the first ACP frame is forwarded without capability rewriting. #[derive(Clone, Copy)] enum ForwardUnchangedReason { ExceededMaximumSize, @@ -71,6 +73,7 @@ where } } +/// Read the next chunk into `first_frame` and decide how to proceed. async fn next_initial_frame_action( buffered_stdin: &mut tokio::io::BufReader, first_frame: &mut Vec, @@ -100,6 +103,7 @@ where }) } +/// Emit a debug trace for a frame forwarded without rewriting. fn log_unmodified_forwarding(reason: ForwardUnchangedReason, bytes: usize) { match reason { ForwardUnchangedReason::ExceededMaximumSize => log_exceeded_maximum_size(bytes), @@ -107,6 +111,7 @@ fn log_unmodified_forwarding(reason: ForwardUnchangedReason, bytes: usize) { } } +/// Emit a debug trace when the first frame exceeded the size limit. fn log_exceeded_maximum_size(bytes: usize) { tracing::debug!( bytes, @@ -114,6 +119,7 @@ fn log_exceeded_maximum_size(bytes: usize) { ); } +/// Emit a debug trace when EOF was reached before a newline was found. fn log_eof_before_newline(bytes: usize) { tracing::debug!( bytes, @@ -121,6 +127,8 @@ fn log_eof_before_newline(bytes: usize) { ); } +/// Write `first_frame` to `input` unchanged after logging the forwarding +/// reason. async fn forward_unmodified_initial_frame( input: &mut Pin>, first_frame: &[u8], @@ -130,6 +138,7 @@ async fn forward_unmodified_initial_frame( input.write_all(first_frame).await } +/// Mask `first_frame` and write the result to `input`. async fn forward_masked_initial_frame( input: &mut Pin>, first_frame: &[u8], @@ -140,12 +149,15 @@ async fn forward_masked_initial_frame( Ok(()) } +/// Emit a debug trace when the first frame was masked before forwarding. fn log_masked_frame_forwarded(masked_frame: &[u8], first_frame: &[u8]) { if masked_frame != first_frame { tracing::debug!("ACP initialize frame masked and forwarded"); } } +/// Fill the internal buffer and copy up to the size limit into `first_frame`. +/// Returns `(bytes_consumed, has_complete_frame)`. async fn read_next_bounded_frame_chunk( buffered_stdin: &mut tokio::io::BufReader, first_frame: &mut Vec, @@ -229,6 +241,9 @@ pub(super) fn split_frame_line_ending(frame: &[u8]) -> (&[u8], &[u8]) { (frame, b"") } +/// Remove `terminal` and `fs` from `params.clientCapabilities` in an ACP +/// `initialize` message. Returns `true` when at least one capability was +/// removed. fn remove_masked_acp_capabilities(message: &mut Value) -> bool { if message.get("method").and_then(Value::as_str) != Some(ACP_INITIALIZE_METHOD) { return false; diff --git a/src/engine/connection/exec/protocol.rs b/src/engine/connection/exec/protocol.rs index 6d0fa2db..da7e1a40 100644 --- a/src/engine/connection/exec/protocol.rs +++ b/src/engine/connection/exec/protocol.rs @@ -63,8 +63,11 @@ const STDIN_SETTLE_TIMEOUT: Duration = Duration::from_millis(50); /// low while preventing unbounded accumulation during high-throughput scenarios. const STDIN_BUFFER_CAPACITY: usize = 65_536; +/// Per-session configuration knobs for protocol-mode exec proxy loops. #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] pub(super) struct ProtocolSessionOptions { + /// When `true`, host stdin is replaced by a held-open no-op reader so that + /// the process's inherited stdin is never forwarded to the container. disable_stdin_forwarding: bool, /// When `true`, the first ACP `initialize` frame is rewritten to remove /// `terminal` and `fs` capabilities before being forwarded to the container. @@ -72,6 +75,7 @@ pub(super) struct ProtocolSessionOptions { } impl ProtocolSessionOptions { + /// Create default protocol-session options with production behaviour. pub(super) const fn new() -> Self { Self { disable_stdin_forwarding: false, @@ -92,6 +96,9 @@ impl ProtocolSessionOptions { } } +/// An `AsyncRead` adapter that stays open indefinitely without producing +/// bytes, used to suppress inherited process stdin while the forwarding task +/// remains alive. struct HeldOpenStdin { // `tokio::io::duplex(1)` gives us the smallest possible in-memory pipe to // keep `reader` and `_writer_guard` alive without producing an immediate @@ -126,12 +133,15 @@ impl ProtocolProxyIo Self { self.options = options; self } } +/// Run a protocol exec session using real host stdio and the given session +/// options. pub(super) async fn run_protocol_session_async_with_options( request: &ExecRequest, output: Pin> + Send>>, @@ -147,6 +157,8 @@ pub(super) async fn run_protocol_session_async_with_options( run_protocol_session_with_io_async(request, output, input, stdio).await } +/// Return the appropriate stdin reader for a protocol session. Yields a +/// held-open no-op reader when stdin forwarding is disabled. fn protocol_host_stdin(options: ProtocolSessionOptions) -> Pin> { if options.disable_stdin_forwarding || stdin_forwarding_disabled_for_tests() { let (writer_guard, reader) = tokio::io::duplex(1); @@ -195,6 +207,8 @@ where stdin_result } +/// Wait for the stdin forwarding task to complete within a short grace +/// period. async fn settle_stdin_forwarding_task( container_id: &str, mut stdin_task: JoinHandle>, @@ -219,6 +233,7 @@ async fn settle_stdin_forwarding_task( classify_stdin_forwarding_task_result(container_id, join_result) } +/// Map a join result from the stdin forwarding task to a `PodbotError`. fn classify_stdin_forwarding_task_result( container_id: &str, join_result: Result, tokio::task::JoinError>, @@ -237,6 +252,7 @@ fn classify_stdin_forwarding_task_result( } } +/// Abort and drop the stdin forwarding task without awaiting it. fn abort_stdin_forwarding_task(stdin_task: JoinHandle>) { if !stdin_task.is_finished() { stdin_task.abort(); @@ -247,6 +263,8 @@ fn abort_stdin_forwarding_task(stdin_task: JoinHandle>) { } } +/// Copy host stdin to the container exec input, optionally rewriting the +/// first ACP `initialize` frame before the raw copy begins. async fn forward_host_stdin_to_exec_async( host_stdin: HostStdin, mut input: Pin>, @@ -270,6 +288,8 @@ where #[path = "protocol_acp_tests.rs"] mod acp_tests; +/// Drain the container output stream, routing each chunk to host stdout or +/// stderr. async fn run_output_loop_async( container_id: &str, output: &mut Pin> + Send>>, @@ -289,6 +309,7 @@ where Ok(()) } +/// Route a single container log-output chunk to the appropriate host stream. async fn handle_log_output_chunk( container_id: &str, chunk: LogOutput, @@ -310,6 +331,8 @@ where } } +/// Write and flush a byte slice to `writer`, mapping I/O failures to +/// `PodbotError`. async fn write_output_chunk( container_id: &str, writer: &mut Writer, diff --git a/src/engine/connection/exec/session.rs b/src/engine/connection/exec/session.rs index 0be3195e..0ef1e966 100644 --- a/src/engine/connection/exec/session.rs +++ b/src/engine/connection/exec/session.rs @@ -6,6 +6,8 @@ use super::protocol::ProtocolSessionOptions; /// stream behaviour. #[derive(Debug, Clone, Copy, Default)] pub(crate) struct ExecSessionOptions { + /// When `true`, protocol-mode sessions replace host stdin with a held-open + /// no-op reader so that the process's inherited stdin is not forwarded. disable_protocol_stdin_forwarding: bool, /// When `true`, the first ACP `initialize` frame sent from host stdin is /// rewritten to remove `terminal` and `fs` capabilities before being @@ -47,6 +49,8 @@ impl ExecSessionOptions { } } +/// Convert exec-session options into the lower-level [`ProtocolSessionOptions`] +/// consumed by the protocol proxy loop. pub(super) const fn protocol_session_options( options: ExecSessionOptions, ) -> ProtocolSessionOptions { From b1884a74d37e526fc6c6be337903fd7270d377c4 Mon Sep 17 00:00:00 2001 From: leynos Date: Mon, 4 May 2026 18:17:34 +0200 Subject: [PATCH 3/4] Document private fields and variants in ACP exec proxy MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add `///` doc comments to the previously undocumented private fields of `ProtocolProxyIo` and `HeldOpenStdin` in `src/engine/connection/exec/protocol.rs`, to the `poll_read` implementation on `impl AsyncRead for HeldOpenStdin`, and to the variants of `InitialFrameAction` and `ForwardUnchangedReason` in `src/engine/connection/exec/acp_helpers.rs`. The pre-existing `//` block comment on `HeldOpenStdin` is replaced by field-level `///` docs that capture the same intent — the duplex reader never yields bytes, and the writer guard keeps the reader from seeing EOF — so docstring coverage clears the threshold without losing the explanation. No logic, signatures, tests, or other comments are altered. --- src/engine/connection/exec/acp_helpers.rs | 5 +++++ src/engine/connection/exec/protocol.rs | 11 +++++++---- 2 files changed, 12 insertions(+), 4 deletions(-) diff --git a/src/engine/connection/exec/acp_helpers.rs b/src/engine/connection/exec/acp_helpers.rs index 9a5012d9..aaa95180 100644 --- a/src/engine/connection/exec/acp_helpers.rs +++ b/src/engine/connection/exec/acp_helpers.rs @@ -35,15 +35,20 @@ pub(super) const ACP_TERMINAL_CAPABILITY: &str = "terminal"; /// Decision produced by each iteration of the first-frame scanning loop. enum InitialFrameAction { + /// Keep reading; the complete first frame has not yet been buffered. Continue, + /// A complete frame was buffered; rewrite it before forwarding. ForwardMasked, + /// Forward the buffered bytes unchanged for the given reason. ForwardUnchanged(ForwardUnchangedReason), } /// Reason why the first ACP frame is forwarded without capability rewriting. #[derive(Clone, Copy)] enum ForwardUnchangedReason { + /// The frame exceeded [`MAX_FIRST_FRAME_BYTES`] before a newline was found. ExceededMaximumSize, + /// Host stdin reached EOF before a newline-delimited frame was complete. EofBeforeNewline, } diff --git a/src/engine/connection/exec/protocol.rs b/src/engine/connection/exec/protocol.rs index da7e1a40..116f9dc9 100644 --- a/src/engine/connection/exec/protocol.rs +++ b/src/engine/connection/exec/protocol.rs @@ -42,9 +42,13 @@ use crate::error::PodbotError; /// Host-side stdio handles used by the protocol byte proxy. pub(super) struct ProtocolProxyIo { + /// Host stdin reader supplied to the forwarding task. stdin: HostStdin, + /// Host stdout writer used for container stdout and console output. stdout: HostStdout, + /// Host stderr writer used for container stderr and error diagnostics. stderr: HostStderr, + /// Per-session configuration knobs applied to this proxy run. options: ProtocolSessionOptions, } @@ -100,15 +104,14 @@ impl ProtocolSessionOptions { /// bytes, used to suppress inherited process stdin while the forwarding task /// remains alive. struct HeldOpenStdin { - // `tokio::io::duplex(1)` gives us the smallest possible in-memory pipe to - // keep `reader` and `_writer_guard` alive without producing an immediate - // EOF. No bytes are ever written through this seam; the 1-byte buffer just - // satisfies Tokio's duplex constructor while avoiding wasted memory. + /// The read half of the in-memory duplex pipe; never yields bytes. reader: tokio::io::DuplexStream, + /// Keeps the write half alive so `reader` never sees EOF. _writer_guard: tokio::io::DuplexStream, } impl tokio::io::AsyncRead for HeldOpenStdin { + /// Delegates to the inner duplex reader, which blocks indefinitely without producing bytes. fn poll_read( mut self: Pin<&mut Self>, cx: &mut Context<'_>, From c951451d6c332c46672dc1a55828d6dbbec5f72c Mon Sep 17 00:00:00 2001 From: leynos Date: Tue, 5 May 2026 11:59:57 +0200 Subject: [PATCH 4/4] Document concurrency model in ACP exec protocol module Co-authored-by: factory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com> --- src/engine/connection/exec/protocol.rs | 37 ++++++++++++++++++++++++++ 1 file changed, 37 insertions(+) diff --git a/src/engine/connection/exec/protocol.rs b/src/engine/connection/exec/protocol.rs index 116f9dc9..602599e2 100644 --- a/src/engine/connection/exec/protocol.rs +++ b/src/engine/connection/exec/protocol.rs @@ -19,6 +19,43 @@ //! `handle_log_output_chunk` for `LogOutput::StdOut` and `LogOutput::Console` //! messages from the container. All other code paths route to stderr or return //! errors without touching stdout. +//! +//! ## Concurrency Model +//! +//! `run_protocol_session_with_io_async` runs two concurrent paths: +//! +//! 1. **Stdin forwarding task** — spawned via `spawn_stdin_forwarding_task`, +//! which takes ownership of the host-stdin reader and the container-input +//! writer. The task runs on the Tokio thread pool and is represented by a +//! `JoinHandle>`. Ownership of the handle is retained by the +//! caller function for the lifetime of the session. +//! +//! 2. **Output loop** — driven directly on the caller task via +//! `run_output_loop_async`. It polls the container output stream to +//! completion before any stdin shutdown logic runs. +//! +//! ### Task coordination +//! +//! The output loop is awaited to completion first. Once it returns (success or +//! error), `settle_stdin_forwarding_task` awaits the stdin handle with a +//! [`STDIN_SETTLE_TIMEOUT`] deadline of 50 ms. This grace period covers the +//! common case where container EOF propagates to the forwarding task within +//! normal pipe-flush timings. +//! +//! ### Timeout and cancellation +//! +//! If the grace period expires, `abort_stdin_forwarding_task` calls +//! `JoinHandle::abort` and immediately drops the handle **without awaiting +//! it**. Awaiting after abort would block indefinitely if host stdin is stalled +//! in a non-cancellable kernel read; dropping the handle mirrors the +//! attached-session teardown path and keeps shutdown bounded. +//! +//! A `JoinError` carrying `is_cancelled()` is mapped to `Ok(())` because an +//! explicit abort is an intentional shutdown signal, not a failure. +//! +//! When `disable_stdin_forwarding` is set, the forwarding task reads from a +//! `HeldOpenStdin` duplex adapter that never yields bytes and never closes. +//! In that mode a timeout is expected and is silently treated as success. use std::io; use std::pin::Pin;