feat(engine): a fleet worker can carry an ACP session

The fleet host already supervises a program with process groups, a
parent-death pipe, stop/restart, memory and logs. It wired stdio for logging,
so an agent that speaks the Agent Client Protocol had no channel. This adds a
transport to the worker start request:

- `FleetWorkerTransport::Logs` is today's behaviour, unchanged.
- `FleetWorkerTransport::JsonRpc` gives the child piped stdin/stdout and hands
  the caller a `FleetWorkerIo` exactly once. Parent death still works: the
  adapter holds the write end of stdin, so its death reaches the child as EOF.
  The SSH adapter refuses this transport rather than pretending to support it.
- stderr goes to the worker log in both shapes.

`fleet::acp` speaks the protocol over that channel: newline-framed JSON-RPC,
initialize, session/new, session/prompt, session/cancel, and
session/request_permission answered by policy (allow once, reject once,
cancel). Filesystem and terminal client methods are advertised false, no
credential is read, and no update is mapped onto a journal item yet.

Verified with a dependency-free fixture agent through the real host, under the
repo's own test environment (RUST_MIN_STACK=16 MiB, which every fleet test
needs): streamed text, an edit tool call writes a file only when allowed and
nothing when rejected, an unlisted environment variable never reaches the
child, a cancel settles a prompt that never answers, and a Logs worker still
writes its log and has no duplex io. 3 tests, all passing.

crates/engine/src/fleet/acp.rs

@@ -0,0 +1,449 @@
+//! A narrow Agent Client Protocol client over a fleet worker's duplex stdio.
+//!
+//! The fleet host owns the process: groups, parent-death, stop, restart and
+//! logs. This module owns only the wire: newline-framed JSON-RPC, one request
+//! at a time, and the one client-side request the protocol needs in this slice
+//! (`session/request_permission`), answered by a fixed policy rather than a
+//! person.
+//!
+//! Deliberately absent: filesystem and terminal client methods (advertised
+//! false), elicitation, session load/fork, and any credential handling. The
+//! agent runs in its own process with its own tools, and its stderr stays in
+//! the worker log.
+
+use std::io::Write;
+use std::path::{Path, PathBuf};
+use std::sync::mpsc::{channel, Receiver};
+use std::sync::{Arc, Mutex};
+use std::time::{Duration, Instant};
+
+use serde_json::{json, Value};
+
+use super::host::FleetWorkerIo;
+
+const CLIENT_NAME: &str = "codewhale";
+
+/// How the client answers `session/request_permission`.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum PermissionPolicy {
+    /// Choose the first option the agent labels `allow_once`.
+    AllowOnce,
+    /// Choose the first option the agent labels `reject_once`.
+    RejectOnce,
+    /// Answer `cancelled`, the protocol's "no decision" outcome.
+    Cancel,
+}
+
+impl PermissionPolicy {
+    fn option_kind(self) -> &'static str {
+        match self {
+            PermissionPolicy::AllowOnce => "allow_once",
+            PermissionPolicy::RejectOnce => "reject_once",
+            PermissionPolicy::Cancel => "",
+        }
+    }
+}
+
+#[derive(Debug)]
+pub struct AcpError {
+    message: String,
+}
+
+impl AcpError {
+    fn new(message: impl Into<String>) -> Self {
+        Self { message: message.into() }
+    }
+
+    fn timeout(method: &str) -> Self {
+        Self::new(format!("{method} did not answer before the deadline"))
+    }
+}
+
+impl std::fmt::Display for AcpError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str(&self.message)
+    }
+}
+
+impl std::error::Error for AcpError {}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct AgentIdentity {
+    pub name: String,
+    pub version: String,
+    pub protocol_version: u64,
+}
+
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+pub struct PromptOutcome {
+    pub stop_reason: String,
+    pub text: String,
+    pub tool_calls: Vec<String>,
+    pub decisions: Vec<String>,
+}
+
+/// One live ACP conversation over a worker's pipes.
+pub struct AcpClient {
+    stdin: Arc<Mutex<std::process::ChildStdin>>,
+    lines: Receiver<String>,
+    next_id: u64,
+}
+
+/// A write-only handle that can send `session/cancel` while a prompt blocks
+/// the client that owns the reader.
+#[derive(Clone)]
+pub struct AcpCanceller {
+    stdin: Arc<Mutex<std::process::ChildStdin>>,
+    session: String,
+}
+
+impl AcpCanceller {
+    pub fn cancel(&self) -> Result<(), AcpError> {
+        let mut stdin = self.stdin.lock().map_err(|_| AcpError::new("the agent's stdin lock was poisoned"))?;
+        let line = serde_json::to_string(&json!({"jsonrpc": "2.0", "method": "session/cancel", "params": {"sessionId": self.session}}))
+            .map_err(|err| AcpError::new(format!("encoding a cancel failed: {err}")))?;
+        stdin
+            .write_all(format!("{line}\n").as_bytes())
+            .and_then(|()| stdin.flush())
+            .map_err(|err| AcpError::new(format!("the agent's stdin closed: {err}")))
+    }
+}
+
+impl std::fmt::Debug for AcpClient {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("AcpClient").field("next_id", &self.next_id).finish_non_exhaustive()
+    }
+}
+
+impl AcpClient {
+    pub fn new(io: FleetWorkerIo) -> Self {
+        let FleetWorkerIo { stdin, stdout } = io;
+        let (tx, lines) = channel();
+        std::thread::spawn(move || {
+            use std::io::BufRead as _;
+            let mut stdout = stdout;
+            let mut line = String::new();
+            loop {
+                line.clear();
+                match stdout.read_line(&mut line) {
+                    Ok(0) | Err(_) => break,
+                    Ok(_) => {
+                        let trimmed = line.trim_end().to_string();
+                        if trimmed.is_empty() {
+                            continue;
+                        }
+                        if tx.send(trimmed).is_err() {
+                            break;
+                        }
+                    }
+                }
+            }
+        });
+        Self { stdin: Arc::new(Mutex::new(stdin)), lines, next_id: 1 }
+    }
+
+    fn write(&self, message: &Value) -> Result<(), AcpError> {
+        let mut line = serde_json::to_string(message).map_err(|err| AcpError::new(format!("encoding a request failed: {err}")))?;
+        line.push('\n');
+        let mut stdin = self.stdin.lock().map_err(|_| AcpError::new("the agent's stdin lock was poisoned"))?;
+        stdin
+            .write_all(line.as_bytes())
+            .and_then(|()| stdin.flush())
+            .map_err(|err| AcpError::new(format!("the agent's stdin closed: {err}")))
+    }
+
+    /// A handle that can settle a blocked prompt from another thread.
+    pub fn canceller(&self, session: &str) -> AcpCanceller {
+        AcpCanceller { stdin: Arc::clone(&self.stdin), session: session.to_string() }
+    }
+
+    fn next(&mut self) -> u64 {
+        let id = self.next_id;
+        self.next_id += 1;
+        id
+    }
+
+    fn read(&mut self, deadline: Instant, method: &str) -> Result<Value, AcpError> {
+        let left = deadline.saturating_duration_since(Instant::now());
+        if left.is_zero() {
+            return Err(AcpError::timeout(method));
+        }
+        let line = self.lines.recv_timeout(left).map_err(|_| AcpError::timeout(method))?;
+        serde_json::from_str(&line).map_err(|err| AcpError::new(format!("the agent sent a line that is not JSON: {err}")))
+    }
+
+    /// Send a request and wait for its response, answering the agent's
+    /// permission requests with `policy` and reporting every other
+    /// notification to `on_update`.
+    fn request<F>(&mut self, method: &str, params: Value, policy: PermissionPolicy, deadline: Instant, mut on_update: F) -> Result<Value, AcpError>
+    where
+        F: FnMut(&Value),
+    {
+        let id = self.next();
+        self.write(&json!({"jsonrpc": "2.0", "id": id, "method": method, "params": params}))?;
+        loop {
+            let message = self.read(deadline, method)?;
+            if message.get("id").and_then(Value::as_u64) == Some(id) && (message.get("result").is_some() || message.get("error").is_some()) {
+                if let Some(error) = message.get("error") {
+                    let text = error.get("message").and_then(Value::as_str).unwrap_or("the agent refused");
+                    return Err(AcpError::new(format!("{method} failed: {text}")));
+                }
+                return Ok(message.get("result").cloned().unwrap_or(Value::Null));
+            }
+            let Some(incoming) = message.get("method").and_then(Value::as_str) else { continue };
+            let incoming_id = message.get("id").cloned();
+            let incoming_params = message.get("params").cloned().unwrap_or(Value::Null);
+            if incoming == "session/request_permission" {
+                self.answer_permission(incoming_id, &incoming_params, policy)?;
+            } else {
+                on_update(&json!({"method": incoming, "params": incoming_params}));
+            }
+        }
+    }
+
+    fn answer_permission(&mut self, id: Option<Value>, params: &Value, policy: PermissionPolicy) -> Result<(), AcpError> {
+        let Some(id) = id else { return Ok(()) };
+        let outcome = if policy == PermissionPolicy::Cancel {
+            json!({"outcome": "cancelled"})
+        } else {
+            let wanted = policy.option_kind();
+            let chosen = params
+                .get("options")
+                .and_then(Value::as_array)
+                .and_then(|options| {
+                    options.iter().find(|option| option.get("kind").and_then(Value::as_str) == Some(wanted))
+                })
+                .and_then(|option| option.get("optionId").and_then(Value::as_str));
+            match chosen {
+                Some(option_id) => json!({"outcome": "selected", "optionId": option_id}),
+                None => json!({"outcome": "cancelled"}),
+            }
+        };
+        self.write(&json!({"jsonrpc": "2.0", "id": id, "result": {"outcome": outcome}}))
+    }
+
+    pub fn initialize(&mut self, timeout: Duration) -> Result<AgentIdentity, AcpError> {
+        let deadline = Instant::now() + timeout;
+        let result = self.request(
+            "initialize",
+            json!({
+                "protocolVersion": 1,
+                "clientCapabilities": {"fs": {"readTextFile": false, "writeTextFile": false}, "terminal": false},
+                "clientInfo": {"name": CLIENT_NAME, "version": env!("CARGO_PKG_VERSION")}
+            }),
+            PermissionPolicy::Cancel,
+            deadline,
+            |_| {},
+        )?;
+        let info = result.get("agentInfo").cloned().unwrap_or(Value::Null);
+        Ok(AgentIdentity {
+            name: info.get("name").and_then(Value::as_str).unwrap_or("agent").to_string(),
+            version: info.get("version").and_then(Value::as_str).unwrap_or("0").to_string(),
+            protocol_version: result.get("protocolVersion").and_then(Value::as_u64).unwrap_or(1),
+        })
+    }
+
+    pub fn new_session(&mut self, cwd: &Path, timeout: Duration) -> Result<String, AcpError> {
+        let deadline = Instant::now() + timeout;
+        let result = self.request(
+            "session/new",
+            json!({"cwd": cwd.display().to_string(), "mcpServers": []}),
+            PermissionPolicy::Cancel,
+            deadline,
+            |_| {},
+        )?;
+        result
+            .get("sessionId")
+            .and_then(Value::as_str)
+            .map(str::to_string)
+            .ok_or_else(|| AcpError::new("session/new did not name a session"))
+    }
+
+    /// One prompt. Streams are collected into the outcome, which is enough for
+    /// a caller that has not yet mapped ACP updates onto journal items.
+    pub fn prompt(&mut self, session: &str, text: &str, policy: PermissionPolicy, timeout: Duration) -> Result<PromptOutcome, AcpError> {
+        let deadline = Instant::now() + timeout;
+        let mut outcome = PromptOutcome::default();
+        let result = self.request(
+            "session/prompt",
+            json!({"sessionId": session, "prompt": [{"type": "text", "text": text}]}),
+            policy,
+            deadline,
+            |message| {
+                if message.get("method").and_then(Value::as_str) != Some("session/update") {
+                    return;
+                }
+                let Some(update) = message.get("params").and_then(|params| params.get("update")) else { return };
+                match update.get("sessionUpdate").and_then(Value::as_str).unwrap_or("") {
+                    "agent_message_chunk" => {
+                        if let Some(chunk) = update.get("content").and_then(|content| content.get("text")).and_then(Value::as_str) {
+                            outcome.text.push_str(chunk);
+                        }
+                    }
+                    "tool_call" => {
+                        let title = update.get("title").and_then(Value::as_str).unwrap_or("tool");
+                        let kind = update.get("kind").and_then(Value::as_str).unwrap_or("other");
+                        outcome.tool_calls.push(format!("{kind}:{title}"));
+                    }
+                    _ => {}
+                }
+            },
+        )?;
+        outcome.stop_reason = result.get("stopReason").and_then(Value::as_str).unwrap_or("end_turn").to_string();
+        Ok(outcome)
+    }
+
+    pub fn cancel(&self, session: &str) -> Result<(), AcpError> {
+        self.write(&json!({"jsonrpc": "2.0", "method": "session/cancel", "params": {"sessionId": session}}))
+    }
+
+    /// Record a decision the caller made on the agent's behalf. The client
+    /// answers permissions inline, so this only exists for callers that want
+    /// the option id in their own record.
+    pub fn permission_option(params: &Value, policy: PermissionPolicy) -> Option<String> {
+        let wanted = policy.option_kind();
+        params
+            .get("options")
+            .and_then(Value::as_array)
+            .and_then(|options| options.iter().find(|option| option.get("kind").and_then(Value::as_str) == Some(wanted)))
+            .and_then(|option| option.get("optionId").and_then(Value::as_str))
+            .map(str::to_string)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::fleet::host::{
+        FleetHostAdapter, FleetHostWorkerState, FleetWorkerCommand, FleetWorkerStartRequest,
+        FleetWorkerTransport, LocalProcessFleetHostAdapter,
+    };
+    use std::collections::{BTreeMap, BTreeSet};
+
+    fn fixture() -> PathBuf {
+        PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures/acp-agent.mjs")
+    }
+
+    fn node() -> Option<String> {
+        let output = std::process::Command::new("node").arg("--version").output().ok()?;
+        output.status.success().then(|| "node".to_string())
+    }
+
+    fn json_rpc_request(env: BTreeMap<String, String>, allow: BTreeSet<String>) -> FleetWorkerStartRequest {
+        let mut request = FleetWorkerStartRequest::new(
+            "w-acp",
+            FleetWorkerCommand::new("node", [fixture().display().to_string()]),
+        );
+        request.transport = FleetWorkerTransport::JsonRpc;
+        request.env = env;
+        request.env_allowlist = allow;
+        request
+    }
+
+    #[test]
+    fn a_json_rpc_worker_speaks_acp_end_to_end() {
+        let Some(_node) = node() else {
+            return;
+        };
+        let workspace = tempfile::tempdir().unwrap();
+        let mut host = LocalProcessFleetHostAdapter::new(workspace.path());
+        let mut request = json_rpc_request(
+            BTreeMap::from([("CODEWHALE_TEST_SECRET".to_string(), "must-not-travel".to_string())]),
+            BTreeSet::from(["PATH".to_string()]),
+        );
+        request.cwd = Some(workspace.path().to_path_buf());
+        host.start_worker(request).unwrap();
+
+        let io = host.take_worker_io("w-acp").expect("a JsonRpc worker yields its pipes");
+        assert!(host.take_worker_io("w-acp").is_none(), "the stream is taken exactly once");
+        let mut client = AcpClient::new(io);
+
+        let identity = client.initialize(Duration::from_secs(20)).unwrap();
+        assert_eq!(identity.name, "fixture-agent");
+        assert_eq!(identity.protocol_version, 1);
+
+        let session = client.new_session(workspace.path(), Duration::from_secs(20)).unwrap();
+        let allowed = client
+            .prompt(&session, "hello fleet", PermissionPolicy::AllowOnce, Duration::from_secs(20))
+            .unwrap();
+        assert_eq!(allowed.stop_reason, "end_turn");
+        assert!(allowed.text.contains("hello fleet"), "streamed text: {:?}", allowed.text);
+        assert!(
+            allowed.tool_calls.iter().any(|call| call == "edit:Write fixture.txt"),
+            "tool calls: {:?}",
+            allowed.tool_calls
+        );
+        assert!(workspace.path().join("fixture.txt").exists(), "an allowed edit wrote its file");
+
+        std::fs::remove_file(workspace.path().join("fixture.txt")).unwrap();
+        let rejected = client
+            .prompt(&session, "hello again", PermissionPolicy::RejectOnce, Duration::from_secs(20))
+            .unwrap();
+        assert_eq!(rejected.stop_reason, "end_turn");
+        assert!(!workspace.path().join("fixture.txt").exists(), "a rejected edit wrote nothing");
+
+        let dumped = std::fs::read_to_string(workspace.path().join("fixture-env.json")).unwrap();
+        assert!(
+            !dumped.contains("CODEWHALE_TEST_SECRET"),
+            "an unlisted variable never reached the agent"
+        );
+
+        let status = host.stop_worker("w-acp").unwrap();
+        assert_ne!(status.state, FleetHostWorkerState::Running);
+    }
+
+    #[test]
+    fn a_logs_worker_keeps_its_log_and_has_no_duplex_io() {
+        let workspace = tempfile::tempdir().unwrap();
+        let mut host = LocalProcessFleetHostAdapter::new(workspace.path());
+        let mut request = FleetWorkerStartRequest::new(
+            "w-logs",
+            FleetWorkerCommand::new("sh", ["-c", "echo logged-line"]),
+        );
+        request.cwd = Some(workspace.path().to_path_buf());
+        let handle = host.start_worker(request).unwrap();
+        assert!(host.take_worker_io("w-logs").is_none(), "a Logs worker has no duplex io");
+
+        let deadline = std::time::Instant::now() + Duration::from_secs(10);
+        let mut log = String::new();
+        while std::time::Instant::now() < deadline {
+            log = host.read_logs("w-logs", 4096).unwrap();
+            if log.contains("logged-line") {
+                break;
+            }
+            std::thread::sleep(Duration::from_millis(50));
+        }
+        assert!(log.contains("logged-line"), "log was {log:?}");
+        assert!(handle.log_path.exists());
+    }
+
+    #[test]
+    fn a_cancel_settles_a_prompt_that_never_answers() {
+        let Some(_node) = node() else {
+            return;
+        };
+        let workspace = tempfile::tempdir().unwrap();
+        let mut host = LocalProcessFleetHostAdapter::new(workspace.path());
+        let mut request = json_rpc_request(
+            BTreeMap::from([("HANG".to_string(), "1".to_string())]),
+            BTreeSet::from(["HANG".to_string()]),
+        );
+        request.cwd = Some(workspace.path().to_path_buf());
+        host.start_worker(request).unwrap();
+
+        let io = host.take_worker_io("w-acp").expect("duplex io");
+        let mut client = AcpClient::new(io);
+        client.initialize(Duration::from_secs(20)).unwrap();
+        let session = client.new_session(workspace.path(), Duration::from_secs(20)).unwrap();
+        let canceller = client.canceller(&session);
+        let prompt_session = session.clone();
+        let blocked = std::thread::spawn(move || {
+            client.prompt(&prompt_session, "hang", PermissionPolicy::AllowOnce, Duration::from_secs(30))
+        });
+        std::thread::sleep(Duration::from_millis(400));
+        canceller.cancel().unwrap();
+        let outcome = blocked.join().unwrap().unwrap();
+        assert_eq!(outcome.stop_reason, "cancelled");
+        host.stop_worker("w-acp").unwrap();
+    }
+}

crates/engine/src/fleet/host.rs

@@ -98,6 +98,31 @@ impl FleetWorkerCommand {
     }
 }
 
+/// How the adapter wires a worker's stdio.
+///
+/// `Logs` is the dispatcher shape: stdin is the parent-death pipe and both
+/// output streams go to the worker log. `JsonRpc` keeps the parent-death
+/// property (the write end of stdin lives in this adapter) but hands the
+/// caller a duplex channel for a line-framed protocol such as ACP.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub enum FleetWorkerTransport {
+    #[default]
+    Logs,
+    JsonRpc,
+}
+
+/// The duplex stdio of a `JsonRpc` worker, taken once by its owner.
+pub struct FleetWorkerIo {
+    pub stdin: std::process::ChildStdin,
+    pub stdout: std::io::BufReader<std::process::ChildStdout>,
+}
+
+impl std::fmt::Debug for FleetWorkerIo {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("FleetWorkerIo").finish_non_exhaustive()
+    }
+}
+
 #[derive(Debug, Clone)]
 pub struct FleetWorkerStartRequest {
     pub worker_id: String,
@@ -107,6 +132,7 @@ pub struct FleetWorkerStartRequest {
     pub env: BTreeMap<String, String>,
     pub env_allowlist: BTreeSet<String>,
     pub log_limit_bytes: usize,
+    pub transport: FleetWorkerTransport,
 }
 
 impl FleetWorkerStartRequest {
@@ -119,6 +145,7 @@ impl FleetWorkerStartRequest {
             env: BTreeMap::new(),
             env_allowlist: BTreeSet::new(),
             log_limit_bytes: DEFAULT_LOG_LIMIT_BYTES,
+            transport: FleetWorkerTransport::Logs,
         }
     }
 }
@@ -170,6 +197,12 @@ pub trait FleetHostAdapter {
     fn restart_worker(&mut self, worker_id: &str) -> FleetHostResult<FleetWorkerHandle>;
     fn stop_worker(&mut self, worker_id: &str) -> FleetHostResult<FleetHostWorkerStatus>;
     fn cleanup_worker(&mut self, worker_id: &str) -> FleetHostResult<()>;
+
+    /// Take the duplex stdio of a `JsonRpc` worker. The first call wins, so
+    /// two readers can never split one stream.
+    fn take_worker_io(&mut self, _worker_id: &str) -> Option<FleetWorkerIo> {
+        None
+    }
 }
 
 #[derive(Debug)]
@@ -193,6 +226,7 @@ struct LocalWorkerProcess {
     stopped: bool,
     last_exit: Option<ExitStatus>,
     last_memory_mb: Option<u64>,
+    io: Option<FleetWorkerIo>,
     /// When `ps` last sampled this worker. Status polls can run every few
     /// milliseconds; memory is display data, so one sample a second is ample.
     last_memory_sample: Option<std::time::Instant>,
@@ -242,25 +276,42 @@ impl LocalProcessFleetHostAdapter {
             .map_err(|err| FleetHostError::retryable(format!("cloning worker log: {err}")))?;
 
         let mut command = Command::new(&request.command.program);
-        // Parent-death watch (R7): the worker's stdin is the read end of a pipe
-        // whose write end lives only in this adapter process. If the manager
-        // dies (crash, kill, power loss), the kernel closes the write end, the
-        // worker sees stdin EOF, and `--parent-death-watch` shuts the worker
-        // tree down instead of letting it spend forever. Windows workers are
-        // contained in a Job Object that the OS terminates on parent death.
+        // Parent-death watch (R7): a `Logs` worker's stdin is the read end of a
+        // pipe whose write end lives only in this adapter process. If the
+        // manager dies (crash, kill, power loss), the kernel closes the write
+        // end, the worker sees stdin EOF, and `--parent-death-watch` shuts the
+        // worker tree down instead of letting it spend forever. A `JsonRpc`
+        // worker keeps the same property: this adapter holds the write end of
+        // its stdin, so adapter death still reaches it as EOF. Windows workers
+        // are contained in a Job Object that the OS terminates on parent death.
+        let duplex = request.transport == FleetWorkerTransport::JsonRpc;
         #[cfg(unix)]
-        let (parent_death_writer, stdin) = {
-            let (reader, writer) = std::io::pipe().map_err(|err| {
-                FleetHostError::retryable(format!("creating parent-death pipe: {err}"))
-            })?;
-            (Some(writer), std::process::Stdio::from(reader))
+        let mut parent_death_writer = None;
+        let stdin = if duplex {
+            Stdio::piped()
+        } else {
+            #[cfg(unix)]
+            {
+                let (reader, writer) = std::io::pipe().map_err(|err| {
+                    FleetHostError::retryable(format!("creating parent-death pipe: {err}"))
+                })?;
+                parent_death_writer = Some(writer);
+                std::process::Stdio::from(reader)
+            }
+            #[cfg(not(unix))]
+            {
+                Stdio::null()
+            }
+        };
+        let stdout = if duplex {
+            Stdio::piped()
+        } else {
+            Stdio::from(log)
         };
-        #[cfg(not(unix))]
-        let stdin = Stdio::null();
         command
             .args(&request.command.args)
             .stdin(stdin)
-            .stdout(Stdio::from(log))
+            .stdout(stdout)
             .stderr(Stdio::from(stderr))
             .env_clear()
             .envs(env);
@@ -288,6 +339,8 @@ impl LocalProcessFleetHostAdapter {
         let child = command.spawn().map_err(|err| {
             classify_spawn_error(err, format!("starting worker {}", request.worker_id))
         })?;
+        #[cfg(unix)]
+        let mut child = child;
         #[cfg(windows)]
         let (child, windows_job) = attach_fleet_windows_job(child).map_err(|err| {
             FleetHostError::retryable(format!(
@@ -296,6 +349,27 @@ impl LocalProcessFleetHostAdapter {
             ))
         })?;
         let pid = child.id();
+        let io = match request.transport {
+            FleetWorkerTransport::JsonRpc => {
+                let stdin = child.stdin.take().ok_or_else(|| {
+                    FleetHostError::retryable(format!(
+                        "worker {} has no stdin pipe",
+                        request.worker_id
+                    ))
+                })?;
+                let stdout = child.stdout.take().ok_or_else(|| {
+                    FleetHostError::retryable(format!(
+                        "worker {} has no stdout pipe",
+                        request.worker_id
+                    ))
+                })?;
+                Some(FleetWorkerIo {
+                    stdin,
+                    stdout: std::io::BufReader::new(stdout),
+                })
+            }
+            FleetWorkerTransport::Logs => None,
+        };
         let handle = FleetWorkerHandle {
             transport_receipt: None,
             worker_id: request.worker_id.clone(),
@@ -321,6 +395,7 @@ impl LocalProcessFleetHostAdapter {
                 stopped: false,
                 last_exit: None,
                 last_memory_mb: None,
+                io,
                 last_memory_sample: None,
             },
         );
@@ -499,6 +574,10 @@ impl FleetHostAdapter for LocalProcessFleetHostAdapter {
         self.processes.remove(worker_id);
         Ok(())
     }
+
+    fn take_worker_io(&mut self, worker_id: &str) -> Option<FleetWorkerIo> {
+        self.processes.get_mut(worker_id)?.io.take()
+    }
 }
 
 #[derive(Debug, Clone)]
@@ -748,6 +827,11 @@ impl SshFleetHostAdapter {
         &self,
         request: FleetWorkerStartRequest,
     ) -> FleetHostResult<FleetWorkerStartRequest> {
+        if request.transport != FleetWorkerTransport::Logs {
+            return Err(FleetHostError::configuration(
+                "a JSON-RPC worker cannot run over the SSH host adapter yet",
+            ));
+        }
         let command = self.build_ssh_command(&request)?;
         let mut env = ssh_client_env();
         env.extend(filtered_env(&request.env, &self.config.env_allowlist)?);
@@ -760,6 +844,7 @@ impl SshFleetHostAdapter {
             env,
             env_allowlist,
             log_limit_bytes: request.log_limit_bytes,
+            transport: request.transport,
         })
     }
 

crates/engine/src/fleet/mod.rs

@@ -1,5 +1,6 @@
 //! Agent Fleet control plane — local-first manager, ledger, and workers.
 
+pub mod acp;
 pub mod alerts;
 pub mod artifacts;
 pub mod capability_badges;

crates/engine/tests/fixtures/acp-agent.mjs

@@ -0,0 +1,114 @@
+// A deterministic Agent Client Protocol agent for the fleet transport test.
+// Speaks newline-framed JSON-RPC on stdio. No dependencies, no network.
+import { writeFileSync } from 'node:fs';
+import { join } from 'node:path';
+
+const HANG = process.env.HANG === '1';
+const envPath = join(process.cwd(), 'fixture-env.json');
+writeFileSync(envPath, JSON.stringify(process.env));
+
+let buffer = '';
+let sessionId = null;
+let cwd = process.cwd();
+let permissionWaiter = null;
+let pendingPrompt = null;
+
+const send = (message) => process.stdout.write(`${JSON.stringify(message)}\n`);
+const result = (id, value) => send({ jsonrpc: '2.0', id, result: value });
+const notify = (method, params) => send({ jsonrpc: '2.0', method, params });
+
+function askPermission(toolCall) {
+  return new Promise((resolve) => {
+    permissionWaiter = resolve;
+    send({
+      jsonrpc: '2.0',
+      id: 9001,
+      method: 'session/request_permission',
+      params: {
+        sessionId,
+        toolCall,
+        options: [
+          { optionId: 'allow', name: 'Allow once', kind: 'allow_once' },
+          { optionId: 'reject', name: 'Reject once', kind: 'reject_once' },
+        ],
+      },
+    });
+  });
+}
+
+async function prompt(request) {
+  const text = (request.params?.prompt ?? []).map((block) => block.text ?? '').join('');
+  for (const chunk of [`You said: `, text, `.`]) {
+    notify('session/update', { sessionId, update: { sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: chunk } } });
+  }
+  const toolCall = { toolCallId: 'call-1', title: 'Write fixture.txt', kind: 'edit', status: 'pending' };
+  notify('session/update', { sessionId, update: { sessionUpdate: 'tool_call', ...toolCall } });
+  const answer = await askPermission(toolCall);
+  const outcome = answer?.outcome ?? {};
+  if (outcome.outcome === 'selected' && outcome.optionId === 'allow') {
+    writeFileSync(join(cwd, 'fixture.txt'), 'written by the fixture agent\n');
+    notify('session/update', { sessionId, update: { sessionUpdate: 'tool_call_update', toolCallId: 'call-1', status: 'completed' } });
+  } else {
+    notify('session/update', { sessionId, update: { sessionUpdate: 'tool_call_update', toolCallId: 'call-1', status: 'failed' } });
+  }
+  result(request.id, { stopReason: 'end_turn' });
+}
+
+function handle(message) {
+  if (message.id !== undefined && message.method === undefined) {
+    if (message.id === 9001 && permissionWaiter) {
+      const resolve = permissionWaiter;
+      permissionWaiter = null;
+      resolve(message.result);
+    }
+    return;
+  }
+  switch (message.method) {
+    case 'initialize':
+      result(message.id, {
+        protocolVersion: 1,
+        agentInfo: { name: 'fixture-agent', version: '1.0.0' },
+        authMethods: [],
+        agentCapabilities: { loadSession: true },
+      });
+      break;
+    case 'session/new':
+      sessionId = 'fixture-session-1';
+      cwd = message.params?.cwd ?? cwd;
+      result(message.id, { sessionId });
+      break;
+    case 'session/prompt':
+      if (HANG) {
+        pendingPrompt = message;
+        break;
+      }
+      void prompt(message);
+      break;
+    case 'session/cancel':
+      if (pendingPrompt) {
+        result(pendingPrompt.id, { stopReason: 'cancelled' });
+        pendingPrompt = null;
+      }
+      break;
+    default:
+      if (message.id !== undefined) {
+        send({ jsonrpc: '2.0', id: message.id, error: { code: -32601, message: `no method ${message.method}` } });
+      }
+  }
+}
+
+process.stdin.on('data', (chunk) => {
+  buffer += chunk;
+  let index;
+  while ((index = buffer.indexOf('\n')) >= 0) {
+    const line = buffer.slice(0, index);
+    buffer = buffer.slice(index + 1);
+    if (!line.trim()) continue;
+    try {
+      handle(JSON.parse(line));
+    } catch (error) {
+      process.stderr.write(`fixture: bad line: ${String(error)}\n`);
+    }
+  }
+});
+process.stdin.on('end', () => process.exit(0));