Compare commits

..
9 changed files with 373 additions and 39 deletions
+28
View File
@@ -66,6 +66,23 @@ pub fn decide(idle_for: Duration, awaiting_input: bool) -> ActivityState {
} }
} }
/// Derive a structured session's sidebar state from its ACP projection.
pub fn decide_structured(
turn_active: bool,
pending_approval: bool,
turn_failed: bool,
) -> ActivityState {
if turn_failed {
ActivityState::Error
} else if pending_approval {
ActivityState::AwaitingInput
} else if turn_active {
ActivityState::Active
} else {
ActivityState::Idle
}
}
/// How long a session's pane must be continuously focused (a client's pane, /// How long a session's pane must be continuously focused (a client's pane,
/// not a sidebar highlight) before an `AwaitingInput` glow clears: the /// not a sidebar highlight) before an `AwaitingInput` glow clears: the
/// operator has looked, so the rail stops asking for attention. /// operator has looked, so the rail stops asking for attention.
@@ -136,6 +153,17 @@ mod tests {
); );
} }
#[test]
fn structured_activity_follows_lifecycle_with_approval_precedence() {
assert_eq!(decide_structured(false, false, false), ActivityState::Idle);
assert_eq!(decide_structured(true, false, false), ActivityState::Active);
assert_eq!(
decide_structured(true, true, false),
ActivityState::AwaitingInput
);
assert_eq!(decide_structured(false, false, true), ActivityState::Error);
}
#[test] #[test]
fn report_is_in_force_until_exactly_its_expiry() { fn report_is_in_force_until_exactly_its_expiry() {
let ttl = Duration::from_secs(300); let ttl = Duration::from_secs(300);
+4 -5
View File
@@ -307,11 +307,10 @@ async fn rename_session(
} }
} }
/// `POST /api/sessions/:id/activity` (#105): an agent reports its own state. /// `POST /api/sessions/:id/activity` (#105): an agent reports a terminal
/// The report holds above the daemon's inference for a TTL, then inference /// session's state. Structured sessions acknowledge and ignore these reports
/// resumes — an agent that dies mid-run cannot pin its session at `Error` /// because their activity comes from ACP. Unknown session is 404; an unknown
/// forever. Unknown session is 404; an unknown state value is rejected by the /// state value is rejected by the JSON decoder (422). Answers `204 No Content`.
/// JSON decoder (422) rather than coerced. Answers `204 No Content`.
async fn report_activity( async fn report_activity(
State(s): State<LocalApiState>, State(s): State<LocalApiState>,
Path(id): Path<Uuid>, Path(id): Path<Uuid>,
+14 -1
View File
@@ -4,7 +4,8 @@ use std::collections::{HashMap, HashSet};
use maestro_acp::schema::{ContentChunk, SessionNotification, SessionUpdate, ToolCall}; use maestro_acp::schema::{ContentChunk, SessionNotification, SessionUpdate, ToolCall};
use maestro_protocol::{ use maestro_protocol::{
ReducedState, TranscriptDelta, TranscriptRow, TranscriptRowKind, TranscriptSnapshot, ActivityState, ReducedState, TranscriptDelta, TranscriptRow, TranscriptRowKind,
TranscriptSnapshot,
}; };
use serde_json::Value; use serde_json::Value;
@@ -36,6 +37,7 @@ pub struct Projection {
pending_tool_updates: HashMap<String, Vec<maestro_acp::schema::ToolCallUpdateFields>>, pending_tool_updates: HashMap<String, Vec<maestro_acp::schema::ToolCallUpdateFields>>,
state: ReducedState, state: ReducedState,
active_turn: Option<u64>, active_turn: Option<u64>,
turn_failed: bool,
next_turn: u64, next_turn: u64,
turn_tool_calls: HashSet<String>, turn_tool_calls: HashSet<String>,
} }
@@ -109,11 +111,21 @@ impl Projection {
self.active_turn.is_some() self.active_turn.is_some()
} }
/// Sidebar activity derived from the live turn projection.
pub fn activity(&self) -> ActivityState {
crate::activity::decide_structured(
self.active_turn.is_some(),
self.state.pending_approval.is_some(),
self.turn_failed,
)
}
/// Start a prompt turn and return its monotonically increasing token. /// Start a prompt turn and return its monotonically increasing token.
pub fn begin_turn(&mut self) -> (u64, ProjectionUpdate) { pub fn begin_turn(&mut self) -> (u64, ProjectionUpdate) {
self.next_turn += 1; self.next_turn += 1;
let turn = self.next_turn; let turn = self.next_turn;
self.active_turn = Some(turn); self.active_turn = Some(turn);
self.turn_failed = false;
self.state.turn.responding = true; self.state.turn.responding = true;
self.state.turn.thinking = true; self.state.turn.thinking = true;
self.turn_tool_calls.clear(); self.turn_tool_calls.clear();
@@ -141,6 +153,7 @@ impl Projection {
return None; return None;
} }
self.active_turn = None; self.active_turn = None;
self.turn_failed = lifecycle == TurnLifecycle::Failure;
self.state.turn.responding = false; self.state.turn.responding = false;
self.state.turn.thinking = false; self.state.turn.thinking = false;
self.turn_tool_calls.clear(); self.turn_tool_calls.clear();
+111 -19
View File
@@ -535,6 +535,8 @@ impl SessionManager {
return Ok(false); return Ok(false);
}; };
let _ = stream.projected_live.send(projected); let _ = stream.projected_live.send(projected);
drop(stream);
sync_structured_activity(&self.inner, id).await;
Ok(true) Ok(true)
} }
@@ -558,6 +560,8 @@ impl SessionManager {
} }
let (turn, update) = stream.projection.begin_turn(); let (turn, update) = stream.projection.begin_turn();
let _ = stream.projected_live.send(update); let _ = stream.projected_live.send(update);
drop(stream);
sync_structured_activity(&self.inner, id).await;
Ok(BeginAcpTurn::Started(acp, turn)) Ok(BeginAcpTurn::Started(acp, turn))
} }
@@ -614,6 +618,8 @@ impl SessionManager {
}; };
manager.finish_queued_acp_turn(id, turn, lifecycle).await; manager.finish_queued_acp_turn(id, turn, lifecycle).await;
}); });
drop(stream);
sync_structured_activity(&self.inner, id).await;
return Ok(QueueAcpPrompt::Queued); return Ok(QueueAcpPrompt::Queued);
} }
Err(message) => { Err(message) => {
@@ -625,6 +631,8 @@ impl SessionManager {
"newly claimed turn remains current while stream is locked", "newly claimed turn remains current while stream is locked",
); );
let _ = stream.projected_live.send(projected); let _ = stream.projected_live.send(projected);
drop(stream);
sync_structured_activity(&self.inner, id).await;
return Ok(QueueAcpPrompt::Failed(message)); return Ok(QueueAcpPrompt::Failed(message));
} }
} }
@@ -669,6 +677,8 @@ impl SessionManager {
if let Some(projected) = stream.projection.turn_lifecycle(lifecycle, turn) { if let Some(projected) = stream.projection.turn_lifecycle(lifecycle, turn) {
let _ = stream.projected_live.send(projected); let _ = stream.projected_live.send(projected);
} }
drop(stream);
sync_structured_activity(&self.inner, id).await;
} }
async fn fail_cancelled_acp_turn( async fn fail_cancelled_acp_turn(
@@ -741,6 +751,7 @@ impl SessionManager {
.expect("active turn remains current while stream is locked"); .expect("active turn remains current while stream is locked");
let _ = stream.projected_live.send(projected); let _ = stream.projected_live.send(projected);
drop(stream); drop(stream);
sync_structured_activity(&self.inner, id).await;
if let Some(turn) = cancelled_queued_turn { if let Some(turn) = cancelled_queued_turn {
self.schedule_cancelled_acp_turn_failure(id, event_stream, turn); self.schedule_cancelled_acp_turn_failure(id, event_stream, turn);
} }
@@ -789,6 +800,8 @@ impl SessionManager {
if let Some(update) = stream.projection.resolve_approval(nonce) { if let Some(update) = stream.projection.resolve_approval(nonce) {
let _ = stream.projected_live.send(update); let _ = stream.projected_live.send(update);
} }
drop(stream);
sync_structured_activity(&self.inner, id).await;
Ok(Some(())) Ok(Some(()))
} }
@@ -1133,13 +1146,13 @@ impl SessionManager {
SessionStatus::Running SessionStatus::Running
}, },
created_at: Utc::now(), created_at: Utc::now(),
// A freshly-spawned session is producing its initial output; the // A terminal spawn begins active until its watcher sees otherwise;
// activity watcher refines `foreground`/`activity` on its next tick. // a structured spawn has no turn yet and begins idle.
foreground: None, foreground: None,
activity: if acp_failure.is_some() { activity: match (req.view, acp_failure.is_some()) {
ActivityState::Error (_, true) => ActivityState::Error,
} else { (SessionView::Structured, false) => ActivityState::Idle,
ActivityState::Active (SessionView::Terminal, false) => ActivityState::Active,
}, },
activity_detail: acp_failure, activity_detail: acp_failure,
last_activity_at: Utc::now(), last_activity_at: Utc::now(),
@@ -1201,7 +1214,7 @@ impl SessionManager {
} }
if let Some((acp, events)) = event_pump { if let Some((acp, events)) = event_pump {
let done = start_acp_event_pump(&self.inner, id, &acp, events.clone()); let done = start_acp_event_pump(&self.inner, id, &acp, events.clone());
start_acp_permission_pump(&acp, events); start_acp_permission_pump(&self.inner, id, &acp, events);
if let Some(record) = self.inner.sessions.lock().await.get_mut(&id) { if let Some(record) = self.inner.sessions.lock().await.get_mut(&id) {
record.acp_event_pump = Some(done); record.acp_event_pump = Some(done);
} }
@@ -1331,6 +1344,7 @@ impl SessionManager {
return; return;
}; };
run_liveness_pass(&inner).await; run_liveness_pass(&inner).await;
run_structured_activity_pass(&inner).await;
// Same tick, no new timer (NAP-340): refresh activity for the // Same tick, no new timer (NAP-340): refresh activity for the
// survivors and emit SessionUpdated on any transition. // survivors and emit SessionUpdated on any transition.
run_activity_pass(&inner).await; run_activity_pass(&inner).await;
@@ -1499,11 +1513,15 @@ impl SessionManager {
SessionStatus::Dead SessionStatus::Dead
}; };
if recoverable { if recoverable {
session.activity = if failure.is_some() { let activity = if failure.is_some() {
ActivityState::Error ActivityState::Error
} else { } else {
ActivityState::Active ActivityState::Idle
}; };
if session.activity != activity {
session.activity_since = Utc::now();
}
session.activity = activity;
session.activity_detail = failure; session.activity_detail = failure;
} }
@@ -1590,7 +1608,7 @@ impl SessionManager {
); );
if let Some((acp, events)) = event_pump { if let Some((acp, events)) = event_pump {
let done = start_acp_event_pump(&self.inner, id, &acp, events.clone()); let done = start_acp_event_pump(&self.inner, id, &acp, events.clone());
start_acp_permission_pump(&acp, events); start_acp_permission_pump(&self.inner, id, &acp, events);
if let Some(record) = self.inner.sessions.lock().await.get_mut(&id) { if let Some(record) = self.inner.sessions.lock().await.get_mut(&id) {
record.acp_event_pump = Some(done); record.acp_event_pump = Some(done);
} }
@@ -1750,12 +1768,10 @@ impl SessionManager {
Ok(session) Ok(session)
} }
/// Accept an agent-reported activity state (#105), the only path that can /// Accept an agent-reported activity state (#105) for a terminal session.
/// produce `ActivityState::Error`. The report outranks the activity pass's /// Structured sessions derive their state from ACP instead, so their
/// own inference until its TTL lapses — so an agent that dies mid-run /// inherited hooks are acknowledged but ignored. An unknown session is
/// cannot pin its session at `Error` forever. An unknown session is /// [`SessionError::NotFound`]; the reported `detail` is capped server-side.
/// [`SessionError::NotFound`]; the reported `detail` is capped
/// server-side.
/// ///
/// A report that changes the effective state restarts `activity_since` /// A report that changes the effective state restarts `activity_since`
/// (#104's clock); re-reporting an unchanged state only extends the TTL /// (#104's clock); re-reporting an unchanged state only extends the TTL
@@ -1769,6 +1785,9 @@ impl SessionManager {
let updated = { let updated = {
let mut sessions = self.inner.sessions.lock().await; let mut sessions = self.inner.sessions.lock().await;
let rec = sessions.get_mut(&id).ok_or(SessionError::NotFound(id))?; let rec = sessions.get_mut(&id).ok_or(SessionError::NotFound(id))?;
if rec.session.view == SessionView::Structured {
return Ok(());
}
let detail = crate::activity::truncate_detail(detail); let detail = crate::activity::truncate_detail(detail);
let changed = rec.session.activity != state || rec.session.activity_detail != detail; let changed = rec.session.activity != state || rec.session.activity_detail != detail;
if rec.session.activity != state { if rec.session.activity != state {
@@ -1983,10 +2002,51 @@ fn apply_persisted_event(projection: &mut Projection, event: &LoggedEvent) -> Pr
} }
} }
fn start_acp_permission_pump(acp: &Arc<AcpSession>, events: Arc<Mutex<AcpEventStream>>) { async fn sync_structured_activity(inner: &Arc<Inner>, id: Uuid) {
let updated = {
let mut sessions = inner.sessions.lock().await;
let Some(record) = sessions.get_mut(&id) else {
return;
};
if record.session.view != SessionView::Structured {
return;
}
let raw = if record.session.status == SessionStatus::Dead {
ActivityState::Error
} else if let Some(events) = &record.acp_events {
events.lock().await.projection.activity()
} else {
ActivityState::Error
};
let activity = crate::activity::apply_focus_suppression(
raw,
record.focused_since,
std::time::Instant::now(),
&mut record.awaiting_suppressed,
);
if record.session.activity == activity {
None
} else {
record.session.activity = activity;
record.session.activity_since = Utc::now();
Some(record.session.clone())
}
};
if let Some(session) = updated {
let _ = inner.events.send(CatalogEvent::SessionUpdated { session });
}
}
fn start_acp_permission_pump(
inner: &Arc<Inner>,
id: Uuid,
acp: &Arc<AcpSession>,
events: Arc<Mutex<AcpEventStream>>,
) {
let Some(mut requests) = acp.take_permission_requests() else { let Some(mut requests) = acp.take_permission_requests() else {
return; return;
}; };
let weak_inner = Arc::downgrade(inner);
tokio::spawn(async move { tokio::spawn(async move {
while let Some(PermissionRequest { while let Some(PermissionRequest {
turn, turn,
@@ -2023,12 +2083,20 @@ fn start_acp_permission_pump(acp: &Arc<AcpSession>, events: Arc<Mutex<AcpEventSt
if let Err(error) = events.log.append(&event) { if let Err(error) = events.log.append(&event) {
tracing::error!(%error, "approval history append failed"); tracing::error!(%error, "approval history append failed");
cancel_pending_approvals(&mut events); cancel_pending_approvals(&mut events);
drop(events);
if let Some(inner) = weak_inner.upgrade() {
sync_structured_activity(&inner, id).await;
}
continue; continue;
} }
let seq = events.log.last_seq(); let seq = events.log.last_seq();
let _ = events.live.send(LoggedEvent { seq, event }); let _ = events.live.send(LoggedEvent { seq, event });
} }
let _ = events.projected_live.send(update); let _ = events.projected_live.send(update);
drop(events);
if let Some(inner) = weak_inner.upgrade() {
sync_structured_activity(&inner, id).await;
}
} }
}); });
} }
@@ -2176,7 +2244,7 @@ async fn run_liveness_pass(inner: &Arc<Inner>) {
} }
} }
/// Refresh each live session's inferred activity (NAP-340) and emit /// Refresh each live terminal session's inferred activity (NAP-340) and emit
/// `SessionUpdated` only when `activity` or `foreground` actually changes — so a /// `SessionUpdated` only when `activity` or `foreground` actually changes — so a
/// steady state produces no diffs. Runs on the existing liveness tick. /// steady state produces no diffs. Runs on the existing liveness tick.
async fn run_activity_pass(inner: &Arc<Inner>) { async fn run_activity_pass(inner: &Arc<Inner>) {
@@ -2192,7 +2260,10 @@ async fn run_activity_pass(inner: &Arc<Inner>) {
// kept as-is so the UI can still show what it was doing when it died. // kept as-is so the UI can still show what it was doing when it died.
sessions sessions
.iter() .iter()
.filter(|(_, r)| r.session.status == SessionStatus::Running) .filter(|(_, r)| {
r.session.status == SessionStatus::Running
&& r.session.view == SessionView::Terminal
})
.map(|(id, _)| *id) .map(|(id, _)| *id)
.collect() .collect()
}; };
@@ -2290,6 +2361,21 @@ async fn run_activity_pass(inner: &Arc<Inner>) {
} }
} }
/// Reapply focus suppression to structured activity on the liveness cadence.
async fn run_structured_activity_pass(inner: &Arc<Inner>) {
let ids: Vec<Uuid> = inner
.sessions
.lock()
.await
.iter()
.filter(|(_, record)| record.session.view == SessionView::Structured)
.map(|(id, _)| *id)
.collect();
for id in ids {
sync_structured_activity(inner, id).await;
}
}
/// Refresh each running session's blocked-work indicator (#83): what its /// Refresh each running session's blocked-work indicator (#83): what its
/// worktree holds right now that would block an unforced kill. Rides the /// worktree holds right now that would block an unforced kill. Rides the
/// liveness tick on its own (slower) cadence — see /// liveness tick on its own (slower) cadence — see
@@ -2391,6 +2477,12 @@ async fn mark_session_dead_with_worktree_settlement(
match sessions.get_mut(&id) { match sessions.get_mut(&id) {
Some(record) if record.session.status == SessionStatus::Running => { Some(record) if record.session.status == SessionStatus::Running => {
record.session.status = SessionStatus::Dead; record.session.status = SessionStatus::Dead;
if record.session.view == SessionView::Structured
&& record.session.activity != ActivityState::Error
{
record.session.activity = ActivityState::Error;
record.session.activity_since = Utc::now();
}
// Release anyone attached; the scrollback survives for a later // Release anyone attached; the scrollback survives for a later
// attach. // attach.
record.bridge.seal(); record.bridge.seal();
+92
View File
@@ -136,6 +136,7 @@ async fn structured_spawn_starts_the_adapter_in_cwd_with_session_env() {
assert_eq!(session.view, SessionView::Structured); assert_eq!(session.view, SessionView::Structured);
assert_eq!(session.status, SessionStatus::Running); assert_eq!(session.status, SessionStatus::Running);
assert_eq!(session.activity, ActivityState::Idle);
let acp_id = session.acp_session_id.clone().expect("acp session id"); let acp_id = session.acp_session_id.clone().expect("acp session id");
assert!(acp_id.starts_with("fake-"), "{acp_id}"); assert!(acp_id.starts_with("fake-"), "{acp_id}");
@@ -204,6 +205,77 @@ async fn structured_spawn_starts_the_adapter_in_cwd_with_session_env() {
.exists()); .exists());
} }
#[tokio::test]
async fn structured_activity_tracks_turn_completion_and_failure() {
let fx = DaemonFixture::new();
let fake = Fake::new();
let cwd = TempDir::new().unwrap();
let mut env = fake.env();
env.insert("FAKE_ACP_DELAY_PROMPT_MS".into(), "100".into());
env.insert("FAKE_ACP_PROMPT_FAIL_ONCE".into(), "1".into());
let session = fx
.manager
.create(structured(cwd.path(), env))
.await
.unwrap();
assert!(matches!(
fx.manager
.queue_acp_prompt(session.id, "fail".into())
.await
.unwrap(),
maestro_daemon::session::QueueAcpPrompt::Queued
));
assert_eq!(
fx.manager
.snapshot()
.await
.into_iter()
.find(|candidate| candidate.id == session.id)
.unwrap()
.activity,
ActivityState::Active
);
wait_for_async(Duration::from_secs(2), || async {
fx.manager
.snapshot()
.await
.into_iter()
.find(|candidate| candidate.id == session.id)
.is_some_and(|candidate| candidate.activity == ActivityState::Error)
})
.await;
assert!(matches!(
fx.manager
.queue_acp_prompt(session.id, "succeed".into())
.await
.unwrap(),
maestro_daemon::session::QueueAcpPrompt::Queued
));
assert_eq!(
fx.manager
.snapshot()
.await
.into_iter()
.find(|candidate| candidate.id == session.id)
.unwrap()
.activity,
ActivityState::Active
);
wait_for_async(Duration::from_secs(2), || async {
fx.manager
.snapshot()
.await
.into_iter()
.find(|candidate| candidate.id == session.id)
.is_some_and(|candidate| candidate.activity == ActivityState::Idle)
})
.await;
fx.manager.kill(session.id, false).await.unwrap();
}
#[tokio::test] #[tokio::test]
async fn projection_snapshot_and_live_stream_share_one_fold() { async fn projection_snapshot_and_live_stream_share_one_fold() {
let fx = DaemonFixture::new(); let fx = DaemonFixture::new();
@@ -476,6 +548,16 @@ async fn recovery_loads_the_acp_session_without_duplicating_history() {
revived.status(session.id).await, revived.status(session.id).await,
Some(SessionStatus::Running) Some(SessionStatus::Running)
); );
assert_eq!(
revived
.snapshot()
.await
.into_iter()
.find(|candidate| candidate.id == session.id)
.unwrap()
.activity,
ActivityState::Idle
);
let log = fake.log(); let log = fake.log();
assert!( assert!(
log.contains(&format!("recv session/load sessionId={acp_id}")), log.contains(&format!("recv session/load sessionId={acp_id}")),
@@ -925,6 +1007,16 @@ async fn adapter_exit_marks_the_session_dead_and_keeps_it_listed() {
fx.manager.status(session.id).await == Some(SessionStatus::Dead) fx.manager.status(session.id).await == Some(SessionStatus::Dead)
}) })
.await; .await;
assert_eq!(
fx.manager
.snapshot()
.await
.into_iter()
.find(|candidate| candidate.id == session.id)
.unwrap()
.activity,
ActivityState::Error
);
assert!(fx.manager.has(session.id).await); assert!(fx.manager.has(session.id).await);
assert!( assert!(
!fx.sidecar_dir() !fx.sidecar_dir()
+95 -1
View File
@@ -11,7 +11,8 @@ use common::DaemonFixture;
use futures_util::{SinkExt, StreamExt}; use futures_util::{SinkExt, StreamExt};
use maestro_daemon::local_api; use maestro_daemon::local_api;
use maestro_protocol::{ use maestro_protocol::{
ApiError, CreateSessionRequest, Session, SessionStatus, SessionView, WorktreeSpec, ActivityState, ApiError, CreateSessionRequest, Session, SessionStatus, SessionView,
WorktreeSpec,
}; };
use tempfile::TempDir; use tempfile::TempDir;
use tokio::net::TcpListener; use tokio::net::TcpListener;
@@ -167,6 +168,29 @@ async fn structured_session_with_env(h: &Harness, env: HashMap<String, String>)
.id .id
} }
async fn wait_for_session_activity(h: &Harness, id: Uuid, expected: ActivityState) {
tokio::time::timeout(Duration::from_secs(2), async {
loop {
let sessions: Vec<Session> = reqwest::get(url(h.addr, "/api/sessions"))
.await
.unwrap()
.json()
.await
.unwrap();
if sessions
.iter()
.find(|session| session.id == id)
.is_some_and(|session| session.activity == expected)
{
return;
}
tokio::time::sleep(Duration::from_millis(10)).await;
}
})
.await
.unwrap_or_else(|_| panic!("session {id} did not reach {expected:?}"));
}
#[tokio::test] #[tokio::test]
async fn structured_ws_sends_snapshot_then_live_projection() { async fn structured_ws_sends_snapshot_then_live_projection() {
let h = spawn("naps-laptop").await; let h = spawn("naps-laptop").await;
@@ -318,6 +342,7 @@ async fn structured_approval_is_secret_single_use_and_race_safe() {
}) })
.await .await
.unwrap(); .unwrap();
wait_for_session_activity(&h, id, ActivityState::AwaitingInput).await;
assert!( assert!(
!fake.log().contains(&nonce), !fake.log().contains(&nonce),
"daemon nonce leaked to ACP agent" "daemon nonce leaked to ACP agent"
@@ -373,6 +398,7 @@ async fn structured_approval_is_secret_single_use_and_race_safe() {
}) })
.await .await
.expect("resolved approval clears the reduced state"); .expect("resolved approval clears the reduced state");
wait_for_session_activity(&h, id, ActivityState::Idle).await;
assert_eq!( assert_eq!(
client client
.post(url( .post(url(
@@ -429,6 +455,7 @@ async fn structured_prompt_failure_clears_the_turn() {
}) })
.await .await
.expect("failed turn clears promptly"); .expect("failed turn clears promptly");
wait_for_session_activity(&h, id, ActivityState::Error).await;
let second = client let second = client
.post(url(h.addr, &path)) .post(url(h.addr, &path))
.json(&serde_json::json!({"prompt":"available"})) .json(&serde_json::json!({"prompt":"available"}))
@@ -1350,6 +1377,73 @@ async fn events_ws_sends_a_snapshot_then_streams_deltas() {
} }
} }
#[tokio::test]
async fn events_ws_streams_structured_activity_without_hook_overrides() {
let h = spawn("konishi").await;
let fake = FakeAcp::new();
let mut env = fake.env();
env.insert("FAKE_ACP_DELAY_PROMPT_MS".into(), "500".into());
let (mut ws, _) = connect_async(ws_url(h.addr, "/api/events"))
.await
.expect("events ws");
assert!(matches!(
next_event(&mut ws).await,
maestro_protocol::CatalogEvent::Snapshot { .. }
));
let id = structured_session_with_env(&h, env).await;
let maestro_protocol::CatalogEvent::SessionAdded { session } = next_event(&mut ws).await else {
panic!("expected structured SessionAdded")
};
assert_eq!(session.activity, ActivityState::Idle);
let client = reqwest::Client::new();
let prompt = client
.post(url(
h.addr,
&format!("/api/sessions/{id}/structured/prompt"),
))
.json(&serde_json::json!({"prompt": "hello"}))
.send()
.await
.unwrap();
assert_eq!(prompt.status(), reqwest::StatusCode::NO_CONTENT);
let report = client
.post(url(h.addr, &format!("/api/sessions/{id}/activity")))
.json(&serde_json::json!({"state": "Error"}))
.send()
.await
.unwrap();
assert_eq!(report.status(), reqwest::StatusCode::NO_CONTENT);
wait_for_session_activity(&h, id, ActivityState::Active).await;
let mut activities = Vec::new();
tokio::time::timeout(Duration::from_secs(2), async {
loop {
if let maestro_protocol::CatalogEvent::SessionUpdated { session } =
next_event(&mut ws).await
{
activities.push(session.activity);
if session.activity == ActivityState::Idle {
return;
}
}
}
})
.await
.expect("structured turn returns to idle");
assert!(
activities.contains(&ActivityState::Active),
"{activities:?}"
);
assert!(
!activities.contains(&ActivityState::Error),
"{activities:?}"
);
h.manager.kill(id, false).await.unwrap();
}
/// Each subscriber has its own lag domain now that the publisher task is gone: /// Each subscriber has its own lag domain now that the publisher task is gone:
/// two clients connected at different times both get their own Snapshot. /// two clients connected at different times both get their own Snapshot.
#[tokio::test] #[tokio::test]
+6 -7
View File
@@ -156,9 +156,9 @@ pub enum SessionStatus {
Dead, Dead,
} }
/// Inferred session activity, orthogonal to [`SessionStatus`] (NAP-340). The /// Session activity, orthogonal to [`SessionStatus`] (NAP-340). Terminal
/// daemon derives this from tmux pane state, PTY quiescence, and `/proc` /// sessions use pane/process inference and agent reports; structured sessions
/// inspection — never by parsing the terminal byte stream. /// use their ACP turn lifecycle (docs/SIDEBAR.md).
/// ///
/// Deserializes leniently (#110, SIDEBAR.md Precedence): an unknown variant /// Deserializes leniently (#110, SIDEBAR.md Precedence): an unknown variant
/// from a newer daemon becomes [`ActivityState::Idle`] instead of failing the /// from a newer daemon becomes [`ActivityState::Idle`] instead of failing the
@@ -166,15 +166,14 @@ pub enum SessionStatus {
/// forgiven. /// forgiven.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Default)] #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Default)]
pub enum ActivityState { pub enum ActivityState {
/// Producing output, or the foreground process is busy. /// Producing output, the foreground process is busy, or an ACP turn is active.
Active, Active,
/// Foreground process alive but quiescent. The forward-compatible default. /// Foreground process alive but quiescent. The forward-compatible default.
#[default] #[default]
Idle, Idle,
/// Blocked on a `read()` from the pty — waiting on the user. /// Blocked on terminal input or waiting for an ACP approval.
AwaitingInput, AwaitingInput,
/// The agent running in the session reported a failure (#102). Never /// An agent reported a failure, or a structured session/turn failed.
/// inferred by the daemon; only set when an agent reports it (#105).
Error, Error,
} }
+22 -5
View File
@@ -231,7 +231,24 @@ cutover points land within one tick.
## Precedence ## Precedence
An agent-reported `ActivityState` (#105) overrides the daemon's inference for ### Activity source by session kind
The daemon supplies one `activity` field to both clients, but its source depends
on the session kind:
| Session kind | Source |
|---|---|
| Terminal | The existing agent-report override and five-minute TTL, falling back to PTY quiescence, foreground process and `/proc` input-block inference. |
| Structured | ACP state directly: a pending approval is `AwaitingInput`; otherwise an in-flight turn is `Active`; no turn is `Idle`. A failed turn or dead structured session is `Error`, and the next turn clears a failed-turn error. |
Structured sessions never enter the terminal inference path and ignore activity
reports inherited through agent hooks. They have no PTY whose output or blocked
read could describe their state. This is a second source, not a replacement for
the terminal source.
### Terminal report override
For terminal sessions, an agent-reported `ActivityState` (#105) overrides the daemon's inference for
**five minutes** (the TTL is fixed by the daemon; re-reporting extends it). **five minutes** (the TTL is fixed by the daemon; re-reporting extends it).
When the TTL expires, inference resumes and the session's reported detail — When the TTL expires, inference resumes and the session's reported detail —
if any — clears with the report. Clients never blend the two: whichever source if any — clears with the report. Clients never blend the two: whichever source
@@ -281,10 +298,10 @@ session is genuinely `Active` again — new work, not merely a refocus. Only
after that fresh `Active` cycle does the *next* `AwaitingInput` glow again. after that fresh `Active` cycle does the *next* `AwaitingInput` glow again.
Unfocusing and refocusing alone never re-arms it. Unfocusing and refocusing alone never re-arms it.
This applies uniformly to `AwaitingInput` from either source — the daemon's This applies uniformly to `AwaitingInput` from terminal inference, an agent
own inference or an agent's report (#105) — since the rail draws them report (#105), or a structured session's pending approval, since the rail draws
identically. It does not touch `Error`; that state's own TTL (below) is them identically. It does not touch `Error`; terminal reports keep their TTL,
unrelated and unaffected by focus. while a structured failed-turn error clears when the next turn starts.
Each client reports its own pane's focus with `POST /api/sessions/:id/focus` Each client reports its own pane's focus with `POST /api/sessions/:id/focus`
(`{"focused": bool}`) whenever the pane it is showing changes — on attach, on (`{"focused": bool}`) whenever the pane it is showing changes — on attach, on
+1 -1
View File
@@ -343,7 +343,7 @@ never reached disk), and only then reopens the log for further appends. What
the adapter streams back while its own `session/load` is in flight is the adapter streams back while its own `session/load` is in flight is
discarded rather than appended, so the replayed history is not duplicated discarded rather than appended, so the replayed history is not duplicated
(see [The ACP handshake](#the-acp-handshake)). Success flips the session back (see [The ACP handshake](#the-acp-handshake)). Success flips the session back
to `Running`/`Active`; failure (no stored `acp_session_id`, or the handshake to `Running`/`Idle`; failure (no stored `acp_session_id`, or the handshake
itself failing) leaves it `Dead` with the failure reason as itself failing) leaves it `Dead` with the failure reason as
`activity_detail` — the same "tell the operator what to fix" pattern as a `activity_detail` — the same "tell the operator what to fix" pattern as a
missing adapter binary. missing adapter binary.