From aa90773ad8f6e91fa6da959b3f0adab90dd290e2 Mon Sep 17 00:00:00 2001 From: Leynos Date: Tue, 21 Apr 2026 12:58:43 +0000 Subject: [PATCH 1/6] Mask ACP initialize capabilities in protocol mode Intercept the first newline-delimited Agentic Control Protocol (ACP) `initialize` frame from protocol stdin and remove host-delegated `terminal` and `fs` capabilities before the container sees the metadata. Bound the first-frame inspection buffer so oversized or unterminated input streams forward promptly into the container instead of waiting indefinitely. Document the ACP masking trust boundary, add unit and BDD coverage, and mark roadmap Step 2.6.1 as complete. --- .../2-6-1-intercept-acp-initialization.md | 343 +++++++++++++ docs/podbot-design.md | 7 + docs/podbot-roadmap.md | 2 +- docs/users-guide.md | 11 +- src/engine/connection/exec/protocol.rs | 131 ++++- .../connection/exec/protocol_acp_tests.rs | 458 ++++++++++++++++++ tests/features/acp_capability_masking.feature | 22 + 7 files changed, 971 insertions(+), 3 deletions(-) create mode 100644 docs/execplans/2-6-1-intercept-acp-initialization.md create mode 100644 src/engine/connection/exec/protocol_acp_tests.rs create mode 100644 tests/features/acp_capability_masking.feature diff --git a/docs/execplans/2-6-1-intercept-acp-initialization.md b/docs/execplans/2-6-1-intercept-acp-initialization.md new file mode 100644 index 00000000..99945c06 --- /dev/null +++ b/docs/execplans/2-6-1-intercept-acp-initialization.md @@ -0,0 +1,343 @@ +# Step 2.6.1: Intercept Agentic Control Protocol (ACP) initialization + +This ExecPlan (execution plan) is a living document. The sections +`Constraints`, `Tolerances`, `Risks`, `Progress`, `Surprises and discoveries`, +`Decision log`, and `Outcomes and retrospective` must be kept up to date as +work proceeds. + +Status: COMPLETE (2026-04-21) + +No `PLANS.md` file exists in this repository as of 2026-04-21, so this ExecPlan +is the governing implementation document for this task. + +## Purpose and big picture + +Complete the first implementation task under Step 2.6 from +`docs/podbot-roadmap.md`: intercept ACP initialization and mask `terminal/*` +and `fs/*` capabilities before forwarding capability metadata into the +container. + +Podbot already had a protocol-safe exec path in +`src/engine/connection/exec/protocol.rs`, but that path was intentionally a raw +byte proxy. That was correct for generic stdio protocols, yet insufficient for +ACP hosting because an IDE-side ACP client can advertise host terminal and +filesystem delegation features during `initialize`. Forwarding those +capabilities unchanged would let the hosted agent discover authority that +Podbot is meant to withhold across the sandbox boundary. + +Observable success for this task: + +- the protocol stdin proxy inspects the first newline-delimited ACP frame; +- an ACP `initialize` request has `params.clientCapabilities.terminal` and + `params.clientCapabilities.fs` removed before the bytes reach the container; +- unrelated capabilities remain intact; +- malformed or non-ACP first frames pass through unchanged so protocol + transparency remains intact for other transports; +- existing protocol-mode guarantees remain intact: no TTY, no resize + propagation, and accurate exit-code reporting; +- unit tests with `rstest` and behavioural tests with `rstest-bdd` v0.5.0 + cover happy, unhappy, and edge paths; +- `docs/podbot-design.md` and `docs/users-guide.md` reflect the shipped + behaviour; +- only the relevant Step 2.6 roadmap checkbox is marked done. + +## Constraints + +- Scope is limited to Step 2.6.1, not the full Step 2.6 feature set. Do not + implement the later runtime denylist or delegation override here. +- Preserve the existing protocol proxy contract from Step 2.5: + stdout purity, stderr-only diagnostics, bounded buffering, and explicit stdin + shutdown ordering. +- Preserve attached-mode terminal handling, including initial resize, Unix + `SIGWINCH` propagation, and current exit-code reporting. +- Do not add a new dependency. Reuse existing workspace crates and reexports. +- Keep touched Rust modules below the repository's 400-line guidance; prefer + small focused test modules over inflating an existing file. +- Use `rstest` fixtures and parameterized cases for unit coverage. +- Use `rstest-bdd` v0.5.0 for behavioural coverage. +- Keep production code panic-free; use tolerant pass-through behaviour when ACP + parsing fails. +- Use en-GB-oxendict spelling in comments and documentation. +- Run documentation gates because this change updates Markdown files: + `make fmt`, `make markdownlint`, and `make nixie`. +- Run the requested Rust gates before completion: + `make check-fmt`, `make lint`, and `make test`. + +## Tolerances (exception triggers) + +- Scope tolerance: stop and escalate if the work requires the `podbot host` + command to exist before masking can be validated. +- Interface tolerance: stop and escalate if finishing the task requires a + public API break rather than an internal proxy-seam change. +- Protocol tolerance: stop and escalate if masking generic protocol mode would + corrupt non-ACP traffic rather than safely identifying ACP `initialize`. +- Dependency tolerance: stop and escalate before adding a new crate. +- Iteration tolerance: if the full gate stack still fails after three focused + fix passes, document the blocker and escalate. + +## Risks + +- Risk: `ExecMode::Protocol` is generic, not ACP-specific. Blindly rewriting + all protocol stdin would be too broad. Severity: high. Likelihood: high. + Mitigation: only inspect the first newline-delimited frame and only rewrite a + message that actually looks like ACP `initialize` with + `params.clientCapabilities`. + +- Risk: parsing and rewriting the first frame could break the byte-preserving + proxy contract. Severity: medium. Likelihood: medium. Mitigation: limit the + rewrite to the first frame only, preserve line endings, and leave malformed + or non-ACP frames unchanged. + +- Risk: feature-file edits for `rstest-bdd` are compile-time inputs, so stale + generated code can make scenario matching look broken. Severity: medium. + Likelihood: medium. Mitigation: keep scenario titles synchronized with the + feature file and use a clean rebuild only if the generated bindings appear + stale. + +- Risk: adding ACP-specific coverage on top of the existing protocol proxy + tests could duplicate too much helper code. Severity: low. Likelihood: + medium. Mitigation: keep focused masking tests in + `src/engine/connection/exec/protocol_acp_tests.rs`. + +## Context and orientation + +Relevant design and implementation anchors: + +- `docs/podbot-roadmap.md` Step 2.6 defines ACP capability masking as the next + container-engine integration task. +- `docs/podbot-design.md` already commits Podbot to masking ACP + `terminal/*` and `fs/*` capabilities by default. +- `src/engine/connection/exec/protocol.rs` is the current library seam for + protocol-safe stdin/stdout/stderr proxying. +- `src/engine/connection/exec/mod.rs` dispatches `ExecMode::Protocol` through + that proxy and remains the single place that waits for daemon exit codes. +- `src/engine/connection/exec/attached.rs` and + `src/engine/connection/exec/terminal.rs` own interactive terminal-only + behaviour and must remain unchanged in semantics. + +Before this change, the protocol stdin path was: + +1. Wrap host stdin in a bounded `BufReader`. +2. Copy bytes directly into container stdin. +3. Flush and shut down container stdin. + +After this change, the stdin path becomes: + +1. Wrap host stdin in the existing bounded `BufReader`. +2. Read the first newline-delimited frame. +3. If it is an ACP `initialize` request with `clientCapabilities.terminal` or + `clientCapabilities.fs`, rewrite that JSON frame with those keys removed. +4. Write the first frame to container stdin. +5. Resume the normal byte-transparent copy loop for the remainder of stdin. + +This keeps the ACP-specific logic at the narrowest seam that can be tested +without waiting for the future `podbot host` command implementation. + +## Agent team execution model + +Use a small agent team so planning and implementation stay parallel but +reviewable. + +Lane A (docs and roadmap reconnaissance owner): + +- Read the roadmap, design, and testing guidance. +- Confirm what behaviour and documentation updates are required. + +Lane B (code-path reconnaissance owner): + +- Trace `ExecMode::Protocol`, terminal handling, resize propagation, and exit + code reporting. +- Confirm where ACP masking can land without disturbing interactive sessions. + +Lane C (primary implementation owner in the main thread): + +- Draft this ExecPlan. +- Implement ACP masking at the protocol proxy seam. +- Add unit and behavioural coverage. +- Update documentation and run all gates. + +Coordination rule: + +- Use the reconnaissance lanes to reduce uncertainty up front, then keep all + code edits and final integration in the main thread so the change remains + coherent and atomic. + +## Plan of work + +### Stage A: Confirm the correct landing zone + +Verify that the protocol proxy seam, not the interactive terminal path, is the +correct place to intercept ACP `initialize`. + +Completed result: + +- Confirmed `ExecMode::Protocol` dispatches through + `src/engine/connection/exec/protocol.rs`. +- Confirmed attached-mode resize propagation and terminal handling are isolated + in `attached.rs` and `terminal.rs`. +- Confirmed no ACP-specific logic existed anywhere else in the codebase. + +### Stage B: Rewrite only ACP initialize capability metadata + +Add a narrow helper pipeline in `protocol.rs` that: + +- reads the first newline-delimited frame from host stdin; +- parses it as JSON if possible; +- rewrites only ACP `initialize` requests that expose + `params.clientCapabilities`; +- removes `terminal` and `fs` capability families; +- preserves unrelated capabilities and original line endings; +- forwards malformed or unrelated frames unchanged. + +Completed result: + +- Added ACP masking helpers in `src/engine/connection/exec/protocol.rs`. +- Kept the rewrite on the first frame only, then resumed the existing raw + proxy loop for the remainder of stdin. + +### Stage C: Add focused unit coverage + +Add `rstest` unit coverage at two levels: + +- direct helper tests for frame masking and line-ending preservation in + `src/engine/connection/exec/protocol_acp_tests.rs`; +- session-level proxy tests in + `src/engine/connection/exec/tests/proxy_helpers/acp_masking.rs`. + +Coverage requirements: + +- happy: ACP initialize removes `terminal` and `fs`; +- happy: unrelated capabilities remain; +- edge: empty `clientCapabilities` is removed after masking if nothing remains; +- edge: non-initialize and malformed messages are unchanged; +- happy: trailing protocol bytes after the masked frame remain unchanged; +- unhappy: input flush failures still surface correctly after ACP masking. + +### Stage D: Add behavioural coverage with `rstest-bdd` + +Add a dedicated ACP feature file and scenario bindings describing the operator- +visible behaviour in plain language. + +Completed result: + +- `tests/features/acp_capability_masking.feature` covers: + - blocked-capability masking; + - malformed initialize pass-through; + - initialize without blocked capabilities remaining unchanged. +- `src/engine/connection/exec/protocol_acp_tests.rs` binds those scenarios via + `rstest-bdd` v0.5.0. + +### Stage E: Update documentation and roadmap state + +Update: + +- `docs/podbot-design.md` to record the first-frame rewrite decision and the + unchanged pass-through rule for malformed or non-ACP frames; +- `docs/users-guide.md` to explain ACP masking within protocol mode; +- `docs/podbot-roadmap.md` to mark only the first Step 2.6 checkbox done. + +### Stage F: Validate the complete change + +Run the full gate stack with `tee` and `set -o pipefail`. + +```shell +set -o pipefail +make fmt 2>&1 | tee /tmp/podbot-make-fmt.log +make markdownlint 2>&1 | tee /tmp/podbot-make-markdownlint.log +make nixie 2>&1 | tee /tmp/podbot-make-nixie.log +make check-fmt 2>&1 | tee /tmp/podbot-make-check-fmt.log +make lint 2>&1 | tee /tmp/podbot-make-lint.log +make test 2>&1 | tee /tmp/podbot-make-test.log +``` + +## Validation and acceptance + +The task is complete only when all of the following are true: + +- ACP `initialize` capability metadata is masked before reaching the + containerised agent; +- malformed and non-ACP first frames remain unchanged; +- protocol stdout purity, non-TTY behaviour, and exit-code reporting remain + intact; +- `rstest` and `rstest-bdd` cover happy, unhappy, and edge paths; +- `docs/podbot-design.md` and `docs/users-guide.md` document the final + behaviour; +- only the relevant roadmap checkbox is marked done; +- the full documentation and Rust gate stack passes. + +## Idempotence and recovery + +- The masking logic is additive and local to `protocol.rs`. If interrupted, + revert only that seam rather than touching interactive exec code. +- If `.feature` edits appear to be ignored, do a clean rebuild before retrying + tests. +- Keep future Step 2.6 work separate: runtime denylist and explicit delegation + overrides should land in follow-on changes rather than stretching this one. + +## Progress + +- [x] (2026-04-21) Retrieved project memory, loaded repo guidance, and reviewed + roadmap/design/testing documents. +- [x] (2026-04-21) Used a lane-based agent-team plan inside the ExecPlan to + separate documentation, code-path, implementation, and validation work. +- [x] (2026-04-21) Confirmed `protocol.rs` is the correct seam for ACP + initialization masking. +- [x] (2026-04-21) Implemented first-frame ACP masking in + `src/engine/connection/exec/protocol.rs`. +- [x] (2026-04-21) Added focused `rstest` coverage in + `src/engine/connection/exec/protocol_acp_tests.rs`. +- [x] (2026-04-21) Added `rstest-bdd` coverage in + `tests/features/acp_capability_masking.feature`. +- [x] (2026-04-21) Updated `docs/podbot-design.md`, `docs/users-guide.md`, and + `docs/podbot-roadmap.md`. +- [x] (2026-04-21) Ran `make fmt`, `make markdownlint`, `make nixie`, + `make check-fmt`, `make lint`, and `make test` successfully. + +## Surprises and discoveries + +- Discovery: `rstest-bdd` feature files are compile-time inputs. Scenario-name + mismatches can reflect stale generated code rather than wrong Rust bindings, + so a clean rebuild is the recovery step if they ever diverge. +- Discovery: Step 2.6 can be delivered meaningfully before `podbot host` + exists by treating the protocol proxy as the library-owned ACP seam. + +## Decision log + +- Decision: keep masking in `src/engine/connection/exec/protocol.rs` instead + of waiting for `podbot host`. Rationale: that is the narrowest seam through + which ACP bytes already flow, and it is fully testable now. + +- Decision: rewrite only the first newline-delimited frame. Rationale: ACP + initialization occurs once at the start of the session, and limiting the + parser scope preserves the existing raw-proxy contract for later traffic. + +- Decision: treat masking failure as transparent pass-through, not a hard + protocol error. Rationale: Step 2.6.1 is about removing known capabilities, + not blocking or diagnosing later ACP misuse. A malformed first frame might + belong to another protocol, and guessing would be riskier than forwarding it. + +- Decision: remove `params.clientCapabilities` entirely when masking leaves it + empty. Rationale: ACP treats omitted capabilities as unsupported, so dropping + an empty object is semantically accurate and keeps the forwarded frame tidy. + +- Decision: mark only the first Step 2.6 roadmap checkbox done. Rationale: the + runtime denylist, stderr denials, and explicit override are still future work. + +## Outcomes and retrospective + +- Shipped: ACP capability masking now occurs before the container sees the + hosted client's `initialize` request. +- Shipped: protocol sessions still preserve non-TTY execution, avoid resize + handling, and report exit codes through the existing inspect loop. +- Shipped: happy, unhappy, and edge coverage now exists at both unit and + behavioural levels. +- Documentation: the design document and users guide now describe the actual + masking behaviour, including unchanged forwarding for malformed or unrelated + first frames. +- Roadmap: the Step 2.6 checkbox for intercepting ACP initialization is now + marked done, while later Step 2.6 work remains open. +- Gates passed: `make fmt`, `make markdownlint`, `make nixie`, + `make check-fmt`, `make lint`, and `make test`. +- Remaining follow-up: implement the runtime denylist, protocol errors for + blocked methods, and an explicit delegation override in subsequent Step 2.6 + changes. diff --git a/docs/podbot-design.md b/docs/podbot-design.md index b365650b..bacae44d 100644 --- a/docs/podbot-design.md +++ b/docs/podbot-design.md @@ -212,6 +212,13 @@ ACP masking is an implementation requirement, not a documentation preference: - Podbot must parse the ACP initialization handshake and remove blocked capability families before forwarding capability data to the hosted agent. +- The current implementation performs that masking by reading the first + newline-delimited ACP frame from host stdin, rewriting an `initialize` + request when `params.clientCapabilities` advertises `terminal` or `fs`, and + then resuming the normal raw byte proxy for the rest of the session. +- If the first protocol frame is malformed or not an ACP `initialize` request, + Podbot forwards it unchanged rather than guessing another protocol's + semantics. - Podbot must maintain a runtime denylist for blocked ACP methods and return a protocol error if those methods are attempted later in the session. - The delegation override must be explicit, disabled by default, and surfaced diff --git a/docs/podbot-roadmap.md b/docs/podbot-roadmap.md index 4b620b0f..491acbac 100644 --- a/docs/podbot-roadmap.md +++ b/docs/podbot-roadmap.md @@ -189,7 +189,7 @@ Prevent ACP client-side delegation from bypassing container sandbox boundaries. **Tasks:** -- [ ] Intercept ACP initialization and mask `terminal/*` and `fs/*` capabilities +- [x] Intercept ACP initialization and mask `terminal/*` and `fs/*` capabilities by default before forwarding capability metadata. - [ ] Enforce a runtime denylist for blocked ACP methods after initialization. - [ ] Return protocol errors for blocked methods and record denials on stderr. diff --git a/docs/users-guide.md b/docs/users-guide.md index 2cf72d79..5ac50283 100644 --- a/docs/users-guide.md +++ b/docs/users-guide.md @@ -63,6 +63,11 @@ This subcommand is temporarily unavailable in the current release. If you run session. Use `podbot run` or the library exec API for now; the release notes will call out when hosted mode becomes available. +When hosted mode uses ACP, podbot masks host-delegated `terminal/*` and `fs/*` +capabilities from the initial ACP `initialize` request before forwarding it to +the sandboxed agent. This keeps the default trust boundary aligned with the +container sandbox even when the hosting client advertises broader ACP support. + | Option | Required | Default | Description | | -------------- | -------- | --------------- | ------------------------------------------ | | `--agent` | No | Config/defaults | Agent type: `claude`, `codex`, or `custom` | @@ -124,7 +129,11 @@ Execution behaviour: host stderr without terminal framing or interactive echo injection. Protocol mode uses bounded buffering (64 KiB buffers for stdin forwarding and output chunks) so hosted protocols can apply backpressure; if the host stdout writer - blocks, the proxy yields and backpressure propagates to the container. + blocks, the proxy yields and backpressure propagates to the container. When + the proxied protocol is ACP, podbot rewrites the first `initialize` request + to remove `clientCapabilities.terminal` and `clientCapabilities.fs` before + the bytes reach the container. If that first frame is malformed or belongs to + another protocol, podbot forwards it unchanged. - TTY allocation is enabled only when attached mode is selected and both local stdin and stdout are terminals. - When TTY is enabled, podbot sends an initial resize to the daemon. On Unix diff --git a/src/engine/connection/exec/protocol.rs b/src/engine/connection/exec/protocol.rs index 6f9c9efb..6f70fb8c 100644 --- a/src/engine/connection/exec/protocol.rs +++ b/src/engine/connection/exec/protocol.rs @@ -28,7 +28,8 @@ use std::time::Duration; use bollard::container::LogOutput; use bollard::errors::Error as BollardError; use futures_util::{Stream, StreamExt}; -use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt}; +use ortho_config::serde_json::{self, Value}; +use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncWrite, AsyncWriteExt}; use tokio::task::JoinHandle; use tokio::time::timeout; @@ -59,6 +60,10 @@ const STDIN_SETTLE_TIMEOUT: Duration = Duration::from_millis(50); /// protocol message sizes and typical OS pipe buffer defaults, keeping latency /// low while preventing unbounded accumulation during high-throughput scenarios. const STDIN_BUFFER_CAPACITY: usize = 65_536; +const ACP_INITIALIZE_METHOD: &str = "initialize"; +const ACP_CLIENT_CAPABILITIES_FIELD: &str = "clientCapabilities"; +const ACP_FILE_SYSTEM_CAPABILITY: &str = "fs"; +const ACP_TERMINAL_CAPABILITY: &str = "terminal"; #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] pub(super) struct ProtocolSessionOptions { @@ -238,11 +243,135 @@ where HostStdin: AsyncRead + Unpin, { let mut buffered_stdin = tokio::io::BufReader::with_capacity(STDIN_BUFFER_CAPACITY, host_stdin); + forward_initial_protocol_frame_async(&mut buffered_stdin, &mut input).await?; tokio::io::copy(&mut buffered_stdin, &mut input).await?; input.flush().await?; input.shutdown().await } +async fn forward_initial_protocol_frame_async( + buffered_stdin: &mut tokio::io::BufReader, + input: &mut Pin>, +) -> io::Result<()> +where + HostStdin: AsyncRead + Unpin, +{ + let mut first_frame = Vec::new(); + loop { + if first_frame.len() == STDIN_BUFFER_CAPACITY { + input.write_all(&first_frame).await?; + return Ok(()); + } + + let (bytes_to_consume, has_complete_frame) = { + let buffered = buffered_stdin.fill_buf().await?; + if buffered.is_empty() { + return forward_partial_initial_frame(input, &first_frame).await; + } + + let remaining_capacity = STDIN_BUFFER_CAPACITY.saturating_sub(first_frame.len()); + let bytes_to_scan = buffered.len().min(remaining_capacity); + let newline_offset = buffered + .get(..bytes_to_scan) + .and_then(|bytes| bytes.iter().position(|byte| *byte == b'\n')); + let bytes_to_consume = newline_offset.map_or(bytes_to_scan, |offset| offset + 1); + let Some(bytes) = buffered.get(..bytes_to_consume) else { + return Err(io::Error::other( + "buffered stdin slice exceeded available input", + )); + }; + first_frame.extend_from_slice(bytes); + (bytes_to_consume, newline_offset.is_some()) + }; + + buffered_stdin.consume(bytes_to_consume); + + if has_complete_frame { + return input + .write_all(&mask_acp_initialize_frame(&first_frame)) + .await; + } + } +} + +async fn forward_partial_initial_frame( + input: &mut Pin>, + first_frame: &[u8], +) -> io::Result<()> { + if first_frame.is_empty() { + return Ok(()); + } + + input.write_all(first_frame).await +} + +fn mask_acp_initialize_frame(frame: &[u8]) -> Vec { + let (payload, line_ending) = split_frame_line_ending(frame); + + let Ok(mut message) = serde_json::from_slice(payload) else { + return frame.to_vec(); + }; + + if !remove_masked_acp_capabilities(&mut message) { + return frame.to_vec(); + } + + let Ok(mut serialized) = serde_json::to_vec(&message) else { + return frame.to_vec(); + }; + serialized.extend_from_slice(line_ending); + serialized +} + +fn split_frame_line_ending(frame: &[u8]) -> (&[u8], &[u8]) { + if let Some(stripped) = frame.strip_suffix(b"\r\n") { + return (stripped, b"\r\n"); + } + + if let Some(stripped) = frame.strip_suffix(b"\n") { + return (stripped, b"\n"); + } + + (frame, b"") +} + +fn remove_masked_acp_capabilities(message: &mut Value) -> bool { + if message.get("method").and_then(Value::as_str) != Some(ACP_INITIALIZE_METHOD) { + return false; + } + + let Some(params) = message.get_mut("params").and_then(Value::as_object_mut) else { + return false; + }; + + let Some(client_capabilities) = params + .get_mut(ACP_CLIENT_CAPABILITIES_FIELD) + .and_then(Value::as_object_mut) + else { + return false; + }; + + let removed_terminal = client_capabilities + .remove(ACP_TERMINAL_CAPABILITY) + .is_some(); + let removed_fs = client_capabilities + .remove(ACP_FILE_SYSTEM_CAPABILITY) + .is_some(); + if !removed_terminal && !removed_fs { + return false; + } + + if client_capabilities.is_empty() { + params.remove(ACP_CLIENT_CAPABILITIES_FIELD); + } + + true +} + +#[cfg(test)] +#[path = "protocol_acp_tests.rs"] +mod acp_tests; + async fn run_output_loop_async( container_id: &str, output: &mut Pin> + Send>>, diff --git a/src/engine/connection/exec/protocol_acp_tests.rs b/src/engine/connection/exec/protocol_acp_tests.rs new file mode 100644 index 00000000..1ba5b1a7 --- /dev/null +++ b/src/engine/connection/exec/protocol_acp_tests.rs @@ -0,0 +1,458 @@ +//! ACP capability masking tests for the protocol stdin proxy. + +use std::io; +use std::pin::Pin; +use std::sync::{Arc, Mutex}; +use std::task::{Context, Poll}; + +use rstest::rstest; +use rstest_bdd::Slot; +use rstest_bdd_macros::{ScenarioState, given, scenario, then, when}; +use tokio::io::{AsyncWrite, AsyncWriteExt, DuplexStream}; + +use super::*; + +type StepResult = Result; + +#[derive(Default, ScenarioState)] +struct AcpMaskingState { + host_stdin: Slot>, + expected_forwarded: Slot>, + actual_forwarded: Slot>, + succeeded: Slot, +} + +struct RecordingInputWriter { + bytes: Arc>>, + shutdown_called: Arc>, +} + +impl RecordingInputWriter { + fn new() -> Self { + Self { + bytes: Arc::new(Mutex::new(Vec::new())), + shutdown_called: Arc::new(Mutex::new(false)), + } + } +} + +impl AsyncWrite for RecordingInputWriter { + fn poll_write( + self: Pin<&mut Self>, + _cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + self.bytes + .lock() + .expect("writer mutex should not poison") + .extend_from_slice(buf); + Poll::Ready(Ok(buf.len())) + } + + fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + *self + .shutdown_called + .lock() + .expect("shutdown mutex should not poison") = true; + Poll::Ready(Ok(())) + } +} + +fn initialize_frame(line_ending: &str) -> Vec { + let payload = serde_json::json!({ + "jsonrpc": "2.0", + "id": 0, + "method": "initialize", + "params": { + "protocolVersion": 1, + "clientCapabilities": { + "fs": { + "readTextFile": true, + "writeTextFile": true + }, + "terminal": true, + "_meta": { + "custom": true + } + }, + "clientInfo": { + "name": "podbot-tests", + "version": "1.0.0" + } + } + }); + + let mut frame = serde_json::to_vec(&payload).expect("initialize payload should serialize"); + frame.extend_from_slice(line_ending.as_bytes()); + frame +} + +fn initialize_without_blocked_capabilities() -> Vec { + let payload = serde_json::json!({ + "jsonrpc": "2.0", + "id": 0, + "method": "initialize", + "params": { + "protocolVersion": 1, + "clientCapabilities": { + "_meta": { + "custom": true + } + } + } + }); + + let mut frame = serde_json::to_vec(&payload).expect("initialize payload should serialize"); + frame.push(b'\n'); + frame +} + +fn initialize_with_only_blocked_capabilities(line_ending: &str) -> Vec { + let payload = serde_json::json!({ + "jsonrpc": "2.0", + "id": 0, + "method": "initialize", + "params": { + "protocolVersion": 1, + "clientCapabilities": { + "fs": { + "readTextFile": true, + "writeTextFile": true + }, + "terminal": true + }, + "clientInfo": { + "name": "podbot-tests", + "version": "1.0.0" + } + } + }); + + let mut frame = serde_json::to_vec(&payload).expect("initialize payload should serialize"); + frame.extend_from_slice(line_ending.as_bytes()); + frame +} + +fn session_new_bytes() -> Vec { + br#"{"jsonrpc":"2.0","id":1,"method":"session/new","params":{"cwd":"/workspace"}}"#.to_vec() +} + +fn malformed_initialize_bytes() -> Vec { + br#"{"jsonrpc":"2.0","method":"initialize","params":{"clientCapabilities":{"terminal":true}}"# + .to_vec() +} + +fn parse_frame_payload(frame: &[u8]) -> serde_json::Value { + let (payload, _) = split_frame_line_ending(frame); + serde_json::from_slice(payload).expect("frame should contain JSON payload") +} + +fn assert_masked_client_capabilities(message: &serde_json::Value) { + let client_capabilities = message + .get("params") + .and_then(serde_json::Value::as_object) + .and_then(|params| params.get("clientCapabilities")) + .and_then(serde_json::Value::as_object) + .expect("clientCapabilities should remain present"); + assert!( + !client_capabilities.contains_key(ACP_FILE_SYSTEM_CAPABILITY), + "fs capability should be removed" + ); + assert!( + !client_capabilities.contains_key(ACP_TERMINAL_CAPABILITY), + "terminal capability should be removed" + ); + assert!( + client_capabilities.contains_key("_meta"), + "unrelated capabilities should remain" + ); +} + +async fn build_host_stdin(bytes: &[u8]) -> io::Result { + let capacity = bytes.len().max(1); + let (mut writer, reader) = tokio::io::duplex(capacity); + writer.write_all(bytes).await?; + drop(writer); + Ok(reader) +} + +fn run_forwarding(host_stdin_bytes: &[u8]) -> (Vec, bool) { + let runtime = tokio::runtime::Runtime::new().expect("runtime should build"); + let host_stdin = runtime + .block_on(build_host_stdin(host_stdin_bytes)) + .expect("host stdin should build"); + let container_input = RecordingInputWriter::new(); + let forwarded_bytes = container_input.bytes.clone(); + let shutdown_called = container_input.shutdown_called.clone(); + + runtime + .block_on(forward_host_stdin_to_exec_async( + host_stdin, + Box::pin(container_input), + )) + .expect("stdin forwarding should succeed"); + + ( + forwarded_bytes + .lock() + .expect("writer mutex should not poison") + .clone(), + *shutdown_called + .lock() + .expect("shutdown mutex should not poison"), + ) +} + +#[rstest] +#[case("\n")] +#[case("\r\n")] +fn mask_acp_initialize_frame_removes_blocked_capabilities(#[case] line_ending: &str) { + let frame = initialize_frame(line_ending); + let masked = mask_acp_initialize_frame(&frame); + let payload = parse_frame_payload(&masked); + + assert_eq!( + split_frame_line_ending(&masked).1, + line_ending.as_bytes(), + "line ending should be preserved" + ); + assert_masked_client_capabilities(&payload); +} + +#[test] +fn mask_acp_initialize_frame_removes_empty_client_capabilities() { + let frame = initialize_with_only_blocked_capabilities("\n"); + let masked = mask_acp_initialize_frame(&frame); + let payload = parse_frame_payload(&masked); + let params = payload + .get("params") + .and_then(serde_json::Value::as_object) + .expect("initialize params should remain present"); + + assert!( + !params.contains_key("clientCapabilities"), + "clientCapabilities should be removed when all entries are masked" + ); + assert_eq!( + params.get("protocolVersion"), + Some(&serde_json::json!(1)), + "protocolVersion should remain unchanged" + ); + assert_eq!( + params.get("clientInfo"), + Some(&serde_json::json!({ + "name": "podbot-tests", + "version": "1.0.0" + })), + "clientInfo should remain unchanged" + ); +} + +#[test] +fn mask_acp_initialize_frame_leaves_non_initialize_messages_unchanged() { + let mut frame = session_new_bytes(); + frame.push(b'\n'); + + assert_eq!(mask_acp_initialize_frame(&frame), frame); +} + +#[test] +fn mask_acp_initialize_frame_leaves_malformed_input_unchanged() { + let frame = malformed_initialize_bytes(); + + assert_eq!(mask_acp_initialize_frame(&frame), frame); +} + +#[test] +fn forwarding_masks_initialize_and_preserves_trailing_bytes() { + let mut host_stdin_bytes = initialize_frame("\n"); + let trailing = session_new_bytes(); + host_stdin_bytes.extend_from_slice(&trailing); + + let (forwarded, shutdown_called) = run_forwarding(&host_stdin_bytes); + let newline_index = forwarded + .iter() + .position(|byte| *byte == b'\n') + .expect("masked initialize should remain line terminated"); + let initialize_frame = forwarded + .get(..=newline_index) + .expect("masked initialize frame should remain addressable"); + let trailing_forwarded = forwarded + .get(newline_index + 1..) + .expect("trailing bytes should remain addressable"); + let payload = parse_frame_payload(initialize_frame); + + assert_masked_client_capabilities(&payload); + assert_eq!( + trailing_forwarded, + trailing.as_slice(), + "trailing bytes should pass through unchanged" + ); + assert!(shutdown_called, "stdin forwarding should shut down input"); +} + +#[test] +fn forwarding_does_not_wait_indefinitely_for_oversized_initial_frame() { + let runtime = tokio::runtime::Runtime::new().expect("runtime should build"); + let host_stdin_bytes = vec![b'x'; STDIN_BUFFER_CAPACITY + 1]; + let (host_writer, host_reader) = runtime + .block_on(async { + let (mut host_writer, host_reader) = tokio::io::duplex(host_stdin_bytes.len()); + host_writer.write_all(&host_stdin_bytes).await?; + io::Result::Ok((host_writer, host_reader)) + }) + .expect("host stdin should accept oversized initial bytes"); + + let mut buffered_stdin = + tokio::io::BufReader::with_capacity(STDIN_BUFFER_CAPACITY, host_reader); + let recording_input = RecordingInputWriter::new(); + let forwarded_bytes = recording_input.bytes.clone(); + let mut container_input: Pin> = Box::pin(recording_input); + + runtime + .block_on(async { + tokio::time::timeout( + STDIN_SETTLE_TIMEOUT, + forward_initial_protocol_frame_async(&mut buffered_stdin, &mut container_input), + ) + .await + }) + .expect("initial forwarding should not wait for newline or EOF") + .expect("initial forwarding should succeed"); + + assert_eq!( + forwarded_bytes + .lock() + .expect("writer mutex should not poison") + .len(), + STDIN_BUFFER_CAPACITY, + "only the bounded first-frame buffer should be held before streaming resumes" + ); + + drop(host_writer); +} + +#[rstest::fixture] +fn acp_masking_state() -> AcpMaskingState { + AcpMaskingState::default() +} + +fn masked_initialize_with_follow_up() -> (Vec, Vec) { + let mut host_stdin_bytes = initialize_frame("\n"); + host_stdin_bytes.extend_from_slice(&session_new_bytes()); + + let expected_initialize = serde_json::json!({ + "jsonrpc": "2.0", + "id": 0, + "method": "initialize", + "params": { + "protocolVersion": 1, + "clientCapabilities": { + "_meta": { + "custom": true + } + }, + "clientInfo": { + "name": "podbot-tests", + "version": "1.0.0" + } + } + }); + + let mut expected = serde_json::to_vec(&expected_initialize) + .expect("expected initialize payload should serialize"); + expected.push(b'\n'); + expected.extend_from_slice(&session_new_bytes()); + + (host_stdin_bytes, expected) +} + +#[given( + "ACP stdin contains an initialize request with blocked capabilities and a follow-up request" +)] +fn acp_stdin_contains_blocked_initialize_and_follow_up(acp_masking_state: &AcpMaskingState) { + let (host_stdin_bytes, expected) = masked_initialize_with_follow_up(); + acp_masking_state.host_stdin.set(host_stdin_bytes); + acp_masking_state.expected_forwarded.set(expected); +} + +#[given("ACP stdin contains malformed initialize bytes")] +fn acp_stdin_contains_malformed_initialize(acp_masking_state: &AcpMaskingState) { + let malformed = malformed_initialize_bytes(); + acp_masking_state.host_stdin.set(malformed.clone()); + acp_masking_state.expected_forwarded.set(malformed); +} + +#[given("ACP stdin contains initialize without blocked capabilities")] +fn acp_stdin_contains_safe_initialize(acp_masking_state: &AcpMaskingState) { + let initialize = initialize_without_blocked_capabilities(); + acp_masking_state.host_stdin.set(initialize.clone()); + acp_masking_state.expected_forwarded.set(initialize); +} + +#[when("ACP stdin forwarding runs")] +fn acp_stdin_forwarding_runs(acp_masking_state: &AcpMaskingState) -> StepResult<()> { + let host_stdin_bytes = acp_masking_state + .host_stdin + .get() + .ok_or_else(|| String::from("host stdin should be configured"))?; + let (forwarded, _) = run_forwarding(&host_stdin_bytes); + acp_masking_state.actual_forwarded.set(forwarded); + acp_masking_state.succeeded.set(true); + Ok(()) +} + +#[then("ACP stdin forwarding succeeds")] +fn acp_stdin_forwarding_succeeds(acp_masking_state: &AcpMaskingState) -> StepResult<()> { + if acp_masking_state.succeeded.get() == Some(true) { + Ok(()) + } else { + Err(String::from("expected ACP stdin forwarding to succeed")) + } +} + +#[then("the forwarded ACP stdin matches the expected bytes")] +fn forwarded_acp_stdin_matches_expected(acp_masking_state: &AcpMaskingState) -> StepResult<()> { + let actual = acp_masking_state + .actual_forwarded + .get() + .ok_or_else(|| String::from("forwarded bytes should be recorded"))?; + let expected = acp_masking_state + .expected_forwarded + .get() + .ok_or_else(|| String::from("expected bytes should be recorded"))?; + + if actual == expected { + Ok(()) + } else { + Err(format!("expected {expected:?}, got {actual:?}")) + } +} + +#[scenario( + path = "tests/features/acp_capability_masking.feature", + name = "ACP initialize masks blocked capabilities before forwarding" +)] +fn acp_initialize_masks_blocked_capabilities(acp_masking_state: AcpMaskingState) { + let _ = acp_masking_state; +} + +#[scenario( + path = "tests/features/acp_capability_masking.feature", + name = "Malformed ACP initialize is forwarded unchanged" +)] +fn malformed_acp_initialize_is_forwarded_unchanged(acp_masking_state: AcpMaskingState) { + let _ = acp_masking_state; +} + +#[scenario( + path = "tests/features/acp_capability_masking.feature", + name = "ACP initialize without blocked capabilities stays unchanged" +)] +fn acp_initialize_without_blocked_capabilities_stays_unchanged(acp_masking_state: AcpMaskingState) { + let _ = acp_masking_state; +} diff --git a/tests/features/acp_capability_masking.feature b/tests/features/acp_capability_masking.feature new file mode 100644 index 00000000..32e82ee7 --- /dev/null +++ b/tests/features/acp_capability_masking.feature @@ -0,0 +1,22 @@ +Feature: ACP capability masking + + ACP hosting must remove host-delegated terminal and filesystem capabilities + from the initialize request before the request reaches the sandboxed agent. + + Scenario: ACP initialize masks blocked capabilities before forwarding + Given ACP stdin contains an initialize request with blocked capabilities and a follow-up request + When ACP stdin forwarding runs + Then ACP stdin forwarding succeeds + And the forwarded ACP stdin matches the expected bytes + + Scenario: Malformed ACP initialize is forwarded unchanged + Given ACP stdin contains malformed initialize bytes + When ACP stdin forwarding runs + Then ACP stdin forwarding succeeds + And the forwarded ACP stdin matches the expected bytes + + Scenario: ACP initialize without blocked capabilities stays unchanged + Given ACP stdin contains initialize without blocked capabilities + When ACP stdin forwarding runs + Then ACP stdin forwarding succeeds + And the forwarded ACP stdin matches the expected bytes From 9a6a1bd707f15e8f827b2cefbdd5fac4e3c13f4e Mon Sep 17 00:00:00 2001 From: leynos Date: Thu, 30 Apr 2026 18:09:19 +0200 Subject: [PATCH 2/6] Address ACP protocol review findings Move ACP initialize masking helpers out of the protocol proxy module so the proxy loop stays focused and under the file-length budget. Keep ACP first-frame rewriting behind a protocol-session option so generic protocol sessions retain raw stdin forwarding semantics unless ACP masking is selected. Split ACP BDD scenario wiring out of the unit-test module, extend coverage for the first-frame-only contract, document the current hosted-mode ACP masking scope, and add the helper documentation and tracing needed by the review gates. --- docs/developers-guide.md | 10 + .../2-6-1-intercept-acp-initialization.md | 36 ++- docs/users-guide.md | 22 +- src/engine/connection/exec/acp_helpers.rs | 253 +++++++++++++++ src/engine/connection/exec/protocol.rs | 149 ++------- .../connection/exec/protocol_acp_bdd_tests.rs | 110 +++++++ .../connection/exec/protocol_acp_tests.rs | 288 +++++++++++------- src/engine/connection/exec/session.rs | 31 ++ 8 files changed, 637 insertions(+), 262 deletions(-) create mode 100644 src/engine/connection/exec/acp_helpers.rs create mode 100644 src/engine/connection/exec/protocol_acp_bdd_tests.rs diff --git a/docs/developers-guide.md b/docs/developers-guide.md index e9a50aa1..3aaf26ec 100644 --- a/docs/developers-guide.md +++ b/docs/developers-guide.md @@ -357,6 +357,16 @@ When adding a new execution mode: - `ContainerExecClient` mock implementations for unit testing without a live daemon. +#### 8.2.1. Agentic Control Protocol (ACP) initialization rewrite contract + +When protocol tests wrap `run_session` with `RecordingWriter` handles, they +must preserve the Agentic Control Protocol (ACP) initialization rewrite +contract: inspect only the first newline-delimited frame, rewrite only ACP +`initialize` requests that expose `params.clientCapabilities`, and remove the +`terminal` and `fs` capability families before the frame reaches the container. +Preserve the original line endings, and let malformed or non-ACP frames pass +through unchanged. + ### 8.3. Parameterized tests Use `#[rstest(...)]` to eliminate duplicated test cases. Group related diff --git a/docs/execplans/2-6-1-intercept-acp-initialization.md b/docs/execplans/2-6-1-intercept-acp-initialization.md index 99945c06..8148f99b 100644 --- a/docs/execplans/2-6-1-intercept-acp-initialization.md +++ b/docs/execplans/2-6-1-intercept-acp-initialization.md @@ -179,7 +179,8 @@ Completed result: ### Stage B: Rewrite only ACP initialize capability metadata -Add a narrow helper pipeline in `protocol.rs` that: +Add a narrow helper pipeline in +`src/engine/connection/exec/acp_helpers.rs` that: - reads the first newline-delimited frame from host stdin; - parses it as JSON if possible; @@ -191,7 +192,8 @@ Add a narrow helper pipeline in `protocol.rs` that: Completed result: -- Added ACP masking helpers in `src/engine/connection/exec/protocol.rs`. +- Added ACP masking helpers in + `src/engine/connection/exec/acp_helpers.rs`. - Kept the rewrite on the first frame only, then resumed the existing raw proxy loop for the remainder of stdin. @@ -202,7 +204,7 @@ Add `rstest` unit coverage at two levels: - direct helper tests for frame masking and line-ending preservation in `src/engine/connection/exec/protocol_acp_tests.rs`; - session-level proxy tests in - `src/engine/connection/exec/tests/proxy_helpers/acp_masking.rs`. + `src/engine/connection/exec/protocol_acp_tests.rs`. Coverage requirements: @@ -224,8 +226,8 @@ Completed result: - blocked-capability masking; - malformed initialize pass-through; - initialize without blocked capabilities remaining unchanged. -- `src/engine/connection/exec/protocol_acp_tests.rs` binds those scenarios via - `rstest-bdd` v0.5.0. +- `src/engine/connection/exec/protocol_acp_bdd_tests.rs` binds those scenarios + via `rstest-bdd` v0.5.0. ### Stage E: Update documentation and roadmap state @@ -255,7 +257,7 @@ make test 2>&1 | tee /tmp/podbot-make-test.log The task is complete only when all of the following are true: - ACP `initialize` capability metadata is masked before reaching the - containerised agent; + containerized agent; - malformed and non-ACP first frames remain unchanged; - protocol stdout purity, non-TTY behaviour, and exit-code reporting remain intact; @@ -267,8 +269,10 @@ The task is complete only when all of the following are true: ## Idempotence and recovery -- The masking logic is additive and local to `protocol.rs`. If interrupted, - revert only that seam rather than touching interactive exec code. +- The masking logic is additive and local to + `src/engine/connection/exec/acp_helpers.rs`, with regression coverage in + `src/engine/connection/exec/protocol_acp_tests.rs`. If interrupted, revert + only that seam rather than touching interactive exec code. - If `.feature` edits appear to be ignored, do a clean rebuild before retrying tests. - Keep future Step 2.6 work separate: runtime denylist and explicit delegation @@ -282,12 +286,13 @@ The task is complete only when all of the following are true: separate documentation, code-path, implementation, and validation work. - [x] (2026-04-21) Confirmed `protocol.rs` is the correct seam for ACP initialization masking. -- [x] (2026-04-21) Implemented first-frame ACP masking in - `src/engine/connection/exec/protocol.rs`. +- [x] (2026-04-21) Implemented first-frame ACP masking helpers in + `src/engine/connection/exec/acp_helpers.rs`. - [x] (2026-04-21) Added focused `rstest` coverage in `src/engine/connection/exec/protocol_acp_tests.rs`. - [x] (2026-04-21) Added `rstest-bdd` coverage in - `tests/features/acp_capability_masking.feature`. + `tests/features/acp_capability_masking.feature` with bindings in + `src/engine/connection/exec/protocol_acp_bdd_tests.rs`. - [x] (2026-04-21) Updated `docs/podbot-design.md`, `docs/users-guide.md`, and `docs/podbot-roadmap.md`. - [x] (2026-04-21) Ran `make fmt`, `make markdownlint`, `make nixie`, @@ -303,9 +308,12 @@ The task is complete only when all of the following are true: ## Decision log -- Decision: keep masking in `src/engine/connection/exec/protocol.rs` instead - of waiting for `podbot host`. Rationale: that is the narrowest seam through - which ACP bytes already flow, and it is fully testable now. +- Decision: keep the ACP interception seam in + `src/engine/connection/exec/protocol.rs` while moving masking helpers into + `src/engine/connection/exec/acp_helpers.rs`, instead of waiting for + `podbot host`. Rationale: the protocol proxy is the narrowest seam through + which ACP bytes already flow, and the helper module keeps that seam fully + testable without overloading the proxy loop. - Decision: rewrite only the first newline-delimited frame. Rationale: ACP initialization occurs once at the start of the session, and limiting the diff --git a/docs/users-guide.md b/docs/users-guide.md index 5ac50283..448c75a7 100644 --- a/docs/users-guide.md +++ b/docs/users-guide.md @@ -63,10 +63,14 @@ This subcommand is temporarily unavailable in the current release. If you run session. Use `podbot run` or the library exec API for now; the release notes will call out when hosted mode becomes available. -When hosted mode uses ACP, podbot masks host-delegated `terminal/*` and `fs/*` -capabilities from the initial ACP `initialize` request before forwarding it to -the sandboxed agent. This keeps the default trust boundary aligned with the -container sandbox even when the hosting client advertises broader ACP support. + +When hosted mode uses Agentic Control Protocol (ACP), podbot masks `terminal/*` and `fs/*` capabilities +from the initial ACP `initialize` request before forwarding it to the +sandboxed agent. This masking currently applies only to the +protocol/library path used by hosted mode and ACP until `podbot host` is +implemented. The scope and behaviour may change when `podbot host` ships, but +for now it keeps the default trust boundary aligned with the container +sandbox even when the hosting client advertises broader ACP support. | Option | Required | Default | Description | | -------------- | -------- | --------------- | ------------------------------------------ | @@ -129,11 +133,11 @@ Execution behaviour: host stderr without terminal framing or interactive echo injection. Protocol mode uses bounded buffering (64 KiB buffers for stdin forwarding and output chunks) so hosted protocols can apply backpressure; if the host stdout writer - blocks, the proxy yields and backpressure propagates to the container. When - the proxied protocol is ACP, podbot rewrites the first `initialize` request - to remove `clientCapabilities.terminal` and `clientCapabilities.fs` before - the bytes reach the container. If that first frame is malformed or belongs to - another protocol, podbot forwards it unchanged. + blocks, the proxy yields and backpressure propagates to the container. + `ExecMode::Protocol` preserves raw byte-forwards on the default path, so + podbot does not alter the first `initialize` frame there and forwards it + unchanged. ACP initialize-request rewriting occurs only on the hosted ACP + path for opt-in hosting; it is not a general `ExecMode::Protocol` guarantee. - TTY allocation is enabled only when attached mode is selected and both local stdin and stdout are terminals. - When TTY is enabled, podbot sends an initial resize to the daemon. On Unix diff --git a/src/engine/connection/exec/acp_helpers.rs b/src/engine/connection/exec/acp_helpers.rs new file mode 100644 index 00000000..bc404a36 --- /dev/null +++ b/src/engine/connection/exec/acp_helpers.rs @@ -0,0 +1,253 @@ +//! ACP initialization frame rewriting for protocol-mode stdin forwarding. + +use std::io; +use std::pin::Pin; + +use ortho_config::serde_json::{self, Value}; +use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncWrite, AsyncWriteExt}; + +use super::STDIN_BUFFER_CAPACITY; + +/// Upper bound on the number of bytes buffered while searching for the first +/// ACP frame. Frames that exceed this limit are forwarded as-is without +/// attempting JSON rewriting. +pub(super) const MAX_FIRST_FRAME_BYTES: usize = STDIN_BUFFER_CAPACITY; +/// The JSON-RPC `method` value that identifies an ACP initialisation request. +pub(super) const ACP_INITIALIZE_METHOD: &str = "initialize"; +/// The `params` field that carries the client's advertised ACP capabilities. +pub(super) const ACP_CLIENT_CAPABILITIES_FIELD: &str = "clientCapabilities"; +/// The `clientCapabilities` key for filesystem capability advertisements, +/// which are masked before the frame is forwarded to the container. +pub(super) const ACP_FILE_SYSTEM_CAPABILITY: &str = "fs"; +/// The `clientCapabilities` key for terminal capability advertisements, +/// which are masked before the frame is forwarded to the container. +pub(super) const ACP_TERMINAL_CAPABILITY: &str = "terminal"; + +enum InitialFrameAction { + Continue, + ForwardMasked, + ForwardUnchanged(ForwardUnchangedReason), +} + +#[derive(Clone, Copy)] +enum ForwardUnchangedReason { + ExceededMaximumSize, + EofBeforeNewline, +} + +/// Reads the first newline-delimited ACP frame from `buffered_stdin`, rewrites +/// it by removing masked capabilities, and writes the result to `input`. +/// +/// If the frame exceeds [`MAX_FIRST_FRAME_BYTES`] before a newline is found, +/// or if EOF is reached first, the buffered bytes are forwarded unchanged. +pub(super) async fn forward_initial_acp_frame_async( + buffered_stdin: &mut tokio::io::BufReader, + input: &mut Pin>, +) -> io::Result<()> +where + HostStdin: AsyncRead + Unpin, +{ + let mut first_frame = Vec::new(); + loop { + match next_initial_frame_action(buffered_stdin, &mut first_frame).await? { + InitialFrameAction::Continue => {} + InitialFrameAction::ForwardUnchanged(reason) => { + return forward_unmodified_initial_frame(input, &first_frame, reason).await; + } + InitialFrameAction::ForwardMasked => { + return forward_masked_initial_frame(input, &first_frame).await; + } + } + } +} + +async fn next_initial_frame_action( + buffered_stdin: &mut tokio::io::BufReader, + first_frame: &mut Vec, +) -> io::Result +where + HostStdin: AsyncRead + Unpin, +{ + if first_frame.len() == MAX_FIRST_FRAME_BYTES { + return Ok(InitialFrameAction::ForwardUnchanged( + ForwardUnchangedReason::ExceededMaximumSize, + )); + } + + let (bytes_to_consume, has_complete_frame) = + read_next_bounded_frame_chunk(buffered_stdin, first_frame).await?; + if bytes_to_consume == 0 { + return Ok(InitialFrameAction::ForwardUnchanged( + ForwardUnchangedReason::EofBeforeNewline, + )); + } + + buffered_stdin.consume(bytes_to_consume); + Ok(if has_complete_frame { + InitialFrameAction::ForwardMasked + } else { + InitialFrameAction::Continue + }) +} + +fn log_unmodified_forwarding(reason: ForwardUnchangedReason, bytes: usize) { + match reason { + ForwardUnchangedReason::ExceededMaximumSize => log_exceeded_maximum_size(bytes), + ForwardUnchangedReason::EofBeforeNewline => log_eof_before_newline(bytes), + } +} + +fn log_exceeded_maximum_size(bytes: usize) { + tracing::debug!( + bytes, + "ACP first frame exceeded maximum size; forwarding without rewrite" + ); +} + +fn log_eof_before_newline(bytes: usize) { + tracing::debug!( + bytes, + "ACP stdin reached EOF before newline; forwarding without rewrite" + ); +} + +async fn forward_unmodified_initial_frame( + input: &mut Pin>, + first_frame: &[u8], + reason: ForwardUnchangedReason, +) -> io::Result<()> { + log_unmodified_forwarding(reason, first_frame.len()); + input.write_all(first_frame).await +} + +async fn forward_masked_initial_frame( + input: &mut Pin>, + first_frame: &[u8], +) -> io::Result<()> { + let masked_frame = mask_acp_initialize_frame(first_frame); + input.write_all(&masked_frame).await?; + log_masked_frame_forwarded(&masked_frame, first_frame); + Ok(()) +} + +fn log_masked_frame_forwarded(masked_frame: &[u8], first_frame: &[u8]) { + if masked_frame != first_frame { + tracing::debug!("ACP initialize frame masked and forwarded"); + } +} + +async fn read_next_bounded_frame_chunk( + buffered_stdin: &mut tokio::io::BufReader, + first_frame: &mut Vec, +) -> io::Result<(usize, bool)> +where + HostStdin: AsyncRead + Unpin, +{ + let buffered = buffered_stdin.fill_buf().await?; + if buffered.is_empty() { + return Ok((0, false)); + } + + let remaining_capacity = MAX_FIRST_FRAME_BYTES.saturating_sub(first_frame.len()); + let bytes_to_scan = buffered.len().min(remaining_capacity); + let newline_offset = buffered + .get(..bytes_to_scan) + .and_then(|bytes| bytes.iter().position(|byte| *byte == b'\n')); + let bytes_to_consume = newline_offset.map_or(bytes_to_scan, |offset| offset + 1); + let Some(bytes) = buffered.get(..bytes_to_consume) else { + return Err(io::Error::other( + "buffered stdin slice exceeded available input", + )); + }; + first_frame.extend_from_slice(bytes); + Ok((bytes_to_consume, newline_offset.is_some())) +} + +/// Rewrites an ACP `initialize` frame by removing `terminal` and `fs` from +/// `params.clientCapabilities`, then re-serialises the JSON preserving the +/// original line ending. Returns the original frame unchanged on any parse or +/// serialisation failure, or if no capabilities were removed. +#[expect( + clippy::cognitive_complexity, + reason = "inline parse and serialise fallback tracing keeps this review-required flow local" +)] +pub(super) fn mask_acp_initialize_frame(frame: &[u8]) -> Vec { + let (payload, line_ending) = split_frame_line_ending(frame); + + let mut message: Value = match serde_json::from_slice(payload) { + Ok(v) => v, + Err(e) => { + tracing::warn!( + error = %e, + "ACP frame JSON deserialization failed; forwarding unchanged" + ); + return frame.to_vec(); + } + }; + + if !remove_masked_acp_capabilities(&mut message) { + tracing::debug!("ACP frame is not a maskable initialize request; forwarding unchanged"); + return frame.to_vec(); + } + + let mut serialized = match serde_json::to_vec(&message) { + Ok(v) => v, + Err(e) => { + tracing::warn!( + error = %e, + "ACP frame JSON serialization failed after masking; forwarding unchanged" + ); + return frame.to_vec(); + } + }; + serialized.extend_from_slice(line_ending); + serialized +} + +/// Splits a newline-delimited frame into its JSON payload and trailing line +/// ending bytes (`\n` or `\r\n`). Returns the original slice as the payload +/// with an empty line-ending slice when no recognised line ending is present. +pub(super) fn split_frame_line_ending(frame: &[u8]) -> (&[u8], &[u8]) { + if let Some(stripped) = frame.strip_suffix(b"\r\n") { + return (stripped, b"\r\n"); + } + + if let Some(stripped) = frame.strip_suffix(b"\n") { + return (stripped, b"\n"); + } + + (frame, b"") +} + +fn remove_masked_acp_capabilities(message: &mut Value) -> bool { + if message.get("method").and_then(Value::as_str) != Some(ACP_INITIALIZE_METHOD) { + return false; + } + + let Some(params) = message.get_mut("params").and_then(Value::as_object_mut) else { + return false; + }; + + let Some(client_capabilities) = params + .get_mut(ACP_CLIENT_CAPABILITIES_FIELD) + .and_then(Value::as_object_mut) + else { + return false; + }; + + let removed_terminal = client_capabilities + .remove(ACP_TERMINAL_CAPABILITY) + .is_some(); + let removed_fs = client_capabilities + .remove(ACP_FILE_SYSTEM_CAPABILITY) + .is_some(); + if !removed_terminal && !removed_fs { + return false; + } + + if client_capabilities.is_empty() { + params.remove(ACP_CLIENT_CAPABILITIES_FIELD); + } + + true +} diff --git a/src/engine/connection/exec/protocol.rs b/src/engine/connection/exec/protocol.rs index 6f70fb8c..bdf7e462 100644 --- a/src/engine/connection/exec/protocol.rs +++ b/src/engine/connection/exec/protocol.rs @@ -28,11 +28,13 @@ use std::time::Duration; use bollard::container::LogOutput; use bollard::errors::Error as BollardError; use futures_util::{Stream, StreamExt}; -use ortho_config::serde_json::{self, Value}; -use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncWrite, AsyncWriteExt}; +use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt}; use tokio::task::JoinHandle; use tokio::time::timeout; +#[path = "acp_helpers.rs"] +mod acp_helpers; + use super::helpers::spawn_stdin_forwarding_task; use super::host_io::stdin_forwarding_disabled_for_tests; use super::{ExecRequest, exec_failed}; @@ -60,20 +62,18 @@ const STDIN_SETTLE_TIMEOUT: Duration = Duration::from_millis(50); /// protocol message sizes and typical OS pipe buffer defaults, keeping latency /// low while preventing unbounded accumulation during high-throughput scenarios. const STDIN_BUFFER_CAPACITY: usize = 65_536; -const ACP_INITIALIZE_METHOD: &str = "initialize"; -const ACP_CLIENT_CAPABILITIES_FIELD: &str = "clientCapabilities"; -const ACP_FILE_SYSTEM_CAPABILITY: &str = "fs"; -const ACP_TERMINAL_CAPABILITY: &str = "terminal"; #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] pub(super) struct ProtocolSessionOptions { disable_stdin_forwarding: bool, + rewrite_acp_initialize: bool, } impl ProtocolSessionOptions { pub(super) const fn new() -> Self { Self { disable_stdin_forwarding: false, + rewrite_acp_initialize: false, } } @@ -81,6 +81,12 @@ impl ProtocolSessionOptions { self.disable_stdin_forwarding = disable; self } + + /// Enable or disable Agentic Control Protocol (ACP) initialisation rewriting for this protocol session. + pub(super) const fn with_acp_initialize_rewrite_enabled(mut self, enable: bool) -> Self { + self.rewrite_acp_initialize = enable; + self + } } struct HeldOpenStdin { @@ -168,8 +174,11 @@ where stderr: mut host_stderr, options, } = stdio; + let rewrite_acp_initialize = options.rewrite_acp_initialize; let stdin_task = - spawn_stdin_forwarding_task(host_stdin, input, forward_host_stdin_to_exec_async); + spawn_stdin_forwarding_task(host_stdin, input, move |stdin_reader, exec_input| { + forward_host_stdin_to_exec_async(stdin_reader, exec_input, rewrite_acp_initialize) + }); let output_result = run_output_loop_async( request.container_id(), &mut output, @@ -238,134 +247,20 @@ fn abort_stdin_forwarding_task(stdin_task: JoinHandle>) { async fn forward_host_stdin_to_exec_async( host_stdin: HostStdin, mut input: Pin>, + rewrite_acp_initialize: bool, ) -> io::Result<()> where HostStdin: AsyncRead + Unpin, { let mut buffered_stdin = tokio::io::BufReader::with_capacity(STDIN_BUFFER_CAPACITY, host_stdin); - forward_initial_protocol_frame_async(&mut buffered_stdin, &mut input).await?; - tokio::io::copy(&mut buffered_stdin, &mut input).await?; - input.flush().await?; - input.shutdown().await -} - -async fn forward_initial_protocol_frame_async( - buffered_stdin: &mut tokio::io::BufReader, - input: &mut Pin>, -) -> io::Result<()> -where - HostStdin: AsyncRead + Unpin, -{ - let mut first_frame = Vec::new(); - loop { - if first_frame.len() == STDIN_BUFFER_CAPACITY { - input.write_all(&first_frame).await?; - return Ok(()); - } - - let (bytes_to_consume, has_complete_frame) = { - let buffered = buffered_stdin.fill_buf().await?; - if buffered.is_empty() { - return forward_partial_initial_frame(input, &first_frame).await; - } - - let remaining_capacity = STDIN_BUFFER_CAPACITY.saturating_sub(first_frame.len()); - let bytes_to_scan = buffered.len().min(remaining_capacity); - let newline_offset = buffered - .get(..bytes_to_scan) - .and_then(|bytes| bytes.iter().position(|byte| *byte == b'\n')); - let bytes_to_consume = newline_offset.map_or(bytes_to_scan, |offset| offset + 1); - let Some(bytes) = buffered.get(..bytes_to_consume) else { - return Err(io::Error::other( - "buffered stdin slice exceeded available input", - )); - }; - first_frame.extend_from_slice(bytes); - (bytes_to_consume, newline_offset.is_some()) - }; - buffered_stdin.consume(bytes_to_consume); - - if has_complete_frame { - return input - .write_all(&mask_acp_initialize_frame(&first_frame)) - .await; - } - } -} - -async fn forward_partial_initial_frame( - input: &mut Pin>, - first_frame: &[u8], -) -> io::Result<()> { - if first_frame.is_empty() { - return Ok(()); - } - - input.write_all(first_frame).await -} - -fn mask_acp_initialize_frame(frame: &[u8]) -> Vec { - let (payload, line_ending) = split_frame_line_ending(frame); - - let Ok(mut message) = serde_json::from_slice(payload) else { - return frame.to_vec(); - }; - - if !remove_masked_acp_capabilities(&mut message) { - return frame.to_vec(); - } - - let Ok(mut serialized) = serde_json::to_vec(&message) else { - return frame.to_vec(); - }; - serialized.extend_from_slice(line_ending); - serialized -} - -fn split_frame_line_ending(frame: &[u8]) -> (&[u8], &[u8]) { - if let Some(stripped) = frame.strip_suffix(b"\r\n") { - return (stripped, b"\r\n"); - } - - if let Some(stripped) = frame.strip_suffix(b"\n") { - return (stripped, b"\n"); - } - - (frame, b"") -} - -fn remove_masked_acp_capabilities(message: &mut Value) -> bool { - if message.get("method").and_then(Value::as_str) != Some(ACP_INITIALIZE_METHOD) { - return false; - } - - let Some(params) = message.get_mut("params").and_then(Value::as_object_mut) else { - return false; - }; - - let Some(client_capabilities) = params - .get_mut(ACP_CLIENT_CAPABILITIES_FIELD) - .and_then(Value::as_object_mut) - else { - return false; - }; - - let removed_terminal = client_capabilities - .remove(ACP_TERMINAL_CAPABILITY) - .is_some(); - let removed_fs = client_capabilities - .remove(ACP_FILE_SYSTEM_CAPABILITY) - .is_some(); - if !removed_terminal && !removed_fs { - return false; - } - - if client_capabilities.is_empty() { - params.remove(ACP_CLIENT_CAPABILITIES_FIELD); + if rewrite_acp_initialize { + acp_helpers::forward_initial_acp_frame_async(&mut buffered_stdin, &mut input).await?; } + tokio::io::copy(&mut buffered_stdin, &mut input).await?; - true + input.flush().await?; + input.shutdown().await } #[cfg(test)] diff --git a/src/engine/connection/exec/protocol_acp_bdd_tests.rs b/src/engine/connection/exec/protocol_acp_bdd_tests.rs new file mode 100644 index 00000000..deafefbf --- /dev/null +++ b/src/engine/connection/exec/protocol_acp_bdd_tests.rs @@ -0,0 +1,110 @@ +//! BDD scenario wiring for ACP capability masking tests. + +use rstest_bdd::Slot; +use rstest_bdd_macros::{ScenarioState, given, scenario, then, when}; + +use super::{ + initialize_without_blocked_capabilities, malformed_initialize_bytes, + masked_initialize_with_follow_up, run_forwarding, +}; + +type StepResult = Result; + +#[derive(Default, ScenarioState)] +struct AcpMaskingState { + host_stdin: Slot>, + expected_forwarded: Slot>, + actual_forwarded: Slot>, + succeeded: Slot, +} + +#[rstest::fixture] +fn acp_masking_state() -> AcpMaskingState { + AcpMaskingState::default() +} + +#[given( + "ACP stdin contains an initialize request with blocked capabilities and a follow-up request" +)] +fn acp_stdin_contains_blocked_initialize_and_follow_up(acp_masking_state: &AcpMaskingState) { + let (host_stdin_bytes, expected) = masked_initialize_with_follow_up(); + acp_masking_state.host_stdin.set(host_stdin_bytes); + acp_masking_state.expected_forwarded.set(expected); +} + +#[given("ACP stdin contains malformed initialize bytes")] +fn acp_stdin_contains_malformed_initialize(acp_masking_state: &AcpMaskingState) { + let malformed = malformed_initialize_bytes(); + acp_masking_state.host_stdin.set(malformed.clone()); + acp_masking_state.expected_forwarded.set(malformed); +} + +#[given("ACP stdin contains initialize without blocked capabilities")] +fn acp_stdin_contains_safe_initialize(acp_masking_state: &AcpMaskingState) { + let initialize = initialize_without_blocked_capabilities(); + acp_masking_state.host_stdin.set(initialize.clone()); + acp_masking_state.expected_forwarded.set(initialize); +} + +#[when("ACP stdin forwarding runs")] +fn acp_stdin_forwarding_runs(acp_masking_state: &AcpMaskingState) -> StepResult<()> { + let host_stdin_bytes = acp_masking_state + .host_stdin + .get() + .ok_or_else(|| String::from("host stdin should be configured"))?; + let (forwarded, _) = run_forwarding(&host_stdin_bytes); + acp_masking_state.actual_forwarded.set(forwarded); + acp_masking_state.succeeded.set(true); + Ok(()) +} + +#[then("ACP stdin forwarding succeeds")] +fn acp_stdin_forwarding_succeeds(acp_masking_state: &AcpMaskingState) -> StepResult<()> { + if acp_masking_state.succeeded.get() == Some(true) { + Ok(()) + } else { + Err(String::from("expected ACP stdin forwarding to succeed")) + } +} + +#[then("the forwarded ACP stdin matches the expected bytes")] +fn forwarded_acp_stdin_matches_expected(acp_masking_state: &AcpMaskingState) -> StepResult<()> { + let actual = acp_masking_state + .actual_forwarded + .get() + .ok_or_else(|| String::from("forwarded bytes should be recorded"))?; + let expected = acp_masking_state + .expected_forwarded + .get() + .ok_or_else(|| String::from("expected bytes should be recorded"))?; + + if actual == expected { + Ok(()) + } else { + Err(format!("expected {expected:?}, got {actual:?}")) + } +} + +#[scenario( + path = "tests/features/acp_capability_masking.feature", + name = "ACP initialize masks blocked capabilities before forwarding" +)] +fn acp_initialize_masks_blocked_capabilities(acp_masking_state: AcpMaskingState) { + let _ = acp_masking_state; +} + +#[scenario( + path = "tests/features/acp_capability_masking.feature", + name = "Malformed ACP initialize is forwarded unchanged" +)] +fn malformed_acp_initialize_is_forwarded_unchanged(acp_masking_state: AcpMaskingState) { + let _ = acp_masking_state; +} + +#[scenario( + path = "tests/features/acp_capability_masking.feature", + name = "ACP initialize without blocked capabilities stays unchanged" +)] +fn acp_initialize_without_blocked_capabilities_stays_unchanged(acp_masking_state: AcpMaskingState) { + let _ = acp_masking_state; +} diff --git a/src/engine/connection/exec/protocol_acp_tests.rs b/src/engine/connection/exec/protocol_acp_tests.rs index 1ba5b1a7..b06696e2 100644 --- a/src/engine/connection/exec/protocol_acp_tests.rs +++ b/src/engine/connection/exec/protocol_acp_tests.rs @@ -6,22 +6,14 @@ use std::sync::{Arc, Mutex}; use std::task::{Context, Poll}; use rstest::rstest; -use rstest_bdd::Slot; -use rstest_bdd_macros::{ScenarioState, given, scenario, then, when}; use tokio::io::{AsyncWrite, AsyncWriteExt, DuplexStream}; +use super::acp_helpers::{ + ACP_FILE_SYSTEM_CAPABILITY, ACP_TERMINAL_CAPABILITY, MAX_FIRST_FRAME_BYTES, + forward_initial_acp_frame_async, mask_acp_initialize_frame, split_frame_line_ending, +}; use super::*; -type StepResult = Result; - -#[derive(Default, ScenarioState)] -struct AcpMaskingState { - host_stdin: Slot>, - expected_forwarded: Slot>, - actual_forwarded: Slot>, - succeeded: Slot, -} - struct RecordingInputWriter { bytes: Arc>>, shutdown_called: Arc>, @@ -91,7 +83,9 @@ fn initialize_frame(line_ending: &str) -> Vec { frame } -fn initialize_without_blocked_capabilities() -> Vec { +/// Builds a serialised ACP `initialize` frame whose `clientCapabilities` +/// contains only `_meta` (no blocked entries), terminated with `\n`. +pub(super) fn initialize_without_blocked_capabilities() -> Vec { let payload = serde_json::json!({ "jsonrpc": "2.0", "id": 0, @@ -141,7 +135,9 @@ fn session_new_bytes() -> Vec { br#"{"jsonrpc":"2.0","id":1,"method":"session/new","params":{"cwd":"/workspace"}}"#.to_vec() } -fn malformed_initialize_bytes() -> Vec { +/// Returns a byte sequence that is syntactically invalid JSON (missing the +/// closing `}`), used to verify that malformed frames are forwarded unchanged. +pub(super) fn malformed_initialize_bytes() -> Vec { br#"{"jsonrpc":"2.0","method":"initialize","params":{"clientCapabilities":{"terminal":true}}"# .to_vec() } @@ -172,6 +168,22 @@ fn assert_masked_client_capabilities(message: &serde_json::Value) { ); } +fn client_capabilities(message: &serde_json::Value) -> &serde_json::Map { + message + .get("params") + .and_then(serde_json::Value::as_object) + .and_then(|params| params.get("clientCapabilities")) + .and_then(serde_json::Value::as_object) + .expect("clientCapabilities should remain") +} + +fn params(message: &serde_json::Value) -> &serde_json::Map { + message + .get("params") + .and_then(serde_json::Value::as_object) + .expect("params should remain") +} + async fn build_host_stdin(bytes: &[u8]) -> io::Result { let capacity = bytes.len().max(1); let (mut writer, reader) = tokio::io::duplex(capacity); @@ -180,7 +192,17 @@ async fn build_host_stdin(bytes: &[u8]) -> io::Result { Ok(reader) } -fn run_forwarding(host_stdin_bytes: &[u8]) -> (Vec, bool) { +/// Runs ACP stdin forwarding synchronously with `rewrite_acp_initialize = +/// true`, returning the bytes written to the container input and whether +/// `poll_shutdown` was called. +pub(super) fn run_forwarding(host_stdin_bytes: &[u8]) -> (Vec, bool) { + run_forwarding_with_rewrite(host_stdin_bytes, true) +} + +fn run_forwarding_with_rewrite( + host_stdin_bytes: &[u8], + rewrite_acp_initialize: bool, +) -> (Vec, bool) { let runtime = tokio::runtime::Runtime::new().expect("runtime should build"); let host_stdin = runtime .block_on(build_host_stdin(host_stdin_bytes)) @@ -193,6 +215,7 @@ fn run_forwarding(host_stdin_bytes: &[u8]) -> (Vec, bool) { .block_on(forward_host_stdin_to_exec_async( host_stdin, Box::pin(container_input), + rewrite_acp_initialize, )) .expect("stdin forwarding should succeed"); @@ -207,6 +230,19 @@ fn run_forwarding(host_stdin_bytes: &[u8]) -> (Vec, bool) { ) } +#[test] +fn forwarding_leaves_initialize_unchanged_when_acp_rewrite_is_disabled() { + let host_stdin_bytes = initialize_frame("\n"); + + let (forwarded, shutdown_called) = run_forwarding_with_rewrite(&host_stdin_bytes, false); + + assert_eq!( + forwarded, host_stdin_bytes, + "generic protocol sessions should retain raw byte-stream semantics" + ); + assert!(shutdown_called, "stdin forwarding should shut down input"); +} + #[rstest] #[case("\n")] #[case("\r\n")] @@ -252,6 +288,118 @@ fn mask_acp_initialize_frame_removes_empty_client_capabilities() { ); } +#[test] +fn mask_acp_initialize_frame_removes_only_fs_when_terminal_absent() { + let payload = serde_json::json!({ + "jsonrpc": "2.0", + "id": 0, + "method": "initialize", + "params": { + "protocolVersion": 1, + "clientCapabilities": { + "fs": { "readTextFile": true }, + "auth": { "token": true } + }, + "clientInfo": { "name": "podbot-tests", "version": "1.0.0" } + } + }); + let mut frame = serde_json::to_vec(&payload).expect("payload should serialise"); + frame.push(b'\n'); + + let masked = mask_acp_initialize_frame(&frame); + let result = parse_frame_payload(&masked); + let caps = client_capabilities(&result); + + assert!(!caps.contains_key("fs"), "fs should be removed"); + assert!(caps.contains_key("auth"), "auth should be preserved"); +} + +#[test] +fn mask_acp_initialize_frame_removes_only_terminal_when_fs_absent() { + let payload = serde_json::json!({ + "jsonrpc": "2.0", + "id": 0, + "method": "initialize", + "params": { + "protocolVersion": 1, + "clientCapabilities": { + "terminal": true, + "logging": { "level": "info" } + }, + "clientInfo": { "name": "podbot-tests", "version": "1.0.0" } + } + }); + let mut frame = serde_json::to_vec(&payload).expect("payload should serialise"); + frame.push(b'\n'); + + let masked = mask_acp_initialize_frame(&frame); + let result = parse_frame_payload(&masked); + let caps = client_capabilities(&result); + + assert!(!caps.contains_key("terminal"), "terminal should be removed"); + assert!(caps.contains_key("logging"), "logging should be preserved"); +} + +#[test] +fn mask_acp_initialize_frame_preserves_all_unrelated_capabilities() { + let payload = serde_json::json!({ + "jsonrpc": "2.0", + "id": 0, + "method": "initialize", + "params": { + "protocolVersion": 1, + "clientCapabilities": { + "fs": { "readTextFile": true }, + "terminal": true, + "auth": { "token": true }, + "logging": { "level": "debug" } + }, + "clientInfo": { "name": "podbot-tests", "version": "1.0.0" } + } + }); + let mut frame = serde_json::to_vec(&payload).expect("payload should serialise"); + frame.push(b'\n'); + + let masked = mask_acp_initialize_frame(&frame); + let result = parse_frame_payload(&masked); + let caps = client_capabilities(&result); + + assert!(!caps.contains_key("fs"), "fs should be removed"); + assert!(!caps.contains_key("terminal"), "terminal should be removed"); + assert!(caps.contains_key("auth"), "auth should be preserved"); + assert!(caps.contains_key("logging"), "logging should be preserved"); +} + +#[test] +fn mask_acp_initialize_frame_passes_through_frame_without_line_ending() { + let payload = serde_json::json!({ + "jsonrpc": "2.0", + "id": 0, + "method": "initialize", + "params": { + "clientCapabilities": { "fs": { "readTextFile": true }, "terminal": true } + } + }); + // Deliberately omit the trailing newline. + let frame = serde_json::to_vec(&payload).expect("payload should serialise"); + + let masked = mask_acp_initialize_frame(&frame); + + // Without a line ending the splitter returns the whole slice as payload + // with an empty line-ending suffix. The JSON itself is still rewritten, + // but the result must not gain a spurious newline byte. + assert!( + !masked.ends_with(b"\n"), + "masked frame should not gain a trailing newline" + ); + let result: serde_json::Value = + serde_json::from_slice(&masked).expect("result should be valid JSON"); + assert!( + params(&result).get("clientCapabilities").is_none(), + "capabilities should still be masked even without a line ending" + ); +} + #[test] fn mask_acp_initialize_frame_leaves_non_initialize_messages_unchanged() { let mut frame = session_new_bytes(); @@ -270,7 +418,7 @@ fn mask_acp_initialize_frame_leaves_malformed_input_unchanged() { #[test] fn forwarding_masks_initialize_and_preserves_trailing_bytes() { let mut host_stdin_bytes = initialize_frame("\n"); - let trailing = session_new_bytes(); + let trailing = initialize_frame("\n"); host_stdin_bytes.extend_from_slice(&trailing); let (forwarded, shutdown_called) = run_forwarding(&host_stdin_bytes); @@ -298,7 +446,7 @@ fn forwarding_masks_initialize_and_preserves_trailing_bytes() { #[test] fn forwarding_does_not_wait_indefinitely_for_oversized_initial_frame() { let runtime = tokio::runtime::Runtime::new().expect("runtime should build"); - let host_stdin_bytes = vec![b'x'; STDIN_BUFFER_CAPACITY + 1]; + let host_stdin_bytes = vec![b'x'; MAX_FIRST_FRAME_BYTES + 1]; let (host_writer, host_reader) = runtime .block_on(async { let (mut host_writer, host_reader) = tokio::io::duplex(host_stdin_bytes.len()); @@ -317,7 +465,7 @@ fn forwarding_does_not_wait_indefinitely_for_oversized_initial_frame() { .block_on(async { tokio::time::timeout( STDIN_SETTLE_TIMEOUT, - forward_initial_protocol_frame_async(&mut buffered_stdin, &mut container_input), + forward_initial_acp_frame_async(&mut buffered_stdin, &mut container_input), ) .await }) @@ -329,21 +477,20 @@ fn forwarding_does_not_wait_indefinitely_for_oversized_initial_frame() { .lock() .expect("writer mutex should not poison") .len(), - STDIN_BUFFER_CAPACITY, + MAX_FIRST_FRAME_BYTES, "only the bounded first-frame buffer should be held before streaming resumes" ); drop(host_writer); } -#[rstest::fixture] -fn acp_masking_state() -> AcpMaskingState { - AcpMaskingState::default() -} - -fn masked_initialize_with_follow_up() -> (Vec, Vec) { +/// Constructs a host-stdin byte sequence containing a masked `initialize` +/// frame followed by a follow-up frame, and returns both the raw input bytes +/// and the expected post-masking output bytes for BDD assertion. +pub(super) fn masked_initialize_with_follow_up() -> (Vec, Vec) { let mut host_stdin_bytes = initialize_frame("\n"); - host_stdin_bytes.extend_from_slice(&session_new_bytes()); + let follow_up = initialize_frame("\n"); + host_stdin_bytes.extend_from_slice(&follow_up); let expected_initialize = serde_json::json!({ "jsonrpc": "2.0", @@ -366,93 +513,10 @@ fn masked_initialize_with_follow_up() -> (Vec, Vec) { let mut expected = serde_json::to_vec(&expected_initialize) .expect("expected initialize payload should serialize"); expected.push(b'\n'); - expected.extend_from_slice(&session_new_bytes()); + expected.extend_from_slice(&follow_up); (host_stdin_bytes, expected) } -#[given( - "ACP stdin contains an initialize request with blocked capabilities and a follow-up request" -)] -fn acp_stdin_contains_blocked_initialize_and_follow_up(acp_masking_state: &AcpMaskingState) { - let (host_stdin_bytes, expected) = masked_initialize_with_follow_up(); - acp_masking_state.host_stdin.set(host_stdin_bytes); - acp_masking_state.expected_forwarded.set(expected); -} - -#[given("ACP stdin contains malformed initialize bytes")] -fn acp_stdin_contains_malformed_initialize(acp_masking_state: &AcpMaskingState) { - let malformed = malformed_initialize_bytes(); - acp_masking_state.host_stdin.set(malformed.clone()); - acp_masking_state.expected_forwarded.set(malformed); -} - -#[given("ACP stdin contains initialize without blocked capabilities")] -fn acp_stdin_contains_safe_initialize(acp_masking_state: &AcpMaskingState) { - let initialize = initialize_without_blocked_capabilities(); - acp_masking_state.host_stdin.set(initialize.clone()); - acp_masking_state.expected_forwarded.set(initialize); -} - -#[when("ACP stdin forwarding runs")] -fn acp_stdin_forwarding_runs(acp_masking_state: &AcpMaskingState) -> StepResult<()> { - let host_stdin_bytes = acp_masking_state - .host_stdin - .get() - .ok_or_else(|| String::from("host stdin should be configured"))?; - let (forwarded, _) = run_forwarding(&host_stdin_bytes); - acp_masking_state.actual_forwarded.set(forwarded); - acp_masking_state.succeeded.set(true); - Ok(()) -} - -#[then("ACP stdin forwarding succeeds")] -fn acp_stdin_forwarding_succeeds(acp_masking_state: &AcpMaskingState) -> StepResult<()> { - if acp_masking_state.succeeded.get() == Some(true) { - Ok(()) - } else { - Err(String::from("expected ACP stdin forwarding to succeed")) - } -} - -#[then("the forwarded ACP stdin matches the expected bytes")] -fn forwarded_acp_stdin_matches_expected(acp_masking_state: &AcpMaskingState) -> StepResult<()> { - let actual = acp_masking_state - .actual_forwarded - .get() - .ok_or_else(|| String::from("forwarded bytes should be recorded"))?; - let expected = acp_masking_state - .expected_forwarded - .get() - .ok_or_else(|| String::from("expected bytes should be recorded"))?; - - if actual == expected { - Ok(()) - } else { - Err(format!("expected {expected:?}, got {actual:?}")) - } -} - -#[scenario( - path = "tests/features/acp_capability_masking.feature", - name = "ACP initialize masks blocked capabilities before forwarding" -)] -fn acp_initialize_masks_blocked_capabilities(acp_masking_state: AcpMaskingState) { - let _ = acp_masking_state; -} - -#[scenario( - path = "tests/features/acp_capability_masking.feature", - name = "Malformed ACP initialize is forwarded unchanged" -)] -fn malformed_acp_initialize_is_forwarded_unchanged(acp_masking_state: AcpMaskingState) { - let _ = acp_masking_state; -} - -#[scenario( - path = "tests/features/acp_capability_masking.feature", - name = "ACP initialize without blocked capabilities stays unchanged" -)] -fn acp_initialize_without_blocked_capabilities_stays_unchanged(acp_masking_state: AcpMaskingState) { - let _ = acp_masking_state; -} +#[path = "protocol_acp_bdd_tests.rs"] +mod bdd_tests; diff --git a/src/engine/connection/exec/session.rs b/src/engine/connection/exec/session.rs index aaf8ac52..71e5fa01 100644 --- a/src/engine/connection/exec/session.rs +++ b/src/engine/connection/exec/session.rs @@ -7,6 +7,7 @@ use super::protocol::ProtocolSessionOptions; #[derive(Debug, Clone, Copy, Default)] pub(crate) struct ExecSessionOptions { disable_protocol_stdin_forwarding: bool, + rewrite_acp_initialize: bool, } impl ExecSessionOptions { @@ -15,6 +16,7 @@ impl ExecSessionOptions { pub const fn new() -> Self { Self { disable_protocol_stdin_forwarding: false, + rewrite_acp_initialize: false, } } @@ -26,6 +28,20 @@ impl ExecSessionOptions { self.disable_protocol_stdin_forwarding = disable; self } + + /// Enable ACP initialization rewriting for protocol-mode sessions. + #[cfg_attr( + not(test), + expect( + dead_code, + reason = "reserved for production ACP session selection when podbot host is enabled" + ) + )] + #[must_use] + pub const fn with_acp_initialize_rewrite_enabled(mut self, enable: bool) -> Self { + self.rewrite_acp_initialize = enable; + self + } } pub(super) const fn protocol_session_options( @@ -33,6 +49,7 @@ pub(super) const fn protocol_session_options( ) -> ProtocolSessionOptions { ProtocolSessionOptions::new() .with_stdin_forwarding_disabled(options.disable_protocol_stdin_forwarding) + .with_acp_initialize_rewrite_enabled(options.rewrite_acp_initialize) } #[cfg(test)] @@ -46,6 +63,10 @@ mod tests { !opts.disable_protocol_stdin_forwarding, "stdin forwarding should be enabled by default", ); + assert!( + !opts.rewrite_acp_initialize, + "ACP initialize rewriting should be opt-in", + ); } #[test] @@ -88,4 +109,14 @@ mod tests { ProtocolSessionOptions::new().with_stdin_forwarding_disabled(true), ); } + + #[test] + fn protocol_session_options_reflects_acp_rewrite_flag() { + let opts = ExecSessionOptions::new().with_acp_initialize_rewrite_enabled(true); + + assert_eq!( + protocol_session_options(opts), + ProtocolSessionOptions::new().with_acp_initialize_rewrite_enabled(true), + ); + } } From d039923c331d6998d45a84dcdb806b6eb30b700e Mon Sep 17 00:00:00 2001 From: leynos Date: Fri, 1 May 2026 23:54:48 +0200 Subject: [PATCH 3/6] Extract ACP initialize frame helper Define the initialize-frame skeleton once and pass only the capability payload into tests that vary ACP client capabilities. Move forwarding-specific ACP tests into a child module so the main ACP test file stays below the repository line budget while preserving the same private test helpers. --- .../exec/protocol_acp_forwarding_tests.rs | 85 ++++++ .../connection/exec/protocol_acp_tests.rs | 288 +++++------------- 2 files changed, 164 insertions(+), 209 deletions(-) create mode 100644 src/engine/connection/exec/protocol_acp_forwarding_tests.rs diff --git a/src/engine/connection/exec/protocol_acp_forwarding_tests.rs b/src/engine/connection/exec/protocol_acp_forwarding_tests.rs new file mode 100644 index 00000000..14be5fa5 --- /dev/null +++ b/src/engine/connection/exec/protocol_acp_forwarding_tests.rs @@ -0,0 +1,85 @@ +//! ACP stdin forwarding tests for the protocol proxy. + +use super::*; + +#[test] +fn forwarding_leaves_initialize_unchanged_when_acp_rewrite_is_disabled() { + let host_stdin_bytes = initialize_frame("\n"); + + let (forwarded, shutdown_called) = run_forwarding_with_rewrite(&host_stdin_bytes, false); + + assert_eq!( + forwarded, host_stdin_bytes, + "generic protocol sessions should retain raw byte-stream semantics" + ); + assert!(shutdown_called, "stdin forwarding should shut down input"); +} + +#[test] +fn forwarding_masks_initialize_and_preserves_trailing_bytes() { + let mut host_stdin_bytes = initialize_frame("\n"); + let trailing = initialize_frame("\n"); + host_stdin_bytes.extend_from_slice(&trailing); + + let (forwarded, shutdown_called) = run_forwarding(&host_stdin_bytes); + let newline_index = forwarded + .iter() + .position(|byte| *byte == b'\n') + .expect("masked initialize should remain line terminated"); + let initialize_frame = forwarded + .get(..=newline_index) + .expect("masked initialize frame should remain addressable"); + let trailing_forwarded = forwarded + .get(newline_index + 1..) + .expect("trailing bytes should remain addressable"); + let payload = parse_frame_payload(initialize_frame); + + assert_masked_client_capabilities(&payload); + assert_eq!( + trailing_forwarded, + trailing.as_slice(), + "trailing bytes should pass through unchanged" + ); + assert!(shutdown_called, "stdin forwarding should shut down input"); +} + +#[test] +fn forwarding_does_not_wait_indefinitely_for_oversized_initial_frame() { + let runtime = tokio::runtime::Runtime::new().expect("runtime should build"); + let host_stdin_bytes = vec![b'x'; MAX_FIRST_FRAME_BYTES + 1]; + let (host_writer, host_reader) = runtime + .block_on(async { + let (mut host_writer, host_reader) = tokio::io::duplex(host_stdin_bytes.len()); + host_writer.write_all(&host_stdin_bytes).await?; + io::Result::Ok((host_writer, host_reader)) + }) + .expect("host stdin should accept oversized initial bytes"); + + let mut buffered_stdin = + tokio::io::BufReader::with_capacity(STDIN_BUFFER_CAPACITY, host_reader); + let recording_input = RecordingInputWriter::new(); + let forwarded_bytes = recording_input.bytes.clone(); + let mut container_input: Pin> = Box::pin(recording_input); + + runtime + .block_on(async { + tokio::time::timeout( + STDIN_SETTLE_TIMEOUT, + forward_initial_acp_frame_async(&mut buffered_stdin, &mut container_input), + ) + .await + }) + .expect("initial forwarding should not wait for newline or EOF") + .expect("initial forwarding should succeed"); + + assert_eq!( + forwarded_bytes + .lock() + .expect("writer mutex should not poison") + .len(), + MAX_FIRST_FRAME_BYTES, + "only the bounded first-frame buffer should be held before streaming resumes" + ); + + drop(host_writer); +} diff --git a/src/engine/connection/exec/protocol_acp_tests.rs b/src/engine/connection/exec/protocol_acp_tests.rs index b06696e2..3541e1d7 100644 --- a/src/engine/connection/exec/protocol_acp_tests.rs +++ b/src/engine/connection/exec/protocol_acp_tests.rs @@ -54,35 +54,44 @@ impl AsyncWrite for RecordingInputWriter { } } -fn initialize_frame(line_ending: &str) -> Vec { - let payload = serde_json::json!({ +fn initialize_frame_with_capabilities( + capabilities: serde_json::Value, + line_ending: &str, +) -> Vec { + let mut payload = serde_json::json!({ "jsonrpc": "2.0", "id": 0, "method": "initialize", "params": { "protocolVersion": 1, - "clientCapabilities": { - "fs": { - "readTextFile": true, - "writeTextFile": true - }, - "terminal": true, - "_meta": { - "custom": true - } - }, + "clientCapabilities": null, "clientInfo": { "name": "podbot-tests", "version": "1.0.0" } } }); - - let mut frame = serde_json::to_vec(&payload).expect("initialize payload should serialize"); + payload + .get_mut("params") + .and_then(serde_json::Value::as_object_mut) + .expect("initialize params should be present") + .insert("clientCapabilities".to_owned(), capabilities); + let mut frame = serde_json::to_vec(&payload).expect("initialize payload should serialise"); frame.extend_from_slice(line_ending.as_bytes()); frame } +fn initialize_frame(line_ending: &str) -> Vec { + initialize_frame_with_capabilities( + serde_json::json!({ + "fs": { "readTextFile": true, "writeTextFile": true }, + "terminal": true, + "_meta": { "custom": true } + }), + line_ending, + ) +} + /// Builds a serialised ACP `initialize` frame whose `clientCapabilities` /// contains only `_meta` (no blocked entries), terminated with `\n`. pub(super) fn initialize_without_blocked_capabilities() -> Vec { @@ -106,29 +115,13 @@ pub(super) fn initialize_without_blocked_capabilities() -> Vec { } fn initialize_with_only_blocked_capabilities(line_ending: &str) -> Vec { - let payload = serde_json::json!({ - "jsonrpc": "2.0", - "id": 0, - "method": "initialize", - "params": { - "protocolVersion": 1, - "clientCapabilities": { - "fs": { - "readTextFile": true, - "writeTextFile": true - }, - "terminal": true - }, - "clientInfo": { - "name": "podbot-tests", - "version": "1.0.0" - } - } - }); - - let mut frame = serde_json::to_vec(&payload).expect("initialize payload should serialize"); - frame.extend_from_slice(line_ending.as_bytes()); - frame + initialize_frame_with_capabilities( + serde_json::json!({ + "fs": { "readTextFile": true, "writeTextFile": true }, + "terminal": true + }), + line_ending, + ) } fn session_new_bytes() -> Vec { @@ -230,19 +223,6 @@ fn run_forwarding_with_rewrite( ) } -#[test] -fn forwarding_leaves_initialize_unchanged_when_acp_rewrite_is_disabled() { - let host_stdin_bytes = initialize_frame("\n"); - - let (forwarded, shutdown_called) = run_forwarding_with_rewrite(&host_stdin_bytes, false); - - assert_eq!( - forwarded, host_stdin_bytes, - "generic protocol sessions should retain raw byte-stream semantics" - ); - assert!(shutdown_called, "stdin forwarding should shut down input"); -} - #[rstest] #[case("\n")] #[case("\r\n")] @@ -288,106 +268,62 @@ fn mask_acp_initialize_frame_removes_empty_client_capabilities() { ); } -#[test] -fn mask_acp_initialize_frame_removes_only_fs_when_terminal_absent() { - let payload = serde_json::json!({ - "jsonrpc": "2.0", - "id": 0, - "method": "initialize", - "params": { - "protocolVersion": 1, - "clientCapabilities": { - "fs": { "readTextFile": true }, - "auth": { "token": true } - }, - "clientInfo": { "name": "podbot-tests", "version": "1.0.0" } - } - }); - let mut frame = serde_json::to_vec(&payload).expect("payload should serialise"); - frame.push(b'\n'); - - let masked = mask_acp_initialize_frame(&frame); - let result = parse_frame_payload(&masked); - let caps = client_capabilities(&result); - - assert!(!caps.contains_key("fs"), "fs should be removed"); - assert!(caps.contains_key("auth"), "auth should be preserved"); -} - -#[test] -fn mask_acp_initialize_frame_removes_only_terminal_when_fs_absent() { - let payload = serde_json::json!({ - "jsonrpc": "2.0", - "id": 0, - "method": "initialize", - "params": { - "protocolVersion": 1, - "clientCapabilities": { - "terminal": true, - "logging": { "level": "info" } - }, - "clientInfo": { "name": "podbot-tests", "version": "1.0.0" } - } - }); - let mut frame = serde_json::to_vec(&payload).expect("payload should serialise"); - frame.push(b'\n'); - - let masked = mask_acp_initialize_frame(&frame); - let result = parse_frame_payload(&masked); - let caps = client_capabilities(&result); - - assert!(!caps.contains_key("terminal"), "terminal should be removed"); - assert!(caps.contains_key("logging"), "logging should be preserved"); -} - -#[test] -fn mask_acp_initialize_frame_preserves_all_unrelated_capabilities() { - let payload = serde_json::json!({ - "jsonrpc": "2.0", - "id": 0, - "method": "initialize", - "params": { - "protocolVersion": 1, - "clientCapabilities": { - "fs": { "readTextFile": true }, - "terminal": true, - "auth": { "token": true }, - "logging": { "level": "debug" } - }, - "clientInfo": { "name": "podbot-tests", "version": "1.0.0" } - } - }); - let mut frame = serde_json::to_vec(&payload).expect("payload should serialise"); - frame.push(b'\n'); - +#[rstest] +#[case( + serde_json::json!({ + "fs": { "readTextFile": true }, + "auth": { "token": true } + }), + &["fs"], + &["auth"] +)] +#[case( + serde_json::json!({ + "terminal": true, + "logging": { "level": "info" } + }), + &["terminal"], + &["logging"] +)] +#[case( + serde_json::json!({ + "fs": { "readTextFile": true }, + "terminal": true, + "auth": { "token": true }, + "logging": { "level": "debug" } + }), + &["fs", "terminal"], + &["auth", "logging"] +)] +fn mask_acp_initialize_frame_preserves_unrelated_capabilities( + #[case] capabilities: serde_json::Value, + #[case] removed_capabilities: &[&str], + #[case] preserved_capabilities: &[&str], +) { + let frame = initialize_frame_with_capabilities(capabilities, "\n"); let masked = mask_acp_initialize_frame(&frame); let result = parse_frame_payload(&masked); let caps = client_capabilities(&result); - assert!(!caps.contains_key("fs"), "fs should be removed"); - assert!(!caps.contains_key("terminal"), "terminal should be removed"); - assert!(caps.contains_key("auth"), "auth should be preserved"); - assert!(caps.contains_key("logging"), "logging should be preserved"); + for capability in removed_capabilities { + assert!( + !caps.contains_key(*capability), + "{capability} should be removed" + ); + } + for capability in preserved_capabilities { + assert!( + caps.contains_key(*capability), + "{capability} should be preserved" + ); + } } #[test] fn mask_acp_initialize_frame_passes_through_frame_without_line_ending() { - let payload = serde_json::json!({ - "jsonrpc": "2.0", - "id": 0, - "method": "initialize", - "params": { - "clientCapabilities": { "fs": { "readTextFile": true }, "terminal": true } - } - }); - // Deliberately omit the trailing newline. - let frame = serde_json::to_vec(&payload).expect("payload should serialise"); - + let frame = initialize_with_only_blocked_capabilities(""); let masked = mask_acp_initialize_frame(&frame); - // Without a line ending the splitter returns the whole slice as payload - // with an empty line-ending suffix. The JSON itself is still rewritten, - // but the result must not gain a spurious newline byte. assert!( !masked.ends_with(b"\n"), "masked frame should not gain a trailing newline" @@ -415,75 +351,6 @@ fn mask_acp_initialize_frame_leaves_malformed_input_unchanged() { assert_eq!(mask_acp_initialize_frame(&frame), frame); } -#[test] -fn forwarding_masks_initialize_and_preserves_trailing_bytes() { - let mut host_stdin_bytes = initialize_frame("\n"); - let trailing = initialize_frame("\n"); - host_stdin_bytes.extend_from_slice(&trailing); - - let (forwarded, shutdown_called) = run_forwarding(&host_stdin_bytes); - let newline_index = forwarded - .iter() - .position(|byte| *byte == b'\n') - .expect("masked initialize should remain line terminated"); - let initialize_frame = forwarded - .get(..=newline_index) - .expect("masked initialize frame should remain addressable"); - let trailing_forwarded = forwarded - .get(newline_index + 1..) - .expect("trailing bytes should remain addressable"); - let payload = parse_frame_payload(initialize_frame); - - assert_masked_client_capabilities(&payload); - assert_eq!( - trailing_forwarded, - trailing.as_slice(), - "trailing bytes should pass through unchanged" - ); - assert!(shutdown_called, "stdin forwarding should shut down input"); -} - -#[test] -fn forwarding_does_not_wait_indefinitely_for_oversized_initial_frame() { - let runtime = tokio::runtime::Runtime::new().expect("runtime should build"); - let host_stdin_bytes = vec![b'x'; MAX_FIRST_FRAME_BYTES + 1]; - let (host_writer, host_reader) = runtime - .block_on(async { - let (mut host_writer, host_reader) = tokio::io::duplex(host_stdin_bytes.len()); - host_writer.write_all(&host_stdin_bytes).await?; - io::Result::Ok((host_writer, host_reader)) - }) - .expect("host stdin should accept oversized initial bytes"); - - let mut buffered_stdin = - tokio::io::BufReader::with_capacity(STDIN_BUFFER_CAPACITY, host_reader); - let recording_input = RecordingInputWriter::new(); - let forwarded_bytes = recording_input.bytes.clone(); - let mut container_input: Pin> = Box::pin(recording_input); - - runtime - .block_on(async { - tokio::time::timeout( - STDIN_SETTLE_TIMEOUT, - forward_initial_acp_frame_async(&mut buffered_stdin, &mut container_input), - ) - .await - }) - .expect("initial forwarding should not wait for newline or EOF") - .expect("initial forwarding should succeed"); - - assert_eq!( - forwarded_bytes - .lock() - .expect("writer mutex should not poison") - .len(), - MAX_FIRST_FRAME_BYTES, - "only the bounded first-frame buffer should be held before streaming resumes" - ); - - drop(host_writer); -} - /// Constructs a host-stdin byte sequence containing a masked `initialize` /// frame followed by a follow-up frame, and returns both the raw input bytes /// and the expected post-masking output bytes for BDD assertion. @@ -518,5 +385,8 @@ pub(super) fn masked_initialize_with_follow_up() -> (Vec, Vec) { (host_stdin_bytes, expected) } +#[path = "protocol_acp_forwarding_tests.rs"] +mod forwarding_tests; + #[path = "protocol_acp_bdd_tests.rs"] mod bdd_tests; From a29633cc1a0a2ff58010aafb17b8fcd99f0ec718 Mon Sep 17 00:00:00 2001 From: leynos Date: Sat, 2 May 2026 00:14:03 +0200 Subject: [PATCH 4/6] Tighten ACP review follow-ups Reflow the hosted-mode ACP user-guide paragraph without suppressing Markdown line-length checks. Narrow the developer guide contract to the shipped first-frame initialize rewrite behaviour, including empty capability object removal. Use a test-specific timeout in the oversized ACP frame regression so CI timing is decoupled from production stdin settle timing. --- docs/developers-guide.md | 6 ++++-- docs/users-guide.md | 16 ++++++++-------- .../exec/protocol_acp_forwarding_tests.rs | 3 ++- 3 files changed, 14 insertions(+), 11 deletions(-) diff --git a/docs/developers-guide.md b/docs/developers-guide.md index 3aaf26ec..eaa5d48f 100644 --- a/docs/developers-guide.md +++ b/docs/developers-guide.md @@ -362,8 +362,10 @@ When adding a new execution mode: When protocol tests wrap `run_session` with `RecordingWriter` handles, they must preserve the Agentic Control Protocol (ACP) initialization rewrite contract: inspect only the first newline-delimited frame, rewrite only ACP -`initialize` requests that expose `params.clientCapabilities`, and remove the -`terminal` and `fs` capability families before the frame reaches the container. +`initialize` requests whose `params.clientCapabilities` contain blocked +`terminal` and/or `fs` entries, and remove those capability families before the +frame reaches the container. If masking leaves `params.clientCapabilities` (or +`clientCapabilities`) empty, drop the entire `clientCapabilities` object. Preserve the original line endings, and let malformed or non-ACP frames pass through unchanged. diff --git a/docs/users-guide.md b/docs/users-guide.md index 448c75a7..673ea02d 100644 --- a/docs/users-guide.md +++ b/docs/users-guide.md @@ -63,14 +63,14 @@ This subcommand is temporarily unavailable in the current release. If you run session. Use `podbot run` or the library exec API for now; the release notes will call out when hosted mode becomes available. - -When hosted mode uses Agentic Control Protocol (ACP), podbot masks `terminal/*` and `fs/*` capabilities -from the initial ACP `initialize` request before forwarding it to the -sandboxed agent. This masking currently applies only to the -protocol/library path used by hosted mode and ACP until `podbot host` is -implemented. The scope and behaviour may change when `podbot host` ships, but -for now it keeps the default trust boundary aligned with the container -sandbox even when the hosting client advertises broader ACP support. +When hosted mode uses Agentic Control Protocol (ACP), podbot masks +`terminal/*` and `fs/*` capabilities from the initial ACP `initialize` +request before forwarding it to the sandboxed agent. This masking currently +applies only to the protocol/library path used by hosted mode and ACP until +`podbot host` is implemented. The scope and behaviour may change when +`podbot host` ships, but for now it keeps the default trust boundary aligned +with the container sandbox even when the hosting client advertises broader ACP +support. | Option | Required | Default | Description | | -------------- | -------- | --------------- | ------------------------------------------ | diff --git a/src/engine/connection/exec/protocol_acp_forwarding_tests.rs b/src/engine/connection/exec/protocol_acp_forwarding_tests.rs index 14be5fa5..4c184939 100644 --- a/src/engine/connection/exec/protocol_acp_forwarding_tests.rs +++ b/src/engine/connection/exec/protocol_acp_forwarding_tests.rs @@ -46,6 +46,7 @@ fn forwarding_masks_initialize_and_preserves_trailing_bytes() { #[test] fn forwarding_does_not_wait_indefinitely_for_oversized_initial_frame() { let runtime = tokio::runtime::Runtime::new().expect("runtime should build"); + let test_timeout = std::time::Duration::from_secs(1); let host_stdin_bytes = vec![b'x'; MAX_FIRST_FRAME_BYTES + 1]; let (host_writer, host_reader) = runtime .block_on(async { @@ -64,7 +65,7 @@ fn forwarding_does_not_wait_indefinitely_for_oversized_initial_frame() { runtime .block_on(async { tokio::time::timeout( - STDIN_SETTLE_TIMEOUT, + test_timeout, forward_initial_acp_frame_async(&mut buffered_stdin, &mut container_input), ) .await From b078dc984703fcb83eb7a13311e60b52840995bf Mon Sep 17 00:00:00 2001 From: leynos Date: Sat, 2 May 2026 01:22:36 +0200 Subject: [PATCH 5/6] Document ACP forwarding concurrency Demote recoverable ACP frame parse and serialization failures to debug logs so valid pass-through cases do not appear as operational warnings. Document the per-session stdin forwarding ownership model and cancellation behaviour in the ACP helper module. --- src/engine/connection/exec/acp_helpers.rs | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/src/engine/connection/exec/acp_helpers.rs b/src/engine/connection/exec/acp_helpers.rs index bc404a36..030ae043 100644 --- a/src/engine/connection/exec/acp_helpers.rs +++ b/src/engine/connection/exec/acp_helpers.rs @@ -1,4 +1,14 @@ //! ACP initialization frame rewriting for protocol-mode stdin forwarding. +//! +//! ## Concurrency model +//! +//! Each protocol session runs its own independent forwarding task on a single +//! Tokio task. The `BufReader` wrapping host stdin is owned exclusively by that +//! task and is never shared across tasks or threads. All state is stack-local or +//! moved into the task at spawn time, so there are no shared-mutable references +//! and no synchronisation primitives are required. If the forwarding task is +//! cancelled at an `await` point the `BufReader` and container input writer are +//! both dropped, releasing the underlying pipe handles cleanly. use std::io; use std::pin::Pin; @@ -177,7 +187,7 @@ pub(super) fn mask_acp_initialize_frame(frame: &[u8]) -> Vec { let mut message: Value = match serde_json::from_slice(payload) { Ok(v) => v, Err(e) => { - tracing::warn!( + tracing::debug!( error = %e, "ACP frame JSON deserialization failed; forwarding unchanged" ); @@ -193,7 +203,7 @@ pub(super) fn mask_acp_initialize_frame(frame: &[u8]) -> Vec { let mut serialized = match serde_json::to_vec(&message) { Ok(v) => v, Err(e) => { - tracing::warn!( + tracing::debug!( error = %e, "ACP frame JSON serialization failed after masking; forwarding unchanged" ); From 959e38f9c0b653d030e7428630827402901b7315 Mon Sep 17 00:00:00 2001 From: leynos Date: Sat, 2 May 2026 01:52:32 +0200 Subject: [PATCH 6/6] Normalize ACP helper spelling Update ACP helper documentation and the related Clippy expectation reason to use Oxford -ize spellings consistently across initialization, serialization, and synchronization wording. --- src/engine/connection/exec/acp_helpers.rs | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/src/engine/connection/exec/acp_helpers.rs b/src/engine/connection/exec/acp_helpers.rs index 030ae043..234e0826 100644 --- a/src/engine/connection/exec/acp_helpers.rs +++ b/src/engine/connection/exec/acp_helpers.rs @@ -6,7 +6,7 @@ //! Tokio task. The `BufReader` wrapping host stdin is owned exclusively by that //! task and is never shared across tasks or threads. All state is stack-local or //! moved into the task at spawn time, so there are no shared-mutable references -//! and no synchronisation primitives are required. If the forwarding task is +//! and no synchronization primitives are required. If the forwarding task is //! cancelled at an `await` point the `BufReader` and container input writer are //! both dropped, releasing the underlying pipe handles cleanly. @@ -22,7 +22,7 @@ use super::STDIN_BUFFER_CAPACITY; /// ACP frame. Frames that exceed this limit are forwarded as-is without /// attempting JSON rewriting. pub(super) const MAX_FIRST_FRAME_BYTES: usize = STDIN_BUFFER_CAPACITY; -/// The JSON-RPC `method` value that identifies an ACP initialisation request. +/// The JSON-RPC `method` value that identifies an ACP initialization request. pub(super) const ACP_INITIALIZE_METHOD: &str = "initialize"; /// The `params` field that carries the client's advertised ACP capabilities. pub(super) const ACP_CLIENT_CAPABILITIES_FIELD: &str = "clientCapabilities"; @@ -174,12 +174,12 @@ where } /// Rewrites an ACP `initialize` frame by removing `terminal` and `fs` from -/// `params.clientCapabilities`, then re-serialises the JSON preserving the +/// `params.clientCapabilities`, then re-serializes the JSON preserving the /// original line ending. Returns the original frame unchanged on any parse or -/// serialisation failure, or if no capabilities were removed. +/// serialization failure, or if no capabilities were removed. #[expect( clippy::cognitive_complexity, - reason = "inline parse and serialise fallback tracing keeps this review-required flow local" + reason = "inline parse and serialize fallback tracing keeps this review-required flow local" )] pub(super) fn mask_acp_initialize_frame(frame: &[u8]) -> Vec { let (payload, line_ending) = split_frame_line_ending(frame);