feat(maestro): add maestro-acp, the ACP client per structured session (#134)
New crate wrapping agent-client-protocol 2.1 in the client role. spawn() resolves claude-code-acp on the PATH the child would run with, starts it in the session cwd with the session env, runs initialize + session/new, and returns an AcpSession holding the connection, the ACP session id, and an unbounded channel of raw session/update notifications for #135. initialize declares fs.readTextFile, fs.writeTextFile and terminal as false so the adapter uses Claude Code's own tools. request_permission is a stub that declines until #139. terminate() sends session/cancel, waits, SIGTERMs, then SIGKILLs. The child stays in the daemon's process group and gets SIGTERM on daemon death via PR_SET_PDEATHSIG. CLAUDECODE is stripped from the child env: the adapter refuses to start when it believes it is nested in another Claude Code session, which is how a daemon launched from one behaves. Tests run against tests/fake_acp_agent.py, a JSON-RPC-over-stdio stand-in that logs what it receives; real_adapter_handshake is #[ignore]d.
This commit is contained in:
Miguel Palhas
@@ -0,0 +1,20 @@
|
||||
[package]
|
||||
name = "maestro-acp"
|
||||
version.workspace = true
|
||||
edition.workspace = true
|
||||
license.workspace = true
|
||||
repository.workspace = true
|
||||
publish.workspace = true
|
||||
description = "ACP client role: one claude-code-acp child per structured session"
|
||||
|
||||
[dependencies]
|
||||
agent-client-protocol.workspace = true
|
||||
libc.workspace = true
|
||||
serde_json.workspace = true
|
||||
thiserror.workspace = true
|
||||
tokio.workspace = true
|
||||
tokio-util.workspace = true
|
||||
tracing.workspace = true
|
||||
|
||||
[dev-dependencies]
|
||||
tempfile.workspace = true
|
||||
@@ -0,0 +1,468 @@
|
||||
//! ACP client role for the daemon (#134): one `claude-code-acp` child per
|
||||
//! structured session.
|
||||
//!
|
||||
//! The daemon is the client; the adapter is the agent. [`spawn`] starts the
|
||||
//! adapter as a child process with stdio pipes, in the session's `cwd` with
|
||||
//! the session's env, runs the `initialize` + `session/new` handshake, and
|
||||
//! hands back an [`AcpSession`] that owns the connection for the life of the
|
||||
//! session.
|
||||
//!
|
||||
//! **The handshake declares almost nothing.** `initialize` reports
|
||||
//! `fs.readTextFile`, `fs.writeTextFile` and `terminal` as `false`, so the
|
||||
//! adapter uses Claude Code's own Read / Write / Bash instead of asking the
|
||||
//! daemon to act on its behalf — exactly what already happens in the
|
||||
//! terminal view. `session/request_permission` is the one client method that
|
||||
//! gets a real implementation, in #139; here it is a stub that declines.
|
||||
//!
|
||||
//! Every `session/update` notification the adapter sends lands on an
|
||||
//! unbounded channel the session owns ([`AcpSession::take_updates`]). Nothing
|
||||
//! is dropped while nobody reads: #135 takes the receiver and persists each
|
||||
//! notification to the per-session event log before broadcasting it.
|
||||
//!
|
||||
//! The child is a child of the daemon, full stop. It shares the daemon's
|
||||
//! process group, is marked to receive `SIGTERM` when the daemon dies
|
||||
//! (`PR_SET_PDEATHSIG` on Linux), and [`AcpSession::terminate`] is what both
|
||||
//! `DELETE /api/sessions/:id` and daemon shutdown call: `session/cancel`, a
|
||||
//! short wait, then `SIGTERM`, then `SIGKILL` as the last resort.
|
||||
|
||||
#![deny(missing_debug_implementations)]
|
||||
|
||||
use std::collections::{HashMap, VecDeque};
|
||||
use std::ffi::OsString;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::process::{ExitStatus, Stdio};
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::Duration;
|
||||
|
||||
use agent_client_protocol::schema::v1::{
|
||||
CancelNotification, ClientCapabilities, FileSystemCapabilities, Implementation,
|
||||
InitializeRequest, NewSessionRequest, RequestPermissionOutcome, RequestPermissionRequest,
|
||||
RequestPermissionResponse, SessionId, SessionNotification,
|
||||
};
|
||||
use agent_client_protocol::schema::ProtocolVersion;
|
||||
use agent_client_protocol::{ByteStreams, Client};
|
||||
use tokio::io::AsyncBufReadExt;
|
||||
use tokio::sync::{mpsc, oneshot, watch};
|
||||
use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt};
|
||||
|
||||
pub use agent_client_protocol::schema::v1 as schema;
|
||||
|
||||
/// The adapter binary, resolved on `PATH`.
|
||||
pub const ADAPTER_BIN: &str = "claude-code-acp";
|
||||
|
||||
/// How to install the adapter when it is missing. Named in the error so an
|
||||
/// operator staring at a failed session knows what to run.
|
||||
pub const INSTALL_COMMAND: &str =
|
||||
"npm install -g --prefix ~/.local @zed-industries/claude-code-acp";
|
||||
|
||||
/// How long `initialize` + `session/new` may take before the spawn is
|
||||
/// declared failed. The real adapter boots a node process and the Claude
|
||||
/// Code SDK behind it; a slow machine takes seconds, never a minute.
|
||||
pub const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(60);
|
||||
|
||||
/// How long [`AcpSession::terminate`] waits after `session/cancel` before
|
||||
/// sending `SIGTERM`. Cancel is a courtesy that lets the adapter finish the
|
||||
/// turn it is on; the signal is what actually ends it.
|
||||
pub const CANCEL_GRACE: Duration = Duration::from_millis(500);
|
||||
|
||||
/// How long [`AcpSession::terminate`] waits after `SIGTERM` before `SIGKILL`.
|
||||
pub const TERM_GRACE: Duration = Duration::from_secs(2);
|
||||
|
||||
/// Bytes of the child's stderr kept for the failure message.
|
||||
const STDERR_TAIL_BYTES: usize = 4096;
|
||||
|
||||
/// What [`spawn`] needs to start one adapter.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct SpawnConfig {
|
||||
/// Working directory of the child and `cwd` of `session/new`.
|
||||
pub cwd: PathBuf,
|
||||
/// Environment laid over the daemon's own for the child. A `PATH` in
|
||||
/// here is also what the adapter is resolved against, since that is the
|
||||
/// `PATH` the child would see.
|
||||
pub env: HashMap<String, String>,
|
||||
/// Upper bound on the handshake; [`HANDSHAKE_TIMEOUT`] by default.
|
||||
pub handshake_timeout: Duration,
|
||||
}
|
||||
|
||||
impl SpawnConfig {
|
||||
pub fn new(cwd: impl Into<PathBuf>, env: HashMap<String, String>) -> Self {
|
||||
Self {
|
||||
cwd: cwd.into(),
|
||||
env,
|
||||
handshake_timeout: HANDSHAKE_TIMEOUT,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum AcpError {
|
||||
/// The adapter is not on `PATH`. The message names the install command
|
||||
/// so the failed session is actionable rather than silently dead.
|
||||
#[error("{ADAPTER_BIN} not found on PATH; install it with: {INSTALL_COMMAND}")]
|
||||
AdapterMissing,
|
||||
#[error("failed to start {ADAPTER_BIN}: {0}")]
|
||||
Spawn(#[source] std::io::Error),
|
||||
/// The adapter answered the handshake with an error.
|
||||
#[error("ACP handshake failed: {0}")]
|
||||
Handshake(String),
|
||||
/// The adapter went away before the handshake finished. Carries the
|
||||
/// tail of its stderr, which is where a missing Claude login shows up.
|
||||
#[error("{ADAPTER_BIN} exited during the ACP handshake ({status}){stderr}")]
|
||||
ExitedEarly { status: String, stderr: String },
|
||||
#[error("ACP handshake timed out after {0:?}")]
|
||||
HandshakeTimeout(Duration),
|
||||
}
|
||||
|
||||
/// Resolve [`ADAPTER_BIN`] against the `PATH` the child would run with: the
|
||||
/// session env's `PATH` when it sets one, the daemon's own otherwise.
|
||||
pub fn resolve_adapter(env: &HashMap<String, String>) -> Option<PathBuf> {
|
||||
let path: OsString = match env.get("PATH") {
|
||||
Some(p) => OsString::from(p),
|
||||
None => std::env::var_os("PATH")?,
|
||||
};
|
||||
std::env::split_paths(&path)
|
||||
.filter(|dir| !dir.as_os_str().is_empty())
|
||||
.map(|dir| dir.join(ADAPTER_BIN))
|
||||
.find(|candidate| is_executable(candidate))
|
||||
}
|
||||
|
||||
fn is_executable(path: &Path) -> bool {
|
||||
use std::os::unix::fs::PermissionsExt;
|
||||
std::fs::metadata(path)
|
||||
.map(|m| m.is_file() && m.permissions().mode() & 0o111 != 0)
|
||||
.unwrap_or(false)
|
||||
}
|
||||
|
||||
enum Command {
|
||||
Cancel,
|
||||
Close,
|
||||
}
|
||||
|
||||
/// Why the handshake did not produce a session id.
|
||||
enum ReadyError {
|
||||
/// The adapter answered with an error.
|
||||
Handshake(String),
|
||||
/// The transport closed under the request: the child is gone.
|
||||
Closed,
|
||||
}
|
||||
|
||||
fn ready_error(e: agent_client_protocol::Error) -> ReadyError {
|
||||
if agent_client_protocol::is_incoming_transport_closed(&e) {
|
||||
ReadyError::Closed
|
||||
} else {
|
||||
ReadyError::Handshake(e.to_string())
|
||||
}
|
||||
}
|
||||
|
||||
/// A live ACP connection to one adapter child. Dropping it sends `SIGTERM`
|
||||
/// if the child is still running; [`AcpSession::terminate`] is the polite
|
||||
/// version.
|
||||
#[derive(Debug)]
|
||||
pub struct AcpSession {
|
||||
session_id: String,
|
||||
pid: u32,
|
||||
commands: mpsc::UnboundedSender<Command>,
|
||||
updates: Mutex<Option<mpsc::UnboundedReceiver<SessionNotification>>>,
|
||||
exit: watch::Receiver<Option<ExitStatus>>,
|
||||
}
|
||||
|
||||
impl AcpSession {
|
||||
/// The id `session/new` returned. What a respawn passes to
|
||||
/// `session/load` (#135).
|
||||
pub fn session_id(&self) -> &str {
|
||||
&self.session_id
|
||||
}
|
||||
|
||||
/// The adapter's pid.
|
||||
pub fn pid(&self) -> u32 {
|
||||
self.pid
|
||||
}
|
||||
|
||||
/// Whether the child has been reaped.
|
||||
pub fn has_exited(&self) -> bool {
|
||||
self.exit.borrow().is_some()
|
||||
}
|
||||
|
||||
/// Wait for the child to exit.
|
||||
pub async fn wait(&self) -> Option<ExitStatus> {
|
||||
wait_exit(&self.exit).await
|
||||
}
|
||||
|
||||
/// The stream of raw `session/update` notifications, handed out once.
|
||||
/// Notifications that arrived before the take are still queued on it.
|
||||
pub fn take_updates(&self) -> Option<mpsc::UnboundedReceiver<SessionNotification>> {
|
||||
self.updates.lock().expect("updates mutex").take()
|
||||
}
|
||||
|
||||
/// Send `session/cancel`. Fire-and-forget: a closed connection means the
|
||||
/// child is already gone, which is the outcome cancel was after.
|
||||
pub fn cancel(&self) {
|
||||
let _ = self.commands.send(Command::Cancel);
|
||||
}
|
||||
|
||||
/// End the child: `session/cancel`, wait `cancel_grace`, `SIGTERM`, wait
|
||||
/// `term_grace`, `SIGKILL`. Returns once the child has been reaped.
|
||||
pub async fn terminate(&self, cancel_grace: Duration, term_grace: Duration) {
|
||||
if self.has_exited() {
|
||||
return;
|
||||
}
|
||||
self.cancel();
|
||||
if tokio::time::timeout(cancel_grace, self.wait())
|
||||
.await
|
||||
.is_ok()
|
||||
{
|
||||
return;
|
||||
}
|
||||
self.signal(libc::SIGTERM);
|
||||
if tokio::time::timeout(term_grace, self.wait()).await.is_ok() {
|
||||
return;
|
||||
}
|
||||
tracing::warn!(
|
||||
pid = self.pid,
|
||||
"{ADAPTER_BIN} ignored SIGTERM; sending SIGKILL"
|
||||
);
|
||||
self.signal(libc::SIGKILL);
|
||||
let _ = self.commands.send(Command::Close);
|
||||
self.wait().await;
|
||||
}
|
||||
|
||||
fn signal(&self, sig: libc::c_int) {
|
||||
signal_pid(self.pid, sig);
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for AcpSession {
|
||||
fn drop(&mut self) {
|
||||
let _ = self.commands.send(Command::Close);
|
||||
if !self.has_exited() {
|
||||
self.signal(libc::SIGTERM);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Start the adapter for one session and run the handshake.
|
||||
///
|
||||
/// Fails without a running child: on any error the process is already
|
||||
/// signalled, so a failed session never leaves a stray adapter behind.
|
||||
pub async fn spawn(config: SpawnConfig) -> Result<AcpSession, AcpError> {
|
||||
let adapter = resolve_adapter(&config.env).ok_or(AcpError::AdapterMissing)?;
|
||||
|
||||
let mut cmd = tokio::process::Command::new(&adapter);
|
||||
cmd.current_dir(&config.cwd)
|
||||
.envs(&config.env)
|
||||
// Claude Code refuses to start when it thinks it is nested inside
|
||||
// another Claude Code session; a daemon launched from one would
|
||||
// otherwise never get a structured session up.
|
||||
.env_remove("CLAUDECODE")
|
||||
.stdin(Stdio::piped())
|
||||
.stdout(Stdio::piped())
|
||||
.stderr(Stdio::piped())
|
||||
// If the daemon is dropped without a graceful shutdown the runtime
|
||||
// still tears the task holding this `Child` down, and that kills it.
|
||||
.kill_on_drop(true);
|
||||
#[cfg(target_os = "linux")]
|
||||
{
|
||||
// Same process group as the daemon (no setpgid), and SIGTERM when the
|
||||
// daemon dies however it dies: an agent that outlives its daemon is
|
||||
// exactly the orphan #134 rules out.
|
||||
// SAFETY: prctl is async-signal-safe and touches no shared state.
|
||||
unsafe {
|
||||
cmd.pre_exec(|| {
|
||||
libc::prctl(libc::PR_SET_PDEATHSIG, libc::SIGTERM);
|
||||
Ok(())
|
||||
});
|
||||
}
|
||||
}
|
||||
let mut child = cmd.spawn().map_err(AcpError::Spawn)?;
|
||||
let pid = child.id().expect("child pid is known before wait");
|
||||
let stdin = child.stdin.take().expect("piped stdin");
|
||||
let stdout = child.stdout.take().expect("piped stdout");
|
||||
let stderr = child.stderr.take().expect("piped stderr");
|
||||
|
||||
let stderr_tail = Arc::new(Mutex::new(VecDeque::<u8>::new()));
|
||||
tokio::spawn(drain_stderr(stderr, pid, stderr_tail.clone()));
|
||||
|
||||
let (exit_tx, exit_rx) = watch::channel(None);
|
||||
tokio::spawn(async move {
|
||||
let status = child.wait().await.ok();
|
||||
tracing::debug!(pid, ?status, "{ADAPTER_BIN} exited");
|
||||
let _ = exit_tx.send(status);
|
||||
});
|
||||
|
||||
let (updates_tx, updates_rx) = mpsc::unbounded_channel();
|
||||
let (commands_tx, commands_rx) = mpsc::unbounded_channel();
|
||||
let (ready_tx, ready_rx) = oneshot::channel();
|
||||
let transport = ByteStreams::new(stdin.compat_write(), stdout.compat());
|
||||
tokio::spawn(run_connection(
|
||||
transport,
|
||||
config.cwd.clone(),
|
||||
updates_tx,
|
||||
commands_rx,
|
||||
ready_tx,
|
||||
));
|
||||
|
||||
let outcome = tokio::time::timeout(config.handshake_timeout, ready_rx).await;
|
||||
let failure = match outcome {
|
||||
Ok(Ok(Ok(session_id))) => {
|
||||
return Ok(AcpSession {
|
||||
session_id,
|
||||
pid,
|
||||
commands: commands_tx,
|
||||
updates: Mutex::new(Some(updates_rx)),
|
||||
exit: exit_rx,
|
||||
});
|
||||
}
|
||||
Ok(Ok(Err(ReadyError::Handshake(message)))) => AcpError::Handshake(message),
|
||||
Ok(Ok(Err(ReadyError::Closed))) | Ok(Err(_)) => {
|
||||
// The connection ended before the handshake did: the child went
|
||||
// away. Reap it so the status is real, then report.
|
||||
let status = wait_exit(&exit_rx).await;
|
||||
let stderr = {
|
||||
let mut tail = stderr_tail.lock().expect("stderr tail");
|
||||
tail.make_contiguous();
|
||||
String::from_utf8_lossy(tail.as_slices().0)
|
||||
.trim()
|
||||
.to_string()
|
||||
};
|
||||
AcpError::ExitedEarly {
|
||||
status: status.map_or_else(|| "unknown status".into(), |s| s.to_string()),
|
||||
stderr: if stderr.is_empty() {
|
||||
String::new()
|
||||
} else {
|
||||
format!(": {stderr}")
|
||||
},
|
||||
}
|
||||
}
|
||||
Err(_) => AcpError::HandshakeTimeout(config.handshake_timeout),
|
||||
};
|
||||
// No half-started adapter survives a failed spawn.
|
||||
let _ = commands_tx.send(Command::Close);
|
||||
signal_pid(pid, libc::SIGTERM);
|
||||
if tokio::time::timeout(TERM_GRACE, wait_exit(&exit_rx))
|
||||
.await
|
||||
.is_err()
|
||||
{
|
||||
signal_pid(pid, libc::SIGKILL);
|
||||
wait_exit(&exit_rx).await;
|
||||
}
|
||||
Err(failure)
|
||||
}
|
||||
|
||||
async fn wait_exit(exit: &watch::Receiver<Option<ExitStatus>>) -> Option<ExitStatus> {
|
||||
let mut exit = exit.clone();
|
||||
let status = exit.wait_for(|s| s.is_some()).await.ok()?;
|
||||
*status
|
||||
}
|
||||
|
||||
fn signal_pid(pid: u32, sig: libc::c_int) {
|
||||
// ESRCH just means the child is already gone.
|
||||
// SAFETY: plain syscall on a pid we spawned and have not reaped.
|
||||
unsafe {
|
||||
libc::kill(pid as libc::pid_t, sig);
|
||||
}
|
||||
}
|
||||
|
||||
/// The capabilities the daemon declares: none of the filesystem or terminal
|
||||
/// surface, so the adapter falls back to Claude Code's own tools.
|
||||
fn client_capabilities() -> ClientCapabilities {
|
||||
ClientCapabilities::new()
|
||||
.fs(FileSystemCapabilities::new()
|
||||
.read_text_file(false)
|
||||
.write_text_file(false))
|
||||
.terminal(false)
|
||||
}
|
||||
|
||||
async fn run_connection<T>(
|
||||
transport: T,
|
||||
cwd: PathBuf,
|
||||
updates_tx: mpsc::UnboundedSender<SessionNotification>,
|
||||
mut commands_rx: mpsc::UnboundedReceiver<Command>,
|
||||
ready_tx: oneshot::Sender<Result<String, ReadyError>>,
|
||||
) where
|
||||
T: agent_client_protocol::ConnectTo<Client> + 'static,
|
||||
{
|
||||
let result = Client
|
||||
.builder()
|
||||
.name("maestro")
|
||||
.on_receive_notification(
|
||||
async move |notification: SessionNotification, _cx| {
|
||||
// A dropped receiver only means nobody wants updates any more.
|
||||
let _ = updates_tx.send(notification);
|
||||
Ok(())
|
||||
},
|
||||
agent_client_protocol::on_receive_notification!(),
|
||||
)
|
||||
.on_receive_request(
|
||||
async move |request: RequestPermissionRequest, responder, _cx| {
|
||||
// Stub until #139: decline rather than grant something the
|
||||
// operator never saw.
|
||||
tracing::warn!(
|
||||
session = %request.session_id,
|
||||
"session/request_permission declined: not implemented yet (#139)"
|
||||
);
|
||||
responder.respond(RequestPermissionResponse::new(
|
||||
RequestPermissionOutcome::Cancelled,
|
||||
))
|
||||
},
|
||||
agent_client_protocol::on_receive_request!(),
|
||||
)
|
||||
.connect_with(transport, async move |cx| {
|
||||
let init = InitializeRequest::new(ProtocolVersion::V1)
|
||||
.client_capabilities(client_capabilities())
|
||||
.client_info(Implementation::new("maestro", env!("CARGO_PKG_VERSION")));
|
||||
if let Err(e) = cx.send_request(init).block_task().await {
|
||||
let _ = ready_tx.send(Err(ready_error(e)));
|
||||
return Ok(());
|
||||
}
|
||||
let created = match cx
|
||||
.send_request(NewSessionRequest::new(cwd))
|
||||
.block_task()
|
||||
.await
|
||||
{
|
||||
Ok(r) => r,
|
||||
Err(e) => {
|
||||
let _ = ready_tx.send(Err(ready_error(e)));
|
||||
return Ok(());
|
||||
}
|
||||
};
|
||||
let session_id = created.session_id.0.to_string();
|
||||
let _ = ready_tx.send(Ok(session_id.clone()));
|
||||
|
||||
loop {
|
||||
tokio::select! {
|
||||
cmd = commands_rx.recv() => match cmd {
|
||||
Some(Command::Cancel) => {
|
||||
let _ = cx.send_notification(CancelNotification::new(
|
||||
SessionId::new(session_id.as_str()),
|
||||
));
|
||||
}
|
||||
Some(Command::Close) | None => break,
|
||||
},
|
||||
_ = cx.incoming_closed() => break,
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
})
|
||||
.await;
|
||||
if let Err(e) = result {
|
||||
tracing::debug!(error = %e, "ACP connection ended with an error");
|
||||
}
|
||||
}
|
||||
|
||||
async fn drain_stderr(
|
||||
stderr: tokio::process::ChildStderr,
|
||||
pid: u32,
|
||||
tail: Arc<Mutex<VecDeque<u8>>>,
|
||||
) {
|
||||
let mut lines = tokio::io::BufReader::new(stderr).lines();
|
||||
while let Ok(Some(line)) = lines.next_line().await {
|
||||
tracing::debug!(pid, "{ADAPTER_BIN}: {line}");
|
||||
let mut tail = tail.lock().expect("stderr tail");
|
||||
tail.extend(line.as_bytes());
|
||||
tail.push_back(b'\n');
|
||||
while tail.len() > STDERR_TAIL_BYTES {
|
||||
tail.pop_front();
|
||||
}
|
||||
tail.make_contiguous();
|
||||
}
|
||||
}
|
||||
Executable
+98
@@ -0,0 +1,98 @@
|
||||
#!/usr/bin/env python3
|
||||
"""A deterministic stand-in for `claude-code-acp` (#134).
|
||||
|
||||
Speaks just enough JSON-RPC over stdio for the daemon's handshake:
|
||||
answers `initialize` and `session/new`, logs every method it receives, and
|
||||
exits on SIGTERM. Tests point `PATH` at a directory holding this file under
|
||||
the adapter's name, so the real adapter (which needs Claude auth and is
|
||||
slow) never runs in the deterministic suite.
|
||||
|
||||
Environment knobs, all optional:
|
||||
|
||||
FAKE_ACP_LOG append one line per event to this file:
|
||||
start pid=<pid> cwd=<cwd>
|
||||
env <NAME>=<value> (for FAKE_ACP_ECHO_ENV)
|
||||
caps <json of clientCapabilities>
|
||||
recv <method>
|
||||
signal SIGTERM
|
||||
FAKE_ACP_ECHO_ENV name of one env var to log at start
|
||||
FAKE_ACP_INIT_FAIL=1 answer `initialize` with a JSON-RPC error
|
||||
FAKE_ACP_EXIT_AFTER_NEW=1 exit 0 right after answering `session/new`
|
||||
FAKE_ACP_EXIT_ON_CANCEL=1 exit 0 on `session/cancel` (default: keep running
|
||||
so the SIGTERM path is exercised)
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import signal
|
||||
import sys
|
||||
|
||||
LOG = os.environ.get("FAKE_ACP_LOG")
|
||||
|
||||
|
||||
def log(line):
|
||||
if not LOG:
|
||||
return
|
||||
with open(LOG, "a") as f:
|
||||
f.write(line + "\n")
|
||||
f.flush()
|
||||
|
||||
|
||||
def send(obj):
|
||||
sys.stdout.write(json.dumps(obj) + "\n")
|
||||
sys.stdout.flush()
|
||||
|
||||
|
||||
def on_sigterm(_signum, _frame):
|
||||
log("signal SIGTERM")
|
||||
sys.exit(0)
|
||||
|
||||
|
||||
signal.signal(signal.SIGTERM, on_sigterm)
|
||||
|
||||
log(f"start pid={os.getpid()} cwd={os.getcwd()}")
|
||||
echo = os.environ.get("FAKE_ACP_ECHO_ENV")
|
||||
if echo:
|
||||
log(f"env {echo}={os.environ.get(echo, '')}")
|
||||
|
||||
session_id = f"fake-{os.getpid()}"
|
||||
|
||||
for raw in sys.stdin:
|
||||
raw = raw.strip()
|
||||
if not raw:
|
||||
continue
|
||||
msg = json.loads(raw)
|
||||
method = msg.get("method")
|
||||
if method is None:
|
||||
# A response to something we sent; nothing to do.
|
||||
continue
|
||||
log(f"recv {method}")
|
||||
rid = msg.get("id")
|
||||
params = msg.get("params") or {}
|
||||
|
||||
if method == "initialize":
|
||||
log("caps " + json.dumps(params.get("clientCapabilities", {}), sort_keys=True))
|
||||
if os.environ.get("FAKE_ACP_INIT_FAIL") == "1":
|
||||
send({"jsonrpc": "2.0", "id": rid,
|
||||
"error": {"code": -32000, "message": "fake init failure"}})
|
||||
continue
|
||||
send({"jsonrpc": "2.0", "id": rid, "result": {
|
||||
"protocolVersion": params.get("protocolVersion", 1),
|
||||
"agentCapabilities": {"loadSession": True},
|
||||
"agentInfo": {"name": "fake-acp-agent", "version": "0.0.0"},
|
||||
"authMethods": [],
|
||||
}})
|
||||
elif method == "session/new":
|
||||
send({"jsonrpc": "2.0", "id": rid, "result": {"sessionId": session_id}})
|
||||
send({"jsonrpc": "2.0", "method": "session/update", "params": {
|
||||
"sessionId": session_id,
|
||||
"update": {"sessionUpdate": "agent_message_chunk",
|
||||
"content": {"type": "text", "text": "hello from fake"}},
|
||||
}})
|
||||
if os.environ.get("FAKE_ACP_EXIT_AFTER_NEW") == "1":
|
||||
sys.exit(0)
|
||||
elif method == "session/cancel":
|
||||
if os.environ.get("FAKE_ACP_EXIT_ON_CANCEL") == "1":
|
||||
sys.exit(0)
|
||||
elif rid is not None:
|
||||
send({"jsonrpc": "2.0", "id": rid,
|
||||
"error": {"code": -32601, "message": f"method not found: {method}"}})
|
||||
@@ -0,0 +1,292 @@
|
||||
//! `maestro-acp` against the fake adapter in `tests/fake_acp_agent.py`, plus
|
||||
//! one ignored test against the real `claude-code-acp`.
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use maestro_acp::schema::{ContentBlock, SessionUpdate};
|
||||
use maestro_acp::{AcpError, SpawnConfig, ADAPTER_BIN, INSTALL_COMMAND};
|
||||
use tempfile::TempDir;
|
||||
|
||||
fn fake_agent() -> PathBuf {
|
||||
Path::new(env!("CARGO_MANIFEST_DIR")).join("tests/fake_acp_agent.py")
|
||||
}
|
||||
|
||||
/// A temp `bin/` holding the fake agent under the adapter's name, plus a
|
||||
/// log file the agent appends to. `PATH` for the spawn is just that dir.
|
||||
struct Fake {
|
||||
dir: TempDir,
|
||||
}
|
||||
|
||||
impl Fake {
|
||||
fn new() -> Self {
|
||||
let dir = TempDir::new().unwrap();
|
||||
let bin = dir.path().join("bin");
|
||||
std::fs::create_dir_all(&bin).unwrap();
|
||||
std::fs::copy(fake_agent(), bin.join(ADAPTER_BIN)).unwrap();
|
||||
Self { dir }
|
||||
}
|
||||
|
||||
fn log_path(&self) -> PathBuf {
|
||||
self.dir.path().join("agent.log")
|
||||
}
|
||||
|
||||
fn log(&self) -> String {
|
||||
std::fs::read_to_string(self.log_path()).unwrap_or_default()
|
||||
}
|
||||
|
||||
/// The fake's `bin/` first, so it shadows any real adapter, then the
|
||||
/// process `PATH` so the script's `python3` still resolves.
|
||||
fn env(&self) -> HashMap<String, String> {
|
||||
let path = format!(
|
||||
"{}:{}",
|
||||
self.dir.path().join("bin").display(),
|
||||
std::env::var("PATH").unwrap_or_default()
|
||||
);
|
||||
HashMap::from([
|
||||
("PATH".to_string(), path),
|
||||
(
|
||||
"FAKE_ACP_LOG".to_string(),
|
||||
self.log_path().to_string_lossy().into_owned(),
|
||||
),
|
||||
])
|
||||
}
|
||||
|
||||
fn config(&self, cwd: &Path) -> SpawnConfig {
|
||||
let mut config = SpawnConfig::new(cwd, self.env());
|
||||
config.handshake_timeout = Duration::from_secs(10);
|
||||
config
|
||||
}
|
||||
|
||||
async fn wait_for_log(&self, needle: &str) {
|
||||
let deadline = Instant::now() + Duration::from_secs(5);
|
||||
while Instant::now() < deadline {
|
||||
if self.log().contains(needle) {
|
||||
return;
|
||||
}
|
||||
tokio::time::sleep(Duration::from_millis(20)).await;
|
||||
}
|
||||
panic!("log never contained {needle:?}:\n{}", self.log());
|
||||
}
|
||||
}
|
||||
|
||||
fn pid_alive(pid: u32) -> bool {
|
||||
Path::new(&format!("/proc/{pid}")).exists()
|
||||
&& std::fs::read_to_string(format!("/proc/{pid}/stat"))
|
||||
.map(|s| !s.contains(") Z "))
|
||||
.unwrap_or(false)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn spawn_runs_adapter_in_cwd_with_env_and_restrictive_caps() {
|
||||
let fake = Fake::new();
|
||||
let cwd = TempDir::new().unwrap();
|
||||
let mut config = fake.config(cwd.path());
|
||||
config
|
||||
.env
|
||||
.insert("FAKE_ACP_ECHO_ENV".into(), "MAESTRO_SESSION_ID".into());
|
||||
config
|
||||
.env
|
||||
.insert("MAESTRO_SESSION_ID".into(), "sess-123".into());
|
||||
|
||||
let session = maestro_acp::spawn(config).await.expect("spawn");
|
||||
assert!(session.session_id().starts_with("fake-"));
|
||||
assert!(pid_alive(session.pid()));
|
||||
|
||||
let log = fake.log();
|
||||
let cwd_canon = cwd.path().canonicalize().unwrap();
|
||||
assert!(
|
||||
log.contains(&format!("cwd={}", cwd_canon.display())),
|
||||
"child cwd missing from log:\n{log}"
|
||||
);
|
||||
assert!(log.contains("env MAESTRO_SESSION_ID=sess-123"), "{log}");
|
||||
assert!(log.contains("recv initialize"), "{log}");
|
||||
assert!(log.contains("recv session/new"), "{log}");
|
||||
let caps = log
|
||||
.lines()
|
||||
.find_map(|l| l.strip_prefix("caps "))
|
||||
.expect("caps line");
|
||||
let caps: serde_json::Value = serde_json::from_str(caps).unwrap();
|
||||
assert_eq!(caps["fs"]["readTextFile"], false);
|
||||
assert_eq!(caps["fs"]["writeTextFile"], false);
|
||||
assert_eq!(caps["terminal"], false);
|
||||
|
||||
session
|
||||
.terminate(Duration::from_millis(100), Duration::from_secs(2))
|
||||
.await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn missing_adapter_names_the_install_command() {
|
||||
let empty = TempDir::new().unwrap();
|
||||
let cwd = TempDir::new().unwrap();
|
||||
let env = HashMap::from([(
|
||||
"PATH".to_string(),
|
||||
empty.path().to_string_lossy().into_owned(),
|
||||
)]);
|
||||
let err = maestro_acp::spawn(SpawnConfig::new(cwd.path(), env))
|
||||
.await
|
||||
.expect_err("no adapter on PATH");
|
||||
assert!(matches!(err, AcpError::AdapterMissing));
|
||||
let msg = err.to_string();
|
||||
assert!(msg.contains(ADAPTER_BIN), "{msg}");
|
||||
assert!(msg.contains(INSTALL_COMMAND), "{msg}");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn session_updates_reach_the_taken_receiver() {
|
||||
let fake = Fake::new();
|
||||
let cwd = TempDir::new().unwrap();
|
||||
let session = maestro_acp::spawn(fake.config(cwd.path()))
|
||||
.await
|
||||
.expect("spawn");
|
||||
|
||||
let mut updates = session.take_updates().expect("first take");
|
||||
assert!(session.take_updates().is_none(), "handed out once");
|
||||
let first = tokio::time::timeout(Duration::from_secs(5), updates.recv())
|
||||
.await
|
||||
.expect("update within 5s")
|
||||
.expect("channel open");
|
||||
assert_eq!(first.session_id.0.as_ref(), session.session_id());
|
||||
match first.update {
|
||||
SessionUpdate::AgentMessageChunk(chunk) => match chunk.content {
|
||||
ContentBlock::Text(t) => assert_eq!(t.text, "hello from fake"),
|
||||
other => panic!("unexpected content {other:?}"),
|
||||
},
|
||||
other => panic!("unexpected update {other:?}"),
|
||||
}
|
||||
|
||||
session
|
||||
.terminate(Duration::from_millis(100), Duration::from_secs(2))
|
||||
.await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn terminate_sends_cancel_then_sigterm() {
|
||||
let fake = Fake::new();
|
||||
let cwd = TempDir::new().unwrap();
|
||||
let session = maestro_acp::spawn(fake.config(cwd.path()))
|
||||
.await
|
||||
.expect("spawn");
|
||||
let pid = session.pid();
|
||||
|
||||
session
|
||||
.terminate(Duration::from_millis(200), Duration::from_secs(2))
|
||||
.await;
|
||||
assert!(session.has_exited());
|
||||
fake.wait_for_log("signal SIGTERM").await;
|
||||
|
||||
let log = fake.log();
|
||||
let cancel = log.find("recv session/cancel").expect("cancel logged");
|
||||
let term = log.find("signal SIGTERM").expect("SIGTERM logged");
|
||||
assert!(cancel < term, "cancel must precede SIGTERM:\n{log}");
|
||||
assert!(!pid_alive(pid), "child {pid} still alive");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn terminate_is_satisfied_by_an_exit_on_cancel() {
|
||||
let fake = Fake::new();
|
||||
let cwd = TempDir::new().unwrap();
|
||||
let mut config = fake.config(cwd.path());
|
||||
config
|
||||
.env
|
||||
.insert("FAKE_ACP_EXIT_ON_CANCEL".into(), "1".into());
|
||||
let session = maestro_acp::spawn(config).await.expect("spawn");
|
||||
|
||||
session
|
||||
.terminate(Duration::from_secs(2), Duration::from_secs(2))
|
||||
.await;
|
||||
assert!(session.has_exited());
|
||||
let log = fake.log();
|
||||
assert!(log.contains("recv session/cancel"), "{log}");
|
||||
assert!(!log.contains("signal SIGTERM"), "cancel was enough:\n{log}");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn handshake_error_fails_the_spawn_and_reaps_the_child() {
|
||||
let fake = Fake::new();
|
||||
let cwd = TempDir::new().unwrap();
|
||||
let mut config = fake.config(cwd.path());
|
||||
config.env.insert("FAKE_ACP_INIT_FAIL".into(), "1".into());
|
||||
|
||||
let err = maestro_acp::spawn(config).await.expect_err("init fails");
|
||||
assert!(matches!(err, AcpError::Handshake(_)), "{err}");
|
||||
assert!(err.to_string().contains("fake init failure"), "{err}");
|
||||
// Closing the connection gives the fake EOF on stdin; the SIGTERM that
|
||||
// follows may find it already gone. Either way it must not survive.
|
||||
let pid: u32 = fake
|
||||
.log()
|
||||
.lines()
|
||||
.find_map(|l| l.strip_prefix("start pid="))
|
||||
.and_then(|rest| rest.split(' ').next())
|
||||
.and_then(|p| p.parse().ok())
|
||||
.expect("pid in log");
|
||||
let deadline = Instant::now() + Duration::from_secs(5);
|
||||
while pid_alive(pid) && Instant::now() < deadline {
|
||||
tokio::time::sleep(Duration::from_millis(20)).await;
|
||||
}
|
||||
assert!(!pid_alive(pid), "child {pid} outlived the failed spawn");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn early_exit_is_observed_after_a_successful_handshake() {
|
||||
let fake = Fake::new();
|
||||
let cwd = TempDir::new().unwrap();
|
||||
let mut config = fake.config(cwd.path());
|
||||
config
|
||||
.env
|
||||
.insert("FAKE_ACP_EXIT_AFTER_NEW".into(), "1".into());
|
||||
let session = maestro_acp::spawn(config).await.expect("spawn");
|
||||
|
||||
let status = tokio::time::timeout(Duration::from_secs(5), session.wait())
|
||||
.await
|
||||
.expect("exit within 5s")
|
||||
.expect("status");
|
||||
assert!(status.success(), "{status}");
|
||||
assert!(session.has_exited());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn child_exit_before_handshake_reports_exited_early() {
|
||||
// A "adapter" that dies before answering anything.
|
||||
let dir = TempDir::new().unwrap();
|
||||
let bin = dir.path().join("bin");
|
||||
std::fs::create_dir_all(&bin).unwrap();
|
||||
let script = bin.join(ADAPTER_BIN);
|
||||
std::fs::write(&script, "#!/bin/sh\necho 'not logged in' >&2\nexit 3\n").unwrap();
|
||||
{
|
||||
use std::os::unix::fs::PermissionsExt;
|
||||
std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).unwrap();
|
||||
}
|
||||
let cwd = TempDir::new().unwrap();
|
||||
let env = HashMap::from([("PATH".to_string(), bin.to_string_lossy().into_owned())]);
|
||||
let mut config = SpawnConfig::new(cwd.path(), env);
|
||||
config.handshake_timeout = Duration::from_secs(10);
|
||||
|
||||
let err = maestro_acp::spawn(config).await.expect_err("child died");
|
||||
match &err {
|
||||
AcpError::ExitedEarly { status, stderr } => {
|
||||
assert!(status.contains('3'), "{status}");
|
||||
assert!(stderr.contains("not logged in"), "{stderr}");
|
||||
}
|
||||
other => panic!("unexpected {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
/// End to end against the installed `claude-code-acp`. Needs Claude auth,
|
||||
/// takes seconds, so it is opt-in: `cargo test -p maestro-acp -- --ignored`.
|
||||
#[tokio::test]
|
||||
#[ignore]
|
||||
async fn real_adapter_handshake() {
|
||||
let cwd = TempDir::new().unwrap();
|
||||
let session = maestro_acp::spawn(SpawnConfig::new(cwd.path(), HashMap::new()))
|
||||
.await
|
||||
.expect("real claude-code-acp handshake");
|
||||
assert!(!session.session_id().is_empty());
|
||||
assert!(pid_alive(session.pid()));
|
||||
session
|
||||
.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
|
||||
.await;
|
||||
assert!(session.has_exited());
|
||||
}
|
||||
Reference in New Issue
Block a user