Document ACP capability masking modules and session knobs

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

src/engine/connection/exec/acp_helpers.rs

@@ -33,15 +33,22 @@ 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 {
+    /// 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,
 }
 
@@ -71,6 +78,7 @@ where
     }
 }
 
+/// Read the next chunk into `first_frame` and decide how to proceed.
 async fn next_initial_frame_action<HostStdin>(
     buffered_stdin: &mut tokio::io::BufReader<HostStdin>,
     first_frame: &mut Vec<u8>,
@@ -100,27 +108,32 @@ 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),
         ForwardUnchangedReason::EofBeforeNewline => log_eof_before_newline(bytes),
     }
 }
 
+/// Emit a debug trace when the first frame exceeded the size limit.
 fn log_exceeded_maximum_size(bytes: usize) {
     tracing::debug!(
         bytes,
         "ACP first frame exceeded maximum size; forwarding without rewrite"
     );
 }
 
+/// Emit a debug trace when EOF was reached before a newline was found.
 fn log_eof_before_newline(bytes: usize) {
     tracing::debug!(
         bytes,
         "ACP stdin reached EOF before newline; forwarding without rewrite"
     );
 }
 
+/// Write `first_frame` to `input` unchanged after logging the forwarding
+/// reason.
 async fn forward_unmodified_initial_frame(
     input: &mut Pin<Box<dyn AsyncWrite + Send>>,
     first_frame: &[u8],
@@ -130,6 +143,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<Box<dyn AsyncWrite + Send>>,
     first_frame: &[u8],
@@ -140,12 +154,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<HostStdin>(
     buffered_stdin: &mut tokio::io::BufReader<HostStdin>,
     first_frame: &mut Vec<u8>,
@@ -229,6 +246,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;

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<io::Result<()>>`. 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;
@@ -42,9 +79,13 @@ use crate::error::PodbotError;
 
 /// Host-side stdio handles used by the protocol byte proxy.
 pub(super) struct ProtocolProxyIo<HostStdin, HostStdout, HostStderr> {
+    /// 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,
 }
 
@@ -63,20 +104,27 @@ 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.
     rewrite_acp_initialize: bool,
 }
 
 impl ProtocolSessionOptions {
+    /// Create default protocol-session options with production behaviour.
     pub(super) const fn new() -> Self {
         Self {
             disable_stdin_forwarding: false,
             rewrite_acp_initialize: false,
         }
     }
 
+    /// 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
@@ -89,16 +137,18 @@ 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
-    // 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<'_>,
@@ -123,12 +173,15 @@ impl<HostStdin, HostStdout, HostStderr> ProtocolProxyIo<HostStdin, HostStdout, H
         }
     }
 
+    /// Attach session options to this host-IO bundle.
     pub(super) const fn with_options(mut self, options: ProtocolSessionOptions) -> 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<Box<dyn Stream<Item = Result<LogOutput, BollardError>> + Send>>,
@@ -144,6 +197,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<Box<dyn AsyncRead + Send>> {
     if options.disable_stdin_forwarding || stdin_forwarding_disabled_for_tests() {
         let (writer_guard, reader) = tokio::io::duplex(1);
@@ -192,6 +247,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<io::Result<()>>,
@@ -216,6 +273,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<io::Result<()>, tokio::task::JoinError>,
@@ -234,6 +292,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<io::Result<()>>) {
     if !stdin_task.is_finished() {
         stdin_task.abort();
@@ -244,6 +303,8 @@ fn abort_stdin_forwarding_task(stdin_task: JoinHandle<io::Result<()>>) {
     }
 }
 
+/// 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<HostStdin>(
     host_stdin: HostStdin,
     mut input: Pin<Box<dyn AsyncWrite + Send>>,
@@ -267,6 +328,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<HostStdout, HostStderr>(
     container_id: &str,
     output: &mut Pin<Box<dyn Stream<Item = Result<LogOutput, BollardError>> + Send>>,
@@ -286,6 +349,7 @@ where
     Ok(())
 }
 
+/// Route a single container log-output chunk to the appropriate host stream.
 async fn handle_log_output_chunk<HostStdout, HostStderr>(
     container_id: &str,
     chunk: LogOutput,
@@ -307,6 +371,8 @@ where
     }
 }
 
+/// Write and flush a byte slice to `writer`, mapping I/O failures to
+/// `PodbotError`.
 async fn write_output_chunk<Writer>(
     container_id: &str,
     writer: &mut Writer,

src/engine/connection/exec/session.rs

@@ -6,7 +6,12 @@ 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
+    /// forwarded to the container.
     rewrite_acp_initialize: bool,
 }
 
@@ -44,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 {