1089 lines
43 KiB
Rust
1089 lines
43 KiB
Rust
//! ACP client role for the daemon: one provider adapter 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 the provider's own file and command tools 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 with
|
|
//! a real implementation.
|
|
//!
|
|
//! 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.
|
|
//!
|
|
//! A respawn after a daemon restart sets [`SpawnConfig::load_session`] and
|
|
//! the handshake runs `session/load` with the stored id instead of
|
|
//! `session/new`. The history the agent streams back while that request is
|
|
//! in flight is *not* delivered: the daemon's event log already holds it,
|
|
//! and replaying it would double every line.
|
|
//!
|
|
//! 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::atomic::{AtomicBool, Ordering};
|
|
use std::sync::{Arc, Mutex};
|
|
use std::time::Duration;
|
|
|
|
use agent_client_protocol::schema::v1::{
|
|
CancelNotification, ClientCapabilities, ConfigOptionUpdate, ContentBlock, CurrentModeUpdate,
|
|
FileSystemCapabilities, Implementation, InitializeRequest, PromptRequest,
|
|
RequestPermissionOutcome, RequestPermissionRequest, RequestPermissionResponse,
|
|
SessionConfigKind, SessionConfigOption, SessionConfigOptionCategory, SessionConfigOptionValue,
|
|
SessionConfigSelectOption, SessionId, SessionModeState, SessionNotification, SessionUpdate,
|
|
SetSessionConfigOptionRequest, SetSessionModeRequest,
|
|
};
|
|
use agent_client_protocol::schema::ProtocolVersion;
|
|
use agent_client_protocol::{ByteStreams, Client, JsonRpcRequest, JsonRpcResponse};
|
|
use serde::{Deserialize, Serialize};
|
|
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;
|
|
|
|
/// ACP adapter selected from the structured session's resolved command.
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
|
pub enum AcpAdapter {
|
|
ClaudeCode,
|
|
Codex,
|
|
}
|
|
|
|
impl AcpAdapter {
|
|
pub const fn binary(self) -> &'static str {
|
|
match self {
|
|
Self::ClaudeCode => "claude-code-acp",
|
|
Self::Codex => "codex-acp",
|
|
}
|
|
}
|
|
|
|
pub const fn install_command(self) -> &'static str {
|
|
match self {
|
|
Self::ClaudeCode => "npm install -g --prefix ~/.local @zed-industries/claude-code-acp",
|
|
Self::Codex => "npm install -g --prefix ~/.local @agentclientprotocol/codex-acp",
|
|
}
|
|
}
|
|
}
|
|
|
|
impl std::fmt::Display for AcpAdapter {
|
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
f.write_str(match self {
|
|
Self::ClaudeCode => "Claude Code",
|
|
Self::Codex => "Codex",
|
|
})
|
|
}
|
|
}
|
|
|
|
/// Compatibility aliases for Claude Code callers.
|
|
pub const ADAPTER_BIN: &str = "claude-code-acp";
|
|
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 adapters boot a node process and their provider
|
|
/// client 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;
|
|
const LEGACY_MODEL_CONFIG_ID: &str = "__maestro_legacy_model";
|
|
const LEGACY_MODE_CONFIG_ID: &str = "__maestro_legacy_mode";
|
|
|
|
#[derive(Debug, Clone, Serialize, Deserialize, JsonRpcRequest)]
|
|
#[request(method = "session/new", response = ExtendedNewSessionResponse)]
|
|
#[serde(rename_all = "camelCase")]
|
|
struct ExtendedNewSessionRequest {
|
|
cwd: PathBuf,
|
|
mcp_servers: Vec<serde_json::Value>,
|
|
}
|
|
|
|
#[derive(Debug, Clone, Serialize, Deserialize, JsonRpcResponse)]
|
|
#[serde(rename_all = "camelCase")]
|
|
struct ExtendedNewSessionResponse {
|
|
session_id: SessionId,
|
|
#[serde(default)]
|
|
modes: Option<SessionModeState>,
|
|
#[serde(default)]
|
|
config_options: Option<Vec<SessionConfigOption>>,
|
|
#[serde(default)]
|
|
models: Option<LegacyModelState>,
|
|
}
|
|
|
|
#[derive(Debug, Clone, Serialize, Deserialize, JsonRpcRequest)]
|
|
#[request(method = "session/load", response = ExtendedLoadSessionResponse)]
|
|
#[serde(rename_all = "camelCase")]
|
|
struct ExtendedLoadSessionRequest {
|
|
session_id: SessionId,
|
|
cwd: PathBuf,
|
|
mcp_servers: Vec<serde_json::Value>,
|
|
}
|
|
|
|
#[derive(Debug, Clone, Serialize, Deserialize, JsonRpcResponse)]
|
|
#[serde(rename_all = "camelCase")]
|
|
struct ExtendedLoadSessionResponse {
|
|
#[serde(default)]
|
|
modes: Option<SessionModeState>,
|
|
#[serde(default)]
|
|
config_options: Option<Vec<SessionConfigOption>>,
|
|
#[serde(default)]
|
|
models: Option<LegacyModelState>,
|
|
}
|
|
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
#[serde(rename_all = "camelCase")]
|
|
struct LegacyModelState {
|
|
current_model_id: String,
|
|
available_models: Vec<LegacyModel>,
|
|
}
|
|
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
#[serde(rename_all = "camelCase")]
|
|
struct LegacyModel {
|
|
model_id: String,
|
|
name: String,
|
|
#[serde(default)]
|
|
description: Option<String>,
|
|
}
|
|
|
|
#[derive(Debug, Clone, Serialize, Deserialize, JsonRpcRequest)]
|
|
#[request(method = "session/set_model", response = LegacySetModelResponse)]
|
|
#[serde(rename_all = "camelCase")]
|
|
struct LegacySetModelRequest {
|
|
session_id: SessionId,
|
|
model_id: String,
|
|
}
|
|
|
|
#[derive(Debug, Clone, Serialize, Deserialize, JsonRpcResponse)]
|
|
struct LegacySetModelResponse {}
|
|
|
|
/// 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,
|
|
/// An existing ACP session id to resume with `session/load` (#135)
|
|
/// instead of opening a fresh one. The adapter must advertise
|
|
/// `loadSession`; the handshake fails otherwise.
|
|
pub load_session: Option<String>,
|
|
/// The provider-specific adapter to launch.
|
|
pub adapter: AcpAdapter,
|
|
}
|
|
|
|
impl SpawnConfig {
|
|
pub fn new(cwd: impl Into<PathBuf>, env: HashMap<String, String>) -> Self {
|
|
Self {
|
|
cwd: cwd.into(),
|
|
env,
|
|
handshake_timeout: HANDSHAKE_TIMEOUT,
|
|
load_session: None,
|
|
adapter: AcpAdapter::ClaudeCode,
|
|
}
|
|
}
|
|
|
|
/// Resume `session_id` with `session/load` instead of `session/new`.
|
|
pub fn load_session(mut self, session_id: impl Into<String>) -> Self {
|
|
self.load_session = Some(session_id.into());
|
|
self
|
|
}
|
|
|
|
pub fn adapter(mut self, adapter: AcpAdapter) -> Self {
|
|
self.adapter = adapter;
|
|
self
|
|
}
|
|
}
|
|
|
|
/// A blocked ACP permission request delivered to the daemon.
|
|
#[derive(Debug)]
|
|
pub struct PermissionRequest {
|
|
/// The prompt turn that was active when ACP delivered this callback.
|
|
pub turn: u64,
|
|
pub request: RequestPermissionRequest,
|
|
pub response: oneshot::Sender<RequestPermissionOutcome>,
|
|
}
|
|
|
|
#[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} ACP adapter ({bin}) not found on PATH; install it with: {install_command}", bin = adapter.binary(), install_command = adapter.install_command())]
|
|
AdapterMissing { adapter: AcpAdapter },
|
|
#[error("failed to start {adapter} ACP adapter ({bin}): {source}", bin = adapter.binary())]
|
|
Spawn {
|
|
adapter: AcpAdapter,
|
|
#[source]
|
|
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} ACP adapter ({bin}) exited during the ACP handshake ({status}){stderr}", bin = adapter.binary())]
|
|
ExitedEarly {
|
|
adapter: AcpAdapter,
|
|
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(adapter: AcpAdapter, 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.binary()))
|
|
.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,
|
|
Prompt {
|
|
text: String,
|
|
turn: u64,
|
|
done: oneshot::Sender<Result<(), String>>,
|
|
},
|
|
SetConfig {
|
|
config_id: String,
|
|
value: SessionConfigOptionValue,
|
|
done: oneshot::Sender<Result<(), String>>,
|
|
},
|
|
Close,
|
|
}
|
|
|
|
struct ConnectionConfig {
|
|
cwd: PathBuf,
|
|
load_session: Option<String>,
|
|
adapter: AcpAdapter,
|
|
}
|
|
|
|
enum ChildCommand {
|
|
Signal(libc::c_int),
|
|
Reap(oneshot::Sender<bool>),
|
|
}
|
|
|
|
#[derive(Debug, Default)]
|
|
struct PermissionTurnState {
|
|
active: Option<u64>,
|
|
}
|
|
|
|
impl PermissionTurnState {
|
|
fn begin(&mut self, turn: u64) {
|
|
self.active = Some(turn);
|
|
}
|
|
|
|
fn cancel_active(&mut self) {
|
|
self.active = None;
|
|
}
|
|
|
|
fn finish(&mut self, turn: u64) {
|
|
if self.active == Some(turn) {
|
|
self.active = None;
|
|
}
|
|
}
|
|
|
|
fn callback_turn(&self) -> Option<u64> {
|
|
self.active
|
|
}
|
|
}
|
|
|
|
/// 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>,
|
|
child_commands: mpsc::UnboundedSender<ChildCommand>,
|
|
updates: Mutex<Option<mpsc::UnboundedReceiver<SessionNotification>>>,
|
|
permission_requests: Mutex<Option<mpsc::UnboundedReceiver<PermissionRequest>>>,
|
|
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()
|
|
}
|
|
|
|
/// Reap the child if it has already exited.
|
|
pub async fn reap_if_exited(&self) -> bool {
|
|
if self.has_exited() {
|
|
return true;
|
|
}
|
|
let (sent, received) = oneshot::channel();
|
|
if self.child_commands.send(ChildCommand::Reap(sent)).is_err() {
|
|
return self.has_exited();
|
|
}
|
|
received.await.unwrap_or_else(|_| self.has_exited())
|
|
}
|
|
|
|
/// 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()
|
|
}
|
|
|
|
pub fn take_permission_requests(&self) -> Option<mpsc::UnboundedReceiver<PermissionRequest>> {
|
|
self.permission_requests
|
|
.lock()
|
|
.expect("permission requests")
|
|
.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);
|
|
}
|
|
|
|
/// Send a user prompt and receive completion once the agent ends the turn.
|
|
pub fn prompt(
|
|
&self,
|
|
text: String,
|
|
turn: u64,
|
|
) -> Result<oneshot::Receiver<Result<(), String>>, String> {
|
|
let (done_tx, done_rx) = oneshot::channel();
|
|
self.commands
|
|
.send(Command::Prompt {
|
|
text,
|
|
turn,
|
|
done: done_tx,
|
|
})
|
|
.map_err(|_| "ACP connection is closed".to_string())?;
|
|
Ok(done_rx)
|
|
}
|
|
|
|
/// Change one option advertised by the ACP agent.
|
|
pub fn set_config(
|
|
&self,
|
|
config_id: String,
|
|
value: SessionConfigOptionValue,
|
|
) -> Result<oneshot::Receiver<Result<(), String>>, String> {
|
|
let (done_tx, done_rx) = oneshot::channel();
|
|
self.commands
|
|
.send(Command::SetConfig {
|
|
config_id,
|
|
value,
|
|
done: done_tx,
|
|
})
|
|
.map_err(|_| "ACP connection is closed".to_string())?;
|
|
Ok(done_rx)
|
|
}
|
|
|
|
/// 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) {
|
|
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) {
|
|
let _ = self.child_commands.send(ChildCommand::Signal(sig));
|
|
}
|
|
}
|
|
|
|
impl Drop for AcpSession {
|
|
fn drop(&mut self) {
|
|
self.signal(libc::SIGTERM);
|
|
let _ = self.commands.send(Command::Close);
|
|
}
|
|
}
|
|
|
|
/// 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 provider = config.adapter;
|
|
let adapter = resolve_adapter(provider, &config.env)
|
|
.ok_or(AcpError::AdapterMissing { adapter: provider })?;
|
|
|
|
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(|source| AcpError::Spawn {
|
|
adapter: provider,
|
|
source,
|
|
})?;
|
|
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(), provider));
|
|
|
|
let (exit_tx, exit_rx) = watch::channel(None);
|
|
let (child_commands_tx, child_commands_rx) = mpsc::unbounded_channel();
|
|
tokio::spawn(supervise_child(
|
|
child,
|
|
pid,
|
|
child_commands_rx,
|
|
exit_tx,
|
|
provider,
|
|
));
|
|
|
|
let (updates_tx, updates_rx) = mpsc::unbounded_channel();
|
|
let (permission_tx, permission_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,
|
|
ConnectionConfig {
|
|
cwd: config.cwd.clone(),
|
|
load_session: config.load_session.clone(),
|
|
adapter: provider,
|
|
},
|
|
updates_tx,
|
|
permission_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,
|
|
child_commands: child_commands_tx,
|
|
updates: Mutex::new(Some(updates_rx)),
|
|
permission_requests: Mutex::new(Some(permission_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 {
|
|
adapter: provider,
|
|
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);
|
|
let _ = child_commands_tx.send(ChildCommand::Signal(libc::SIGTERM));
|
|
if tokio::time::timeout(TERM_GRACE, wait_exit(&exit_rx))
|
|
.await
|
|
.is_err()
|
|
{
|
|
let _ = child_commands_tx.send(ChildCommand::Signal(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
|
|
}
|
|
|
|
async fn supervise_child(
|
|
mut child: tokio::process::Child,
|
|
pid: u32,
|
|
mut commands: mpsc::UnboundedReceiver<ChildCommand>,
|
|
exit_tx: watch::Sender<Option<ExitStatus>>,
|
|
adapter: AcpAdapter,
|
|
) {
|
|
loop {
|
|
tokio::select! {
|
|
status = child.wait() => {
|
|
let status = status.ok();
|
|
tracing::debug!(pid, ?status, provider = %adapter, "ACP adapter exited");
|
|
let _ = exit_tx.send(status);
|
|
return;
|
|
}
|
|
command = commands.recv() => match command {
|
|
Some(ChildCommand::Signal(sig)) => {
|
|
// This task alone owns `child`. It cannot reap the child
|
|
// between receiving this command and sending the signal,
|
|
// so its PID cannot be reused in that window.
|
|
unsafe { libc::kill(pid as libc::pid_t, sig) };
|
|
}
|
|
Some(ChildCommand::Reap(done)) => match child.try_wait() {
|
|
Ok(Some(status)) => {
|
|
let _ = done.send(true);
|
|
tracing::debug!(pid, ?status, provider = %adapter, "ACP adapter exited");
|
|
let _ = exit_tx.send(Some(status));
|
|
return;
|
|
}
|
|
Ok(None) => {
|
|
let _ = done.send(false);
|
|
}
|
|
Err(error) => {
|
|
tracing::warn!(pid, %error, provider = %adapter, "could not reap ACP adapter");
|
|
let _ = done.send(false);
|
|
}
|
|
},
|
|
None => {
|
|
// The session vanished without a command receiver. This
|
|
// task still owns the unreaped child, so this cannot hit
|
|
// a reused PID.
|
|
unsafe { libc::kill(pid as libc::pid_t, libc::SIGTERM) };
|
|
let status = child.wait().await.ok();
|
|
tracing::debug!(pid, ?status, provider = %adapter, "ACP adapter exited");
|
|
let _ = exit_tx.send(status);
|
|
return;
|
|
}
|
|
},
|
|
}
|
|
}
|
|
}
|
|
|
|
/// 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)
|
|
}
|
|
|
|
fn advertised_config_options(
|
|
mut options: Vec<SessionConfigOption>,
|
|
modes: Option<&SessionModeState>,
|
|
models: Option<&LegacyModelState>,
|
|
) -> Vec<SessionConfigOption> {
|
|
let has_model = options
|
|
.iter()
|
|
.any(|option| matches!(option.category, Some(SessionConfigOptionCategory::Model)));
|
|
let has_mode = options
|
|
.iter()
|
|
.any(|option| matches!(option.category, Some(SessionConfigOptionCategory::Mode)));
|
|
if let Some(models) = models.filter(|_| !has_model) {
|
|
let mut choices = models
|
|
.available_models
|
|
.iter()
|
|
.map(|model| {
|
|
let mut option =
|
|
SessionConfigSelectOption::new(model.model_id.clone(), &model.name);
|
|
if model.model_id == "default" {
|
|
option.name = "Opus (default)".into();
|
|
}
|
|
option.description.clone_from(&model.description);
|
|
option
|
|
})
|
|
.collect::<Vec<_>>();
|
|
if !models
|
|
.available_models
|
|
.iter()
|
|
.any(|model| model.model_id == "fable")
|
|
{
|
|
choices.push(SessionConfigSelectOption::new("fable", "Fable"));
|
|
}
|
|
options.push(
|
|
SessionConfigOption::select(
|
|
LEGACY_MODEL_CONFIG_ID,
|
|
"Model",
|
|
models.current_model_id.clone(),
|
|
choices,
|
|
)
|
|
.category(SessionConfigOptionCategory::Model),
|
|
);
|
|
}
|
|
if let Some(modes) = modes.filter(|_| !has_mode) {
|
|
let choices = modes
|
|
.available_modes
|
|
.iter()
|
|
.map(|mode| {
|
|
let mut option = SessionConfigSelectOption::new(mode.id.0.clone(), &mode.name);
|
|
option.description.clone_from(&mode.description);
|
|
option
|
|
})
|
|
.collect::<Vec<_>>();
|
|
options.push(
|
|
SessionConfigOption::select(
|
|
LEGACY_MODE_CONFIG_ID,
|
|
"Mode",
|
|
modes.current_mode_id.0.clone(),
|
|
choices,
|
|
)
|
|
.category(SessionConfigOptionCategory::Mode),
|
|
);
|
|
}
|
|
options
|
|
}
|
|
|
|
fn replace_config_value(
|
|
options: &Arc<Mutex<Vec<SessionConfigOption>>>,
|
|
config_id: &str,
|
|
value: &SessionConfigOptionValue,
|
|
) -> Vec<SessionConfigOption> {
|
|
let mut options = options.lock().expect("config options");
|
|
if let Some(option) = options
|
|
.iter_mut()
|
|
.find(|option| option.id.0.as_ref() == config_id)
|
|
{
|
|
match (&mut option.kind, value) {
|
|
(SessionConfigKind::Select(select), SessionConfigOptionValue::ValueId { value }) => {
|
|
select.current_value = value.clone();
|
|
}
|
|
(SessionConfigKind::Boolean(boolean), SessionConfigOptionValue::Boolean { value }) => {
|
|
boolean.current_value = *value;
|
|
}
|
|
_ => {}
|
|
}
|
|
}
|
|
options.clone()
|
|
}
|
|
|
|
fn config_notification(session_id: &str, options: Vec<SessionConfigOption>) -> SessionNotification {
|
|
SessionNotification::new(
|
|
SessionId::new(session_id),
|
|
SessionUpdate::ConfigOptionUpdate(ConfigOptionUpdate::new(options)),
|
|
)
|
|
}
|
|
|
|
async fn run_connection<T>(
|
|
transport: T,
|
|
config: ConnectionConfig,
|
|
updates_tx: mpsc::UnboundedSender<SessionNotification>,
|
|
permission_requests: mpsc::UnboundedSender<PermissionRequest>,
|
|
mut commands_rx: mpsc::UnboundedReceiver<Command>,
|
|
ready_tx: oneshot::Sender<Result<String, ReadyError>>,
|
|
) where
|
|
T: agent_client_protocol::ConnectTo<Client> + 'static,
|
|
{
|
|
// Set while `session/load` is in flight. Notification handlers run in
|
|
// the dispatch loop, in arrival order, so everything the agent streams
|
|
// before its load response sees the flag up and is dropped: that is the
|
|
// history the event log already holds (#135).
|
|
let replaying = Arc::new(AtomicBool::new(false));
|
|
let replaying_seen = replaying.clone();
|
|
// ACP callbacks have no turn id. The daemon never queues a later prompt
|
|
// until a cancelled prompt settles, so this active turn is unambiguous.
|
|
let permission_turn = Arc::new(Mutex::new(PermissionTurnState::default()));
|
|
let permission_turn_seen = permission_turn.clone();
|
|
let received_updates = updates_tx.clone();
|
|
let result = Client
|
|
.builder()
|
|
.name("maestro")
|
|
.on_receive_notification(
|
|
async move |notification: SessionNotification, _cx| {
|
|
if replaying_seen.load(Ordering::SeqCst) {
|
|
tracing::trace!("dropping replayed notification during session/load");
|
|
return Ok(());
|
|
}
|
|
// A dropped receiver only means nobody wants updates any more.
|
|
let _ = received_updates.send(notification);
|
|
Ok(())
|
|
},
|
|
agent_client_protocol::on_receive_notification!(),
|
|
)
|
|
.on_receive_request(
|
|
async move |request: RequestPermissionRequest, responder, _cx| {
|
|
let turn = permission_turn_seen
|
|
.lock()
|
|
.expect("permission turn state")
|
|
.callback_turn();
|
|
let Some(turn) = turn else {
|
|
return responder.respond(RequestPermissionResponse::new(
|
|
RequestPermissionOutcome::Cancelled,
|
|
));
|
|
};
|
|
let (response, received) = oneshot::channel();
|
|
if permission_requests
|
|
.send(PermissionRequest {
|
|
turn,
|
|
request,
|
|
response,
|
|
})
|
|
.is_err()
|
|
{
|
|
return responder.respond(RequestPermissionResponse::new(
|
|
RequestPermissionOutcome::Cancelled,
|
|
));
|
|
}
|
|
let outcome = received.await.unwrap_or(RequestPermissionOutcome::Cancelled);
|
|
responder.respond(RequestPermissionResponse::new(outcome))
|
|
},
|
|
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")));
|
|
let initialized = match cx.send_request(init).block_task().await {
|
|
Ok(r) => r,
|
|
Err(e) => {
|
|
let _ = ready_tx.send(Err(ready_error(e)));
|
|
return Ok(());
|
|
}
|
|
};
|
|
let (session_id, modes, initial_options, models) = match config.load_session {
|
|
Some(existing) => {
|
|
if !initialized.agent_capabilities.load_session {
|
|
let _ = ready_tx.send(Err(ReadyError::Handshake(format!(
|
|
"{} ACP adapter ({}) does not advertise loadSession; cannot resume session {existing}",
|
|
config.adapter,
|
|
config.adapter.binary()
|
|
))));
|
|
return Ok(());
|
|
}
|
|
replaying.store(true, Ordering::SeqCst);
|
|
let (loaded_tx, loaded_rx) = oneshot::channel();
|
|
let replaying_done = replaying.clone();
|
|
if let Err(e) = cx
|
|
.send_request(ExtendedLoadSessionRequest {
|
|
session_id: SessionId::new(existing.as_str()),
|
|
cwd: config.cwd,
|
|
mcp_servers: vec![],
|
|
})
|
|
.on_receiving_result(move |result| {
|
|
replaying_done.store(false, Ordering::SeqCst);
|
|
async move {
|
|
let _ = loaded_tx.send(result);
|
|
Ok(())
|
|
}
|
|
})
|
|
{
|
|
let _ = ready_tx.send(Err(ready_error(e)));
|
|
return Ok(());
|
|
}
|
|
let loaded = match loaded_rx.await {
|
|
Ok(Ok(loaded)) => loaded,
|
|
Ok(Err(e)) => {
|
|
let _ = ready_tx.send(Err(ready_error(e)));
|
|
return Ok(());
|
|
}
|
|
Err(_) => {
|
|
let _ = ready_tx.send(Err(ReadyError::Handshake(
|
|
"session/load response was not delivered".into(),
|
|
)));
|
|
return Ok(());
|
|
}
|
|
};
|
|
(existing, loaded.modes, loaded.config_options.unwrap_or_default(), loaded.models)
|
|
}
|
|
None => match cx
|
|
.send_request(ExtendedNewSessionRequest {
|
|
cwd: config.cwd,
|
|
mcp_servers: vec![],
|
|
})
|
|
.block_task()
|
|
.await
|
|
{
|
|
Ok(created) => (
|
|
created.session_id.0.to_string(),
|
|
created.modes,
|
|
created.config_options.unwrap_or_default(),
|
|
created.models,
|
|
),
|
|
Err(e) => {
|
|
let _ = ready_tx.send(Err(ready_error(e)));
|
|
return Ok(());
|
|
}
|
|
},
|
|
};
|
|
let config_options = Arc::new(Mutex::new(advertised_config_options(
|
|
initial_options,
|
|
modes.as_ref(),
|
|
models.as_ref(),
|
|
)));
|
|
if let Some(modes) = &modes {
|
|
let _ = updates_tx.send(SessionNotification::new(
|
|
SessionId::new(session_id.as_str()),
|
|
SessionUpdate::CurrentModeUpdate(CurrentModeUpdate::new(
|
|
modes.current_mode_id.clone(),
|
|
)),
|
|
));
|
|
}
|
|
if !config_options.lock().expect("config options").is_empty() {
|
|
let _ = updates_tx.send(config_notification(
|
|
&session_id,
|
|
config_options.lock().expect("config options").clone(),
|
|
));
|
|
}
|
|
let _ = ready_tx.send(Ok(session_id.clone()));
|
|
|
|
loop {
|
|
tokio::select! {
|
|
cmd = commands_rx.recv() => match cmd {
|
|
Some(Command::Cancel) => {
|
|
permission_turn
|
|
.lock()
|
|
.expect("permission turn state")
|
|
.cancel_active();
|
|
let _ = cx.send_notification(CancelNotification::new(
|
|
SessionId::new(session_id.as_str()),
|
|
));
|
|
}
|
|
Some(Command::Prompt { text, turn, done }) => {
|
|
permission_turn
|
|
.lock()
|
|
.expect("permission turn state")
|
|
.begin(turn);
|
|
let request = PromptRequest::new(
|
|
SessionId::new(session_id.as_str()),
|
|
vec![ContentBlock::from(text)],
|
|
);
|
|
let done = Arc::new(Mutex::new(Some(done)));
|
|
let callback_done = done.clone();
|
|
let callback_turn = permission_turn.clone();
|
|
if let Err(error) = cx.send_request(request).on_receiving_result(move |result| {
|
|
let callback_done = callback_done.clone();
|
|
let callback_turn = callback_turn.clone();
|
|
async move {
|
|
callback_turn
|
|
.lock()
|
|
.expect("permission turn state")
|
|
.finish(turn);
|
|
if let Some(done) = callback_done.lock().expect("prompt sender").take() {
|
|
let _ = done.send(result.map(|_| ()).map_err(|e| e.to_string()));
|
|
}
|
|
Ok(())
|
|
}
|
|
}) {
|
|
permission_turn
|
|
.lock()
|
|
.expect("permission turn state")
|
|
.finish(turn);
|
|
if let Some(done) = done.lock().expect("prompt sender").take() {
|
|
let _ = done.send(Err(error.to_string()));
|
|
}
|
|
}
|
|
}
|
|
Some(Command::SetConfig { config_id, value, done }) => {
|
|
let done = Arc::new(Mutex::new(Some(done)));
|
|
let callback_done = done.clone();
|
|
let callback_options = config_options.clone();
|
|
let callback_updates = updates_tx.clone();
|
|
let callback_session = session_id.clone();
|
|
let callback_config_id = config_id.clone();
|
|
let callback_value = value.clone();
|
|
let sent = if config_id == LEGACY_MODEL_CONFIG_ID {
|
|
let Some(model_id) = value.as_value_id() else {
|
|
let _ = done.lock().expect("config sender").take()
|
|
.map(|done| done.send(Err("model must be a string value".into())));
|
|
continue;
|
|
};
|
|
cx.send_request(LegacySetModelRequest {
|
|
session_id: SessionId::new(session_id.as_str()),
|
|
model_id: model_id.0.to_string(),
|
|
}).on_receiving_result(move |result| async move {
|
|
if result.is_ok() {
|
|
let options = replace_config_value(&callback_options, &callback_config_id, &callback_value);
|
|
let _ = callback_updates.send(config_notification(&callback_session, options));
|
|
}
|
|
if let Some(done) = callback_done.lock().expect("config sender").take() {
|
|
let _ = done.send(result.map(|_| ()).map_err(|error| error.to_string()));
|
|
}
|
|
Ok(())
|
|
})
|
|
} else if config_id == LEGACY_MODE_CONFIG_ID {
|
|
let Some(mode_id) = value.as_value_id() else {
|
|
let _ = done.lock().expect("config sender").take()
|
|
.map(|done| done.send(Err("mode must be a string value".into())));
|
|
continue;
|
|
};
|
|
cx.send_request(SetSessionModeRequest::new(
|
|
SessionId::new(session_id.as_str()),
|
|
mode_id.0.to_string(),
|
|
)).on_receiving_result(move |result| async move {
|
|
if result.is_ok() {
|
|
let options = replace_config_value(&callback_options, &callback_config_id, &callback_value);
|
|
let _ = callback_updates.send(config_notification(&callback_session, options));
|
|
let _ = callback_updates.send(SessionNotification::new(
|
|
SessionId::new(callback_session.as_str()),
|
|
SessionUpdate::CurrentModeUpdate(CurrentModeUpdate::new(
|
|
callback_value.as_value_id().expect("checked mode value").0.clone(),
|
|
)),
|
|
));
|
|
}
|
|
if let Some(done) = callback_done.lock().expect("config sender").take() {
|
|
let _ = done.send(result.map(|_| ()).map_err(|error| error.to_string()));
|
|
}
|
|
Ok(())
|
|
})
|
|
} else {
|
|
cx.send_request(SetSessionConfigOptionRequest::new(
|
|
SessionId::new(session_id.as_str()),
|
|
config_id,
|
|
value,
|
|
))
|
|
.on_receiving_result(move |result| async move {
|
|
if let Ok(response) = &result {
|
|
let mut current = callback_options.lock().expect("config options");
|
|
current.retain(|option| option.id.0.as_ref().starts_with("__maestro_legacy_"));
|
|
current.extend(response.config_options.clone());
|
|
let _ = callback_updates.send(config_notification(&callback_session, current.clone()));
|
|
}
|
|
if let Some(done) = callback_done.lock().expect("config sender").take() {
|
|
let _ = done.send(result.map(|_| ()).map_err(|error| error.to_string()));
|
|
}
|
|
Ok(())
|
|
})
|
|
};
|
|
if let Err(error) = sent {
|
|
if let Some(done) = done.lock().expect("config sender").take() {
|
|
let _ = done.send(Err(error.to_string()));
|
|
}
|
|
}
|
|
}
|
|
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>>>,
|
|
adapter: AcpAdapter,
|
|
) {
|
|
let mut lines = tokio::io::BufReader::new(stderr).lines();
|
|
while let Ok(Some(line)) = lines.next_line().await {
|
|
tracing::debug!(pid, provider = %adapter, "ACP adapter: {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();
|
|
}
|
|
}
|