fix(maestro): mark graceful ACP recovery
This commit is contained in:
@@ -930,9 +930,9 @@ impl SessionManager {
|
||||
adopted += 1;
|
||||
}
|
||||
|
||||
// Structured sessions have no tmux process to discover. Their
|
||||
// sidecars are the recovery inventory: when an ACP id exists, start a
|
||||
// fresh adapter process and load that conversation into it.
|
||||
// Structured sessions have no tmux process to discover. A durable
|
||||
// marker written during graceful shutdown is the only evidence that a
|
||||
// sidecar may be resumed; older and naturally-dead sidecars stay dead.
|
||||
for mut session in self
|
||||
.inner
|
||||
.sidecar
|
||||
@@ -945,6 +945,7 @@ impl SessionManager {
|
||||
}
|
||||
|
||||
let id = session.id;
|
||||
let recoverable = self.inner.sidecar.take_recovery_marker(id)?;
|
||||
let mut session_env = sanitize_env(&session.env)?;
|
||||
let fifo_path = self.inner.fifo_dir.join(format!("{id}.fifo"));
|
||||
let bridge =
|
||||
@@ -960,40 +961,50 @@ impl SessionManager {
|
||||
session_env.insert("MAESTRO_URL".into(), url);
|
||||
}
|
||||
|
||||
let (acp, failure) = match session.acp_session_id.clone() {
|
||||
Some(acp_session_id) => {
|
||||
let config = SpawnConfig::new(
|
||||
&session.cwd,
|
||||
session_env
|
||||
.iter()
|
||||
.map(|(key, value)| (key.clone(), value.clone()))
|
||||
.collect(),
|
||||
)
|
||||
.load_session(acp_session_id);
|
||||
match maestro_acp::spawn(config).await {
|
||||
Ok(acp) => (Some(Arc::new(acp)), None),
|
||||
Err(error) => {
|
||||
tracing::warn!(%id, %error, "structured session: adapter failed to resume");
|
||||
(None, Some(error.to_string()))
|
||||
let (acp, failure) = if recoverable {
|
||||
match session.acp_session_id.clone() {
|
||||
Some(acp_session_id) => {
|
||||
let config = SpawnConfig::new(
|
||||
&session.cwd,
|
||||
session_env
|
||||
.iter()
|
||||
.map(|(key, value)| (key.clone(), value.clone()))
|
||||
.collect(),
|
||||
)
|
||||
.load_session(acp_session_id);
|
||||
match maestro_acp::spawn(config).await {
|
||||
Ok(acp) => (Some(Arc::new(acp)), None),
|
||||
Err(error) => {
|
||||
tracing::warn!(%id, %error, "structured session: adapter failed to resume");
|
||||
(None, Some(error.to_string()))
|
||||
}
|
||||
}
|
||||
}
|
||||
None => (
|
||||
None,
|
||||
Some("structured session has no stored ACP session id".into()),
|
||||
),
|
||||
}
|
||||
None => (
|
||||
None,
|
||||
Some("structured session has no stored ACP session id".into()),
|
||||
),
|
||||
} else {
|
||||
(None, None)
|
||||
};
|
||||
session.status = if failure.is_some() {
|
||||
session.status = if recoverable {
|
||||
if failure.is_some() {
|
||||
SessionStatus::Dead
|
||||
} else {
|
||||
SessionStatus::Running
|
||||
}
|
||||
} else {
|
||||
SessionStatus::Dead
|
||||
} else {
|
||||
SessionStatus::Running
|
||||
};
|
||||
session.activity = if failure.is_some() {
|
||||
ActivityState::Error
|
||||
} else {
|
||||
ActivityState::Active
|
||||
};
|
||||
session.activity_detail = failure;
|
||||
if recoverable {
|
||||
session.activity = if failure.is_some() {
|
||||
ActivityState::Error
|
||||
} else {
|
||||
ActivityState::Active
|
||||
};
|
||||
session.activity_detail = failure;
|
||||
}
|
||||
|
||||
let acp_events = if let Some(acp) = &acp {
|
||||
match self.inner.sidecar.open_events(id) {
|
||||
@@ -1724,6 +1735,7 @@ async fn mark_session_dead(inner: &Arc<Inner>, id: Uuid) {
|
||||
/// exit: the sidecar is the recovery inventory, including its worktree.
|
||||
async fn mark_session_dead_after_shutdown(inner: &Arc<Inner>, id: Uuid) {
|
||||
mark_session_dead_with_worktree_settlement(inner, id, false).await;
|
||||
let _ = inner.sidecar.mark_for_recovery(id);
|
||||
}
|
||||
|
||||
async fn mark_session_dead_with_worktree_settlement(
|
||||
|
||||
@@ -56,6 +56,28 @@ impl SidecarStore {
|
||||
self.dir.join(format!("{id}.events.jsonl"))
|
||||
}
|
||||
|
||||
/// Durable evidence that graceful daemon shutdown ended this structured
|
||||
/// session and the next daemon may resume it.
|
||||
pub fn recovery_path_for(&self, id: Uuid) -> PathBuf {
|
||||
self.dir.join(format!("{id}.recover"))
|
||||
}
|
||||
|
||||
/// Mark `id` as eligible for one structured-session recovery attempt.
|
||||
pub fn mark_for_recovery(&self, id: Uuid) -> Result<(), SidecarError> {
|
||||
std::fs::create_dir_all(&self.dir)?;
|
||||
std::fs::write(self.recovery_path_for(id), [])?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Consume the graceful-shutdown recovery marker for `id`.
|
||||
pub fn take_recovery_marker(&self, id: Uuid) -> Result<bool, SidecarError> {
|
||||
match std::fs::remove_file(self.recovery_path_for(id)) {
|
||||
Ok(()) => Ok(true),
|
||||
Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(false),
|
||||
Err(error) => Err(error.into()),
|
||||
}
|
||||
}
|
||||
|
||||
/// Open the event log for `id` for appending, creating it (and the
|
||||
/// directory) when missing. One handle per live session: the caller
|
||||
/// keeps it for the session's lifetime rather than reopening per
|
||||
@@ -164,10 +186,11 @@ impl SidecarStore {
|
||||
}
|
||||
}
|
||||
|
||||
/// Delete the sidecar and the event log for `id`. No-op for whichever
|
||||
/// does not exist.
|
||||
/// Delete the sidecar, event log, and recovery marker for `id`. No-op for
|
||||
/// whichever does not exist.
|
||||
pub fn delete(&self, id: Uuid) -> Result<(), SidecarError> {
|
||||
remove_if_present(&self.events_path_for(id))?;
|
||||
remove_if_present(&self.recovery_path_for(id))?;
|
||||
remove_if_present(&self.path_for(id))
|
||||
}
|
||||
|
||||
|
||||
@@ -232,6 +232,12 @@ async fn recovery_loads_the_acp_session_without_duplicating_history() {
|
||||
let old_pid = fake.pid();
|
||||
fx.manager.shutdown().await;
|
||||
wait_for_exit(old_pid).await;
|
||||
assert!(
|
||||
fx.sidecar_dir()
|
||||
.join(format!("{}.recover", session.id))
|
||||
.exists(),
|
||||
"graceful shutdown must record recovery provenance"
|
||||
);
|
||||
|
||||
let revived = SessionManager::new(
|
||||
Arc::new(fx.tmux.tmux.clone()),
|
||||
@@ -600,7 +606,27 @@ async fn adapter_exit_marks_the_session_dead_and_keeps_it_listed() {
|
||||
})
|
||||
.await;
|
||||
assert!(fx.manager.has(session.id).await);
|
||||
assert!(
|
||||
!fx.sidecar_dir()
|
||||
.join(format!("{}.recover", session.id))
|
||||
.exists(),
|
||||
"a naturally exited adapter is never marked for recovery"
|
||||
);
|
||||
|
||||
fx.manager.kill(session.id, false).await.expect("dismiss");
|
||||
assert!(!fx.manager.has(session.id).await);
|
||||
let revived = SessionManager::new(
|
||||
Arc::new(fx.tmux.tmux.clone()),
|
||||
SidecarStore::new(fx.sidecar_dir()),
|
||||
WorktreeManager::new(None),
|
||||
fx.fifo_dir(),
|
||||
);
|
||||
assert_eq!(revived.recover().await.expect("recover"), 1);
|
||||
assert_eq!(revived.status(session.id).await, Some(SessionStatus::Dead));
|
||||
assert!(
|
||||
!fake.log().contains("recv session/load"),
|
||||
"a naturally dead adapter must not be resumed: {}",
|
||||
fake.log()
|
||||
);
|
||||
|
||||
revived.kill(session.id, false).await.expect("dismiss");
|
||||
assert!(!revived.has(session.id).await);
|
||||
}
|
||||
|
||||
@@ -68,15 +68,30 @@ fn delete_removes_file_and_is_idempotent() {
|
||||
let s = sample(id);
|
||||
|
||||
store.write(&s).unwrap();
|
||||
store.mark_for_recovery(id).unwrap();
|
||||
assert!(store.path_for(id).exists());
|
||||
assert!(store.recovery_path_for(id).exists());
|
||||
|
||||
store.delete(id).unwrap();
|
||||
assert!(!store.path_for(id).exists());
|
||||
assert!(!store.recovery_path_for(id).exists());
|
||||
|
||||
// second delete is a no-op
|
||||
store.delete(id).unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn recovery_marker_is_consumed_once() {
|
||||
let dir = TempDir::new().unwrap();
|
||||
let store = SidecarStore::new(dir.path());
|
||||
let id = Uuid::new_v4();
|
||||
|
||||
assert!(!store.take_recovery_marker(id).unwrap());
|
||||
store.mark_for_recovery(id).unwrap();
|
||||
assert!(store.take_recovery_marker(id).unwrap());
|
||||
assert!(!store.take_recovery_marker(id).unwrap());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn load_all_recovers_every_sidecar_and_skips_garbage() {
|
||||
let dir = TempDir::new().unwrap();
|
||||
|
||||
Reference in New Issue
Block a user