naps62/maestro
Archived
1
0
Fork 0
Code Issues 9 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(maestro): derive structured activity

Browse Source
This commit is contained in:
Miguel Palhas
2026-09-05 04:08:43 +01:00
parent 3e98b3ab24
commit 8f5d96a95b
9 changed files with 373 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/maestro-daemon/src/session.rs
+111 -19
View File
@@ -535,6 +535,8 @@ impl SessionManager {
return Ok(false);
};
let _ = stream.projected_live.send(projected);
drop(stream);
sync_structured_activity(&self.inner, id).await;
Ok(true)
}
@@ -558,6 +560,8 @@ impl SessionManager {
}
let (turn, update) = stream.projection.begin_turn();
let _ = stream.projected_live.send(update);
drop(stream);
sync_structured_activity(&self.inner, id).await;
Ok(BeginAcpTurn::Started(acp, turn))
}
@@ -614,6 +618,8 @@ impl SessionManager {
};
manager.finish_queued_acp_turn(id, turn, lifecycle).await;
});
drop(stream);
sync_structured_activity(&self.inner, id).await;
return Ok(QueueAcpPrompt::Queued);
}
Err(message) => {
@@ -625,6 +631,8 @@ impl SessionManager {
"newly claimed turn remains current while stream is locked",
);
let _ = stream.projected_live.send(projected);
drop(stream);
sync_structured_activity(&self.inner, id).await;
return Ok(QueueAcpPrompt::Failed(message));
}
}
@@ -669,6 +677,8 @@ impl SessionManager {
if let Some(projected) = stream.projection.turn_lifecycle(lifecycle, turn) {
let _ = stream.projected_live.send(projected);
}
drop(stream);
sync_structured_activity(&self.inner, id).await;
}
async fn fail_cancelled_acp_turn(
@@ -741,6 +751,7 @@ impl SessionManager {
.expect("active turn remains current while stream is locked");
let _ = stream.projected_live.send(projected);
drop(stream);
sync_structured_activity(&self.inner, id).await;
if let Some(turn) = cancelled_queued_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) {
let _ = stream.projected_live.send(update);
}
drop(stream);
sync_structured_activity(&self.inner, id).await;
Ok(Some(()))
}
@@ -1133,13 +1146,13 @@ impl SessionManager {
SessionStatus::Running
},
created_at: Utc::now(),
// A freshly-spawned session is producing its initial output; the
// activity watcher refines `foreground`/`activity` on its next tick.
// A terminal spawn begins active until its watcher sees otherwise;
// a structured spawn has no turn yet and begins idle.
foreground: None,
activity: if acp_failure.is_some() {
ActivityState::Error
} else {
ActivityState::Active
activity: match (req.view, acp_failure.is_some()) {
(_, true) => ActivityState::Error,
(SessionView::Structured, false) => ActivityState::Idle,
(SessionView::Terminal, false) => ActivityState::Active,
},
activity_detail: acp_failure,
last_activity_at: Utc::now(),
@@ -1201,7 +1214,7 @@ impl SessionManager {
}
if let Some((acp, events)) = event_pump {
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) {
record.acp_event_pump = Some(done);
}
@@ -1331,6 +1344,7 @@ impl SessionManager {
return;
};
run_liveness_pass(&inner).await;
run_structured_activity_pass(&inner).await;
// Same tick, no new timer (NAP-340): refresh activity for the
// survivors and emit SessionUpdated on any transition.
run_activity_pass(&inner).await;
@@ -1499,11 +1513,15 @@ impl SessionManager {
SessionStatus::Dead
};
if recoverable {
session.activity = if failure.is_some() {
let activity = if failure.is_some() {
ActivityState::Error
} else {
ActivityState::Active
ActivityState::Idle
};
if session.activity != activity {
session.activity_since = Utc::now();
}
session.activity = activity;
session.activity_detail = failure;
}
@@ -1590,7 +1608,7 @@ impl SessionManager {
);
if let Some((acp, events)) = event_pump {
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) {
record.acp_event_pump = Some(done);
}
@@ -1750,12 +1768,10 @@ impl SessionManager {
Ok(session)
}
/// Accept an agent-reported activity state (#105), the only path that can
/// produce `ActivityState::Error`. The report outranks the activity pass's
/// own inference until its TTL lapses — so an agent that dies mid-run
/// cannot pin its session at `Error` forever. An unknown session is
/// [`SessionError::NotFound`]; the reported `detail` is capped
/// server-side.
/// Accept an agent-reported activity state (#105) for a terminal session.
/// Structured sessions derive their state from ACP instead, so their
/// inherited hooks are acknowledged but ignored. An unknown session is
/// [`SessionError::NotFound`]; the reported `detail` is capped server-side.
///
/// A report that changes the effective state restarts `activity_since`
/// (#104's clock); re-reporting an unchanged state only extends the TTL
@@ -1769,6 +1785,9 @@ impl SessionManager {
let updated = {
let mut sessions = self.inner.sessions.lock().await;
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 changed = rec.session.activity != state || rec.session.activity_detail != detail;
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 {
return;
};
let weak_inner = Arc::downgrade(inner);
tokio::spawn(async move {
while let Some(PermissionRequest {
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) {
tracing::error!(%error, "approval history append failed");
cancel_pending_approvals(&mut events);
drop(events);
if let Some(inner) = weak_inner.upgrade() {
sync_structured_activity(&inner, id).await;
}
continue;
}
let seq = events.log.last_seq();
let _ = events.live.send(LoggedEvent { seq, event });
}
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
/// steady state produces no diffs. Runs on the existing liveness tick.
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.
sessions
.iter()
.filter(|(_, r)| r.session.status == SessionStatus::Running)
.filter(|(_, r)| {
r.session.status == SessionStatus::Running
&& r.session.view == SessionView::Terminal
})
.map(|(id, _)| *id)
.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
/// worktree holds right now that would block an unforced kill. Rides the
/// 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) {
Some(record) if record.session.status == SessionStatus::Running => {
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
// attach.
record.bridge.seal();