Compare commits

..
Author SHA1 Message Date
Miguel Palhas 2e7c5f7f58 fix: recover PWA after backgrounding
ci / web (push) Successful in 48s
ci / flake (push) Successful in 2m56s
ci / rust (push) Successful in 6m4s
2026-09-05 18:10:21 +01:00
Miguel Palhas 3529947497 Merge branch 'fix/codex-attach-scroll'
ci / web (push) Successful in 1m3s
ci / flake (push) Successful in 3m3s
ci / rust (push) Successful in 7m35s
2026-09-05 03:49:35 +01:00
Miguel Palhas 4851ad9101 fix(maestro): follow initial attach replay 2026-09-05 03:48:35 +01:00
Miguel Palhas 3cb4bebd1e fix(maestro): update second stale worktree test for branch suffixing
ci / web (push) Successful in 54s
ci / flake (push) Successful in 3m8s
ci / rust (push) Successful in 8m30s
second_spawn_on_same_branch_is_rejected asserted the same pre-#77
outright-rejection contract as the unit test fixed in 0e20baf, just at
the e2e/session level — also broken by 49ab7e9's branch-collision
suffixing and also already failing on main.

Renamed to second_spawn_on_same_branch_gets_suffixed and rewritten to
assert the actual behavior: the second spawn succeeds on a `-2`
suffixed branch and directory, and the first worktree is untouched.
2026-09-04 20:08:16 +01:00
Miguel Palhas 0e20baf7c2 fix(maestro): update stale worktree test for branch-collision suffixing
ci / web (push) Failing after 38s
ci / flake (push) Successful in 4m17s
ci / rust (push) Failing after 7m40s
add_fails_when_branch_already_exists asserted the pre-#77 behavior:
that a branch collision errors outright. Since 49ab7e9, add_new
sidesteps a branch collision with a numeric suffix instead, so the
test was failing CI on main regardless of any given change.

Replaces it with a test asserting the actual (and intended) suffixing
behavior.
2026-09-04 19:57:33 +01:00
Miguel Palhas af8acbfbf2 test(maestro): verify umbrella binary routes each subcommand to its crate
ci / web (push) Successful in 1m10s
ci / rust (push) Failing after 3m17s
ci / flake (push) Successful in 7m6s
Closes #10

The compile_error guard and per-feature build already had CI coverage
(the "feature-gate matrix" job). What was missing was proof that a
subcommand actually reaches its own crate's dispatch at runtime, not
just that clap parses it.

A malformed --config file makes cli/daemon/tui fail identically at
config load, before touching tmux, the network, or a terminal. That
turns "clap parsed it but the handler never ran" (a usage error, exit
2) into a distinguishable outcome from "it dispatched into the real
crate" (a `toml decode` runtime error via main's error printer, exit
1). The absence-side tests are feature-gated to only exist in a
single-feature build, so they run under CI's existing per-feature
matrix rather than the default all-features build.
2026-09-04 19:47:32 +01:00
Miguel Palhas 1ad7845f51 fix(maestro): refresh attach header when a session is renamed
Header name/command were baked into innerHTML at connection creation
and never revisited. ConnectionPool.mount() reused the cached
connection on later calls but dropped the fresh info, so a header
never picked up haiku autoname's rename.
2026-09-04 19:38:09 +01:00
44 changed files with 363 additions and 2354 deletions
Generated
+57 -745
View File
File diff suppressed because it is too large Load Diff
-3
View File
@@ -10,7 +10,6 @@ repository = "https://git.naps.pt/naps62/maestro"
publish = false
[workspace.dependencies]
agent-client-protocol = "2.1"
axum = { version = "0.7", features = ["ws"] }
chrono = { version = "0.4", default-features = false, features = ["clock", "serde"] }
clap = { version = "4.5", features = ["derive", "env"] }
@@ -18,7 +17,6 @@ crossterm = { version = "0.28", features = ["event-stream"] }
dirs = "5"
futures-util = { version = "0.3", default-features = false, features = ["sink", "std"] }
libc = "0.2"
maestro-acp = { path = "crates/maestro-acp" }
maestro-config = { path = "crates/maestro-config" }
# `default-features = false` so each consumer opts into `haiku-autoname`
# through its own feature of the same name (#131); a workspace entry that
@@ -40,7 +38,6 @@ tempfile = "3"
thiserror = "1"
tokio = { version = "1", features = ["macros", "rt-multi-thread", "sync", "time", "net", "io-util", "fs", "signal", "process"] }
tokio-tungstenite = "0.21"
tokio-util = { version = "0.7", features = ["compat"] }
toml = "0.8"
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter"] }
-20
View File
@@ -1,20 +0,0 @@
[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
-468
View File
@@ -1,468 +0,0 @@
//! 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();
}
}
@@ -1,98 +0,0 @@
#!/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}"}})
-292
View File
@@ -1,292 +0,0 @@
//! `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());
}
-3
View File
@@ -51,7 +51,6 @@ pub async fn run_spawn(client: &DaemonClient, args: SpawnArgs) -> Result<Session
};
let body = CreateSessionRequest {
view: Default::default(),
cwd: args.cwd.clone(),
command: args.command.clone(),
env,
@@ -237,7 +236,6 @@ mod tests {
fn session(cwd: &str, command: &str, id: u128) -> Session {
Session {
view: Default::default(),
id: Uuid::from_u128(id),
cwd: PathBuf::from(cwd),
command: command.into(),
@@ -254,7 +252,6 @@ mod tests {
current_cwd: None,
blocked_work: None,
last_focused_at: None,
acp_session_id: None,
}
}
@@ -91,7 +91,6 @@ async fn spawn_round_trips_through_the_daemon() {
let created = mock
.client()
.create_session(&maestro_protocol::CreateSessionRequest {
view: Default::default(),
cwd: "/home/mock".into(),
command: "bash".into(),
env: Default::default(),
-2
View File
@@ -28,7 +28,6 @@ dirs.workspace = true
futures-util.workspace = true
include_dir = { workspace = true, optional = true }
libc.workspace = true
maestro-acp.workspace = true
mime_guess = { workspace = true, optional = true }
maestro-config.workspace = true
maestro-hooks = { workspace = true, default-features = false }
@@ -43,7 +42,6 @@ tracing-subscriber.workspace = true
uuid.workspace = true
[dev-dependencies]
libc.workspace = true
reqwest.workspace = true
tokio-tungstenite.workspace = true
urlencoding = "2"
+3 -14
View File
@@ -106,7 +106,7 @@ async fn run(config_override: Option<PathBuf>) -> Result<(), Box<dyn Error>> {
projects.start_background_refresh();
let router = local_api::router(local_api::state_for(
manager.clone(),
manager,
config.hostname.clone(),
config.home_dir.clone(),
config.spawn_presets.clone(),
@@ -118,25 +118,14 @@ async fn run(config_override: Option<PathBuf>) -> Result<(), Box<dyn Error>> {
let bound = listener.local_addr()?;
tracing::info!(addr = %bound, "API listening");
// Ctrl-C at a terminal or SIGTERM from systemd: either one is a
// shutdown, and both must reach the adapter children below (#134). The
// unit's `KillMode=process` signals only the daemon, so nothing else
// would end them.
let shutdown = async {
let mut sigterm = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
.expect("install SIGTERM handler");
tokio::select! {
_ = tokio::signal::ctrl_c() => tracing::info!("ctrl-c received — shutting down"),
_ = sigterm.recv() => tracing::info!("SIGTERM received — shutting down"),
}
let _ = tokio::signal::ctrl_c().await;
tracing::info!("ctrl-c received — shutting down");
};
axum::serve(listener, router)
.tcp_nodelay(true)
.with_graceful_shutdown(shutdown)
.await?;
// tmux sessions outlive the daemon by design; structured sessions'
// adapter children do not (#134).
manager.shutdown().await;
Ok(())
}
+31 -155
View File
@@ -12,10 +12,9 @@ use std::sync::Arc;
use std::time::Duration;
use chrono::Utc;
use maestro_acp::{AcpSession, SpawnConfig};
use maestro_protocol::{
ActivityState, CatalogEvent, CreateSessionRequest, ErrorInfo, NameSource, Session,
SessionStatus, SessionView, WorktreeSpec, NAME_KEY, NAME_SOURCE_KEY,
SessionStatus, WorktreeSpec, NAME_KEY, NAME_SOURCE_KEY,
};
use tokio::sync::{broadcast, Mutex};
use uuid::Uuid;
@@ -88,12 +87,6 @@ pub struct SessionRecord {
/// glow, this holds through focus loss until the session does new work.
/// Daemon-local; never rides the wire.
pub awaiting_suppressed: bool,
/// The `claude-code-acp` child behind a [`SessionView::Structured`]
/// session (#134): its connection, pid and the raw notification stream.
/// `None` for terminal sessions and for a structured session whose
/// adapter never came up. Shared so a kill or shutdown can terminate it
/// outside the sessions lock. Daemon-local; never rides the wire.
pub acp: Option<Arc<AcpSession>>,
}
#[derive(Debug, thiserror::Error)]
@@ -294,47 +287,6 @@ impl SessionManager {
sessions.values().map(|r| r.session.clone()).collect()
}
/// The raw `session/update` stream of a structured session (#134),
/// handed out once; `None` for a terminal session, a session whose
/// adapter never came up, or a stream already taken. Notifications that
/// arrived before the take are still queued. #135 takes it to feed the
/// per-session event log.
pub async fn take_acp_updates(
&self,
id: Uuid,
) -> Option<tokio::sync::mpsc::UnboundedReceiver<maestro_acp::schema::SessionNotification>>
{
let sessions = self.inner.sessions.lock().await;
sessions.get(&id)?.acp.as_ref()?.take_updates()
}
/// End every adapter child before the daemon exits (#134): each gets
/// `session/cancel`, then SIGTERM, concurrently. The sessions stay in
/// the map, marked Dead and written through to their sidecars, so what
/// is on disk after the restart is truthful. tmux sessions are left
/// alone — surviving a restart is what they are for.
pub async fn shutdown(&self) {
let children: Vec<(Uuid, Arc<AcpSession>)> = {
let sessions = self.inner.sessions.lock().await;
sessions
.iter()
.filter(|(_, r)| r.session.status == SessionStatus::Running)
.filter_map(|(id, r)| r.acp.clone().map(|acp| (*id, acp)))
.collect()
};
if children.is_empty() {
return;
}
tracing::info!(count = children.len(), "shutdown: ending adapter children");
let terminations = children
.iter()
.map(|(_, acp)| acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE));
futures_util::future::join_all(terminations).await;
for (id, _) in children {
mark_session_dead(&self.inner, id).await;
}
}
/// Subscribe to a session's PTY broadcast. `None` if the session is
/// unknown to this daemon.
pub async fn subscribe_pty(&self, id: Uuid) -> Option<broadcast::Receiver<Vec<u8>>> {
@@ -365,7 +317,7 @@ impl SessionManager {
// replayed from it renders as garbage until something forces a
// repaint. A dead session has no pane left to capture, so it
// keeps the byte tail.
let alive = rec.session.status == SessionStatus::Running && rec.acp.is_none();
let alive = rec.session.status == SessionStatus::Running;
// The capture shells out to tmux while the replay lock is held,
// which parks the reader for the length of one tmux round-trip.
// That only happens on attach, and the alternative is a seam in
@@ -475,48 +427,20 @@ impl SessionManager {
{
session_env.insert("MAESTRO_URL".into(), url);
}
// A structured session (#134) gets no tmux pane: the agent is a
// child of the daemon, driven over ACP. An adapter that cannot come
// up still yields a session — Dead, with the reason on it — so the
// operator sees what to fix instead of a silently missing session.
let (acp, acp_failure) = match req.view {
SessionView::Structured => {
match maestro_acp::spawn(SpawnConfig::new(
&effective_cwd,
session_env
.iter()
.map(|(k, v)| (k.clone(), v.clone()))
.collect(),
))
.await
{
Ok(acp) => (Some(Arc::new(acp)), None),
Err(e) => {
tracing::warn!(%id, error = %e, "structured session: adapter failed to start");
(None, Some(e.to_string()))
}
}
}
SessionView::Terminal => {
let spawned = {
let tmux = self.inner.tmux.clone();
let cwd = effective_cwd.clone();
let command = req.command.clone();
let sink = pipe_pane_command(&fifo_path);
let env = session_env.clone();
tmux_blocking(move || tmux.new_session_piped(id, &cwd, &command, &env, &sink))
.await
};
if let Err(e) = spawned {
// `bridge` drops at the end of this block, closing the
// FIFO reader.
let _ = std::fs::remove_file(&fifo_path);
self.rollback_worktree(&worktree_base_repo, &worktree_path);
return Err(SessionError::Tmux(e));
}
(None, None)
}
let spawned = {
let tmux = self.inner.tmux.clone();
let cwd = effective_cwd.clone();
let command = req.command.clone();
let sink = pipe_pane_command(&fifo_path);
let env = session_env.clone();
tmux_blocking(move || tmux.new_session_piped(id, &cwd, &command, &env, &sink)).await
};
if let Err(e) = spawned {
// `bridge` drops at the end of this block, closing the FIFO reader.
let _ = std::fs::remove_file(&fifo_path);
self.rollback_worktree(&worktree_base_repo, &worktree_path);
return Err(SessionError::Tmux(e));
}
// Capture spawn-history inputs before `req` is partially moved into
// `session`. `history_cwd` is the *chosen* working dir (the request
@@ -539,21 +463,13 @@ impl SessionManager {
base_repo: base_repo.clone(),
},
),
status: if acp_failure.is_some() {
SessionStatus::Dead
} else {
SessionStatus::Running
},
status: SessionStatus::Running,
created_at: Utc::now(),
// A freshly-spawned session is producing its initial output; the
// activity watcher refines `foreground`/`activity` on its next tick.
foreground: None,
activity: if acp_failure.is_some() {
ActivityState::Error
} else {
ActivityState::Active
},
activity_detail: acp_failure,
activity: ActivityState::Active,
activity_detail: None,
last_activity_at: Utc::now(),
activity_since: chrono::Utc::now(),
@@ -562,23 +478,13 @@ impl SessionManager {
// fresh tree has nothing in the way anyway.
blocked_work: None,
last_focused_at: None,
acp_session_id: acp.as_ref().map(|a| a.session_id().to_string()),
view: req.view,
};
if let Err(e) = self.inner.sidecar.write(&session) {
// The process already started; end it so no ghost survives a
// tmux already spawned the session; kill it so no ghost survives a
// failed sidecar write, then unwind the FIFO + worktree.
match &acp {
Some(acp) => {
acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
.await
}
None => {
let tmux = self.inner.tmux.clone();
let _ = tmux_blocking(move || tmux.kill_session(id)).await;
}
}
let tmux = self.inner.tmux.clone();
let _ = tmux_blocking(move || tmux.kill_session(id)).await;
let _ = std::fs::remove_file(&fifo_path);
self.rollback_worktree(&worktree_base_repo, &worktree_path);
return Err(SessionError::Sidecar(e));
@@ -600,7 +506,6 @@ impl SessionManager {
naming: false,
focused_since: None,
awaiting_suppressed: false,
acp,
},
);
}
@@ -670,9 +575,7 @@ impl SessionManager {
.worktree_base_repo
.clone()
.zip(record.worktree_path.clone()),
// A structured session has no pane to kill; its child is
// ended through the ACP handle in `cleanup_record`.
record.session.status == SessionStatus::Running && record.acp.is_none(),
record.session.status == SessionStatus::Running,
),
}
};
@@ -780,10 +683,6 @@ impl SessionManager {
current_cwd: None,
blocked_work: None,
last_focused_at: None,
acp_session_id: None,
// Unknown on adoption without a sidecar; matches every
// other field's forward-compat default.
view: maestro_protocol::SessionView::Terminal,
},
};
let fifo_path = self.inner.fifo_dir.join(format!("{id}.fifo"));
@@ -815,9 +714,6 @@ impl SessionManager {
naming: false,
focused_since: None,
awaiting_suppressed: false,
// A structured session's adapter died with the previous
// daemon; only tmux-backed sessions can be adopted.
acp: None,
},
);
let _ = self
@@ -1187,36 +1083,21 @@ async fn run_liveness_pass(inner: &Arc<Inner>) {
// be competing to start the tmux server).
// Only sessions we still believe are running can transition; a session
// already marked Dead must not re-emit an update on every 2s tick.
// A structured session (#134) is not in tmux at all: its liveness is its
// adapter child, checked through the ACP handle instead.
let (tmux_backed, acp_backed): (Vec<Uuid>, Vec<(Uuid, Arc<AcpSession>)>) = {
let known: Vec<Uuid> = {
let sessions = inner.sessions.lock().await;
if sessions.is_empty() {
return;
}
let running = sessions
sessions
.iter()
.filter(|(_, r)| r.session.status == SessionStatus::Running);
let mut tmux_backed = Vec::new();
let mut acp_backed = Vec::new();
for (id, record) in running {
match &record.acp {
Some(acp) => acp_backed.push((*id, acp.clone())),
None => tmux_backed.push(*id),
}
}
(tmux_backed, acp_backed)
.filter(|(_, r)| r.session.status == SessionStatus::Running)
.map(|(id, _)| *id)
.collect()
};
for (id, acp) in acp_backed {
if acp.has_exited() {
mark_session_dead(inner, id).await;
}
}
if tmux_backed.is_empty() {
if known.is_empty() {
return;
}
let live_ids = {
let tmux = inner.tmux.clone();
match tmux_blocking(move || tmux.list_sessions()).await {
@@ -1229,7 +1110,7 @@ async fn run_liveness_pass(inner: &Arc<Inner>) {
};
let live: HashSet<Uuid> = live_ids.into_iter().collect();
for id in tmux_backed {
for id in known {
if !live.contains(&id) {
mark_session_dead(inner, id).await;
}
@@ -1435,7 +1316,7 @@ async fn mark_session_dead(inner: &Arc<Inner>, id: Uuid) {
let Some((mut session, worktree)) = flipped else {
return;
};
tracing::info!(%id, "session process gone — marking dead");
tracing::info!(%id, "session vanished from tmux — marking dead");
if let Some(wt) = &worktree {
if settle_worktree_on_exit(inner, id, wt).await {
@@ -1537,11 +1418,6 @@ async fn cleanup_record(
let tmux = inner.tmux.clone();
let _ = tmux_blocking(move || tmux.kill_session(id)).await;
}
if let Some(acp) = &record.acp {
// `session/cancel`, a short wait, then SIGTERM (#134).
acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
.await;
}
let _ = std::fs::remove_file(&record.fifo_path);
if let Err(e) = inner.sidecar.delete(id) {
tracing::warn!(error = ?e, %id, "cleanup: failed to delete sidecar");
-296
View File
@@ -1,296 +0,0 @@
//! Structured sessions (#134): a `claude-code-acp` child per session instead
//! of a tmux pane. Runs against the fake adapter in
//! `crates/maestro-acp/tests/fake_acp_agent.py`; nothing here needs Claude.
mod common;
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::time::{Duration, Instant};
use common::{wait_for_async, DaemonFixture};
use maestro_acp::{ADAPTER_BIN, INSTALL_COMMAND};
use maestro_protocol::{ActivityState, CreateSessionRequest, SessionStatus, SessionView};
use tempfile::TempDir;
fn fake_agent() -> PathBuf {
Path::new(env!("CARGO_MANIFEST_DIR")).join("../maestro-acp/tests/fake_acp_agent.py")
}
/// The fake adapter installed as `claude-code-acp` in a temp `bin/`, with
/// the log file it appends to. `env()` is what a spawn request carries.
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()
}
/// Fake `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(),
),
])
}
/// The pid the fake logged at start, once it has.
fn pid(&self) -> u32 {
self.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")
}
}
fn structured(cwd: &Path, env: HashMap<String, String>) -> CreateSessionRequest {
CreateSessionRequest {
view: SessionView::Structured,
cwd: cwd.to_path_buf(),
command: "claude".into(),
env,
metadata: HashMap::new(),
worktree: None,
}
}
fn pid_alive(pid: u32) -> bool {
std::fs::read_to_string(format!("/proc/{pid}/stat"))
.map(|s| !s.contains(") Z "))
.unwrap_or(false)
}
async fn wait_for_exit(pid: u32) {
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), "pid {pid} still alive");
}
#[tokio::test]
async fn structured_spawn_starts_the_adapter_in_cwd_with_session_env() {
let fx = DaemonFixture::new();
let fake = Fake::new();
let cwd = TempDir::new().unwrap();
let mut env = fake.env();
env.insert("FAKE_ACP_ECHO_ENV".into(), "MAESTRO_SESSION_ID".into());
let session = fx
.manager
.create(structured(cwd.path(), env))
.await
.expect("create");
assert_eq!(session.view, SessionView::Structured);
assert_eq!(session.status, SessionStatus::Running);
let acp_id = session.acp_session_id.clone().expect("acp session id");
assert!(acp_id.starts_with("fake-"), "{acp_id}");
// The child ran in the session cwd with the session's env — including
// the identity the daemon injects, so the agent knows which session it
// is (#103).
let log = fake.log();
let cwd_canon = cwd.path().canonicalize().unwrap();
assert!(
log.contains(&format!("cwd={}", cwd_canon.display())),
"{log}"
);
assert!(
log.contains(&format!("env MAESTRO_SESSION_ID={}", session.id)),
"{log}"
);
// No tmux pane was created for it: the agent is the daemon's child.
let panes = fx.tmux.tmux.list_sessions().expect("list");
assert!(
!panes.contains(&session.id),
"structured session got a tmux pane"
);
// The sidecar carries the id a respawn will pass to session/load (#135).
let sidecar: maestro_protocol::Session = serde_json::from_slice(
&std::fs::read(fx.sidecar_dir().join(format!("{}.json", session.id))).unwrap(),
)
.unwrap();
assert_eq!(sidecar.acp_session_id.as_deref(), Some(acp_id.as_str()));
// The raw notification stream is reachable for #135, handed out once.
let mut updates = fx
.manager
.take_acp_updates(session.id)
.await
.expect("updates receiver");
assert!(fx.manager.take_acp_updates(session.id).await.is_none());
let first = tokio::time::timeout(Duration::from_secs(5), updates.recv())
.await
.expect("an update within 5s")
.expect("channel open");
assert_eq!(first.session_id.0.as_ref(), acp_id);
fx.manager.kill(session.id, false).await.expect("kill");
}
#[tokio::test]
async fn missing_adapter_yields_a_failed_session_naming_the_install_command() {
let fx = DaemonFixture::new();
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 session = fx
.manager
.create(structured(cwd.path(), env))
.await
.expect("a failed session, not an error");
assert_eq!(session.status, SessionStatus::Dead);
assert_eq!(session.activity, ActivityState::Error);
let detail = session.activity_detail.clone().expect("actionable detail");
assert!(detail.contains(ADAPTER_BIN), "{detail}");
assert!(detail.contains(INSTALL_COMMAND), "{detail}");
assert_eq!(session.acp_session_id, None);
// It is a real session: listed, on disk, and dismissable like any other
// dead one.
assert!(fx.manager.has(session.id).await);
assert!(fx
.sidecar_dir()
.join(format!("{}.json", session.id))
.exists());
fx.manager.kill(session.id, false).await.expect("dismiss");
assert!(!fx.manager.has(session.id).await);
}
#[tokio::test]
async fn kill_sends_cancel_then_sigterm_and_removes_the_sidecar() {
let fx = DaemonFixture::new();
let fake = Fake::new();
let cwd = TempDir::new().unwrap();
let mut events = fx.manager.events();
let session = fx
.manager
.create(structured(cwd.path(), fake.env()))
.await
.expect("create");
let pid = fake.pid();
assert!(pid_alive(pid));
fx.manager.kill(session.id, false).await.expect("kill");
wait_for_exit(pid).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!(!fx.manager.has(session.id).await);
assert!(!fx
.sidecar_dir()
.join(format!("{}.json", session.id))
.exists());
let mut removed = false;
while let Ok(ev) = events.try_recv() {
if matches!(ev, maestro_protocol::CatalogEvent::SessionRemoved { id } if id == session.id) {
removed = true;
}
}
assert!(removed, "SessionRemoved never broadcast");
}
#[tokio::test]
async fn shutdown_reaps_every_adapter_child() {
let fx = DaemonFixture::new();
let cwd = TempDir::new().unwrap();
let a = Fake::new();
let b = Fake::new();
let sa = fx
.manager
.create(structured(cwd.path(), a.env()))
.await
.expect("create a");
let sb = fx
.manager
.create(structured(cwd.path(), b.env()))
.await
.expect("create b");
let (pa, pb) = (a.pid(), b.pid());
assert!(pid_alive(pa) && pid_alive(pb));
fx.manager.shutdown().await;
wait_for_exit(pa).await;
wait_for_exit(pb).await;
assert!(a.log().contains("signal SIGTERM"), "{}", a.log());
assert!(b.log().contains("signal SIGTERM"), "{}", b.log());
// Still listed, now Dead: the sidecar is truthful for whoever reads it
// after the restart (#135).
assert_eq!(fx.manager.status(sa.id).await, Some(SessionStatus::Dead));
assert_eq!(fx.manager.status(sb.id).await, Some(SessionStatus::Dead));
let sidecar: maestro_protocol::Session = serde_json::from_slice(
&std::fs::read(fx.sidecar_dir().join(format!("{}.json", sa.id))).unwrap(),
)
.unwrap();
assert_eq!(sidecar.status, SessionStatus::Dead);
}
#[tokio::test]
async fn adapter_exit_marks_the_session_dead_and_keeps_it_listed() {
let fx = DaemonFixture::new();
let fake = Fake::new();
let cwd = TempDir::new().unwrap();
let session = fx
.manager
.create(structured(cwd.path(), fake.env()))
.await
.expect("create");
fx.manager.start_liveness_watcher(Duration::from_millis(50));
// The agent dies out from under the daemon: the liveness pass, not tmux,
// is what notices for a structured session.
let pid = fake.pid();
unsafe {
libc::kill(pid as libc::pid_t, libc::SIGTERM);
}
wait_for_exit(pid).await;
wait_for_async(Duration::from_secs(3), || async {
fx.manager.status(session.id).await == Some(SessionStatus::Dead)
})
.await;
assert!(fx.manager.has(session.id).await);
fx.manager.kill(session.id, false).await.expect("dismiss");
assert!(!fx.manager.has(session.id).await);
}
@@ -42,7 +42,6 @@ async fn fixture(ttl: Duration) -> Fixture {
async fn spawn_sleeping(manager: &SessionManager) -> Uuid {
manager
.create(CreateSessionRequest {
view: Default::default(),
cwd: "/tmp".into(),
command: "sleep 60".into(),
env: Default::default(),
-1
View File
@@ -75,7 +75,6 @@ async fn create_bash_session(h: &Harness) -> Uuid {
let s = h
.manager
.create(CreateSessionRequest {
view: Default::default(),
cwd: cwd_path,
command: "bash --noprofile --norc".into(),
env: HashMap::new(),
-1
View File
@@ -50,7 +50,6 @@ async fn spawn(manager: &SessionManager, name: Option<&str>) -> Uuid {
};
manager
.create(CreateSessionRequest {
view: Default::default(),
cwd: "/tmp".into(),
command: "sleep 60".into(),
env: Default::default(),
@@ -19,7 +19,6 @@ use tempfile::TempDir;
fn request(cwd: &std::path::Path, repo: &std::path::Path, branch: &str) -> CreateSessionRequest {
CreateSessionRequest {
view: Default::default(),
cwd: cwd.to_path_buf(),
command: "sleep 30".into(),
env: HashMap::new(),
-10
View File
@@ -174,7 +174,6 @@ async fn list_sessions_starts_empty_and_reflects_creates() {
assert!(empty.is_empty());
let req = CreateSessionRequest {
view: Default::default(),
cwd: "/tmp".into(),
command: "sleep 60".into(),
env: HashMap::new(),
@@ -391,7 +390,6 @@ async fn patch_session_renames_it() {
let session = h
.manager
.create(CreateSessionRequest {
view: Default::default(),
cwd: "/tmp".into(),
command: "sleep 60".into(),
env: HashMap::new(),
@@ -434,7 +432,6 @@ async fn post_focus_stamps_last_focused_at() {
let session = h
.manager
.create(CreateSessionRequest {
view: Default::default(),
cwd: "/tmp".into(),
command: "sleep 60".into(),
env: HashMap::new(),
@@ -468,7 +465,6 @@ async fn delete_owned_session_kills_it() {
let session = h
.manager
.create(CreateSessionRequest {
view: Default::default(),
cwd: "/tmp".into(),
command: "sleep 60".into(),
env: HashMap::new(),
@@ -490,7 +486,6 @@ async fn delete_owned_session_kills_it() {
async fn spawn_via_manager(h: &Harness) -> Uuid {
h.manager
.create(CreateSessionRequest {
view: Default::default(),
cwd: "/tmp".into(),
command: "sleep 60".into(),
env: HashMap::new(),
@@ -600,7 +595,6 @@ async fn delete_a_session_whose_worktree_holds_unsaved_work_returns_409_with_the
let session = h
.manager
.create(CreateSessionRequest {
view: Default::default(),
cwd: repo.path().to_path_buf(),
command: "sleep 60".into(),
env: HashMap::new(),
@@ -653,7 +647,6 @@ async fn delete_with_force_ends_the_session_and_keeps_the_worktree() {
let session = h
.manager
.create(CreateSessionRequest {
view: Default::default(),
cwd: repo.path().to_path_buf(),
command: "sleep 60".into(),
env: HashMap::new(),
@@ -687,7 +680,6 @@ async fn attach_ws_streams_pty_bytes() {
let session = h
.manager
.create(CreateSessionRequest {
view: Default::default(),
cwd: "/tmp".into(),
command: "bash --noprofile --norc".into(),
env: HashMap::new(),
@@ -756,7 +748,6 @@ async fn events_ws_sends_a_snapshot_then_streams_deltas() {
let created = h
.manager
.create(CreateSessionRequest {
view: Default::default(),
cwd: cwd.path().to_path_buf(),
command: "sleep 30".into(),
env: HashMap::new(),
@@ -801,7 +792,6 @@ async fn every_events_subscriber_gets_its_own_snapshot() {
h.manager
.create(CreateSessionRequest {
view: Default::default(),
cwd: cwd.path().to_path_buf(),
command: "sleep 30".into(),
env: HashMap::new(),
-1
View File
@@ -51,7 +51,6 @@ async fn spawn_named(manager: &SessionManager, name: Option<&str>) -> Uuid {
};
manager
.create(CreateSessionRequest {
view: Default::default(),
cwd: "/tmp".into(),
command: "sleep 60".into(),
env: Default::default(),
@@ -64,7 +64,6 @@ impl Fx {
fn request(&self, command: &str) -> CreateSessionRequest {
CreateSessionRequest {
view: Default::default(),
cwd: self.tmp.path().to_path_buf(),
command: command.into(),
env: HashMap::new(),
@@ -22,7 +22,6 @@ async fn create_then_kill_round_trip() {
let session = fx
.manager
.create(CreateSessionRequest {
view: Default::default(),
cwd: cwd.path().to_path_buf(),
command: "sleep 30".into(),
env: HashMap::new(),
@@ -64,7 +63,6 @@ async fn create_appends_to_spawn_history() {
let session = fx
.manager
.create(CreateSessionRequest {
view: Default::default(),
cwd: cwd.path().to_path_buf(),
command: "sleep 30".into(),
env: HashMap::new(),
@@ -94,7 +92,6 @@ async fn create_without_history_logs_nothing() {
let cwd = TempDir::new().unwrap();
fx.manager
.create(CreateSessionRequest {
view: Default::default(),
cwd: cwd.path().to_path_buf(),
command: "sleep 30".into(),
env: HashMap::new(),
@@ -117,7 +114,6 @@ async fn recover_adopts_existing_tmux_sessions() {
let s = fx
.manager
.create(CreateSessionRequest {
view: Default::default(),
cwd: cwd.path().to_path_buf(),
command: "sleep 30".into(),
env: HashMap::new(),
@@ -157,7 +153,6 @@ async fn create_captures_initial_output_in_replay_buffer() {
let s = fx
.manager
.create(CreateSessionRequest {
view: Default::default(),
cwd: cwd.path().to_path_buf(),
// The shell prints the marker before doing anything else and
// then sleeps so the session stays alive. Before NAP-253 this
@@ -205,7 +200,6 @@ async fn attach_replay_is_the_current_screen_not_the_byte_tail() {
let s = fx
.manager
.create(CreateSessionRequest {
view: Default::default(),
cwd: cwd.path().to_path_buf(),
// Print a line, then switch to the alternate screen the way a
// full-screen program does on startup. Only the second marker is
@@ -264,7 +258,6 @@ async fn attach_replay_carries_the_panes_mouse_modes() {
let s = fx
.manager
.create(CreateSessionRequest {
view: Default::default(),
cwd: cwd.path().to_path_buf(),
// Alternate screen, then button-event tracking (?1002) with SGR
// coordinates (?1006) — what tmux and agent TUIs actually request.
@@ -318,7 +311,6 @@ async fn recover_re_pipes_pane_so_output_reaches_subscribers() {
let s = fx
.manager
.create(CreateSessionRequest {
view: Default::default(),
cwd: cwd.path().to_path_buf(),
command: "bash --noprofile --norc".into(),
env: HashMap::new(),
@@ -381,7 +373,6 @@ async fn pty_broadcast_delivers_bytes_to_subscribers() {
let s = fx
.manager
.create(CreateSessionRequest {
view: Default::default(),
cwd: cwd.path().to_path_buf(),
command: "bash --noprofile --norc".into(),
env: HashMap::new(),
@@ -432,7 +423,6 @@ async fn liveness_watcher_marks_dead_sessions_and_keeps_them_until_dismissed() {
let s = fx
.manager
.create(CreateSessionRequest {
view: Default::default(),
cwd: cwd.path().to_path_buf(),
command: "sleep 30".into(),
env: HashMap::new(),
@@ -498,7 +488,6 @@ async fn watcher_refreshes_the_blocked_work_indicator() {
fx.manager
.create(CreateSessionRequest {
view: Default::default(),
cwd: repo.path().to_path_buf(),
command: "sleep 30".into(),
env: HashMap::new(),
@@ -591,7 +580,6 @@ async fn kill_removes_worktree_even_when_spec_says_keep() {
let s = fx
.manager
.create(CreateSessionRequest {
view: Default::default(),
cwd: repo.path().to_path_buf(),
command: "sleep 30".into(),
env: HashMap::new(),
@@ -629,7 +617,6 @@ async fn killing_refuses_a_worktree_with_unsaved_work_and_keeps_the_session() {
let s = fx
.manager
.create(CreateSessionRequest {
view: Default::default(),
cwd: repo.path().to_path_buf(),
command: "sleep 30".into(),
env: HashMap::new(),
@@ -693,7 +680,6 @@ async fn killing_refuses_until_the_branch_is_pushed() {
let s = fx
.manager
.create(CreateSessionRequest {
view: Default::default(),
cwd: repo.path().to_path_buf(),
command: "sleep 30".into(),
env: HashMap::new(),
@@ -743,7 +729,6 @@ async fn force_killing_a_guarded_worktree_leaves_it_on_disk() {
let s = fx
.manager
.create(CreateSessionRequest {
view: Default::default(),
cwd: repo.path().to_path_buf(),
command: "sleep 30".into(),
env: HashMap::new(),
@@ -789,7 +774,6 @@ async fn force_on_a_clear_worktree_removes_it_as_usual() {
let s = fx
.manager
.create(CreateSessionRequest {
view: Default::default(),
cwd: repo.path().to_path_buf(),
command: "sleep 30".into(),
env: HashMap::new(),
@@ -830,7 +814,6 @@ async fn killing_an_ignored_only_worktree_succeeds() {
let s = fx
.manager
.create(CreateSessionRequest {
view: Default::default(),
cwd: repo.path().to_path_buf(),
command: "sleep 30".into(),
env: HashMap::new(),
@@ -872,7 +855,6 @@ async fn natural_exit_of_a_guarded_worktree_leaves_it_on_disk() {
let s = fx
.manager
.create(CreateSessionRequest {
view: Default::default(),
cwd: repo.path().to_path_buf(),
command: "sleep 30".into(),
env: HashMap::new(),
@@ -934,7 +916,6 @@ async fn natural_exit_of_a_pristine_worktree_removes_it() {
let s = fx
.manager
.create(CreateSessionRequest {
view: Default::default(),
cwd: repo.path().to_path_buf(),
command: "sleep 30".into(),
env: HashMap::new(),
@@ -995,7 +976,6 @@ async fn liveness_watcher_leaves_running_sessions_alone() {
let s = fx
.manager
.create(CreateSessionRequest {
view: Default::default(),
cwd: cwd.path().to_path_buf(),
command: "sleep 30".into(),
env: HashMap::new(),
@@ -1030,7 +1010,6 @@ async fn worktree_base_repo_records_the_main_checkout_not_the_request() {
let first = fx
.manager
.create(CreateSessionRequest {
view: Default::default(),
cwd: repo.path().to_path_buf(),
command: "sleep 30".into(),
env: HashMap::new(),
@@ -1048,7 +1027,6 @@ async fn worktree_base_repo_records_the_main_checkout_not_the_request() {
let second = fx
.manager
.create(CreateSessionRequest {
view: Default::default(),
cwd: inner_worktree.clone(),
command: "sleep 30".into(),
env: HashMap::new(),
@@ -1087,7 +1065,6 @@ async fn spawned_sessions_export_maestro_session_id_and_url() {
let s = fx
.manager
.create(CreateSessionRequest {
view: Default::default(),
cwd: cwd.path().to_path_buf(),
command: "sleep 30".into(),
env: HashMap::new(),
@@ -1126,7 +1103,6 @@ async fn a_freshly_created_session_has_never_been_focused() {
let s = fx
.manager
.create(CreateSessionRequest {
view: Default::default(),
cwd: cwd.path().to_path_buf(),
command: "sleep 30".into(),
env: HashMap::new(),
@@ -1152,7 +1128,6 @@ async fn set_focused_true_stamps_last_focused_at() {
let s = fx
.manager
.create(CreateSessionRequest {
view: Default::default(),
cwd: cwd.path().to_path_buf(),
command: "sleep 30".into(),
env: HashMap::new(),
@@ -1201,7 +1176,6 @@ async fn attaching_stamps_last_focused_at() {
let s = fx
.manager
.create(CreateSessionRequest {
view: Default::default(),
cwd: cwd.path().to_path_buf(),
command: "sleep 30".into(),
env: HashMap::new(),
+1 -34
View File
@@ -7,7 +7,7 @@ use std::path::PathBuf;
use chrono::{DateTime, Utc};
use maestro_daemon::sidecar::SidecarStore;
use maestro_protocol::{Session, SessionStatus, SessionView};
use maestro_protocol::{Session, SessionStatus};
use serde_json::Value;
use tempfile::TempDir;
use uuid::Uuid;
@@ -32,8 +32,6 @@ fn sample(id: Uuid) -> Session {
current_cwd: Some(PathBuf::from("/tmp/work/crates")),
blocked_work: None,
last_focused_at: None,
acp_session_id: None,
view: SessionView::Terminal,
}
}
@@ -105,37 +103,6 @@ fn load_all_returns_empty_when_directory_missing() {
assert!(store.load_all().unwrap().is_empty());
}
#[test]
fn sidecar_missing_view_key_reads_as_terminal() {
// A pre-#133 sidecar on disk has no `view` field at all.
let dir = TempDir::new().unwrap();
let store = SidecarStore::new(dir.path());
let id = Uuid::new_v4();
let mut value = serde_json::to_value(sample(id)).unwrap();
value.as_object_mut().unwrap().remove("view");
std::fs::write(store.path_for(id), serde_json::to_vec(&value).unwrap()).unwrap();
let loaded = store.read(id).expect("read").expect("present");
assert_eq!(loaded.view, SessionView::Terminal);
}
#[test]
fn sidecar_structured_view_round_trips() {
let dir = TempDir::new().unwrap();
let store = SidecarStore::new(dir.path());
let id = Uuid::new_v4();
let mut s = sample(id);
s.view = SessionView::Structured;
store.write(&s).expect("write");
let loaded = store.read(id).expect("read");
assert_eq!(
loaded.as_ref().map(|s| s.view),
Some(SessionView::Structured)
);
}
#[test]
fn write_creates_parent_directory() {
let dir = TempDir::new().unwrap();
+16 -6
View File
@@ -405,15 +405,17 @@ fn run_git(dir: &std::path::Path, args: &[&str]) {
);
}
/// 49ab7e9: a new-branch add onto a name that already exists suffixes the
/// branch (and so the directory) instead of failing, per docs/WORKTREES.md.
/// `git worktree add -b` alone fails outright when the branch already exists,
/// even with the target directory free — a random handle commonly collides
/// with one an earlier, already-removed worktree left behind. `add_new`
/// sidesteps it on the branch name itself instead of erroring (#77).
#[test]
fn add_new_suffixes_when_branch_already_exists() {
fn add_new_suffixes_branch_name_on_branch_collision() {
let repo_dir = TempDir::new().unwrap();
let repo = repo_dir.path().to_path_buf();
init_repo(&repo);
// pre-create the branch
// pre-create the branch, unused by any worktree.
let status = Command::new("git")
.arg("-C")
.arg(&repo)
@@ -425,9 +427,17 @@ fn add_new_suffixes_when_branch_already_exists() {
let wt_root = TempDir::new().unwrap();
let mgr = WorktreeManager::new(Some(wt_root.path().to_path_buf()));
let path = mgr.add_new(&repo, "feat/dup").unwrap();
let path = mgr
.add_new(&repo, "feat/dup")
.expect("sidesteps the branch collision");
assert!(path.ends_with("feat/dup-2"), "{}", path.display());
assert!(path.is_dir());
assert!(path.exists());
// The pre-existing branch is untouched; the new worktree checked out the
// suffixed one instead of erroring or adopting `feat/dup`.
let branches = worktree_branches(&repo);
assert!(branches.contains(&"feat/dup-2".to_string()), "{branches:?}");
assert!(!branches.contains(&"feat/dup".to_string()), "{branches:?}");
}
/// The project picker lists linked worktrees as pickable rows, so `base_repo`
+32 -35
View File
@@ -27,7 +27,6 @@ async fn bring_up() -> (Daemon, TestClient) {
fn create_req(cwd: &std::path::Path, command: &str) -> CreateSessionRequest {
CreateSessionRequest {
view: Default::default(),
cwd: cwd.to_path_buf(),
command: command.into(),
env: HashMap::new(),
@@ -50,7 +49,6 @@ fn worktree_spec(base_repo: &std::path::Path, branch: &str, create_branch: bool)
/// Spawn request carrying a worktree spec against `base`.
fn worktree_req(base: &std::path::Path, branch: &str, create_branch: bool) -> CreateSessionRequest {
CreateSessionRequest {
view: Default::default(),
cwd: base.to_path_buf(),
command: "sleep 30".into(),
env: HashMap::new(),
@@ -145,7 +143,6 @@ async fn kill_removes_clean_worktree() {
init_repo(base.path());
let req = CreateSessionRequest {
view: Default::default(),
cwd: base.path().to_path_buf(),
command: "sleep 30".into(),
env: HashMap::new(),
@@ -199,7 +196,6 @@ async fn kill_refuses_worktree_with_uncommitted_changes() {
.join("unsaved");
let req = CreateSessionRequest {
view: Default::default(),
cwd: base.path().to_path_buf(),
command: "sleep 30".into(),
env: HashMap::new(),
@@ -258,7 +254,6 @@ async fn spawn_on_existing_branch_checks_it_out() {
run_git(&["branch", "feature/existing"], base.path());
let req = CreateSessionRequest {
view: Default::default(),
cwd: base.path().to_path_buf(),
command: "sleep 30".into(),
env: HashMap::new(),
@@ -328,13 +323,11 @@ async fn spawn_new_branch_creates_worktree_and_runs_in_it() {
client.kill_session(s.id).await.expect("kill");
}
/// #66: a second spawn onto an *existing* branch that already has a worktree
/// is rejected with a structured worktree error; the first worktree stays
/// intact. (A second `create_branch: true` spawn is a different case: 49ab7e9
/// suffixes the new branch name instead, see
/// `colliding_branch_name_gets_suffixed_too`.)
/// #66/#77: a second spawn onto the same branch used to be rejected outright;
/// since the branch-collision suffixing landed, it succeeds instead — the
/// second worktree lands on `<branch>-2` and the first stays untouched.
#[tokio::test(flavor = "multi_thread")]
async fn second_spawn_on_same_branch_is_rejected() {
async fn second_spawn_on_same_branch_gets_suffixed() {
let (_daemon, client) = bring_up().await;
let base = TempDir::new().unwrap();
@@ -344,28 +337,33 @@ async fn second_spawn_on_same_branch_is_rejected() {
.spawn_session(&worktree_req(base.path(), "feature/dup", true))
.await
.expect("first spawn");
let dup = client
.spawn_session(&worktree_req(base.path(), "feature/dup", false))
.await;
match dup {
Err(maestro_e2e::ClientError::Api(maestro_protocol::ClientError::Api { info, .. })) => {
assert_eq!(
info.code, "worktree_failed",
"expected worktree error: {info:?}"
);
}
other => panic!("expected rejection for duplicate branch, got {other:?}"),
}
let second = client
.spawn_session(&worktree_req(base.path(), "feature/dup", true))
.await
.expect("second spawn suffixes instead of rejecting");
// Exactly one registration, one directory, still usable.
// Count branch records only: the worktree path now contains the branch
// name too (`worktrees/feature/dup`), so a substring match over every line
// would count the same tree twice.
let holders = git_worktree_list(base.path())
.iter()
.filter(|w| w.as_str() == "branch refs/heads/feature/dup")
.count();
assert_eq!(holders, 1, "duplicate spawn must not register another tree");
let suffixed = base.path().join("worktrees").join("feature").join("dup-2");
assert_eq!(
second.cwd, suffixed,
"second worktree must land on the suffixed path"
);
assert!(suffixed.is_dir());
// Both registrations exist: the original branch, untouched, and the
// suffixed one the second spawn created.
let registrations = git_worktree_list(base.path());
assert!(
registrations
.iter()
.any(|l| l.as_str() == "branch refs/heads/feature/dup"),
"original branch must stay registered: {registrations:?}"
);
assert!(
registrations
.iter()
.any(|l| l.as_str() == "branch refs/heads/feature/dup-2"),
"suffixed branch must be registered too: {registrations:?}"
);
assert!(
base.path()
.join("worktrees")
@@ -376,7 +374,8 @@ async fn second_spawn_on_same_branch_is_rejected() {
);
let _origin = throwaway_origin(base.path());
client.kill_session(first.id).await.expect("kill");
client.kill_session(first.id).await.expect("kill first");
client.kill_session(second.id).await.expect("kill second");
}
/// #66/#56: when the client picks a *linked worktree* as base_repo, placement
@@ -762,7 +761,6 @@ async fn natural_exit_settles_dirty_and_pristine_worktrees_differently() {
let _dirty = client
.spawn_session(&CreateSessionRequest {
view: Default::default(),
cwd: dirty_repo.path().to_path_buf(),
command: "sh -c 'echo unsaved > scratch.txt'".into(),
env: HashMap::new(),
@@ -811,7 +809,6 @@ async fn natural_exit_settles_dirty_and_pristine_worktrees_differently() {
fn create_req_cmd(cwd: &std::path::Path, command: &str, branch: &str) -> CreateSessionRequest {
CreateSessionRequest {
view: Default::default(),
cwd: cwd.to_path_buf(),
command: command.into(),
env: HashMap::new(),
-2
View File
@@ -268,7 +268,6 @@ mod tests {
fn session(cwd: &str) -> Session {
Session {
view: Default::default(),
id: Uuid::from_u128(1),
cwd: PathBuf::from(cwd),
command: "bash".into(),
@@ -285,7 +284,6 @@ mod tests {
current_cwd: None,
blocked_work: None,
last_focused_at: None,
acp_session_id: None,
}
}
-78
View File
@@ -116,33 +116,6 @@ pub struct Session {
/// sorts a never-focused session after every session that has one.
#[serde(default)]
pub last_focused_at: Option<DateTime<Utc>>,
/// Whether this session's process is a plain terminal (PTY bytes through
/// tmux) or a structured agent driven over ACP rows (#133). Fixed at
/// spawn time and never changes for the life of the session. Defaults to
/// `Terminal` on the wire so every existing client and payload from a
/// daemon that predates the field keeps working untouched.
#[serde(default)]
pub view: SessionView,
/// The ACP session id the `claude-code-acp` adapter handed back from
/// `session/new` (#134). Only a [`SessionView::Structured`] session has
/// one, and only once its handshake succeeded; a later respawn passes it
/// to `session/load` so the conversation continues (#135). Distinct from
/// the hook-reported Claude Code conversation id (#161): that one is what
/// `claude --resume` takes, this one is what the ACP adapter takes, and
/// the two are minted by different processes with different lifetimes.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub acp_session_id: Option<String>,
}
/// A session's process shape (#133): a plain terminal, or a structured agent
/// driven over ACP rows. Chosen at spawn and immutable for the session's
/// life. `Terminal` is the default — every sidecar and `CreateSessionRequest`
/// written before this field existed meant a terminal session.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
pub enum SessionView {
#[default]
Terminal,
Structured,
}
/// Default for [`Session::last_activity_at`] when an older payload omits it.
@@ -379,11 +352,6 @@ pub struct CreateSessionRequest {
pub metadata: HashMap<String, Value>,
#[serde(default)]
pub worktree: Option<WorktreeSpec>,
/// The process shape to spawn (#133). Defaulted so every existing client
/// and `curl` against the API keeps requesting a `Terminal` session
/// untouched.
#[serde(default)]
pub view: SessionView,
}
/// Result of `GET /api/dirs`: the immediate subdirectories of one directory.
@@ -894,8 +862,6 @@ mod tests {
current_cwd: Some(PathBuf::from("/home/naps62/projects/maestro/crates")),
blocked_work: None,
last_focused_at: None,
view: SessionView::Terminal,
acp_session_id: None,
}
}
@@ -1107,7 +1073,6 @@ mod tests {
branch: "feat/foo".into(),
create_branch: true,
}),
view: SessionView::Terminal,
};
assert_eq!(req, round_trip(&req));
}
@@ -1122,49 +1087,6 @@ mod tests {
assert!(req.worktree.is_none());
assert!(req.env.is_empty());
assert!(req.metadata.is_empty());
assert_eq!(req.view, SessionView::Terminal);
}
#[test]
fn session_view_defaults_to_terminal_on_legacy_payload() {
// A pre-#133 daemon or client omits `view`; it must deserialize to
// `Terminal` rather than failing, and `Structured` round-trips.
let json = serde_json::json!({
"id": Uuid::from_u128(1),
"cwd": "/tmp",
"command": "bash",
"status": "Running",
"created_at": "2026-05-24T12:00:00Z"
});
let s: Session = serde_json::from_value(json).unwrap();
assert_eq!(s.view, SessionView::Terminal);
let mut structured = sample_session();
structured.view = SessionView::Structured;
assert_eq!(structured, round_trip(&structured));
}
#[test]
fn acp_session_id_defaults_to_none_and_round_trips() {
// A payload from before #134 carries no `acp_session_id`; it must
// read as `None`, stay off the wire while unset, and round-trip
// once the adapter has handed one back.
let json = serde_json::json!({
"id": Uuid::from_u128(1),
"cwd": "/tmp",
"command": "bash",
"status": "Running",
"created_at": "2026-05-24T12:00:00Z"
});
let s: Session = serde_json::from_value(json).unwrap();
assert_eq!(s.acp_session_id, None);
let serialized = serde_json::to_value(&s).unwrap();
assert!(serialized.get("acp_session_id").is_none());
let mut structured = sample_session();
structured.view = SessionView::Structured;
structured.acp_session_id = Some("sess_01HZX".into());
assert_eq!(structured, round_trip(&structured));
}
/// A client that predates the new/existing discriminator sends no
-2
View File
@@ -31,7 +31,6 @@ use uuid::Uuid;
/// Build a plausible `Session` for a test to hand to [`MockDaemon::add_session`].
pub fn session(id: Uuid, cwd: &str, command: &str) -> Session {
Session {
view: Default::default(),
id,
cwd: cwd.into(),
command: command.into(),
@@ -48,7 +47,6 @@ pub fn session(id: Uuid, cwd: &str, command: &str) -> Session {
current_cwd: None,
blocked_work: None,
last_focused_at: None,
acp_session_id: None,
}
}
-13
View File
@@ -1882,7 +1882,6 @@ mod tests {
sessions: ids
.iter()
.map(|id| Session {
view: Default::default(),
id: *id,
command: "bash".into(),
cwd: std::path::PathBuf::from("/"),
@@ -1899,7 +1898,6 @@ mod tests {
current_cwd: None,
blocked_work: None,
last_focused_at: None,
acp_session_id: None,
})
.collect(),
});
@@ -2065,7 +2063,6 @@ mod tests {
let mut app = make_app();
app.catalog.apply_event(CatalogEvent::Snapshot {
sessions: vec![Session {
view: Default::default(),
id: sid,
command: "bash".into(),
cwd: std::path::PathBuf::from("/"),
@@ -2082,7 +2079,6 @@ mod tests {
current_cwd: None,
blocked_work: None,
last_focused_at: None,
acp_session_id: None,
}],
});
app.dispatch(Action::SidebarNext);
@@ -2094,7 +2090,6 @@ mod tests {
use chrono::Utc;
use maestro_protocol::{Session, SessionStatus};
Session {
view: Default::default(),
id: sid,
command: "bash".into(),
cwd: cwd.into(),
@@ -2111,7 +2106,6 @@ mod tests {
current_cwd: None,
blocked_work: None,
last_focused_at: None,
acp_session_id: None,
}
}
@@ -2783,7 +2777,6 @@ mod tests {
);
app.dispatch(Action::OpenSpawnModal);
let req = CreateSessionRequest {
view: Default::default(),
cwd: std::path::PathBuf::from("/srv/work"),
command: "bash".into(),
env: Default::default(),
@@ -3202,7 +3195,6 @@ mod tests {
use chrono::Utc;
use maestro_protocol::{CatalogEvent, Session, SessionStatus};
let make = |n: u128, name: &str, focused: Option<chrono::DateTime<Utc>>| Session {
view: Default::default(),
id: Uuid::from_u128(n),
command: "bash".into(),
cwd: std::path::PathBuf::from("/"),
@@ -3222,7 +3214,6 @@ mod tests {
current_cwd: None,
blocked_work: None,
last_focused_at: focused,
acp_session_id: None,
};
let older = Utc::now() - chrono::Duration::seconds(60);
app.catalog.apply_event(CatalogEvent::Snapshot {
@@ -3484,7 +3475,6 @@ mod tests {
let sid = Uuid::from_u128(7);
app.catalog.apply_event(CatalogEvent::Snapshot {
sessions: vec![Session {
view: Default::default(),
id: sid,
command: "bash".into(),
cwd: std::path::PathBuf::from("/"),
@@ -3502,7 +3492,6 @@ mod tests {
current_cwd: None,
blocked_work: None,
last_focused_at: None,
acp_session_id: None,
}],
});
app.dispatch(Action::KillHighlighted);
@@ -3522,7 +3511,6 @@ mod tests {
};
app.catalog.apply_event(CatalogEvent::Snapshot {
sessions: vec![Session {
view: Default::default(),
id: sid,
command: "bash".into(),
cwd: std::path::PathBuf::from("/"),
@@ -3539,7 +3527,6 @@ mod tests {
current_cwd: None,
blocked_work: None,
last_focused_at: None,
acp_session_id: None,
}],
});
}
-2
View File
@@ -207,7 +207,6 @@ mod tests {
fn session(id: u128, command: &str) -> Session {
Session {
view: Default::default(),
id: Uuid::from_u128(id),
command: command.into(),
cwd: std::path::PathBuf::from("/tmp"),
@@ -224,7 +223,6 @@ mod tests {
current_cwd: None,
blocked_work: None,
last_focused_at: None,
acp_session_id: None,
}
}
-2
View File
@@ -793,7 +793,6 @@ mod tests {
fn session(n: u128, command: &str) -> Session {
Session {
view: Default::default(),
id: Uuid::from_u128(n),
cwd: PathBuf::from("/home/me/dev/api"),
command: command.to_string(),
@@ -810,7 +809,6 @@ mod tests {
current_cwd: None,
blocked_work: None,
last_focused_at: None,
acp_session_id: None,
}
}
-1
View File
@@ -694,7 +694,6 @@ impl DraftPane {
}
let worktree = self.build_worktree()?;
Ok(CreateSessionRequest {
view: Default::default(),
cwd,
command: command.to_string(),
env: Default::default(),
-1
View File
@@ -108,7 +108,6 @@ impl FastForwardDefaults {
return None;
}
Some(CreateSessionRequest {
view: Default::default(),
cwd: project.path.clone(),
command: command.to_string(),
env: Default::default(),
-4
View File
@@ -484,7 +484,6 @@ mod tests {
let mut metadata = HashMap::new();
metadata.insert("name".into(), serde_json::Value::String(name.into()));
Session {
view: Default::default(),
id: Uuid::from_u128(id),
command: "bash".into(),
cwd: std::path::PathBuf::from("/tmp"),
@@ -501,7 +500,6 @@ mod tests {
current_cwd: None,
blocked_work: None,
last_focused_at: None,
acp_session_id: None,
}
}
@@ -1252,7 +1250,6 @@ mod tests {
#[test]
fn falls_back_to_command_when_name_metadata_is_absent() {
let s = Session {
view: Default::default(),
id: Uuid::from_u128(1),
command: "vim".into(),
cwd: std::path::PathBuf::from("/tmp"),
@@ -1269,7 +1266,6 @@ mod tests {
current_cwd: None,
blocked_work: None,
last_focused_at: None,
acp_session_id: None,
};
let sb = Sidebar {
sessions: vec![s],
+3
View File
@@ -29,3 +29,6 @@ tracing-subscriber.workspace = true
maestro-cli = { path = "../maestro-cli", optional = true, default-features = false }
maestro-daemon = { path = "../maestro-daemon", optional = true, default-features = false }
maestro-tui = { path = "../maestro-tui", optional = true }
[dev-dependencies]
tempfile.workspace = true
+102
View File
@@ -0,0 +1,102 @@
//! Process-level routing for the umbrella `maestro` binary (#10): each
//! enabled top-level subcommand must reach its own crate's runtime dispatch,
//! not just parse successfully.
//!
//! A malformed `--config` file makes every crate's dispatch fail identically
//! at config load, before any subcommand touches tmux, the network, or a
//! terminal. That lets the assertions tell "clap parsed it but the handler
//! never ran" (a usage error, exit 2) apart from "it dispatched into the real
//! crate" (a `toml decode` runtime error via `main`'s `maestro: {e}` printer,
//! exit 1).
//!
//! The absence tests below only compile when the matching feature is off, so
//! they run under CI's per-feature-subset matrix (`--no-default-features
//! --features cli|daemon|tui`) rather than the default all-features build.
use std::io::Write;
use std::path::PathBuf;
use std::process::{Command, Output};
fn bin() -> PathBuf {
PathBuf::from(env!("CARGO_BIN_EXE_maestro"))
}
fn bad_config() -> tempfile::NamedTempFile {
let mut f = tempfile::NamedTempFile::new().expect("create temp config file");
write!(f, "this is not valid toml {{{{{{").expect("write temp config");
f
}
fn run(args: &[&str]) -> Output {
Command::new(bin())
.args(args)
.output()
.unwrap_or_else(|e| panic!("spawn `maestro {args:?}`: {e}"))
}
/// Real dispatch happened: exit 1, `main`'s `"maestro: {e}"` prefix, and the
/// config crate's error surfacing through — never clap's own usage-error
/// page, which would mean the subcommand never reached its handler.
fn assert_dispatched(out: &Output) {
let stderr = String::from_utf8_lossy(&out.stderr);
assert_eq!(out.status.code(), Some(1), "stderr:\n{stderr}");
assert!(stderr.starts_with("maestro: "), "stderr:\n{stderr}");
assert!(stderr.contains("toml decode"), "stderr:\n{stderr}");
assert!(!stderr.contains("Usage:"), "stderr:\n{stderr}");
}
/// Never reached a handler: clap rejected the subcommand outright because the
/// feature that defines it is off. Only used by the `not(feature = "...")`
/// tests below, which themselves only exist in a single-feature build.
#[cfg(not(all(feature = "cli", feature = "daemon", feature = "tui")))]
fn assert_unrecognized(out: &Output, name: &str) {
let stderr = String::from_utf8_lossy(&out.stderr);
assert_eq!(out.status.code(), Some(2), "stderr:\n{stderr}");
assert!(
stderr.contains("unrecognized subcommand"),
"stderr:\n{stderr}"
);
assert!(stderr.contains(name), "stderr:\n{stderr}");
}
#[cfg(feature = "cli")]
#[test]
fn list_dispatches_into_maestro_cli() {
let cfg = bad_config();
let out = run(&["list", "--config", cfg.path().to_str().unwrap()]);
assert_dispatched(&out);
}
#[cfg(not(feature = "cli"))]
#[test]
fn list_is_unrecognized_without_cli_feature() {
assert_unrecognized(&run(&["list"]), "list");
}
#[cfg(feature = "daemon")]
#[test]
fn daemon_dispatches_into_maestro_daemon() {
let cfg = bad_config();
let out = run(&["daemon", "--config", cfg.path().to_str().unwrap()]);
assert_dispatched(&out);
}
#[cfg(not(feature = "daemon"))]
#[test]
fn daemon_is_unrecognized_without_daemon_feature() {
assert_unrecognized(&run(&["daemon"]), "daemon");
}
#[cfg(feature = "tui")]
#[test]
fn tui_dispatches_into_maestro_tui() {
let cfg = bad_config();
let out = run(&["tui", "--config", cfg.path().to_str().unwrap()]);
assert_dispatched(&out);
}
#[cfg(not(feature = "tui"))]
#[test]
fn tui_is_unrecognized_without_tui_feature() {
assert_unrecognized(&run(&["tui"]), "tui");
}
-23
View File
@@ -188,29 +188,6 @@ An unknown name is rejected as an `[tui].theme` configuration error.
The full action vocabulary is the `Action` enum in
`crates/maestro-tui/src/action.rs`.
## Structured sessions: `claude-code-acp`
A session spawned with `view: Structured` (#134) runs Claude Code as a child
of the daemon over the Agent Client Protocol instead of in a tmux pane. The
daemon resolves the adapter binary, `claude-code-acp`, on its `PATH` at spawn
time. Install it once per host:
```sh
npm install -g --prefix ~/.local @zed-industries/claude-code-acp
```
That puts `claude-code-acp` in `~/.local/bin`, which must be on the daemon's
`PATH` (a systemd user unit does not read your shell profile; set
`Environment=PATH=…` in the unit or install somewhere already on it). The
adapter needs a logged-in Claude Code (`claude /login`).
When the adapter is missing the spawn still creates the session, marked dead
with the install command above as its activity detail, so the sidebar says
what to run. The adapter is a child of the daemon: it ends with a `DELETE
/api/sessions/:id` (`session/cancel`, then `SIGTERM`) and with the daemon
itself. Terminal sessions keep surviving restarts through tmux; structured
ones do not yet.
## The browser UI
The daemon serves the SPA off its own listener. Open
+35 -8
View File
@@ -43,6 +43,10 @@ export interface AttachConnection {
* handling — the leader's literal Ctrl-K escape hatch
* (docs/COMMAND-BAR.md). */
sendInput(data: string): void;
/** Refresh the header's name/command (e.g. after autoname renames the
* session) — the connection outlives mount/unmount, so this is the only
* way an already-open header picks up a new name. */
updateInfo(info: AttachInfo): void;
/** Fully disconnect: close the WS, dispose the terminal, remove the element. */
dispose(): void;
}
@@ -85,21 +89,25 @@ export function createAttachConnection(
el.className = "attach";
// Useful identity over a bare UUID: name · command. The full id stays in the
// tooltip for when you really need it.
const titleHtml = info
? `<span class="at-name">${escapeHtml(info.name)}</span>` +
(info.command && info.command !== info.name ? commandFragment(info.command) : "")
: `session <span class="sid">${escapeHtml(sessionId.slice(0, 8))}</span>`;
const titleAttr = info ? `${info.name} · ${info.command} (${sessionId})` : sessionId;
const titleMarkup = (i?: AttachInfo): { html: string; attr: string } => ({
html: i
? `<span class="at-name">${escapeHtml(i.name)}</span>` +
(i.command && i.command !== i.name ? commandFragment(i.command) : "")
: `session <span class="sid">${escapeHtml(sessionId.slice(0, 8))}</span>`,
attr: i ? `${i.name} · ${i.command} (${sessionId})` : sessionId,
});
const initialTitle = titleMarkup(info);
el.innerHTML = `
<header class="attach-header">
<button type="button" class="btn btn-ghost btn-icon sessions-btn" aria-label="Show sessions" title="Show sessions">☰</button>
<span class="attach-title" title="${escapeHtml(titleAttr)}">${titleHtml}</span>
<span class="attach-title" title="${escapeHtml(initialTitle.attr)}">${initialTitle.html}</span>
<span class="attach-state" data-state="connecting">connecting…</span>
<button class="btn btn-ghost btn-icon disconnect" type="button" aria-label="Disconnect" title="Disconnect this session">⏻</button>
</header>
<div class="attach-body"></div>
`;
const titleEl = el.querySelector(".attach-title") as HTMLSpanElement;
const body = el.querySelector(".attach-body") as HTMLDivElement;
const stateEl = el.querySelector(".attach-state") as HTMLSpanElement;
const disconnectBtn = el.querySelector(".disconnect") as HTMLButtonElement;
@@ -359,6 +367,9 @@ export function createAttachConnection(
let ws: WebSocket;
let reconnectMs = INITIAL_RECONNECT_MS;
let reconnectTimer: number | null = null;
// Snapshot replay starts at row zero, which can leave Codex at its oldest line.
// Follow it once per socket; later scrolling remains under the operator's control.
let awaitingReplay = true;
const sendControl = (msg: ClientControl) => {
if (ws.readyState !== WebSocket.OPEN) return;
@@ -378,6 +389,7 @@ export function createAttachConnection(
};
function connect(): void {
awaitingReplay = true;
ws = new WebSocket(attachWsUrl(sessionId));
ws.binaryType = "arraybuffer";
@@ -398,10 +410,20 @@ export function createAttachConnection(
return;
}
if (ev.data instanceof ArrayBuffer) {
term.write(new Uint8Array(ev.data));
const replay = awaitingReplay;
awaitingReplay = false;
term.write(new Uint8Array(ev.data), () => {
if (replay) term.scrollToBottom();
});
} else if (ev.data instanceof Blob) {
// Some browsers may default to Blob despite binaryType="arraybuffer".
ev.data.arrayBuffer().then((b) => term.write(new Uint8Array(b)));
void ev.data.arrayBuffer().then((b) => {
const replay = awaitingReplay;
awaitingReplay = false;
term.write(new Uint8Array(b), () => {
if (replay) term.scrollToBottom();
});
});
}
};
@@ -499,6 +521,11 @@ export function createAttachConnection(
// The closure above: the same bytes-out path keystrokes and the mobile key
// bar already take.
sendInput,
updateInfo(next: AttachInfo) {
const { html, attr } = titleMarkup(next);
titleEl.innerHTML = html;
titleEl.title = attr;
},
dispose() {
if (disposed) return;
disposed = true;
+1
View File
@@ -16,6 +16,7 @@ vi.mock("./attach.js", () => ({
sendInput(data: string) {
sent.push({ id, data });
},
updateInfo() {},
dispose() {},
}),
}));
+1
View File
@@ -79,6 +79,7 @@ export class ConnectionPool {
if (conn) {
this.cache.delete(id);
this.cache.set(id, conn);
if (info) conn.updateInfo(info);
} else {
conn = createAttachConnection(
id,
+25
View File
@@ -130,6 +130,31 @@ describe("DaemonLink", () => {
expect(link.status().state).toBe("online");
});
it("rechecks immediately when the browser returns from the background", async () => {
const sched = manual();
let reply: DaemonInfo | null = info("aaa", "i1");
const link = new DaemonLink({
fetchInfo: async () => {
if (!reply) throw new Error("down");
return reply;
},
scheduler: sched,
reload: () => {},
});
await link.start();
reply = null;
link.noteDisconnected();
sched.run();
await Promise.resolve();
expect(link.status().state).toBe("down");
reply = info("aaa", "i1");
link.recover();
await Promise.resolve();
expect(link.status().state).toBe("online");
});
it("reloads the page when the daemon comes back on a new commit", async () => {
const sched = manual();
const reload = vi.fn();
+11
View File
@@ -148,6 +148,17 @@ export class DaemonLink {
void this.probe();
}
/**
* Recheck liveness after the browser resumes from the background. Mobile
* browsers may suspend both WebSocket events and timers while a PWA is
* hidden, leaving a stale `down` state after the daemon is already back.
*/
recover(): void {
if (this.state === "updating") return;
this.noteDisconnected();
void this.probe();
}
/**
* One `/api/info` round-trip. Resolves `true` when the daemon answered.
*
+15
View File
@@ -132,6 +132,21 @@ describe("CatalogStore reducer", () => {
});
describe("CatalogStore reconnect", () => {
it("replaces a suspended socket immediately", () => {
const first = new FakeSocket();
const second = new FakeSocket();
const { store } = storeWith([first, second]);
store.start();
store.reconnect();
expect(first.closed).toBe(true);
second.open();
// A late close from the old socket is deliberately ignored.
first.drop();
expect(store.backoff()).toBe(INITIAL_BACKOFF_MS);
});
it("doubles the delay per attempt and caps it", () => {
const sockets = Array.from({ length: 8 }, () => new FakeSocket());
const { store, scheduler } = storeWith(sockets);
+17
View File
@@ -59,6 +59,23 @@ export class CatalogStore {
this.connect();
}
/** Replace a socket that may have gone stale while the browser was suspended. */
reconnect(): void {
if (this.reconnectTimer !== null) {
this.scheduler.clearTimer(this.reconnectTimer);
this.reconnectTimer = null;
}
if (this.ws) {
// A close event from the superseded socket must not schedule a second
// connection alongside the new one.
this.ws.onclose = null;
this.ws.onerror = null;
this.ws.close();
this.ws = null;
}
this.connect();
}
/** Current reconnect delay, for tests asserting the backoff schedule. */
backoff(): number {
return this.backoffMs;
+13
View File
@@ -107,6 +107,19 @@ void link.start();
store.start();
// Installed mobile browsers suspend timers and WebSockets in the background.
// Replacing the socket on return avoids leaving the restart overlay up after a
// transient lost connection has already recovered.
function recoverFromBackground(): void {
if (document.visibilityState !== "visible") return;
link.recover();
store.reconnect();
}
document.addEventListener("visibilitychange", recoverFromBackground);
window.addEventListener("pageshow", recoverFromBackground);
window.addEventListener("online", recoverFromBackground);
// Dev-only: `?mock` seeds a representative catalog so the UI can be reviewed
// without a running daemon. Stripped from production builds by Vite.
if (import.meta.env.DEV && location.search.includes("mock")) {
-4
View File
@@ -56,10 +56,6 @@ export interface Session {
* how the command bar's empty query sorts a never-focused session after every
* session that has one (`docs/COMMAND-BAR.md`). */
last_focused_at?: IsoTimestamp | null;
/** The ACP session id the `claude-code-acp` adapter returned from
* `session/new` (#134). Only a `Structured` session whose handshake
* succeeded has one; a respawn passes it to `session/load` (#135). */
acp_session_id?: string | null;
}
/** A session's daemon-created worktree (`SessionWorktree` on the wire). */