Compare commits
10
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
cf322746a7 | ||
|
|
b25e733539 | ||
|
|
5a861976d7 | ||
|
|
bfe4965081 | ||
|
|
9d7ed8c071 | ||
|
|
88a40c1ea7 | ||
|
|
7049b5f1c0 | ||
|
|
c277ad630c | ||
|
|
fc9c62d46b | ||
|
|
3fe560d67e |
Generated
+1
@@ -1558,6 +1558,7 @@ version = "0.1.0"
|
|||||||
dependencies = [
|
dependencies = [
|
||||||
"chrono",
|
"chrono",
|
||||||
"futures-util",
|
"futures-util",
|
||||||
|
"libc",
|
||||||
"maestro-protocol",
|
"maestro-protocol",
|
||||||
"maestro-testing",
|
"maestro-testing",
|
||||||
"maestro-tui",
|
"maestro-tui",
|
||||||
|
|||||||
@@ -17,6 +17,8 @@ Environment knobs, all optional:
|
|||||||
signal SIGTERM
|
signal SIGTERM
|
||||||
FAKE_ACP_ECHO_ENV name of one env var to log at start
|
FAKE_ACP_ECHO_ENV name of one env var to log at start
|
||||||
FAKE_ACP_INIT_FAIL=1 answer `initialize` with a JSON-RPC error
|
FAKE_ACP_INIT_FAIL=1 answer `initialize` with a JSON-RPC error
|
||||||
|
FAKE_ACP_DELAY_SESSION_NEW_MS=n wait n milliseconds before answering
|
||||||
|
`session/new`
|
||||||
FAKE_ACP_EXIT_AFTER_NEW=1 exit 0 right after answering `session/new`
|
FAKE_ACP_EXIT_AFTER_NEW=1 exit 0 right after answering `session/new`
|
||||||
FAKE_ACP_EXIT_ON_CANCEL=1 exit 0 on `session/cancel` (default: keep running
|
FAKE_ACP_EXIT_ON_CANCEL=1 exit 0 on `session/cancel` (default: keep running
|
||||||
so the SIGTERM path is exercised)
|
so the SIGTERM path is exercised)
|
||||||
@@ -36,6 +38,7 @@ import json
|
|||||||
import os
|
import os
|
||||||
import signal
|
import signal
|
||||||
import sys
|
import sys
|
||||||
|
import time
|
||||||
|
|
||||||
LOG = os.environ.get("FAKE_ACP_LOG")
|
LOG = os.environ.get("FAKE_ACP_LOG")
|
||||||
|
|
||||||
@@ -113,6 +116,7 @@ for raw in sys.stdin:
|
|||||||
"authMethods": [],
|
"authMethods": [],
|
||||||
}})
|
}})
|
||||||
elif method == "session/new":
|
elif method == "session/new":
|
||||||
|
time.sleep(int(os.environ.get("FAKE_ACP_DELAY_SESSION_NEW_MS", "0")) / 1000)
|
||||||
send({"jsonrpc": "2.0", "id": rid, "result": {"sessionId": session_id}})
|
send({"jsonrpc": "2.0", "id": rid, "result": {"sessionId": session_id}})
|
||||||
send(chunk("hello from fake"))
|
send(chunk("hello from fake"))
|
||||||
if os.environ.get("FAKE_ACP_EXIT_AFTER_NEW") == "1":
|
if os.environ.get("FAKE_ACP_EXIT_AFTER_NEW") == "1":
|
||||||
|
|||||||
@@ -122,21 +122,23 @@ async fn run(config_override: Option<PathBuf>) -> Result<(), Box<dyn Error>> {
|
|||||||
// shutdown, and both must reach the adapter children below (#134). The
|
// shutdown, and both must reach the adapter children below (#134). The
|
||||||
// unit's `KillMode=process` signals only the daemon, so nothing else
|
// unit's `KillMode=process` signals only the daemon, so nothing else
|
||||||
// would end them.
|
// would end them.
|
||||||
let shutdown = async {
|
let shutdown_manager = manager.clone();
|
||||||
|
let shutdown = async move {
|
||||||
let mut sigterm = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
|
let mut sigterm = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
|
||||||
.expect("install SIGTERM handler");
|
.expect("install SIGTERM handler");
|
||||||
tokio::select! {
|
tokio::select! {
|
||||||
_ = tokio::signal::ctrl_c() => tracing::info!("ctrl-c received — shutting down"),
|
_ = tokio::signal::ctrl_c() => tracing::info!("ctrl-c received — shutting down"),
|
||||||
_ = sigterm.recv() => tracing::info!("SIGTERM received — shutting down"),
|
_ = sigterm.recv() => tracing::info!("SIGTERM received — shutting down"),
|
||||||
}
|
}
|
||||||
|
// Axum waits for upgraded WebSockets to drain. Structured children
|
||||||
|
// must be settled before that wait, while the signal path still owns
|
||||||
|
// the shutdown order (#179).
|
||||||
|
shutdown_manager.shutdown().await;
|
||||||
};
|
};
|
||||||
|
|
||||||
axum::serve(listener, router)
|
axum::serve(listener, router)
|
||||||
.tcp_nodelay(true)
|
.tcp_nodelay(true)
|
||||||
.with_graceful_shutdown(shutdown)
|
.with_graceful_shutdown(shutdown)
|
||||||
.await?;
|
.await?;
|
||||||
// tmux sessions outlive the daemon by design; structured sessions'
|
|
||||||
// adapter children do not (#134).
|
|
||||||
manager.shutdown().await;
|
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -170,6 +170,7 @@ fn session_error_status(code: &str) -> StatusCode {
|
|||||||
// The session outlived its pane: nothing to type into, and nothing
|
// The session outlived its pane: nothing to type into, and nothing
|
||||||
// the daemon can do about it.
|
// the daemon can do about it.
|
||||||
"not_running" => StatusCode::CONFLICT,
|
"not_running" => StatusCode::CONFLICT,
|
||||||
|
"shutting_down" => StatusCode::SERVICE_UNAVAILABLE,
|
||||||
_ => StatusCode::BAD_GATEWAY,
|
_ => StatusCode::BAD_GATEWAY,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -7,7 +7,7 @@
|
|||||||
|
|
||||||
use std::collections::{HashMap, HashSet};
|
use std::collections::{HashMap, HashSet};
|
||||||
use std::path::PathBuf;
|
use std::path::PathBuf;
|
||||||
use std::sync::atomic::{AtomicU64, Ordering};
|
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
use std::time::Duration;
|
use std::time::Duration;
|
||||||
|
|
||||||
@@ -127,6 +127,8 @@ pub enum SessionError {
|
|||||||
/// a stray keystroke on an attach socket, which is dropped.
|
/// a stray keystroke on an attach socket, which is dropped.
|
||||||
#[error("session is no longer running: {0}")]
|
#[error("session is no longer running: {0}")]
|
||||||
NotRunning(Uuid),
|
NotRunning(Uuid),
|
||||||
|
#[error("daemon is shutting down")]
|
||||||
|
ShuttingDown,
|
||||||
/// The kill was refused (#79): the session's worktree holds unsaved work.
|
/// The kill was refused (#79): the session's worktree holds unsaved work.
|
||||||
/// Nothing happened — the session keeps running and the tree is untouched,
|
/// Nothing happened — the session keeps running and the tree is untouched,
|
||||||
/// so it stays the operator's handle on their own unfinished work.
|
/// so it stays the operator's handle on their own unfinished work.
|
||||||
@@ -144,6 +146,7 @@ impl SessionError {
|
|||||||
SessionError::NotFound(_) => "not_found",
|
SessionError::NotFound(_) => "not_found",
|
||||||
SessionError::InvalidEnv(_) => "invalid_env",
|
SessionError::InvalidEnv(_) => "invalid_env",
|
||||||
SessionError::NotRunning(_) => "not_running",
|
SessionError::NotRunning(_) => "not_running",
|
||||||
|
SessionError::ShuttingDown => "shutting_down",
|
||||||
SessionError::KillGuarded(_) => "kill_guarded",
|
SessionError::KillGuarded(_) => "kill_guarded",
|
||||||
};
|
};
|
||||||
let unsaved_work = match self {
|
let unsaved_work = match self {
|
||||||
@@ -195,6 +198,12 @@ struct Inner {
|
|||||||
/// client that reconnects never collides with the registration its
|
/// client that reconnects never collides with the registration its
|
||||||
/// previous socket has not finished releasing.
|
/// previous socket has not finished releasing.
|
||||||
next_viewport: AtomicU64,
|
next_viewport: AtomicU64,
|
||||||
|
/// Once daemon shutdown starts, it owns structured-session finalization.
|
||||||
|
/// Background passes must not classify those deliberate exits as natural.
|
||||||
|
shutting_down: AtomicBool,
|
||||||
|
/// Serializes structured creation with shutdown so every adapter child is
|
||||||
|
/// either rejected or included in shutdown's final drain.
|
||||||
|
structured_lifecycle: Mutex<()>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl SessionManager {
|
impl SessionManager {
|
||||||
@@ -237,6 +246,8 @@ impl SessionManager {
|
|||||||
events,
|
events,
|
||||||
viewports: Mutex::new(Viewports::new()),
|
viewports: Mutex::new(Viewports::new()),
|
||||||
next_viewport: AtomicU64::new(0),
|
next_viewport: AtomicU64::new(0),
|
||||||
|
shutting_down: AtomicBool::new(false),
|
||||||
|
structured_lifecycle: Mutex::new(()),
|
||||||
}),
|
}),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -349,13 +360,18 @@ impl SessionManager {
|
|||||||
/// is on disk after the restart is truthful. tmux sessions are left
|
/// is on disk after the restart is truthful. tmux sessions are left
|
||||||
/// alone — surviving a restart is what they are for.
|
/// alone — surviving a restart is what they are for.
|
||||||
pub async fn shutdown(&self) {
|
pub async fn shutdown(&self) {
|
||||||
|
if self.inner.shutting_down.swap(true, Ordering::AcqRel) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
let _lifecycle = self.inner.structured_lifecycle.lock().await;
|
||||||
let children: Vec<(Uuid, Arc<AcpSession>, Option<watch::Receiver<bool>>)> = {
|
let children: Vec<(Uuid, Arc<AcpSession>, Option<watch::Receiver<bool>>)> = {
|
||||||
let sessions = self.inner.sessions.lock().await;
|
let sessions = self.inner.sessions.lock().await;
|
||||||
sessions
|
sessions
|
||||||
.iter()
|
.iter()
|
||||||
.filter_map(|(id, r)| {
|
.filter_map(|(id, r)| {
|
||||||
r.acp
|
(r.session.status == SessionStatus::Running)
|
||||||
.clone()
|
.then_some(())
|
||||||
|
.and_then(|()| r.acp.clone())
|
||||||
.map(|acp| (*id, acp, r.acp_event_pump.clone()))
|
.map(|acp| (*id, acp, r.acp_event_pump.clone()))
|
||||||
})
|
})
|
||||||
.collect()
|
.collect()
|
||||||
@@ -374,7 +390,7 @@ impl SessionManager {
|
|||||||
.map(wait_for_event_pump);
|
.map(wait_for_event_pump);
|
||||||
futures_util::future::join_all(drains).await;
|
futures_util::future::join_all(drains).await;
|
||||||
for (id, _, _) in children {
|
for (id, _, _) in children {
|
||||||
mark_session_dead(&self.inner, id).await;
|
mark_session_dead_after_shutdown(&self.inner, id).await;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -457,6 +473,17 @@ impl SessionManager {
|
|||||||
|
|
||||||
/// Spawn a new session. On success returns the freshly-created `Session`.
|
/// Spawn a new session. On success returns the freshly-created `Session`.
|
||||||
pub async fn create(&self, req: CreateSessionRequest) -> Result<Session, SessionError> {
|
pub async fn create(&self, req: CreateSessionRequest) -> Result<Session, SessionError> {
|
||||||
|
if req.view == SessionView::Structured {
|
||||||
|
let _lifecycle = self.inner.structured_lifecycle.lock().await;
|
||||||
|
if self.inner.shutting_down.load(Ordering::Acquire) {
|
||||||
|
return Err(SessionError::ShuttingDown);
|
||||||
|
}
|
||||||
|
return self.create_inner(req).await;
|
||||||
|
}
|
||||||
|
self.create_inner(req).await
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn create_inner(&self, req: CreateSessionRequest) -> Result<Session, SessionError> {
|
||||||
let (effective_cwd, worktree_path, worktree_base_repo) = match &req.worktree {
|
let (effective_cwd, worktree_path, worktree_base_repo) = match &req.worktree {
|
||||||
Some(spec) => {
|
Some(spec) => {
|
||||||
let WorktreeSpec {
|
let WorktreeSpec {
|
||||||
@@ -1438,6 +1465,9 @@ async fn probe_worktree(
|
|||||||
/// One pass of the liveness watcher: diff the in-memory session set against
|
/// One pass of the liveness watcher: diff the in-memory session set against
|
||||||
/// the tmux-reported one, and reap anything we still track but tmux has lost.
|
/// the tmux-reported one, and reap anything we still track but tmux has lost.
|
||||||
async fn run_liveness_pass(inner: &Arc<Inner>) {
|
async fn run_liveness_pass(inner: &Arc<Inner>) {
|
||||||
|
if inner.shutting_down.load(Ordering::Acquire) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
// Snapshot the in-memory ids first; if we own no sessions there is
|
// Snapshot the in-memory ids first; if we own no sessions there is
|
||||||
// nothing to reap and no reason to spawn a tmux subprocess (this also
|
// nothing to reap and no reason to spawn a tmux subprocess (this also
|
||||||
// avoids racing with the very first `new-session` that would otherwise
|
// avoids racing with the very first `new-session` that would otherwise
|
||||||
@@ -1497,6 +1527,9 @@ async fn run_liveness_pass(inner: &Arc<Inner>) {
|
|||||||
/// `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>) {
|
||||||
|
if inner.shutting_down.load(Ordering::Acquire) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
let ids: Vec<Uuid> = {
|
let ids: Vec<Uuid> = {
|
||||||
let sessions = inner.sessions.lock().await;
|
let sessions = inner.sessions.lock().await;
|
||||||
if sessions.is_empty() {
|
if sessions.is_empty() {
|
||||||
@@ -1539,6 +1572,9 @@ async fn run_activity_pass(inner: &Arc<Inner>) {
|
|||||||
|
|
||||||
let updated = {
|
let updated = {
|
||||||
let mut sessions = inner.sessions.lock().await;
|
let mut sessions = inner.sessions.lock().await;
|
||||||
|
if inner.shutting_down.load(Ordering::Acquire) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
let Some(rec) = sessions.get_mut(&id) else {
|
let Some(rec) = sessions.get_mut(&id) else {
|
||||||
continue;
|
continue;
|
||||||
};
|
};
|
||||||
@@ -1613,6 +1649,9 @@ async fn run_activity_pass(inner: &Arc<Inner>) {
|
|||||||
/// error must not read as "nothing blocks". Like the activity pass, this only
|
/// error must not read as "nothing blocks". Like the activity pass, this only
|
||||||
/// emits `SessionUpdated` when the value actually changed.
|
/// emits `SessionUpdated` when the value actually changed.
|
||||||
async fn run_blocked_work_pass(inner: &Arc<Inner>) {
|
async fn run_blocked_work_pass(inner: &Arc<Inner>) {
|
||||||
|
if inner.shutting_down.load(Ordering::Acquire) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
let targets: Vec<(Uuid, (PathBuf, PathBuf))> = {
|
let targets: Vec<(Uuid, (PathBuf, PathBuf))> = {
|
||||||
let sessions = inner.sessions.lock().await;
|
let sessions = inner.sessions.lock().await;
|
||||||
sessions
|
sessions
|
||||||
@@ -1640,6 +1679,9 @@ async fn run_blocked_work_pass(inner: &Arc<Inner>) {
|
|||||||
};
|
};
|
||||||
let updated = {
|
let updated = {
|
||||||
let mut sessions = inner.sessions.lock().await;
|
let mut sessions = inner.sessions.lock().await;
|
||||||
|
if inner.shutting_down.load(Ordering::Acquire) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
let Some(rec) = sessions.get_mut(&id) else {
|
let Some(rec) = sessions.get_mut(&id) else {
|
||||||
continue;
|
continue;
|
||||||
};
|
};
|
||||||
@@ -1656,7 +1698,7 @@ async fn run_blocked_work_pass(inner: &Arc<Inner>) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Mark a session whose tmux session has disappeared as [`SessionStatus::Dead`]
|
/// Mark a session whose backing process has disappeared as [`SessionStatus::Dead`]
|
||||||
/// and settle its worktree per the natural-exit rule (#82).
|
/// and settle its worktree per the natural-exit rule (#82).
|
||||||
///
|
///
|
||||||
/// The record, its [`PtyBridge`] and its sidecar all survive, so a client can
|
/// The record, its [`PtyBridge`] and its sidecar all survive, so a client can
|
||||||
@@ -1668,9 +1710,27 @@ async fn run_blocked_work_pass(inner: &Arc<Inner>) {
|
|||||||
/// alone and stays attached to the dead record for a later forced dismissal
|
/// alone and stays attached to the dead record for a later forced dismissal
|
||||||
/// (#81), which leaves it on disk too.
|
/// (#81), which leaves it on disk too.
|
||||||
///
|
///
|
||||||
/// Dead sessions do not survive a daemon restart: [`SessionManager::recover`]
|
/// Terminal sessions that die naturally do not survive a daemon restart:
|
||||||
/// adopts only what tmux still holds.
|
/// [`SessionManager::recover`] adopts only what tmux still holds. Structured
|
||||||
|
/// sessions stopped for daemon shutdown retain their sidecars for ACP recovery.
|
||||||
async fn mark_session_dead(inner: &Arc<Inner>, id: Uuid) {
|
async fn mark_session_dead(inner: &Arc<Inner>, id: Uuid) {
|
||||||
|
if inner.shutting_down.load(Ordering::Acquire) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
mark_session_dead_with_worktree_settlement(inner, id, true).await;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// A daemon restart terminates ACP children, but it is not a natural adapter
|
||||||
|
/// 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;
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn mark_session_dead_with_worktree_settlement(
|
||||||
|
inner: &Arc<Inner>,
|
||||||
|
id: Uuid,
|
||||||
|
settle_worktree: bool,
|
||||||
|
) {
|
||||||
let flipped = {
|
let flipped = {
|
||||||
let mut sessions = inner.sessions.lock().await;
|
let mut sessions = inner.sessions.lock().await;
|
||||||
match sessions.get_mut(&id) {
|
match sessions.get_mut(&id) {
|
||||||
@@ -1694,17 +1754,19 @@ async fn mark_session_dead(inner: &Arc<Inner>, id: Uuid) {
|
|||||||
};
|
};
|
||||||
tracing::info!(%id, "session process gone — marking dead");
|
tracing::info!(%id, "session process gone — marking dead");
|
||||||
|
|
||||||
if let Some(wt) = &worktree {
|
if settle_worktree {
|
||||||
if settle_worktree_on_exit(inner, id, wt).await {
|
if let Some(wt) = &worktree {
|
||||||
// The tree is settled (removed); a stale "in the way" would
|
if settle_worktree_on_exit(inner, id, wt).await {
|
||||||
// describe something the session no longer holds.
|
// The tree is settled (removed); a stale "in the way" would
|
||||||
session.blocked_work = None;
|
// describe something the session no longer holds.
|
||||||
let mut sessions = inner.sessions.lock().await;
|
session.blocked_work = None;
|
||||||
if let Some(record) = sessions.get_mut(&id) {
|
let mut sessions = inner.sessions.lock().await;
|
||||||
record.session.worktree = None;
|
if let Some(record) = sessions.get_mut(&id) {
|
||||||
record.session.blocked_work = None;
|
record.session.worktree = None;
|
||||||
record.worktree_path = None;
|
record.session.blocked_work = None;
|
||||||
record.worktree_base_repo = None;
|
record.worktree_path = None;
|
||||||
|
record.worktree_base_repo = None;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -9,13 +9,15 @@ use std::path::{Path, PathBuf};
|
|||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
use std::time::{Duration, Instant};
|
use std::time::{Duration, Instant};
|
||||||
|
|
||||||
use common::{wait_for_async, DaemonFixture};
|
use common::{init_bare_repo, init_repo, push_all_to_origin, wait_for_async, DaemonFixture};
|
||||||
use maestro_acp::schema::{ContentBlock, SessionUpdate};
|
use maestro_acp::schema::{ContentBlock, SessionUpdate};
|
||||||
use maestro_acp::{ADAPTER_BIN, INSTALL_COMMAND};
|
use maestro_acp::{ADAPTER_BIN, INSTALL_COMMAND};
|
||||||
use maestro_daemon::sidecar::{EventCursor, SidecarStore};
|
use maestro_daemon::sidecar::{EventCursor, SidecarStore};
|
||||||
use maestro_daemon::worktree::WorktreeManager;
|
use maestro_daemon::worktree::WorktreeManager;
|
||||||
use maestro_daemon::SessionManager;
|
use maestro_daemon::SessionManager;
|
||||||
use maestro_protocol::{ActivityState, CreateSessionRequest, SessionStatus, SessionView};
|
use maestro_protocol::{
|
||||||
|
ActivityState, CreateSessionRequest, SessionStatus, SessionView, WorktreeSpec,
|
||||||
|
};
|
||||||
use tempfile::TempDir;
|
use tempfile::TempDir;
|
||||||
|
|
||||||
fn fake_agent() -> PathBuf {
|
fn fake_agent() -> PathBuf {
|
||||||
@@ -285,6 +287,76 @@ async fn recovery_loads_the_acp_session_without_duplicating_history() {
|
|||||||
revived.kill(session.id, false).await.expect("kill");
|
revived.kill(session.id, false).await.expect("kill");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn shutdown_preserves_a_pristine_structured_worktree_for_recovery() {
|
||||||
|
let fx = DaemonFixture::new();
|
||||||
|
let fake = Fake::new();
|
||||||
|
let repo = TempDir::new().unwrap();
|
||||||
|
init_repo(repo.path());
|
||||||
|
let origin = TempDir::new().unwrap();
|
||||||
|
init_bare_repo(origin.path());
|
||||||
|
|
||||||
|
let branch = "feature/restart";
|
||||||
|
let worktree = repo
|
||||||
|
.path()
|
||||||
|
.join("worktrees")
|
||||||
|
.join("feature")
|
||||||
|
.join("restart");
|
||||||
|
let session = fx
|
||||||
|
.manager
|
||||||
|
.create(CreateSessionRequest {
|
||||||
|
view: SessionView::Structured,
|
||||||
|
cwd: repo.path().to_path_buf(),
|
||||||
|
command: "claude".into(),
|
||||||
|
env: fake.env(),
|
||||||
|
metadata: HashMap::new(),
|
||||||
|
worktree: Some(WorktreeSpec {
|
||||||
|
base_repo: repo.path().to_path_buf(),
|
||||||
|
branch: branch.into(),
|
||||||
|
create_branch: true,
|
||||||
|
}),
|
||||||
|
})
|
||||||
|
.await
|
||||||
|
.expect("create");
|
||||||
|
assert_eq!(session.cwd, worktree);
|
||||||
|
push_all_to_origin(repo.path(), origin.path());
|
||||||
|
|
||||||
|
fx.manager.shutdown().await;
|
||||||
|
|
||||||
|
assert!(
|
||||||
|
worktree.is_dir(),
|
||||||
|
"shutdown must not settle a pristine worktree"
|
||||||
|
);
|
||||||
|
let sidecar: maestro_protocol::Session = serde_json::from_slice(
|
||||||
|
&std::fs::read(fx.sidecar_dir().join(format!("{}.json", session.id))).unwrap(),
|
||||||
|
)
|
||||||
|
.unwrap();
|
||||||
|
assert_eq!(
|
||||||
|
sidecar.worktree.as_ref().map(|wt| &wt.path),
|
||||||
|
Some(&worktree)
|
||||||
|
);
|
||||||
|
|
||||||
|
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::Running)
|
||||||
|
);
|
||||||
|
assert!(fake.log().contains("recv session/load"), "{}", fake.log());
|
||||||
|
assert!(worktree.is_dir(), "recovery must retain the worktree");
|
||||||
|
|
||||||
|
revived.kill(session.id, false).await.expect("dismiss");
|
||||||
|
assert!(
|
||||||
|
!worktree.exists(),
|
||||||
|
"operator dismissal still cleans a pristine tree"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn failed_recovery_cleans_its_fifo() {
|
async fn failed_recovery_cleans_its_fifo() {
|
||||||
let fx = DaemonFixture::new();
|
let fx = DaemonFixture::new();
|
||||||
@@ -426,6 +498,50 @@ async fn shutdown_reaps_every_adapter_child() {
|
|||||||
assert_eq!(sidecar.status, SessionStatus::Dead);
|
assert_eq!(sidecar.status, SessionStatus::Dead);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn shutdown_waits_for_structured_creation_and_rejects_new_creates() {
|
||||||
|
let fx = DaemonFixture::new();
|
||||||
|
let fake = Fake::new();
|
||||||
|
let cwd = TempDir::new().unwrap();
|
||||||
|
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 });
|
||||||
|
wait_for_async(Duration::from_secs(5), || async {
|
||||||
|
fake.log().contains("recv session/new")
|
||||||
|
})
|
||||||
|
.await;
|
||||||
|
|
||||||
|
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"
|
||||||
|
);
|
||||||
|
|
||||||
|
let session = create.await.expect("create task").expect("create");
|
||||||
|
let pid = fake.pid();
|
||||||
|
shutdown.await.expect("shutdown task");
|
||||||
|
|
||||||
|
wait_for_exit(pid).await;
|
||||||
|
assert!(fake.log().contains("signal SIGTERM"), "{}", fake.log());
|
||||||
|
assert_eq!(
|
||||||
|
fx.manager.status(session.id).await,
|
||||||
|
Some(SessionStatus::Dead)
|
||||||
|
);
|
||||||
|
assert!(matches!(
|
||||||
|
fx.manager.create(structured(cwd.path(), fake.env())).await,
|
||||||
|
Err(maestro_daemon::SessionError::ShuttingDown)
|
||||||
|
));
|
||||||
|
|
||||||
|
// A repeated signal must not attempt to terminate the child again.
|
||||||
|
fx.manager.shutdown().await;
|
||||||
|
assert_eq!(fake.log().matches("signal SIGTERM").count(), 1);
|
||||||
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn shutdown_drains_final_notifications_to_the_log() {
|
async fn shutdown_drains_final_notifications_to_the_log() {
|
||||||
let fx = DaemonFixture::new();
|
let fx = DaemonFixture::new();
|
||||||
|
|||||||
@@ -13,6 +13,7 @@ path = "src/lib.rs"
|
|||||||
[dependencies]
|
[dependencies]
|
||||||
chrono.workspace = true
|
chrono.workspace = true
|
||||||
futures-util.workspace = true
|
futures-util.workspace = true
|
||||||
|
libc.workspace = true
|
||||||
maestro-protocol = { workspace = true, features = ["client"] }
|
maestro-protocol = { workspace = true, features = ["client"] }
|
||||||
maestro-testing.workspace = true
|
maestro-testing.workspace = true
|
||||||
serde.workspace = true
|
serde.workspace = true
|
||||||
|
|||||||
@@ -87,6 +87,14 @@ impl Daemon {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Send the daemon the same SIGTERM systemd uses, without waiting for its
|
||||||
|
/// graceful HTTP shutdown to finish.
|
||||||
|
pub fn send_sigterm(&self) {
|
||||||
|
let child = self.child.as_ref().expect("daemon is running");
|
||||||
|
let result = unsafe { libc::kill(child.id() as libc::pid_t, libc::SIGTERM) };
|
||||||
|
assert_eq!(result, 0, "send SIGTERM to daemon");
|
||||||
|
}
|
||||||
|
|
||||||
/// Re-spawn the daemon on the same port + state. The previous tmux
|
/// Re-spawn the daemon on the same port + state. The previous tmux
|
||||||
/// server (under `TMUX_TMPDIR`) is left running so `recover()` can adopt
|
/// server (under `TMUX_TMPDIR`) is left running so `recover()` can adopt
|
||||||
/// its sessions.
|
/// its sessions.
|
||||||
|
|||||||
@@ -15,7 +15,7 @@ use std::time::{Duration, Instant};
|
|||||||
use futures_util::{SinkExt, StreamExt};
|
use futures_util::{SinkExt, StreamExt};
|
||||||
use maestro_e2e::util::wait_for_async;
|
use maestro_e2e::util::wait_for_async;
|
||||||
use maestro_e2e::{Daemon, TestClient};
|
use maestro_e2e::{Daemon, TestClient};
|
||||||
use maestro_protocol::{CreateSessionRequest, WorktreeSpec};
|
use maestro_protocol::{CreateSessionRequest, SessionStatus, SessionView, WorktreeSpec};
|
||||||
use tempfile::TempDir;
|
use tempfile::TempDir;
|
||||||
use tokio_tungstenite::tungstenite::Message as WsMessage;
|
use tokio_tungstenite::tungstenite::Message as WsMessage;
|
||||||
|
|
||||||
@@ -36,6 +36,61 @@ fn create_req(cwd: &std::path::Path, command: &str) -> CreateSessionRequest {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
struct FakeAcp {
|
||||||
|
_dir: TempDir,
|
||||||
|
env: HashMap<String, String>,
|
||||||
|
}
|
||||||
|
|
||||||
|
fn fake_acp(log: &std::path::Path) -> FakeAcp {
|
||||||
|
let dir = tempfile::tempdir().expect("fake adapter dir");
|
||||||
|
let adapter = dir.path().join("claude-code-acp");
|
||||||
|
std::fs::copy(
|
||||||
|
concat!(
|
||||||
|
env!("CARGO_MANIFEST_DIR"),
|
||||||
|
"/../maestro-acp/tests/fake_acp_agent.py"
|
||||||
|
),
|
||||||
|
&adapter,
|
||||||
|
)
|
||||||
|
.expect("copy fake adapter");
|
||||||
|
let mut permissions = std::fs::metadata(&adapter)
|
||||||
|
.expect("adapter metadata")
|
||||||
|
.permissions();
|
||||||
|
#[cfg(unix)]
|
||||||
|
{
|
||||||
|
use std::os::unix::fs::PermissionsExt;
|
||||||
|
permissions.set_mode(0o755);
|
||||||
|
}
|
||||||
|
std::fs::set_permissions(&adapter, permissions).expect("make fake adapter executable");
|
||||||
|
let env = HashMap::from([
|
||||||
|
(
|
||||||
|
"PATH".into(),
|
||||||
|
format!(
|
||||||
|
"{}:{}",
|
||||||
|
dir.path().display(),
|
||||||
|
std::env::var("PATH").unwrap_or_default()
|
||||||
|
),
|
||||||
|
),
|
||||||
|
("FAKE_ACP_LOG".into(), log.to_string_lossy().into_owned()),
|
||||||
|
]);
|
||||||
|
FakeAcp { _dir: dir, env }
|
||||||
|
}
|
||||||
|
|
||||||
|
fn fake_acp_pid(log: &std::path::Path) -> Option<u32> {
|
||||||
|
std::fs::read_to_string(log).ok()?.lines().find_map(|line| {
|
||||||
|
line.strip_prefix("start pid=")?
|
||||||
|
.split(' ')
|
||||||
|
.next()?
|
||||||
|
.parse()
|
||||||
|
.ok()
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
fn pid_alive(pid: u32) -> bool {
|
||||||
|
std::fs::read_to_string(format!("/proc/{pid}/stat"))
|
||||||
|
.map(|stat| !stat.contains(") Z "))
|
||||||
|
.unwrap_or(false)
|
||||||
|
}
|
||||||
|
|
||||||
/// A worktree spec built through its wire shape. The daemon and clients ship
|
/// A worktree spec built through its wire shape. The daemon and clients ship
|
||||||
/// together (#87 dropped `cleanup_on_kill` from the struct), so the JSON
|
/// together (#87 dropped `cleanup_on_kill` from the struct), so the JSON
|
||||||
/// object carries only the fields the lifecycle actually honours.
|
/// object carries only the fields the lifecycle actually honours.
|
||||||
@@ -995,6 +1050,66 @@ async fn daemon_restart_recovers_session() {
|
|||||||
client.kill_session(s.id).await.expect("kill");
|
client.kill_session(s.id).await.expect("kill");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test(flavor = "multi_thread")]
|
||||||
|
async fn sigterm_stops_structured_child_before_events_ws_drains() {
|
||||||
|
let (mut daemon, client) = bring_up().await;
|
||||||
|
let cwd = TempDir::new().unwrap();
|
||||||
|
let log = cwd.path().join("fake-acp.log");
|
||||||
|
let fake = fake_acp(&log);
|
||||||
|
let session = client
|
||||||
|
.spawn_session(&CreateSessionRequest {
|
||||||
|
view: SessionView::Structured,
|
||||||
|
cwd: cwd.path().to_path_buf(),
|
||||||
|
command: "claude".into(),
|
||||||
|
env: fake.env.clone(),
|
||||||
|
metadata: HashMap::new(),
|
||||||
|
worktree: None,
|
||||||
|
})
|
||||||
|
.await
|
||||||
|
.expect("spawn structured session");
|
||||||
|
let ws_url = daemon.client().events_url().expect("events URL");
|
||||||
|
let (_events_ws, _) = tokio_tungstenite::connect_async(ws_url)
|
||||||
|
.await
|
||||||
|
.expect("events WS handshake");
|
||||||
|
wait_for_async("fake ACP starts", Duration::from_secs(5), || {
|
||||||
|
let log = log.clone();
|
||||||
|
async move { fake_acp_pid(&log).is_some() }
|
||||||
|
})
|
||||||
|
.await;
|
||||||
|
let pid = fake_acp_pid(&log).expect("fake ACP pid");
|
||||||
|
|
||||||
|
daemon.send_sigterm();
|
||||||
|
|
||||||
|
wait_for_async(
|
||||||
|
"fake ACP exits after SIGTERM",
|
||||||
|
Duration::from_secs(5),
|
||||||
|
|| async { !pid_alive(pid) },
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
let sidecar_path = daemon
|
||||||
|
.state_dir()
|
||||||
|
.join("sessions")
|
||||||
|
.join(format!("{}.json", session.id));
|
||||||
|
wait_for_async(
|
||||||
|
"dead structured session sidecar",
|
||||||
|
Duration::from_secs(5),
|
||||||
|
|| {
|
||||||
|
let sidecar_path = sidecar_path.clone();
|
||||||
|
async move {
|
||||||
|
std::fs::read(&sidecar_path)
|
||||||
|
.ok()
|
||||||
|
.and_then(|json| {
|
||||||
|
serde_json::from_slice::<maestro_protocol::Session>(&json).ok()
|
||||||
|
})
|
||||||
|
.is_some_and(|saved| saved.status == SessionStatus::Dead)
|
||||||
|
}
|
||||||
|
},
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
|
||||||
|
daemon.stop();
|
||||||
|
}
|
||||||
|
|
||||||
#[tokio::test(flavor = "multi_thread")]
|
#[tokio::test(flavor = "multi_thread")]
|
||||||
async fn concurrent_spawns_each_get_unique_ids() {
|
async fn concurrent_spawns_each_get_unique_ids() {
|
||||||
let (_daemon, client) = bring_up().await;
|
let (_daemon, client) = bring_up().await;
|
||||||
|
|||||||
+4
-2
@@ -208,8 +208,10 @@ When the adapter is missing the spawn still creates the session, marked dead
|
|||||||
with the install command above as its activity detail, so the sidebar says
|
with the install command above as its activity detail, so the sidebar says
|
||||||
what to run. The adapter is a child of the daemon: it ends with a `DELETE
|
what to run. The adapter is a child of the daemon: it ends with a `DELETE
|
||||||
/api/sessions/:id` (`session/cancel`, then `SIGTERM`) and with the daemon
|
/api/sessions/:id` (`session/cancel`, then `SIGTERM`) and with the daemon
|
||||||
itself. Terminal sessions keep surviving restarts through tmux; structured
|
itself. A daemon restart ends an in-flight structured turn, but preserves its
|
||||||
ones do not yet.
|
worktree, session record, and event-log transcript. The replacement daemon
|
||||||
|
starts a new adapter, resumes it with `session/load`, and replays the
|
||||||
|
transcript. Terminal sessions keep surviving restarts through tmux.
|
||||||
|
|
||||||
## The browser UI
|
## The browser UI
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user