Compare commits

..
Author SHA1 Message Date
Miguel Palhas 0292f373f7 test(daemon): cover projection lifecycle 2026-09-05 01:31:20 +01:00
Miguel Palhas eea337ce36 fix(daemon): scope active tools to turn 2026-09-05 01:30:16 +01:00
Miguel Palhas fd3ea64949 fix(daemon): reconcile transcript projections 2026-09-05 01:26:12 +01:00
Miguel Palhas 046c110460 feat(maestro): project ACP transcript state 2026-09-05 01:13:39 +01:00
Miguel Palhas 6eb65f1057 Merge structured-view/183-dismissal-race into blitz/structured-view 2026-09-05 00:57:46 +01:00
Miguel Palhas 3a68c1dd2f fix(maestro): preserve dismissed recovery cleanup 2026-09-05 00:57:36 +01:00
Miguel Palhas fde1954eac Merge structured-view/182-natural-exit-race into blitz/structured-view 2026-09-05 00:48:22 +01:00
Miguel Palhas 25435c34b2 fix(maestro): classify exited ACP children 2026-09-05 00:47:51 +01:00
Miguel Palhas 4a956e9c38 Merge structured-view/181-recovery-provenance into blitz/structured-view 2026-09-05 00:36:19 +01:00
Miguel Palhas 2a49814572 fix(maestro): mark graceful ACP recovery 2026-09-05 00:35:50 +01:00
Miguel Palhas d1aa56bab3 Merge structured-view/180-shutdown-create into blitz/structured-view 2026-09-05 00:26:07 +01:00
Miguel Palhas cf322746a7 fix(maestro): serialize structured shutdown 2026-09-05 00:22:59 +01:00
Miguel Palhas b25e733539 Merge structured-view/179-shutdown-drain into blitz/structured-view 2026-09-05 00:11:13 +01:00
Miguel Palhas 5a861976d7 test(maestro): cover signal shutdown drain 2026-09-05 00:07:19 +01:00
Miguel Palhas bfe4965081 fix(maestro): stop ACP before WS drain 2026-09-05 00:07:19 +01:00
Miguel Palhas 9d7ed8c071 fix(maestro): serialize shutdown finalization 2026-09-05 00:07:19 +01:00
Miguel Palhas 88a40c1ea7 Merge structured-view/177-restart-worktree into blitz/structured-view 2026-09-04 23:58:43 +01:00
Miguel Palhas 7049b5f1c0 docs(maestro): document structured restart recovery 2026-09-04 23:53:20 +01:00
Miguel Palhas c277ad630c fix(maestro): retain worktree on ACP restart 2026-09-04 23:53:20 +01:00
Miguel Palhas fc9c62d46b test(maestro): cover structured restart worktree 2026-09-04 23:53:20 +01:00
Miguel Palhas 3fe560d67e Merge structured-view/178-pid-safe into blitz/structured-view 2026-09-04 23:46:32 +01:00
Miguel Palhas b2b33535ae fix(maestro): own ACP child signals 2026-09-04 23:42:17 +01:00
Miguel Palhas 4ea6ddcaee Merge structured-view/135-event-log into blitz/structured-view 2026-09-04 23:30:15 +01:00
Miguel Palhas e9142bb027 fix(maestro): close ACP replay seams
Serialize replay reads with event appends, acquire the live receiver under that lock, and drain the event pump during shutdown. Recovery now removes its FIFO when opening the log fails.
2026-09-04 23:25:46 +01:00
Miguel Palhas d11098a478 feat(maestro): persist live ACP events
Own the ACP update receiver inside SessionManager, append each notification before broadcasting it, and recover structured sidecars by loading their stored ACP session id without duplicating replayed history.
2026-09-04 23:20:48 +01:00
Miguel Palhas 61776b4e0a fix(maestro): preserve post-load events
Switch replay suppression off inside the ordered session/load response callback, before the dispatcher can deliver the next live notification.
2026-09-04 23:18:59 +01:00
Miguel Palhas eabd21bccf feat(maestro): load existing ACP sessions
Resume a stored ACP session through session/load and discard the history replayed during the load handshake so the durable event log remains the single transcript source.
2026-09-04 23:13:45 +01:00
Miguel PalhasandClaude Fable 5.1 fd54d4f833 feat(maestro): add the per-session event log to SidecarStore (#135)
<sidecar_dir>/<id>.events.jsonl holds one session/update notification
per line; seq is the line's 1-based position. open_events returns an
append handle held for the session's lifetime, and terminates a
partial last line first so a hard kill mid-write never glues the next
event onto the fragment. read_events takes a Since (forward, exclusive)
or Before (backward, exclusive, None for the tail) cursor with a limit;
unparseable lines keep their seq and are skipped. delete removes the
log with the sidecar; load_all ignores .jsonl.

Each append is a single write_all to the kernel, no fsync: a killed
daemon loses nothing already written, and an fsync per streamed chunk
is the wrong trade for a log that is not meant to survive power loss.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
2026-09-04 22:14:33 +01:00
Miguel Palhas 78b5abb0cf Merge structured-view/134-acp-subprocess into blitz/structured-view
Closes #134
2026-09-04 22:06:46 +01:00
Miguel Palhas 6d8b49b75f docs(maestro): document the claude-code-acp install (#134) 2026-09-04 22:03:23 +01:00
Miguel Palhas 6c20a99d58 feat(maestro): end adapter children on daemon shutdown (#134)
SIGTERM now counts as a shutdown alongside ctrl-c: systemd stops the
unit with SIGTERM, and with KillMode=process nothing else would reach
the claude-code-acp children. After the listener drains the manager
terminates every adapter before the process exits.
2026-09-04 22:03:23 +01:00
Miguel Palhas b1718559df feat(maestro): spawn a claude-code-acp child per structured session (#134)
A Structured spawn gets no tmux pane: the daemon starts the adapter in
the session cwd with the session env (including MAESTRO_SESSION_ID and
MAESTRO_URL), runs the handshake, and records the ACP session id on the
sidecar. An adapter that cannot come up still yields a session, Dead
with activity Error and the reason as activity_detail, so a missing
adapter reads as the install command in the sidebar rather than as a
session that never appeared.

The liveness pass checks a structured session's child instead of the
tmux list. kill sends session/cancel, waits, then SIGTERMs through the
ACP handle. SessionManager::shutdown ends every adapter concurrently
and marks those sessions Dead on disk. take_acp_updates hands #135 the
raw session/update stream once, with earlier notifications still queued.
2026-09-04 22:02:39 +01:00
Miguel Palhas 540a3025b1 feat(maestro): add maestro-acp, the ACP client per structured session (#134)
New crate wrapping agent-client-protocol 2.1 in the client role. spawn()
resolves claude-code-acp on the PATH the child would run with, starts it
in the session cwd with the session env, runs initialize + session/new,
and returns an AcpSession holding the connection, the ACP session id,
and an unbounded channel of raw session/update notifications for #135.

initialize declares fs.readTextFile, fs.writeTextFile and terminal as
false so the adapter uses Claude Code's own tools. request_permission is
a stub that declines until #139.

terminate() sends session/cancel, waits, SIGTERMs, then SIGKILLs. The
child stays in the daemon's process group and gets SIGTERM on daemon
death via PR_SET_PDEATHSIG. CLAUDECODE is stripped from the child env:
the adapter refuses to start when it believes it is nested in another
Claude Code session, which is how a daemon launched from one behaves.

Tests run against tests/fake_acp_agent.py, a JSON-RPC-over-stdio stand-in
that logs what it receives; real_adapter_handshake is #[ignore]d.
2026-09-04 20:13:30 +01:00
Miguel Palhas c0da16658a feat(maestro): add acp_session_id to the session sidecar (#134)
A Structured session's adapter hands back an ACP session id from
session/new; a respawn will pass it to session/load (#135). This is a
new field rather than #161's agent_session_id: that one is the Claude
Code conversation id the activity hook reports and `claude --resume`
takes, minted by Claude Code itself. The ACP id is minted by the
adapter, only exists for Structured sessions, and is what the adapter
takes back. Overloading one field with two ids from two processes
would leave a respawn unable to tell which kind it holds.
2026-09-04 20:06:12 +01:00
Miguel Palhas eb4655a9d2 test(maestro): update collision tests for branch suffixing
49ab7e9 made a new-branch worktree add suffix the branch name instead of
failing, and documented it, but left two tests asserting the old hard
rejection. The rejection contract (#66) still holds for an existing
branch that is already checked out, so that test now exercises that
case.
2026-09-04 20:02:20 +01:00
Miguel Palhas c76059ec66 Merge structured-view/133-session-view into blitz/structured-view
Closes #133
2026-09-04 19:57:18 +01:00
Miguel Palhas f02168ffd5 feat(maestro): add SessionView to the protocol and sidecar
Adds `SessionView { Terminal, Structured }` to maestro-protocol, a
defaulted `view` field on `Session` and `CreateSessionRequest`, and
sidecar round-trip tests for both a legacy payload with no `view` key
and a `Structured` session. Nothing reads the field yet; every other
struct literal across the workspace picks up `Default::default()`
(Terminal) to keep compiling.

Gitea #133.
2026-09-04 19:41:29 +01:00
Miguel PalhasandClaude Sonnet 5 f0c272e0b1 feat(maestro): copy terminal selection to clipboard
ci / web (push) Successful in 28s
ci / rust (push) Failing after 2m6s
ci / flake (push) Successful in 2m14s
xterm.js has no built-in copy-on-select; mirror onSelectionChange to
navigator.clipboard.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-09-04 14:33:24 +01:00
Miguel Palhas 624f6f19c8 Merge remote-tracking branch 'origin/fermata'
ci / web (push) Failing after 13s
ci / flake (push) Successful in 3m12s
ci / rust (push) Failing after 3m26s
2026-09-04 13:32:05 +01:00
Miguel Palhas 08161ae042 Merge remote-tracking branch 'origin/main' into fermata 2026-09-04 13:30:19 +01:00
Miguel PalhasandClaude Sonnet 5 1e3c7b294d fix(maestro): send plain Escape to the pane, move focus.toggle behind the leader
Escape unfocusing the pane made it impossible to send a literal Esc into
the attached session. Removes pane.focusList and its Escape binding, and
rebinds focus.toggle from ctrl-shift-h (claimed by the browser's History
shortcut) to leader-h, consistent with the other leader escape hatches.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-09-04 13:30:14 +01:00
Miguel PalhasandClaude Sonnet 5 49ab7e9c81 fix(maestro): suffix branch name on new-branch worktree collision
ci / rust (push) Failing after 3m26s
ci / web (push) Successful in 43s
ci / flake (push) Successful in 3m54s
git worktree add -b fails outright when the branch already exists, even
if the target directory is free. A random fast-forward handle commonly
collides with a branch an earlier, already-removed worktree left behind.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-09-04 13:29:49 +01:00
Miguel PalhasandClaude Sonnet 5 0cc25d7e1c fix(maestro): equalize split widths after spawning a session
ci / rust (push) Successful in 9m12s
ci / web (push) Successful in 33s
ci / flake (push) Successful in 2m37s
splitLeaf always halved just the target pane, so repeated splits left
one pane at 50% and the rest squeezed into shrinking fractions instead
of equal columns.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-09-04 13:29:03 +01:00
Miguel PalhasandClaude Opus 5 706942ddd7 feat(maestro): install activity hooks for codex too
ci / web (push) Successful in 51s
ci / flake (push) Successful in 3m52s
ci / rust (push) Successful in 6m47s
The hook installer only ever wrote ~/.claude/settings.json, so a codex
session reported no activity and never posted its first prompt — leaving
it on the daemon's generated handle with no haiku name.

Adds a Target enum (claude-code, codex) over the hook file's location and
event set. One script and one marker serve both; codex swaps Notification
for PermissionRequest and has no StopFailure, so it never reports Error.
Both agents spell the prompt event UserPromptSubmit and put the text in
the payload's `prompt`, so autoname needs no change.

Codex will not run a hook it has not been told to trust; `maestro hooks
install` says so, and the approval stays the operator's.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-04 10:44:14 +01:00
Miguel Palhas 52877abe29 fix(maestro): focus the kill button in kill dialogs
ci / web (push) Successful in 37s
ci / flake (push) Successful in 2m50s
ci / rust (push) Successful in 5m46s
Enter now confirms the kill the operator just asked for; Escape and the scrim still cancel.
2026-09-04 08:24:32 +01:00
70 changed files with 5501 additions and 2739 deletions
-1
View File
@@ -14,7 +14,6 @@ maestro daemon tmux wrapper + session API + SPA on 127.0.0.1:7070
├── DELETE /api/sessions/:id kill
├── POST /api/sessions/:id/input type text into a session
├── POST /api/sessions/:id/prompt an agent reports its first prompt (#131)
├── POST /api/sessions/:id/resume respawn a dormant session, same id (#166)
├── GET /api/info hostname, version, home dir, spawn presets
├── GET /api/dirs directory listing (spawn autocomplete)
├── GET /api/recent-dirs spawn history
Generated
+747 -57
View File
File diff suppressed because it is too large Load Diff
+3
View File
@@ -10,6 +10,7 @@ repository = "https://git.naps.pt/naps62/maestro"
publish = false
[workspace.dependencies]
agent-client-protocol = { version = "2.1", features = ["unstable_tool_call_name"] }
axum = { version = "0.7", features = ["ws"] }
chrono = { version = "0.4", default-features = false, features = ["clock", "serde"] }
clap = { version = "4.5", features = ["derive", "env"] }
@@ -17,6 +18,7 @@ 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
@@ -38,6 +40,7 @@ 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
@@ -0,0 +1,20 @@
[package]
name = "maestro-acp"
version.workspace = true
edition.workspace = true
license.workspace = true
repository.workspace = true
publish.workspace = true
description = "ACP client role: one claude-code-acp child per structured session"
[dependencies]
agent-client-protocol.workspace = true
libc.workspace = true
serde_json.workspace = true
thiserror.workspace = true
tokio.workspace = true
tokio-util.workspace = true
tracing.workspace = true
[dev-dependencies]
tempfile.workspace = true
+594
View File
@@ -0,0 +1,594 @@
//! ACP client role for the daemon (#134): one `claude-code-acp` child per
//! structured session.
//!
//! The daemon is the client; the adapter is the agent. [`spawn`] starts the
//! adapter as a child process with stdio pipes, in the session's `cwd` with
//! the session's env, runs the `initialize` + `session/new` handshake, and
//! hands back an [`AcpSession`] that owns the connection for the life of the
//! session.
//!
//! **The handshake declares almost nothing.** `initialize` reports
//! `fs.readTextFile`, `fs.writeTextFile` and `terminal` as `false`, so the
//! adapter uses Claude Code's own Read / Write / Bash instead of asking the
//! daemon to act on its behalf — exactly what already happens in the
//! terminal view. `session/request_permission` is the one client method that
//! gets a real implementation, in #139; here it is a stub that declines.
//!
//! Every `session/update` notification the adapter sends lands on an
//! unbounded channel the session owns ([`AcpSession::take_updates`]). Nothing
//! is dropped while nobody reads: #135 takes the receiver and persists each
//! notification to the per-session event log before broadcasting it.
//!
//! A respawn after a daemon restart sets [`SpawnConfig::load_session`] and
//! the handshake runs `session/load` with the stored id instead of
//! `session/new`. The history the agent streams back while that request is
//! in flight is *not* delivered: the daemon's event log already holds it,
//! and replaying it would double every line.
//!
//! The child is a child of the daemon, full stop. It shares the daemon's
//! process group, is marked to receive `SIGTERM` when the daemon dies
//! (`PR_SET_PDEATHSIG` on Linux), and [`AcpSession::terminate`] is what both
//! `DELETE /api/sessions/:id` and daemon shutdown call: `session/cancel`, a
//! short wait, then `SIGTERM`, then `SIGKILL` as the last resort.
#![deny(missing_debug_implementations)]
use std::collections::{HashMap, VecDeque};
use std::ffi::OsString;
use std::path::{Path, PathBuf};
use std::process::{ExitStatus, Stdio};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use agent_client_protocol::schema::v1::{
CancelNotification, ClientCapabilities, FileSystemCapabilities, Implementation,
InitializeRequest, LoadSessionRequest, 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,
/// An existing ACP session id to resume with `session/load` (#135)
/// instead of opening a fresh one. The adapter must advertise
/// `loadSession`; the handshake fails otherwise.
pub load_session: Option<String>,
}
impl SpawnConfig {
pub fn new(cwd: impl Into<PathBuf>, env: HashMap<String, String>) -> Self {
Self {
cwd: cwd.into(),
env,
handshake_timeout: HANDSHAKE_TIMEOUT,
load_session: None,
}
}
/// Resume `session_id` with `session/load` instead of `session/new`.
pub fn load_session(mut self, session_id: impl Into<String>) -> Self {
self.load_session = Some(session_id.into());
self
}
}
#[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,
}
enum ChildCommand {
Signal(libc::c_int),
Reap(oneshot::Sender<bool>),
}
/// Why the handshake did not produce a session id.
enum ReadyError {
/// The adapter answered with an error.
Handshake(String),
/// The transport closed under the request: the child is gone.
Closed,
}
fn ready_error(e: agent_client_protocol::Error) -> ReadyError {
if agent_client_protocol::is_incoming_transport_closed(&e) {
ReadyError::Closed
} else {
ReadyError::Handshake(e.to_string())
}
}
/// A live ACP connection to one adapter child. Dropping it sends `SIGTERM`
/// if the child is still running; [`AcpSession::terminate`] is the polite
/// version.
#[derive(Debug)]
pub struct AcpSession {
session_id: String,
pid: u32,
commands: mpsc::UnboundedSender<Command>,
child_commands: mpsc::UnboundedSender<ChildCommand>,
updates: Mutex<Option<mpsc::UnboundedReceiver<SessionNotification>>>,
exit: watch::Receiver<Option<ExitStatus>>,
}
impl AcpSession {
/// The id `session/new` returned. What a respawn passes to
/// `session/load` (#135).
pub fn session_id(&self) -> &str {
&self.session_id
}
/// The adapter's pid.
pub fn pid(&self) -> u32 {
self.pid
}
/// Whether the child has been reaped.
pub fn has_exited(&self) -> bool {
self.exit.borrow().is_some()
}
/// Reap the child if it has already exited.
pub async fn reap_if_exited(&self) -> bool {
if self.has_exited() {
return true;
}
let (sent, received) = oneshot::channel();
if self.child_commands.send(ChildCommand::Reap(sent)).is_err() {
return self.has_exited();
}
received.await.unwrap_or_else(|_| self.has_exited())
}
/// Wait for the child to exit.
pub async fn wait(&self) -> Option<ExitStatus> {
wait_exit(&self.exit).await
}
/// The stream of raw `session/update` notifications, handed out once.
/// Notifications that arrived before the take are still queued on it.
pub fn take_updates(&self) -> Option<mpsc::UnboundedReceiver<SessionNotification>> {
self.updates.lock().expect("updates mutex").take()
}
/// 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) {
self.cancel();
if tokio::time::timeout(cancel_grace, self.wait())
.await
.is_ok()
{
return;
}
self.signal(libc::SIGTERM);
if tokio::time::timeout(term_grace, self.wait()).await.is_ok() {
return;
}
tracing::warn!(
pid = self.pid,
"{ADAPTER_BIN} ignored SIGTERM; sending SIGKILL"
);
self.signal(libc::SIGKILL);
let _ = self.commands.send(Command::Close);
self.wait().await;
}
fn signal(&self, sig: libc::c_int) {
let _ = self.child_commands.send(ChildCommand::Signal(sig));
}
}
impl Drop for AcpSession {
fn drop(&mut self) {
self.signal(libc::SIGTERM);
let _ = self.commands.send(Command::Close);
}
}
/// Start the adapter for one session and run the handshake.
///
/// Fails without a running child: on any error the process is already
/// signalled, so a failed session never leaves a stray adapter behind.
pub async fn spawn(config: SpawnConfig) -> Result<AcpSession, AcpError> {
let adapter = resolve_adapter(&config.env).ok_or(AcpError::AdapterMissing)?;
let mut cmd = tokio::process::Command::new(&adapter);
cmd.current_dir(&config.cwd)
.envs(&config.env)
// Claude Code refuses to start when it thinks it is nested inside
// another Claude Code session; a daemon launched from one would
// otherwise never get a structured session up.
.env_remove("CLAUDECODE")
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
// If the daemon is dropped without a graceful shutdown the runtime
// still tears the task holding this `Child` down, and that kills it.
.kill_on_drop(true);
#[cfg(target_os = "linux")]
{
// Same process group as the daemon (no setpgid), and SIGTERM when the
// daemon dies however it dies: an agent that outlives its daemon is
// exactly the orphan #134 rules out.
// SAFETY: prctl is async-signal-safe and touches no shared state.
unsafe {
cmd.pre_exec(|| {
libc::prctl(libc::PR_SET_PDEATHSIG, libc::SIGTERM);
Ok(())
});
}
}
let mut child = cmd.spawn().map_err(AcpError::Spawn)?;
let pid = child.id().expect("child pid is known before wait");
let stdin = child.stdin.take().expect("piped stdin");
let stdout = child.stdout.take().expect("piped stdout");
let stderr = child.stderr.take().expect("piped stderr");
let stderr_tail = Arc::new(Mutex::new(VecDeque::<u8>::new()));
tokio::spawn(drain_stderr(stderr, pid, stderr_tail.clone()));
let (exit_tx, exit_rx) = watch::channel(None);
let (child_commands_tx, child_commands_rx) = mpsc::unbounded_channel();
tokio::spawn(supervise_child(child, pid, child_commands_rx, exit_tx));
let (updates_tx, updates_rx) = mpsc::unbounded_channel();
let (commands_tx, commands_rx) = mpsc::unbounded_channel();
let (ready_tx, ready_rx) = oneshot::channel();
let transport = ByteStreams::new(stdin.compat_write(), stdout.compat());
tokio::spawn(run_connection(
transport,
config.cwd.clone(),
config.load_session.clone(),
updates_tx,
commands_rx,
ready_tx,
));
let outcome = tokio::time::timeout(config.handshake_timeout, ready_rx).await;
let failure = match outcome {
Ok(Ok(Ok(session_id))) => {
return Ok(AcpSession {
session_id,
pid,
commands: commands_tx,
child_commands: child_commands_tx,
updates: Mutex::new(Some(updates_rx)),
exit: exit_rx,
});
}
Ok(Ok(Err(ReadyError::Handshake(message)))) => AcpError::Handshake(message),
Ok(Ok(Err(ReadyError::Closed))) | Ok(Err(_)) => {
// The connection ended before the handshake did: the child went
// away. Reap it so the status is real, then report.
let status = wait_exit(&exit_rx).await;
let stderr = {
let mut tail = stderr_tail.lock().expect("stderr tail");
tail.make_contiguous();
String::from_utf8_lossy(tail.as_slices().0)
.trim()
.to_string()
};
AcpError::ExitedEarly {
status: status.map_or_else(|| "unknown status".into(), |s| s.to_string()),
stderr: if stderr.is_empty() {
String::new()
} else {
format!(": {stderr}")
},
}
}
Err(_) => AcpError::HandshakeTimeout(config.handshake_timeout),
};
// No half-started adapter survives a failed spawn.
let _ = commands_tx.send(Command::Close);
let _ = child_commands_tx.send(ChildCommand::Signal(libc::SIGTERM));
if tokio::time::timeout(TERM_GRACE, wait_exit(&exit_rx))
.await
.is_err()
{
let _ = child_commands_tx.send(ChildCommand::Signal(libc::SIGKILL));
wait_exit(&exit_rx).await;
}
Err(failure)
}
async fn wait_exit(exit: &watch::Receiver<Option<ExitStatus>>) -> Option<ExitStatus> {
let mut exit = exit.clone();
let status = exit.wait_for(|s| s.is_some()).await.ok()?;
*status
}
async fn supervise_child(
mut child: tokio::process::Child,
pid: u32,
mut commands: mpsc::UnboundedReceiver<ChildCommand>,
exit_tx: watch::Sender<Option<ExitStatus>>,
) {
loop {
tokio::select! {
status = child.wait() => {
let status = status.ok();
tracing::debug!(pid, ?status, "{ADAPTER_BIN} exited");
let _ = exit_tx.send(status);
return;
}
command = commands.recv() => match command {
Some(ChildCommand::Signal(sig)) => {
// This task alone owns `child`. It cannot reap the child
// between receiving this command and sending the signal,
// so its PID cannot be reused in that window.
unsafe { libc::kill(pid as libc::pid_t, sig) };
}
Some(ChildCommand::Reap(done)) => match child.try_wait() {
Ok(Some(status)) => {
let _ = done.send(true);
tracing::debug!(pid, ?status, "{ADAPTER_BIN} exited");
let _ = exit_tx.send(Some(status));
return;
}
Ok(None) => {
let _ = done.send(false);
}
Err(error) => {
tracing::warn!(pid, %error, "could not reap {ADAPTER_BIN}");
let _ = done.send(false);
}
},
None => {
// The session vanished without a command receiver. This
// task still owns the unreaped child, so this cannot hit
// a reused PID.
unsafe { libc::kill(pid as libc::pid_t, libc::SIGTERM) };
let status = child.wait().await.ok();
tracing::debug!(pid, ?status, "{ADAPTER_BIN} exited");
let _ = exit_tx.send(status);
return;
}
},
}
}
}
/// The capabilities the daemon declares: none of the filesystem or terminal
/// surface, so the adapter falls back to Claude Code's own tools.
fn client_capabilities() -> ClientCapabilities {
ClientCapabilities::new()
.fs(FileSystemCapabilities::new()
.read_text_file(false)
.write_text_file(false))
.terminal(false)
}
async fn run_connection<T>(
transport: T,
cwd: PathBuf,
load_session: Option<String>,
updates_tx: mpsc::UnboundedSender<SessionNotification>,
mut commands_rx: mpsc::UnboundedReceiver<Command>,
ready_tx: oneshot::Sender<Result<String, ReadyError>>,
) where
T: agent_client_protocol::ConnectTo<Client> + 'static,
{
// Set while `session/load` is in flight. Notification handlers run in
// the dispatch loop, in arrival order, so everything the agent streams
// before its load response sees the flag up and is dropped: that is the
// history the event log already holds (#135).
let replaying = Arc::new(AtomicBool::new(false));
let replaying_seen = replaying.clone();
let result = Client
.builder()
.name("maestro")
.on_receive_notification(
async move |notification: SessionNotification, _cx| {
if replaying_seen.load(Ordering::SeqCst) {
tracing::trace!("dropping replayed notification during session/load");
return Ok(());
}
// A dropped receiver only means nobody wants updates any more.
let _ = updates_tx.send(notification);
Ok(())
},
agent_client_protocol::on_receive_notification!(),
)
.on_receive_request(
async move |request: RequestPermissionRequest, responder, _cx| {
// 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")));
let initialized = match cx.send_request(init).block_task().await {
Ok(r) => r,
Err(e) => {
let _ = ready_tx.send(Err(ready_error(e)));
return Ok(());
}
};
let session_id = match load_session {
Some(existing) => {
if !initialized.agent_capabilities.load_session {
let _ = ready_tx.send(Err(ReadyError::Handshake(format!(
"{ADAPTER_BIN} does not advertise loadSession; cannot resume session {existing}"
))));
return Ok(());
}
replaying.store(true, Ordering::SeqCst);
let (loaded_tx, loaded_rx) = oneshot::channel();
let replaying_done = replaying.clone();
if let Err(e) = cx
.send_request(LoadSessionRequest::new(existing.clone(), cwd))
.on_receiving_result(move |result| {
replaying_done.store(false, Ordering::SeqCst);
async move {
let _ = loaded_tx.send(result);
Ok(())
}
})
{
let _ = ready_tx.send(Err(ready_error(e)));
return Ok(());
}
match loaded_rx.await {
Ok(Ok(_)) => {}
Ok(Err(e)) => {
let _ = ready_tx.send(Err(ready_error(e)));
return Ok(());
}
Err(_) => {
let _ = ready_tx.send(Err(ReadyError::Handshake(
"session/load response was not delivered".into(),
)));
return Ok(());
}
}
existing
}
None => match cx
.send_request(NewSessionRequest::new(cwd))
.block_task()
.await
{
Ok(created) => created.session_id.0.to_string(),
Err(e) => {
let _ = ready_tx.send(Err(ready_error(e)));
return Ok(());
}
},
};
let _ = ready_tx.send(Ok(session_id.clone()));
loop {
tokio::select! {
cmd = commands_rx.recv() => match cmd {
Some(Command::Cancel) => {
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();
}
}
+146
View File
@@ -0,0 +1,146 @@
#!/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_DELAY_SESSION_NEW_MS=n wait n milliseconds before answering
`session/new`
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)
FAKE_ACP_IGNORE_EOF=1 keep running after stdin closes, until signalled
FAKE_ACP_EMIT_ON_CANCEL=1 emit a final notification before handling cancel
FAKE_ACP_NO_LOAD=1 advertise `loadSession: false`
FAKE_ACP_LOAD_REPLAY=n on `session/load`, replay n `agent_message_chunk`
notifications ("replayed <i>") before answering,
the way a real agent streams history back
FAKE_ACP_LOAD_FAIL=1 answer `session/load` with a JSON-RPC error
Signals: SIGUSR1 emits one `agent_message_chunk` notification ("tick <n>",
counting from 1) and logs `emit tick <n>`, so a test can make the agent
speak on demand (#135).
"""
import json
import os
import signal
import sys
import time
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)
session_id = f"fake-{os.getpid()}"
ticks = 0
def chunk(text):
return {"jsonrpc": "2.0", "method": "session/update", "params": {
"sessionId": session_id,
"update": {"sessionUpdate": "agent_message_chunk",
"content": {"type": "text", "text": text}},
}}
def on_sigusr1(_signum, _frame):
global ticks
ticks += 1
send(chunk(f"tick {ticks}"))
log(f"emit tick {ticks}")
signal.signal(signal.SIGUSR1, on_sigusr1)
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, '')}")
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": os.environ.get("FAKE_ACP_NO_LOAD") != "1"},
"agentInfo": {"name": "fake-acp-agent", "version": "0.0.0"},
"authMethods": [],
}})
elif method == "session/new":
time.sleep(int(os.environ.get("FAKE_ACP_DELAY_SESSION_NEW_MS", "0")) / 1000)
send({"jsonrpc": "2.0", "id": rid, "result": {"sessionId": session_id}})
send(chunk("hello from fake"))
if os.environ.get("FAKE_ACP_EXIT_AFTER_NEW") == "1":
sys.exit(0)
elif method == "session/load":
session_id = params.get("sessionId")
log(f"recv session/load sessionId={session_id}")
if os.environ.get("FAKE_ACP_LOAD_FAIL") == "1":
send({"jsonrpc": "2.0", "id": rid,
"error": {"code": -32000, "message": "fake load failure"}})
continue
for i in range(int(os.environ.get("FAKE_ACP_LOAD_REPLAY", "0"))):
send(chunk(f"replayed {i + 1}"))
send({"jsonrpc": "2.0", "id": rid, "result": {}})
send(chunk("hello after load"))
elif method == "session/cancel":
if os.environ.get("FAKE_ACP_EMIT_ON_CANCEL") == "1":
send(chunk("final on 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}"}})
if os.environ.get("FAKE_ACP_IGNORE_EOF") == "1":
while True:
signal.pause()
+394
View File
@@ -0,0 +1,394 @@
//! `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(&self) -> u32 {
self.log()
.lines()
.find_map(|line| line.strip_prefix("start pid="))
.and_then(|rest| rest.split(' ').next())
.and_then(|pid| pid.parse().ok())
.expect("pid in log")
}
async fn wait_for_exit(&self) {
let pid = self.pid();
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} still alive");
}
}
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:?}"),
}
}
#[tokio::test]
async fn drop_terminates_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_IGNORE_EOF".into(), "1".into());
let session = maestro_acp::spawn(config).await.expect("spawn");
drop(session);
fake.wait_for_log("signal SIGTERM").await;
fake.wait_for_exit().await;
}
fn chunk_text(n: &maestro_acp::schema::SessionNotification) -> String {
match &n.update {
SessionUpdate::AgentMessageChunk(chunk) => match &chunk.content {
ContentBlock::Text(t) => t.text.clone(),
other => panic!("unexpected content {other:?}"),
},
other => panic!("unexpected update {other:?}"),
}
}
#[tokio::test]
async fn load_session_resumes_the_stored_id_and_drops_the_replay() {
let fake = Fake::new();
let cwd = TempDir::new().unwrap();
let mut config = fake.config(cwd.path()).load_session("fake-old");
config.env.insert("FAKE_ACP_LOAD_REPLAY".into(), "3".into());
let session = maestro_acp::spawn(config).await.expect("spawn");
assert_eq!(session.session_id(), "fake-old");
let log = fake.log();
assert!(
log.contains("recv session/load sessionId=fake-old"),
"{log}"
);
assert!(!log.contains("recv session/new"), "{log}");
// The three replayed chunks never reach the receiver: the event log
// already holds them. The first notification after the response does.
let mut updates = session.take_updates().expect("updates");
let first = tokio::time::timeout(Duration::from_secs(5), updates.recv())
.await
.expect("update within 5s")
.expect("channel open");
assert_eq!(chunk_text(&first), "hello after load");
// Nothing queued behind it either.
tokio::time::sleep(Duration::from_millis(100)).await;
assert!(updates.try_recv().is_err(), "replay leaked through");
session
.terminate(Duration::from_millis(100), Duration::from_secs(2))
.await;
}
#[tokio::test]
async fn load_session_requires_the_load_capability() {
let fake = Fake::new();
let cwd = TempDir::new().unwrap();
let mut config = fake.config(cwd.path()).load_session("fake-old");
config.env.insert("FAKE_ACP_NO_LOAD".into(), "1".into());
let err = maestro_acp::spawn(config)
.await
.expect_err("no loadSession");
assert!(matches!(err, AcpError::Handshake(_)), "{err}");
assert!(err.to_string().contains("loadSession"), "{err}");
assert!(!fake.log().contains("recv session/load"), "{}", fake.log());
}
#[tokio::test]
async fn load_session_error_fails_the_spawn() {
let fake = Fake::new();
let cwd = TempDir::new().unwrap();
let mut config = fake.config(cwd.path()).load_session("fake-old");
config.env.insert("FAKE_ACP_LOAD_FAIL".into(), "1".into());
let err = maestro_acp::spawn(config).await.expect_err("load fails");
assert!(matches!(err, AcpError::Handshake(_)), "{err}");
assert!(err.to_string().contains("fake load failure"), "{err}");
}
/// 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());
}
+23 -11
View File
@@ -96,20 +96,32 @@ pub async fn dispatch(
}
/// `maestro hooks install` (#106): write the shipped hook script under the
/// user's data dir and register it in their Claude Code settings. Idempotent
/// and foreign-hook-preserving — the guarantees live in [`maestro_hooks`].
/// user's data dir and register it in every agent's hook file. Idempotent and
/// foreign-hook-preserving — the guarantees live in [`maestro_hooks`]. One
/// agent failing is reported and does not stop the others.
#[allow(clippy::result_large_err)]
fn run_hooks_install() -> Result<(), CliError> {
let home = dirs::home_dir().ok_or(CliError::NoHome)?;
let changed = maestro_hooks::install_for_home(&home)?;
if changed {
println!(
"installed claude-code activity hook: {} → {}",
maestro_hooks::script_path(&home).display(),
maestro_hooks::settings_path(&home).display()
);
} else {
println!("claude-code activity hook already up to date");
let script = maestro_hooks::script_path(&home);
for (target, result) in maestro_hooks::install_for_home(&home) {
let agent = target.label();
let settings = target.settings_path(&home);
match result {
Ok(true) => {
println!(
"installed {agent} activity hook: {} → {}",
script.display(),
settings.display()
);
if target == maestro_hooks::Target::Codex {
// Codex will not run a hook it has not been told to
// trust, and that approval is the operator's to give.
println!(" codex will ask you to trust this hook the next time it starts");
}
}
Ok(false) => println!("{agent} activity hook already up to date"),
Err(e) => eprintln!("{agent} activity hook not installed: {e}"),
}
}
Ok(())
}
+3 -2
View File
@@ -51,6 +51,7 @@ 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,
@@ -188,7 +189,6 @@ fn print_session_table(out: &mut impl Write, sessions: &[Session], info: &Daemon
let status = match s.status {
SessionStatus::Running => "running",
SessionStatus::Dead => "dead",
SessionStatus::Dormant => "dormant",
};
let activity = match s.activity {
ActivityState::Active => "active",
@@ -237,6 +237,7 @@ 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(),
@@ -253,7 +254,7 @@ mod tests {
current_cwd: None,
blocked_work: None,
last_focused_at: None,
agent_session_id: None,
acp_session_id: None,
}
}
@@ -28,7 +28,6 @@ async fn info_carries_version_and_presets() {
id: "claude".into(),
label: "Claude".into(),
command: "claude".into(),
resume_command: None,
}],
..Default::default()
});
@@ -92,6 +91,7 @@ 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(),
-140
View File
@@ -129,80 +129,6 @@ pub struct DaemonSection {
/// `Some(0)` disables history entirely (no log written, no recents read).
#[serde(default)]
pub spawn_history_days: Option<u32>,
/// What recovery does with sessions that outlived their tmux server —
/// after a host reboot, or any restart that finds a sidecar with no pane
/// (#167). Default [`ResumeOnBoot::Dormant`].
#[serde(default, deserialize_with = "deserialize_resume_on_boot")]
pub resume_on_boot: Option<ResumeOnBoot>,
}
impl DaemonSection {
/// The configured `[daemon].resume_on_boot`, or the safe default:
/// recovered-but-dead sessions wait for the operator.
pub fn resume_on_boot(&self) -> ResumeOnBoot {
self.resume_on_boot.unwrap_or_default()
}
}
/// `[daemon].resume_on_boot` — what recovery does with a session whose tmux
/// pane is gone but whose sidecar is still on disk (#167).
///
/// Resuming agents unattended seconds after a reboot is a foot-gun: they run
/// with whatever flags they were spawned with, with nobody watching. So the
/// default waits for a human, and `auto` is opted into per host.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub enum ResumeOnBoot {
/// Register each one `Dormant` and wait for the operator to resume it.
#[default]
Dormant,
/// Resume each one during recovery, on the same path a client's
/// `POST /api/sessions/:id/resume` takes.
Auto,
/// Prune the sidecars instead: a reboot is a clean slate.
Never,
}
/// Every `[daemon].resume_on_boot` name and its variant, in documentation
/// order.
pub const RESUME_ON_BOOT_POLICIES: &[(&str, ResumeOnBoot)] = &[
("dormant", ResumeOnBoot::Dormant),
("auto", ResumeOnBoot::Auto),
("never", ResumeOnBoot::Never),
];
impl ResumeOnBoot {
/// The name this policy is spelled with in the config file and in logs.
pub fn as_str(&self) -> &'static str {
match self {
ResumeOnBoot::Dormant => "dormant",
ResumeOnBoot::Auto => "auto",
ResumeOnBoot::Never => "never",
}
}
}
fn deserialize_resume_on_boot<'de, D>(deserializer: D) -> Result<Option<ResumeOnBoot>, D::Error>
where
D: Deserializer<'de>,
{
let raw = Option::<String>::deserialize(deserializer)?;
raw.map(|name| {
RESUME_ON_BOOT_POLICIES
.iter()
.find(|(n, _)| *n == name)
.map(|(_, policy)| *policy)
.ok_or_else(|| {
let known = RESUME_ON_BOOT_POLICIES
.iter()
.map(|(n, _)| format!("{n:?}"))
.collect::<Vec<_>>()
.join(", ");
D::Error::custom(format!(
"invalid [daemon].resume_on_boot {name:?}; expected one of {known}"
))
})
})
.transpose()
}
/// `[tui]` — `maestro tui` settings (NAP-275 epic, phase 2 NAP-277).
@@ -387,19 +313,16 @@ pub fn builtin_presets() -> Vec<SpawnPreset> {
id: "claude".into(),
label: "Claude".into(),
command: "claude".into(),
resume_command: Some("claude --resume {agent_session}".into()),
},
SpawnPreset {
id: "opencode".into(),
label: "opencode".into(),
command: "opencode".into(),
resume_command: None,
},
SpawnPreset {
id: "zsh".into(),
label: "Shell".into(),
command: "zsh".into(),
resume_command: None,
},
]
}
@@ -411,7 +334,6 @@ pub fn custom_preset() -> SpawnPreset {
id: CUSTOM_PRESET_ID.into(),
label: "Custom…".into(),
command: String::new(),
resume_command: None,
}
}
@@ -585,18 +507,6 @@ auto_install = false
assert!(custom.command.is_empty());
}
#[test]
fn builtin_claude_preset_resumes_by_agent_session() {
let claude = builtin_presets()
.into_iter()
.find(|preset| preset.id == "claude")
.expect("Claude preset");
assert_eq!(
claude.resume_command.as_deref(),
Some("claude --resume {agent_session}")
);
}
#[test]
fn spawn_presets_override_replaces_builtins() {
// Present list replaces the built-ins entirely; custom still appended.
@@ -605,7 +515,6 @@ auto_install = false
id: "fish".into(),
label: "fish".into(),
command: "fish".into(),
resume_command: None,
}]),
..Default::default()
};
@@ -642,7 +551,6 @@ auto_install = false
id: " ".into(),
label: "x".into(),
command: "x".into(),
resume_command: None,
}]),
..Default::default()
};
@@ -653,7 +561,6 @@ auto_install = false
id: "x".into(),
label: "x".into(),
command: "".into(),
resume_command: None,
}]),
..Default::default()
};
@@ -671,13 +578,11 @@ auto_install = false
id: "a".into(),
label: "a".into(),
command: "a".into(),
resume_command: None,
},
SpawnPreset {
id: "a".into(),
label: "a2".into(),
command: "a2".into(),
resume_command: None,
},
]),
..Default::default()
@@ -692,7 +597,6 @@ auto_install = false
id: "custom".into(),
label: "Custom".into(),
command: "whatever".into(),
resume_command: None,
}]),
..Default::default()
};
@@ -728,27 +632,6 @@ presets = [
assert_eq!(ids, ["claude", "zsh", "custom"]);
}
#[test]
fn spawn_preset_parses_resume_command_from_issue_example() {
let cfg = toml::from_str::<MaestroConfigFile>(
r#"
[spawn]
presets = [
{ id = "claude", command = "claude --dangerously-skip-permissions", resume_command = "claude --dangerously-skip-permissions --resume {agent_session}" }
]
"#,
)
.expect("issue #165 example parses");
let preset = &cfg.spawn.presets.expect("configured presets")[0];
assert_eq!(preset.id, "claude");
assert_eq!(preset.command, "claude --dangerously-skip-permissions");
assert_eq!(
preset.resume_command.as_deref(),
Some("claude --dangerously-skip-permissions --resume {agent_session}")
);
}
#[test]
fn spawn_project_scan_fields_default_to_none() {
// Absent [spawn].project_* → None; the daemon applies its own defaults.
@@ -849,27 +732,4 @@ bind_addr = "127.0.0.1:9090"
);
assert_eq!(cfg.resolve_daemon_url(None), "http://explicit:1");
}
#[test]
fn resume_on_boot_defaults_to_dormant_and_parses_every_policy() {
assert_eq!(
MaestroConfigFile::default().daemon.resume_on_boot(),
ResumeOnBoot::Dormant,
"an absent policy is the safe one"
);
for (name, expected) in RESUME_ON_BOOT_POLICIES {
let cfg = file_with(&format!("[daemon]\nresume_on_boot = \"{name}\"\n"));
assert_eq!(cfg.daemon.resume_on_boot(), *expected, "{name}");
assert_eq!(expected.as_str(), *name);
}
}
#[test]
fn resume_on_boot_rejects_an_unknown_policy() {
let err = toml::from_str::<MaestroConfigFile>("[daemon]\nresume_on_boot = \"maybe\"\n")
.expect_err("unknown policy must not silently fall back");
let msg = err.to_string();
assert!(msg.contains("resume_on_boot"), "{msg}");
assert!(msg.contains("\"auto\""), "{msg}");
}
}
+2
View File
@@ -28,6 +28,7 @@ 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 }
@@ -42,6 +43,7 @@ tracing-subscriber.workspace = true
uuid.workspace = true
[dev-dependencies]
libc.workspace = true
reqwest.workspace = true
tokio-tungstenite.workspace = true
urlencoding = "2"
-65
View File
@@ -1,65 +0,0 @@
//! Tells a daemon restart from a host reboot (#164).
//!
//! `/proc/sys/kernel/random/boot_id` is fixed for the life of the kernel and
//! changes on every boot. [`SessionManager::recover`] compares it against the
//! value it stored on the previous run and logs which of the two it was.
//!
//! It does not decide liveness (#174). A boot id changes without the tmux
//! server dying whenever the state directory is restored, a VM or container
//! snapshot is rolled back, or the `boot_id` file is truncated — so recovery
//! always probes `tmux list-sessions` and the boot id only says whether a
//! server that did not answer is coming back.
use std::path::Path;
/// Where the daemon reads the running kernel's boot id. Behind a trait so
/// tests can simulate a changed (or unreadable) boot id without rebooting the
/// machine.
pub trait BootIdSource: std::fmt::Debug + Send + Sync + 'static {
/// Current boot id, or `None` if it can't be determined (non-Linux,
/// restricted `/proc`, missing or empty file).
fn current(&self) -> Option<String>;
}
/// Production source: `/proc/sys/kernel/random/boot_id`.
#[derive(Debug, Default)]
pub struct ProcBootId;
impl BootIdSource for ProcBootId {
fn current(&self) -> Option<String> {
read_boot_id(Path::new("/proc/sys/kernel/random/boot_id"))
}
}
fn read_boot_id(path: &Path) -> Option<String> {
let contents = std::fs::read_to_string(path).ok()?;
let trimmed = contents.trim();
(!trimmed.is_empty()).then(|| trimmed.to_string())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn reads_trimmed_contents() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("boot_id");
std::fs::write(&path, "abc-123\n").unwrap();
assert_eq!(read_boot_id(&path).as_deref(), Some("abc-123"));
}
#[test]
fn missing_file_returns_none() {
let dir = tempfile::tempdir().unwrap();
assert_eq!(read_boot_id(&dir.path().join("missing")), None);
}
#[test]
fn empty_file_returns_none() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("boot_id");
std::fs::write(&path, "\n").unwrap();
assert_eq!(read_boot_id(&path), None);
}
}
+34 -22
View File
@@ -51,17 +51,23 @@ async fn run(config_override: Option<PathBuf>) -> Result<(), Box<dyn Error>> {
// First-start hook auto-install (#106). Best-effort, before serving: a
// session spawned right away should already carry the hook.
if config.hooks_auto_install {
match crate::hooks::auto_install(&config.home_dir) {
crate::hooks::AutoInstallOutcome::Installed => {
tracing::info!("installed claude-code activity hook (#106)");
for (target, outcome) in crate::hooks::auto_install(&config.home_dir) {
let agent = target.label();
match outcome {
crate::hooks::AutoInstallOutcome::Installed => {
tracing::info!(agent, "installed activity hook (#106)");
}
crate::hooks::AutoInstallOutcome::Upgraded => {
tracing::info!(agent, "updated activity hook to this build (#106)");
}
crate::hooks::AutoInstallOutcome::Failed => {
tracing::warn!(
agent,
"hooks auto-install failed; run `maestro hooks install` manually"
);
}
_ => {}
}
crate::hooks::AutoInstallOutcome::Upgraded => {
tracing::info!("updated claude-code activity hook to this build (#106)");
}
crate::hooks::AutoInstallOutcome::Failed => {
tracing::warn!("hooks auto-install failed; run `maestro hooks install` manually");
}
_ => {}
}
}
// No worktree dir to pre-create: without an override the worktrees live
@@ -83,14 +89,7 @@ async fn run(config_override: Option<PathBuf>) -> Result<(), Box<dyn Error>> {
.with_spawn_history(crate::spawn_history::SpawnHistory::new(
config.spawn_history_path(),
config.spawn_history_days,
))
.with_boot_id(
config.boot_id_path(),
std::sync::Arc::new(crate::boot_id::ProcBootId),
)
// #167: recovery needs the presets to resolve a resumed session's command,
// so `auto` runs the same resume a client's POST would.
.with_resume_policy(config.resume_on_boot, config.spawn_presets.clone());
));
// Spawned sessions learn where to report back (#103). Derived from the
// configured bind address; set before any spawn can happen.
manager.set_daemon_url(format!("http://{}", config.bind_addr));
@@ -107,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,
manager.clone(),
config.hostname.clone(),
config.home_dir.clone(),
config.spawn_presets.clone(),
@@ -119,9 +118,22 @@ async fn run(config_override: Option<PathBuf>) -> Result<(), Box<dyn Error>> {
let bound = listener.local_addr()?;
tracing::info!(addr = %bound, "API listening");
let shutdown = async {
let _ = tokio::signal::ctrl_c().await;
tracing::info!("ctrl-c received — shutting down");
// 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_manager = manager.clone();
let shutdown = async move {
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"),
}
// Axum waits for upgraded WebSockets to drain. Structured children
// must be settled before that wait, while the signal path still owns
// the shutdown order (#179).
shutdown_manager.shutdown().await;
};
axum::serve(listener, router)
-41
View File
@@ -97,10 +97,6 @@ pub struct DaemonConfig {
/// never binds it itself; it serves it so the web client and the TUI read
/// one knob.
pub leader: String,
/// What recovery does with sessions whose tmux pane is gone but whose
/// sidecar survives (#167): `[daemon].resume_on_boot`, default
/// [`maestro_config::ResumeOnBoot::Dormant`].
pub resume_on_boot: maestro_config::ResumeOnBoot,
}
impl DaemonConfig {
@@ -116,12 +112,6 @@ impl DaemonConfig {
pub fn spawn_history_path(&self) -> PathBuf {
self.state_dir.join("spawn-history.jsonl")
}
/// Where recovery records the kernel boot id it saw last (#164):
/// `<state_dir>/boot_id`.
pub fn boot_id_path(&self) -> PathBuf {
self.state_dir.join("boot_id")
}
}
/// Env-var overrides applied by [`load_config`]. Tests pass explicit
@@ -153,8 +143,6 @@ pub fn load_config(path: &Path, env: EnvOverrides) -> Result<DaemonConfig, Confi
// validation (#22/#26).
let spawn_presets = file.spawn.resolve()?;
let resume_on_boot = daemon.resume_on_boot();
let hostname = daemon.hostname.unwrap_or_else(system_hostname);
let home_dir = dirs::home_dir()
.map(|p| p.to_string_lossy().into_owned())
@@ -215,7 +203,6 @@ pub fn load_config(path: &Path, env: EnvOverrides) -> Result<DaemonConfig, Confi
project_scan_depth,
hooks_auto_install: file.hooks.auto_install(),
leader: file.tui.leader().to_string(),
resume_on_boot,
})
}
@@ -296,34 +283,6 @@ auto_install = false
assert!(!cfg.hooks_auto_install);
}
#[test]
fn resume_on_boot_defaults_to_dormant_and_reads_the_file() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("config.toml");
std::fs::write(&path, "").unwrap();
assert_eq!(
load_config(&path, EnvOverrides::default())
.unwrap()
.resume_on_boot,
maestro_config::ResumeOnBoot::Dormant
);
std::fs::write(
&path,
r#"
[daemon]
resume_on_boot = "auto"
"#,
)
.unwrap();
assert_eq!(
load_config(&path, EnvOverrides::default())
.unwrap()
.resume_on_boot,
maestro_config::ResumeOnBoot::Auto
);
}
#[test]
fn empty_file_defaults_everything() {
let dir = tempfile::tempdir().unwrap();
+107 -37
View File
@@ -1,37 +1,53 @@
//! First-start hook auto-install (#106).
//!
//! The daemon runs [`auto_install`] once at startup: unless the user's
//! `~/.claude/settings.json` already carries exactly what this build ships
//! (and `[hooks].auto_install` hasn't opted out — checked by the caller), the
//! same installer `maestro hooks install` uses writes it. That covers
//! upgrades as well as first starts: an install from a maestro that had fewer
//! events — #131 added `UserPromptSubmit` to a hook #106 had already put in
//! everyone's settings — is rewritten rather than left half-wired. Best-effort:
//! a failure is warned about and never blocks startup. The same global hook
//! serves every Claude Code session on the machine, so nothing here targets
//! repositories or worktrees.
//! The daemon runs [`auto_install`] once at startup: for every agent maestro
//! knows how to hook ([`maestro_hooks::Target`]), unless that agent's hook
//! file already carries exactly what this build ships (and `[hooks]
//! .auto_install` hasn't opted out — checked by the caller), the same
//! installer `maestro hooks install` uses writes it. That covers upgrades as
//! well as first starts: an install from a maestro that had fewer events —
//! #131 added `UserPromptSubmit` to a hook #106 had already put in everyone's
//! settings — is rewritten rather than left half-wired. Best-effort: a
//! failure is warned about and never blocks startup, and one target failing
//! does not stop the others. The same global hook serves every session on the
//! machine, so nothing here targets repositories or worktrees.
use std::path::Path;
/// Install the Claude Code activity hook for `home_dir` unless the current
/// one is already there or the home dir is unresolvable (`""`). Returns what
/// happened; an older install being brought up to date is
/// [`AutoInstallOutcome::Upgraded`], so the log can say so.
pub fn auto_install(home_dir: &str) -> AutoInstallOutcome {
use maestro_hooks::Target;
/// Install the activity hook for every [`Target`] under `home_dir`, skipping
/// any whose install is already current. Returns what happened per target —
/// an older install being brought up to date is
/// [`AutoInstallOutcome::Upgraded`], so the log can say so. An unresolvable
/// home dir (`""`) reports every target as [`AutoInstallOutcome::Skipped`].
pub fn auto_install(home_dir: &str) -> Vec<(Target, AutoInstallOutcome)> {
Target::ALL
.iter()
.map(|&target| (target, auto_install_target(target, home_dir)))
.collect()
}
fn auto_install_target(target: Target, home_dir: &str) -> AutoInstallOutcome {
if home_dir.is_empty() {
return AutoInstallOutcome::Skipped;
}
let settings = maestro_hooks::settings_path(Path::new(home_dir));
let script = maestro_hooks::script_path(Path::new(home_dir));
if maestro_hooks::is_current(&settings, &script) {
let home = Path::new(home_dir);
let settings = target.settings_path(home);
let script = maestro_hooks::script_path(home);
if maestro_hooks::is_current(target, &settings, &script) {
return AutoInstallOutcome::AlreadyPresent;
}
let stale = maestro_hooks::is_installed(&settings);
match maestro_hooks::install_for_home(Path::new(home_dir)) {
match maestro_hooks::install_target(target, home) {
Ok(_) if stale => AutoInstallOutcome::Upgraded,
Ok(_) => AutoInstallOutcome::Installed,
Err(e) => {
tracing::warn!(error = %e, "hooks auto-install failed (non-fatal)");
tracing::warn!(
target = target.label(),
error = %e,
"hooks auto-install failed (non-fatal)"
);
AutoInstallOutcome::Failed
}
}
@@ -52,15 +68,50 @@ pub enum AutoInstallOutcome {
mod tests {
use super::*;
/// The outcome recorded for one target.
fn outcome(outcomes: &[(Target, AutoInstallOutcome)], target: Target) -> AutoInstallOutcome {
outcomes
.iter()
.find(|(t, _)| *t == target)
.map(|(_, o)| *o)
.expect("every target is reported")
}
/// Everything happens under a temp `$HOME`; the developer's own
/// `~/.claude/settings.json` is never read or written.
/// `~/.claude/settings.json` and `~/.codex/hooks.json` are never read or
/// written.
#[test]
fn installs_when_settings_carry_no_maestro_hook() {
let dir = tempfile::tempdir().unwrap();
let outcome = auto_install(dir.path().to_str().unwrap());
assert_eq!(outcome, AutoInstallOutcome::Installed);
let settings = dir.path().join(".claude/settings.json");
assert!(maestro_hooks::is_installed(&settings));
let outcomes = auto_install(dir.path().to_str().unwrap());
for target in Target::ALL {
assert_eq!(
outcome(&outcomes, target),
AutoInstallOutcome::Installed,
"{}",
target.label()
);
assert!(maestro_hooks::is_installed(
&target.settings_path(dir.path())
));
}
}
/// Every agent maestro knows about is hooked, not just Claude Code: a
/// codex session reports activity and its first prompt the same way.
#[test]
fn codex_gets_its_own_hook_file() {
let dir = tempfile::tempdir().unwrap();
auto_install(dir.path().to_str().unwrap());
let hooks = dir.path().join(".codex/hooks.json");
let root: serde_json::Value =
serde_json::from_str(&std::fs::read_to_string(&hooks).unwrap()).unwrap();
let events = root["hooks"].as_object().unwrap();
assert!(events.contains_key("PermissionRequest"));
assert!(
!events.contains_key("Notification"),
"codex has no Notification"
);
}
#[test]
@@ -69,14 +120,22 @@ mod tests {
let home = dir.path().to_str().unwrap();
auto_install(home);
// A current install is left exactly as it is.
let settings = dir.path().join(".claude/settings.json");
let before = std::fs::read_to_string(&settings).unwrap();
assert_eq!(auto_install(home), AutoInstallOutcome::AlreadyPresent);
assert_eq!(
before,
std::fs::read_to_string(&settings).unwrap(),
"already-installed settings must not be rewritten"
);
let before: Vec<String> = Target::ALL
.iter()
.map(|t| std::fs::read_to_string(t.settings_path(dir.path())).unwrap())
.collect();
let outcomes = auto_install(home);
for (i, target) in Target::ALL.iter().enumerate() {
assert_eq!(
outcome(&outcomes, *target),
AutoInstallOutcome::AlreadyPresent
);
assert_eq!(
before[i],
std::fs::read_to_string(target.settings_path(dir.path())).unwrap(),
"already-installed settings must not be rewritten"
);
}
}
/// An install from an older maestro — present, but missing an event this
@@ -87,7 +146,7 @@ mod tests {
let dir = tempfile::tempdir().unwrap();
let home = dir.path().to_str().unwrap();
auto_install(home);
let settings = dir.path().join(".claude/settings.json");
let settings = Target::ClaudeCode.settings_path(dir.path());
let script = maestro_hooks::script_path(dir.path());
let mut root: serde_json::Value =
@@ -96,12 +155,23 @@ mod tests {
std::fs::write(&settings, serde_json::to_string_pretty(&root).unwrap()).unwrap();
std::fs::write(&script, "#!/bin/sh\nexit 0\n").unwrap();
assert_eq!(auto_install(home), AutoInstallOutcome::Upgraded);
assert!(maestro_hooks::is_current(&settings, &script));
let outcomes = auto_install(home);
assert_eq!(
outcome(&outcomes, Target::ClaudeCode),
AutoInstallOutcome::Upgraded
);
assert!(maestro_hooks::is_current(
Target::ClaudeCode,
&settings,
&script
));
}
#[test]
fn unresolvable_home_is_skipped() {
assert_eq!(auto_install(""), AutoInstallOutcome::Skipped);
let outcomes = auto_install("");
for target in Target::ALL {
assert_eq!(outcome(&outcomes, target), AutoInstallOutcome::Skipped);
}
}
}
+1 -2
View File
@@ -12,7 +12,6 @@ pub mod attach;
/// Naming a session after its first prompt (#131).
#[cfg(feature = "haiku-autoname")]
pub mod autoname;
pub mod boot_id;
pub mod branches;
pub mod cli;
pub mod config;
@@ -22,10 +21,10 @@ pub mod hooks;
pub mod local_api;
pub mod panes;
pub mod proc;
pub mod projection;
pub mod projects;
pub mod pty;
pub mod release;
pub mod resume;
pub mod session;
pub mod sidecar;
pub mod spawn_history;
+2 -40
View File
@@ -52,7 +52,6 @@ pub fn router(state: LocalApiState) -> Router {
.route(routes::patterns::SESSION_ACTIVITY, post(report_activity))
.route(routes::patterns::SESSION_INPUT, post(send_input))
.route(routes::patterns::SESSION_FOCUS, post(set_focus))
.route(routes::patterns::SESSION_RESUME, post(resume_session))
.route(
routes::patterns::SESSION,
delete(kill_session).patch(rename_session),
@@ -171,9 +170,7 @@ fn session_error_status(code: &str) -> StatusCode {
// The session outlived its pane: nothing to type into, and nothing
// the daemon can do about it.
"not_running" => StatusCode::CONFLICT,
// Resume refusals (#166): the session exists and the request was
// well-formed, it just cannot be brought back in this state.
"already_running" | "cwd_missing" => StatusCode::CONFLICT,
"shutting_down" => StatusCode::SERVICE_UNAVAILABLE,
_ => StatusCode::BAD_GATEWAY,
}
}
@@ -311,11 +308,7 @@ async fn report_activity(
Path(id): Path<Uuid>,
Json(body): Json<ReportActivityRequest>,
) -> Response {
match s
.manager
.report_activity(id, body.state, body.detail, body.session_id)
.await
{
match s.manager.report_activity(id, body.state, body.detail).await {
Ok(()) => StatusCode::NO_CONTENT.into_response(),
Err(e) => {
let info = e.to_error_info();
@@ -372,27 +365,6 @@ async fn set_focus(
}
}
/// `POST /api/sessions/:id/resume` (#166): respawn a session that has no pane
/// under its own id and sidecar, so its name, `created_at`, env and worktree
/// ownership survive. Answers with the updated session, which also goes out as
/// `SessionUpdated` to every events subscriber.
///
/// Refused with `409` when the session is already running or its `cwd` is
/// gone; `404` when there is no such session.
async fn resume_session(State(s): State<LocalApiState>, Path(id): Path<Uuid>) -> Response {
match s.manager.resume(id, &s.presets).await {
Ok(session) => (StatusCode::OK, Json(session)).into_response(),
Err(e) => {
let info = e.to_error_info();
error_response(
session_error_status(&info.code),
info.code.as_str(),
info.message.as_str(),
)
}
}
}
fn error_response(status: StatusCode, code: &str, message: &str) -> Response {
(status, Json(ApiError::new(code, message))).into_response()
}
@@ -435,16 +407,6 @@ async fn attach_ws(
ws: axum::extract::WebSocketUpgrade,
) -> Result<Response, Response> {
let Some((replay, rx)) = s.manager.subscribe_pty_with_replay(id).await else {
// #166: a session the daemon knows but has no bridge for — a dormant
// one — is not missing, it is not running. Reporting it as unknown
// left the client unable to tell "gone" from "resumable".
if s.manager.has(id).await {
return Err(error_response(
StatusCode::CONFLICT,
"not_attachable",
"session has no live pane; resume it first",
));
}
return Err(error_response(
StatusCode::NOT_FOUND,
"unknown_session",
+608
View File
@@ -0,0 +1,608 @@
//! Fold persisted ACP notifications into client-neutral transcript and state.
use std::collections::{HashMap, HashSet};
use maestro_acp::schema::{ContentChunk, SessionNotification, SessionUpdate, ToolCall};
use maestro_protocol::{
ReducedState, TranscriptDelta, TranscriptRow, TranscriptRowKind, TranscriptSnapshot,
};
use serde_json::Value;
/// A transcript delta paired with the state after the same ACP notification.
#[derive(Debug, Clone, PartialEq)]
pub struct ProjectionUpdate {
pub transcript: Option<TranscriptDelta>,
pub state: ReducedState,
}
/// A prompt lifecycle transition supplied by the prompt transport (#138).
///
/// ACP notifications do not delimit turns, so replaying them must not infer
/// an in-flight prompt from an agent chunk.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TurnLifecycle {
Begin,
Finish,
Cancel,
Failure,
}
/// The one-pass ACP fold used for both persisted replay and live events.
#[derive(Debug, Default)]
pub struct Projection {
rows: Vec<TranscriptRow>,
row_index: HashMap<String, usize>,
tools: HashMap<String, ToolCall>,
pending_tool_updates: HashMap<String, Vec<maestro_acp::schema::ToolCallUpdateFields>>,
state: ReducedState,
turn_active: bool,
turn_tool_calls: HashSet<String>,
}
impl Projection {
/// Fold one durable event and return its client-visible changes.
pub fn apply(&mut self, seq: u64, notification: &SessionNotification) -> ProjectionUpdate {
let transcript = match &notification.update {
SessionUpdate::UserMessageChunk(chunk) => {
self.content(seq, chunk, TranscriptRowKind::User, "user")
}
SessionUpdate::AgentMessageChunk(chunk) => {
self.agent_text();
self.content(seq, chunk, TranscriptRowKind::AgentText, "agent")
}
SessionUpdate::AgentThoughtChunk(chunk) => {
self.agent_thought();
self.content(seq, chunk, TranscriptRowKind::AgentThought, "thought")
}
SessionUpdate::ToolCall(tool) => self.tool_call(tool),
SessionUpdate::ToolCallUpdate(update) => self.tool_update(update),
SessionUpdate::Plan(plan) => {
let data = to_value(plan);
self.state.plan = Some(data.clone());
Some(self.replace("plan".into(), TranscriptRowKind::Plan, data))
}
SessionUpdate::UsageUpdate(usage) => {
let value = to_value(usage);
self.state.usage = Some(value.clone());
self.state.context = Some(value);
None
}
SessionUpdate::CurrentModeUpdate(mode) => {
self.state.mode = Some(mode.current_mode_id.to_string());
None
}
SessionUpdate::ConfigOptionUpdate(options) => {
let value = to_value(options);
self.state.model = model_from_options(&value).or_else(|| self.state.model.clone());
None
}
// ACP deliberately adds notification variants over time. Known but
// non-transcript updates become a generic notice instead of being
// dropped or forcing clients to understand ACP's enum.
_ => Some(self.append(
format!("notice:{seq}"),
TranscriptRowKind::Notice,
to_value(&notification.update),
)),
};
ProjectionUpdate {
transcript,
state: self.state.clone(),
}
}
/// The current transcript at `cursor`.
pub fn snapshot(&self, cursor: u64) -> TranscriptSnapshot {
TranscriptSnapshot {
cursor,
rows: self.rows.clone(),
}
}
/// The state folded over every event seen so far.
pub fn state(&self) -> ReducedState {
self.state.clone()
}
/// Apply a transport-owned prompt lifecycle transition.
pub fn turn_lifecycle(&mut self, lifecycle: TurnLifecycle) -> ProjectionUpdate {
match lifecycle {
TurnLifecycle::Begin => {
self.turn_active = true;
self.state.turn.responding = true;
self.state.turn.thinking = true;
self.turn_tool_calls.clear();
self.state.turn.active_tool_calls.clear();
}
TurnLifecycle::Finish | TurnLifecycle::Cancel | TurnLifecycle::Failure => {
self.turn_active = false;
self.state.turn.responding = false;
self.state.turn.thinking = false;
self.turn_tool_calls.clear();
self.state.turn.active_tool_calls.clear();
}
}
ProjectionUpdate {
transcript: None,
state: self.state(),
}
}
fn content(
&mut self,
seq: u64,
chunk: &ContentChunk,
kind: TranscriptRowKind,
prefix: &str,
) -> Option<TranscriptDelta> {
let id = chunk
.message_id
.as_ref()
.map(|id| format!("{prefix}:{id}"))
.unwrap_or_else(|| format!("{prefix}:{seq}"));
let data = to_value(chunk);
if let Some(index) = self.row_index.get(&id).copied() {
let row = &mut self.rows[index];
append_text(&mut row.data, &data);
return Some(TranscriptDelta::Patch { row: row.clone() });
}
Some(self.append(id, kind, data))
}
fn tool_call(&mut self, tool: &ToolCall) -> Option<TranscriptDelta> {
let id = format!("tool:{}", tool.tool_call_id);
let tool_id = tool.tool_call_id.to_string();
let mut canonical = tool.clone();
if let Some(updates) = self.pending_tool_updates.get(&tool_id) {
for fields in updates {
canonical.update(fields.clone());
}
}
let kind = if tool_has_diff(tool) {
TranscriptRowKind::ToolDiff
} else {
TranscriptRowKind::ToolCall
};
self.tools.insert(tool_id, canonical.clone());
self.track_tool_call(&canonical);
Some(self.replace(
id,
if tool_has_diff(&canonical) {
TranscriptRowKind::ToolDiff
} else {
kind
},
to_value(&canonical),
))
}
fn tool_update(
&mut self,
update: &maestro_acp::schema::ToolCallUpdate,
) -> Option<TranscriptDelta> {
let tool_id = update.tool_call_id.to_string();
let Some(tool) = self.tools.get_mut(&tool_id) else {
self.pending_tool_updates
.entry(tool_id)
.or_default()
.push(update.fields.clone());
return None;
};
tool.update(update.fields.clone());
self.pending_tool_updates
.entry(tool_id.clone())
.or_default()
.push(update.fields.clone());
let row_id = format!("tool:{tool_id}");
let index = *self
.row_index
.get(&row_id)
.expect("a tracked tool call always has a transcript row");
let row = {
let row = &mut self.rows[index];
row.kind = if tool_has_diff(tool) {
TranscriptRowKind::ToolDiff
} else {
TranscriptRowKind::ToolCall
};
row.data = to_value(tool);
row.clone()
};
self.sync_active_tool(&tool_id);
Some(TranscriptDelta::Patch { row })
}
fn append(&mut self, id: String, kind: TranscriptRowKind, data: Value) -> TranscriptDelta {
let row = TranscriptRow { id, kind, data };
self.row_index.insert(row.id.clone(), self.rows.len());
self.rows.push(row.clone());
TranscriptDelta::Append { row }
}
fn replace(&mut self, id: String, kind: TranscriptRowKind, data: Value) -> TranscriptDelta {
let Some(index) = self.row_index.get(&id).copied() else {
return self.append(id, kind, data);
};
let row = &mut self.rows[index];
row.kind = kind;
row.data = data;
TranscriptDelta::Patch { row: row.clone() }
}
fn agent_text(&mut self) {
if self.turn_active {
self.state.turn.responding = true;
self.state.turn.thinking = false;
}
}
fn agent_thought(&mut self) {
if self.turn_active {
self.state.turn.responding = true;
self.state.turn.thinking = true;
}
}
fn track_tool_call(&mut self, tool: &ToolCall) {
if self.turn_active {
self.turn_tool_calls.insert(tool.tool_call_id.to_string());
self.sync_active_tool(&tool.tool_call_id.to_string());
}
}
fn sync_active_tool(&mut self, tool_id: &str) {
if !self.turn_active || !self.turn_tool_calls.contains(tool_id) {
return;
}
let active = self.tools.get(tool_id).is_some_and(|tool| {
matches!(
tool.status,
maestro_acp::schema::ToolCallStatus::Pending
| maestro_acp::schema::ToolCallStatus::InProgress
)
});
let active_tool_calls = &mut self.state.turn.active_tool_calls;
if active {
if !active_tool_calls.iter().any(|id| id == tool_id) {
active_tool_calls.push(tool_id.to_owned());
active_tool_calls.sort();
}
} else {
active_tool_calls.retain(|id| id != tool_id);
}
}
}
fn to_value(value: &impl serde::Serialize) -> Value {
serde_json::to_value(value).expect("ACP schema values serialize")
}
fn tool_has_diff(tool: &ToolCall) -> bool {
tool.content
.iter()
.any(|content| matches!(content, maestro_acp::schema::ToolCallContent::Diff(_)))
}
fn append_text(existing: &mut Value, next: &Value) {
let Some(existing_text) = existing.pointer_mut("/content/text") else {
*existing = next.clone();
return;
};
let Some(next_text) = next.pointer("/content/text").and_then(Value::as_str) else {
*existing = next.clone();
return;
};
if let Some(text) = existing_text.as_str() {
*existing_text = Value::String(format!("{text}{next_text}"));
} else {
*existing = next.clone();
}
}
fn model_from_options(value: &Value) -> Option<String> {
value
.get("configOptions")?
.as_array()?
.iter()
.find(|option| {
option.get("id").and_then(Value::as_str) == Some("model")
|| option.get("category").and_then(Value::as_str) == Some("model")
})?
.get("currentValue")?
.as_str()
.map(str::to_owned)
}
#[cfg(test)]
mod tests {
use maestro_acp::schema::SessionNotification;
use maestro_protocol::{TranscriptDelta, TranscriptRowKind};
use super::{Projection, TurnLifecycle};
fn event(value: serde_json::Value) -> SessionNotification {
serde_json::from_value(serde_json::json!({
"sessionId": "acp-1",
"update": value,
}))
.unwrap()
}
#[test]
fn folds_content_tool_diff_plan_and_notice_rows() {
let mut projection = Projection::default();
let events = [
event(
serde_json::json!({"sessionUpdate":"user_message_chunk","content":{"type":"text","text":"hi"}}),
),
event(
serde_json::json!({"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"hello"}}),
),
event(
serde_json::json!({"sessionUpdate":"agent_thought_chunk","content":{"type":"text","text":"think"}}),
),
event(
serde_json::json!({"sessionUpdate":"tool_call","toolCallId":"read-1","title":"Read","kind":"read"}),
),
event(
serde_json::json!({"sessionUpdate":"tool_call","toolCallId":"edit-1","title":"Edit","content":[{"type":"diff","path":"x.rs","oldText":"a","newText":"b"}]}),
),
event(serde_json::json!({"sessionUpdate":"plan","entries":[]})),
event(
serde_json::json!({"sessionUpdate":"available_commands_update","availableCommands":[]}),
),
];
let kinds: Vec<_> = events
.iter()
.enumerate()
.filter_map(|(i, event)| projection.apply((i + 1) as u64, event).transcript)
.map(|delta| match delta {
TranscriptDelta::Append { row } => row.kind,
other => panic!("expected append, got {other:?}"),
})
.collect();
assert_eq!(
kinds,
[
TranscriptRowKind::User,
TranscriptRowKind::AgentText,
TranscriptRowKind::AgentThought,
TranscriptRowKind::ToolCall,
TranscriptRowKind::ToolDiff,
TranscriptRowKind::Plan,
TranscriptRowKind::Notice,
]
);
}
#[test]
fn tool_update_patches_the_original_stable_row() {
let mut projection = Projection::default();
let first = projection.apply(
4,
&event(serde_json::json!({
"sessionUpdate":"tool_call", "toolCallId":"bash-7", "title":"Run"
})),
);
let updated = projection.apply(
5,
&event(serde_json::json!({
"sessionUpdate":"tool_call_update", "toolCallId":"bash-7", "status":"completed"
})),
);
let id = match first.transcript.unwrap() {
TranscriptDelta::Append { row } => row.id,
other => panic!("expected append, got {other:?}"),
};
match updated.transcript.unwrap() {
TranscriptDelta::Patch { row } => {
assert_eq!(row.id, id);
assert_eq!(row.data["status"], "completed");
}
other => panic!("expected patch, got {other:?}"),
}
assert_eq!(projection.snapshot(5).rows.len(), 1);
}
#[test]
fn duplicate_tool_calls_patch_one_stable_row() {
let mut projection = Projection::default();
projection.apply(
1,
&event(serde_json::json!({
"sessionUpdate":"tool_call", "toolCallId":"bash-7", "title":"Run"
})),
);
let duplicate = projection.apply(
2,
&event(serde_json::json!({
"sessionUpdate":"tool_call", "toolCallId":"bash-7", "title":"Run again"
})),
);
match duplicate.transcript.unwrap() {
TranscriptDelta::Patch { row } => {
assert_eq!(row.id, "tool:bash-7");
assert_eq!(row.data["title"], "Run again");
}
other => panic!("expected patch, got {other:?}"),
}
assert_eq!(projection.snapshot(2).rows.len(), 1);
}
#[test]
fn early_tool_update_is_reconciled_when_the_call_arrives() {
let mut projection = Projection::default();
let early = projection.apply(
1,
&event(serde_json::json!({
"sessionUpdate":"tool_call_update", "toolCallId":"bash-7", "status":"completed"
})),
);
assert_eq!(early.transcript, None);
let call = projection.apply(
2,
&event(serde_json::json!({
"sessionUpdate":"tool_call", "toolCallId":"bash-7", "title":"Run", "status":"pending"
})),
);
match call.transcript.unwrap() {
TranscriptDelta::Append { row } => {
assert_eq!(row.id, "tool:bash-7");
assert_eq!(row.data["status"], "completed");
}
other => panic!("expected append, got {other:?}"),
}
assert!(projection.state().turn.active_tool_calls.is_empty());
}
#[test]
fn unknown_tool_stays_generic_and_keeps_its_raw_name() {
let mut projection = Projection::default();
let update = projection.apply(
1,
&event(serde_json::json!({
"sessionUpdate":"tool_call", "toolCallId":"custom-1", "title":"Custom",
"name":"vendor.custom", "kind":"future_kind"
})),
);
match update.transcript.unwrap() {
TranscriptDelta::Append { row } => {
assert_eq!(row.kind, TranscriptRowKind::ToolCall);
assert_eq!(row.data["name"], "vendor.custom");
}
other => panic!("expected generic tool row, got {other:?}"),
}
}
#[test]
fn folds_usage_mode_model_and_plan_into_whole_state() {
let mut projection = Projection::default();
for (seq, update) in [
serde_json::json!({"sessionUpdate":"usage_update","used":12,"size":200}),
serde_json::json!({"sessionUpdate":"current_mode_update","currentModeId":"plan"}),
serde_json::json!({"sessionUpdate":"config_option_update","configOptions":[{"id":"model","name":"Model","category":"model","type":"select","currentValue":"opus","options":[]}]}),
serde_json::json!({"sessionUpdate":"plan","entries":[]}),
]
.into_iter()
.enumerate()
{
projection.apply((seq + 1) as u64, &event(update));
}
let state = projection.state();
assert_eq!(state.usage.as_ref().unwrap()["used"], 12);
assert_eq!(state.context.as_ref().unwrap()["size"], 200);
assert_eq!(state.mode.as_deref(), Some("plan"));
assert_eq!(state.model.as_deref(), Some("opus"));
assert_eq!(
state.plan.as_ref().unwrap()["entries"],
serde_json::json!([])
);
}
#[test]
fn plans_replace_one_stable_row_and_state() {
let mut projection = Projection::default();
let first = projection.apply(
1,
&event(serde_json::json!({"sessionUpdate":"plan","entries":[{"content":"first","priority":"high","status":"pending"}]})),
);
assert!(matches!(
first.transcript,
Some(TranscriptDelta::Append { .. })
));
let replacement = projection.apply(
2,
&event(serde_json::json!({"sessionUpdate":"plan","entries":[{"content":"second","priority":"high","status":"completed"}]})),
);
match replacement.transcript.unwrap() {
TranscriptDelta::Patch { row } => {
assert_eq!(row.id, "plan");
assert_eq!(row.data["entries"][0]["content"], "second");
}
other => panic!("expected patch, got {other:?}"),
}
assert_eq!(projection.snapshot(2).rows.len(), 1);
assert_eq!(
projection.state().plan.unwrap()["entries"][0]["content"],
"second"
);
}
#[test]
fn turn_lifecycle_gates_chunks_and_normalizes_terminal_states() {
let mut projection = Projection::default();
projection.apply(
1,
&event(serde_json::json!({"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"replay"}})),
);
assert_eq!(projection.state().turn, Default::default());
projection.turn_lifecycle(TurnLifecycle::Begin);
assert!(projection.state().turn.responding);
assert!(projection.state().turn.thinking);
projection.apply(
2,
&event(serde_json::json!({"sessionUpdate":"agent_thought_chunk","content":{"type":"text","text":"think"}})),
);
assert!(projection.state().turn.thinking);
projection.apply(
3,
&event(serde_json::json!({"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"answer"}})),
);
assert!(projection.state().turn.responding);
assert!(!projection.state().turn.thinking);
for terminal in [
TurnLifecycle::Finish,
TurnLifecycle::Cancel,
TurnLifecycle::Failure,
] {
projection.turn_lifecycle(TurnLifecycle::Begin);
projection.turn_lifecycle(terminal);
assert!(!projection.state().turn.responding);
assert!(!projection.state().turn.thinking);
}
}
fn terminal_turn_drops_unfinished_tools(
terminal: TurnLifecycle,
unfinished_id: &str,
next_id: &str,
) {
let mut projection = Projection::default();
projection.turn_lifecycle(TurnLifecycle::Begin);
projection.apply(
1,
&event(serde_json::json!({
"sessionUpdate":"tool_call", "toolCallId":unfinished_id,
"title":"Unfinished", "status":"in_progress"
})),
);
assert_eq!(projection.state().turn.active_tool_calls, [unfinished_id]);
projection.turn_lifecycle(terminal);
assert!(projection.state().turn.active_tool_calls.is_empty());
assert_eq!(projection.snapshot(1).rows.len(), 1);
projection.turn_lifecycle(TurnLifecycle::Begin);
projection.apply(
2,
&event(serde_json::json!({
"sessionUpdate":"tool_call", "toolCallId":next_id,
"title":"Next", "status":"pending"
})),
);
assert_eq!(projection.state().turn.active_tool_calls, [next_id]);
}
#[test]
fn cancelled_turn_does_not_reactivate_unfinished_tools() {
terminal_turn_drops_unfinished_tools(TurnLifecycle::Cancel, "cancelled-1", "next-1");
}
#[test]
fn failed_turn_does_not_reactivate_unfinished_tools() {
terminal_turn_drops_unfinished_tools(TurnLifecycle::Failure, "failed-1", "next-1");
}
}
-99
View File
@@ -1,99 +0,0 @@
//! Choosing the command a resumed session comes back on (#166).
//!
//! A session records the command it was spawned with. Bringing it back means
//! running that preset's `resume_command` (#165) instead, so an agent picks up
//! its old conversation rather than starting a fresh one. Everything else
//! about the session — id, sidecar, name, `created_at`, env, worktree — is
//! untouched, so only the command is decided here.
use maestro_protocol::{Session, SpawnPreset};
/// The command to respawn `session` with.
///
/// The preset is matched on the command the session was spawned with, which is
/// what the sidecar holds. A session with no matching preset, a preset with no
/// `resume_command`, or a `resume_command` needing an agent session id the
/// session never reported all fall back to the original command: respawning
/// what the operator asked for is always honest, where a half-filled resume
/// template is not.
pub fn resume_command(session: &Session, presets: &[SpawnPreset]) -> String {
presets
.iter()
.find(|p| p.command == session.command)
.and_then(|p| p.resume_command_for(session.agent_session_id.as_deref()))
.unwrap_or_else(|| session.command.clone())
}
#[cfg(test)]
mod tests {
use super::*;
use std::collections::HashMap;
fn presets() -> Vec<SpawnPreset> {
vec![
SpawnPreset {
id: "claude".into(),
label: "Claude".into(),
command: "claude".into(),
resume_command: Some("claude --resume {agent_session}".into()),
},
SpawnPreset {
id: "zsh".into(),
label: "Shell".into(),
command: "zsh".into(),
resume_command: None,
},
]
}
fn session(command: &str, agent_session_id: Option<&str>) -> Session {
Session {
id: uuid::Uuid::new_v4(),
cwd: "/tmp".into(),
command: command.into(),
env: HashMap::new(),
metadata: HashMap::new(),
worktree: None,
status: maestro_protocol::SessionStatus::Dormant,
created_at: chrono::Utc::now(),
foreground: None,
activity: maestro_protocol::ActivityState::Idle,
activity_detail: None,
last_activity_at: chrono::Utc::now(),
activity_since: chrono::Utc::now(),
current_cwd: None,
blocked_work: None,
last_focused_at: None,
agent_session_id: agent_session_id.map(str::to_string),
}
}
#[test]
fn uses_the_matching_presets_resume_command() {
assert_eq!(
resume_command(&session("claude", Some("conv-7")), &presets()),
"claude --resume conv-7"
);
}
#[test]
fn falls_back_when_the_agent_session_id_is_unknown() {
assert_eq!(
resume_command(&session("claude", None), &presets()),
"claude"
);
}
#[test]
fn falls_back_for_a_preset_without_a_resume_command() {
assert_eq!(resume_command(&session("zsh", None), &presets()), "zsh");
}
#[test]
fn falls_back_when_no_preset_matches() {
assert_eq!(
resume_command(&session("bash --noprofile", Some("conv-7")), &presets()),
"bash --noprofile"
);
}
}
File diff suppressed because it is too large Load Diff
+192 -6
View File
@@ -5,9 +5,22 @@
//! original `cwd`, command, env, metadata, etc. Files are deleted on
//! `KillSession`. Sessions whose sidecar is missing at recovery time are still
//! advertised but with empty metadata (handled by the caller, not here).
//!
//! A structured session (#135) also owns `<id>.events.jsonl` next to its
//! sidecar: every `session/update` notification its adapter sent, one JSON
//! object per line, appended before it is broadcast. The log is the source of
//! truth for the transcript; `seq` is the line's 1-based position, so a
//! client that holds seq `n` asks for [`EventCursor::Since`]`(n)`. A line
//! that does not parse — including the partial line a hard kill leaves at
//! the end — keeps its seq and is skipped on read. `delete` removes the log
//! with the sidecar; `load_all` never reads one as a sidecar.
use std::collections::VecDeque;
use std::fs::{File, OpenOptions};
use std::io::{BufRead, BufReader, Read, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};
use maestro_acp::schema::SessionNotification;
use maestro_protocol::Session;
use uuid::Uuid;
@@ -38,6 +51,120 @@ impl SidecarStore {
self.dir.join(format!("{id}.json"))
}
/// Where the per-session event log lives (#135).
pub fn events_path_for(&self, id: Uuid) -> PathBuf {
self.dir.join(format!("{id}.events.jsonl"))
}
/// Durable evidence that graceful daemon shutdown ended this structured
/// session and the next daemon may resume it.
pub fn recovery_path_for(&self, id: Uuid) -> PathBuf {
self.dir.join(format!("{id}.recover"))
}
/// Mark `id` as eligible for one structured-session recovery attempt.
pub fn mark_for_recovery(&self, id: Uuid) -> Result<(), SidecarError> {
std::fs::create_dir_all(&self.dir)?;
std::fs::write(self.recovery_path_for(id), [])?;
Ok(())
}
/// Consume the graceful-shutdown recovery marker for `id`.
pub fn take_recovery_marker(&self, id: Uuid) -> Result<bool, SidecarError> {
match std::fs::remove_file(self.recovery_path_for(id)) {
Ok(()) => Ok(true),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(false),
Err(error) => Err(error.into()),
}
}
/// Open the event log for `id` for appending, creating it (and the
/// directory) when missing. One handle per live session: the caller
/// keeps it for the session's lifetime rather than reopening per
/// notification.
///
/// A log whose last line has no newline — what a hard kill mid-write
/// leaves behind — is terminated first, so the next append starts a
/// fresh line instead of gluing onto the fragment. The fragment keeps
/// the seq of its line and reads as malformed.
pub fn open_events(&self, id: Uuid) -> Result<EventLog, SidecarError> {
std::fs::create_dir_all(&self.dir)?;
let mut file = OpenOptions::new()
.create(true)
.read(true)
.append(true)
.open(self.events_path_for(id))?;
let mut lines = 0u64;
let mut last = None;
let mut reader = BufReader::new(&mut file);
let mut buf = [0u8; 64 * 1024];
loop {
let n = reader.read(&mut buf)?;
if n == 0 {
break;
}
lines += buf[..n].iter().filter(|&&b| b == b'\n').count() as u64;
last = Some(buf[n - 1]);
}
if last.is_some_and(|b| b != b'\n') {
file.write_all(b"\n")?;
lines += 1;
}
file.seek(SeekFrom::End(0))?;
Ok(EventLog {
file,
last_seq: lines,
})
}
/// Read events from the log for `id`, at most `limit`, in ascending seq
/// order. A missing log is an empty log. Lines that fail to parse are
/// skipped but still consume a seq, so cursors stay stable.
pub fn read_events(
&self,
id: Uuid,
cursor: EventCursor,
limit: usize,
) -> Result<Vec<LoggedEvent>, SidecarError> {
let file = match File::open(self.events_path_for(id)) {
Ok(f) => f,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
Err(e) => return Err(e.into()),
};
if limit == 0 {
return Ok(Vec::new());
}
let mut out: VecDeque<LoggedEvent> = VecDeque::new();
let mut seq = 0u64;
for line in BufReader::new(file).split(b'\n') {
let line = line?;
seq += 1;
let keep = match cursor {
EventCursor::Since(since) => seq > since,
EventCursor::Before(Some(before)) => seq < before,
EventCursor::Before(None) => true,
};
if !keep {
if matches!(cursor, EventCursor::Before(Some(_))) {
break;
}
continue;
}
let Ok(notification) = serde_json::from_slice::<SessionNotification>(&line) else {
continue;
};
out.push_back(LoggedEvent { seq, notification });
match cursor {
EventCursor::Since(_) if out.len() >= limit => break,
EventCursor::Before(_) if out.len() > limit => {
out.pop_front();
}
_ => {}
}
}
Ok(out.into())
}
/// Atomically write `session` to `<dir>/<id>.json`. Creates the directory
/// if it does not exist.
pub fn write(&self, session: &Session) -> Result<(), SidecarError> {
@@ -59,13 +186,12 @@ impl SidecarStore {
}
}
/// Delete the sidecar for `id`. No-op if it does not exist.
/// Delete the sidecar, event log, and recovery marker for `id`. No-op for
/// whichever does not exist.
pub fn delete(&self, id: Uuid) -> Result<(), SidecarError> {
match std::fs::remove_file(self.path_for(id)) {
Ok(()) => Ok(()),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
Err(e) => Err(e.into()),
}
remove_if_present(&self.events_path_for(id))?;
remove_if_present(&self.recovery_path_for(id))?;
remove_if_present(&self.path_for(id))
}
/// Read every sidecar JSON in the directory. Files that fail to parse are
@@ -81,6 +207,8 @@ impl SidecarStore {
for entry in entries {
let entry = entry?;
let path = entry.path();
// `.events.jsonl` logs (#135) sit in the same directory and are
// never sidecars.
if path.extension().and_then(|s| s.to_str()) != Some("json") {
continue;
}
@@ -95,3 +223,61 @@ impl SidecarStore {
Ok(out)
}
}
fn remove_if_present(path: &Path) -> Result<(), SidecarError> {
match std::fs::remove_file(path) {
Ok(()) => Ok(()),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
Err(e) => Err(e.into()),
}
}
/// Which slice of a session's event log to read.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum EventCursor {
/// Forward: events with `seq` strictly greater than the cursor, the
/// first `limit` of them. `Since(0)` is the whole log from the start.
/// What a client catching up after a reconnect asks for.
Since(u64),
/// Backward: events with `seq` strictly less than the cursor, the last
/// `limit` of them, still returned in ascending order. `None` means the
/// tail of the log, for a first paint. The next page passes the smallest
/// seq it received.
Before(Option<u64>),
}
/// One line of a session's event log.
#[derive(Debug, Clone, PartialEq)]
pub struct LoggedEvent {
/// 1-based line position in the log; the cursor for [`EventCursor`].
pub seq: u64,
pub notification: SessionNotification,
}
/// An open, append-only handle on one session's event log. Obtained from
/// [`SidecarStore::open_events`]; held for the session's lifetime.
#[derive(Debug)]
pub struct EventLog {
file: File,
last_seq: u64,
}
impl EventLog {
/// The seq of the last line, `0` for an empty log.
pub fn last_seq(&self) -> u64 {
self.last_seq
}
/// Append one notification as a line and return its seq. The line is
/// handed to the kernel in a single write before this returns, so a
/// caller that broadcasts afterwards never announces an event the log
/// would not replay — short of the machine losing power.
pub fn append(&mut self, notification: &SessionNotification) -> Result<u64, SidecarError> {
let mut line = serde_json::to_vec(notification)?;
line.push(b'\n');
self.file.write_all(&line)?;
self.file.flush()?;
self.last_seq += 1;
Ok(self.last_seq)
}
}
+40 -12
View File
@@ -111,16 +111,33 @@ impl WorktreeManager {
// Clear registrations whose directory is gone first, so a
// previously-leaked entry doesn't push this branch to a suffix.
prune(base_repo)?;
// Anything occupying the target — a tree we left behind, one the
// operator made, an unrelated directory — is sidestepped with a
// numeric suffix on the directory; the branch stays whatever was
// requested (#77). The daemon never adopts what it finds.
let mut path = self.target_path(base_repo, branch);
let mut n = 1;
while path.symlink_metadata().is_ok() {
n += 1;
path = self.target_path_suffixed(base_repo, branch, n);
}
let (branch, path) = if create_branch {
// Worktrees are commonly abandoned without deleting their branch,
// so a random handle can collide with one of its own past runs.
// Unlike the `add_existing` arm below, the suffix goes on the
// branch name itself — the path is derived from it anyway.
let mut candidate = branch.to_string();
let mut path = self.target_path(base_repo, &candidate);
let mut n = 1;
while path.symlink_metadata().is_ok() || branch_exists(base_repo, &candidate) {
n += 1;
candidate = format!("{branch}-{n}");
path = self.target_path(base_repo, &candidate);
}
(candidate, path)
} else {
// Existing branch: the name must stay exact, so only the
// directory — anything occupying it, ours or the operator's — is
// sidestepped with a numeric suffix (#77). The daemon never
// adopts what it finds.
let mut path = self.target_path(base_repo, branch);
let mut n = 1;
while path.symlink_metadata().is_ok() {
n += 1;
path = self.target_path_suffixed(base_repo, branch, n);
}
(branch.to_string(), path)
};
// The parent (`<repo>/worktrees` or the override root) may not exist
// yet; the target itself must not.
if let Some(parent) = path.parent() {
@@ -129,10 +146,10 @@ impl WorktreeManager {
let mut cmd = Command::new("git");
cmd.arg("-C").arg(base_repo).args(["worktree", "add"]);
if create_branch {
cmd.arg("-b").arg(branch).arg(&path);
cmd.arg("-b").arg(&branch).arg(&path);
} else {
// Existing branch: `worktree add <path> <branch>` with no `-b`.
cmd.arg(&path).arg(branch);
cmd.arg(&path).arg(&branch);
}
let out = cmd.output()?;
check(out)?;
@@ -266,6 +283,17 @@ fn branch_relpath(branch: &str) -> Option<PathBuf> {
Some(candidate.to_path_buf())
}
/// Whether `refs/heads/<branch>` already exists in `base_repo`.
fn branch_exists(base_repo: &Path, branch: &str) -> bool {
Command::new("git")
.arg("-C")
.arg(base_repo)
.args(["show-ref", "--verify", "--quiet"])
.arg(format!("refs/heads/{branch}"))
.status()
.is_ok_and(|s| s.success())
}
/// `git -C <base_repo> worktree prune` — drops registrations whose directory
/// no longer exists.
fn prune(base_repo: &Path) -> Result<(), WorktreeError> {
+804
View File
@@ -0,0 +1,804 @@
//! 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::sync::Arc;
use std::time::{Duration, Instant};
use common::{init_bare_repo, init_repo, push_all_to_origin, wait_for_async, DaemonFixture};
use maestro_acp::schema::{ContentBlock, SessionUpdate};
use maestro_acp::{ADAPTER_BIN, INSTALL_COMMAND};
use maestro_daemon::projection::TurnLifecycle;
use maestro_daemon::sidecar::{EventCursor, SidecarStore};
use maestro_daemon::worktree::WorktreeManager;
use maestro_daemon::SessionManager;
use maestro_protocol::{
ActivityState, CreateSessionRequest, SessionStatus, SessionView, TranscriptDelta,
TranscriptRowKind, WorktreeSpec,
};
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()
.filter_map(|line| {
line.strip_prefix("start pid=")?
.split(' ')
.next()?
.parse()
.ok()
})
.next_back()
.expect("pid in log")
}
}
fn event_text(event: &maestro_daemon::sidecar::LoggedEvent) -> &str {
match &event.notification.update {
SessionUpdate::AgentMessageChunk(chunk) => match &chunk.content {
ContentBlock::Text(text) => &text.text,
other => panic!("unexpected content {other:?}"),
},
other => panic!("unexpected update {other:?}"),
}
}
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()));
wait_for_async(Duration::from_secs(5), || async {
fx.manager
.read_acp_events(session.id, EventCursor::Since(0), 10)
.is_ok_and(|events| !events.is_empty())
})
.await;
// Live delivery happens only after the notification is durable.
let (replay, mut updates) = fx
.manager
.subscribe_acp_events_with_replay(session.id, EventCursor::Since(0), 10)
.await
.expect("replay read")
.expect("event subscription");
assert_eq!(replay.len(), 1);
assert!(updates.try_recv().is_err(), "replay duplicated onto live");
unsafe {
libc::kill(fake.pid() as libc::pid_t, libc::SIGUSR1);
}
let live = tokio::time::timeout(Duration::from_secs(5), updates.recv())
.await
.expect("an update within 5s")
.expect("broadcast open");
assert_eq!(event_text(&live), "tick 1");
let stored = fx
.manager
.read_acp_events(session.id, EventCursor::Since(live.seq - 1), 1)
.expect("read event logged before broadcast");
assert_eq!(stored, vec![live]);
fx.manager.kill(session.id, false).await.expect("kill");
assert!(!fx
.sidecar_dir()
.join(format!("{}.events.jsonl", session.id))
.exists());
}
#[tokio::test]
async fn projection_snapshot_and_live_stream_share_one_fold() {
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");
wait_for_async(Duration::from_secs(5), || async {
fx.manager
.read_acp_events(session.id, EventCursor::Since(0), 10)
.is_ok_and(|events| events.len() == 1)
})
.await;
// The cursor does not truncate either projection: a reconnect needs the
// source row for future patches and whole reduced state from the session.
let (snapshot, state, mut live) = fx
.manager
.subscribe_acp_projection(session.id, EventCursor::Since(1))
.await
.expect("projection subscription")
.expect("structured session");
assert_eq!(snapshot.cursor, 1);
assert_eq!(snapshot.rows.len(), 1);
assert_eq!(snapshot.rows[0].kind, TranscriptRowKind::AgentText);
assert_eq!(state.turn, Default::default());
assert!(fx
.manager
.apply_acp_turn_lifecycle(session.id, TurnLifecycle::Begin)
.await
.expect("begin lifecycle"));
let begin = tokio::time::timeout(Duration::from_secs(5), live.recv())
.await
.expect("begin lifecycle broadcast")
.expect("projection broadcast");
assert_eq!(begin.transcript, None);
assert!(begin.state.turn.responding);
assert!(begin.state.turn.thinking);
unsafe {
libc::kill(fake.pid() as libc::pid_t, libc::SIGUSR1);
}
let agent_update = tokio::time::timeout(Duration::from_secs(5), live.recv())
.await
.expect("live projection")
.expect("projection broadcast");
match agent_update.transcript {
Some(TranscriptDelta::Append { row }) => {
assert_eq!(row.kind, TranscriptRowKind::AgentText);
assert_eq!(row.data["content"]["text"], "tick 1");
}
other => panic!("expected one appended transcript row, got {other:?}"),
}
assert!(agent_update.state.turn.responding);
assert!(!agent_update.state.turn.thinking);
assert!(fx
.manager
.apply_acp_turn_lifecycle(session.id, TurnLifecycle::Finish)
.await
.expect("finish lifecycle"));
let finish = tokio::time::timeout(Duration::from_secs(5), live.recv())
.await
.expect("finish lifecycle broadcast")
.expect("projection broadcast");
assert_eq!(finish.transcript, None);
assert_eq!(finish.state.turn, Default::default());
fx.manager.kill(session.id, false).await.expect("kill");
}
#[tokio::test]
async fn recovery_loads_the_acp_session_without_duplicating_history() {
let fx = DaemonFixture::new();
let fake = Fake::new();
let cwd = TempDir::new().unwrap();
let mut env = fake.env();
env.insert("FAKE_ACP_LOAD_REPLAY".into(), "2".into());
let session = fx
.manager
.create(structured(cwd.path(), env))
.await
.expect("create");
let acp_id = session.acp_session_id.clone().expect("acp id");
wait_for_async(Duration::from_secs(5), || async {
fx.manager
.read_acp_events(session.id, EventCursor::Since(0), 10)
.is_ok_and(|events| events.len() == 1)
})
.await;
unsafe {
libc::kill(fake.pid() as libc::pid_t, libc::SIGUSR1);
}
wait_for_async(Duration::from_secs(5), || async {
fx.manager
.read_acp_events(session.id, EventCursor::Since(0), 10)
.is_ok_and(|events| events.len() == 2)
})
.await;
let old_pid = fake.pid();
fx.manager.shutdown().await;
wait_for_exit(old_pid).await;
assert!(
fx.sidecar_dir()
.join(format!("{}.recover", session.id))
.exists(),
"graceful shutdown must record recovery provenance"
);
let revived = SessionManager::new(
Arc::new(fx.tmux.tmux.clone()),
SidecarStore::new(fx.sidecar_dir()),
WorktreeManager::new(None),
fx.fifo_dir(),
);
assert_eq!(revived.recover().await.expect("recover"), 1);
assert_eq!(
revived.status(session.id).await,
Some(SessionStatus::Running)
);
let log = fake.log();
assert!(
log.contains(&format!("recv session/load sessionId={acp_id}")),
"{log}"
);
wait_for_async(Duration::from_secs(5), || async {
revived
.read_acp_events(session.id, EventCursor::Since(0), 10)
.is_ok_and(|events| events.len() == 3)
})
.await;
let (replay, mut live) = revived
.subscribe_acp_events_with_replay(session.id, EventCursor::Since(0), 10)
.await
.expect("replay read")
.expect("recovered event subscription");
assert_eq!(replay.len(), 3);
assert!(live.try_recv().is_err(), "replay duplicated onto live");
let new_pid = fake.pid();
assert_ne!(new_pid, old_pid);
unsafe {
libc::kill(new_pid as libc::pid_t, libc::SIGUSR1);
}
let event = tokio::time::timeout(Duration::from_secs(5), live.recv())
.await
.expect("live event after recovery")
.expect("broadcast open");
assert_eq!(event.seq, 4);
assert_eq!(event_text(&event), "tick 1");
let stored = revived
.read_acp_events(session.id, EventCursor::Since(0), 10)
.expect("recovered log");
assert_eq!(stored.len(), 4, "session/load replay must not be appended");
assert_eq!(
stored.iter().map(event_text).collect::<Vec<_>>(),
["hello from fake", "tick 1", "hello after load", "tick 1"]
);
revived.kill(session.id, false).await.expect("kill");
}
#[tokio::test]
async fn shutdown_preserves_a_pristine_structured_worktree_for_recovery() {
let fx = DaemonFixture::new();
let fake = Fake::new();
let repo = TempDir::new().unwrap();
init_repo(repo.path());
let origin = TempDir::new().unwrap();
init_bare_repo(origin.path());
let branch = "feature/restart";
let worktree = repo
.path()
.join("worktrees")
.join("feature")
.join("restart");
let session = fx
.manager
.create(CreateSessionRequest {
view: SessionView::Structured,
cwd: repo.path().to_path_buf(),
command: "claude".into(),
env: fake.env(),
metadata: HashMap::new(),
worktree: Some(WorktreeSpec {
base_repo: repo.path().to_path_buf(),
branch: branch.into(),
create_branch: true,
}),
})
.await
.expect("create");
assert_eq!(session.cwd, worktree);
push_all_to_origin(repo.path(), origin.path());
fx.manager.shutdown().await;
assert!(
worktree.is_dir(),
"shutdown must not settle a pristine worktree"
);
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.worktree.as_ref().map(|wt| &wt.path),
Some(&worktree)
);
let revived = SessionManager::new(
Arc::new(fx.tmux.tmux.clone()),
SidecarStore::new(fx.sidecar_dir()),
WorktreeManager::new(None),
fx.fifo_dir(),
);
assert_eq!(revived.recover().await.expect("recover"), 1);
assert_eq!(
revived.status(session.id).await,
Some(SessionStatus::Running)
);
assert!(fake.log().contains("recv session/load"), "{}", fake.log());
assert!(worktree.is_dir(), "recovery must retain the worktree");
revived.kill(session.id, false).await.expect("dismiss");
assert!(
!worktree.exists(),
"operator dismissal still cleans a pristine tree"
);
}
#[tokio::test]
async fn failed_recovery_cleans_its_fifo() {
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");
let old_pid = fake.pid();
fx.manager.shutdown().await;
wait_for_exit(old_pid).await;
let events_path = fx
.sidecar_dir()
.join(format!("{}.events.jsonl", session.id));
std::fs::remove_file(&events_path).unwrap();
std::fs::create_dir(&events_path).unwrap();
let revived = SessionManager::new(
Arc::new(fx.tmux.tmux.clone()),
SidecarStore::new(fx.sidecar_dir()),
WorktreeManager::new(None),
fx.fifo_dir(),
);
assert!(revived.recover().await.is_err());
assert!(!fx.fifo_dir().join(format!("{}.fifo", session.id)).exists());
let failed_pid = fake.pid();
assert_ne!(failed_pid, old_pid);
wait_for_exit(failed_pid).await;
}
#[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 dismissal_during_shutdown_does_not_restore_recovery_artifacts() {
let fx = DaemonFixture::new();
let dismissed = Fake::new();
let blocker = Fake::new();
let cwd = TempDir::new().unwrap();
let mut env = dismissed.env();
env.insert("FAKE_ACP_EXIT_ON_CANCEL".into(), "1".into());
let session = fx
.manager
.create(structured(cwd.path(), env))
.await
.expect("create");
fx.manager
.create(structured(cwd.path(), blocker.env()))
.await
.expect("create blocker");
let manager = fx.manager.clone();
let shutdown = tokio::spawn(async move { manager.shutdown().await });
wait_for_async(Duration::from_secs(5), || async {
dismissed.log().contains("recv session/cancel")
})
.await;
fx.manager.kill(session.id, false).await.expect("dismiss");
assert!(
!fx.sidecar_dir()
.join(format!("{}.json", session.id))
.exists(),
"dismissal must finish before shutdown settles its stale snapshot"
);
shutdown.await.expect("shutdown task");
assert!(!fx.manager.has(session.id).await);
assert!(
!fx.sidecar_dir()
.join(format!("{}.recover", session.id))
.exists(),
"dismissal must leave no recovery marker"
);
assert!(
!fx.sidecar_dir()
.join(format!("{}.json", session.id))
.exists(),
"dismissal must leave no sidecar"
);
assert!(
!fx.sidecar_dir()
.join(format!("{}.events.jsonl", session.id))
.exists(),
"dismissal must leave no event log"
);
}
#[tokio::test]
async fn shutdown_does_not_recover_an_adapter_that_already_exited() {
let fx = DaemonFixture::new();
let fake = Fake::new();
let cwd = TempDir::new().unwrap();
let mut env = fake.env();
env.insert("FAKE_ACP_EXIT_AFTER_NEW".into(), "1".into());
let session = fx
.manager
.create(structured(cwd.path(), env))
.await
.expect("create");
wait_for_exit(fake.pid()).await;
fx.manager.shutdown().await;
assert!(
!fx.sidecar_dir()
.join(format!("{}.recover", session.id))
.exists(),
"a naturally exited adapter is not recoverable"
);
let revived = SessionManager::new(
Arc::new(fx.tmux.tmux.clone()),
SidecarStore::new(fx.sidecar_dir()),
WorktreeManager::new(None),
fx.fifo_dir(),
);
assert_eq!(revived.recover().await.expect("recover"), 1);
assert_eq!(revived.status(session.id).await, Some(SessionStatus::Dead));
assert!(
!fake.log().contains("recv session/load"),
"a naturally dead adapter must not be resumed: {}",
fake.log()
);
}
#[tokio::test]
async fn shutdown_waits_for_structured_creation_and_rejects_new_creates() {
let fx = DaemonFixture::new();
let fake = Fake::new();
let cwd = TempDir::new().unwrap();
let mut env = fake.env();
env.insert("FAKE_ACP_DELAY_SESSION_NEW_MS".into(), "500".into());
let create_cwd = cwd.path().to_path_buf();
let manager = fx.manager.clone();
let create = tokio::spawn(async move { manager.create(structured(&create_cwd, env)).await });
wait_for_async(Duration::from_secs(5), || async {
fake.log().contains("recv session/new")
})
.await;
let manager = fx.manager.clone();
let shutdown = tokio::spawn(async move { manager.shutdown().await });
tokio::time::sleep(Duration::from_millis(50)).await;
assert!(
!shutdown.is_finished(),
"shutdown returned before create finished"
);
let session = create.await.expect("create task").expect("create");
let pid = fake.pid();
shutdown.await.expect("shutdown task");
wait_for_exit(pid).await;
assert!(fake.log().contains("signal SIGTERM"), "{}", fake.log());
assert_eq!(
fx.manager.status(session.id).await,
Some(SessionStatus::Dead)
);
assert!(matches!(
fx.manager.create(structured(cwd.path(), fake.env())).await,
Err(maestro_daemon::SessionError::ShuttingDown)
));
// A repeated signal must not attempt to terminate the child again.
fx.manager.shutdown().await;
assert_eq!(fake.log().matches("signal SIGTERM").count(), 1);
}
#[tokio::test]
async fn shutdown_drains_final_notifications_to_the_log() {
let fx = DaemonFixture::new();
let fake = Fake::new();
let cwd = TempDir::new().unwrap();
let mut env = fake.env();
env.insert("FAKE_ACP_EMIT_ON_CANCEL".into(), "1".into());
let session = fx
.manager
.create(structured(cwd.path(), env))
.await
.expect("create");
wait_for_async(Duration::from_secs(5), || async {
fx.manager
.read_acp_events(session.id, EventCursor::Since(0), 10)
.is_ok_and(|events| events.len() == 1)
})
.await;
fx.manager.shutdown().await;
let stored = fx
.manager
.read_acp_events(session.id, EventCursor::Since(0), 10)
.expect("event log after shutdown");
assert_eq!(
stored.iter().map(event_text).collect::<Vec<_>>(),
["hello from fake", "final on cancel"]
);
}
#[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);
assert!(
!fx.sidecar_dir()
.join(format!("{}.recover", session.id))
.exists(),
"a naturally exited adapter is never marked for recovery"
);
let revived = SessionManager::new(
Arc::new(fx.tmux.tmux.clone()),
SidecarStore::new(fx.sidecar_dir()),
WorktreeManager::new(None),
fx.fifo_dir(),
);
assert_eq!(revived.recover().await.expect("recover"), 1);
assert_eq!(revived.status(session.id).await, Some(SessionStatus::Dead));
assert!(
!fake.log().contains("recv session/load"),
"a naturally dead adapter must not be resumed: {}",
fake.log()
);
revived.kill(session.id, false).await.expect("dismiss");
assert!(!revived.has(session.id).await);
}
+6 -85
View File
@@ -18,7 +18,6 @@ use uuid::Uuid;
struct Fixture {
manager: SessionManager,
sidecar: SidecarStore,
_tmux: TmuxFixture,
_state_dir: TempDir,
}
@@ -26,17 +25,15 @@ struct Fixture {
async fn fixture(ttl: Duration) -> Fixture {
let tmux = TmuxFixture::new();
let state_dir = TempDir::new().unwrap();
let sidecar = SidecarStore::new(state_dir.path().join("sessions"));
let manager = SessionManager::new(
Arc::new(tmux.tmux.clone()),
sidecar.clone(),
SidecarStore::new(state_dir.path().join("sessions")),
WorktreeManager::new(None),
state_dir.path().join("fifos"),
)
.with_activity_report_ttl(ttl);
Fixture {
manager,
sidecar,
_tmux: tmux,
_state_dir: state_dir,
}
@@ -45,6 +42,7 @@ 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(),
@@ -70,7 +68,7 @@ async fn report_on_unknown_session_is_not_found() {
let fx = fixture(Duration::from_secs(300)).await;
let err = fx
.manager
.report_activity(Uuid::new_v4(), ActivityState::Idle, None, None)
.report_activity(Uuid::new_v4(), ActivityState::Idle, None)
.await
.unwrap_err();
assert!(matches!(err, maestro_daemon::SessionError::NotFound(_)));
@@ -86,7 +84,6 @@ async fn report_sets_error_state_and_detail() {
id,
ActivityState::Error,
Some("cargo build failed: link error".into()),
None,
)
.await
.unwrap();
@@ -105,7 +102,7 @@ async fn replacing_a_report_replaces_both_state_and_detail() {
let id = spawn_sleeping(&fx.manager).await;
fx.manager
.report_activity(id, ActivityState::Error, Some("first failure".into()), None)
.report_activity(id, ActivityState::Error, Some("first failure".into()))
.await
.unwrap();
fx.manager
@@ -113,7 +110,6 @@ async fn replacing_a_report_replaces_both_state_and_detail() {
id,
ActivityState::AwaitingInput,
Some("needs approval".into()),
None,
)
.await
.unwrap();
@@ -130,7 +126,7 @@ async fn detail_is_capped_server_side() {
let long = "x".repeat(10_000);
fx.manager
.report_activity(id, ActivityState::Error, Some(long), None)
.report_activity(id, ActivityState::Error, Some(long))
.await
.unwrap();
@@ -149,12 +145,7 @@ async fn reported_state_survives_watcher_ticks_until_the_ttl_then_inference_resu
fx.manager.start_liveness_watcher(Duration::from_millis(50));
fx.manager
.report_activity(
id,
ActivityState::Error,
Some("agent died mid-run".into()),
None,
)
.report_activity(id, ActivityState::Error, Some("agent died mid-run".into()))
.await
.unwrap();
@@ -172,73 +163,3 @@ async fn reported_state_survives_watcher_ticks_until_the_ttl_then_inference_resu
})
.await;
}
#[tokio::test]
async fn a_reported_session_id_is_recorded_on_the_session() {
let fx = fixture(Duration::from_secs(300)).await;
let id = spawn_sleeping(&fx.manager).await;
fx.manager
.report_activity(id, ActivityState::Active, None, Some("conv-abc123".into()))
.await
.unwrap();
let s = observed(&fx.manager, id).await;
assert_eq!(s.agent_session_id.as_deref(), Some("conv-abc123"));
}
/// #161: a session id does not change mid-session, so once set it is never
/// overwritten by a later report — even one that names a different id.
#[tokio::test]
async fn a_reported_session_id_is_written_once_and_never_replaced() {
let fx = fixture(Duration::from_secs(300)).await;
let id = spawn_sleeping(&fx.manager).await;
fx.manager
.report_activity(id, ActivityState::Active, None, Some("conv-first".into()))
.await
.unwrap();
fx.manager
.report_activity(id, ActivityState::Active, None, Some("conv-second".into()))
.await
.unwrap();
let s = observed(&fx.manager, id).await;
assert_eq!(s.agent_session_id.as_deref(), Some("conv-first"));
}
#[tokio::test]
async fn a_reported_session_id_survives_recovery_from_the_sidecar() {
let fx = fixture(Duration::from_secs(300)).await;
let id = spawn_sleeping(&fx.manager).await;
fx.manager
.report_activity(id, ActivityState::Active, None, Some("conv-abc123".into()))
.await
.unwrap();
let stored = fx
.sidecar
.read(id)
.expect("sidecar read")
.expect("sidecar present");
assert_eq!(stored.agent_session_id.as_deref(), Some("conv-abc123"));
// The sidecar holding the id is only half the claim: a restarted daemon
// has to carry it back onto the recovered session, which is what a resume
// reads. Build a parallel manager over the same tmux server and state
// dirs and recover through it.
let revived = SessionManager::new(
Arc::new(fx._tmux.tmux.clone()),
fx.sidecar.clone(),
WorktreeManager::new(None),
fx._state_dir.path().join("fifos"),
);
revived.recover().await.expect("recover");
let recovered = observed(&revived, id).await;
assert_eq!(
recovered.agent_session_id.as_deref(),
Some("conv-abc123"),
"the recovered session must carry the agent session id, not just the sidecar"
);
}
+1 -68
View File
@@ -75,6 +75,7 @@ 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(),
@@ -94,74 +95,6 @@ async fn rejects_unknown_session() {
assert!(format!("{err}").contains("404"));
}
/// #166: a session the daemon knows but has no bridge for is not unknown, it
/// is not running. `404 unknown_session` made "gone" and "resumable" look the
/// same to a client that could have offered a resume.
#[tokio::test]
async fn dormant_session_is_not_attachable_rather_than_unknown() {
let h = spawn_daemon_ws().await;
let cwd = TempDir::new().unwrap();
let session = h
.manager
.create(CreateSessionRequest {
cwd: cwd.path().to_path_buf(),
command: "sleep 30".into(),
env: HashMap::new(),
metadata: HashMap::new(),
worktree: None,
})
.await
.unwrap();
// The host reboot: tmux is gone, the sidecar is not, and a fresh manager
// registers the session dormant.
h._fx.tmux.tmux.kill_server().unwrap();
let dormant = SessionManager::new(
std::sync::Arc::new(h._fx.tmux.tmux.clone()),
maestro_daemon::sidecar::SidecarStore::new(h._fx.sidecar_dir()),
maestro_daemon::worktree::WorktreeManager::new(None),
h._fx.fifo_dir(),
);
dormant.recover().await.unwrap();
let dormant_addr = serve_manager(dormant).await;
let err = connect_attach(dormant_addr, session.id).await.unwrap_err();
let tokio_tungstenite::tungstenite::Error::Http(resp) = err else {
panic!("want an HTTP rejection, got {err}");
};
assert_eq!(resp.status().as_u16(), 409);
let body: maestro_protocol::ApiError =
serde_json::from_slice(resp.body().as_deref().expect("error body")).unwrap();
assert_eq!(body.error.code, "not_attachable");
// A session that really is unknown still reads as unknown.
let err = connect_attach(dormant_addr, Uuid::new_v4())
.await
.unwrap_err();
assert!(format!("{err}").contains("404"));
}
/// Serve `manager` on its own listener, for tests that need a second daemon
/// over the same tmux/sidecar state.
async fn serve_manager(manager: SessionManager) -> SocketAddr {
let router = local_api::router(local_api::state_for(
manager,
"test-host",
"/home/test",
Vec::new(),
std::sync::Arc::new(maestro_daemon::projects::ProjectScanner::disabled()),
maestro_config::DEFAULT_LEADER,
));
let listener = TcpListener::bind(SocketAddr::from((Ipv4Addr::LOCALHOST, 0)))
.await
.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move {
let _ = axum::serve(listener, router).await;
});
addr
}
#[tokio::test]
async fn pty_round_trip_through_real_tmux() {
let h = spawn_daemon_ws().await;
+1
View File
@@ -50,6 +50,7 @@ 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,6 +19,7 @@ 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 -27
View File
@@ -174,6 +174,7 @@ 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(),
@@ -390,6 +391,7 @@ 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(),
@@ -432,6 +434,7 @@ 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(),
@@ -465,6 +468,7 @@ 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(),
@@ -486,6 +490,7 @@ 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(),
@@ -526,33 +531,6 @@ async fn post_activity_reports_error_with_detail() {
assert_eq!(s.activity_detail.as_deref(), Some("cargo build failed"));
}
/// #161: a reported `session_id` lands on the session as `agent_session_id`.
#[tokio::test]
async fn post_activity_records_the_reported_session_id() {
let h = spawn("naps-laptop").await;
let id = spawn_via_manager(&h).await;
let resp = reqwest::Client::new()
.post(url(h.addr, &format!("/api/sessions/{id}/activity")))
.json(&serde_json::json!({
"state": "Active",
"session_id": "conv-abc123"
}))
.send()
.await
.unwrap();
assert_eq!(resp.status(), reqwest::StatusCode::NO_CONTENT);
let sessions: Vec<Session> = reqwest::get(url(h.addr, "/api/sessions"))
.await
.unwrap()
.json()
.await
.unwrap();
let s = sessions.iter().find(|s| s.id == id).unwrap();
assert_eq!(s.agent_session_id.as_deref(), Some("conv-abc123"));
}
#[tokio::test]
async fn post_activity_to_unknown_session_returns_404() {
let h = spawn("naps-laptop").await;
@@ -622,6 +600,7 @@ 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(),
@@ -674,6 +653,7 @@ 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(),
@@ -707,6 +687,7 @@ 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(),
@@ -775,6 +756,7 @@ 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(),
@@ -819,6 +801,7 @@ 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,6 +51,7 @@ 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(),
-359
View File
@@ -1,359 +0,0 @@
//! End-to-end tests for `POST /api/sessions/:id/resume` (#166) and the
//! manager call behind it, against real tmux.
//!
//! The point of resume is that the session comes back under the *same* uuid
//! and sidecar — name, `created_at`, env and worktree ownership intact — with
//! only the tmux session and the PTY bridge new.
mod common;
use std::collections::HashMap;
use std::net::{Ipv4Addr, SocketAddr};
use std::sync::Arc;
use std::time::Duration;
use common::{wait_for_async, DaemonFixture};
use maestro_daemon::{local_api, SessionError, SessionManager};
use maestro_protocol::{CreateSessionRequest, Session, SessionStatus, SpawnPreset};
use tempfile::TempDir;
use tokio::net::TcpListener;
use tokio::sync::oneshot;
use uuid::Uuid;
/// A fresh manager over the same tmux server, sidecars and fifos — what the
/// daemon builds after a restart.
fn revive(fx: &DaemonFixture) -> SessionManager {
SessionManager::new(
Arc::new(fx.tmux.tmux.clone()),
maestro_daemon::sidecar::SidecarStore::new(fx.sidecar_dir()),
maestro_daemon::worktree::WorktreeManager::new(None),
fx.fifo_dir(),
)
}
/// Create a session, then simulate the host reboot that leaves it dormant:
/// the tmux server dies, the sidecar stays, a new manager recovers.
async fn dormant_session(
fx: &DaemonFixture,
cwd: &TempDir,
command: &str,
name: &str,
) -> (SessionManager, Session) {
let mut metadata = HashMap::new();
metadata.insert(
maestro_protocol::NAME_KEY.to_string(),
serde_json::Value::String(name.into()),
);
let session = fx
.manager
.create(CreateSessionRequest {
cwd: cwd.path().to_path_buf(),
command: command.into(),
env: HashMap::from([("MAESTRO_TEST_ENV".to_string(), "kept".to_string())]),
metadata,
worktree: None,
})
.await
.expect("create");
fx.tmux.tmux.kill_server().expect("kill tmux server");
let revived = revive(fx);
revived.recover().await.expect("recover");
(revived, session)
}
fn marker_preset(command: &str, marker_dir: &TempDir) -> Vec<SpawnPreset> {
vec![SpawnPreset {
id: "agent".into(),
label: "Agent".into(),
command: command.into(),
resume_command: Some(format!(
"sh -c 'touch {}/resumed-{{agent_session}}; sleep 30'",
marker_dir.path().display()
)),
}]
}
/// The keystone: a dormant session comes back running under its own id, with
/// its name, `created_at`, env and sidecar untouched.
#[tokio::test]
async fn resume_brings_a_dormant_session_back_under_the_same_id() {
let fx = DaemonFixture::new();
let cwd = TempDir::new().unwrap();
let (manager, original) = dormant_session(&fx, &cwd, "sleep 30", "long-runner").await;
assert_eq!(
manager.status(original.id).await,
Some(SessionStatus::Dormant)
);
let resumed = manager.resume(original.id, &[]).await.expect("resume");
assert_eq!(resumed.id, original.id, "same uuid");
assert_eq!(resumed.status, SessionStatus::Running);
assert_eq!(resumed.created_at, original.created_at);
assert_eq!(resumed.cwd, original.cwd);
assert_eq!(resumed.command, original.command);
assert_eq!(resumed.env, original.env);
assert_eq!(
resumed
.metadata
.get(maestro_protocol::NAME_KEY)
.and_then(|v| v.as_str()),
Some("long-runner")
);
// tmux really holds it again, and the sidecar is the same file, updated.
let live = fx.tmux.tmux.list_sessions().expect("list");
assert_eq!(live, vec![original.id]);
let stored = maestro_daemon::sidecar::SidecarStore::new(fx.sidecar_dir())
.read(original.id)
.expect("read sidecar")
.expect("sidecar still there");
assert_eq!(stored.status, SessionStatus::Running);
assert_eq!(stored.created_at, original.created_at);
// And it is attachable again, which a dormant session never was.
assert!(manager
.subscribe_pty_with_replay(original.id)
.await
.is_some());
}
/// The worktree the session owns stays the session's, so killing the resumed
/// session still cleans the tree up.
#[tokio::test]
async fn resume_keeps_worktree_ownership() {
let fx = DaemonFixture::new();
let repo = TempDir::new().unwrap();
common::init_repo(repo.path());
let session = fx
.manager
.create(CreateSessionRequest {
cwd: repo.path().to_path_buf(),
command: "sleep 30".into(),
env: HashMap::new(),
metadata: HashMap::new(),
worktree: Some(maestro_protocol::WorktreeSpec {
base_repo: repo.path().to_path_buf(),
branch: "feature".into(),
create_branch: true,
}),
})
.await
.expect("create");
let worktree = session.worktree.clone().expect("worktree");
fx.tmux.tmux.kill_server().expect("kill tmux server");
let manager = revive(&fx);
manager.recover().await.expect("recover");
let resumed = manager.resume(session.id, &[]).await.expect("resume");
assert_eq!(resumed.worktree, Some(worktree.clone()));
assert_eq!(
resumed.cwd, worktree.path,
"resumed inside its own worktree"
);
}
/// The command comes from the preset's `resume_command`, with the session's
/// own agent conversation id (#161) substituted in.
#[tokio::test]
async fn resume_runs_the_presets_resume_command() {
let fx = DaemonFixture::new();
let cwd = TempDir::new().unwrap();
let markers = TempDir::new().unwrap();
let presets = marker_preset("sleep 30", &markers);
let (manager, session) = dormant_session(&fx, &cwd, "sleep 30", "agentic").await;
// The agent reported its conversation id before the reboot, so it is on
// the sidecar and the resume template can be filled in.
fx.manager
.report_activity(
session.id,
maestro_protocol::ActivityState::Idle,
None,
Some("conv-42".into()),
)
.await
.ok();
// Re-read the sidecar the report updated.
let manager = {
drop(manager);
let m = revive(&fx);
m.recover().await.expect("recover");
m
};
manager.resume(session.id, &presets).await.expect("resume");
let marker = markers.path().join("resumed-conv-42");
wait_for_async(Duration::from_secs(5), || {
let marker = marker.clone();
async move { marker.exists() }
})
.await;
}
/// A session that is already running has nothing to bring back, and
/// respawning would strand the pane that is already there.
#[tokio::test]
async fn resume_refuses_a_running_session() {
let fx = DaemonFixture::new();
let cwd = TempDir::new().unwrap();
let session = fx
.manager
.create(CreateSessionRequest {
cwd: cwd.path().to_path_buf(),
command: "sleep 30".into(),
env: HashMap::new(),
metadata: HashMap::new(),
worktree: None,
})
.await
.expect("create");
let err = fx
.manager
.resume(session.id, &[])
.await
.expect_err("already running");
assert!(matches!(err, SessionError::AlreadyRunning(id) if id == session.id));
assert_eq!(err.to_error_info().code, "already_running");
}
/// Nothing can be resumed in a directory that no longer exists.
#[tokio::test]
async fn resume_refuses_when_the_cwd_is_gone() {
let fx = DaemonFixture::new();
let cwd = TempDir::new().unwrap();
let (manager, session) = dormant_session(&fx, &cwd, "sleep 30", "homeless").await;
std::fs::remove_dir_all(cwd.path()).unwrap();
let err = manager
.resume(session.id, &[])
.await
.expect_err("cwd is gone");
assert_eq!(err.to_error_info().code, "cwd_missing");
assert_eq!(
manager.status(session.id).await,
Some(SessionStatus::Dormant),
"a refused resume leaves the session exactly as it was"
);
}
#[tokio::test]
async fn resume_of_an_unknown_session_is_not_found() {
let fx = DaemonFixture::new();
let err = fx
.manager
.resume(Uuid::new_v4(), &[])
.await
.expect_err("unknown");
assert_eq!(err.to_error_info().code, "not_found");
}
// ---------------------------------------------------------------------------
// HTTP surface
// ---------------------------------------------------------------------------
struct Harness {
addr: SocketAddr,
_shutdown: oneshot::Sender<()>,
}
async fn serve(manager: SessionManager, presets: Vec<SpawnPreset>) -> Harness {
let router = local_api::router(local_api::state_for(
manager,
"test-host",
"/home/test",
presets,
Arc::new(maestro_daemon::projects::ProjectScanner::disabled()),
maestro_config::DEFAULT_LEADER,
));
let listener = TcpListener::bind(SocketAddr::from((Ipv4Addr::LOCALHOST, 0)))
.await
.unwrap();
let addr = listener.local_addr().unwrap();
let (tx, rx) = oneshot::channel::<()>();
tokio::spawn(async move {
let _ = axum::serve(listener, router)
.with_graceful_shutdown(async move {
let _ = rx.await;
})
.await;
});
Harness {
addr,
_shutdown: tx,
}
}
#[tokio::test]
async fn post_resume_returns_the_revived_session() {
let fx = DaemonFixture::new();
let cwd = TempDir::new().unwrap();
let (manager, session) = dormant_session(&fx, &cwd, "sleep 30", "web-resumed").await;
let h = serve(manager, Vec::new()).await;
let resp = reqwest::Client::new()
.post(format!(
"http://{}{}",
h.addr,
maestro_protocol::routes::session_resume(session.id)
))
.send()
.await
.unwrap();
assert_eq!(resp.status(), reqwest::StatusCode::OK);
let body: Session = resp.json().await.unwrap();
assert_eq!(body.id, session.id);
assert_eq!(body.status, SessionStatus::Running);
}
#[tokio::test]
async fn post_resume_on_a_running_session_is_a_conflict() {
let fx = DaemonFixture::new();
let cwd = TempDir::new().unwrap();
let session = fx
.manager
.create(CreateSessionRequest {
cwd: cwd.path().to_path_buf(),
command: "sleep 30".into(),
env: HashMap::new(),
metadata: HashMap::new(),
worktree: None,
})
.await
.expect("create");
let h = serve(fx.manager.clone(), Vec::new()).await;
let resp = reqwest::Client::new()
.post(format!(
"http://{}{}",
h.addr,
maestro_protocol::routes::session_resume(session.id)
))
.send()
.await
.unwrap();
assert_eq!(resp.status(), reqwest::StatusCode::CONFLICT);
let err: maestro_protocol::ApiError = resp.json().await.unwrap();
assert_eq!(err.error.code, "already_running");
}
#[tokio::test]
async fn post_resume_on_an_unknown_session_is_404() {
let fx = DaemonFixture::new();
let h = serve(fx.manager.clone(), Vec::new()).await;
let resp = reqwest::Client::new()
.post(format!(
"http://{}{}",
h.addr,
maestro_protocol::routes::session_resume(Uuid::new_v4())
))
.send()
.await
.unwrap();
assert_eq!(resp.status(), reqwest::StatusCode::NOT_FOUND);
}
@@ -1,161 +0,0 @@
//! `[daemon].resume_on_boot` (#167): what recovery does with sessions whose
//! tmux pane is gone but whose sidecar survived, against real tmux.
//!
//! The default is deliberately the timid one. Resuming agents unattended
//! seconds after a reboot is a foot-gun, so `auto` is opted into per host and
//! `never` throws the sidecars away instead.
mod common;
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
use common::{wait_for_async, DaemonFixture};
use maestro_config::ResumeOnBoot;
use maestro_daemon::SessionManager;
use maestro_protocol::{CreateSessionRequest, Session, SessionStatus, SpawnPreset};
use tempfile::TempDir;
/// A fresh manager over the same tmux server, sidecars and fifos — what the
/// daemon builds after a restart — carrying `policy`.
fn revive(fx: &DaemonFixture, policy: ResumeOnBoot, presets: Vec<SpawnPreset>) -> SessionManager {
SessionManager::new(
Arc::new(fx.tmux.tmux.clone()),
maestro_daemon::sidecar::SidecarStore::new(fx.sidecar_dir()),
maestro_daemon::worktree::WorktreeManager::new(None),
fx.fifo_dir(),
)
.with_resume_policy(policy, presets)
}
/// A preset whose `resume_command` touches a marker file, so a test can see
/// that the resumed session ran the resume command rather than the original.
fn marker_preset(command: &str, marker_dir: &TempDir) -> Vec<SpawnPreset> {
vec![SpawnPreset {
id: "agent".into(),
label: "Agent".into(),
command: command.into(),
resume_command: Some(format!(
"sh -c 'touch {}/resumed; sleep 30'",
marker_dir.path().display()
)),
}]
}
async fn create(fx: &DaemonFixture, cwd: &TempDir, command: &str) -> Session {
fx.manager
.create(CreateSessionRequest {
cwd: cwd.path().to_path_buf(),
command: command.into(),
env: HashMap::new(),
metadata: HashMap::new(),
worktree: None,
})
.await
.expect("create")
}
/// The default: a reboot leaves the session listed but not running, waiting
/// for the operator.
#[tokio::test]
async fn dormant_is_the_default_and_leaves_the_session_waiting() {
let fx = DaemonFixture::new();
let cwd = TempDir::new().unwrap();
let session = create(&fx, &cwd, "sleep 30").await;
fx.tmux.tmux.kill_server().expect("kill tmux server");
let revived = revive(&fx, ResumeOnBoot::Dormant, Vec::new());
revived.recover().await.expect("recover");
assert_eq!(
revived.status(session.id).await,
Some(SessionStatus::Dormant)
);
assert!(fx
.sidecar_dir()
.join(format!("{}.json", session.id))
.exists());
assert!(
fx.tmux.tmux.list_sessions().expect("list").is_empty(),
"nothing was respawned"
);
}
/// `auto` brings each one back itself, down the same path a client's resume
/// takes: same uuid, same sidecar, the preset's `resume_command`.
#[tokio::test]
async fn auto_resumes_every_dormant_session_during_recovery() {
let fx = DaemonFixture::new();
let cwd = TempDir::new().unwrap();
let markers = TempDir::new().unwrap();
let session = create(&fx, &cwd, "sleep 30").await;
fx.tmux.tmux.kill_server().expect("kill tmux server");
let revived = revive(&fx, ResumeOnBoot::Auto, marker_preset("sleep 30", &markers));
revived.recover().await.expect("recover");
assert_eq!(
revived.status(session.id).await,
Some(SessionStatus::Running),
"auto resumed it"
);
assert_eq!(
fx.tmux.tmux.list_sessions().expect("list"),
vec![session.id],
"back under the same uuid"
);
let stored = maestro_daemon::sidecar::SidecarStore::new(fx.sidecar_dir())
.read(session.id)
.expect("read sidecar")
.expect("same sidecar");
assert_eq!(stored.created_at, session.created_at);
assert_eq!(stored.command, "sleep 30", "the original command is kept");
// It really ran the preset's resume command, not the spawn command.
let marker = markers.path().join("resumed");
wait_for_async(Duration::from_secs(5), || {
let marker = marker.clone();
async move { marker.exists() }
})
.await;
assert!(marker.exists(), "resume_command ran");
}
/// `never` makes a reboot a clean slate: the sidecars go, and the session is
/// not listed at all.
#[tokio::test]
async fn never_prunes_the_sidecars_instead() {
let fx = DaemonFixture::new();
let cwd = TempDir::new().unwrap();
let session = create(&fx, &cwd, "sleep 30").await;
fx.tmux.tmux.kill_server().expect("kill tmux server");
let revived = revive(&fx, ResumeOnBoot::Never, Vec::new());
revived.recover().await.expect("recover");
assert!(!revived.has(session.id).await, "not listed");
assert!(
!fx.sidecar_dir()
.join(format!("{}.json", session.id))
.exists(),
"sidecar pruned"
);
assert!(fx.tmux.tmux.list_sessions().expect("list").is_empty());
}
/// The policy only decides what happens to sessions that lost their pane. A
/// session tmux still holds is adopted running by the first pass under every
/// policy — including `never`, which must not delete a live session's sidecar.
#[tokio::test]
async fn the_policy_leaves_sessions_tmux_still_holds_alone() {
let fx = DaemonFixture::new();
let cwd = TempDir::new().unwrap();
let live = create(&fx, &cwd, "sleep 30").await;
let revived = revive(&fx, ResumeOnBoot::Never, Vec::new());
assert_eq!(revived.recover().await.expect("recover"), 1);
assert_eq!(revived.status(live.id).await, Some(SessionStatus::Running));
assert!(fx.sidecar_dir().join(format!("{}.json", live.id)).exists());
}
@@ -64,6 +64,7 @@ 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(),
+26 -444
View File
@@ -9,7 +9,6 @@ use std::sync::Arc;
use std::time::Duration;
use common::{init_bare_repo, init_repo, push_all_to_origin, wait_for_async, DaemonFixture};
use maestro_daemon::sidecar::SidecarStore;
use maestro_daemon::SessionError;
use maestro_protocol::CreateSessionRequest;
use maestro_protocol::{CatalogEvent, SessionStatus, WorktreeSpec};
@@ -23,6 +22,7 @@ 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,6 +64,7 @@ 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(),
@@ -93,6 +94,7 @@ 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(),
@@ -115,6 +117,7 @@ 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(),
@@ -141,390 +144,6 @@ async fn recover_adopts_existing_tmux_sessions() {
assert_eq!(snap[0].command, "sleep 30");
}
/// #162: a host reboot kills the tmux server, so recovery finds nothing to
/// adopt while every sidecar is still on disk. Those sessions come back as
/// `Dormant` — a record of what was running — rather than vanishing from the
/// API and leaving the operator with nothing.
#[tokio::test]
async fn recover_registers_orphaned_sidecars_as_dormant() {
let fx = DaemonFixture::new();
let cwd = TempDir::new().unwrap();
let mut metadata = HashMap::new();
metadata.insert(
maestro_protocol::NAME_KEY.to_string(),
serde_json::Value::String("after-reboot".into()),
);
let s = fx
.manager
.create(CreateSessionRequest {
cwd: cwd.path().to_path_buf(),
command: "sleep 30".into(),
env: HashMap::new(),
metadata,
worktree: None,
})
.await
.expect("create");
// The reboot: the tmux server is gone, the sidecar is untouched.
fx.tmux.tmux.kill_server().expect("kill tmux server");
let revived = revive(&fx);
let adopted = revived.recover().await.expect("recover");
assert_eq!(adopted, 0, "no live tmux session left to adopt");
let snap = revived.snapshot().await;
assert_eq!(snap.len(), 1, "the sidecar's session is still listed");
assert_eq!(snap[0].id, s.id);
assert_eq!(snap[0].status, SessionStatus::Dormant);
assert_eq!(snap[0].command, "sleep 30");
assert_eq!(snap[0].cwd, s.cwd);
assert_eq!(snap[0].created_at, s.created_at);
assert_eq!(
snap[0]
.metadata
.get(maestro_protocol::NAME_KEY)
.and_then(|v| v.as_str()),
Some("after-reboot"),
);
}
/// #162: a session tmux still holds is adopted `Running` by the first pass and
/// must not be re-registered as dormant by the second — the sidecar is on disk
/// for both.
#[tokio::test]
async fn recover_does_not_mark_adopted_sessions_dormant() {
let fx = DaemonFixture::new();
let cwd = TempDir::new().unwrap();
let live = fx
.manager
.create(CreateSessionRequest {
cwd: cwd.path().to_path_buf(),
command: "sleep 30".into(),
env: HashMap::new(),
metadata: HashMap::new(),
worktree: None,
})
.await
.expect("create live");
let orphan = fx
.manager
.create(CreateSessionRequest {
cwd: cwd.path().to_path_buf(),
command: "sleep 31".into(),
env: HashMap::new(),
metadata: HashMap::new(),
worktree: None,
})
.await
.expect("create orphan");
// Only the second session loses its pane.
fx.tmux.tmux.kill_session(orphan.id).expect("kill session");
let revived = revive(&fx);
assert_eq!(revived.recover().await.expect("recover"), 1);
let mut snap = revived.snapshot().await;
snap.sort_by_key(|s| s.command.clone());
assert_eq!(snap.len(), 2);
assert_eq!(snap[0].id, live.id);
assert_eq!(snap[0].status, SessionStatus::Running);
assert_eq!(snap[1].id, orphan.id);
assert_eq!(snap[1].status, SessionStatus::Dormant);
}
/// #163: a sidecar whose `cwd` no longer exists is pruned instead of
/// registered dormant — the worktree it pointed at is gone, so there is
/// nothing left to resume.
#[tokio::test]
async fn recover_deletes_sidecar_when_cwd_gone() {
let fx = DaemonFixture::new();
let cwd = TempDir::new().unwrap();
let cwd_path = cwd.path().to_path_buf();
let s = fx
.manager
.create(CreateSessionRequest {
cwd: cwd_path.clone(),
command: "sleep 30".into(),
env: HashMap::new(),
metadata: HashMap::new(),
worktree: None,
})
.await
.expect("create");
fx.tmux.tmux.kill_server().expect("kill tmux server");
drop(cwd);
std::fs::remove_dir_all(&cwd_path).ok();
assert!(!cwd_path.exists());
let revived = revive(&fx);
let adopted = revived.recover().await.expect("recover");
assert_eq!(adopted, 0);
assert!(!revived.has(s.id).await);
assert!(!fx.sidecar_dir().join(format!("{}.json", s.id)).exists());
}
/// #163: a sidecar whose `cwd` still exists is left alone regardless of age
/// — it is registered dormant like any other orphan.
#[tokio::test]
async fn recover_keeps_dormant_sidecar_when_cwd_exists() {
let fx = DaemonFixture::new();
let cwd = TempDir::new().unwrap();
let s = fx
.manager
.create(CreateSessionRequest {
cwd: cwd.path().to_path_buf(),
command: "sleep 30".into(),
env: HashMap::new(),
metadata: HashMap::new(),
worktree: None,
})
.await
.expect("create");
fx.tmux.tmux.kill_server().expect("kill tmux server");
let revived = revive(&fx);
revived.recover().await.expect("recover");
assert!(revived.has(s.id).await);
assert!(fx.sidecar_dir().join(format!("{}.json", s.id)).exists());
let snap = revived.snapshot().await;
assert_eq!(snap[0].status, SessionStatus::Dormant);
}
/// #164: a fake [`maestro_daemon::boot_id::BootIdSource`] lets tests simulate
/// a changed (or unreadable) boot id without rebooting the machine.
#[derive(Debug)]
struct FakeBootId(Option<String>);
impl maestro_daemon::boot_id::BootIdSource for FakeBootId {
fn current(&self) -> Option<String> {
self.0.clone()
}
}
fn revive_with_boot_id(
fx: &DaemonFixture,
boot_id_path: PathBuf,
current: Option<&str>,
) -> maestro_daemon::SessionManager {
maestro_daemon::SessionManager::new(
Arc::new(fx.tmux.tmux.clone()),
maestro_daemon::sidecar::SidecarStore::new(fx.sidecar_dir()),
maestro_daemon::worktree::WorktreeManager::new(None),
fx.fifo_dir(),
)
.with_boot_id(
boot_id_path,
Arc::new(FakeBootId(current.map(String::from))),
)
}
/// #174: a boot id can change without the tmux server dying — a restored
/// state directory, a VM or container snapshot restore, a truncated
/// `boot_id` file. `tmux list-sessions` stays the single source of truth for
/// liveness, so a session tmux still holds is adopted `Running` no matter
/// what the boot id says. Registering it dormant instead stranded it: resume
/// would collide with the tmux name still in use and fail forever.
#[tokio::test]
async fn recover_adopts_live_session_when_boot_id_changed() {
let fx = DaemonFixture::new();
let cwd = TempDir::new().unwrap();
let s = fx
.manager
.create(CreateSessionRequest {
cwd: cwd.path().to_path_buf(),
command: "sleep 30".into(),
env: HashMap::new(),
metadata: HashMap::new(),
worktree: None,
})
.await
.expect("create");
let boot_id_path = fx.state_dir.path().join("boot_id");
std::fs::write(&boot_id_path, "previous-boot").unwrap();
// tmux session is still alive: the probe runs and finds it.
let revived = revive_with_boot_id(&fx, boot_id_path.clone(), Some("current-boot"));
let adopted = revived.recover().await.expect("recover");
assert_eq!(adopted, 1, "tmux still holds the session; adopt it");
let snap = revived.snapshot().await;
assert_eq!(snap.len(), 1);
assert_eq!(snap[0].id, s.id);
assert_eq!(snap[0].status, SessionStatus::Running);
// The current boot id is now the stored one, for the next comparison.
assert_eq!(
std::fs::read_to_string(&boot_id_path).unwrap().trim(),
"current-boot"
);
}
/// #164, still true after #174: a real reboot takes the tmux server with it,
/// so the probe comes back empty and every sidecar is registered dormant.
#[tokio::test]
async fn recover_registers_dormant_when_boot_id_changed_and_tmux_gone() {
let fx = DaemonFixture::new();
let cwd = TempDir::new().unwrap();
let s = fx
.manager
.create(CreateSessionRequest {
cwd: cwd.path().to_path_buf(),
command: "sleep 30".into(),
env: HashMap::new(),
metadata: HashMap::new(),
worktree: None,
})
.await
.expect("create");
let boot_id_path = fx.state_dir.path().join("boot_id");
std::fs::write(&boot_id_path, "previous-boot").unwrap();
fx.tmux.tmux.kill_server().expect("kill tmux server");
let revived = revive_with_boot_id(&fx, boot_id_path, Some("current-boot"));
let adopted = revived.recover().await.expect("recover");
assert_eq!(adopted, 0, "no tmux server survived the reboot");
let snap = revived.snapshot().await;
assert_eq!(snap.len(), 1);
assert_eq!(snap[0].id, s.id);
assert_eq!(snap[0].status, SessionStatus::Dormant);
}
/// #164: an unchanged boot id is a plain restart — recovery probes tmux as
/// before and adopts the still-live session `Running`.
#[tokio::test]
async fn recover_probes_tmux_when_boot_id_unchanged() {
let fx = DaemonFixture::new();
let cwd = TempDir::new().unwrap();
let s = fx
.manager
.create(CreateSessionRequest {
cwd: cwd.path().to_path_buf(),
command: "sleep 30".into(),
env: HashMap::new(),
metadata: HashMap::new(),
worktree: None,
})
.await
.expect("create");
let boot_id_path = fx.state_dir.path().join("boot_id");
std::fs::write(&boot_id_path, "same-boot").unwrap();
let revived = revive_with_boot_id(&fx, boot_id_path, Some("same-boot"));
let adopted = revived.recover().await.expect("recover");
assert_eq!(adopted, 1);
let snap = revived.snapshot().await;
assert_eq!(snap[0].id, s.id);
assert_eq!(snap[0].status, SessionStatus::Running);
}
/// #164: no stored boot id (first run ever, or an older build that never
/// wrote one) means there is nothing to compare against — falls back to
/// today's probe-based recovery rather than assuming a reboot.
#[tokio::test]
async fn recover_falls_back_to_probe_when_no_stored_boot_id() {
let fx = DaemonFixture::new();
let cwd = TempDir::new().unwrap();
fx.manager
.create(CreateSessionRequest {
cwd: cwd.path().to_path_buf(),
command: "sleep 30".into(),
env: HashMap::new(),
metadata: HashMap::new(),
worktree: None,
})
.await
.expect("create");
let boot_id_path = fx.state_dir.path().join("boot_id");
assert!(!boot_id_path.exists());
let revived = revive_with_boot_id(&fx, boot_id_path.clone(), Some("current-boot"));
let adopted = revived.recover().await.expect("recover");
assert_eq!(adopted, 1, "no stored boot id; falls back to probing tmux");
assert_eq!(
std::fs::read_to_string(&boot_id_path).unwrap().trim(),
"current-boot"
);
}
/// #164: an unreadable current boot id (non-Linux, restricted `/proc`) falls
/// back to today's behaviour rather than failing startup or assuming a
/// reboot.
#[tokio::test]
async fn recover_falls_back_to_probe_when_boot_id_unreadable() {
let fx = DaemonFixture::new();
let cwd = TempDir::new().unwrap();
fx.manager
.create(CreateSessionRequest {
cwd: cwd.path().to_path_buf(),
command: "sleep 30".into(),
env: HashMap::new(),
metadata: HashMap::new(),
worktree: None,
})
.await
.expect("create");
let boot_id_path = fx.state_dir.path().join("boot_id");
std::fs::write(&boot_id_path, "stored-boot").unwrap();
let revived = revive_with_boot_id(&fx, boot_id_path, None);
let adopted = revived.recover().await.expect("recover");
assert_eq!(adopted, 1, "boot id unreadable; falls back to probing tmux");
}
/// #162: dismissing a dormant session is the one thing that can be done to it
/// — no tmux call to fail, and the sidecar goes with it so the next recovery
/// does not resurrect it.
#[tokio::test]
async fn dormant_session_can_be_dismissed() {
let fx = DaemonFixture::new();
let cwd = TempDir::new().unwrap();
let s = fx
.manager
.create(CreateSessionRequest {
cwd: cwd.path().to_path_buf(),
command: "sleep 30".into(),
env: HashMap::new(),
metadata: HashMap::new(),
worktree: None,
})
.await
.expect("create");
fx.tmux.tmux.kill_server().expect("kill tmux server");
let revived = revive(&fx);
revived.recover().await.expect("recover");
revived.kill(s.id, false).await.expect("dismiss");
assert!(!revived.has(s.id).await);
assert!(!fx.sidecar_dir().join(format!("{}.json", s.id)).exists());
}
/// A second `SessionManager` over the same dirs and the same tmux server —
/// what a daemon restart looks like from inside a test.
fn revive(fx: &DaemonFixture) -> maestro_daemon::SessionManager {
maestro_daemon::SessionManager::new(
Arc::new(fx.tmux.tmux.clone()),
maestro_daemon::sidecar::SidecarStore::new(fx.sidecar_dir()),
maestro_daemon::worktree::WorktreeManager::new(None),
fx.fifo_dir(),
)
}
/// Regression for NAP-253: the very first bytes a freshly-spawned command
/// writes must reach the replay buffer. Before the fix, `pipe-pane` was a
/// separate tmux invocation that ran after `new-session` + an async FIFO
@@ -538,6 +157,7 @@ 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
@@ -585,6 +205,7 @@ 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
@@ -643,6 +264,7 @@ 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.
@@ -696,6 +318,7 @@ 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(),
@@ -758,6 +381,7 @@ 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(),
@@ -808,6 +432,7 @@ 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(),
@@ -873,6 +498,7 @@ 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(),
@@ -965,6 +591,7 @@ 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(),
@@ -1002,6 +629,7 @@ 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(),
@@ -1065,6 +693,7 @@ 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(),
@@ -1114,6 +743,7 @@ 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(),
@@ -1159,6 +789,7 @@ 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(),
@@ -1199,6 +830,7 @@ 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(),
@@ -1240,6 +872,7 @@ 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(),
@@ -1301,6 +934,7 @@ 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(),
@@ -1339,65 +973,6 @@ async fn natural_exit_of_a_pristine_worktree_removes_it() {
);
}
/// #172: `mark_session_dead` clears the worktree off the in-memory record
/// when a pristine tree settles, but the local `session` clone it writes to
/// the sidecar and broadcasts must be cleared the same way — otherwise the
/// sidecar keeps naming a tree the daemon just removed, and a recovery from
/// that sidecar advertises a branch and path that no longer exist.
#[tokio::test]
async fn natural_exit_of_a_pristine_worktree_clears_it_from_the_sidecar() {
let fx = DaemonFixture::new();
let repo = TempDir::new().unwrap();
init_repo(repo.path());
let wt_path = repo.path().join("worktrees").join("feat").join("sidecar");
let s = fx
.manager
.create(CreateSessionRequest {
cwd: repo.path().to_path_buf(),
command: "sleep 30".into(),
env: HashMap::new(),
metadata: HashMap::new(),
worktree: Some(WorktreeSpec {
base_repo: repo.path().to_path_buf(),
branch: "feat/sidecar".into(),
create_branch: true,
}),
})
.await
.expect("create");
// Push everything so the branch has no unpushed commits: fully pristine.
let origin = TempDir::new().unwrap();
init_bare_repo(origin.path());
push_all_to_origin(repo.path(), origin.path());
fx.tmux.tmux.kill_session(s.id).expect("kill tmux directly");
fx.manager.start_liveness_watcher(Duration::from_millis(50));
wait_for_async(Duration::from_secs(3), || async {
fx.manager.status(s.id).await == Some(SessionStatus::Dead)
})
.await;
wait_for_async(Duration::from_secs(3), || async {
fx.manager.is_settled(s.id).await == Some(true)
})
.await;
assert!(
!wt_path.exists(),
"a pristine worktree leaves nothing behind"
);
let sidecar = SidecarStore::new(fx.sidecar_dir())
.read(s.id)
.expect("sidecar read")
.expect("sidecar exists");
assert!(
sidecar.worktree.is_none(),
"the sidecar must not keep naming a worktree the daemon already removed"
);
}
fn run_git_in(dir: &std::path::Path, args: &[&str]) {
let out = Command::new("git")
.arg("-C")
@@ -1420,6 +995,7 @@ 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(),
@@ -1454,6 +1030,7 @@ 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(),
@@ -1471,6 +1048,7 @@ 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(),
@@ -1509,6 +1087,7 @@ 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(),
@@ -1547,6 +1126,7 @@ 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(),
@@ -1572,6 +1152,7 @@ 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(),
@@ -1620,6 +1201,7 @@ 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(),
+235 -2
View File
@@ -7,7 +7,7 @@ use std::path::PathBuf;
use chrono::{DateTime, Utc};
use maestro_daemon::sidecar::SidecarStore;
use maestro_protocol::{Session, SessionStatus};
use maestro_protocol::{Session, SessionStatus, SessionView};
use serde_json::Value;
use tempfile::TempDir;
use uuid::Uuid;
@@ -32,7 +32,8 @@ fn sample(id: Uuid) -> Session {
current_cwd: Some(PathBuf::from("/tmp/work/crates")),
blocked_work: None,
last_focused_at: None,
agent_session_id: Some("conv-abc123".into()),
acp_session_id: None,
view: SessionView::Terminal,
}
}
@@ -67,15 +68,30 @@ fn delete_removes_file_and_is_idempotent() {
let s = sample(id);
store.write(&s).unwrap();
store.mark_for_recovery(id).unwrap();
assert!(store.path_for(id).exists());
assert!(store.recovery_path_for(id).exists());
store.delete(id).unwrap();
assert!(!store.path_for(id).exists());
assert!(!store.recovery_path_for(id).exists());
// second delete is a no-op
store.delete(id).unwrap();
}
#[test]
fn recovery_marker_is_consumed_once() {
let dir = TempDir::new().unwrap();
let store = SidecarStore::new(dir.path());
let id = Uuid::new_v4();
assert!(!store.take_recovery_marker(id).unwrap());
store.mark_for_recovery(id).unwrap();
assert!(store.take_recovery_marker(id).unwrap());
assert!(!store.take_recovery_marker(id).unwrap());
}
#[test]
fn load_all_recovers_every_sidecar_and_skips_garbage() {
let dir = TempDir::new().unwrap();
@@ -104,6 +120,37 @@ 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();
@@ -114,3 +161,189 @@ fn write_creates_parent_directory() {
assert!(nested.exists());
assert!(store.path_for(s.id).exists());
}
// ---- per-session event log (#135) ----
use maestro_acp::schema::{ContentBlock, SessionNotification, SessionUpdate, TextContent};
use maestro_daemon::sidecar::EventCursor;
fn note(text: &str) -> SessionNotification {
SessionNotification::new(
"acp-1",
SessionUpdate::AgentMessageChunk(maestro_acp::schema::ContentChunk::new(
ContentBlock::Text(TextContent::new(text)),
)),
)
}
fn text_of(n: &SessionNotification) -> &str {
match &n.update {
SessionUpdate::AgentMessageChunk(chunk) => match &chunk.content {
ContentBlock::Text(t) => &t.text,
other => panic!("unexpected content {other:?}"),
},
other => panic!("unexpected update {other:?}"),
}
}
#[test]
fn append_assigns_line_position_as_seq_and_reads_back() {
let dir = TempDir::new().unwrap();
let store = SidecarStore::new(dir.path());
let id = Uuid::new_v4();
let mut log = store.open_events(id).expect("open");
assert_eq!(log.last_seq(), 0);
assert_eq!(log.append(&note("a")).unwrap(), 1);
assert_eq!(log.append(&note("b")).unwrap(), 2);
assert_eq!(log.append(&note("c")).unwrap(), 3);
assert_eq!(log.last_seq(), 3);
let path = store.events_path_for(id);
assert!(path.ends_with(format!("{id}.events.jsonl")), "{path:?}");
let raw = std::fs::read_to_string(&path).unwrap();
assert_eq!(raw.lines().count(), 3, "one object per line:\n{raw}");
for line in raw.lines() {
let v: Value = serde_json::from_str(line).expect("each line is one JSON object");
assert_eq!(v["sessionId"], "acp-1");
}
let all = store.read_events(id, EventCursor::Since(0), 100).unwrap();
assert_eq!(all.iter().map(|e| e.seq).collect::<Vec<_>>(), vec![1, 2, 3]);
assert_eq!(
all.iter()
.map(|e| text_of(&e.notification))
.collect::<Vec<_>>(),
vec!["a", "b", "c"]
);
}
#[test]
fn since_reads_forward_and_before_reads_backward_with_limit() {
let dir = TempDir::new().unwrap();
let store = SidecarStore::new(dir.path());
let id = Uuid::new_v4();
let mut log = store.open_events(id).unwrap();
for i in 1..=6 {
log.append(&note(&format!("n{i}"))).unwrap();
}
// `since` is exclusive: what a client that already holds seq 2 needs.
let fwd = store.read_events(id, EventCursor::Since(2), 3).unwrap();
assert_eq!(fwd.iter().map(|e| e.seq).collect::<Vec<_>>(), vec![3, 4, 5]);
// Past the end: nothing, not an error.
assert!(store
.read_events(id, EventCursor::Since(6), 3)
.unwrap()
.is_empty());
// `before` is exclusive and pages backward: the `limit` events closest
// to the cursor, still in ascending seq order.
let back = store
.read_events(id, EventCursor::Before(Some(5)), 2)
.unwrap();
assert_eq!(back.iter().map(|e| e.seq).collect::<Vec<_>>(), vec![3, 4]);
// No cursor: the tail of the log, for a first paint.
let tail = store.read_events(id, EventCursor::Before(None), 2).unwrap();
assert_eq!(tail.iter().map(|e| e.seq).collect::<Vec<_>>(), vec![5, 6]);
// Paging past the start runs out cleanly.
let start = store
.read_events(id, EventCursor::Before(Some(2)), 10)
.unwrap();
assert_eq!(start.iter().map(|e| e.seq).collect::<Vec<_>>(), vec![1]);
assert!(store
.read_events(id, EventCursor::Before(Some(1)), 10)
.unwrap()
.is_empty());
}
#[test]
fn missing_log_reads_as_empty() {
let dir = TempDir::new().unwrap();
let store = SidecarStore::new(dir.path());
let id = Uuid::new_v4();
assert!(store
.read_events(id, EventCursor::Since(0), 10)
.unwrap()
.is_empty());
assert!(store
.read_events(id, EventCursor::Before(None), 10)
.unwrap()
.is_empty());
}
#[test]
fn malformed_lines_are_skipped_but_keep_their_seq() {
let dir = TempDir::new().unwrap();
let store = SidecarStore::new(dir.path());
let id = Uuid::new_v4();
let mut log = store.open_events(id).unwrap();
log.append(&note("first")).unwrap();
drop(log);
// Garbage in the middle and a line that is JSON but not a notification.
let path = store.events_path_for(id);
let mut raw = std::fs::read_to_string(&path).unwrap();
raw.push_str("{not json\n");
raw.push_str("{\"unrelated\": true}\n");
std::fs::write(&path, raw).unwrap();
let mut log = store.open_events(id).unwrap();
assert_eq!(log.last_seq(), 3);
assert_eq!(log.append(&note("fourth")).unwrap(), 4);
let all = store.read_events(id, EventCursor::Since(0), 10).unwrap();
assert_eq!(all.iter().map(|e| e.seq).collect::<Vec<_>>(), vec![1, 4]);
assert_eq!(text_of(&all[1].notification), "fourth");
}
#[test]
fn truncated_last_line_still_opens_and_the_next_append_starts_a_fresh_line() {
// A hard kill mid-write leaves a partial line with no newline.
let dir = TempDir::new().unwrap();
let store = SidecarStore::new(dir.path());
let id = Uuid::new_v4();
let mut log = store.open_events(id).unwrap();
log.append(&note("whole")).unwrap();
drop(log);
let path = store.events_path_for(id);
let mut raw = std::fs::read(&path).unwrap();
raw.extend_from_slice(br#"{"sessionId":"acp-1","update":{"sessionUpd"#);
std::fs::write(&path, &raw).unwrap();
// Reads: the partial line is skipped, nothing else is lost.
let all = store.read_events(id, EventCursor::Since(0), 10).unwrap();
assert_eq!(all.iter().map(|e| e.seq).collect::<Vec<_>>(), vec![1]);
// Reopening terminates the partial line so the next append does not
// glue onto it; the partial keeps a seq of its own.
let mut log = store.open_events(id).unwrap();
assert_eq!(log.last_seq(), 2);
assert_eq!(log.append(&note("after")).unwrap(), 3);
let all = store.read_events(id, EventCursor::Since(0), 10).unwrap();
assert_eq!(all.iter().map(|e| e.seq).collect::<Vec<_>>(), vec![1, 3]);
assert_eq!(text_of(&all[1].notification), "after");
let raw = std::fs::read_to_string(&path).unwrap();
assert_eq!(raw.lines().count(), 3, "{raw}");
}
#[test]
fn delete_removes_the_event_log_and_load_all_ignores_it() {
let dir = TempDir::new().unwrap();
let store = SidecarStore::new(dir.path());
let id = Uuid::new_v4();
store.write(&sample(id)).unwrap();
let mut log = store.open_events(id).unwrap();
log.append(&note("x")).unwrap();
drop(log);
assert!(store.events_path_for(id).exists());
let loaded = store.load_all().unwrap();
assert_eq!(loaded.len(), 1, "the .jsonl must not be read as a sidecar");
store.delete(id).unwrap();
assert!(!store.path_for(id).exists());
assert!(!store.events_path_for(id).exists());
// Idempotent with the log already gone.
store.delete(id).unwrap();
}
+6 -4
View File
@@ -405,8 +405,10 @@ 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.
#[test]
fn add_fails_when_branch_already_exists() {
fn add_new_suffixes_when_branch_already_exists() {
let repo_dir = TempDir::new().unwrap();
let repo = repo_dir.path().to_path_buf();
init_repo(&repo);
@@ -423,9 +425,9 @@ fn add_fails_when_branch_already_exists() {
let wt_root = TempDir::new().unwrap();
let mgr = WorktreeManager::new(Some(wt_root.path().to_path_buf()));
let err = mgr.add_new(&repo, "feat/dup").unwrap_err();
let msg = format!("{err}");
assert!(msg.contains("git exited"), "{msg}");
let path = mgr.add_new(&repo, "feat/dup").unwrap();
assert!(path.ends_with("feat/dup-2"), "{}", path.display());
assert!(path.is_dir());
}
/// The project picker lists linked worktrees as pickable rows, so `base_repo`
+1
View File
@@ -13,6 +13,7 @@ path = "src/lib.rs"
[dependencies]
chrono.workspace = true
futures-util.workspace = true
libc.workspace = true
maestro-protocol = { workspace = true, features = ["client"] }
maestro-testing.workspace = true
serde.workspace = true
+8
View File
@@ -87,6 +87,14 @@ impl Daemon {
}
}
/// Send the daemon the same SIGTERM systemd uses, without waiting for its
/// graceful HTTP shutdown to finish.
pub fn send_sigterm(&self) {
let child = self.child.as_ref().expect("daemon is running");
let result = unsafe { libc::kill(child.id() as libc::pid_t, libc::SIGTERM) };
assert_eq!(result, 0, "send SIGTERM to daemon");
}
/// Re-spawn the daemon on the same port + state. The previous tmux
/// server (under `TMUX_TMPDIR`) is left running so `recover()` can adopt
/// its sessions.
+180 -4
View File
@@ -15,7 +15,7 @@ use std::time::{Duration, Instant};
use futures_util::{SinkExt, StreamExt};
use maestro_e2e::util::wait_for_async;
use maestro_e2e::{Daemon, TestClient};
use maestro_protocol::{CreateSessionRequest, WorktreeSpec};
use maestro_protocol::{CreateSessionRequest, SessionStatus, SessionView, WorktreeSpec};
use tempfile::TempDir;
use tokio_tungstenite::tungstenite::Message as WsMessage;
@@ -27,6 +27,7 @@ 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(),
@@ -35,6 +36,61 @@ fn create_req(cwd: &std::path::Path, command: &str) -> CreateSessionRequest {
}
}
struct FakeAcp {
_dir: TempDir,
env: HashMap<String, String>,
}
fn fake_acp(log: &std::path::Path) -> FakeAcp {
let dir = tempfile::tempdir().expect("fake adapter dir");
let adapter = dir.path().join("claude-code-acp");
std::fs::copy(
concat!(
env!("CARGO_MANIFEST_DIR"),
"/../maestro-acp/tests/fake_acp_agent.py"
),
&adapter,
)
.expect("copy fake adapter");
let mut permissions = std::fs::metadata(&adapter)
.expect("adapter metadata")
.permissions();
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
permissions.set_mode(0o755);
}
std::fs::set_permissions(&adapter, permissions).expect("make fake adapter executable");
let env = HashMap::from([
(
"PATH".into(),
format!(
"{}:{}",
dir.path().display(),
std::env::var("PATH").unwrap_or_default()
),
),
("FAKE_ACP_LOG".into(), log.to_string_lossy().into_owned()),
]);
FakeAcp { _dir: dir, env }
}
fn fake_acp_pid(log: &std::path::Path) -> Option<u32> {
std::fs::read_to_string(log).ok()?.lines().find_map(|line| {
line.strip_prefix("start pid=")?
.split(' ')
.next()?
.parse()
.ok()
})
}
fn pid_alive(pid: u32) -> bool {
std::fs::read_to_string(format!("/proc/{pid}/stat"))
.map(|stat| !stat.contains(") Z "))
.unwrap_or(false)
}
/// A worktree spec built through its wire shape. The daemon and clients ship
/// together (#87 dropped `cleanup_on_kill` from the struct), so the JSON
/// object carries only the fields the lifecycle actually honours.
@@ -49,6 +105,7 @@ 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(),
@@ -143,6 +200,7 @@ 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(),
@@ -196,6 +254,7 @@ 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(),
@@ -254,6 +313,7 @@ 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(),
@@ -323,8 +383,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 a branch that already has a worktree is rejected
/// with a structured worktree error; the first worktree stays intact.
/// #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`.)
#[tokio::test(flavor = "multi_thread")]
async fn second_spawn_on_same_branch_is_rejected() {
let (_daemon, client) = bring_up().await;
@@ -337,7 +400,7 @@ async fn second_spawn_on_same_branch_is_rejected() {
.await
.expect("first spawn");
let dup = client
.spawn_session(&worktree_req(base.path(), "feature/dup", true))
.spawn_session(&worktree_req(base.path(), "feature/dup", false))
.await;
match dup {
Err(maestro_e2e::ClientError::Api(maestro_protocol::ClientError::Api { info, .. })) => {
@@ -497,6 +560,57 @@ async fn occupied_path_gets_suffixed_directory_and_both_worktrees_work() {
);
}
/// A random handle can reuse the name of a branch left behind by an earlier,
/// already-removed worktree — `git worktree add -b` refuses outright in that
/// case, even though the directory is free. The daemon suffixes the branch
/// itself (not just the directory) and both land on `-2`.
#[tokio::test(flavor = "multi_thread")]
async fn colliding_branch_name_gets_suffixed_too() {
let (_daemon, client) = bring_up().await;
let base = TempDir::new().unwrap();
init_repo(base.path());
// No worktree directory in the way, but the branch already exists.
run_git(&["branch", "handle"], base.path());
let s = client
.spawn_session(&worktree_req(base.path(), "handle", true))
.await
.expect("spawn onto a taken branch name");
let suffixed = base.path().join("worktrees").join("handle-2");
assert_eq!(
s.cwd, suffixed,
"the directory must follow the suffixed branch name"
);
assert!(suffixed.is_dir());
let registrations = git_worktree_list(base.path());
assert!(
registrations.iter().any(|l| l.contains("handle-2")),
"the -2 branch's worktree must be registered: {registrations:?}"
);
let branches = String::from_utf8(
std::process::Command::new("git")
.arg("-C")
.arg(base.path())
.args(["branch", "--list"])
.output()
.unwrap()
.stdout,
)
.unwrap();
assert!(
branches.contains("handle-2"),
"the branch itself must be named handle-2: {branches}"
);
let _origin = throwaway_origin(base.path());
client.kill_session(s.id).await.expect("kill");
}
/// #77/#88: the target path holds a plain directory the daemon never made and
/// git has never heard of. The daemon neither adopts it nor errors — it builds
/// beside it and leaves every byte of it alone.
@@ -703,6 +817,7 @@ 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(),
@@ -751,6 +866,7 @@ 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(),
@@ -934,6 +1050,66 @@ async fn daemon_restart_recovers_session() {
client.kill_session(s.id).await.expect("kill");
}
#[tokio::test(flavor = "multi_thread")]
async fn sigterm_stops_structured_child_before_events_ws_drains() {
let (mut daemon, client) = bring_up().await;
let cwd = TempDir::new().unwrap();
let log = cwd.path().join("fake-acp.log");
let fake = fake_acp(&log);
let session = client
.spawn_session(&CreateSessionRequest {
view: SessionView::Structured,
cwd: cwd.path().to_path_buf(),
command: "claude".into(),
env: fake.env.clone(),
metadata: HashMap::new(),
worktree: None,
})
.await
.expect("spawn structured session");
let ws_url = daemon.client().events_url().expect("events URL");
let (_events_ws, _) = tokio_tungstenite::connect_async(ws_url)
.await
.expect("events WS handshake");
wait_for_async("fake ACP starts", Duration::from_secs(5), || {
let log = log.clone();
async move { fake_acp_pid(&log).is_some() }
})
.await;
let pid = fake_acp_pid(&log).expect("fake ACP pid");
daemon.send_sigterm();
wait_for_async(
"fake ACP exits after SIGTERM",
Duration::from_secs(5),
|| async { !pid_alive(pid) },
)
.await;
let sidecar_path = daemon
.state_dir()
.join("sessions")
.join(format!("{}.json", session.id));
wait_for_async(
"dead structured session sidecar",
Duration::from_secs(5),
|| {
let sidecar_path = sidecar_path.clone();
async move {
std::fs::read(&sidecar_path)
.ok()
.and_then(|json| {
serde_json::from_slice::<maestro_protocol::Session>(&json).ok()
})
.is_some_and(|saved| saved.status == SessionStatus::Dead)
}
},
)
.await;
daemon.stop();
}
#[tokio::test(flavor = "multi_thread")]
async fn concurrent_spawns_each_get_unique_ids() {
let (_daemon, client) = bring_up().await;
+144 -65
View File
@@ -1,4 +1,4 @@
//! Claude Code activity hook (#106).
//! Agent activity hook (#106).
//!
//! Two halves, both shared by the `maestro hooks install` CLI subcommand and
//! the daemon's first-start auto-install:
@@ -6,28 +6,40 @@
//! * [`HOOK_SCRIPT`] — the hook itself, written to disk at install time. It
//! posts an [`ActivityState`](maestro-protocol) to
//! `$MAESTRO_URL/api/sessions/$MAESTRO_SESSION_ID/activity` (#105). The same
//! global hook runs in every Claude Code session on the machine, so it exits
//! global hook runs in every agent session on the machine, so it exits
//! 0 and silently when either #103 env var is unset, and it never lets a
//! daemon failure escape (short curl timeout, status ignored).
//! * [`install`] — idempotent registration of that script in
//! `~/.claude/settings.json`. Only entries maestro itself wrote (identified
//! by [`HOOK_MARKER`]) are touched; every other hook and every other field
//! in the file is preserved verbatim.
//! * [`install`] — idempotent registration of that script in one [`Target`]'s
//! hook file. Only entries maestro itself wrote (identified by
//! [`HOOK_MARKER`]) are touched; every other hook and every other field in
//! the file is preserved verbatim.
//!
//! One script serves every [`Target`]: Claude Code and Codex read the same
//! `{"hooks": {Event: [{"hooks": [{"type": "command", "command": …}]}]}}`
//! shape and hand the hook the same JSON on stdin. Only the file's location
//! and the event set differ.
use std::path::Path;
/// Marker substring every maestro-installed hook command carries. The
/// installer rewrites or removes exactly these entries — nothing else in the
/// user's settings file.
///
/// It still says `claude` although the script now serves every [`Target`]:
/// the marker is how an install from an older maestro is recognised, and
/// renaming it would orphan every entry already on disk.
pub const HOOK_MARKER: &str = "maestro-claude-hook";
/// File name of the installed script (carries [`HOOK_MARKER`]).
/// File name of the installed script (carries [`HOOK_MARKER`]). One script
/// for every [`Target`] — see the note on [`HOOK_MARKER`] for the name.
pub const SCRIPT_FILE_NAME: &str = "maestro-claude-hook.sh";
/// Hook events installed only with the `haiku-autoname` feature (#131).
/// `UserPromptSubmit` carries the prompt the user just submitted; the script
/// forwards the whole payload to `/api/sessions/:id/prompt`, which names a
/// still-provisionally-named session after its first one.
/// Hook events installed only with the `haiku-autoname` feature (#131), for
/// every [`Target`] — Claude Code and Codex both spell the event
/// `UserPromptSubmit` and both put the submitted text in the payload's
/// `prompt`. The script forwards the whole payload to
/// `/api/sessions/:id/prompt`, which names a still-provisionally-named
/// session after its first one.
#[cfg(feature = "haiku-autoname")]
pub const PROMPT_EVENTS: [(&str, &str); 1] = [("UserPromptSubmit", "Prompt")];
@@ -43,7 +55,7 @@ pub const PROMPT_EVENTS: [(&str, &str); 0] = [];
/// Claude Code's own name for "the turn ended on an API error"; there is no
/// failure signal in `Stop` or `SessionEnd` payloads, so those never report
/// `Error`.
pub const HOOK_EVENTS: [(&str, &str); 5] = [
pub const CLAUDE_EVENTS: [(&str, &str); 5] = [
// Tool starting: the agent is demonstrably producing work, even if the
// tool itself runs for minutes.
("PreToolUse", "Active"),
@@ -58,17 +70,86 @@ pub const HOOK_EVENTS: [(&str, &str); 5] = [
("StopFailure", "Error"),
];
/// Codex hook event → `ActivityState` spelling posted to the daemon.
///
/// Codex's event set is close to Claude Code's but not identical: a pending
/// approval arrives as `PermissionRequest` rather than `Notification`, and
/// there is no `StopFailure`, so a codex session never reports `Error`.
pub const CODEX_EVENTS: [(&str, &str); 4] = [
("PreToolUse", "Active"),
("PostToolUse", "Active"),
// Codex is blocked on the operator approving a tool call.
("PermissionRequest", "AwaitingInput"),
("Stop", "AwaitingInput"),
];
/// An agent whose hook file maestro writes into.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Target {
/// `~/.claude/settings.json`.
ClaudeCode,
/// `~/.codex/hooks.json`. Codex asks the operator to trust a hook it has
/// not seen before, so an install here does nothing until they approve it
/// once inside codex — maestro never writes that approval itself.
Codex,
}
impl Target {
/// Every target an install covers. Both files are written whether or not
/// the agent is on the machine: an unused hook file costs nothing, and
/// the alternative is a session spawned after the daemon started missing
/// the hook because its agent arrived later.
pub const ALL: [Target; 2] = [Target::ClaudeCode, Target::Codex];
/// How the target is named in logs and CLI output.
pub fn label(self) -> &'static str {
match self {
Target::ClaudeCode => "claude-code",
Target::Codex => "codex",
}
}
/// The file this target's hook entries are registered in, under `home`.
pub fn settings_path(self, home: &Path) -> std::path::PathBuf {
match self {
Target::ClaudeCode => home.join(".claude/settings.json"),
Target::Codex => home.join(".codex/hooks.json"),
}
}
/// The activity events this target emits.
pub fn activity_events(self) -> &'static [(&'static str, &'static str)] {
match self {
Target::ClaudeCode => &CLAUDE_EVENTS,
Target::Codex => &CODEX_EVENTS,
}
}
/// Every event installed for this target: its activity events plus, with
/// `haiku-autoname` on, the prompt event both agents spell the same way.
pub fn events(self) -> impl Iterator<Item = (&'static str, &'static str)> {
self.activity_events()
.iter()
.chain(PROMPT_EVENTS.iter())
.copied()
}
}
/// The shipped hook script. POSIX sh so any `/bin/sh` works.
pub const HOOK_SCRIPT: &str = r#"#!/bin/sh
# Installed by maestro (#106): report Claude Code activity to its daemon.
# Exits silently unless running inside a session maestro spawned (#103 env).
# Installed by maestro (#106): report agent activity to its daemon.
# One script for every agent maestro installs hooks into; the state to post
# is chosen by the caller, so the two event sets differ only in what is
# registered where. Exits silently unless running inside a session maestro
# spawned (#103 env).
# Event -> ActivityState mapping:
# PreToolUse -> Active (tool started; agent is working)
# PostToolUse -> Active (tool finished)
# Notification -> AwaitingInput (permission / idle notice)
# Stop -> AwaitingInput (turn ended; waiting on the user)
# StopFailure -> Error (turn ended on an API error)
# Prompt -> the payload is forwarded to the prompt route (#131)
# PreToolUse -> Active (tool started; agent is working)
# PostToolUse -> Active (tool finished)
# Notification -> AwaitingInput (claude code: permission / idle)
# PermissionRequest -> AwaitingInput (codex: waiting on an approval)
# Stop -> AwaitingInput (turn ended; waiting on the user)
# StopFailure -> Error (claude code: turn ended on an error)
# Prompt -> the payload is forwarded to the prompt route (#131)
state="$1"
[ -n "$MAESTRO_SESSION_ID" ] || exit 0
[ -n "$MAESTRO_URL" ] || exit 0
@@ -84,23 +165,15 @@ if [ "$state" = "Prompt" ]; then
>/dev/null 2>&1 || true
exit 0
fi
fields="\"state\":\"$state\""
body="{\"state\":\"$state\"}"
# StopFailure payloads carry the outcome in "error". Only a plain enum
# spelling is reported as the detail — anything else (quotes, backslashes,
# prose) would risk breaking the JSON body, and the spelling already names
# the failure.
detail=$(printf '%s' "$payload" | sed -n 's/.*"error"[[:space:]]*:[[:space:]]*"\([a-z_]*\)".*/\1/p')
if [ -n "$detail" ]; then
fields="$fields,\"detail\":\"$detail\""
body="{\"state\":\"$state\",\"detail\":\"$detail\"}"
fi
# Every Claude Code hook payload carries its own conversation id (#161).
# Constrained to a safe charset for the same reason as "detail" above: it is
# spliced into the JSON body unescaped.
session_id=$(printf '%s' "$payload" | sed -n 's/.*"session_id"[[:space:]]*:[[:space:]]*"\([A-Za-z0-9_-]*\)".*/\1/p')
if [ -n "$session_id" ]; then
fields="$fields,\"session_id\":\"$session_id\""
fi
body="{$fields}"
curl -fsS --connect-timeout 1 --max-time 2 \
-H 'Content-Type: application/json' \
-d "$body" \
@@ -123,34 +196,29 @@ pub fn script_path(home: &Path) -> std::path::PathBuf {
home.join(".local/share/maestro").join(SCRIPT_FILE_NAME)
}
/// The user-level Claude Code settings file for a given home directory.
pub fn settings_path(home: &Path) -> std::path::PathBuf {
home.join(".claude/settings.json")
}
/// Whether `settings_path` already carries a maestro hook (#106). Presence
/// only: an install from an older release answers `true` even when it is
/// missing events this build ships — see [`is_current`].
/// Whether the file at `settings_path` already carries a maestro hook
/// (#106). Presence only: an install from an older release answers `true`
/// even when it is missing events this build ships — see [`is_current`].
pub fn is_installed(settings_path: &Path) -> bool {
settings_json(settings_path)
.map(|root| find_maestro_commands(&root).next().is_some())
.unwrap_or(false)
}
/// Whether what is on disk is what this build ships: the script at
/// `script_path` byte-for-byte, and exactly one settings entry per event in
/// [`HOOK_EVENTS`] and [`PROMPT_EVENTS`], pointing at that script.
/// Whether what is on disk is what this build ships for `target`: the script
/// at `script_path` byte-for-byte, and exactly one settings entry per one of
/// the target's [`Target::events`], pointing at that script.
///
/// This is the upgrade question. A maestro that gains an event — #131 added
/// `UserPromptSubmit` long after #106 had installed itself — leaves every
/// existing user with a present but incomplete install, which
/// [`is_installed`] cannot tell apart from a current one.
pub fn is_current(settings_path: &Path, script_path: &Path) -> bool {
/// This is the upgrade question, asked per [`Target`]. A maestro that gains
/// an event — #131 added `UserPromptSubmit` long after #106 had installed
/// itself — leaves every existing user with a present but incomplete
/// install, which [`is_installed`] cannot tell apart from a current one.
pub fn is_current(target: Target, settings_path: &Path, script_path: &Path) -> bool {
if std::fs::read_to_string(script_path).ok().as_deref() != Some(HOOK_SCRIPT) {
return false;
}
settings_json(settings_path)
.map(|root| maestro_entries(&root) == expected_entries(script_path))
.map(|root| maestro_entries(&root) == expected_entries(target, script_path))
.unwrap_or(false)
}
@@ -181,14 +249,13 @@ fn maestro_entries(root: &serde_json::Value) -> Vec<(String, String)> {
found
}
/// What [`install`] writes for `script_path`, in the same shape.
fn expected_entries(script_path: &Path) -> Vec<(String, String)> {
let mut want: Vec<(String, String)> = HOOK_EVENTS
.iter()
.chain(PROMPT_EVENTS.iter())
/// What [`install`] writes for `target` and `script_path`, in the same shape.
fn expected_entries(target: Target, script_path: &Path) -> Vec<(String, String)> {
let mut want: Vec<(String, String)> = target
.events()
.map(|(event, state)| {
(
(*event).to_string(),
event.to_string(),
format!("{} {state}", script_path.display()),
)
})
@@ -215,28 +282,40 @@ pub fn write_script(path: &Path) -> Result<(), HooksError> {
Ok(())
}
/// Install the full hook setup for `home`: write the script under
/// `.local/share/maestro/` and register it in `.claude/settings.json`.
/// Install the full hook setup for one target under `home`: write the script
/// under `.local/share/maestro/` and register it in the target's hook file.
/// Idempotent; returns whether anything changed on disk.
pub fn install_for_home(home: &Path) -> Result<bool, HooksError> {
pub fn install_target(target: Target, home: &Path) -> Result<bool, HooksError> {
let script = script_path(home);
let settings = settings_path(home);
let settings = target.settings_path(home);
// A script left over from an older release counts as a change even when
// the settings entries pointing at it are already right.
let changed = !is_current(&settings, &script);
let changed = !is_current(target, &settings, &script);
write_script(&script)?;
install(&settings, &script)?;
install(target, &settings, &script)?;
Ok(changed)
}
/// Register the hook command for every event in [`HOOK_EVENTS`] in the
/// settings file at `settings_path`. Idempotent: existing maestro entries are
/// updated in place wherever they sit, stale ones removed, duplicates folded;
/// every non-maestro entry and every unrelated top-level field survives.
/// Install for every [`Target::ALL`], reporting each one separately. One
/// target failing — an unwritable file, unparseable JSON — never stops the
/// others: an agent whose hook could not be written is one agent without
/// activity reporting, not a failed install.
pub fn install_for_home(home: &Path) -> Vec<(Target, Result<bool, HooksError>)> {
Target::ALL
.iter()
.map(|&target| (target, install_target(target, home)))
.collect()
}
/// Register the hook command for every one of `target`'s [`Target::events`]
/// in the settings file at `settings_path`. Idempotent: existing maestro
/// entries are updated in place wherever they sit, stale ones removed,
/// duplicates folded; every non-maestro entry and every unrelated top-level
/// field survives.
///
/// Never inspects or writes anywhere but `settings_path` — callers pass the
/// user's real home or a temp one in tests, never a repository.
pub fn install(settings_path: &Path, script_path: &Path) -> Result<(), HooksError> {
pub fn install(target: Target, settings_path: &Path, script_path: &Path) -> Result<(), HooksError> {
let mut root = settings_json(settings_path)?;
// Strip every maestro entry wherever it currently sits, so an older
@@ -244,7 +323,7 @@ pub fn install(settings_path: &Path, script_path: &Path) -> Result<(), HooksErro
strip_maestro_entries(&mut root);
let hooks = ensure_object(&mut root, "hooks");
for (event, state) in HOOK_EVENTS.iter().chain(PROMPT_EVENTS.iter()).copied() {
for (event, state) in target.events() {
let groups = ensure_array(hooks, event);
let mut group = serde_json::Map::new();
group.insert(
+136 -67
View File
@@ -1,15 +1,24 @@
//! Tests for the shipped hook script (run as a real `/bin/sh` subprocess)
//! and the settings-file installer. Everything happens under a temp `$HOME`;
//! the developer's own `~/.claude/settings.json` is never touched.
//! the developer's own `~/.claude/settings.json` and `~/.codex/hooks.json`
//! are never touched.
use std::io::{Read, Write};
use std::process::{Command, Stdio};
use super::*;
/// How many commands one install writes: every activity event, plus the
/// prompt event when `haiku-autoname` is on (#131).
const INSTALLED_EVENTS: usize = HOOK_EVENTS.len() + PROMPT_EVENTS.len();
/// How many commands one install writes for a target: its activity events,
/// plus the prompt event when `haiku-autoname` is on (#131).
fn installed_events(target: Target) -> usize {
target.events().count()
}
/// The Claude Code hook file under a temp home. Most installer tests only
/// need one target, and this is the one they were written against.
fn settings_path(home: &Path) -> std::path::PathBuf {
Target::ClaudeCode.settings_path(home)
}
fn sh(script: &str, args: &[&str], envs: &[(&str, &str)], payload: &[u8]) -> std::process::Output {
let mut cmd = Command::new("/bin/sh");
@@ -70,7 +79,7 @@ fn owned_envs<'a>(envs: &'a [(&'static str, String)]) -> Vec<(&'static str, &'a
/// exit 0, no output, stdin drained.
#[test]
fn script_exits_silently_without_maestro_env() {
for (event, _) in HOOK_EVENTS {
for (event, _) in CLAUDE_EVENTS {
let out = sh(HOOK_SCRIPT, &[event], &[], br#"{"session_id":"x"}"#);
assert!(out.status.success(), "{event:?}: {:?}", out.stderr);
assert!(out.stdout.is_empty(), "{event:?}: {:?}", out.stdout);
@@ -103,7 +112,7 @@ fn script_posts_mapped_state_to_the_activity_endpoint() {
/// Every event maps to the state the contract specifies.
#[test]
fn event_state_mapping_matches_the_contract() {
let map: std::collections::HashMap<&str, &str> = HOOK_EVENTS.into_iter().collect();
let map: std::collections::HashMap<&str, &str> = CLAUDE_EVENTS.into_iter().collect();
assert_eq!(map["PreToolUse"], "Active");
assert_eq!(map["PostToolUse"], "Active");
assert_eq!(map["Notification"], "AwaitingInput");
@@ -128,41 +137,6 @@ fn script_reports_the_error_spelling_as_detail() {
assert!(request.contains(r#""detail":"rate_limit""#), "{request}");
}
/// #161: the payload's own `session_id` — the agent's conversation id — is
/// parsed out and forwarded in the activity body, so the daemon can persist
/// it on the session.
#[test]
fn script_forwards_the_payloads_session_id_in_the_activity_body() {
let request = capture_request(|url| {
sh(
HOOK_SCRIPT,
&["Active"],
&owned_envs(&session_envs(&url)),
br#"{"session_id":"conv-abc123","hook_event_name":"PreToolUse"}"#,
)
});
assert!(request.contains(r#""state":"Active""#), "{request}");
assert!(
request.contains(r#""session_id":"conv-abc123""#),
"{request}"
);
}
/// A payload without a `session_id` (or one with an unexpected shape) must
/// never produce a broken body — no `session_id` field is posted.
#[test]
fn script_posts_no_session_id_when_the_payload_has_none() {
let request = capture_request(|url| {
sh(
HOOK_SCRIPT,
&["AwaitingInput"],
&owned_envs(&session_envs(&url)),
br#"{"hook_event_name":"Stop","stop_hook_active":false}"#,
)
});
assert!(!request.contains("session_id"), "{request}");
}
/// An `error` value that is not a plain enum spelling (or a payload without
/// one at all) must never produce a broken body — no detail is posted.
#[test]
@@ -238,10 +212,10 @@ fn fresh_install_registers_every_event() {
let dir = tempfile::tempdir().unwrap();
let home = dir.path();
write_script(&script_path(home)).unwrap();
install(&settings_path(home), &script_path(home)).unwrap();
install(Target::ClaudeCode, &settings_path(home), &script_path(home)).unwrap();
let cmds = installed_commands(&settings_path(home));
assert_eq!(cmds.len(), INSTALLED_EVENTS);
assert_eq!(cmds.len(), installed_events(Target::ClaudeCode));
let joined = cmds.join("\n");
assert_eq!(joined.matches("Active").count(), 2);
assert!(joined.contains("Error"));
@@ -256,15 +230,15 @@ fn fresh_install_registers_every_event() {
fn install_is_idempotent() {
let dir = tempfile::tempdir().unwrap();
let home = dir.path();
install(&settings_path(home), &script_path(home)).unwrap();
install(Target::ClaudeCode, &settings_path(home), &script_path(home)).unwrap();
let once = std::fs::read_to_string(settings_path(home)).unwrap();
install(&settings_path(home), &script_path(home)).unwrap();
install(Target::ClaudeCode, &settings_path(home), &script_path(home)).unwrap();
let twice = std::fs::read_to_string(settings_path(home)).unwrap();
assert_eq!(once, twice);
assert_eq!(
installed_commands(&settings_path(home)).len(),
INSTALLED_EVENTS
installed_events(Target::ClaudeCode)
);
}
@@ -292,12 +266,12 @@ fn install_updates_an_older_maestro_hook_in_place() {
.path()
.join(".local/share/maestro")
.join(SCRIPT_FILE_NAME);
install(&settings, &new_script).unwrap();
install(Target::ClaudeCode, &settings, &new_script).unwrap();
let cmds = installed_commands(&settings);
assert_eq!(
cmds.len(),
INSTALLED_EVENTS,
installed_events(Target::ClaudeCode),
"old entry replaced, not kept: {cmds:?}"
);
assert!(!cmds.iter().any(|c| c == old_script));
@@ -333,7 +307,7 @@ fn install_preserves_foreign_hooks_and_removes_stale_maestro_ones() {
.path()
.join(".local/share/maestro")
.join(SCRIPT_FILE_NAME);
install(&settings, &new_script).unwrap();
install(Target::ClaudeCode, &settings, &new_script).unwrap();
let root: serde_json::Value =
serde_json::from_str(&std::fs::read_to_string(&settings).unwrap()).unwrap();
@@ -346,7 +320,7 @@ fn install_preserves_foreign_hooks_and_removes_stale_maestro_ones() {
let cmds = installed_commands(&settings);
assert_eq!(
cmds.iter().filter(|c| c.contains(HOOK_MARKER)).count(),
INSTALLED_EVENTS,
installed_events(Target::ClaudeCode),
"{cmds:?}"
);
}
@@ -356,9 +330,15 @@ fn install_preserves_foreign_hooks_and_removes_stale_maestro_ones() {
fn is_installed_tracks_reality() {
let dir = tempfile::tempdir().unwrap();
let home = dir.path();
assert!(!is_installed(&settings_path(home)));
install_for_home(home).unwrap();
assert!(is_installed(&settings_path(home)));
for target in Target::ALL {
assert!(!is_installed(&target.settings_path(home)));
}
for (_, result) in install_for_home(home) {
result.unwrap();
}
for target in Target::ALL {
assert!(is_installed(&target.settings_path(home)));
}
}
/// The full home-level install leaves the script executable.
@@ -366,7 +346,9 @@ fn is_installed_tracks_reality() {
fn install_for_home_writes_an_executable_script() {
let dir = tempfile::tempdir().unwrap();
let home = dir.path();
install_for_home(home).unwrap();
for (_, result) in install_for_home(home) {
result.unwrap();
}
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
@@ -385,7 +367,7 @@ fn install_for_home_writes_an_executable_script() {
fn install_registers_the_prompt_event() {
let dir = tempfile::tempdir().unwrap();
let home = dir.path();
install(&settings_path(home), &script_path(home)).unwrap();
install(Target::ClaudeCode, &settings_path(home), &script_path(home)).unwrap();
let root: serde_json::Value =
serde_json::from_str(&std::fs::read_to_string(settings_path(home)).unwrap()).unwrap();
@@ -402,7 +384,7 @@ fn install_registers_the_prompt_event() {
fn install_registers_no_prompt_event_without_the_feature() {
let dir = tempfile::tempdir().unwrap();
let home = dir.path();
install(&settings_path(home), &script_path(home)).unwrap();
install(Target::ClaudeCode, &settings_path(home), &script_path(home)).unwrap();
let root: serde_json::Value =
serde_json::from_str(&std::fs::read_to_string(settings_path(home)).unwrap()).unwrap();
@@ -471,8 +453,12 @@ fn the_script_posts_a_prompt_payload_unchanged() {
fn is_current_rejects_an_install_missing_an_event() {
let dir = tempfile::tempdir().unwrap();
let home = dir.path();
install_for_home(home).unwrap();
assert!(is_current(&settings_path(home), &script_path(home)));
install_target(Target::ClaudeCode, home).unwrap();
assert!(is_current(
Target::ClaudeCode,
&settings_path(home),
&script_path(home)
));
// Drop one event, as an older install would have left the file.
let mut root: serde_json::Value =
@@ -485,7 +471,11 @@ fn is_current_rejects_an_install_missing_an_event() {
.unwrap();
assert!(is_installed(&settings_path(home)));
assert!(!is_current(&settings_path(home), &script_path(home)));
assert!(!is_current(
Target::ClaudeCode,
&settings_path(home),
&script_path(home)
));
}
/// A settings file that is complete but points at a script from an older
@@ -494,9 +484,13 @@ fn is_current_rejects_an_install_missing_an_event() {
fn is_current_rejects_a_stale_script() {
let dir = tempfile::tempdir().unwrap();
let home = dir.path();
install_for_home(home).unwrap();
install_target(Target::ClaudeCode, home).unwrap();
std::fs::write(script_path(home), "#!/bin/sh\nexit 0\n").unwrap();
assert!(!is_current(&settings_path(home), &script_path(home)));
assert!(!is_current(
Target::ClaudeCode,
&settings_path(home),
&script_path(home)
));
}
/// Nothing installed at all is not current.
@@ -504,18 +498,93 @@ fn is_current_rejects_a_stale_script() {
fn is_current_rejects_a_missing_install() {
let dir = tempfile::tempdir().unwrap();
let home = dir.path();
assert!(!is_current(&settings_path(home), &script_path(home)));
assert!(!is_current(
Target::ClaudeCode,
&settings_path(home),
&script_path(home)
));
}
/// `install_for_home` reports a stale script as a change, even when the
/// `install_target` reports a stale script as a change, even when the
/// settings entries pointing at it already read correctly.
#[test]
fn install_for_home_reports_a_stale_script_as_a_change() {
fn install_target_reports_a_stale_script_as_a_change() {
let dir = tempfile::tempdir().unwrap();
let home = dir.path();
assert!(install_for_home(home).unwrap());
assert!(!install_for_home(home).unwrap());
assert!(install_target(Target::ClaudeCode, home).unwrap());
assert!(!install_target(Target::ClaudeCode, home).unwrap());
std::fs::write(script_path(home), "#!/bin/sh\nexit 0\n").unwrap();
assert!(install_for_home(home).unwrap());
assert!(install_target(Target::ClaudeCode, home).unwrap());
}
/// Codex is hooked in its own file, with its own event set: the two agents
/// share the script but not the events, so neither file carries the other's
/// spellings (#131 follow-up — a codex session was never named because
/// nothing installed its `UserPromptSubmit` hook).
#[test]
fn each_target_gets_its_own_file_and_events() {
let dir = tempfile::tempdir().unwrap();
let home = dir.path();
for (_, result) in install_for_home(home) {
result.unwrap();
}
let claude = events_in(&Target::ClaudeCode.settings_path(home));
let codex = events_in(&Target::Codex.settings_path(home));
assert_eq!(
Target::Codex.settings_path(home),
home.join(".codex/hooks.json")
);
// Claude Code's failure signal has no codex equivalent, and codex's
// approval notice is spelled differently.
assert!(claude.contains(&"StopFailure".to_string()));
assert!(claude.contains(&"Notification".to_string()));
assert!(!codex.contains(&"StopFailure".to_string()));
assert!(!codex.contains(&"Notification".to_string()));
assert!(codex.contains(&"PermissionRequest".to_string()));
// Both report work and both report a finished turn.
for event in ["PreToolUse", "PostToolUse", "Stop"] {
assert!(claude.contains(&event.to_string()), "claude {event}");
assert!(codex.contains(&event.to_string()), "codex {event}");
}
assert_eq!(
installed_commands(&Target::Codex.settings_path(home)).len(),
installed_events(Target::Codex)
);
}
/// #131: a codex session names itself the same way a Claude Code one does —
/// same event spelling, same script, same `Prompt` argument.
#[cfg(feature = "haiku-autoname")]
#[test]
fn codex_reports_its_first_prompt_too() {
let dir = tempfile::tempdir().unwrap();
let home = dir.path();
install_target(Target::Codex, home).unwrap();
let root: serde_json::Value =
serde_json::from_str(&std::fs::read_to_string(Target::Codex.settings_path(home)).unwrap())
.unwrap();
let cmd = root["hooks"]["UserPromptSubmit"][0]["hooks"][0]["command"]
.as_str()
.unwrap()
.to_string();
assert!(cmd.ends_with(" Prompt"), "{cmd}");
assert!(cmd.contains(SCRIPT_FILE_NAME));
}
/// Every event name a settings file registers a maestro hook under.
fn events_in(settings: &Path) -> Vec<String> {
serde_json::from_str::<serde_json::Value>(&std::fs::read_to_string(settings).unwrap())
.unwrap()
.get("hooks")
.and_then(|h| h.as_object())
.into_iter()
.flatten()
.map(|(event, _)| event.clone())
.collect()
}
+2 -1
View File
@@ -268,6 +268,7 @@ mod tests {
fn session(cwd: &str) -> Session {
Session {
view: Default::default(),
id: Uuid::from_u128(1),
cwd: PathBuf::from(cwd),
command: "bash".into(),
@@ -284,7 +285,7 @@ mod tests {
current_cwd: None,
blocked_work: None,
last_focused_at: None,
agent_session_id: None,
acp_session_id: None,
}
}
+184 -136
View File
@@ -116,13 +116,33 @@ pub struct Session {
/// sorts a never-focused session after every session that has one.
#[serde(default)]
pub last_focused_at: Option<DateTime<Utc>>,
/// The conversation id the agent running in this session reported for
/// itself (#161), parsed from a hook's `session_id` field. Written once,
/// from the first activity report that carries one, and left alone after
/// — an agent's own session id does not change mid-session. `None` for
/// sessions that never run an agent (plain shells) or predate the field.
/// 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 agent_session_id: Option<String>,
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.
@@ -130,38 +150,10 @@ fn epoch() -> DateTime<Utc> {
DateTime::from_timestamp(0, 0).expect("unix epoch is a valid timestamp")
}
/// Lifecycle of a session as the daemon knows it.
///
/// Deserializes leniently, like [`ActivityState`] (#162): an unknown spelling
/// from a newer daemon reads as [`SessionStatus::Dead`] instead of failing the
/// whole `Session`. `Dead` is the safe degrade — an unrecognised session is
/// one this client must not treat as having a live pane to type into.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum SessionStatus {
Running,
Dead,
/// Known from its sidecar, with no tmux session behind it (#162). The
/// daemon registers these at recovery time for every sidecar tmux had
/// nothing to adopt — after a host reboot, that is every session there
/// was. There is no pane and no PTY bridge: a dormant session is a record
/// of what was running, listed so the operator still has it, and the only
/// thing that can be done to it is dismissing it.
Dormant,
}
impl<'de> Deserialize<'de> for SessionStatus {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
let s = String::deserialize(deserializer)?;
Ok(match s.as_str() {
"Running" => Self::Running,
"Dormant" => Self::Dormant,
// Known or not, any other spelling degrades to the inert variant.
_ => Self::Dead,
})
}
}
/// Inferred session activity, orthogonal to [`SessionStatus`] (NAP-340). The
@@ -219,11 +211,6 @@ pub struct ReportActivityRequest {
/// The daemon caps its length server-side.
#[serde(default)]
pub detail: Option<String>,
/// The agent's own conversation id (#161), forwarded from the hook
/// payload's `session_id` field. `None` for hook events that don't carry
/// one, or hooks predating this field.
#[serde(default)]
pub session_id: Option<String>,
}
/// Reject an unknown [`ActivityState`] spelling instead of degrading it to
@@ -392,6 +379,11 @@ 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.
@@ -529,34 +521,8 @@ impl ApiError {
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct SpawnPreset {
pub id: String,
#[serde(default)]
pub label: String,
pub command: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub resume_command: Option<String>,
}
/// Placeholder a preset's `resume_command` may carry for the agent's own
/// conversation id (#165), substituted from [`Session::agent_session_id`].
pub const AGENT_SESSION_PLACEHOLDER: &str = "{agent_session}";
impl SpawnPreset {
/// The command that resumes a session spawned from this preset, with
/// [`AGENT_SESSION_PLACEHOLDER`] filled in from `agent_session`.
///
/// `None` when the preset has no `resume_command` — a plain shell has
/// nothing to resume — and also when the template asks for an agent
/// session id the session never reported: `claude --resume` with an empty
/// id resumes nothing, so the caller is better off respawning the original
/// command than running a broken one.
pub fn resume_command_for(&self, agent_session: Option<&str>) -> Option<String> {
let template = self.resume_command.as_deref()?;
if !template.contains(AGENT_SESSION_PLACEHOLDER) {
return Some(template.to_string());
}
let id = agent_session.filter(|s| !s.is_empty())?;
Some(template.replace(AGENT_SESSION_PLACEHOLDER, id))
}
}
/// Build provenance of the running daemon: the commit it was compiled
@@ -789,6 +755,87 @@ pub enum CatalogEvent {
},
}
// ---------------------------------------------------------------------------
// Structured-session projections
// ---------------------------------------------------------------------------
/// One client-neutral item in a structured session transcript.
///
/// `data` preserves the ACP payload without committing clients to an ACP
/// dependency. Rendering choices, including markdown and tool-card treatment,
/// stay with the client.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct TranscriptRow {
/// Stable within one session, so later deltas can address this row.
pub id: String,
pub kind: TranscriptRowKind,
pub data: Value,
}
/// The semantic shape of a [`TranscriptRow`].
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum TranscriptRowKind {
User,
AgentText,
AgentThought,
ToolCall,
ToolDiff,
Approval,
Plan,
Notice,
}
/// A complete transcript at a durable event-log position.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct TranscriptSnapshot {
pub cursor: u64,
#[serde(default)]
pub rows: Vec<TranscriptRow>,
}
/// A change to a transcript row.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum TranscriptDelta {
Append { row: TranscriptRow },
Patch { row: TranscriptRow },
Remove { id: String },
}
/// Per-turn indicators derived from the ACP stream.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct TurnState {
#[serde(default)]
pub responding: bool,
#[serde(default)]
pub thinking: bool,
#[serde(default)]
pub active_tool_calls: Vec<String>,
}
/// A whole-session view derived from ACP notifications.
///
/// Values whose shape ACP owns are carried as JSON so this wire protocol stays
/// forward-compatible when ACP adds fields or variants.
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub struct ReducedState {
#[serde(default)]
pub turn: TurnState,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub pending_approval: Option<Value>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub plan: Option<Value>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub usage: Option<Value>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub context: Option<Value>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub model: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub mode: Option<String>,
}
// ---------------------------------------------------------------------------
// Client ↔ Daemon control messages (text frames on the attach WS)
// ---------------------------------------------------------------------------
@@ -808,47 +855,6 @@ pub enum ClientControl {
mod tests {
use super::*;
fn preset(resume: Option<&str>) -> SpawnPreset {
SpawnPreset {
id: "claude".into(),
label: "Claude".into(),
command: "claude".into(),
resume_command: resume.map(str::to_string),
}
}
#[test]
fn resume_command_substitutes_the_agent_session() {
let p = preset(Some("claude --resume {agent_session}"));
assert_eq!(
p.resume_command_for(Some("abc-123")).as_deref(),
Some("claude --resume abc-123")
);
}
#[test]
fn resume_command_without_a_placeholder_ignores_the_agent_session() {
let p = preset(Some("opencode --continue"));
assert_eq!(
p.resume_command_for(None).as_deref(),
Some("opencode --continue")
);
}
/// A template that wants an id the session never reported would run
/// `claude --resume` against nothing; the caller respawns instead.
#[test]
fn resume_command_is_none_without_the_id_it_asks_for() {
let p = preset(Some("claude --resume {agent_session}"));
assert!(p.resume_command_for(None).is_none());
assert!(p.resume_command_for(Some("")).is_none());
}
#[test]
fn a_preset_without_a_resume_command_has_none() {
assert!(preset(None).resume_command_for(Some("abc")).is_none());
}
#[test]
fn version_compatibility_is_major_only() {
assert!(is_compatible(PROTOCOL_VERSION));
@@ -969,7 +975,8 @@ mod tests {
current_cwd: Some(PathBuf::from("/home/naps62/projects/maestro/crates")),
blocked_work: None,
last_focused_at: None,
agent_session_id: None,
view: SessionView::Terminal,
acp_session_id: None,
}
}
@@ -1139,7 +1146,6 @@ mod tests {
let full = ReportActivityRequest {
state: ActivityState::AwaitingInput,
detail: Some("waiting on permission".into()),
session_id: Some("conv-abc123".into()),
};
assert_eq!(full, round_trip(&full));
@@ -1168,33 +1174,6 @@ mod tests {
serde_json::to_value(SessionStatus::Dead).unwrap(),
serde_json::json!("Dead")
);
assert_eq!(
serde_json::to_value(SessionStatus::Dormant).unwrap(),
serde_json::json!("Dormant")
);
}
// #162: a session with no tmux behind it still rides the wire whole.
#[test]
fn dormant_session_round_trips() {
let mut s = sample_session();
s.status = SessionStatus::Dormant;
assert_eq!(s, round_trip(&s));
}
// #162: a `SessionStatus` this client does not know must degrade to the
// inert variant, not fail the whole `Session` the way a derived
// `Deserialize` would.
#[test]
fn unknown_session_status_degrades_to_dead() {
assert_eq!(
serde_json::from_value::<SessionStatus>(serde_json::json!("Hibernating")).unwrap(),
SessionStatus::Dead
);
let mut value = serde_json::to_value(sample_session()).unwrap();
value["status"] = serde_json::json!("Hibernating");
let s: Session = serde_json::from_value(value).expect("session survives unknown status");
assert_eq!(s.status, SessionStatus::Dead);
}
#[test]
@@ -1209,6 +1188,7 @@ mod tests {
branch: "feat/foo".into(),
create_branch: true,
}),
view: SessionView::Terminal,
};
assert_eq!(req, round_trip(&req));
}
@@ -1223,6 +1203,74 @@ 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 structured_projection_types_round_trip_and_default() {
let row = TranscriptRow {
id: "tool:read-1".into(),
kind: TranscriptRowKind::ToolCall,
data: serde_json::json!({"name": "vendor.read", "title": "Read"}),
};
let snapshot = TranscriptSnapshot {
cursor: 7,
rows: vec![row.clone()],
};
assert_eq!(snapshot, round_trip(&snapshot));
assert_eq!(
TranscriptDelta::Patch { row: row.clone() },
round_trip(&TranscriptDelta::Patch { row })
);
let state: ReducedState = serde_json::from_value(serde_json::json!({})).unwrap();
assert_eq!(state, ReducedState::default());
assert_eq!(
serde_json::to_value(TranscriptRowKind::AgentThought).unwrap(),
serde_json::json!("agent_thought")
);
}
#[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
-15
View File
@@ -66,12 +66,6 @@ pub fn session_focus(id: Uuid) -> String {
format!("{SESSIONS}/{id}/focus")
}
/// `POST /api/sessions/:id/resume` (#166): respawn a dormant session under its
/// own id, keeping its sidecar, name, worktree and `created_at`.
pub fn session_resume(id: Uuid) -> String {
format!("{SESSIONS}/{id}/resume")
}
/// `WS /api/attach/:id` — raw PTY bytes.
pub fn attach(id: Uuid) -> String {
format!("{API_PREFIX}/attach/{id}")
@@ -85,7 +79,6 @@ pub mod patterns {
pub const SESSION_INPUT: &str = "/api/sessions/:id/input";
pub const SESSION_PROMPT: &str = "/api/sessions/:id/prompt";
pub const SESSION_FOCUS: &str = "/api/sessions/:id/focus";
pub const SESSION_RESUME: &str = "/api/sessions/:id/resume";
pub const ATTACH: &str = "/api/attach/:id";
}
@@ -209,10 +202,6 @@ mod tests {
session_focus(id()),
"/api/sessions/6f1a7c0e-0f5e-4a2b-9a3d-6b1c2d3e4f50/focus"
);
assert_eq!(
session_resume(id()),
"/api/sessions/6f1a7c0e-0f5e-4a2b-9a3d-6b1c2d3e4f50/resume"
);
assert_eq!(
attach(id()),
"/api/attach/6f1a7c0e-0f5e-4a2b-9a3d-6b1c2d3e4f50"
@@ -247,10 +236,6 @@ mod tests {
patterns::SESSION_FOCUS.trim_end_matches(":id/focus"),
session(id()).trim_end_matches(&id().to_string())
);
assert_eq!(
patterns::SESSION_RESUME.trim_end_matches(":id/resume"),
session(id()).trim_end_matches(&id().to_string())
);
assert_eq!(
patterns::ATTACH.trim_end_matches(":id"),
attach(id()).trim_end_matches(&id().to_string())
+2 -1
View File
@@ -31,6 +31,7 @@ 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(),
@@ -47,7 +48,7 @@ pub fn session(id: Uuid, cwd: &str, command: &str) -> Session {
current_cwd: None,
blocked_work: None,
last_focused_at: None,
agent_session_id: None,
acp_session_id: None,
}
}
+13 -6
View File
@@ -1882,6 +1882,7 @@ mod tests {
sessions: ids
.iter()
.map(|id| Session {
view: Default::default(),
id: *id,
command: "bash".into(),
cwd: std::path::PathBuf::from("/"),
@@ -1898,7 +1899,7 @@ mod tests {
current_cwd: None,
blocked_work: None,
last_focused_at: None,
agent_session_id: None,
acp_session_id: None,
})
.collect(),
});
@@ -2064,6 +2065,7 @@ 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("/"),
@@ -2080,7 +2082,7 @@ mod tests {
current_cwd: None,
blocked_work: None,
last_focused_at: None,
agent_session_id: None,
acp_session_id: None,
}],
});
app.dispatch(Action::SidebarNext);
@@ -2092,6 +2094,7 @@ mod tests {
use chrono::Utc;
use maestro_protocol::{Session, SessionStatus};
Session {
view: Default::default(),
id: sid,
command: "bash".into(),
cwd: cwd.into(),
@@ -2108,7 +2111,7 @@ mod tests {
current_cwd: None,
blocked_work: None,
last_focused_at: None,
agent_session_id: None,
acp_session_id: None,
}
}
@@ -2780,6 +2783,7 @@ 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(),
@@ -3198,6 +3202,7 @@ 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("/"),
@@ -3217,7 +3222,7 @@ mod tests {
current_cwd: None,
blocked_work: None,
last_focused_at: focused,
agent_session_id: None,
acp_session_id: None,
};
let older = Utc::now() - chrono::Duration::seconds(60);
app.catalog.apply_event(CatalogEvent::Snapshot {
@@ -3479,6 +3484,7 @@ 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("/"),
@@ -3496,7 +3502,7 @@ mod tests {
current_cwd: None,
blocked_work: None,
last_focused_at: None,
agent_session_id: None,
acp_session_id: None,
}],
});
app.dispatch(Action::KillHighlighted);
@@ -3516,6 +3522,7 @@ mod tests {
};
app.catalog.apply_event(CatalogEvent::Snapshot {
sessions: vec![Session {
view: Default::default(),
id: sid,
command: "bash".into(),
cwd: std::path::PathBuf::from("/"),
@@ -3532,7 +3539,7 @@ mod tests {
current_cwd: None,
blocked_work: None,
last_focused_at: None,
agent_session_id: None,
acp_session_id: None,
}],
});
}
+2 -1
View File
@@ -207,6 +207,7 @@ 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"),
@@ -223,7 +224,7 @@ mod tests {
current_cwd: None,
blocked_work: None,
last_focused_at: None,
agent_session_id: None,
acp_session_id: None,
}
}
+4 -22
View File
@@ -194,9 +194,7 @@ pub enum BarOutcome {
pub fn order_sessions_for_bar(sessions: &[Session]) -> Vec<&Session> {
let mut ordered: Vec<&Session> = sessions.iter().collect();
ordered.sort_by(|a, b| {
// docs/COMMAND-BAR.md: anything with no live pane — dead or dormant
// (#162) — sorts after the sessions you could actually attach to.
let dead = |s: &Session| u8::from(s.status != SessionStatus::Running);
let dead = |s: &Session| u8::from(s.status == SessionStatus::Dead);
dead(a)
.cmp(&dead(b))
.then_with(|| {
@@ -321,7 +319,7 @@ impl CommandBar {
.map(|s| BarEntry {
candidate: MatchCandidate::Session(session_candidate(s, home_dir)),
target: BarTarget::Session(s.id),
dead: s.status != SessionStatus::Running,
dead: s.status == SessionStatus::Dead,
chords: Vec::new(),
})
.collect();
@@ -795,6 +793,7 @@ 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(),
@@ -811,7 +810,7 @@ mod tests {
current_cwd: None,
blocked_work: None,
last_focused_at: None,
agent_session_id: None,
acp_session_id: None,
}
}
@@ -892,23 +891,6 @@ mod tests {
assert!(bar.rows()[1].dead);
}
/// #162: docs/COMMAND-BAR.md ranks by whether you could attach, not by
/// which way the pane went away — a dormant session sorts with the dead.
#[test]
fn dormant_sessions_sort_after_live_ones_however_recent() {
let mut dormant = named(1, "dormant");
dormant.status = SessionStatus::Dormant;
dormant.last_focused_at = Some(at(99));
let live = named(2, "live");
let bar = CommandBar::new(&[dormant, live], &[], "/home/me");
assert_eq!(
ids(bar.rows()),
vec![Uuid::from_u128(2), Uuid::from_u128(1)]
);
assert!(bar.rows()[1].dead);
}
#[test]
fn typing_ranks_through_the_matcher_and_returns_ranges() {
let mut bar = CommandBar::new(
+1 -4
View File
@@ -694,6 +694,7 @@ impl DraftPane {
}
let worktree = self.build_worktree()?;
Ok(CreateSessionRequest {
view: Default::default(),
cwd,
command: command.to_string(),
env: Default::default(),
@@ -1705,7 +1706,6 @@ mod tests {
id: "aider".into(),
label: "Aider".into(),
command: "aider".into(),
resume_command: None,
}]);
assert_eq!(
draft.command(),
@@ -1728,13 +1728,11 @@ mod tests {
id: "claude".into(),
label: "Claude".into(),
command: "claude".into(),
resume_command: None,
},
SpawnPreset {
id: "aider".into(),
label: "Aider".into(),
command: "aider".into(),
resume_command: None,
},
]);
assert_eq!(draft.command(), "aider");
@@ -1754,7 +1752,6 @@ mod tests {
id: "claude".into(),
label: "Claude".into(),
command: "claude".into(),
resume_command: None,
}]);
assert!(draft.is_custom());
assert_eq!(draft.command(), "bash -l");
+1 -1
View File
@@ -108,6 +108,7 @@ impl FastForwardDefaults {
return None;
}
Some(CreateSessionRequest {
view: Default::default(),
cwd: project.path.clone(),
command: command.to_string(),
env: Default::default(),
@@ -183,7 +184,6 @@ mod tests {
id: id.into(),
label: id.into(),
command: command.into(),
resume_command: None,
}
}
+4 -41
View File
@@ -268,8 +268,6 @@ impl Sidebar {
fn name_row(&self, s: &Session, width: usize, focused: bool) -> Vec<Span<'static>> {
let glyph = match (s.status, self.attached.contains(&s.id)) {
(SessionStatus::Dead, _) => "✗",
// #162: known from its sidecar, nothing running behind it.
(SessionStatus::Dormant, _) => "◌",
(SessionStatus::Running, true) => "●",
(SessionStatus::Running, false) => "○",
};
@@ -324,7 +322,6 @@ impl Sidebar {
// on line 1; the kill confirm names the specifics.
let blocked = s.blocked_work.filter(|b| !b.is_clear());
let dead = s.status == SessionStatus::Dead;
let dormant = s.status == SessionStatus::Dormant;
let highlighted = self.highlight == Some(SidebarItem::Session(s.id));
let fill = self.card_fill(s, focused);
let base = card_base(fill, focused);
@@ -364,9 +361,6 @@ impl Sidebar {
if dead {
tail.push(Span::styled(" exited", muted));
}
if dormant {
tail.push(Span::styled(" dormant", muted));
}
// docs/SIDEBAR.md: the card names the program, never its arguments.
let command = Span::styled(format!(" {}", command_program(&s.command)), muted);
@@ -490,6 +484,7 @@ 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"),
@@ -506,7 +501,7 @@ mod tests {
current_cwd: None,
blocked_work: None,
last_focused_at: None,
agent_session_id: None,
acp_session_id: None,
}
}
@@ -1075,39 +1070,6 @@ mod tests {
);
}
/// #162, docs/SIDEBAR.md: a dormant session is known but not running —
/// its own glyph on line 1 and the muted `dormant` fragment on line 2, so
/// it never reads as something that exited.
#[test]
fn dormant_session_reads_as_known_but_not_running() {
let mut s = session_named(1, "claude");
s.status = SessionStatus::Dormant;
s.activity_since = Utc::now() - Duration::seconds(5);
let sb = Sidebar {
sessions: vec![s],
highlight: None,
attached: Default::default(),
pane_session: None,
};
let all = rows(&draw(&sb, 28, 8));
let idx = all.iter().position(|l| l.contains("claude")).unwrap();
assert!(
all[idx].contains('\u{25cc}'),
"expected the dormant glyph in: {:?}",
all[idx]
);
assert!(
all[idx + 1].contains("dormant"),
"expected dormant fragment in: {:?}",
all[idx + 1]
);
assert!(
!all[idx + 1].contains("exited"),
"a dormant session never reads as exited: {:?}",
all[idx + 1]
);
}
/// docs/SIDEBAR.md: line 2 names the program, never its arguments.
#[test]
fn meta_line_drops_the_commands_arguments() {
@@ -1290,6 +1252,7 @@ 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"),
@@ -1306,7 +1269,7 @@ mod tests {
current_cwd: None,
blocked_work: None,
last_focused_at: None,
agent_session_id: None,
acp_session_id: None,
};
let sb = Sidebar {
sessions: vec![s],
+3 -3
View File
@@ -123,9 +123,9 @@ client opening `/api/attach/:id`, and a client reporting focus explicitly at
`POST /api/sessions/:id/focus` when it refocuses a pane that is already open.
**A session that has never been focused sorts after every session that has**,
and among those, most recently created first. Sessions with no live pane —
`Dead`, and `Dormant` (#162) — sort after live ones regardless of recency: a
session you cannot attach to is not where you just were.
and among those, most recently created first. Dead sessions sort after live
ones regardless of recency: a session you cannot attach to is not where you
just were.
Under `>` an empty query lists every command in registry order. Under `~` it
lists recent dirs, most recent first, then the project list.
+25 -20
View File
@@ -188,6 +188,31 @@ 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. A daemon restart ends an in-flight structured turn, but preserves its
worktree, session record, and event-log transcript. The replacement daemon
starts a new adapter, resumes it with `session/load`, and replays the
transcript. Terminal sessions keep surviving restarts through tmux.
## The browser UI
The daemon serves the SPA off its own listener. Open
@@ -243,26 +268,6 @@ Pick `/tmp` if the daemon must outlive your login session and you can't run
tmux sockets with it). The env var `MAESTRO_DAEMON_TMUX_TMPDIR` takes
precedence over the file.
## What happens to sessions after a reboot
A host reboot kills the tmux server, so every session the daemon knew is dead
while its record on disk survives. What recovery does with those is
`[daemon].resume_on_boot`:
```toml
[daemon]
resume_on_boot = "dormant" # "dormant" (default) | "auto" | "never"
```
- `dormant` — the sessions are listed but not running, and you resume the ones
you want from the web UI or the TUI.
- `auto` — the daemon resumes them all itself during startup. Only pick this
on a host where restarting every agent unattended is safe: they come back on
the flags they were spawned with, with nobody watching.
- `never` — the records are deleted and a reboot is a clean slate.
See `docs/RESTART.md` for the full contract.
## What the systemd unit looks like
The wizard writes a *user* unit (`systemctl --user`), since `maestro daemon`
-78
View File
@@ -68,84 +68,6 @@ resolves, so a blocked client is never a trapped one. It clears itself the
moment a probe succeeds — there is no "retry" button, because the client is
already retrying.
## What survives, and what comes back dormant
A daemon restart leaves tmux alone, so recovery re-adopts every
`maestro-<uuid>` session tmux still holds and the operator sees no change
beyond the reconnect above.
A **host reboot** is different: it kills the tmux server, so there is nothing
to adopt while every session's sidecar is still on disk. Recovery always asks
tmux — `tmux list-sessions` is the single source of truth for liveness, and the
kernel's `boot_id` only tells a reboot from a restart in the log (#164, #174).
A boot id changes without the tmux server dying whenever the state directory is
restored, a VM or container snapshot is rolled back, or the `boot_id` file is
truncated; a session tmux still holds is adopted `Running` in every one of
those cases. Recovery makes a
second pass over those sidecars and registers each as a `Dormant` session
(#162) — id, name, cwd, command, worktree and `created_at` intact. A dormant
session has no pane and no PTY bridge: it cannot be attached to, typed into or
resized, and neither watcher pass touches it.
A dormant session is brought back with `POST /api/sessions/:id/resume` (#166),
which respawns tmux under the **same** uuid and sidecar — `metadata.name`,
`created_at`, `env` and worktree ownership all survive, and only the tmux
session and the PTY bridge are new. The command is the matching spawn preset's
`resume_command` with `{agent_session}` filled in from the session's own
`agent_session_id`, so an agent comes back inside its old conversation; a
session with no matching preset, no `resume_command`, or no reported
conversation id is respawned on the command it was created with. `Session::command`
always keeps that original command, so a second resume resolves the same preset.
Resume is refused with `409` for a session that is already running
(`already_running`) and for one whose `cwd` has disappeared (`cwd_missing`) —
neither is something a respawn could fix. The other thing to do with a dormant
session is dismiss it, which removes the sidecar so the next recovery does not
bring it back.
## What recovery does with them: `resume_on_boot`
Whether a recovered-but-dead session is left dormant at all is the operator's
call, per host, via `[daemon].resume_on_boot` (#167):
| value | what recovery does |
| ----------- | --------------------------------------------------------------------- |
| `"dormant"` | **Default.** Register each one dormant and wait for the operator. |
| `"auto"` | Resume each one during recovery, itself. |
| `"never"` | Prune the sidecars instead — a reboot is a clean slate. |
`dormant` is the default because resuming agents unattended is a foot-gun:
they come back on whatever flags they were spawned with, seconds after a
reboot, with nobody watching. `auto` exists for hosts where that is known to
be safe, and it is opted into rather than inherited.
`auto` calls the same `resume` the client's `POST /api/sessions/:id/resume`
does — same uuid, same sidecar, the preset's `resume_command` — so there is
one resume path and not two. A session that fails to resume is logged and
stays dormant; the rest still come back.
`never` prunes the same way #163 does for a missing `cwd`: the sidecar is
deleted and nothing is registered, so the session is not listed and the next
recovery does not find it. Whatever worktree it owned is left on disk.
The policy only decides the fate of sessions with no live pane. A session
tmux still holds is adopted `Running` by the first pass under every policy,
including `never`.
Recovery logs the policy that ran and the number of sessions it acted on, so
a host that resumes agents unattended says so in its own journal.
Attaching to a session that has no bridge answers `409 not_attachable`, not
`404 unknown_session`: the session exists, and a client that cannot tell "gone"
from "not running" cannot offer the resume.
`SessionStatus` is a wire type, so from this build forward it deserializes
leniently on both clients: a spelling the client does not know reads as `Dead`
rather than failing the whole session. That leniency lives in the client's own
binary, so it covers spellings added *after* this build, not this one — a CLI
or TUI still running an older build fails to parse a `Dormant` session, and
with it the whole session list, until it is upgraded. Clients render a dormant session as known-but-not-running —
see `docs/SIDEBAR.md` for the card and `docs/COMMAND-BAR.md` for how it ranks.
## Timing
- **Grace window** — 1200 ms between the socket dropping and the overlay
+12 -13
View File
@@ -56,7 +56,7 @@ itself is not.
Line 2 reads left to right:
```
[<duration>] [<detail>] <command> [<diff stat> | exited | dormant]
[<duration>] [<detail>] <command> [<diff stat> | exited]
```
Its minimum content is therefore just the command.
@@ -104,14 +104,8 @@ fragment; it no longer decides whether line 2 exists.
while an agent-reported state is in force (see Precedence): when the report
lapses at its TTL or is replaced, the detail goes with it. The detail
truncates with `…` ahead of the trailing fragments — a long detail never
pushes `<command>`, `<blocked work>`, `exited` or `dormant` off the line, and
neither client ever lets it wrap.
`dormant` (#162) is the counterpart of `exited` for a session the daemon knows
only from its sidecar: recovery found no tmux session behind it, so it has no
pane, no scrollback, and nothing to attach to. It is not an exit — nothing was
observed ending — so the two fragments never both show. The TUI marks the same
state on line 1 with `◌`, next to `✗` for exited and `●`/`○` for running.
pushes `<command>`, `<blocked work>`, or `exited` off the line, and neither
client ever lets it wrap.
`<diff stat>` is the blocked-work breakdown (#83) in diff notation, replacing
the spelled-out `⚠ 3 uncommitted files, 2 unpushed commits`. It renders only
@@ -215,7 +209,7 @@ the cursor's fill while the pane holds the keyboard, and the web's ring on a
highlighted **folder** row — a folder previews nothing, so it has no fill to
step up to.
The duration text inherits its state's colour; the exited, dormant and command
The duration text inherits its state's colour; the exited and command
fragments use the surface's `muted` role, as do the diff stat's `~`, `?` and
`↑` fragments. Only `+` and `-` step outside the surface's muted role, and they
take the `active` and `error` roles of the same surface.
@@ -256,6 +250,10 @@ Two rulings specific to a reported `Error` (#114):
process changes do not clear it — only the agent itself may withdraw what
the agent itself reported. Re-reporting `Error` extends the TTL rather than
stacking reports.
- **Only Claude Code reports `Error`.** Codex has no counterpart to
`StopFailure` in its hook event set, so a codex turn that ends on an API
error reports `AwaitingInput` and its card carries no error mark. A card
without the mark means "nothing reported a failure", not "nothing failed".
If a client receives an `ActivityState` variant it does not know, it must not
fail rendering: treat it as `Idle` (neutral/dim, line-2 rules apply as usual)
@@ -309,9 +307,10 @@ lands on the first row it was hiding. A highlight parked on a skipped row —
reachable by clicking a folder header — still steps to its neighbour rather
than jumping to the end of the list.
The web client needs a chord to reach the list at all (`Ctrl+Shift+H`, and the
same chord back): an attached terminal owns every plain keystroke. The TUI uses
`ctrl-w h`.
The web client needs a chord to reach the list at all (leader-`h` — `Ctrl+A h`
by default — and the same chord back): an attached terminal owns every plain
keystroke, and Escape is left to reach the pane's shell/app instead. The TUI
uses `ctrl-w h`.
Everything else is the same key on both clients:
+12 -4
View File
@@ -35,10 +35,18 @@ The branch name is used verbatim as a relative path, so `feature/login` becomes
back to a short hash of the branch.
If that path exists, the daemon appends `-2`, then `-3`, and so on until it finds
a free one. The suffix names the **directory only**. The branch is whatever was
requested; a second worktree of the same branch is not what happens here, because
git refuses to check the same branch out twice and the branch picker already
marks branches that are checked out elsewhere.
a free one.
For a *new* branch (`create_branch: true`), the suffix goes on the branch name
itself, and the directory follows since it is derived from the branch — a
random handle can otherwise reuse the name of a branch some earlier, already
removed worktree left behind, and `git worktree add -b` refuses outright even
though the directory is free.
For an *existing* branch (`create_branch: false`) the name must stay exact, so
only the directory is suffixed — a second worktree of the same branch is not
what happens here, because git refuses to check the same branch out twice and
the branch picker already marks branches that are checked out elsewhere.
### The `worktree_root` override
+7 -20
View File
@@ -4,8 +4,6 @@ import { Terminal } from "@xterm/xterm";
import "@xterm/xterm/css/xterm.css";
import { attachWsUrl } from "./routes.js";
import { FOCUS_SIDEBAR_EVENT } from "./catalog.js";
import { CommandSet, defineCommands } from "./commands.js";
import { escapeHtml } from "./html.js";
import { INITIAL_RECONNECT_MS, MAX_RECONNECT_MS } from "./reconnect.js";
import { applyModsToChar, cursorSeq, tildeSeq } from "./keys.js";
@@ -14,15 +12,6 @@ import { splitCommand, toolIconSvg } from "./toolIcon.js";
import type { ClientControl, Uuid } from "./types.js";
import { WebglPolicy } from "./webglRenderer.js";
// The pane's half of the command registry (#148): plain Escape hands the
// keyboard back to the session list, the pair to the list's Tab.
const paneKeys = new CommandSet(
defineCommands({
"pane.focusList": () => document.dispatchEvent(new CustomEvent(FOCUS_SIDEBAR_EVENT)),
}),
{ preventDefault: false },
);
/** One budget for the whole page — see webglRenderer.ts for why contexts are
* rationed rather than handed to every connection. */
const webglPolicy = new WebglPolicy();
@@ -221,15 +210,6 @@ export function createAttachConnection(
}
}
// Plain Escape hands the keyboard back to the session list instead of the
// shell/app. `false` tells xterm to skip its own handling of this event —
// which is also why the set doesn't `preventDefault`: swallowing here is
// xterm's return value, not the browser's.
term.attachCustomKeyEventHandler((e) => {
if (e.type !== "keydown") return true;
return !paneKeys.handle(e);
});
// Touch scrolling: xterm's built-in touch handling is jittery (drags crawl a
// line at a time and stall), so map finger drags ourselves. Two routes:
// • normal buffer → term.scrollLines walks the scrollback.
@@ -470,6 +450,12 @@ export function createAttachConnection(
const resizeDisp = term.onResize(({ cols, rows }) => {
sendControl({ type: "Resize", cols, rows });
});
// xterm has no built-in copy-on-select; mirror the selection to the
// clipboard ourselves (best-effort — insecure contexts lack the API).
const selectionDisp = term.onSelectionChange(() => {
const text = term.getSelection();
if (text) void navigator.clipboard?.writeText(text).catch(() => {});
});
window.addEventListener("resize", onResize);
const visual = window.visualViewport ?? null;
@@ -525,6 +511,7 @@ export function createAttachConnection(
visual?.removeEventListener("resize", onResize);
dataDisp.dispose();
resizeDisp.dispose();
selectionDisp.dispose();
releaseWebgl();
try {
ws.close();
-13
View File
@@ -414,19 +414,6 @@ describe("session cards (#107)", () => {
expect(line2?.textContent).toContain("exited");
});
// #162: a session recovered with no tmux behind it is known but not running,
// and never reads as something that exited.
it("shows dormant on line 2 for dormant sessions", () => {
const s = session({
id: "aaaaaaaa-0000-0000-0000-000000000118",
status: "Dormant",
});
const el = renderSidebar(storeWith([s])).el;
const line2 = rowFor(el, s.id)?.querySelector<HTMLElement>(".card-line2");
expect(line2?.textContent).toContain("dormant");
expect(line2?.textContent).not.toContain("exited");
});
it("carries the state word only in tooltips and aria labels, never as text", () => {
const s = session({
id: "aaaaaaaa-0000-0000-0000-000000000019",
-8
View File
@@ -53,10 +53,6 @@ export function requestOpenCommandBar(): void {
* connection pool. */
export const FOCUS_PANE_EVENT = "catalog-focus-pane";
/** Ask the shell to hand the keyboard back to the session list — what Escape
* does when the pane's terminal is focused. Handled in main.ts. */
export const FOCUS_SIDEBAR_EVENT = "catalog-focus-sidebar";
/** Ask the shell to open the spawn modal seeded from an existing session
* ("Spawn similar"). Detail is the session being cloned. */
export const SPAWN_SIMILAR_EVENT = "catalog-spawn-similar";
@@ -113,9 +109,6 @@ function activityRailMarkup(slug: string, stateWord: string): string {
function metaLine(s: Session, now: number): string {
const dead = s.status === "Dead";
// #162: known from its sidecar with nothing running behind it — a distinct
// fragment, since it never exited.
const dormant = s.status === "Dormant";
const blockers = sessionBlockers(s);
const slug = activityColorSlug(s.activity ?? "Idle");
const held = activityHeldMs(s.activity_since, now);
@@ -130,7 +123,6 @@ function metaLine(s: Session, now: number): string {
commandFragment(s.command),
blockers ? blockedFragment(blockers) : "",
dead ? `<span class="card-flag">exited</span>` : "",
dormant ? `<span class="card-flag">dormant</span>` : "",
];
return `<span class="card-line2" data-activity="${slug}">${fragments.join("")}</span>`;
}
-14
View File
@@ -99,20 +99,6 @@ describe("empty-query ordering", () => {
]);
});
// #162: docs/COMMAND-BAR.md ranks by whether you could attach, not by which
// way the pane went away.
it("sorts dormant sessions after live ones, like dead ones", () => {
const dormantRecent = session("dormant", {
status: "Dormant",
last_focused_at: "2026-06-01T00:00:00Z",
});
const liveStale = session("live", { last_focused_at: "2020-01-01T00:00:00Z" });
expect(orderSessionsForBar([dormantRecent, liveStale]).map((s) => s.id)).toEqual([
"live",
"dormant",
]);
});
it("treats a null last_focused_at as never focused", () => {
const nulled = session("nulled", { last_focused_at: null });
const stamped = session("stamped", { last_focused_at: "2020-01-01T00:00:00Z" });
+3 -5
View File
@@ -142,10 +142,8 @@ function createdAt(s: Session): number {
* matcher's last tie-break preserves. */
export function orderSessionsForBar(sessions: readonly Session[]): Session[] {
return [...sessions].sort((a, b) => {
// docs/COMMAND-BAR.md: anything with no live pane — dead or dormant (#162)
// — sorts after the sessions you could actually attach to.
const deadA = a.status === "Running" ? 0 : 1;
const deadB = b.status === "Running" ? 0 : 1;
const deadA = a.status === "Dead" ? 1 : 0;
const deadB = b.status === "Dead" ? 1 : 0;
if (deadA !== deadB) return deadA - deadB;
const focusA = focusedAt(a);
@@ -352,7 +350,7 @@ function renderRow(result: BarResult, index: number): HTMLElement {
row.dataset.kind = candidate.kind;
if (candidate.kind === "session") {
row.dataset.sessionId = candidate.session.id;
if (candidate.session.status !== "Running") row.classList.add("is-dead");
if (candidate.session.status === "Dead") row.classList.add("is-dead");
} else if (isActionCandidate(candidate)) {
row.dataset.actionId = candidate.action.id;
} else if (candidate.kind === "command") {
+12 -14
View File
@@ -133,31 +133,28 @@ describe("CommandSet", () => {
expect(run).toHaveBeenCalledTimes(2);
});
it("keeps plain keys deaf inside a form field, and chorded ones live", () => {
it("keeps plain keys deaf inside a form field, and leader chords live from the pane", () => {
const plain = vi.fn();
const chorded = vi.fn();
const set = new CommandSet(
defineCommands({ "spawn.open": plain, "focus.toggle": chorded }),
defineCommands({ "spawn.open": plain, "focus.toggle": chorded }, DEFAULT_LEADER),
{ leader: DEFAULT_LEADER },
);
const input = el("<input>");
expect(pressOn(set, input, "n")).toBe(false);
expect(plain).not.toHaveBeenCalled();
expect(pressOn(set, input, "H", { ctrlKey: true, shiftKey: true })).toBe(true);
// focus.toggle hangs off the leader (#166): a plain, unmodified Escape is
// left for the pane's shell/app instead.
const terminal = el('<textarea class="xterm-helper-textarea"></textarea>');
pressOn(set, terminal, "a", { ctrlKey: true });
expect(pressOn(set, terminal, "h")).toBe(true);
expect(chorded).toHaveBeenCalledTimes(1);
});
it("keeps pane commands live inside the terminal's textarea", () => {
const run = vi.fn();
const set = new CommandSet(defineCommands({ "pane.focusList": run }), {
preventDefault: false,
});
it("leaves a plain Escape for the pane instead of binding it", () => {
const set = new CommandSet(defineCommands({ "spawn.open": vi.fn() }));
const textarea = el('<textarea class="xterm-helper-textarea"></textarea>');
const e = key("Escape");
Object.defineProperty(e, "target", { value: textarea });
expect(set.handle(e)).toBe(true);
expect(run).toHaveBeenCalledTimes(1);
// xterm is told to swallow by the return value, not by the browser default.
expect(e.defaultPrevented).toBe(false);
expect(pressOn(set, textarea, "Escape")).toBe(false);
});
});
@@ -173,6 +170,7 @@ describe("the leader", () => {
const byId = new Map(commandSpecs("ctrl-b").map((s) => [s.id, s]));
expect(byId.get("pane.literalCtrlK")?.keys).toEqual(["ctrl-b ctrl-k"]);
expect(byId.get("pane.literalLeader")?.keys).toEqual(["ctrl-b ctrl-b"]);
expect(byId.get("focus.toggle")?.keys).toEqual(["ctrl-b h"]);
// Commands that do not hang off the leader are untouched.
expect(byId.get("bar.open")?.keys).toEqual(["ctrl-k"]);
});
+2 -8
View File
@@ -93,7 +93,8 @@ const SPECS: readonly SpecSource[] = [
id: "focus.toggle",
label: "Move the keyboard between the session list and the pane",
group: "Session list",
keys: ["ctrl-shift-h"],
keys: [`${DEFAULT_LEADER} h`],
leaderKeys: (leader) => [`${leader} h`],
scope: "global",
},
{
@@ -145,13 +146,6 @@ const SPECS: readonly SpecSource[] = [
keys: ["tab"],
scope: "list",
},
{
id: "pane.focusList",
label: "Hand the keyboard back to the session list, from the pane",
group: "Session list",
keys: ["esc"],
scope: "pane",
},
{
id: "help.toggle",
label: "Show or hide this help",
+33
View File
@@ -0,0 +1,33 @@
// @vitest-environment happy-dom
import { beforeEach, describe, expect, it } from "vitest";
import {
closeForceKillDialog,
openForceKillDialog,
openKillConfirmDialog,
} from "./killDialog.js";
describe("kill dialogs", () => {
beforeEach(() => {
closeForceKillDialog();
document.body.innerHTML = "";
});
it("focuses the kill button, so Enter confirms", () => {
const el = openKillConfirmDialog({
name: "alpha",
blockers: null,
onKill: () => Promise.resolve(),
});
expect(document.activeElement).toBe(el.querySelector(".kill-dialog-kill"));
});
it("focuses the force button too", () => {
const el = openForceKillDialog({
name: "alpha",
blockers: null,
reason: "worktree is dirty.",
onForce: () => Promise.resolve(),
});
expect(document.activeElement).toBe(el.querySelector(".kill-dialog-force"));
});
});
+3 -3
View File
@@ -132,9 +132,9 @@ function mount(p: KillDialogSpec): HTMLElement {
};
window.addEventListener("keydown", onKeydown, true);
open = { el, onKeydown };
// Cancel takes the focus in both prompts: a kill is never one keystroke
// away from whatever asked for the dialog.
cancel.focus();
// The acting button takes the focus in both prompts: the operator already
// asked for the kill, so Enter confirms it. Escape and the scrim cancel.
confirm.focus();
return el;
}
+18
View File
@@ -2,6 +2,7 @@ import { describe, expect, it } from "vitest";
import {
emptyLayout,
equalizeRatios,
leaves,
locateSession,
openSessions,
@@ -44,6 +45,23 @@ describe("splitLeaf", () => {
});
});
describe("equalizeRatios", () => {
it("resets nested splits to equal width, not just the newest one", () => {
const tab = tabWithOneLeaf("s1");
const first = leaves(tab.root)[0].id;
splitLeaf(tab, first, "s2", "row", true); // A|B, 0.5/0.5
const second = leaves(tab.root)[1].id;
splitLeaf(tab, second, "s3", "row", true); // A(0.5) | (B|C, 0.5/0.5) -> B,C at 0.25 each
equalizeRatios(tab.root);
const root = tab.root as Extract<typeof tab.root, { kind: "split" }>;
expect(root.ratio).toBeCloseTo(1 / 3);
const inner = root.b as Extract<typeof root.b, { kind: "split" }>;
expect(inner.ratio).toBeCloseTo(1 / 2);
});
});
describe("removeLeaf", () => {
it("collapses the parent split into the surviving sibling", () => {
const tab = tabWithOneLeaf("s1");
+11
View File
@@ -118,6 +118,17 @@ export function removeLeaf(tab: Tab, leafId: string): void {
tab.root = tab.root ? rec(tab.root) : null;
}
/** Reset every split ratio under `node` so all leaves get equal width,
* regardless of split direction or nesting order. Returns the leaf count. */
export function equalizeRatios(node: Node | null): number {
if (!node) return 0;
if (node.kind === "leaf") return 1;
const na = equalizeRatios(node.a);
const nb = equalizeRatios(node.b);
node.ratio = na / (na + nb);
return na + nb;
}
export function addTab(layout: Layout): Tab {
const tab: Tab = {
id: freshId("t"),
+7 -11
View File
@@ -35,7 +35,6 @@ import { loadLayout, saveLayout } from "./persistence.js";
import { createPreviewSchedule } from "./previewSchedule.js";
import {
FOCUS_PANE_EVENT,
FOCUS_SIDEBAR_EVENT,
OPEN_COMMAND_BAR_EVENT,
OPEN_SPAWN_EVENT,
SESSION_SELECT_EVENT,
@@ -440,10 +439,11 @@ const spawnKeys = new CommandSet(
}),
);
// Ctrl+Shift+H moves the keyboard between the session list and the pane, the
// web counterpart of the TUI's `ctrl-w h`. It has to be a chord: an attached
// terminal owns every plain keystroke, so nothing unmodified could ever get
// the focus back out of it.
// Leader-h moves the keyboard between the session list and the pane, the web
// counterpart of the TUI's `ctrl-w h`. It has to be a chord behind the
// leader: an attached terminal owns every plain keystroke, and Ctrl+Shift+H
// is already claimed by the browser (history), so nothing unmodified — or
// modified with a common browser chord — could get the focus back out of it.
//
// This set also owns the leader (`[tui].leader`, `ctrl-a` by default, served
// on `/api/info`): it holds the key as a partial chord and takes select-all off
@@ -528,8 +528,8 @@ window.addEventListener(
);
// Tab in the session list hands the keyboard to the pane; Escape in the pane
// hands it back — the pair the TUI gets from `tab` and `esc`.
// Tab in the session list hands the keyboard to the pane; leader-h (focus.toggle,
// above) hands it back the other way.
document.addEventListener(FOCUS_PANE_EVENT, () => {
// Handing the keyboard to a previewed pane is choosing it: the preview stops
// being provisional rather than snapping back the moment the list blurs.
@@ -537,10 +537,6 @@ document.addEventListener(FOCUS_PANE_EVENT, () => {
if (previewed) focusSession(previewed);
pool.focusTerminal();
});
document.addEventListener(FOCUS_SIDEBAR_EVENT, () => {
drawer.open(); // no-op on desktop; on mobile the list is off-canvas
sidebar.focusList();
});
/** Open/focus a session and reflect it in the shell chrome. Shared by the
* hash router and the sidebar's click-driven path (#133), which can't rely
+4
View File
@@ -10,6 +10,7 @@ import {
activeTab,
addTab,
closeTab,
equalizeRatios,
findLeaf,
freshId,
type Dir,
@@ -457,6 +458,7 @@ export function createMultiView(opts: MultiViewOptions): MultiView {
dropPreview();
const tab = activeTab(layout);
tab.focusedLeafId = splitLeaf(tab, leafId, null, dir, true);
equalizeRatios(tab.root);
render();
}
@@ -542,6 +544,7 @@ export function createMultiView(opts: MultiViewOptions): MultiView {
target.sessionId = sessionId;
} else {
tab.focusedLeafId = splitLeaf(tab, target.id, sessionId, "row", true);
equalizeRatios(tab.root);
}
render();
opts.onOpenSessionsChange?.();
@@ -815,6 +818,7 @@ export function createMultiView(opts: MultiViewOptions): MultiView {
if (source.kind === "pane" && self) removeLeaf(tab, self);
else if (loc) removeLeaf(loc.tab, loc.leaf.id);
tab.focusedLeafId = splitLeaf(tab, targetLeafId, sessionId, dir, after);
equalizeRatios(tab.root);
render();
}
+5 -4
View File
@@ -6,10 +6,7 @@ import { escapeHtml } from "./html.js";
export type Uuid = string;
export type IsoTimestamp = string;
/** `Dormant` (#162) is a session the daemon knows only from its sidecar —
* recovered after the tmux server died under it, so there is no pane behind it
* and nothing to attach to. Listed, never live. */
export type SessionStatus = "Running" | "Dead" | "Dormant";
export type SessionStatus = "Running" | "Dead";
/** Daemon-inferred activity dimension, orthogonal to `SessionStatus` (NAP-340).
* Absent on payloads from daemons that predate the field. `Error` is only set
@@ -59,6 +56,10 @@ 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). */