Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
160 changes: 132 additions & 28 deletions crates/daemon/src/succession.rs
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,8 @@
//! counts as dead until it's seen again.
//!
//! A new session that registers while its predecessor's client dropped only
//! moments ago, inside the grace, gets a fresh nickname first. The sweeper
//! moments ago, inside the grace, or while a predecessor that never had a
//! client is still inside its quiet time (MUS-93), gets a fresh nickname first. The sweeper
//! checks again each tick for the swap window (10 min), and once the
//! predecessor counts as dead the new session takes its nickname. Its fresh
//! nickname stays an alias for it for the alias time (10 min), and its next
Expand Down Expand Up @@ -144,6 +145,46 @@ pub struct Facts<'a> {
pub known: bool,
}

/// The latest sign of life a row shows: its last activity, tool or connection.
fn last_signal(row: &Session) -> i64 {
[
row.last_activity_at,
row.last_tool_at,
Some(row.connected_at),
]
.into_iter()
.flatten()
.max()
.unwrap_or(0)
}

/// MUS-93: when a predecessor that [`verdict`] calls live only because it
/// never had a client and was active inside the quiet time would first count
/// as dead by the same rule, if nothing more is heard from it: its last signal
/// plus the quiet time. `None` when it's live for any other reason, or not live.
///
/// A Claude session restarted before its MCP shim bound leaves such a row. The
/// rule isn't loosened: the sweeper only applies it again later.
pub fn quiet_until(f: &Facts, now: i64, cfg: &Config) -> Option<i64> {
let dead_at = last_signal(judged_by_quiet_time(f)?) + cfg.quiet_ms;
(dead_at > now).then_some(dead_at)
}

/// A Codex row on the roster, while it can't be told from an idle live thread
/// (MUS-77).
fn codex_held(row: &Session) -> bool {
row.harness.as_deref() == Some("codex") && !CODEX_PRESENT_ROWS_CAN_BE_DEAD
}

/// The row, when [`verdict`] judges it by the quiet time alone: it's on the
/// roster, isn't waiting, isn't a held Codex row, and never had a client. Both
/// `verdict` and [`quiet_until`] decide that here, so they can't drift apart.
fn judged_by_quiet_time<'a>(f: &Facts<'a>) -> Option<&'a Session> {
let row = f.row?;
let never_had_a_client = row.client_id.is_none() && f.client_gone_at.is_none();
(!f.waiting && !codex_held(row) && never_had_a_client).then_some(row)
}

/// Whether a predecessor is dead. Pure, so it can be tested on its own.
pub fn verdict(f: &Facts, now: i64, cfg: &Config) -> Verdict {
let Some(row) = f.row else {
Expand All @@ -165,20 +206,27 @@ pub fn verdict(f: &Facts, now: i64, cfg: &Config) -> Verdict {
if row.client_id.is_some() && f.client_live {
return Verdict::Live("its client is connected".into());
}
if row.harness.as_deref() == Some("codex") && !CODEX_PRESENT_ROWS_CAN_BE_DEAD {
if codex_held(row) {
return Verdict::Live(
"a Codex row on the roster can't be told from an idle live thread until MUS-77".into(),
);
}
let last_signal = [
row.last_activity_at,
row.last_tool_at,
Some(row.connected_at),
]
.into_iter()
.flatten()
.max()
.unwrap_or(0);
let last_signal = last_signal(row);
if judged_by_quiet_time(f).is_some() {
let quiet_for = now - last_signal;
return if quiet_for < cfg.quiet_ms {
Verdict::Live(format!(
"it never had a client, and was active {}s ago (quiet time {}s)",
quiet_for / 1000,
cfg.quiet_ms / 1000
))
} else {
Verdict::Dead(format!(
"it never had a client, has no waiter, and has been quiet for {}s",
quiet_for / 1000
))
};
}
match f.client_gone_at {
Some(gone_at) => {
// The hooked_since guard: a session still hooking after its client
Expand All @@ -204,25 +252,11 @@ pub fn verdict(f: &Facts, now: i64, cfg: &Config) -> Verdict {
}
// Its client is gone, but the sweeper hasn't timed the drop yet (it
// ticks every few seconds). Pending, so a quick restart is checked
// again once the drop is timed.
None if row.client_id.is_some() => Verdict::Pending(
// again once the drop is timed. (A row that never had a client was
// judged by the quiet time above.)
None => Verdict::Pending(
"its client isn't connected, and the daemon hasn't timed the drop yet".into(),
),
None => {
let quiet_for = now - last_signal;
if quiet_for < cfg.quiet_ms {
Verdict::Live(format!(
"it never had a client, and was active {}s ago (quiet time {}s)",
quiet_for / 1000,
cfg.quiet_ms / 1000
))
} else {
Verdict::Dead(format!(
"it never had a client, has no waiter, and has been quiet for {}s",
quiet_for / 1000
))
}
}
}
}

Expand Down Expand Up @@ -379,6 +413,19 @@ fn find_predecessor(
);
check_again(before + cfg.settle_ms);
}
// MUS-93: live only because it never had a client and was
// active inside the quiet time. Check again until the quiet
// time has passed, a tick after, never past the swap window:
// a sibling that keeps signalling moves its own time on, and
// keeps its name once the window ends.
if let Some(dead_at) = quiet_until(&facts, now, cfg) {
let tick = crate::cleanup::TICK_INTERVAL.as_millis() as i64;
log::info!(
"[identity] {me} may take over from {id} if it stays quiet until {}s from now: {why}",
(dead_at - now) / 1000
);
check_again((dead_at + tick).min(before + cfg.window_ms));
}
if role {
log::info!("[identity] {me} won't take role {key} over: {id} holds it ({why})");
role_held = true;
Expand Down Expand Up @@ -881,6 +928,63 @@ mod tests {
assert!(dead(verdict(&f, 1_000 + 600_000, &cfg())));
}

/// MUS-93: a row live only because it never had a client and was active
/// inside the quiet time says when the same rule would call it dead, and
/// that time is the verdict's own boundary. Live for any other reason, it
/// says nothing, so nothing is checked again for it.
#[test]
fn a_quiet_row_says_when_it_would_count_as_dead() {
let r = row("claude", None, 1_000);
let f = Facts {
row: Some(&r),
..Default::default()
};
let at = quiet_until(&f, 1_000, &cfg()).expect("live only for the quiet time");
assert_eq!(at, 1_000 + 600_000);
assert!(matches!(verdict(&f, at - 1, &cfg()), Verdict::Live(_)));
assert!(dead(verdict(&f, at, &cfg())));
// Past it, it isn't waiting on the quiet time any more.
assert_eq!(quiet_until(&f, at, &cfg()), None);
// A sibling that signals again moves its own time on.
let later = row("claude", None, 300_000);
let f_later = Facts {
row: Some(&later),
..Default::default()
};
assert_eq!(quiet_until(&f_later, 300_000, &cfg()), Some(900_000));
// Live for another reason: a waiter, a client bound or dropped, a
// Codex row before MUS-77, or no row at all.
let waiting = Facts {
row: Some(&r),
waiting: true,
..Default::default()
};
assert_eq!(quiet_until(&waiting, 1_000, &cfg()), None);
let bound = row("claude", Some("c1"), 1_000);
let connected = Facts {
row: Some(&bound),
client_live: true,
..Default::default()
};
assert_eq!(quiet_until(&connected, 1_000, &cfg()), None);
let dropped = Facts {
row: Some(&r),
client_gone_at: Some(1_000),
..Default::default()
};
assert_eq!(quiet_until(&dropped, 1_000, &cfg()), None);
let codex = row("codex", None, 1_000);
let codex_f = Facts {
row: Some(&codex),
..Default::default()
};
assert_eq!(
quiet_until(&codex_f, 1_000, &cfg()).is_some(),
CODEX_PRESENT_ROWS_CAN_BE_DEAD
);
assert_eq!(quiet_until(&Facts::default(), 1_000, &cfg()), None);
}

/// Flip with `CODEX_PRESENT_ROWS_CAN_BE_DEAD` once MUS-77 is on main.
#[test]
fn codex_rows_wait_for_mus_77() {
Expand Down
129 changes: 129 additions & 0 deletions crates/daemon/tests/identity_quiet.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,129 @@
//! MUS-93: a restart whose predecessor never had a client, and was active
//! inside the quiet time when the new session registered, as a Claude session
//! restarted before its MCP shim bound. In its own test binary, because the
//! sweeper's observations (`succession::observe`) are process-wide.

use std::sync::Arc;

use daemon::{nickname_assign::ensure_assigned, succession};
use muster_entities::{HostInfo, Session, SessionId};
use myko::command::CommandContext;
use myko::server::{MykoServerContext, Persister};
use myko_server::{BlackholePersister, MykoServer};
use uuid::Uuid;

fn setup() -> MykoServerContext {
muster_entities::link();
daemon::link();
let blackhole: Arc<dyn Persister> = Arc::new(BlackholePersister);
let server = MykoServer::builder()
.with_default_persister(blackhole)
.build();
let ctx = server.ctx();
Box::leak(Box::new(server));
ctx
}

/// A Claude session with no client bound, last active at `at`.
fn session(cwd: &str, at: i64) -> Session {
Session {
id: SessionId(Arc::from(Uuid::new_v4().to_string().as_str())),
client_id: None,
pid: 1,
cwd: cwd.into(),
git_branch: None,
current_task: None,
current_task_at: None,
session_name: None,
activity: None,
kind: None,
connected_at: at,
last_activity_at: Some(at),
last_tool: None,
last_tool_at: None,
operator: Some("op@example.com".into()),
host: Some(HostInfo {
name: "a1".into(),
os: String::new(),
arch: String::new(),
}),
project: None,
channels_enabled: None,
acks_pushes: None,
harness: Some("claude".into()),
identity: None,
}
}

fn cmd(ctx: &MykoServerContext) -> CommandContext {
succession::internal_ctx(ctx)
}

/// Register a session and name it, as the session-start hook does.
fn start(ctx: &MykoServerContext, s: &Session) -> String {
ctx.set(s).expect("session SET");
ensure_assigned(&cmd(ctx), s.id.0.as_ref()).expect("named")
}

fn unique_cwd() -> String {
format!("/work/{}", Uuid::new_v4())
}

/// The quiet time and the swap window, at their defaults (600 s each).
const QUIET_MS: i64 = 600_000;
const WINDOW_MS: i64 = 600_000;

#[test]
fn a_predecessor_live_only_for_its_quiet_time_is_still_taken_over() {
let ctx = setup();
let cwd = unique_cwd();
let t0 = chrono::Utc::now().timestamp_millis();
// The old session hooked just now, and its MCP shim never bound before it
// ended: no client, so only the quiet time can tell it's gone.
let old = session(&cwd, t0);
let nick = start(&ctx, &old);
let new = session(&cwd, t0);
let fresh = start(&ctx, &new);
assert_ne!(
fresh, nick,
"a predecessor active inside the quiet time isn't taken over at registration"
);
// Before the quiet time has passed, a late check still finds it live.
succession::recheck_pending(&cmd(&ctx), t0 + QUIET_MS - 1_000);
assert_eq!(
ensure_assigned(&cmd(&ctx), new.id.0.as_ref()).unwrap(),
fresh
);
// Once it has, the late check takes it over: the same rule, applied later.
succession::recheck_pending(&cmd(&ctx), t0 + QUIET_MS);
assert_eq!(
ensure_assigned(&cmd(&ctx), new.id.0.as_ref()).unwrap(),
nick,
"once its quiet time has passed, the new session takes its nickname"
);

// The other side, in the same test because the sweeper's state is shared:
// a sibling that keeps signalling is never taken over, even after the swap
// window, when nothing checks any more.
let cwd = unique_cwd();
let t0 = chrono::Utc::now().timestamp_millis();
let sibling = session(&cwd, t0);
let sibling_nick = start(&ctx, &sibling);
let later = session(&cwd, t0);
let later_nick = start(&ctx, &later);
assert_ne!(later_nick, sibling_nick);
let mut signalled = sibling.clone();
for at in [t0 + 300_000, t0 + 590_000] {
signalled.last_activity_at = Some(at);
ctx.set(&signalled).expect("sibling signals");
succession::recheck_pending(&cmd(&ctx), at + 1_000);
}
succession::recheck_pending(&cmd(&ctx), t0 + WINDOW_MS);
// Past the window it would count as dead, but the check has ended.
succession::recheck_pending(&cmd(&ctx), t0 + 590_000 + QUIET_MS);
assert_eq!(
ensure_assigned(&cmd(&ctx), later.id.0.as_ref()).unwrap(),
later_nick,
"a sibling that kept signalling through the swap window keeps its name"
);
}
18 changes: 18 additions & 0 deletions docs/roadmap.md
Original file line number Diff line number Diff line change
Expand Up @@ -168,6 +168,24 @@ Merged and needing a live run: MUS-36 and MUS-37.
`restart-keeps-identity` failed at the full 20 s on main's push of 65d922b (run 42107003441)
and on #125 at 0adfa2c (run 42111510686), so it's on `FLAKY` for this story as well (#175). The
#125 run printed the daemon's `[identity]` lines for the failed cell, so it can supply that log.
**Cause found** (2026-11-07, plush-tortoise, from #194's runs 42324052913 and 42330531873):
`claude` × `unread-mail-follows-a-restart` failed the suite twice, and `claude` ×
`restart-keeps-identity` once on `FLAKY`. Each failed cell's daemon log says
`predecessor … is Live("it never had a client, and was active 0s ago (quiet time 2s)")`. The
test's old Claude session was restarted before its MCP shim bound, so its row had no client,
and `verdict` called it live inside the quiet time. The sweep schedules a late check only for a
live client or waiter (MUS-92), so this verdict was never checked again, though the same rule
calls the row dead 2 s later. Fixed by checking such a predecessor again until its quiet time
has passed (`succession::quiet_until`), capped at the swap window, with a unit test and
`crates/daemon/tests/identity_quiet.rs`, which fails without the fix. The `claude-tls` cells'
CI failures predate any daemon log, so it isn't known whether they share this cause. The cells
stay on `FLAKY` until they pass in CI on three PRs in a row, as above.
- [ ] **MUS-115: MUS-93's cells come off `FLAKY` after three green PRs in a row.** Split from
MUS-93 at the watchman's read of #200, which verified the cause and the fix. Done when: `claude`
and `claude-tls` × `restart-keeps-identity`, and `claude-tls` × `unread-mail-follows-a-restart`,
have passed in CI on three PRs in a row after #200, and come off `FLAKY`; and whether the
`claude-tls` failures share MUS-93's cause is checked from a failing run's daemon log, if one
comes, or said to be unknown.
- [x] **MUS-47: Hand unread mail to a successor.** From MUS-31. Retained mail reaches
nobody without this. Depends on MUS-3. Done when: a session taking over an identity
receives its predecessor's unread mail, shown by a test.
Expand Down
8 changes: 8 additions & 0 deletions docs/stable-identity.md
Original file line number Diff line number Diff line change
Expand Up @@ -55,6 +55,14 @@ MUS-92). That catches a new session that starts before the old one's socket
has closed. It still has to turn dead by the rules above. A sibling that's
still live after 60 seconds keeps its name.

A predecessor that never had a client, and looks live only because it was
active inside the quiet time, is checked again until the quiet time has
passed, and never after the swap window (MUS-93). That's a Claude session
restarted before its MCP shim bound: its row has no client, so only the quiet
time can tell it's gone. The rule isn't loosened; it's applied again later. A
sibling that keeps signalling moves its own quiet time on, and keeps its name
once the window ends.

## What a takeover does

In one step, the daemon:
Expand Down
Loading