Compare commits

..
5 Commits
Author SHA1 Message Date
Miguel PalhasandClaude Opus 4.6 b339fecad3 feat(maestro): route maestro send through ACP for structured sessions
ci / rust (push) Waiting to run
ci / web (push) Waiting to run
ci / flake (push) Waiting to run
Structured sessions were rejected by `send_input` with NotTerminal.
Now `run_send` checks the session view and routes to
`send_structured_prompt` for structured sessions.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-09-08 13:43:01 +01:00
Miguel PalhasandClaude Opus 4.6 3c10a3a08a fix(maestro): skip Starting sessions in liveness watcher
ci / web (push) Failing after 21s
ci / flake (push) Successful in 3m30s
ci / rust (push) Successful in 6m8s
A structured session still booting its adapter has no ACP handle and
no tmux pane, so the liveness pass misclassified it as a vanished
terminal session and reaped it. Claude Code adapters take long enough
to boot that the 2s liveness tick fires before the handshake finishes.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-09-08 07:25:48 +01:00
Miguel Palhas fd79c8116a Merge remote-tracking branch 'origin/main' into tuba
ci / web (push) Failing after 7s
ci / flake (push) Successful in 3m15s
ci / rust (push) Successful in 5m54s
2026-09-07 21:49:31 +01:00
Miguel PalhasandClaude Opus 4.6 cb8b93f178 test(maestro): update tests for async structured spawn
Tests now assert `Starting` after create and use `wait_for_ready`
before checking post-handshake state. Unsupported command test
expects `BadRequest` instead of a persisted Dead session.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-09-07 21:28:22 +01:00
Miguel PalhasandClaude Opus 4.6 8d35f7f8c8 feat(maestro): async structured session spawn
Register a structured session immediately with `activity: Starting`
and run the ACP handshake in a background task. The HTTP caller gets
an instant response; the user can begin typing a prompt while the
adapter boots.

- Add `ActivityState::Starting` to the protocol
- Open the event log eagerly so the structured WS can connect during
  boot (empty transcript + idle state)
- `queue_acp_prompt` stashes a prompt while `Starting`; the background
  task drains it on successful handshake
- `apply_structured_activity` skips overwriting `Starting`
- `subscribe_acp_projection` allows connecting during `Starting`
- TUI/CLI map `Starting` to idle colour / "starting" label

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-09-07 21:08:08 +01:00
13 changed files with 470 additions and 268 deletions
+2 -2
View File
@@ -41,8 +41,8 @@ pub enum Command {
Attach(AttachArgs),
/// Kill a session by ID.
Kill(KillArgs),
/// Type text into a running session, as if it had been typed on an
/// attached terminal.
/// Send text to a running session. Terminal sessions receive raw
/// keystrokes; structured sessions receive an ACP prompt.
Send(SendArgs),
/// Install or update the agent activity hook (#106).
Hooks(HooksArgs),
+28 -15
View File
@@ -130,9 +130,12 @@ pub fn resolve_session(sessions: &[Session], needle: &str) -> Result<uuid::Uuid,
/// `maestro send`. Answers with the session that was typed into, so the
/// caller can name it once the daemon has accepted the text.
///
/// A bare ID skips the catalog fetch; anything else needs the list to match
/// against. Text comes from the arguments, or from stdin when there is none —
/// the shape a heredoc or a pipe wants.
/// Terminal sessions get raw keystrokes via `send_input`; structured sessions
/// get an ACP prompt via `send_structured_prompt` (`--no-enter` is ignored
/// there — ACP prompts are always complete messages).
///
/// Text comes from the arguments, or from stdin when there is none — the
/// shape a heredoc or a pipe wants.
#[allow(clippy::result_large_err)]
pub async fn run_send(
client: &DaemonClient,
@@ -155,19 +158,28 @@ pub async fn run_send(
return Err(CliError::NothingToSend);
}
let id = match uuid::Uuid::parse_str(args.session.trim()) {
Ok(id) => id,
Err(_) => {
let sessions = client.list_sessions().await?;
resolve_session(&sessions, &args.session)?
}
};
// We always need the session list: even a bare UUID needs the view field
// to decide which endpoint to hit.
let sessions = client.list_sessions().await?;
let id = resolve_session(&sessions, &args.session)?;
let view = sessions
.iter()
.find(|s| s.id == id)
.map(|s| s.view)
.unwrap_or(SessionView::Terminal);
let req = SendInputRequest {
text,
submit: !args.no_enter,
};
client.send_input(id, &req).await?;
match view {
SessionView::Structured => {
client.send_structured_prompt(id, &text).await?;
}
SessionView::Terminal => {
let req = SendInputRequest {
text,
submit: !args.no_enter,
};
client.send_input(id, &req).await?;
}
}
Ok(id)
}
@@ -216,6 +228,7 @@ fn print_session_table(out: &mut impl Write, sessions: &[Session], info: &Daemon
ActivityState::Idle => "idle",
ActivityState::AwaitingInput => "awaiting",
ActivityState::Error => "error",
ActivityState::Starting => "starting",
};
// Show the live foreground command alongside the spawn command when
// the daemon has observed one that differs (NAP-340).
@@ -2,7 +2,7 @@
//! was unit-tested, but nothing exercised the client, the error decoding, or
//! the unreachable-daemon path without compiling and forking a real daemon.
use maestro_protocol::{DaemonClient, DaemonInfo};
use maestro_protocol::{DaemonClient, DaemonInfo, SessionView};
use maestro_testing::{session, MockDaemon};
use uuid::Uuid;
@@ -202,6 +202,35 @@ async fn send_resolves_a_session_by_name() {
assert_eq!(sent, id);
}
/// A structured session routes through the ACP prompt endpoint, not the
/// terminal input one — `--no-enter` is irrelevant (ACP prompts are
/// complete messages).
#[tokio::test]
async fn send_to_structured_session_uses_acp_prompt() {
let mock = MockDaemon::start().await;
let id = Uuid::from_u128(13);
let mut s = session(id, "/tmp", "claude");
s.view = SessionView::Structured;
mock.add_session(s);
let args = maestro_cli::cli::SendArgs {
session: id.to_string(),
text: vec!["fix".into(), "the".into(), "tests".into()],
no_enter: false,
};
let sent = maestro_cli::sessions::run_send(&mock.client(), &args, &mut std::io::empty())
.await
.expect("send");
assert_eq!(sent, id);
// Terminal input must stay empty — the message went through ACP.
assert!(mock.inputs().is_empty());
let prompts = mock.structured_prompts();
assert_eq!(prompts.len(), 1);
assert_eq!(prompts[0].0, id);
assert_eq!(prompts[0].1.prompt, "fix the tests");
}
#[tokio::test]
async fn send_to_an_unknown_session_fails_without_typing_anything() {
let mock = MockDaemon::start().await;
+4 -2
View File
@@ -180,7 +180,7 @@ fn dir_error_status(code: &str) -> StatusCode {
fn session_error_status(code: &str) -> StatusCode {
match code {
"not_found" => StatusCode::NOT_FOUND,
"invalid_env" => StatusCode::BAD_REQUEST,
"invalid_env" | "bad_request" => StatusCode::BAD_REQUEST,
// A guard refusal is the daemon doing its job, not a failure (#79).
"kill_guarded" => StatusCode::CONFLICT,
// The session outlived its pane: nothing to type into, and nothing
@@ -483,7 +483,9 @@ async fn structured_prompt(
Json(body): Json<StructuredPromptRequest>,
) -> Response {
match s.manager.queue_acp_prompt(id, body.prompt).await {
Ok(QueueAcpPrompt::Queued) => StatusCode::NO_CONTENT.into_response(),
Ok(QueueAcpPrompt::Queued) | Ok(QueueAcpPrompt::Pending) => {
StatusCode::NO_CONTENT.into_response()
}
Ok(QueueAcpPrompt::InFlight) => error_response(
StatusCode::CONFLICT,
"turn_in_flight",
+208 -160
View File
@@ -6,12 +6,9 @@
//! handler interact with tmux.
use std::collections::{HashMap, HashSet};
use std::future::Future;
use std::path::PathBuf;
use std::pin::Pin;
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, Weak};
use std::task::{Context, Poll};
use std::time::Duration;
use chrono::Utc;
@@ -125,6 +122,9 @@ pub struct SessionRecord {
/// When `true`, the permission pump auto-responds `AllowAlways` to every
/// ACP permission request without surfacing it to clients.
pub auto_approve: bool,
/// A prompt submitted while the adapter was still booting (`Starting`).
/// The background handshake task drains this on success.
pub pending_prompt: Option<String>,
}
impl SessionRecord {
@@ -134,6 +134,13 @@ impl SessionRecord {
&& self.acp.is_some()
&& self.acp_events.is_some()
}
/// The session is structured and still booting its adapter.
fn is_starting_structured(&self) -> bool {
self.session.view == SessionView::Structured
&& self.session.activity == ActivityState::Starting
&& self.acp.is_none()
}
}
#[derive(Debug)]
@@ -208,6 +215,8 @@ pub enum SessionError {
NotTerminal(Uuid),
#[error("daemon is shutting down")]
ShuttingDown,
#[error("bad request: {0}")]
BadRequest(String),
/// The kill was refused (#79): the session's worktree holds unsaved work.
/// Nothing happened — the session keeps running and the tree is untouched,
/// so it stays the operator's handle on their own unfinished work.
@@ -227,6 +236,8 @@ pub enum BeginAcpTurn {
#[derive(Debug)]
pub enum QueueAcpPrompt {
Queued,
/// The adapter is still booting; the prompt is held until it is ready.
Pending,
InFlight,
Unavailable,
Failed(String),
@@ -253,6 +264,7 @@ impl SessionError {
SessionError::NotRunning(_) => "not_running",
SessionError::NotTerminal(_) => "not_terminal",
SessionError::ShuttingDown => "shutting_down",
SessionError::BadRequest(_) => "bad_request",
SessionError::KillGuarded(_) => "kill_guarded",
};
let unsaved_work = match self {
@@ -320,18 +332,6 @@ struct StructuredLifecycle {
}
impl StructuredLifecycle {
fn admit<'a>(
&'a self,
shutting_down: &AtomicBool,
) -> Result<StructuredCreate<'a>, SessionError> {
let _gate = self.gate.lock().expect("structured lifecycle gate");
if shutting_down.load(Ordering::Acquire) {
return Err(SessionError::ShuttingDown);
}
self.in_flight.fetch_add(1, Ordering::AcqRel);
Ok(StructuredCreate { lifecycle: self })
}
async fn wait_until_settled(&self) {
loop {
let settled = self.settled.notified();
@@ -343,30 +343,18 @@ impl StructuredLifecycle {
}
}
struct StructuredCreate<'a> {
lifecycle: &'a StructuredLifecycle,
}
impl Drop for StructuredCreate<'_> {
fn drop(&mut self) {
if self.lifecycle.in_flight.fetch_sub(1, Ordering::AcqRel) == 1 {
self.lifecycle.settled.notify_one();
}
}
}
struct AdmittedStructuredCreate<'a> {
create: Pin<Box<dyn Future<Output = Result<Session, SessionError>> + Send + 'a>>,
/// Field order matters: dropping a cancelled create rolls back its inner
/// future before this permit releases shutdown's in-flight barrier.
_permit: StructuredCreate<'a>,
}
impl Future for AdmittedStructuredCreate<'_> {
type Output = Result<Session, SessionError>;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
self.create.as_mut().poll(cx)
/// Release the structured-lifecycle in-flight counter that `create` acquires
/// before spawning the background ACP handshake task. Called when the task
/// finishes (success or failure) or when `create_inner` fails before the task
/// is spawned.
fn drop_structured_lifecycle_permit(inner: &Inner) {
if inner
.structured_lifecycle
.in_flight
.fetch_sub(1, Ordering::AcqRel)
== 1
{
inner.structured_lifecycle.settled.notify_one();
}
}
@@ -660,7 +648,9 @@ impl SessionManager {
let stream = {
let sessions = self.inner.sessions.lock().await;
let record = sessions.get(&id).ok_or(SessionError::NotFound(id))?;
if !record.is_live_structured() {
// Allow connecting while the adapter is still booting
// (`Starting`) — the event stream exists, it is just empty.
if !record.is_live_structured() && !record.is_starting_structured() {
return Ok(None);
}
let Some(stream) = &record.acp_events else {
@@ -745,8 +735,14 @@ impl SessionManager {
text: String,
) -> Result<QueueAcpPrompt, SessionError> {
let (acp, stream) = {
let sessions = self.inner.sessions.lock().await;
let record = sessions.get(&id).ok_or(SessionError::NotFound(id))?;
let mut sessions = self.inner.sessions.lock().await;
let record = sessions.get_mut(&id).ok_or(SessionError::NotFound(id))?;
// While the adapter is booting, stash the prompt for the
// background handshake task to drain on success.
if record.is_starting_structured() {
record.pending_prompt = Some(text);
return Ok(QueueAcpPrompt::Pending);
}
if !record.is_live_structured() {
return Ok(QueueAcpPrompt::Unavailable);
}
@@ -1217,17 +1213,42 @@ impl SessionManager {
}
/// Spawn a new session. On success returns the freshly-created `Session`.
///
/// For structured sessions, the ACP handshake runs in a background task
/// after the session is registered — the HTTP caller gets an instant
/// response with `activity: Starting`. The `StructuredLifecycle` gate is
/// held across the background task so `shutdown()` waits for it.
pub async fn create(&self, req: CreateSessionRequest) -> Result<Session, SessionError> {
if req.view == SessionView::Structured {
let permit = self
// Acquire the lifecycle gate synchronously (rejects if shutting
// down). The background handshake task calls
// `drop_structured_lifecycle_permit` when it finishes; we must
// not use the RAII `StructuredCreate` because it would drop at
// the end of `create_inner`, before the handshake completes.
let _gate = self
.inner
.structured_lifecycle
.admit(&self.inner.shutting_down)?;
return AdmittedStructuredCreate {
create: Box::pin(self.create_inner(req)),
_permit: permit,
.gate
.lock()
.expect("structured lifecycle gate");
if self.inner.shutting_down.load(Ordering::Acquire) {
return Err(SessionError::ShuttingDown);
}
.await;
self.inner
.structured_lifecycle
.in_flight
.fetch_add(1, Ordering::AcqRel);
drop(_gate);
return match self.create_inner(req).await {
Ok(session) => Ok(session),
Err(e) => {
// create_inner failed before spawning the background
// task, so release the permit here.
drop_structured_lifecycle_permit(&self.inner);
Err(e)
}
};
}
self.create_inner(req).await
}
@@ -1289,32 +1310,39 @@ impl SessionManager {
{
session_env.insert("MAESTRO_URL".into(), url);
}
// A structured session (#134) gets no tmux pane: the agent is a
// child of the daemon, driven over ACP. An adapter that cannot come
// up still yields a session — Dead, with the reason on it — so the
// operator sees what to fix instead of a silently missing session.
let (acp, acp_failure) = match req.view {
SessionView::Structured => match structured_adapter(&req.command) {
Err(error) => (None, Some(error)),
Ok(adapter) => match maestro_acp::spawn(
SpawnConfig::new(
&effective_cwd,
session_env
.iter()
.map(|(k, v)| (k.clone(), v.clone()))
.collect(),
)
.adapter(adapter),
)
.await
{
Ok(acp) => (Some(Arc::new(acp)), None),
Err(e) => {
tracing::warn!(%id, error = %e, "structured session: adapter failed to start");
(None, Some(e.to_string()))
// ── Structured path: validate adapter, open event log, register
// the session immediately with `Starting`, then handshake in the
// background so the HTTP caller gets an instant response.
//
// ── Terminal path: unchanged — tmux spawn is fast enough to stay
// synchronous.
// Structured: validate adapter, open event log eagerly. Terminal:
// spawn the tmux pane synchronously (it is fast).
let structured_spawn = match req.view {
SessionView::Structured => {
// Fail fast on an unsupported command before touching disk.
let adapter = structured_adapter(&req.command).map_err(SessionError::BadRequest)?;
// Open the event log eagerly so the structured WS can
// connect before the adapter is up.
let acp_events = match self.inner.sidecar.open_events(id) {
Ok(log) => {
let (tx, _) = broadcast::channel(EVENTS_CAPACITY);
let (projected_tx, _) = broadcast::channel(EVENTS_CAPACITY);
Arc::new(Mutex::new(AcpEventStream {
log,
live: tx,
projection: Projection::default(),
projected_live: projected_tx,
pending_approvals: HashMap::new(),
operation: Arc::new(Mutex::new(())),
dispatch: AcpDispatchState::default(),
activity_sequence: 0,
}))
}
},
},
Err(e) => return Err(SessionError::Sidecar(e)),
};
Some((adapter, acp_events))
}
SessionView::Terminal => {
let spawned = {
let tmux = self.inner.tmux.clone();
@@ -1328,53 +1356,10 @@ impl SessionManager {
if let Err(e) = spawned {
return Err(SessionError::Tmux(e));
}
(None, None)
None
}
};
let acp_events = if let Some(acp) = &acp {
let persisted =
match self
.inner
.sidecar
.read_events(id, EventCursor::Since(0), usize::MAX)
{
Ok(events) => events,
Err(error) => {
acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
.await;
return Err(SessionError::Sidecar(error));
}
};
match self.inner.sidecar.open_events(id) {
Ok(log) => {
let (tx, _) = broadcast::channel(EVENTS_CAPACITY);
let (projected_tx, _) = broadcast::channel(EVENTS_CAPACITY);
let mut projection = Projection::default();
for event in &persisted {
apply_persisted_event(&mut projection, event);
}
Some(Arc::new(Mutex::new(AcpEventStream {
log,
live: tx,
projection,
projected_live: projected_tx,
pending_approvals: HashMap::new(),
operation: Arc::new(Mutex::new(())),
dispatch: AcpDispatchState::default(),
activity_sequence: 0,
})))
}
Err(e) => {
acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
.await;
return Err(SessionError::Sidecar(e));
}
}
} else {
None
};
// Capture spawn-history inputs before `req` is partially moved into
// `session`. `history_cwd` is the *chosen* working dir (the request
// `cwd`), not a generated worktree checkout path — that's what the
@@ -1384,10 +1369,13 @@ impl SessionManager {
resolve_name(&mut metadata);
let worktree_branch = req.worktree.as_ref().map(|w| w.branch.clone());
let is_structured = structured_spawn.is_some();
let events_arc = structured_spawn.as_ref().map(|(_, e)| e.clone());
let session = Session {
id,
cwd: effective_cwd,
command: req.command,
cwd: effective_cwd.clone(),
command: req.command.clone(),
env: req.env,
metadata,
worktree: worktree_path.as_ref().zip(worktree_base_repo.as_ref()).map(
@@ -1396,67 +1384,41 @@ impl SessionManager {
base_repo: base_repo.clone(),
},
),
status: if acp_failure.is_some() {
SessionStatus::Dead
} else {
SessionStatus::Running
},
status: SessionStatus::Running,
created_at: Utc::now(),
// A terminal spawn begins active until its watcher sees otherwise;
// a structured spawn has no turn yet and begins idle.
foreground: None,
activity: match (req.view, acp_failure.is_some()) {
(_, true) => ActivityState::Error,
(SessionView::Structured, false) => ActivityState::Idle,
(SessionView::Terminal, false) => ActivityState::Active,
activity: if is_structured {
ActivityState::Starting
} else {
ActivityState::Active
},
activity_detail: acp_failure,
activity_detail: None,
last_activity_at: Utc::now(),
activity_since: chrono::Utc::now(),
current_cwd: None,
// The watcher fills this in on its first blocked-work pass; a
// fresh tree has nothing in the way anyway.
blocked_work: None,
last_focused_at: None,
acp_session_id: acp.as_ref().map(|a| a.session_id().to_string()),
acp_session_id: None,
view: req.view,
};
if let Err(e) = self.inner.sidecar.write(&session) {
// The process already started; end it so no ghost survives a
// failed sidecar write, then unwind the FIFO + worktree.
match &acp {
Some(acp) => {
acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
.await
}
None => {
let tmux = self.inner.tmux.clone();
let _ = tmux_blocking(move || tmux.kill_session(id)).await;
}
if !is_structured {
let tmux = self.inner.tmux.clone();
let _ = tmux_blocking(move || tmux.kill_session(id)).await;
}
return Err(SessionError::Sidecar(e));
}
let event_pump = acp
.as_ref()
.zip(acp_events.as_ref())
.map(|(acp, events)| (acp.clone(), events.clone()));
{
let mut sessions = self.inner.sessions.lock().await;
let event_pump = event_pump.map(|(acp, events)| {
let done = start_acp_event_pump(&self.inner, id, &acp, events.clone());
start_acp_permission_pump(&self.inner, id, &acp, events, req.auto_approve);
done
});
sessions.insert(
id,
SessionRecord {
session: session.clone(),
fifo_path,
worktree_path,
worktree_base_repo,
worktree_path: worktree_path.clone(),
worktree_base_repo: worktree_base_repo.clone(),
bridge,
reported_activity: None,
settled: false,
@@ -1464,11 +1426,12 @@ impl SessionManager {
naming: false,
focused_since: None,
awaiting_suppressed: false,
acp,
acp_events,
acp: None,
acp_events: events_arc,
structured_activity_watermark: None,
acp_event_pump: event_pump,
acp_event_pump: None,
auto_approve: req.auto_approve,
pending_prompt: None,
},
);
}
@@ -1497,6 +1460,78 @@ impl SessionManager {
}
pending.commit();
// ── Background ACP handshake for structured sessions ──
//
// The session is registered and visible to clients. The adapter
// spawn + handshake runs in a background task; on completion it
// fills in `acp`, starts the event/permission pumps, and
// transitions activity from `Starting` to `Idle` (or to `Error`
// on failure). The `StructuredLifecycle` permit is held until the
// task finishes so `shutdown()` waits for it.
if let Some((adapter, events)) = structured_spawn {
let inner = self.inner.clone();
let env: HashMap<String, String> = session_env
.iter()
.map(|(k, v)| (k.clone(), v.clone()))
.collect();
let cwd = effective_cwd;
let auto_approve = req.auto_approve;
tokio::spawn(async move {
let result = maestro_acp::spawn(SpawnConfig::new(&cwd, env).adapter(adapter)).await;
match result {
Ok(acp_session) => {
let acp = Arc::new(acp_session);
let done = start_acp_event_pump(&inner, id, &acp, events.clone());
start_acp_permission_pump(&inner, id, &acp, events.clone(), auto_approve);
// Drain any prompt submitted while Starting.
let pending_prompt = {
let mut sessions = inner.sessions.lock().await;
let Some(record) = sessions.get_mut(&id) else {
// Session was killed while we were booting.
acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
.await;
drop_structured_lifecycle_permit(&inner);
return;
};
record.acp = Some(acp.clone());
record.acp_event_pump = Some(done);
record.session.acp_session_id = Some(acp.session_id().to_string());
record.session.activity = ActivityState::Idle;
record.session.activity_since = Utc::now();
let _ = inner.sidecar.write(&record.session);
let _ = inner.events.send(CatalogEvent::SessionUpdated {
session: record.session.clone(),
});
record.pending_prompt.take()
};
if let Some(prompt) = pending_prompt {
let manager = SessionManager {
inner: inner.clone(),
};
let _ = manager.queue_acp_prompt(id, prompt).await;
}
}
Err(e) => {
tracing::warn!(%id, error = %e, "structured session: adapter failed to start");
let mut sessions = inner.sessions.lock().await;
if let Some(record) = sessions.get_mut(&id) {
record.session.status = SessionStatus::Dead;
record.session.activity = ActivityState::Error;
record.session.activity_detail = Some(e.to_string());
record.session.activity_since = Utc::now();
let _ = inner.sidecar.write(&record.session);
let _ = inner.events.send(CatalogEvent::SessionUpdated {
session: record.session.clone(),
});
}
}
}
drop_structured_lifecycle_permit(&inner);
});
}
Ok(session)
}
@@ -1682,6 +1717,7 @@ impl SessionManager {
structured_activity_watermark: None,
acp_event_pump: None,
auto_approve: false,
pending_prompt: None,
},
);
let _ = self
@@ -1862,6 +1898,7 @@ impl SessionManager {
// Recovery cannot know the original preset's flag; the
// operator re-enables it by restarting with the preset.
auto_approve: false,
pending_prompt: None,
},
);
if let Some((acp, events)) = event_pump {
@@ -2321,6 +2358,11 @@ async fn apply_structured_activity(
if record.session.view != SessionView::Structured {
return;
}
// While the adapter is still booting, the background handshake task
// owns the activity state — the activity pass must not overwrite it.
if record.session.activity == ActivityState::Starting {
return;
}
let current_events = record.acp_events.as_ref();
match (expected_events, current_events) {
(Some((expected, sequence)), Some(current)) if Arc::ptr_eq(expected, current) => {
@@ -2586,6 +2628,12 @@ async fn run_liveness_pass(inner: &Arc<Inner>) {
let mut tmux_backed = Vec::new();
let mut acp_backed = Vec::new();
for (id, record) in running {
if record.is_starting_structured() {
// The adapter hasn't finished its handshake yet — no ACP
// handle and no tmux pane. Skip; the background task owns
// this session's lifecycle until the handshake settles.
continue;
}
match &record.acp {
Some(acp) => acp_backed.push((*id, acp.clone())),
None => tmux_backed.push(*id),
+135 -81
View File
@@ -130,6 +130,25 @@ async fn wait_for_exit(pid: u32) {
assert!(!pid_alive(pid), "pid {pid} still alive");
}
/// Wait for a session's background ACP handshake to complete (activity
/// transitions out of `Starting`). Returns the refreshed session.
async fn wait_for_ready(manager: &SessionManager, id: uuid::Uuid) -> maestro_protocol::Session {
let deadline = Instant::now() + Duration::from_secs(10);
loop {
let session = manager
.snapshot()
.await
.into_iter()
.find(|s| s.id == id)
.expect("session exists");
if session.activity != ActivityState::Starting {
return session;
}
assert!(Instant::now() < deadline, "handshake timed out");
tokio::time::sleep(Duration::from_millis(20)).await;
}
}
#[tokio::test]
async fn structured_spawn_starts_the_adapter_in_cwd_with_session_env() {
let fx = DaemonFixture::new();
@@ -138,7 +157,7 @@ async fn structured_spawn_starts_the_adapter_in_cwd_with_session_env() {
let mut env = fake.env();
env.insert("FAKE_ACP_ECHO_ENV".into(), "MAESTRO_SESSION_ID".into());
let session = fx
let created = fx
.manager
.create(structured_command(
cwd.path(),
@@ -148,8 +167,12 @@ async fn structured_spawn_starts_the_adapter_in_cwd_with_session_env() {
.await
.expect("create");
assert_eq!(session.view, SessionView::Structured);
assert_eq!(session.status, SessionStatus::Running);
assert_eq!(created.view, SessionView::Structured);
assert_eq!(created.status, SessionStatus::Running);
assert_eq!(created.activity, ActivityState::Starting);
assert!(created.acp_session_id.is_none());
let session = wait_for_ready(&fx.manager, created.id).await;
assert_eq!(session.activity, ActivityState::Idle);
let acp_id = session.acp_session_id.clone().expect("acp session id");
assert!(acp_id.starts_with("fake-"), "{acp_id}");
@@ -164,7 +187,7 @@ async fn structured_spawn_starts_the_adapter_in_cwd_with_session_env() {
"{log}"
);
assert!(
log.contains(&format!("env MAESTRO_SESSION_ID={}", session.id)),
log.contains(&format!("env MAESTRO_SESSION_ID={}", created.id)),
"{log}"
);
@@ -232,6 +255,7 @@ async fn structured_activity_tracks_turn_completion_and_failure() {
.create(structured_command(cwd.path(), env, "codex"))
.await
.unwrap();
let session = wait_for_ready(&fx.manager, session.id).await;
assert!(matches!(
fx.manager
@@ -295,11 +319,12 @@ 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
let created = fx
.manager
.create(structured(cwd.path(), fake.env()))
.await
.expect("create");
let session = wait_for_ready(&fx.manager, created.id).await;
wait_for_async(Duration::from_secs(5), || async {
fx.manager
@@ -397,6 +422,7 @@ async fn queued_prompt_is_a_durable_user_transcript_row() {
.create(structured(cwd.path(), fake.env()))
.await
.expect("create");
let session = wait_for_ready(&fx.manager, session.id).await;
assert!(matches!(
fx.manager
@@ -442,6 +468,7 @@ async fn stale_turn_completion_cannot_end_the_next_turn() {
.create(structured(cwd.path(), fake.env()))
.await
.unwrap();
wait_for_ready(&fx.manager, session.id).await;
let maestro_daemon::session::BeginAcpTurn::Started(_, first) =
fx.manager.begin_acp_turn(session.id).await.unwrap()
@@ -485,6 +512,7 @@ async fn cancelled_prompt_that_never_settles_fails_the_session() {
.create(structured(cwd.path(), env))
.await
.unwrap();
wait_for_ready(&fx.manager, session.id).await;
assert!(matches!(
fx.manager
@@ -523,6 +551,7 @@ async fn cancellation_settlement_allows_the_next_prompt_permission() {
.create(structured(cwd.path(), env))
.await
.unwrap();
wait_for_ready(&fx.manager, session.id).await;
assert!(matches!(
fx.manager
@@ -582,11 +611,12 @@ async fn recovery_loads_the_acp_session_without_duplicating_history() {
let mut env = fake.env();
env.insert("FAKE_ACP_LOAD_REPLAY".into(), "2".into());
let session = fx
let created = fx
.manager
.create(structured_command(cwd.path(), env, "codex"))
.await
.expect("create");
let session = wait_for_ready(&fx.manager, created.id).await;
let acp_id = session.acp_session_id.clone().expect("acp id");
wait_for_async(Duration::from_secs(5), || async {
fx.manager
@@ -711,6 +741,7 @@ async fn shutdown_preserves_a_pristine_structured_worktree_for_recovery() {
.await
.expect("create");
assert_eq!(session.cwd, worktree);
wait_for_ready(&fx.manager, session.id).await;
push_all_to_origin(repo.path(), origin.path());
fx.manager.shutdown().await;
@@ -759,6 +790,7 @@ async fn failed_recovery_cleans_its_fifo() {
.create(structured(cwd.path(), fake.env()))
.await
.expect("create");
wait_for_ready(&fx.manager, session.id).await;
let old_pid = fake.pid();
fx.manager.shutdown().await;
wait_for_exit(old_pid).await;
@@ -792,12 +824,15 @@ async fn missing_adapter_yields_a_failed_session_naming_the_install_command() {
empty.path().to_string_lossy().into_owned(),
)]);
let session = fx
let created = fx
.manager
.create(structured_command(cwd.path(), env, "codex"))
.await
.expect("a failed session, not an error");
.expect("create registers a Starting session");
assert_eq!(created.activity, ActivityState::Starting);
// The background handshake fails because the adapter binary is missing.
let session = wait_for_ready(&fx.manager, created.id).await;
assert_eq!(session.status, SessionStatus::Dead);
assert_eq!(session.activity, ActivityState::Error);
let detail = session.activity_detail.clone().expect("actionable detail");
@@ -810,13 +845,13 @@ async fn missing_adapter_yields_a_failed_session_naming_the_install_command() {
// It is a real session: listed, on disk, and dismissable like any other
// dead one.
assert!(fx.manager.has(session.id).await);
assert!(fx.manager.has(created.id).await);
assert!(fx
.sidecar_dir()
.join(format!("{}.json", session.id))
.join(format!("{}.json", created.id))
.exists());
fx.manager.kill(session.id, false).await.expect("dismiss");
assert!(!fx.manager.has(session.id).await);
fx.manager.kill(created.id, false).await.expect("dismiss");
assert!(!fx.manager.has(created.id).await);
}
#[tokio::test]
@@ -825,20 +860,20 @@ async fn unsupported_structured_command_fails_without_launching_claude() {
let fake = Fake::new();
let cwd = TempDir::new().unwrap();
let session = fx
let err = fx
.manager
.create(structured_command(cwd.path(), fake.env(), "opencode"))
.await
.expect("failed structured session is persisted");
.expect_err("unsupported command is rejected at request validation");
assert_eq!(session.status, SessionStatus::Dead);
let detail = session.activity_detail.expect("selection error");
assert!(detail.contains("Claude Code or Codex"), "{detail}");
assert!(
matches!(err, maestro_daemon::SessionError::BadRequest(ref msg) if msg.contains("Claude Code or Codex")),
"{err:?}"
);
assert!(
fake.log().is_empty(),
"unsupported command launched an adapter"
);
fx.manager.kill(session.id, false).await.expect("dismiss");
}
#[tokio::test]
@@ -853,6 +888,7 @@ async fn kill_sends_cancel_then_sigterm_and_removes_the_sidecar() {
.create(structured(cwd.path(), fake.env()))
.await
.expect("create");
wait_for_ready(&fx.manager, session.id).await;
let pid = fake.pid();
assert!(pid_alive(pid));
@@ -895,6 +931,8 @@ async fn shutdown_reaps_every_adapter_child() {
.create(structured(cwd.path(), b.env()))
.await
.expect("create b");
wait_for_ready(&fx.manager, sa.id).await;
wait_for_ready(&fx.manager, sb.id).await;
let (pa, pb) = (a.pid(), b.pid());
assert!(pid_alive(pa) && pid_alive(pb));
@@ -929,10 +967,13 @@ async fn dismissal_during_shutdown_does_not_restore_recovery_artifacts() {
.create(structured(cwd.path(), env))
.await
.expect("create");
fx.manager
let blocker_session = fx
.manager
.create(structured(cwd.path(), blocker.env()))
.await
.expect("create blocker");
wait_for_ready(&fx.manager, session.id).await;
wait_for_ready(&fx.manager, blocker_session.id).await;
let manager = fx.manager.clone();
let shutdown = tokio::spawn(async move { manager.shutdown().await });
@@ -984,6 +1025,7 @@ async fn shutdown_does_not_recover_an_adapter_that_already_exited() {
.create(structured(cwd.path(), env))
.await
.expect("create");
wait_for_ready(&fx.manager, session.id).await;
wait_for_exit(fake.pid()).await;
fx.manager.shutdown().await;
@@ -1017,23 +1059,31 @@ async fn shutdown_waits_for_structured_creation_and_rejects_new_creates() {
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 });
// create() returns immediately; the handshake runs in the background
// and holds a lifecycle permit.
let session = fx
.manager
.create(structured(cwd.path(), env))
.await
.expect("create");
// Wait until the adapter has received session/new (handshake in progress).
wait_for_async(Duration::from_secs(5), || async {
fake.log().contains("recv session/new")
})
.await;
// Shutdown must wait for the in-flight background handshake.
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"
"shutdown returned before handshake finished"
);
let session = create.await.expect("create task").expect("create");
// The handshake completes in the background (500ms delay).
wait_for_ready(&fx.manager, session.id).await;
let pid = fake.pid();
shutdown.await.expect("shutdown task");
@@ -1066,27 +1116,37 @@ async fn structured_handshakes_run_independently() {
gate.to_string_lossy().into_owned(),
);
let first_cwd = cwd.path().to_path_buf();
let manager = fx.manager.clone();
let first =
tokio::spawn(async move { manager.create(structured(&first_cwd, blocked_env)).await });
wait_for_async(Duration::from_secs(5), || async {
blocked.log().contains("recv session/new")
})
.await;
// Both create() calls return immediately with Starting; the first
// adapter's handshake is gated while the second runs freely.
let first = fx
.manager
.create(structured(cwd.path(), blocked_env))
.await
.expect("first create");
let second = fx
.manager
.create(structured(cwd.path(), ready.env()))
.await
.expect("second create");
let second = tokio::time::timeout(
Duration::from_secs(5),
fx.manager.create(structured(cwd.path(), ready.env())),
)
.await
.expect("second handshake was blocked by the first")
.expect("second create");
assert_eq!(second.status, SessionStatus::Running);
assert!(!first.is_finished(), "the gate must still block the first");
// The second handshake completes independently of the first.
let second_ready = wait_for_ready(&fx.manager, second.id).await;
assert_eq!(second_ready.activity, ActivityState::Idle);
// The first handshake is still blocked.
let first_snap = fx
.manager
.snapshot()
.await
.into_iter()
.find(|s| s.id == first.id)
.unwrap();
assert_eq!(first_snap.activity, ActivityState::Starting);
std::fs::write(&gate, b"release").unwrap();
let first = first.await.expect("first task").expect("first create");
let first_ready = wait_for_ready(&fx.manager, first.id).await;
assert_eq!(first_ready.activity, ActivityState::Idle);
fx.manager.kill(first.id, false).await.expect("kill first");
fx.manager
.kill(second.id, false)
@@ -1114,16 +1174,18 @@ async fn shutdown_waits_for_every_admitted_structured_create() {
gate_b.to_string_lossy().into_owned(),
);
let create_a = {
let manager = fx.manager.clone();
let cwd = cwd.path().to_path_buf();
tokio::spawn(async move { manager.create(structured(&cwd, env_a)).await })
};
let create_b = {
let manager = fx.manager.clone();
let cwd = cwd.path().to_path_buf();
tokio::spawn(async move { manager.create(structured(&cwd, env_b)).await })
};
// Both create() calls return immediately; the background handshake
// tasks hold lifecycle permits and are blocked on their gates.
let session_a = fx
.manager
.create(structured(cwd.path(), env_a))
.await
.expect("create a");
let session_b = fx
.manager
.create(structured(cwd.path(), env_b))
.await
.expect("create b");
wait_for_async(Duration::from_secs(5), || async {
a.log().contains("recv session/new") && b.log().contains("recv session/new")
})
@@ -1143,14 +1205,14 @@ async fn shutdown_waits_for_every_admitted_structured_create() {
assert!(!shutdown.is_finished());
std::fs::write(&gate_a, b"release").unwrap();
let session_a = create_a.await.expect("create a task").expect("create a");
wait_for_ready(&fx.manager, session_a.id).await;
assert!(
!shutdown.is_finished(),
"shutdown ignored the second create"
);
std::fs::write(&gate_b, b"release").unwrap();
let session_b = create_b.await.expect("create b task").expect("create b");
wait_for_ready(&fx.manager, session_b.id).await;
shutdown.await.expect("shutdown task");
wait_for_exit(a.pid()).await;
@@ -1166,7 +1228,7 @@ async fn shutdown_waits_for_every_admitted_structured_create() {
}
#[tokio::test]
async fn cancelling_a_structured_handshake_rolls_back_and_settles_admission() {
async fn killing_a_starting_session_rolls_back_and_settles_admission() {
let fx = DaemonFixture::new();
let fake = Fake::new();
let repo = TempDir::new().unwrap();
@@ -1192,40 +1254,30 @@ async fn cancelling_a_structured_handshake_rolls_back_and_settles_admission() {
auto_approve: false,
};
let create = {
let manager = fx.manager.clone();
tokio::spawn(async move { manager.create(request).await })
};
// create() returns immediately with Starting; the handshake is gated.
let session = fx.manager.create(request).await.expect("create");
assert_eq!(session.activity, ActivityState::Starting);
wait_for_async(Duration::from_secs(5), || async {
fake.log().contains("recv session/new")
})
.await;
let id = fake
.log()
.lines()
.find_map(|line| line.strip_prefix("env MAESTRO_SESSION_ID="))
.and_then(|id| uuid::Uuid::parse_str(id).ok())
.expect("session id in adapter log");
let pid = fake.pid();
create.abort();
let shutdown = {
let manager = fx.manager.clone();
tokio::spawn(async move { manager.shutdown().await })
};
assert!(create.await.unwrap_err().is_cancelled());
tokio::time::timeout(Duration::from_secs(5), shutdown)
// Kill the session while its handshake is still in progress.
// force=true because the worktree has an unpushed branch commit.
fx.manager.kill(session.id, true).await.expect("kill");
// Release the gate so the background task can finish and settle.
std::fs::write(&gate, b"release").unwrap();
// Shutdown must not hang — the lifecycle permit is settled.
tokio::time::timeout(Duration::from_secs(5), fx.manager.shutdown())
.await
.expect("cancelled admission kept shutdown waiting")
.expect("shutdown task");
.expect("shutdown hung on a settled permit");
wait_for_exit(pid).await;
wait_for_async(Duration::from_secs(5), || async {
!fx.fifo_dir().join(format!("{id}.fifo")).exists()
&& !repo.path().join("cancelled-create").exists()
})
.await;
assert!(fx.manager.snapshot().await.is_empty());
let id = id.to_string();
let id = session.id.to_string();
assert!(
std::fs::read_dir(fx.sidecar_dir())
.into_iter()
@@ -1235,7 +1287,7 @@ async fn cancelling_a_structured_handshake_rolls_back_and_settles_admission() {
.file_name()
.to_string_lossy()
.starts_with(&id)),
"cancelled create left a sidecar artifact"
"killed Starting session left a sidecar artifact"
);
}
@@ -1252,6 +1304,7 @@ async fn shutdown_drains_final_notifications_to_the_log() {
.create(structured(cwd.path(), env))
.await
.expect("create");
wait_for_ready(&fx.manager, session.id).await;
wait_for_async(Duration::from_secs(5), || async {
fx.manager
.read_acp_events(session.id, EventCursor::Since(0), 10)
@@ -1282,6 +1335,7 @@ async fn adapter_exit_marks_the_session_dead_and_keeps_it_listed() {
.create(structured(cwd.path(), fake.env()))
.await
.expect("create");
wait_for_ready(&fx.manager, session.id).await;
fx.manager.start_liveness_watcher(Duration::from_millis(50));
// The agent dies out from under the daemon: the liveness pass, not tmux,
+11 -2
View File
@@ -154,7 +154,8 @@ async fn structured_session(h: &Harness, fake: &FakeAcp) -> Uuid {
}
async fn structured_session_with_env(h: &Harness, env: HashMap<String, String>) -> Uuid {
h.manager
let id = h
.manager
.create(CreateSessionRequest {
view: SessionView::Structured,
cwd: "/tmp".into(),
@@ -166,7 +167,9 @@ async fn structured_session_with_env(h: &Harness, env: HashMap<String, String>)
})
.await
.unwrap()
.id
.id;
wait_for_session_activity(h, id, ActivityState::Idle).await;
id
}
async fn wait_for_session_activity(h: &Harness, id: Uuid, expected: ActivityState) {
@@ -1494,6 +1497,12 @@ async fn events_ws_streams_structured_activity_without_hook_overrides() {
let maestro_protocol::CatalogEvent::SessionAdded { session } = next_event(&mut ws).await else {
panic!("expected structured SessionAdded")
};
assert_eq!(session.activity, ActivityState::Starting);
// The background handshake completes and sends SessionUpdated → Idle.
let maestro_protocol::CatalogEvent::SessionUpdated { session } = next_event(&mut ws).await
else {
panic!("expected SessionUpdated after handshake")
};
assert_eq!(session.activity, ActivityState::Idle);
let client = reqwest::Client::new();
+7 -1
View File
@@ -175,6 +175,10 @@ pub enum ActivityState {
AwaitingInput,
/// An agent reported a failure, or a structured session/turn failed.
Error,
/// The ACP adapter is booting; the session is registered but cannot accept
/// prompts yet. Older clients that do not know this variant degrade it to
/// `Idle` via the lenient deserializer — acceptable.
Starting,
}
impl<'de> Deserialize<'de> for ActivityState {
@@ -187,6 +191,7 @@ impl<'de> Deserialize<'de> for ActivityState {
"Active" => Self::Active,
"AwaitingInput" => Self::AwaitingInput,
"Error" => Self::Error,
"Starting" => Self::Starting,
// Known or not, any other spelling degrades to the neutral default
_ => Self::Idle,
})
@@ -225,9 +230,10 @@ where
"Idle" => Ok(ActivityState::Idle),
"AwaitingInput" => Ok(ActivityState::AwaitingInput),
"Error" => Ok(ActivityState::Error),
"Starting" => Ok(ActivityState::Starting),
other => Err(D::Error::unknown_variant(
other,
&["Active", "Idle", "AwaitingInput", "Error"],
&["Active", "Idle", "AwaitingInput", "Error", "Starting"],
)),
}
}
+38 -1
View File
@@ -22,7 +22,8 @@ use chrono::Utc;
use futures_util::{SinkExt, StreamExt};
use maestro_protocol::{
routes, ApiError, CatalogEvent, CreateSessionRequest, DaemonClient, DaemonInfo, DirListing,
GitRoot, RecentDirs, SendInputRequest, Session, SessionStatus, PROTOCOL_VERSION,
GitRoot, RecentDirs, SendInputRequest, Session, SessionStatus, StructuredPromptRequest,
PROTOCOL_VERSION,
};
use tokio::net::TcpListener;
use tokio::sync::{broadcast, oneshot};
@@ -60,6 +61,8 @@ struct Inner {
forced_error: Option<(u16, ApiError)>,
/// Everything `POST /api/sessions/:id/input` was asked to type, in order.
inputs: Vec<(Uuid, SendInputRequest)>,
/// Every ACP prompt sent to a structured session, in order.
structured_prompts: Vec<(Uuid, StructuredPromptRequest)>,
}
#[derive(Clone)]
@@ -92,6 +95,10 @@ impl MockDaemon {
.route(routes::SESSIONS, get(list_sessions).post(create_session))
.route(routes::patterns::SESSION, delete(kill_session))
.route(routes::patterns::SESSION_INPUT, post(send_input))
.route(
routes::patterns::STRUCTURED_PROMPT,
post(structured_prompt),
)
.route(routes::INFO, get(daemon_info))
.route(routes::DIRS, get(list_dir))
.route(routes::RECENT_DIRS, get(recent_dirs))
@@ -178,6 +185,16 @@ impl MockDaemon {
self.state.inner.lock().expect("poisoned").inputs.clone()
}
/// Every ACP prompt this daemon received, oldest first.
pub fn structured_prompts(&self) -> Vec<(Uuid, StructuredPromptRequest)> {
self.state
.inner
.lock()
.expect("poisoned")
.structured_prompts
.clone()
}
pub fn session_count(&self) -> usize {
self.state.inner.lock().expect("poisoned").sessions.len()
}
@@ -266,6 +283,26 @@ async fn send_input(
StatusCode::NO_CONTENT.into_response()
}
async fn structured_prompt(
State(s): State<AppState>,
Path(id): Path<Uuid>,
Json(req): Json<StructuredPromptRequest>,
) -> Response {
if let Some(r) = forced(&s) {
return r;
}
let mut guard = s.inner.lock().expect("poisoned");
if !guard.sessions.iter().any(|x| x.id == id) {
return (
StatusCode::NOT_FOUND,
Json(ApiError::new("not_found", "no such session on this daemon")),
)
.into_response();
}
guard.structured_prompts.push((id, req));
StatusCode::NO_CONTENT.into_response()
}
async fn daemon_info(State(s): State<AppState>) -> Response {
if let Some(r) = forced(&s) {
return r;
+3 -2
View File
@@ -82,7 +82,7 @@ impl Surface {
pub fn activity(&self, state: ActivityState) -> Color {
match state {
ActivityState::Active => self.active,
ActivityState::Idle => self.idle,
ActivityState::Idle | ActivityState::Starting => self.idle,
ActivityState::AwaitingInput => self.awaiting,
ActivityState::Error => self.error,
}
@@ -928,9 +928,10 @@ pub fn contrast_ratio(a: Color, b: Color) -> Option<f64> {
mod tests {
use super::*;
const STATES: [ActivityState; 4] = [
const STATES: [ActivityState; 5] = [
ActivityState::Active,
ActivityState::Idle,
ActivityState::Starting,
ActivityState::AwaitingInput,
ActivityState::Error,
];
+2
View File
@@ -979,6 +979,7 @@ mod tests {
for state in [
ActivityState::Active,
ActivityState::Idle,
ActivityState::Starting,
ActivityState::AwaitingInput,
ActivityState::Error,
] {
@@ -1009,6 +1010,7 @@ mod tests {
for state in [
ActivityState::Active,
ActivityState::Idle,
ActivityState::Starting,
ActivityState::AwaitingInput,
ActivityState::Error,
] {
+1
View File
@@ -160,6 +160,7 @@ The state word is never spelled out — the colour carries it.
| ---------------- | --------- | ---------------- |
| `Active` | `--live` | `active` |
| `Idle` | `--muted` | `idle` |
| `Starting` | `--muted` | `idle` |
| `AwaitingInput` | `--warn` | `awaiting` |
| `Error` | `--danger`| `error` |
+1 -1
View File
@@ -11,7 +11,7 @@ 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
* when an agent reports failure (#105); the daemon never infers it. */
export type ActivityState = "Active" | "Idle" | "AwaitingInput" | "Error";
export type ActivityState = "Active" | "Idle" | "AwaitingInput" | "Error" | "Starting";
export interface Session {
id: Uuid;