Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
8f5d96a95b | ||
|
|
3e98b3ab24 |
@@ -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);
|
||||||
|
|||||||
@@ -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>,
|
||||||
|
|||||||
@@ -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();
|
||||||
|
|||||||
@@ -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();
|
||||||
|
|||||||
@@ -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()
|
||||||
|
|||||||
@@ -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]
|
||||||
|
|||||||
@@ -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
@@ -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
|
||||||
|
|||||||
@@ -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.
|
||||||
|
|||||||
Reference in New Issue
Block a user