Compare commits

...
Author SHA1 Message Date
Miguel Palhas bbf622ed9b feat(maestro): select ACP adapter by command 2026-09-06 18:11:04 +01:00
4 changed files with 241 additions and 71 deletions
+98 -29
View File
@@ -58,11 +58,40 @@ use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt};
pub use agent_client_protocol::schema::v1 as schema; pub use agent_client_protocol::schema::v1 as schema;
/// The adapter binary, resolved on `PATH`. /// ACP adapter selected from the structured session's resolved command.
pub const ADAPTER_BIN: &str = "claude-code-acp"; #[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AcpAdapter {
ClaudeCode,
Codex,
}
/// How to install the adapter when it is missing. Named in the error so an impl AcpAdapter {
/// operator staring at a failed session knows what to run. 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 = pub const INSTALL_COMMAND: &str =
"npm install -g --prefix ~/.local @zed-industries/claude-code-acp"; "npm install -g --prefix ~/.local @zed-industries/claude-code-acp";
@@ -166,6 +195,8 @@ pub struct SpawnConfig {
/// instead of opening a fresh one. The adapter must advertise /// instead of opening a fresh one. The adapter must advertise
/// `loadSession`; the handshake fails otherwise. /// `loadSession`; the handshake fails otherwise.
pub load_session: Option<String>, pub load_session: Option<String>,
/// The provider-specific adapter to launch.
pub adapter: AcpAdapter,
} }
impl SpawnConfig { impl SpawnConfig {
@@ -175,6 +206,7 @@ impl SpawnConfig {
env, env,
handshake_timeout: HANDSHAKE_TIMEOUT, handshake_timeout: HANDSHAKE_TIMEOUT,
load_session: None, load_session: None,
adapter: AcpAdapter::ClaudeCode,
} }
} }
@@ -183,6 +215,11 @@ impl SpawnConfig {
self.load_session = Some(session_id.into()); self.load_session = Some(session_id.into());
self self
} }
pub fn adapter(mut self, adapter: AcpAdapter) -> Self {
self.adapter = adapter;
self
}
} }
/// A blocked ACP permission request delivered to the daemon. /// A blocked ACP permission request delivered to the daemon.
@@ -198,31 +235,39 @@ pub struct PermissionRequest {
pub enum AcpError { pub enum AcpError {
/// The adapter is not on `PATH`. The message names the install command /// The adapter is not on `PATH`. The message names the install command
/// so the failed session is actionable rather than silently dead. /// so the failed session is actionable rather than silently dead.
#[error("{ADAPTER_BIN} not found on PATH; install it with: {INSTALL_COMMAND}")] #[error("{adapter} ACP adapter ({bin}) not found on PATH; install it with: {install_command}", bin = adapter.binary(), install_command = adapter.install_command())]
AdapterMissing, AdapterMissing { adapter: AcpAdapter },
#[error("failed to start {ADAPTER_BIN}: {0}")] #[error("failed to start {adapter} ACP adapter ({bin}): {source}", bin = adapter.binary())]
Spawn(#[source] std::io::Error), Spawn {
adapter: AcpAdapter,
#[source]
source: std::io::Error,
},
/// The adapter answered the handshake with an error. /// The adapter answered the handshake with an error.
#[error("ACP handshake failed: {0}")] #[error("ACP handshake failed: {0}")]
Handshake(String), Handshake(String),
/// The adapter went away before the handshake finished. Carries the /// The adapter went away before the handshake finished. Carries the
/// tail of its stderr, which is where a missing Claude login shows up. /// tail of its stderr, which is where a missing Claude login shows up.
#[error("{ADAPTER_BIN} exited during the ACP handshake ({status}){stderr}")] #[error("{adapter} ACP adapter ({bin}) exited during the ACP handshake ({status}){stderr}", bin = adapter.binary())]
ExitedEarly { status: String, stderr: String }, ExitedEarly {
adapter: AcpAdapter,
status: String,
stderr: String,
},
#[error("ACP handshake timed out after {0:?}")] #[error("ACP handshake timed out after {0:?}")]
HandshakeTimeout(Duration), HandshakeTimeout(Duration),
} }
/// Resolve [`ADAPTER_BIN`] against the `PATH` the child would run with: the /// 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. /// session env's `PATH` when it sets one, the daemon's own otherwise.
pub fn resolve_adapter(env: &HashMap<String, String>) -> Option<PathBuf> { pub fn resolve_adapter(adapter: AcpAdapter, env: &HashMap<String, String>) -> Option<PathBuf> {
let path: OsString = match env.get("PATH") { let path: OsString = match env.get("PATH") {
Some(p) => OsString::from(p), Some(p) => OsString::from(p),
None => std::env::var_os("PATH")?, None => std::env::var_os("PATH")?,
}; };
std::env::split_paths(&path) std::env::split_paths(&path)
.filter(|dir| !dir.as_os_str().is_empty()) .filter(|dir| !dir.as_os_str().is_empty())
.map(|dir| dir.join(ADAPTER_BIN)) .map(|dir| dir.join(adapter.binary()))
.find(|candidate| is_executable(candidate)) .find(|candidate| is_executable(candidate))
} }
@@ -248,6 +293,12 @@ enum Command {
Close, Close,
} }
struct ConnectionConfig {
cwd: PathBuf,
load_session: Option<String>,
adapter: AcpAdapter,
}
enum ChildCommand { enum ChildCommand {
Signal(libc::c_int), Signal(libc::c_int),
Reap(oneshot::Sender<bool>), Reap(oneshot::Sender<bool>),
@@ -435,7 +486,9 @@ impl Drop for AcpSession {
/// Fails without a running child: on any error the process is already /// Fails without a running child: on any error the process is already
/// signalled, so a failed session never leaves a stray adapter behind. /// signalled, so a failed session never leaves a stray adapter behind.
pub async fn spawn(config: SpawnConfig) -> Result<AcpSession, AcpError> { pub async fn spawn(config: SpawnConfig) -> Result<AcpSession, AcpError> {
let adapter = resolve_adapter(&config.env).ok_or(AcpError::AdapterMissing)?; 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); let mut cmd = tokio::process::Command::new(&adapter);
cmd.current_dir(&config.cwd) cmd.current_dir(&config.cwd)
@@ -463,18 +516,27 @@ pub async fn spawn(config: SpawnConfig) -> Result<AcpSession, AcpError> {
}); });
} }
} }
let mut child = cmd.spawn().map_err(AcpError::Spawn)?; 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 pid = child.id().expect("child pid is known before wait");
let stdin = child.stdin.take().expect("piped stdin"); let stdin = child.stdin.take().expect("piped stdin");
let stdout = child.stdout.take().expect("piped stdout"); let stdout = child.stdout.take().expect("piped stdout");
let stderr = child.stderr.take().expect("piped stderr"); let stderr = child.stderr.take().expect("piped stderr");
let stderr_tail = Arc::new(Mutex::new(VecDeque::<u8>::new())); let stderr_tail = Arc::new(Mutex::new(VecDeque::<u8>::new()));
tokio::spawn(drain_stderr(stderr, pid, stderr_tail.clone())); tokio::spawn(drain_stderr(stderr, pid, stderr_tail.clone(), provider));
let (exit_tx, exit_rx) = watch::channel(None); let (exit_tx, exit_rx) = watch::channel(None);
let (child_commands_tx, child_commands_rx) = mpsc::unbounded_channel(); let (child_commands_tx, child_commands_rx) = mpsc::unbounded_channel();
tokio::spawn(supervise_child(child, pid, child_commands_rx, exit_tx)); tokio::spawn(supervise_child(
child,
pid,
child_commands_rx,
exit_tx,
provider,
));
let (updates_tx, updates_rx) = mpsc::unbounded_channel(); let (updates_tx, updates_rx) = mpsc::unbounded_channel();
let (permission_tx, permission_rx) = mpsc::unbounded_channel(); let (permission_tx, permission_rx) = mpsc::unbounded_channel();
@@ -483,8 +545,11 @@ pub async fn spawn(config: SpawnConfig) -> Result<AcpSession, AcpError> {
let transport = ByteStreams::new(stdin.compat_write(), stdout.compat()); let transport = ByteStreams::new(stdin.compat_write(), stdout.compat());
tokio::spawn(run_connection( tokio::spawn(run_connection(
transport, transport,
config.cwd.clone(), ConnectionConfig {
config.load_session.clone(), cwd: config.cwd.clone(),
load_session: config.load_session.clone(),
adapter: provider,
},
updates_tx, updates_tx,
permission_tx, permission_tx,
commands_rx, commands_rx,
@@ -517,6 +582,7 @@ pub async fn spawn(config: SpawnConfig) -> Result<AcpSession, AcpError> {
.to_string() .to_string()
}; };
AcpError::ExitedEarly { AcpError::ExitedEarly {
adapter: provider,
status: status.map_or_else(|| "unknown status".into(), |s| s.to_string()), status: status.map_or_else(|| "unknown status".into(), |s| s.to_string()),
stderr: if stderr.is_empty() { stderr: if stderr.is_empty() {
String::new() String::new()
@@ -551,12 +617,13 @@ async fn supervise_child(
pid: u32, pid: u32,
mut commands: mpsc::UnboundedReceiver<ChildCommand>, mut commands: mpsc::UnboundedReceiver<ChildCommand>,
exit_tx: watch::Sender<Option<ExitStatus>>, exit_tx: watch::Sender<Option<ExitStatus>>,
adapter: AcpAdapter,
) { ) {
loop { loop {
tokio::select! { tokio::select! {
status = child.wait() => { status = child.wait() => {
let status = status.ok(); let status = status.ok();
tracing::debug!(pid, ?status, "{ADAPTER_BIN} exited"); tracing::debug!(pid, ?status, provider = %adapter, "ACP adapter exited");
let _ = exit_tx.send(status); let _ = exit_tx.send(status);
return; return;
} }
@@ -570,7 +637,7 @@ async fn supervise_child(
Some(ChildCommand::Reap(done)) => match child.try_wait() { Some(ChildCommand::Reap(done)) => match child.try_wait() {
Ok(Some(status)) => { Ok(Some(status)) => {
let _ = done.send(true); let _ = done.send(true);
tracing::debug!(pid, ?status, "{ADAPTER_BIN} exited"); tracing::debug!(pid, ?status, provider = %adapter, "ACP adapter exited");
let _ = exit_tx.send(Some(status)); let _ = exit_tx.send(Some(status));
return; return;
} }
@@ -578,7 +645,7 @@ async fn supervise_child(
let _ = done.send(false); let _ = done.send(false);
} }
Err(error) => { Err(error) => {
tracing::warn!(pid, %error, "could not reap {ADAPTER_BIN}"); tracing::warn!(pid, %error, provider = %adapter, "could not reap ACP adapter");
let _ = done.send(false); let _ = done.send(false);
} }
}, },
@@ -588,7 +655,7 @@ async fn supervise_child(
// a reused PID. // a reused PID.
unsafe { libc::kill(pid as libc::pid_t, libc::SIGTERM) }; unsafe { libc::kill(pid as libc::pid_t, libc::SIGTERM) };
let status = child.wait().await.ok(); let status = child.wait().await.ok();
tracing::debug!(pid, ?status, "{ADAPTER_BIN} exited"); tracing::debug!(pid, ?status, provider = %adapter, "ACP adapter exited");
let _ = exit_tx.send(status); let _ = exit_tx.send(status);
return; return;
} }
@@ -704,8 +771,7 @@ fn config_notification(session_id: &str, options: Vec<SessionConfigOption>) -> S
async fn run_connection<T>( async fn run_connection<T>(
transport: T, transport: T,
cwd: PathBuf, config: ConnectionConfig,
load_session: Option<String>,
updates_tx: mpsc::UnboundedSender<SessionNotification>, updates_tx: mpsc::UnboundedSender<SessionNotification>,
permission_requests: mpsc::UnboundedSender<PermissionRequest>, permission_requests: mpsc::UnboundedSender<PermissionRequest>,
mut commands_rx: mpsc::UnboundedReceiver<Command>, mut commands_rx: mpsc::UnboundedReceiver<Command>,
@@ -779,11 +845,13 @@ async fn run_connection<T>(
return Ok(()); return Ok(());
} }
}; };
let (session_id, modes, initial_options, models) = match load_session { let (session_id, modes, initial_options, models) = match config.load_session {
Some(existing) => { Some(existing) => {
if !initialized.agent_capabilities.load_session { if !initialized.agent_capabilities.load_session {
let _ = ready_tx.send(Err(ReadyError::Handshake(format!( let _ = ready_tx.send(Err(ReadyError::Handshake(format!(
"{ADAPTER_BIN} does not advertise loadSession; cannot resume session {existing}" "{} ACP adapter ({}) does not advertise loadSession; cannot resume session {existing}",
config.adapter,
config.adapter.binary()
)))); ))));
return Ok(()); return Ok(());
} }
@@ -793,7 +861,7 @@ async fn run_connection<T>(
if let Err(e) = cx if let Err(e) = cx
.send_request(ExtendedLoadSessionRequest { .send_request(ExtendedLoadSessionRequest {
session_id: SessionId::new(existing.as_str()), session_id: SessionId::new(existing.as_str()),
cwd, cwd: config.cwd,
mcp_servers: vec![], mcp_servers: vec![],
}) })
.on_receiving_result(move |result| { .on_receiving_result(move |result| {
@@ -824,7 +892,7 @@ async fn run_connection<T>(
} }
None => match cx None => match cx
.send_request(ExtendedNewSessionRequest { .send_request(ExtendedNewSessionRequest {
cwd, cwd: config.cwd,
mcp_servers: vec![], mcp_servers: vec![],
}) })
.block_task() .block_task()
@@ -1004,10 +1072,11 @@ async fn drain_stderr(
stderr: tokio::process::ChildStderr, stderr: tokio::process::ChildStderr,
pid: u32, pid: u32,
tail: Arc<Mutex<VecDeque<u8>>>, tail: Arc<Mutex<VecDeque<u8>>>,
adapter: AcpAdapter,
) { ) {
let mut lines = tokio::io::BufReader::new(stderr).lines(); let mut lines = tokio::io::BufReader::new(stderr).lines();
while let Ok(Some(line)) = lines.next_line().await { while let Ok(Some(line)) = lines.next_line().await {
tracing::debug!(pid, "{ADAPTER_BIN}: {line}"); tracing::debug!(pid, provider = %adapter, "ACP adapter: {line}");
let mut tail = tail.lock().expect("stderr tail"); let mut tail = tail.lock().expect("stderr tail");
tail.extend(line.as_bytes()); tail.extend(line.as_bytes());
tail.push_back(b'\n'); tail.push_back(b'\n');
+30 -3
View File
@@ -6,7 +6,7 @@ use std::path::{Path, PathBuf};
use std::time::{Duration, Instant}; use std::time::{Duration, Instant};
use maestro_acp::schema::{ContentBlock, SessionUpdate}; use maestro_acp::schema::{ContentBlock, SessionUpdate};
use maestro_acp::{AcpError, SpawnConfig, ADAPTER_BIN, INSTALL_COMMAND}; use maestro_acp::{AcpAdapter, AcpError, SpawnConfig, ADAPTER_BIN, INSTALL_COMMAND};
use tempfile::TempDir; use tempfile::TempDir;
fn fake_agent() -> PathBuf { fn fake_agent() -> PathBuf {
@@ -25,6 +25,7 @@ impl Fake {
let bin = dir.path().join("bin"); let bin = dir.path().join("bin");
std::fs::create_dir_all(&bin).unwrap(); std::fs::create_dir_all(&bin).unwrap();
std::fs::copy(fake_agent(), bin.join(ADAPTER_BIN)).unwrap(); std::fs::copy(fake_agent(), bin.join(ADAPTER_BIN)).unwrap();
std::fs::copy(fake_agent(), bin.join(AcpAdapter::Codex.binary())).unwrap();
Self { dir } Self { dir }
} }
@@ -146,12 +147,38 @@ async fn missing_adapter_names_the_install_command() {
let err = maestro_acp::spawn(SpawnConfig::new(cwd.path(), env)) let err = maestro_acp::spawn(SpawnConfig::new(cwd.path(), env))
.await .await
.expect_err("no adapter on PATH"); .expect_err("no adapter on PATH");
assert!(matches!(err, AcpError::AdapterMissing)); assert!(matches!(
err,
AcpError::AdapterMissing {
adapter: AcpAdapter::ClaudeCode
}
));
let msg = err.to_string(); let msg = err.to_string();
assert!(msg.contains(ADAPTER_BIN), "{msg}"); assert!(msg.contains(ADAPTER_BIN), "{msg}");
assert!(msg.contains(INSTALL_COMMAND), "{msg}"); assert!(msg.contains(INSTALL_COMMAND), "{msg}");
} }
#[test]
fn codex_adapter_has_its_own_binary_and_install_guidance() {
assert_eq!(AcpAdapter::Codex.binary(), "codex-acp");
assert!(AcpAdapter::Codex
.install_command()
.contains("@agentclientprotocol/codex-acp"));
}
#[tokio::test]
async fn spawn_selects_the_codex_adapter() {
let fake = Fake::new();
let cwd = TempDir::new().unwrap();
let session = maestro_acp::spawn(fake.config(cwd.path()).adapter(AcpAdapter::Codex))
.await
.expect("spawn Codex adapter");
assert!(fake.log().contains("recv session/new"));
session
.terminate(Duration::from_millis(100), Duration::from_secs(2))
.await;
}
#[tokio::test] #[tokio::test]
async fn session_updates_reach_the_taken_receiver() { async fn session_updates_reach_the_taken_receiver() {
let fake = Fake::new(); let fake = Fake::new();
@@ -341,7 +368,7 @@ async fn child_exit_before_handshake_reports_exited_early() {
let err = maestro_acp::spawn(config).await.expect_err("child died"); let err = maestro_acp::spawn(config).await.expect_err("child died");
match &err { match &err {
AcpError::ExitedEarly { status, stderr } => { AcpError::ExitedEarly { status, stderr, .. } => {
assert!(status.contains('3'), "{status}"); assert!(status.contains('3'), "{status}");
assert!(stderr.contains("not logged in"), "{stderr}"); assert!(stderr.contains("not logged in"), "{stderr}");
} }
+66 -30
View File
@@ -15,7 +15,8 @@ use std::task::{Context, Poll};
use std::time::Duration; use std::time::Duration;
use chrono::Utc; use chrono::Utc;
use maestro_acp::{AcpSession, PermissionRequest, SpawnConfig}; use maestro_acp::{AcpAdapter, AcpSession, PermissionRequest, SpawnConfig};
use maestro_protocol::display::command_tool;
use maestro_protocol::{ use maestro_protocol::{
ActivityState, CatalogEvent, CreateSessionRequest, ErrorInfo, NameSource, Session, ActivityState, CatalogEvent, CreateSessionRequest, ErrorInfo, NameSource, Session,
SessionStatus, SessionView, StructuredReplayEvent, TranscriptDelta, TranscriptRow, SessionStatus, SessionView, StructuredReplayEvent, TranscriptDelta, TranscriptRow,
@@ -59,6 +60,16 @@ pub const DEFAULT_BLOCKED_WORK_INTERVAL: Duration = Duration::from_secs(15);
/// actually go. /// actually go.
const SUBMIT_DELAY: Duration = Duration::from_millis(150); const SUBMIT_DELAY: Duration = Duration::from_millis(150);
fn structured_adapter(command: &str) -> Result<AcpAdapter, String> {
match command_tool(command) {
Some("claude") => Ok(AcpAdapter::ClaudeCode),
Some("codex") => Ok(AcpAdapter::Codex),
_ => Err(format!(
"structured view supports Claude Code or Codex commands; cannot select an ACP adapter for {command:?}"
)),
}
}
/// Per-session daemon-local bookkeeping that complements the public `Session`. /// Per-session daemon-local bookkeeping that complements the public `Session`.
#[derive(Debug)] #[derive(Debug)]
pub struct SessionRecord { pub struct SessionRecord {
@@ -1250,14 +1261,18 @@ impl SessionManager {
// up still yields a session — Dead, with the reason on it — so the // up still yields a session — Dead, with the reason on it — so the
// operator sees what to fix instead of a silently missing session. // operator sees what to fix instead of a silently missing session.
let (acp, acp_failure) = match req.view { let (acp, acp_failure) = match req.view {
SessionView::Structured => { SessionView::Structured => match structured_adapter(&req.command) {
match maestro_acp::spawn(SpawnConfig::new( Err(error) => (None, Some(error)),
&effective_cwd, Ok(adapter) => match maestro_acp::spawn(
session_env SpawnConfig::new(
.iter() &effective_cwd,
.map(|(k, v)| (k.clone(), v.clone())) session_env
.collect(), .iter()
)) .map(|(k, v)| (k.clone(), v.clone()))
.collect(),
)
.adapter(adapter),
)
.await .await
{ {
Ok(acp) => (Some(Arc::new(acp)), None), Ok(acp) => (Some(Arc::new(acp)), None),
@@ -1265,8 +1280,8 @@ impl SessionManager {
tracing::warn!(%id, error = %e, "structured session: adapter failed to start"); tracing::warn!(%id, error = %e, "structured session: adapter failed to start");
(None, Some(e.to_string())) (None, Some(e.to_string()))
} }
} },
} },
SessionView::Terminal => { SessionView::Terminal => {
let spawned = { let spawned = {
let tmux = self.inner.tmux.clone(); let tmux = self.inner.tmux.clone();
@@ -1674,23 +1689,30 @@ impl SessionManager {
let (acp, failure) = if recoverable { let (acp, failure) = if recoverable {
match session.acp_session_id.clone() { match session.acp_session_id.clone() {
Some(acp_session_id) => { Some(acp_session_id) => match structured_adapter(&session.command) {
let config = SpawnConfig::new( Err(error) => {
&session.cwd, tracing::warn!(%id, %error, "structured session: unsupported command during recovery");
session_env (None, Some(error))
.iter() }
.map(|(key, value)| (key.clone(), value.clone())) Ok(adapter) => {
.collect(), let config = SpawnConfig::new(
) &session.cwd,
.load_session(acp_session_id); session_env
match maestro_acp::spawn(config).await { .iter()
Ok(acp) => (Some(Arc::new(acp)), None), .map(|(key, value)| (key.clone(), value.clone()))
Err(error) => { .collect(),
tracing::warn!(%id, %error, "structured session: adapter failed to resume"); )
(None, Some(error.to_string())) .load_session(acp_session_id)
.adapter(adapter);
match maestro_acp::spawn(config).await {
Ok(acp) => (Some(Arc::new(acp)), None),
Err(error) => {
tracing::warn!(%id, %error, "structured session: adapter failed to resume");
(None, Some(error.to_string()))
}
} }
} }
} },
None => ( None => (
None, None,
Some("structured session has no stored ACP session id".into()), Some("structured session has no stored ACP session id".into()),
@@ -2969,15 +2991,18 @@ mod tests {
use std::task::Poll; use std::task::Poll;
use std::time::Duration; use std::time::Duration;
use maestro_acp::schema::{ use maestro_acp::{
PermissionOption, PermissionOptionKind, RequestPermissionOutcome, RequestPermissionRequest, schema::{
ToolCallUpdate, ToolCallUpdateFields, PermissionOption, PermissionOptionKind, RequestPermissionOutcome,
RequestPermissionRequest, ToolCallUpdate, ToolCallUpdateFields,
},
AcpAdapter,
}; };
use maestro_protocol::{ActivityState, CatalogEvent, CreateSessionRequest, SessionView}; use maestro_protocol::{ActivityState, CatalogEvent, CreateSessionRequest, SessionView};
use super::{ use super::{
apply_structured_activity, cancel_pending_approvals, permission_outcome, apply_structured_activity, cancel_pending_approvals, permission_outcome,
record_permission_request, run_structured_activity_pass, record_permission_request, run_structured_activity_pass, structured_adapter,
sync_structured_activity_from_stream, AcpDispatchState, AcpEventStream, ApprovalDecision, sync_structured_activity_from_stream, AcpDispatchState, AcpEventStream, ApprovalDecision,
RecordOutcome, SessionManager, RecordOutcome, SessionManager,
}; };
@@ -3001,6 +3026,17 @@ mod tests {
) )
} }
#[test]
fn structured_adapter_uses_the_shared_command_resolution() {
assert_eq!(structured_adapter("claude"), Ok(AcpAdapter::ClaudeCode));
assert_eq!(
structured_adapter("sudo -u agent -- /opt/bin/codex"),
Ok(AcpAdapter::Codex)
);
assert!(structured_adapter("opencode").is_err());
assert!(structured_adapter("claude && codex").is_err());
}
fn stream(log: EventLog) -> AcpEventStream { fn stream(log: EventLog) -> AcpEventStream {
let (live, _live_rx) = tokio::sync::broadcast::channel(16); let (live, _live_rx) = tokio::sync::broadcast::channel(16);
let (projected_live, _projected_rx) = tokio::sync::broadcast::channel(16); let (projected_live, _projected_rx) = tokio::sync::broadcast::channel(16);
+47 -9
View File
@@ -11,7 +11,7 @@ use std::time::{Duration, Instant};
use common::{init_bare_repo, init_repo, push_all_to_origin, wait_for_async, DaemonFixture}; use common::{init_bare_repo, init_repo, push_all_to_origin, wait_for_async, DaemonFixture};
use maestro_acp::schema::{ContentBlock, SessionUpdate}; use maestro_acp::schema::{ContentBlock, SessionUpdate};
use maestro_acp::{ADAPTER_BIN, INSTALL_COMMAND}; use maestro_acp::{AcpAdapter, ADAPTER_BIN};
use maestro_daemon::projection::TurnLifecycle; use maestro_daemon::projection::TurnLifecycle;
use maestro_daemon::sidecar::{EventCursor, SidecarStore}; use maestro_daemon::sidecar::{EventCursor, SidecarStore};
use maestro_daemon::worktree::WorktreeManager; use maestro_daemon::worktree::WorktreeManager;
@@ -38,6 +38,7 @@ impl Fake {
let bin = dir.path().join("bin"); let bin = dir.path().join("bin");
std::fs::create_dir_all(&bin).unwrap(); std::fs::create_dir_all(&bin).unwrap();
std::fs::copy(fake_agent(), bin.join(ADAPTER_BIN)).unwrap(); std::fs::copy(fake_agent(), bin.join(ADAPTER_BIN)).unwrap();
std::fs::copy(fake_agent(), bin.join(AcpAdapter::Codex.binary())).unwrap();
Self { dir } Self { dir }
} }
@@ -96,10 +97,18 @@ fn event_text(event: &maestro_daemon::sidecar::LoggedEvent) -> &str {
} }
fn structured(cwd: &Path, env: HashMap<String, String>) -> CreateSessionRequest { fn structured(cwd: &Path, env: HashMap<String, String>) -> CreateSessionRequest {
structured_command(cwd, env, "claude")
}
fn structured_command(
cwd: &Path,
env: HashMap<String, String>,
command: &str,
) -> CreateSessionRequest {
CreateSessionRequest { CreateSessionRequest {
view: SessionView::Structured, view: SessionView::Structured,
cwd: cwd.to_path_buf(), cwd: cwd.to_path_buf(),
command: "claude".into(), command: command.into(),
env, env,
metadata: HashMap::new(), metadata: HashMap::new(),
worktree: None, worktree: None,
@@ -130,7 +139,11 @@ async fn structured_spawn_starts_the_adapter_in_cwd_with_session_env() {
let session = fx let session = fx
.manager .manager
.create(structured(cwd.path(), env)) .create(structured_command(
cwd.path(),
env,
"codex --dangerously-bypass-approvals-and-sandbox",
))
.await .await
.expect("create"); .expect("create");
@@ -215,7 +228,7 @@ async fn structured_activity_tracks_turn_completion_and_failure() {
env.insert("FAKE_ACP_PROMPT_FAIL_ONCE".into(), "1".into()); env.insert("FAKE_ACP_PROMPT_FAIL_ONCE".into(), "1".into());
let session = fx let session = fx
.manager .manager
.create(structured(cwd.path(), env)) .create(structured_command(cwd.path(), env, "codex"))
.await .await
.unwrap(); .unwrap();
@@ -570,7 +583,7 @@ async fn recovery_loads_the_acp_session_without_duplicating_history() {
let session = fx let session = fx
.manager .manager
.create(structured(cwd.path(), env)) .create(structured_command(cwd.path(), env, "codex"))
.await .await
.expect("create"); .expect("create");
let acp_id = session.acp_session_id.clone().expect("acp id"); let acp_id = session.acp_session_id.clone().expect("acp id");
@@ -684,7 +697,7 @@ async fn shutdown_preserves_a_pristine_structured_worktree_for_recovery() {
.create(CreateSessionRequest { .create(CreateSessionRequest {
view: SessionView::Structured, view: SessionView::Structured,
cwd: repo.path().to_path_buf(), cwd: repo.path().to_path_buf(),
command: "claude".into(), command: "codex".into(),
env: fake.env(), env: fake.env(),
metadata: HashMap::new(), metadata: HashMap::new(),
worktree: Some(WorktreeSpec { worktree: Some(WorktreeSpec {
@@ -779,15 +792,18 @@ async fn missing_adapter_yields_a_failed_session_naming_the_install_command() {
let session = fx let session = fx
.manager .manager
.create(structured(cwd.path(), env)) .create(structured_command(cwd.path(), env, "codex"))
.await .await
.expect("a failed session, not an error"); .expect("a failed session, not an error");
assert_eq!(session.status, SessionStatus::Dead); assert_eq!(session.status, SessionStatus::Dead);
assert_eq!(session.activity, ActivityState::Error); assert_eq!(session.activity, ActivityState::Error);
let detail = session.activity_detail.clone().expect("actionable detail"); let detail = session.activity_detail.clone().expect("actionable detail");
assert!(detail.contains(ADAPTER_BIN), "{detail}"); assert!(detail.contains(AcpAdapter::Codex.binary()), "{detail}");
assert!(detail.contains(INSTALL_COMMAND), "{detail}"); assert!(
detail.contains(AcpAdapter::Codex.install_command()),
"{detail}"
);
assert_eq!(session.acp_session_id, None); assert_eq!(session.acp_session_id, None);
// It is a real session: listed, on disk, and dismissable like any other // It is a real session: listed, on disk, and dismissable like any other
@@ -801,6 +817,28 @@ async fn missing_adapter_yields_a_failed_session_naming_the_install_command() {
assert!(!fx.manager.has(session.id).await); assert!(!fx.manager.has(session.id).await);
} }
#[tokio::test]
async fn unsupported_structured_command_fails_without_launching_claude() {
let fx = DaemonFixture::new();
let fake = Fake::new();
let cwd = TempDir::new().unwrap();
let session = fx
.manager
.create(structured_command(cwd.path(), fake.env(), "opencode"))
.await
.expect("failed structured session is persisted");
assert_eq!(session.status, SessionStatus::Dead);
let detail = session.activity_detail.expect("selection error");
assert!(detail.contains("Claude Code or Codex"), "{detail}");
assert!(
fake.log().is_empty(),
"unsupported command launched an adapter"
);
fx.manager.kill(session.id, false).await.expect("dismiss");
}
#[tokio::test] #[tokio::test]
async fn kill_sends_cancel_then_sigterm_and_removes_the_sidecar() { async fn kill_sends_cancel_then_sigterm_and_removes_the_sidecar() {
let fx = DaemonFixture::new(); let fx = DaemonFixture::new();