Compare 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 = [
|
||||
"chrono",
|
||||
"futures-util",
|
||||
"libc",
|
||||
"maestro-protocol",
|
||||
"maestro-testing",
|
||||
"maestro-tui",
|
||||
|
||||
@@ -17,6 +17,8 @@ Environment knobs, all optional:
|
||||
signal SIGTERM
|
||||
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_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_ON_CANCEL=1 exit 0 on `session/cancel` (default: keep running
|
||||
so the SIGTERM path is exercised)
|
||||
@@ -36,6 +38,7 @@ import json
|
||||
import os
|
||||
import signal
|
||||
import sys
|
||||
import time
|
||||
|
||||
LOG = os.environ.get("FAKE_ACP_LOG")
|
||||
|
||||
@@ -113,6 +116,7 @@ for raw in sys.stdin:
|
||||
"authMethods": [],
|
||||
}})
|
||||
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(chunk("hello from fake"))
|
||||
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
|
||||
// unit's `KillMode=process` signals only the daemon, so nothing else
|
||||
// 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())
|
||||
.expect("install SIGTERM handler");
|
||||
tokio::select! {
|
||||
_ = tokio::signal::ctrl_c() => tracing::info!("ctrl-c 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)
|
||||
.tcp_nodelay(true)
|
||||
.with_graceful_shutdown(shutdown)
|
||||
.await?;
|
||||
// tmux sessions outlive the daemon by design; structured sessions'
|
||||
// adapter children do not (#134).
|
||||
manager.shutdown().await;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -170,6 +170,7 @@ fn session_error_status(code: &str) -> StatusCode {
|
||||
// The session outlived its pane: nothing to type into, and nothing
|
||||
// the daemon can do about it.
|
||||
"not_running" => StatusCode::CONFLICT,
|
||||
"shutting_down" => StatusCode::SERVICE_UNAVAILABLE,
|
||||
_ => StatusCode::BAD_GATEWAY,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -7,7 +7,7 @@
|
||||
|
||||
use std::collections::{HashMap, HashSet};
|
||||
use std::path::PathBuf;
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
@@ -127,6 +127,8 @@ pub enum SessionError {
|
||||
/// a stray keystroke on an attach socket, which is dropped.
|
||||
#[error("session is no longer running: {0}")]
|
||||
NotRunning(Uuid),
|
||||
#[error("daemon is shutting down")]
|
||||
ShuttingDown,
|
||||
/// The kill was refused (#79): the session's worktree holds unsaved work.
|
||||
/// Nothing happened — the session keeps running and the tree is untouched,
|
||||
/// so it stays the operator's handle on their own unfinished work.
|
||||
@@ -144,6 +146,7 @@ impl SessionError {
|
||||
SessionError::NotFound(_) => "not_found",
|
||||
SessionError::InvalidEnv(_) => "invalid_env",
|
||||
SessionError::NotRunning(_) => "not_running",
|
||||
SessionError::ShuttingDown => "shutting_down",
|
||||
SessionError::KillGuarded(_) => "kill_guarded",
|
||||
};
|
||||
let unsaved_work = match self {
|
||||
@@ -195,6 +198,12 @@ struct Inner {
|
||||
/// client that reconnects never collides with the registration its
|
||||
/// previous socket has not finished releasing.
|
||||
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 {
|
||||
@@ -237,6 +246,8 @@ impl SessionManager {
|
||||
events,
|
||||
viewports: Mutex::new(Viewports::new()),
|
||||
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
|
||||
/// alone — surviving a restart is what they are for.
|
||||
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 sessions = self.inner.sessions.lock().await;
|
||||
sessions
|
||||
.iter()
|
||||
.filter_map(|(id, r)| {
|
||||
r.acp
|
||||
.clone()
|
||||
(r.session.status == SessionStatus::Running)
|
||||
.then_some(())
|
||||
.and_then(|()| r.acp.clone())
|
||||
.map(|acp| (*id, acp, r.acp_event_pump.clone()))
|
||||
})
|
||||
.collect()
|
||||
@@ -374,7 +390,7 @@ impl SessionManager {
|
||||
.map(wait_for_event_pump);
|
||||
futures_util::future::join_all(drains).await;
|
||||
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`.
|
||||
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 {
|
||||
Some(spec) => {
|
||||
let WorktreeSpec {
|
||||
@@ -1438,6 +1465,9 @@ async fn probe_worktree(
|
||||
/// 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.
|
||||
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
|
||||
// nothing to reap and no reason to spawn a tmux subprocess (this also
|
||||
// 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
|
||||
/// steady state produces no diffs. Runs on the existing liveness tick.
|
||||
async fn run_activity_pass(inner: &Arc<Inner>) {
|
||||
if inner.shutting_down.load(Ordering::Acquire) {
|
||||
return;
|
||||
}
|
||||
let ids: Vec<Uuid> = {
|
||||
let sessions = inner.sessions.lock().await;
|
||||
if sessions.is_empty() {
|
||||
@@ -1539,6 +1572,9 @@ async fn run_activity_pass(inner: &Arc<Inner>) {
|
||||
|
||||
let updated = {
|
||||
let mut sessions = inner.sessions.lock().await;
|
||||
if inner.shutting_down.load(Ordering::Acquire) {
|
||||
return;
|
||||
}
|
||||
let Some(rec) = sessions.get_mut(&id) else {
|
||||
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
|
||||
/// emits `SessionUpdated` when the value actually changed.
|
||||
async fn run_blocked_work_pass(inner: &Arc<Inner>) {
|
||||
if inner.shutting_down.load(Ordering::Acquire) {
|
||||
return;
|
||||
}
|
||||
let targets: Vec<(Uuid, (PathBuf, PathBuf))> = {
|
||||
let sessions = inner.sessions.lock().await;
|
||||
sessions
|
||||
@@ -1640,6 +1679,9 @@ async fn run_blocked_work_pass(inner: &Arc<Inner>) {
|
||||
};
|
||||
let updated = {
|
||||
let mut sessions = inner.sessions.lock().await;
|
||||
if inner.shutting_down.load(Ordering::Acquire) {
|
||||
return;
|
||||
}
|
||||
let Some(rec) = sessions.get_mut(&id) else {
|
||||
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).
|
||||
///
|
||||
/// 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
|
||||
/// (#81), which leaves it on disk too.
|
||||
///
|
||||
/// Dead sessions do not survive a daemon restart: [`SessionManager::recover`]
|
||||
/// adopts only what tmux still holds.
|
||||
/// Terminal sessions that die naturally do not survive a daemon restart:
|
||||
/// [`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) {
|
||||
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 mut sessions = inner.sessions.lock().await;
|
||||
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");
|
||||
|
||||
if let Some(wt) = &worktree {
|
||||
if settle_worktree_on_exit(inner, id, wt).await {
|
||||
// The tree is settled (removed); a stale "in the way" would
|
||||
// describe something the session no longer holds.
|
||||
session.blocked_work = None;
|
||||
let mut sessions = inner.sessions.lock().await;
|
||||
if let Some(record) = sessions.get_mut(&id) {
|
||||
record.session.worktree = None;
|
||||
record.session.blocked_work = None;
|
||||
record.worktree_path = None;
|
||||
record.worktree_base_repo = None;
|
||||
if settle_worktree {
|
||||
if let Some(wt) = &worktree {
|
||||
if settle_worktree_on_exit(inner, id, wt).await {
|
||||
// The tree is settled (removed); a stale "in the way" would
|
||||
// describe something the session no longer holds.
|
||||
session.blocked_work = None;
|
||||
let mut sessions = inner.sessions.lock().await;
|
||||
if let Some(record) = sessions.get_mut(&id) {
|
||||
record.session.worktree = None;
|
||||
record.session.blocked_work = 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::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::{ADAPTER_BIN, INSTALL_COMMAND};
|
||||
use maestro_daemon::sidecar::{EventCursor, SidecarStore};
|
||||
use maestro_daemon::worktree::WorktreeManager;
|
||||
use maestro_daemon::SessionManager;
|
||||
use maestro_protocol::{ActivityState, CreateSessionRequest, SessionStatus, SessionView};
|
||||
use maestro_protocol::{
|
||||
ActivityState, CreateSessionRequest, SessionStatus, SessionView, WorktreeSpec,
|
||||
};
|
||||
use tempfile::TempDir;
|
||||
|
||||
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");
|
||||
}
|
||||
|
||||
#[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]
|
||||
async fn failed_recovery_cleans_its_fifo() {
|
||||
let fx = DaemonFixture::new();
|
||||
@@ -426,6 +498,50 @@ async fn shutdown_reaps_every_adapter_child() {
|
||||
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]
|
||||
async fn shutdown_drains_final_notifications_to_the_log() {
|
||||
let fx = DaemonFixture::new();
|
||||
|
||||
@@ -13,6 +13,7 @@ path = "src/lib.rs"
|
||||
[dependencies]
|
||||
chrono.workspace = true
|
||||
futures-util.workspace = true
|
||||
libc.workspace = true
|
||||
maestro-protocol = { workspace = true, features = ["client"] }
|
||||
maestro-testing.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
|
||||
/// server (under `TMUX_TMPDIR`) is left running so `recover()` can adopt
|
||||
/// its sessions.
|
||||
|
||||
@@ -15,7 +15,7 @@ use std::time::{Duration, Instant};
|
||||
use futures_util::{SinkExt, StreamExt};
|
||||
use maestro_e2e::util::wait_for_async;
|
||||
use maestro_e2e::{Daemon, TestClient};
|
||||
use maestro_protocol::{CreateSessionRequest, WorktreeSpec};
|
||||
use maestro_protocol::{CreateSessionRequest, SessionStatus, SessionView, WorktreeSpec};
|
||||
use tempfile::TempDir;
|
||||
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
|
||||
/// together (#87 dropped `cleanup_on_kill` from the struct), so the JSON
|
||||
/// 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");
|
||||
}
|
||||
|
||||
#[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")]
|
||||
async fn concurrent_spawns_each_get_unique_ids() {
|
||||
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
|
||||
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
|
||||
itself. Terminal sessions keep surviving restarts through tmux; structured
|
||||
ones do not yet.
|
||||
itself. A daemon restart ends an in-flight structured turn, but preserves its
|
||||
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
|
||||
|
||||
|
||||
Reference in New Issue
Block a user