722 lines
28 KiB
Rust
722 lines
28 KiB
Rust
//! 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.
|
|
//!
|
|
//! 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, ContentBlock, FileSystemCapabilities, Implementation,
|
|
InitializeRequest, LoadSessionRequest, NewSessionRequest, PromptRequest,
|
|
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,
|
|
/// 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>,
|
|
}
|
|
|
|
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,
|
|
}
|
|
}
|
|
|
|
/// 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
|
|
}
|
|
}
|
|
|
|
/// 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_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,
|
|
Prompt {
|
|
text: String,
|
|
turn: u64,
|
|
done: oneshot::Sender<Result<(), String>>,
|
|
},
|
|
Close,
|
|
}
|
|
|
|
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)
|
|
}
|
|
|
|
/// 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 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);
|
|
let (child_commands_tx, child_commands_rx) = mpsc::unbounded_channel();
|
|
tokio::spawn(supervise_child(child, pid, child_commands_rx, exit_tx));
|
|
|
|
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,
|
|
config.cwd.clone(),
|
|
config.load_session.clone(),
|
|
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 {
|
|
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>>,
|
|
) {
|
|
loop {
|
|
tokio::select! {
|
|
status = child.wait() => {
|
|
let status = status.ok();
|
|
tracing::debug!(pid, ?status, "{ADAPTER_BIN} 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, "{ADAPTER_BIN} exited");
|
|
let _ = exit_tx.send(Some(status));
|
|
return;
|
|
}
|
|
Ok(None) => {
|
|
let _ = done.send(false);
|
|
}
|
|
Err(error) => {
|
|
tracing::warn!(pid, %error, "could not reap {ADAPTER_BIN}");
|
|
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, "{ADAPTER_BIN} 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)
|
|
}
|
|
|
|
async fn run_connection<T>(
|
|
transport: T,
|
|
cwd: PathBuf,
|
|
load_session: Option<String>,
|
|
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 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 _ = updates_tx.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 = match load_session {
|
|
Some(existing) => {
|
|
if !initialized.agent_capabilities.load_session {
|
|
let _ = ready_tx.send(Err(ReadyError::Handshake(format!(
|
|
"{ADAPTER_BIN} does not advertise loadSession; cannot resume session {existing}"
|
|
))));
|
|
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(LoadSessionRequest::new(existing.clone(), cwd))
|
|
.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(());
|
|
}
|
|
match loaded_rx.await {
|
|
Ok(Ok(_)) => {}
|
|
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
|
|
}
|
|
None => match cx
|
|
.send_request(NewSessionRequest::new(cwd))
|
|
.block_task()
|
|
.await
|
|
{
|
|
Ok(created) => created.session_id.0.to_string(),
|
|
Err(e) => {
|
|
let _ = ready_tx.send(Err(ready_error(e)));
|
|
return Ok(());
|
|
}
|
|
},
|
|
};
|
|
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::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();
|
|
}
|
|
}
|