diff --git a/crates/protocol/src/event.rs b/crates/protocol/src/event.rs index 38d14afc..ccb29433 100644 --- a/crates/protocol/src/event.rs +++ b/crates/protocol/src/event.rs @@ -346,6 +346,12 @@ pub enum AnsweredBy { pub enum TimeoutKind { Inactivity, Total, + /// The run's owning leader has held no live connection for the full + /// disconnect grace window. Distinct from `inactivity`/`total`, + /// which describe the worker's own liveness -- this describes the + /// leader's, and is the fact a grace-window teardown journals before + /// the run is settled. + LeaderGone, } // Mirrored 1:1 from the TUI adapter's own internal `GateKind` diff --git a/crates/runtime/src/adapter/activity.rs b/crates/runtime/src/adapter/activity.rs index c1ae3e67..3f1a38d4 100644 --- a/crates/runtime/src/adapter/activity.rs +++ b/crates/runtime/src/adapter/activity.rs @@ -130,6 +130,22 @@ impl ActivityClock { match kind { TimeoutKind::Inactivity => entry.inactivity_journaled = true, TimeoutKind::Total => entry.total_journaled = true, + // This clock tracks worker/PTY liveness only -- a + // leader's connection state is `ipc::leader_registry`'s + // concern, tracked separately and never through here. + // Should be unreachable on that invariant (the sweep's + // own `since_ms` computation already skips this kind + // before ever calling here), but this is a `pub fn` a + // future caller could reach directly with a bad `kind` -- + // logging and doing nothing costs a call that flags + // nothing, never a panic in a long-lived background task. + TimeoutKind::LeaderGone => { + tracing::error!( + run_id = %run_id, + "mark_journaled called with LeaderGone -- ActivityClock has no \ + leader-connection state to flag; ignoring" + ); + } } } } diff --git a/crates/runtime/src/domain/repository.rs b/crates/runtime/src/domain/repository.rs index 3e8f3402..7e25f088 100644 --- a/crates/runtime/src/domain/repository.rs +++ b/crates/runtime/src/domain/repository.rs @@ -1836,46 +1836,11 @@ impl<'c> DomainRepository<'c> { )?)) } - /// ADR-0027 wave 3's inactivity backstop: settles a run whose vendor - /// finished a turn that the leader then never closed. - /// - /// Settles to `lost`, never `succeeded`. The runtime observed only - /// silence, and ADR-0015's precedent is to name that uncertainty rather - /// than guess an outcome -- a success reached by nobody looking would be - /// a fabricated fact. The run's journaled result text stays readable - /// either way, since `run/result` reads any terminal state. - /// - /// Returns `None` unless the run is parked in `waitingUser` **and** - /// flagged `turnSettled`. That pairing is the whole safety property: a - /// `waitingUser` that means "the worker asked a question" or "an - /// approval is pending" must never be settled by a timeout, and neither - /// must a `waitingPeer` or a `paused` run. The read and the transition - /// share one transaction so a leader finishing the run concurrently - /// cannot be raced. - pub fn settle_abandoned_turn( - &mut self, - run_id: RunId, - ) -> Result, DomainError> { - let Some((state, turn_settled)): Option<(String, i64)> = self - .conn - .query_row( - "SELECT state, flags_turn_settled FROM runs WHERE run_id = ?1", - [run_id.to_string()], - |row| Ok((row.get(0)?, row.get(1)?)), - ) - .optional()? - else { - return Ok(None); - }; - if state != "waitingUser" || turn_settled == 0 { - return Ok(None); - } - let to = RunState::try_from("lost").map_err(|_| DomainError::NotFound { - kind: "run-state", - id: "lost".to_string(), - })?; - self.transition_run(run_id, &to, None).map(Some) - } + // `settle_abandoned_turn` (ADR-0027 wave 3's inactivity backstop, + // which settled a leader-abandoned turn to `lost` after inactivity + // alone) is removed: the maintainer ruled that silence is never + // evidence a leader gave up, only its connection actually being gone + // is. See ADR-0036 and `crate::ipc::leader_registry`. /// Updates a message's delivery state. Emits the matching `Message*` /// event. @@ -3706,86 +3671,6 @@ mod tests { assert_eq!(turns_used, 0, "answers must not consume turns"); } - /// The backstop settles a turn nobody closed -- to `lost`, because - /// silence is not evidence of success. - #[test] - fn the_backstop_settles_an_abandoned_settled_turn_as_lost() { - let mut conn = open_test_db(); - let project_id = ProjectId::new(); - let (run_id, _task_id, _worker_id) = seed_budgeted_run(&mut conn, project_id, 10); - let mut repo = DomainRepository::new(&mut conn, project_id); - for state in ["starting", "working", "waitingUser"] { - let target = RunState::try_from(state).unwrap(); - repo.transition_run(run_id, &target, None) - .expect("drive to waitingUser"); - } - repo.set_run_flag(run_id, RunFlag::TurnSettled, true) - .expect("mark the wait as a finished turn"); - - let committed = repo - .settle_abandoned_turn(run_id) - .expect("settle") - .expect("an abandoned settled turn must be settled"); - match committed.envelope.event { - RuntimeEvent::RunEvent { ref state, .. } => assert_eq!(state.as_str(), "lost"), - ref other => panic!("expected a RunEvent, got {other:?}"), - } - // This layer only settles the state; clearing the marker is the - // sweep's job, so it is still set here. - assert!(repo.read_run_flags(run_id).unwrap().turn_settled); - } - - /// The safety property: a wait that is not a finished turn -- a - /// question, an approval -- must survive the backstop untouched. - #[test] - fn the_backstop_never_settles_a_wait_that_is_not_a_finished_turn() { - let mut conn = open_test_db(); - let project_id = ProjectId::new(); - let (run_id, _task_id, _worker_id) = seed_budgeted_run(&mut conn, project_id, 10); - let mut repo = DomainRepository::new(&mut conn, project_id); - for state in ["starting", "working", "waitingUser"] { - let target = RunState::try_from(state).unwrap(); - repo.transition_run(run_id, &target, None) - .expect("drive to waitingUser"); - } - // No turnSettled flag: this run is waiting on a human, not holding - // a finished answer. - - assert!( - repo.settle_abandoned_turn(run_id) - .expect("settle") - .is_none(), - "a question wait must never be settled by a timeout" - ); - assert!( - !repo.read_run_flags(run_id).unwrap().turn_settled, - "sanity: the flag really is unset" - ); - } - - /// A run still working is not abandoned, however quiet it has been. - #[test] - fn the_backstop_never_settles_a_working_run() { - let mut conn = open_test_db(); - let project_id = ProjectId::new(); - let (run_id, _task_id, _worker_id) = seed_budgeted_run(&mut conn, project_id, 10); - let mut repo = DomainRepository::new(&mut conn, project_id); - for state in ["starting", "working"] { - let target = RunState::try_from(state).unwrap(); - repo.transition_run(run_id, &target, None) - .expect("drive to working"); - } - repo.set_run_flag(run_id, RunFlag::TurnSettled, true) - .expect("set the flag even though the state does not match"); - - assert!( - repo.settle_abandoned_turn(run_id) - .expect("settle") - .is_none(), - "the state and the flag must BOTH hold; the flag alone is not enough" - ); - } - #[test] fn timeout_facts_skip_terminal_runs_but_journal_for_live_ones() { let mut conn = open_test_db(); diff --git a/crates/runtime/src/ipc/connection.rs b/crates/runtime/src/ipc/connection.rs index 647c7cbd..502ae55d 100644 --- a/crates/runtime/src/ipc/connection.rs +++ b/crates/runtime/src/ipc/connection.rs @@ -88,6 +88,43 @@ pub(crate) async fn handle(stream: UnixStream, ctx: ConnContext, shared: Arc 2 -> 1`, never touching +//! zero; a reconnect racing the other way (a fresh `0 -> 1` racing a +//! stale decrement already in flight) still resolves correctly, because +//! plain addition does not care which side's `+1`/`-1` is observed +//! first. +//! +//! A bare `count == 0` is not enough to decide "gone for the grace +//! window", though -- see [`Entry::zero_since`]'s own doc comment for +//! why, and [`LeaderRegistry::gone_for_at_least`] for the check that +//! actually answers it. + +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +/// How long a run's owning leader may hold no live connection at all +/// before the run is settled as unrendered (`RunState::unrendered_verdict`). +/// +/// The maintainer's ruling: a slot is held for at most a minute after a +/// leader genuinely leaves, which is short enough that an abandoned run +/// does not tie up a concurrency slot indefinitely, and long enough to +/// cover a quick OMP restart or reconnect -- the ordinary case this must +/// never fire for. A daemon restart is charged the same way (see +/// `LeaderRegistry::seed_disconnected_since`'s own doc comment for why +/// the clock starts at the daemon's own startup instant, never at the +/// run's last-seen time). +pub(crate) const LEADER_DISCONNECT_GRACE_WINDOW: Duration = Duration::from_secs(60); + +/// One instance id's live-connection bookkeeping. +#[derive(Default)] +struct Entry { + count: usize, + /// The instant `count` last dropped to zero, cleared the moment it + /// rises above zero again. A grace-window timer wakes long after it + /// was spawned and cannot tell, from `count == 0` alone, whether that + /// zero is the SAME one it was spawned to watch or a fresher one from + /// an intervening reconnect-then-disconnect cycle that happened to + /// also land on zero. Recording the instant of the transition (not + /// just the fact of it) is what lets the timer ask the actual + /// question: has the CURRENT absence lasted the full window, not + /// merely "is the count zero right now". + zero_since: Option, +} + +/// Refcounted live-connection registry, keyed by leader instance id. +/// Cloning shares the same underlying map (cheap, `Arc`-backed) -- every +/// clone observes and mutates the same state. +#[derive(Clone, Default)] +pub(crate) struct LeaderRegistry { + entries: Arc>>, +} + +impl LeaderRegistry { + pub(crate) fn new() -> Self { + Self::default() + } + + /// Registers one live connection for `instance_id`, returning a guard + /// that deregisters it on drop -- including on an early `return` out + /// of the connection's dispatch loop or a panic unwind, never only on + /// the loop's "normal" exit. See [`LeaderConnectionGuard`]'s own doc + /// comment for why that has to be structural. + /// + /// `on_gone` fires exactly when THIS drop is the one that brings the + /// count to zero (never on a decrement that leaves it above zero, and + /// never more than once per 1->0 transition) -- the caller's hook to + /// arm a grace-window timer. Fired from inside `Drop::drop` itself, + /// so arming is exactly as structural as the decrement it rides on: + /// a future early `return` that would otherwise skip an explicit + /// "now arm a timer" statement cannot skip this one, because there + /// is no separate statement to skip. This module knows nothing about + /// what the timer does or what "settling" means -- that stays in + /// `ipc::connection`/`service::orchestration`, which is what `on_gone` + /// is for. + pub(crate) fn register( + &self, + instance_id: String, + on_gone: impl Fn(String) + Send + Sync + 'static, + ) -> LeaderConnectionGuard { + let mut entries = self + .entries + .lock() + .expect("leader registry mutex never poisoned"); + let entry = entries.entry(instance_id.clone()).or_default(); + entry.count += 1; + entry.zero_since = None; + drop(entries); + LeaderConnectionGuard { + registry: self.clone(), + instance_id, + on_gone: Arc::new(on_gone), + } + } + + /// Decrements `instance_id`'s live count, returning `true` exactly + /// when this decrement is the one that brought it to zero (the + /// signal [`LeaderConnectionGuard::drop`] uses to fire `on_gone`). + fn deregister_reporting_zero_transition(&self, instance_id: &str) -> bool { + let mut entries = self + .entries + .lock() + .expect("leader registry mutex never poisoned"); + let Some(entry) = entries.get_mut(instance_id) else { + return false; + }; + entry.count = entry.count.saturating_sub(1); + if entry.count == 0 { + entry.zero_since = Some(Instant::now()); + true + } else { + false + } + } + + /// Seeds `instance_id` as having been at zero live connections since + /// `since`, WITHOUT creating a live connection -- the daemon-restart + /// case: at startup there are no connections at all yet, but a + /// parked run's owning leader was disconnected (from THIS daemon's + /// perspective) at the moment this process started, not whenever it + /// last happened to be observed. A leader cannot reconnect to a + /// daemon that is not running, so the clock for "how long has the + /// leader been gone" starts at startup, never at the run's own + /// last-seen time -- charging the daemon's own downtime against the + /// leader would settle live runs on every restart after a long stop, + /// which is exactly the destructive direction this feature exists to + /// avoid. + /// + /// A no-op if `instance_id` already has a live connection (count > 0) + /// by the time this runs -- seeding never clobbers a real connection + /// that beat it to registering. + pub(crate) fn seed_disconnected_since(&self, instance_id: String, since: Instant) { + let mut entries = self + .entries + .lock() + .expect("leader registry mutex never poisoned"); + let entry = entries.entry(instance_id).or_default(); + if entry.count == 0 { + entry.zero_since.get_or_insert(since); + } + } + + /// Whether `instance_id` has had zero live connections continuously + /// for at least `grace`, since the transition [`Self::gone_for_at_least`]'s + /// caller is actually asking about -- not merely whether the count + /// happens to read zero at the moment of the call. `zero_since` is + /// cleared by any intervening reconnect (see [`Self::register`]), so + /// a stale timer waking against a fresher zero correctly sees that + /// fresher zero hasn't aged enough yet, rather than settling early + /// against a zero that isn't the one it was spawned for. + pub(crate) fn gone_for_at_least(&self, instance_id: &str, grace: Duration) -> bool { + let entries = self + .entries + .lock() + .expect("leader registry mutex never poisoned"); + match entries.get(instance_id) { + Some(entry) => { + entry.count == 0 + && entry + .zero_since + .is_some_and(|since| since.elapsed() >= grace) + } + // No entry at all means nobody has ever registered OR seeded + // this id -- an unestablished question, not evidence of + // absence. Both real callers (`register`'s connection path, + // `seed_disconnected_since`'s restart path) always create the + // entry before ever asking this question, so this branch is a + // guard against a future caller's misuse, never a path either + // takes today. The destructive answer (`true`, which a caller + // reads as license to settle) must never be the default for a + // question nobody has established an answer to -- `false` + // costs nothing on the paths that exist, and refuses to + // fabricate an instant, grace-free settlement for one that + // doesn't. + None => { + tracing::warn!( + instance_id, + "gone_for_at_least asked about an instance id with no registry entry at \ + all -- neither a live connection nor a restart seed ever created one; \ + answering false rather than settling on an unestablished question" + ); + false + } + } + } +} + +/// Deregisters its connection's slot from a [`LeaderRegistry`] on drop. +/// +/// The existing `active_connections` counter (`server.rs`) gets this +/// "always decremented, however the connection ends" property for free, +/// because its own decrement lives in the SPAWN WRAPPER around the whole +/// connection future (`Server::admit`), which runs regardless of how +/// that future resolves. This bookkeeping instead has to live INSIDE the +/// connection's own future (`connection::handle`), because it needs the +/// negotiated instance id, which the spawn wrapper never sees -- the +/// `initialize` handshake that produces it happens deep inside `handle`, +/// not in `admit`. Living inside the future means every `break`/`return` +/// out of `handle`'s dispatch loop has to reach the decrement, and that +/// is a property of today's control flow, not of the type -- the next +/// early `return` added to that function would otherwise leak a +/// registration silently. A leaked registration means the count never +/// reaches zero, which means this leader is never considered gone: the +/// feature fails CLOSED, silently, and no test would notice unless it +/// happened to exercise that exact exit path. A drop guard makes the +/// invariant structural instead of conventional. +pub(crate) struct LeaderConnectionGuard { + registry: LeaderRegistry, + instance_id: String, + on_gone: Arc, +} + +impl Drop for LeaderConnectionGuard { + fn drop(&mut self) { + if self + .registry + .deregister_reporting_zero_transition(&self.instance_id) + { + (self.on_gone)(self.instance_id.clone()); + } + } +} + +#[cfg(test)] +mod tests { + use std::sync::{Arc, Mutex}; + use std::time::Duration; + + use super::{LeaderConnectionGuard, LeaderRegistry}; + + const GRACE: Duration = Duration::from_millis(20); + + /// Registers with a no-op `on_gone`, for tests that only care about + /// the count/`zero_since` bookkeeping, not the callback. + fn reg(registry: &LeaderRegistry, id: &str) -> LeaderConnectionGuard { + registry.register(id.to_string(), |_| {}) + } + + #[test] + fn a_single_connection_is_not_gone_while_live() { + let registry = LeaderRegistry::new(); + let _guard = reg(®istry, "leader-1"); + assert!(!registry.gone_for_at_least("leader-1", GRACE)); + } + + #[test] + fn dropping_the_only_connection_eventually_reads_gone() { + let registry = LeaderRegistry::new(); + let guard = reg(®istry, "leader-1"); + drop(guard); + assert!( + !registry.gone_for_at_least("leader-1", GRACE), + "not yet -- the window hasn't elapsed" + ); + std::thread::sleep(GRACE * 2); + assert!(registry.gone_for_at_least("leader-1", GRACE)); + } + + /// A reconnect landing before the old connection's own teardown runs: + /// count goes 1 -> 2 -> 1, never touching zero. Never gone. + #[test] + fn a_reconnect_racing_the_previous_teardown_does_not_strand_the_leader() { + let registry = LeaderRegistry::new(); + let old = reg(®istry, "leader-1"); + let new = reg(®istry, "leader-1"); // reconnect arrives first + drop(old); // old connection's teardown lands second + assert!( + !registry.gone_for_at_least("leader-1", GRACE), + "one live connection (`new`) remains" + ); + std::thread::sleep(GRACE * 2); + assert!( + !registry.gone_for_at_least("leader-1", GRACE), + "still live after the window elapses" + ); + drop(new); + } + + /// The race the other direction: the OLD connection's decrement is + /// the one to land second even though the reconnect happened first in + /// wall-clock terms -- still resolves correctly under commutative + /// arithmetic. + #[test] + fn a_reconnect_before_a_delayed_old_teardown_still_leaves_the_leader_live() { + let registry = LeaderRegistry::new(); + let old = reg(®istry, "leader-1"); + let new = reg(®istry, "leader-1"); + // old's teardown is delayed (simulated by ordering the drop last) + drop(old); + assert!(!registry.gone_for_at_least("leader-1", GRACE)); + drop(new); + } + + #[test] + fn a_zero_that_reconnects_and_disconnects_again_within_the_window_resets_the_clock() { + let registry = LeaderRegistry::new(); + let first = reg(®istry, "leader-1"); + drop(first); // 1 -> 0, zero_since = now + std::thread::sleep(GRACE); // the original zero is now old enough... + let second = reg(®istry, "leader-1"); // ...but a reconnect arrives + drop(second); // 1 -> 0 again -- a FRESH zero_since + // The fresh zero has not aged the full window yet, even though + // enough wall-clock time has passed since the ORIGINAL zero that + // a stale timer checking only `count == 0` would have settled. + assert!(!registry.gone_for_at_least("leader-1", GRACE)); + } + + #[test] + fn seeding_a_never_connected_instance_treats_it_as_gone_since_the_given_instant() { + let registry = LeaderRegistry::new(); + let since = std::time::Instant::now() - GRACE * 2; + registry.seed_disconnected_since("leader-1".to_string(), since); + assert!(registry.gone_for_at_least("leader-1", GRACE)); + } + + #[test] + fn seeding_never_clobbers_a_connection_that_already_registered() { + let registry = LeaderRegistry::new(); + let _guard = reg(®istry, "leader-1"); + let since = std::time::Instant::now() - GRACE * 2; + registry.seed_disconnected_since("leader-1".to_string(), since); + assert!( + !registry.gone_for_at_least("leader-1", GRACE), + "a live connection must never be treated as gone" + ); + } + + /// Both real callers (a live connection's `register`, restart's + /// `seed_disconnected_since`) always create the entry before ever + /// asking `gone_for_at_least` -- so an id with no entry at all is + /// only reachable by a future caller's misuse, never by either path + /// that exists today. `false` is the answer that refuses to settle + /// on that unestablished question, rather than fabricating an + /// instant, grace-free "gone" for it. + #[test] + fn an_unregistered_instance_id_never_reads_as_gone() { + let registry = LeaderRegistry::new(); + assert!(!registry.gone_for_at_least("never-seen", GRACE)); + } + + #[test] + fn on_gone_fires_exactly_once_on_the_true_1_to_0_transition() { + let registry = LeaderRegistry::new(); + let fired: Arc>> = Arc::new(Mutex::new(Vec::new())); + let recorded = Arc::clone(&fired); + let a = registry.register("leader-1".to_string(), move |id| { + recorded.lock().unwrap().push(id) + }); + let recorded2 = Arc::clone(&fired); + let b = registry.register("leader-1".to_string(), move |id| { + recorded2.lock().unwrap().push(id) + }); + + drop(a); // 2 -> 1: not zero, on_gone must not fire + assert!( + fired.lock().unwrap().is_empty(), + "a decrement that leaves the count above zero must never fire on_gone" + ); + + drop(b); // 1 -> 0: the true transition + assert_eq!(*fired.lock().unwrap(), vec!["leader-1".to_string()]); + } +} diff --git a/crates/runtime/src/ipc/mod.rs b/crates/runtime/src/ipc/mod.rs index f772a2a5..7d9432de 100644 --- a/crates/runtime/src/ipc/mod.rs +++ b/crates/runtime/src/ipc/mod.rs @@ -14,6 +14,7 @@ //! guarantees that. mod connection; +pub(crate) mod leader_registry; mod server; use std::sync::Arc; @@ -208,6 +209,14 @@ pub struct ServerConfig { /// Effective retention policy for `retention/clean`. The daemon sets /// this from `crew.json`; absent test/embedded servers refuse the RPC. pub retention: Option, + /// How long a run's owning leader may hold no live connection before + /// its non-terminal runs are settled (see + /// [`leader_registry::LEADER_DISCONNECT_GRACE_WINDOW`]'s own doc + /// comment for the production value and its rationale). A field + /// rather than a bare constant reference at each call site so tests + /// can shrink it to a few milliseconds instead of waiting out the + /// real window. + pub leader_disconnect_grace: std::time::Duration, } impl Default for ServerConfig { @@ -229,6 +238,7 @@ impl Default for ServerConfig { activity_clock: None, pane_reopen: None, retention: None, + leader_disconnect_grace: leader_registry::LEADER_DISCONNECT_GRACE_WINDOW, } } } diff --git a/crates/runtime/src/ipc/server.rs b/crates/runtime/src/ipc/server.rs index 72152609..0271f0be 100644 --- a/crates/runtime/src/ipc/server.rs +++ b/crates/runtime/src/ipc/server.rs @@ -28,6 +28,17 @@ pub(crate) struct Shared { /// Number of connections currently admitted and being served. Used to /// decide whether the runtime is idle. pub(crate) active_connections: Arc, + /// Refcounted live-connection registry keyed by leader instance id -- + /// distinct from [`Self::active_connections`], which counts every + /// connection regardless of identity. Populated from inside + /// [`super::connection::handle`], not [`Server::admit`]: the + /// negotiated instance id doesn't exist until the `initialize` + /// handshake completes, deep inside `handle`, so unlike + /// `active_connections` this cannot be bookkept from the spawn + /// wrapper. See `leader_registry`'s own module doc for the refcount + /// rationale and the disconnect-grace-window mechanism this exists + /// for. + pub(crate) leader_registry: super::leader_registry::LeaderRegistry, /// Fired by an in-band `runtime/shutdown` request to trigger a graceful /// shutdown of the accept loop. pub(crate) shutdown: Arc, @@ -286,6 +297,7 @@ impl Server { started_at: Instant::now(), events_tx, active_connections: Arc::new(AtomicUsize::new(0)), + leader_registry: super::leader_registry::LeaderRegistry::new(), shutdown: Arc::new(Notify::new()), orchestration, coordination, @@ -373,6 +385,35 @@ impl Server { Arc::clone(&self.shared.violation) } + /// Routes every orchestration method to the domain repository -- + /// exposed (like [`Server::coordination_broker`]) so the daemon-restart + /// leader-grace-window seeding in `lifecycle.rs` can settle a run + /// through the SAME service instance every live disconnect uses, + /// never a second one. + #[must_use] + pub fn orchestration_service(&self) -> Arc { + Arc::clone(&self.shared.orchestration) + } + + /// The refcounted leader-connection registry -- exposed so + /// `lifecycle.rs` can seed it with every non-terminal run's owner at + /// startup (the daemon-restart case: nothing has connected yet, but a + /// leader may already be gone from this process's own perspective). + #[must_use] + pub(crate) fn leader_registry(&self) -> super::leader_registry::LeaderRegistry { + self.shared.leader_registry.clone() + } + + /// The configured disconnect grace window -- same value the live + /// per-connection teardown in `connection::handle` waits out, so the + /// restart-seeding sweep in `lifecycle.rs` never drifts from it (and + /// tests can shrink it via [`super::ServerConfig::leader_disconnect_grace`] + /// instead of waiting out the real production window). + #[must_use] + pub(crate) fn leader_disconnect_grace(&self) -> std::time::Duration { + self.shared.config.leader_disconnect_grace + } + /// Accepts and serves connections until `shutdown` resolves. /// /// Each accepted connection has its peer credentials read *before* any diff --git a/crates/runtime/src/lifecycle.rs b/crates/runtime/src/lifecycle.rs index 0d392b60..8a524c99 100644 --- a/crates/runtime/src/lifecycle.rs +++ b/crates/runtime/src/lifecycle.rs @@ -526,6 +526,24 @@ pub async fn serve(opts: &ServeOptions) -> Result<(), ServeError> { } }); + // Leader-disconnect grace window, daemon-restart case: at startup no + // connection has registered for ANY instance id yet, so every leader + // that owns a non-terminal run is -- from THIS process's own + // perspective -- disconnected as of right now. Seeding and arming is + // `OrchestrationService::seed_leader_registry_after_restart`'s own + // job (see its doc comment for the full "why the clock starts at + // startup, never at a run's last-seen activity" reasoning) -- + // extracted there, rather than inlined here, so a test can drive the + // restart case directly against a seeded database. + server + .orchestration_service() + .seed_leader_registry_after_restart( + &server.leader_registry(), + server.leader_disconnect_grace(), + Instant::now(), + ) + .await; + // The dashboard is an opt-in, read-only, localhost-only projection; a // bind failure (port taken) degrades to "no dashboard" rather than // failing a daemon that orchestrates fine without it. diff --git a/crates/runtime/src/service/orchestration.rs b/crates/runtime/src/service/orchestration.rs index b7185db1..6a2c7766 100644 --- a/crates/runtime/src/service/orchestration.rs +++ b/crates/runtime/src/service/orchestration.rs @@ -8,6 +8,7 @@ //! evidence, monotonic revision, connected-instance identity). use std::sync::Arc; +use std::time::{Duration, Instant}; use crew_protocol::{ ApprovalId, ApprovalRequest, CrewMethod, DeliveryState, EventEnvelope, IsolationKind, @@ -1529,14 +1530,40 @@ impl OrchestrationService { params: &Value, ) -> Result { let run_id = parse_run_id(params.get("runId"))?; - let project_id = self.project_id; let to = RunState::try_from("cancelled").expect("cancelled is valid"); - let principal_instance_id = principal.instance_id.clone(); + self.cancel_and_settle_run(run_id, to, Some(principal.instance_id.clone())) + .await + } + + /// The mechanical guts shared by two callers with different reasons + /// for the same shape of transition: `run_cancel` (an explicit + /// request, always `to: "cancelled"`, checked against the requesting + /// principal's ownership) and [`Self::settle_leader_gone`] (a + /// system-initiated settlement once a run's owning leader has been + /// gone for the full disconnect grace window, `to: + /// RunState::unrendered_verdict()`, no principal to check -- + /// `principal_instance_id: None` bypasses `transition_run`'s + /// ownership check the same way `settle_abandoned_turn` used to). + /// + /// Transitions and broadcasts first, then best-effort kills the + /// vendor subprocess (`CancelScope::Worker`) -- a kill failure never + /// fails the call (the transition already committed); it sets + /// `degradedControl` instead. Lease/slot release is not this + /// function's job: it happens downstream, evidence-driven, once the + /// adapter's own exit/cancel produces a `ProcessExited`/turn-boundary + /// event (see `adapter::registry::watch_slot`'s own doc comment). + async fn cancel_and_settle_run( + &self, + run_id: RunId, + to: RunState, + principal_instance_id: Option, + ) -> Result { + let project_id = self.project_id; let mut result = self .db .run_domain_op(Box::new(move |conn| { let mut repo = DomainRepository::new(conn, project_id); - repo.transition_run(run_id, &to, Some(&principal_instance_id)) + repo.transition_run(run_id, &to, principal_instance_id.as_deref()) .map(|c| embed_envelope(json!({ "sequence": c.sequence }), &c.envelope)) })) .await @@ -1549,10 +1576,10 @@ impl OrchestrationService { { // A real kill failure (an absent adapter is the clean // `CancelOutcome::NoRunningAdapter`, not an `Err`): the run is - // journaled `cancelled` but a vendor process may still be - // live. Make that visible to `run/get` and the monitor via - // `degradedControl`, mirroring the policy-violation - // path's treatment -- guarded write, journaled, broadcast. + // journaled but a vendor process may still be live. Make that + // visible to `run/get` and the monitor via `degradedControl`, + // mirroring the policy-violation path's treatment -- guarded + // write, journaled, broadcast. tracing::warn!(error = %err, run_id = %run_id, "failed to cancel running adapter subprocess"); match self .db @@ -1574,6 +1601,210 @@ impl OrchestrationService { Ok(result) } + + /// Settles every non-terminal run owned by `instance_id` that is + /// STILL abandoned, once its grace-window timer (`ipc::leader_registry`) + /// has confirmed it held no live connection for the full window. + /// Called from exactly two places, both making that same initial + /// check themselves right before calling this: a live disconnect's + /// own timer (`ipc::connection::handle`'s `on_gone` callback), and + /// the daemon-startup seed for runs whose owner was never reconnected + /// after a restart ([`Self::seed_leader_registry_after_restart`]). + /// + /// "Abandoned" means gone NOW, having been gone the full window -- + /// never merely "was gone for the window at some earlier instant". + /// The caller's own check is a cheap early-out (skips the query below + /// entirely in the common case: nothing owned, or the leader never + /// reconnects) but is not enough on its own -- the query plus two DB + /// round trips per owned run is not instantaneous, and a leader with + /// several parked runs can reconnect partway through this loop. So + /// `leader_registry.gone_for_at_least` is re-checked immediately + /// before EACH run's own settle, inside the loop below, not only + /// once at entry: a leader that comes back mid-loop leaves every run + /// from that point on untouched, exactly as if it had never been + /// gone. `grace` is also used for the journaled fact's `since_ms`, + /// taking the caller's word for how long the window it waited out + /// actually was (the configured value, which a test may have shrunk + /// well below the production + /// [`crate::ipc::leader_registry::LEADER_DISCONNECT_GRACE_WINDOW`]). + /// + /// For each owned run still confirmed abandoned: journals a `WorkerTimeout { kind: LeaderGone }` + /// fact FIRST (the same "report before deciding" shape the inactivity + /// sweep uses for `Inactivity`/`Total`), then transitions to + /// `RunState::unrendered_verdict()` -- never a literal `"cancelled"`, + /// so the one function that owns that word (`run.rs`) is the only + /// place it can ever change. A run that already went terminal between + /// the query and this loop (the leader finished it just in time) is + /// silently skipped: `transition_run`'s own guarded write refuses an + /// illegal edge, and that refusal is exactly correct here, not an + /// error to surface. + pub(crate) async fn settle_leader_gone( + &self, + instance_id: &str, + grace: Duration, + leader_registry: &crate::ipc::leader_registry::LeaderRegistry, + ) { + let project_id = self.project_id; + let run_ids: Vec = match self + .db + .run_domain_op(query::owned_nonterminal_run_ids_op( + instance_id.to_string(), + project_id, + )) + .await + { + Ok(value) => value + .as_array() + .map(|a| { + a.iter() + .filter_map(Value::as_str) + .map(str::to_string) + .collect() + }) + .unwrap_or_default(), + Err(err) => { + tracing::warn!(error = %err, instance_id, "failed to look up runs owned by a gone leader"); + return; + } + }; + + for run_id_str in run_ids { + let Ok(run_id) = RunId::parse(&run_id_str) else { + continue; + }; + + // Re-checked immediately before THIS run's settle, not only + // once by the caller before the query above: the query plus + // two DB round trips per run is not instantaneous, and a + // leader with several parked runs can reconnect partway + // through this loop. "Abandoned" means gone NOW, having been + // gone the full window -- never merely "was gone for the + // window at some earlier instant" the caller happened to + // observe. The caller's own check stays, as the cheap + // early-out that skips this query entirely in the common + // case (nothing owned, or the leader never reconnects); this + // one is what actually protects a leader that comes back + // mid-loop, run by run, rather than only at the start of it. + if !leader_registry.gone_for_at_least(instance_id, grace) { + tracing::info!( + instance_id, + run_id = %run_id, + "leader reconnected mid-settle; leaving this run untouched" + ); + continue; + } + + let mut timeout_fact = match self + .db + .run_domain_op(Box::new(move |conn| { + let mut repo = DomainRepository::new(conn, project_id); + repo.record_worker_timeout_if_live( + run_id, + crew_protocol::TimeoutKind::LeaderGone, + u64::try_from(grace.as_millis()).unwrap_or(u64::MAX), + ) + .map(|committed| { + committed + .map(|c| embed_envelope(json!({ "sequence": c.sequence }), &c.envelope)) + .unwrap_or(Value::Null) + }) + })) + .await + { + Ok(value) => value, + Err(err) => { + tracing::warn!(error = %err, run_id = %run_id, "failed to journal a leaderGone timeout fact"); + Value::Null + } + }; + if !timeout_fact.is_null() { + self.broadcast(&mut timeout_fact); + } + + if let Err(err) = self + .cancel_and_settle_run(run_id, RunState::unrendered_verdict(), None) + .await + { + tracing::warn!( + error = %err.message, + run_id = %run_id, + instance_id, + "failed to settle a run whose leader is gone" + ); + } + } + } + + /// The daemon-restart half of the disconnect-grace-window feature: + /// at startup, nothing has connected to `leader_registry` yet, so + /// every leader that owns a non-terminal run is -- from THIS + /// process's own perspective -- disconnected as of `startup_instant`. + /// Seeds each such owner into the registry and arms the same + /// grace-window timer a live disconnect would, so a leader that + /// reconnects within the window clears it exactly as an ordinary + /// reconnect does (`LeaderRegistry::register`'s own `zero_since = + /// None`), and one that never returns gets its runs settled through + /// [`Self::settle_leader_gone`] -- the same path a live disconnect + /// uses -- rather than being stranded forever, which is the gap this + /// method exists to close. + /// + /// `startup_instant`, never a run's own last-seen activity: a leader + /// cannot reconnect to a daemon that is not running, so any time + /// this daemon spent stopped is the daemon's own absence, not the + /// leader's, and must never be charged against the grace window -- + /// doing so would settle live, reconnecting leaders' runs on every + /// restart that followed a stop longer than the window, which is + /// exactly the destructive direction this feature exists to avoid. + /// See [`crate::ipc::leader_registry::LeaderRegistry::seed_disconnected_since`]'s + /// own doc comment for the same reasoning at the mechanism that + /// enforces it. + /// + /// Extracted from `lifecycle::serve` as its own method (rather than + /// inlined at the one call site) so a test can drive it directly + /// against a seeded database, without spinning up a second full + /// daemon process to exercise a restart. + pub(crate) async fn seed_leader_registry_after_restart( + self: &Arc, + leader_registry: &crate::ipc::leader_registry::LeaderRegistry, + grace: Duration, + startup_instant: Instant, + ) { + let owners: Vec = match self + .db + .run_domain_op(query::distinct_owners_of_nonterminal_runs_op( + self.project_id, + )) + .await + { + Ok(value) => value + .as_array() + .map(|a| { + a.iter() + .filter_map(Value::as_str) + .map(str::to_string) + .collect() + }) + .unwrap_or_default(), + Err(err) => { + tracing::warn!(error = %err, "failed to list owners of non-terminal runs at startup"); + Vec::new() + } + }; + + for owner in owners { + leader_registry.seed_disconnected_since(owner.clone(), startup_instant); + let leader_registry = leader_registry.clone(); + let orchestration = Arc::clone(self); + tokio::spawn(async move { + tokio::time::sleep(grace).await; + if leader_registry.gone_for_at_least(&owner, grace) { + orchestration + .settle_leader_gone(&owner, grace, &leader_registry) + .await; + } + }); + } + } // ------------------------------------------------------- workspace /// Builds a materializer carrying this daemon's configured copy @@ -3491,3 +3722,336 @@ fn parse_message_kind(value: Option<&Value>) -> Result, + run_driver: Option>, + ) -> Arc { + let db = Arc::new(DatabaseHandle::start(db_path).await.expect("db starts")); + let violation = Arc::new(crate::policy::ViolationService::new( + db.clone(), + project_id, + events_tx.clone(), + None, + crate::config::NestedViolationAction::default(), + crate::security::redaction::Redactor::new(), + )); + let lease_service = Arc::new( + crate::workspace::LeaseService::open_in_memory(project_id).expect("lease service"), + ); + Arc::new(OrchestrationService::new( + db, + project_id, + run_driver, + Arc::new(crate::approval::NoopApprovalCallback), + violation, + events_tx, + lease_service, + Arc::new(ArtifactStore::new()), + std::path::PathBuf::new(), + crate::config::crew::Limits::default().turn_budget_per_subtask, + Arc::new(crate::adapter::ActivityClock::new()), + )) + } + + /// Creates a task owned by `owner`, a worker, and a run submitted + /// (and left) at `queued` -- non-terminal, no adapter ever started -- + /// against `orchestration`'s own database. Returns the run id. + async fn seed_nonterminal_run(orchestration: &OrchestrationService, owner: &str) -> RunId { + let project_id = orchestration.project_id; + let db = Arc::clone(&orchestration.db); + let owner = owner.to_string(); + db.run_domain_op(Box::new(move |conn| { + let mut repo = DomainRepository::new(conn, project_id); + let task_id = TaskId::new(); + repo.upsert_task( + task_id, + &TaskRef { + owner_client_instance_id: owner.clone(), + revision: 1, + }, + )?; + let worker_id = WorkerId::new(); + repo.create_worker(&Worker { + worker_id, + profile_ref: WorkerProfileRef { + id: worker_id, + fingerprint: "sha256:fake".into(), + adapter: "fake".into(), + model: "test".into(), + permission_envelope: serde_json::json!({}), + }, + parent_worker_id: None, + created_at: Timestamp::now(), + })?; + let run_id = RunId::new(); + repo.submit_run( + &Run { + run_id, + task_id, + worker_id, + state: RunState::try_from("queued").expect("queued is valid"), + flags: RunFlags::default(), + vendor_session_id: None, + started_at: None, + completed_at: None, + }, + None, + None, + )?; + Ok(serde_json::json!(run_id.to_string())) + })) + .await + .expect("seed run") + .as_str() + .and_then(|s| RunId::parse(s).ok()) + .expect("run id round-trips") + } + + async fn run_state(orchestration: &OrchestrationService, run_id: RunId) -> String { + orchestration + .db + .run_domain_op(query::run_get_op(run_id)) + .await + .expect("run_get")["state"] + .as_str() + .expect("state is a string") + .to_string() + } + + #[tokio::test] + async fn a_restart_settles_a_run_whose_leader_never_reconnects_after_the_window() { + let dir = tempfile::tempdir().expect("tempdir"); + let project_id = ProjectId::new(); + let (events_tx, mut events_rx) = broadcast::channel(16); + let orchestration = + build_orchestration(&dir.path().join("db.sqlite"), project_id, events_tx, None).await; + let run_id = seed_nonterminal_run(&orchestration, "leader-gone").await; + + let leader_registry = LeaderRegistry::new(); + let grace = Duration::from_millis(20); + // Nothing has connected to `leader_registry` since it was just + // constructed -- exactly the daemon-restart precondition: an + // empty in-memory registry sitting on top of a database that + // still has this run's owner recorded. + orchestration + .seed_leader_registry_after_restart(&leader_registry, grace, Instant::now()) + .await; + + // Past the window, with margin: the leader never reconnects. + tokio::time::sleep(grace * 5).await; + + assert_eq!( + run_state(&orchestration, run_id).await, + "cancelled", + "a leader that never returns after a restart must be settled via unrendered_verdict()" + ); + + let fact = events_rx.try_recv().expect("a fact was broadcast"); + match fact.event { + crew_protocol::RuntimeEvent::WorkerTimeout { + run_id: fact_run_id, + kind, + .. + } => { + assert_eq!(fact_run_id, run_id); + assert_eq!(kind, crew_protocol::TimeoutKind::LeaderGone); + } + other => panic!("expected a WorkerTimeout{{kind: LeaderGone}} fact, got {other:?}"), + } + } + + #[tokio::test] + async fn a_restart_leaves_a_run_untouched_when_its_leader_reconnects_within_the_window() { + let dir = tempfile::tempdir().expect("tempdir"); + let project_id = ProjectId::new(); + let (events_tx, mut events_rx) = broadcast::channel(16); + let orchestration = + build_orchestration(&dir.path().join("db.sqlite"), project_id, events_tx, None).await; + let run_id = seed_nonterminal_run(&orchestration, "leader-returns").await; + + let leader_registry = LeaderRegistry::new(); + let grace = Duration::from_millis(60); + orchestration + .seed_leader_registry_after_restart(&leader_registry, grace, Instant::now()) + .await; + + // The leader reconnects well within the window and stays connected + // (the guard is held for the rest of the test) -- an ordinary + // reconnect, which must clear `zero_since` exactly as a live + // disconnect's own reconnect does. + let _reconnected = leader_registry.register("leader-returns".to_string(), |_| {}); + + tokio::time::sleep(grace * 3).await; + + assert_eq!( + run_state(&orchestration, run_id).await, + "queued", + "a leader that reconnects within the window must leave its runs untouched" + ); + assert!( + events_rx.try_recv().is_err(), + "a reconnect within the window must journal nothing" + ); + } + + /// A [`RunDriver`] whose `cancel_run` blocks on its FIRST call until + /// the test releases it, notifying the test the instant that call + /// starts. Gives a test deterministic control over exactly when + /// `settle_leader_gone`'s loop has finished processing one run -- + /// `cancel_and_settle_run` already commits the run's own + /// `transition_run` (and broadcasts it) *before* calling + /// `cancel_run`, so "the first `cancel_run` call has started" is + /// proof the first run is already settled in the database, without + /// guessing at timing. + struct BlockingOnFirstCancelDriver { + first_call_started: Arc, + release_first_call: Arc, + calls: std::sync::atomic::AtomicUsize, + } + + impl RunDriver for BlockingOnFirstCancelDriver { + fn active_run_count(&self) -> usize { + 0 + } + + fn start( + &self, + _ctx: crate::service::RunDriverContext, + ) -> crate::service::AdapterFuture<'static, Result<(), String>> { + Box::pin(async { Ok(()) }) + } + + fn send_follow_up( + &self, + _run_id: RunId, + _task_id: TaskId, + _worker_id: WorkerId, + _prompt: String, + _kind: crew_protocol::MessageKind, + ) -> crate::service::AdapterFuture<'static, Result<(), String>> { + Box::pin(async { Ok(()) }) + } + + fn running_adapter(&self, _run_id: RunId) -> Option> { + None + } + + fn cancel_run( + &self, + _run_id: RunId, + _scope: CancelScope, + ) -> crate::service::AdapterFuture<'static, Result> + { + let first_call_started = Arc::clone(&self.first_call_started); + let release_first_call = Arc::clone(&self.release_first_call); + let is_first = self.calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst) == 0; + Box::pin(async move { + if is_first { + first_call_started.notify_one(); + release_first_call.notified().await; + } + Ok(crate::service::CancelOutcome::Cancelled) + }) + } + } + + /// The re-check staff's review named directly: `settle_leader_gone`'s + /// caller-side check (before the DB query for owned runs) is not + /// enough on its own for a leader with more than one parked run, + /// because the query plus per-run DB round trips are not + /// instantaneous. This proves the loop's OWN re-check (immediately + /// before each run's settle) actually stops the run reached AFTER a + /// reconnect, using a blocking driver rather than a guessed sleep to + /// land the reconnect deterministically between the two runs' + /// settles -- never racing on timing. + #[tokio::test] + async fn a_leader_reconnecting_mid_settle_leaves_the_later_run_untouched() { + let dir = tempfile::tempdir().expect("tempdir"); + let project_id = ProjectId::new(); + let (events_tx, _events_rx) = broadcast::channel(16); + let driver = Arc::new(BlockingOnFirstCancelDriver { + first_call_started: Arc::new(tokio::sync::Notify::new()), + release_first_call: Arc::new(tokio::sync::Notify::new()), + calls: std::sync::atomic::AtomicUsize::new(0), + }); + let orchestration = build_orchestration( + &dir.path().join("db.sqlite"), + project_id, + events_tx, + Some(Arc::clone(&driver) as Arc), + ) + .await; + + let first_run = seed_nonterminal_run(&orchestration, "leader-mid").await; + let second_run = seed_nonterminal_run(&orchestration, "leader-mid").await; + + let leader_registry = LeaderRegistry::new(); + let grace = Duration::from_millis(20); + // Already gone for the full window from the start -- the + // caller-side check this loop's own re-check exists alongside. + leader_registry.seed_disconnected_since("leader-mid".to_string(), Instant::now() - grace); + + let orch = Arc::clone(&orchestration); + let registry_for_task = leader_registry.clone(); + let settle_task = tokio::spawn(async move { + orch.settle_leader_gone("leader-mid", grace, ®istry_for_task) + .await; + }); + + // Waits for proof the first run is already settled (its + // `cancel_run` call has started, which only happens after its + // own `transition_run` commit), not a guessed duration. + driver.first_call_started.notified().await; + + // The leader reconnects here -- strictly between the first run's + // settle (already committed) and the second's (not yet reached: + // the loop is still blocked inside the first run's `cancel_run`). + let _reconnected = leader_registry.register("leader-mid".to_string(), |_| {}); + driver.release_first_call.notify_one(); + + settle_task.await.expect("settle_leader_gone task"); + + // `owned_nonterminal_run_ids_op` carries no `ORDER BY`, so which + // of the two rows the loop reaches first is not a guarantee this + // test should depend on -- only that exactly one of them was + // already settled when the reconnect landed (the one whose + // `cancel_run` blocked the loop) and the other was reached only + // after, and so must survive. + let first_state = run_state(&orchestration, first_run).await; + let second_state = run_state(&orchestration, second_run).await; + let states = [first_state.as_str(), second_state.as_str()]; + assert_eq!( + states.iter().filter(|s| **s == "cancelled").count(), + 1, + "exactly the run reached before the reconnect must be settled: {states:?}" + ); + assert_eq!( + states.iter().filter(|s| **s == "queued").count(), + 1, + "exactly the run reached after the reconnect must be left untouched: {states:?}" + ); + } +} diff --git a/crates/runtime/src/service/query.rs b/crates/runtime/src/service/query.rs index a35880c1..242e87f1 100644 --- a/crates/runtime/src/service/query.rs +++ b/crates/runtime/src/service/query.rs @@ -404,6 +404,65 @@ pub fn owned_run_ids_op( }) } +/// The run states [`crew_protocol::RunState::is_terminal`] considers +/// terminal, duplicated here because this is raw SQL rather than a Rust +/// match -- kept as one named constant, not inlined at each use, so the +/// two queries below (and any future one) cannot drift from each other +/// even if they drift from `is_terminal` itself. If `is_terminal`'s own +/// set ever changes, this is the one place to update. +const TERMINAL_STATES_SQL: &str = "'succeeded','failed','cancelled','lost'"; + +/// Non-terminal run ids owned (via `tasks.owner_client_instance_id`) by +/// `owner_instance_id` -- the leader-disconnect-grace-window teardown's +/// own query: once a leader has been gone for the grace window, these +/// are exactly the runs it left behind that still need settling. Unlike +/// [`owned_run_ids_op`], this filters to non-terminal runs directly in +/// SQL rather than leaving the caller to filter afterward -- the caller +/// here (`settle_leader_gone`) has no other use for a terminal run's id. +pub fn owned_nonterminal_run_ids_op( + owner_instance_id: String, + project_id: ProjectId, +) -> DomainClosure { + Box::new(move |conn| { + let sql = format!( + "SELECT r.run_id FROM runs r JOIN tasks t ON r.task_id = t.task_id \ + WHERE t.project_id = ?1 AND t.owner_client_instance_id = ?2 \ + AND r.state NOT IN ({TERMINAL_STATES_SQL})" + ); + let mut stmt = conn.prepare(&sql)?; + let ids: Vec = stmt + .query_map( + rusqlite::params![project_id.to_string(), owner_instance_id], + |row| row.get(0), + )? + .filter_map(|r| r.ok()) + .collect(); + Ok(json!(ids)) + }) +} + +/// Every DISTINCT owner instance id with at least one non-terminal run, +/// project-wide -- the daemon-restart seed: at startup nothing has +/// connected yet, so every one of these owners is, from this process's +/// own perspective, disconnected as of right now (see +/// `ipc::leader_registry::LeaderRegistry::seed_disconnected_since`'s own +/// doc comment for why the clock starts at startup, not at the run's +/// last-seen time). +pub fn distinct_owners_of_nonterminal_runs_op(project_id: ProjectId) -> DomainClosure { + Box::new(move |conn| { + let sql = format!( + "SELECT DISTINCT t.owner_client_instance_id FROM runs r JOIN tasks t ON r.task_id = t.task_id \ + WHERE t.project_id = ?1 AND r.state NOT IN ({TERMINAL_STATES_SQL})" + ); + let mut stmt = conn.prepare(&sql)?; + let owners: Vec = stmt + .query_map(rusqlite::params![project_id.to_string()], |row| row.get(0))? + .filter_map(|r| r.ok()) + .collect(); + Ok(json!(owners)) + }) +} + /// Confirms `expected_instance_id` currently owns the task that owns /// `run_id`, for `workspace/acquire`'s ownership arbitration. /// diff --git a/crates/runtime/src/timeout_sweep.rs b/crates/runtime/src/timeout_sweep.rs index d8665798..d8b749b7 100644 --- a/crates/runtime/src/timeout_sweep.rs +++ b/crates/runtime/src/timeout_sweep.rs @@ -13,6 +13,17 @@ //! Every journal decision re-checks liveness inside the database actor //! closure (`record_worker_timeout_if_live`), so a run that settled //! between the clock snapshot and the write never receives a timeout fact. +//! +//! This sweep never transitions a run's state on its own -- a parked run +//! whose leader is still connected is never settled here, however long +//! it stays silent. It used to: an inactivity backstop settled a +//! leader-abandoned turn to `lost` (ADR-0027 wave 3). The maintainer +//! ruled that out -- silence alone is never evidence the leader gave up, +//! only the leader's connection actually being gone is -- so settlement +//! now happens exclusively through `crate::ipc::leader_registry`'s +//! disconnect grace window (see ADR-0036). `lost` keeps its narrower, +//! original meaning (ADR-0023): the evidence itself is missing or +//! ambiguous, not "the leader went quiet." use std::time::{Duration, Instant}; @@ -51,6 +62,28 @@ pub async fn sweep_once( let since_ms = match kind { TimeoutKind::Inactivity => millis_since(activity.last_activity, now), TimeoutKind::Total => millis_since(activity.started_at, now), + // `due_timeouts` (the only producer of `kind` here) only + // ever returns `Inactivity`/`Total` -- it has no notion of + // a leader's connection state at all. `LeaderGone` is + // journaled from the leader-grace-window teardown + // directly (`service::orchestration::settle_leader_gone`), + // never through this worker-liveness sweep. This arm + // should be unreachable on that invariant, but this loop + // runs inside a long-lived background task, not a test -- + // panicking here would silently take the whole sweep down + // rather than fail loudly where someone would notice. Log + // and skip this one kind instead, so a future producer + // added here in error degrades to a missed fact, not a + // dead daemon. + TimeoutKind::LeaderGone => { + tracing::error!( + run_id = %run_id, + "due_timeouts produced LeaderGone, which it should never be able to \ + observe -- skipping this kind rather than journaling a nonsensical \ + since_ms" + ); + continue; + } }; let outcome = db .run_domain_op(Box::new({ @@ -81,19 +114,20 @@ pub async fn sweep_once( // Mark only after a successful commit: a failed append // leaves the flag unset so the next tick retries. clock.mark_journaled(&run_id, kind); - // ADR-0027 wave 3: the inactivity backstop. A TUI run - // parked by a finished turn that the leader never - // closed is settled here rather than sitting forever. - // Deliberately narrow -- `settle_abandoned_turn` - // requires BOTH `waitingUser` and `turnSettled`, so a - // question wait, an approval wait, `waitingPeer` and - // `paused` are all left alone, and the runtime's - // "journal, never decide" stance still holds for - // everything except the one case the leader provably - // walked away from. - if matches!(kind, TimeoutKind::Inactivity) { - settle_abandoned_turn(db, project_id, events_tx, clock, run_id).await; - } + // A parked run whose leader is still alive (connected, + // regardless of how long it stays silent) is never + // settled here -- only the fact above is journaled. + // Settlement now happens exclusively through the + // leader-disconnect grace window + // (`ipc::leader_registry`, + // `OrchestrationService::settle_leader_gone`): the + // inactivity backstop that used to settle a + // leader-abandoned turn to `lost` after this same + // `Inactivity` fact (ADR-0027 wave 3) was removed + // once the maintainer ruled that the runtime must + // never decide a leader gave up based on silence + // alone -- only on its connection actually being + // gone. See ADR-0036. } Ok(_) => { // The run is terminal or gone: drop its clock so this @@ -112,76 +146,3 @@ pub async fn sweep_once( } } } - -/// Settles a run parked by a finished turn the leader never closed, and -/// clears the marker that described that pause. -/// -/// Split out of [`sweep_once`] so the "did anything actually need -/// settling?" decision lives next to its own commit-and-broadcast, and so -/// the sweep's per-timeout loop stays readable. Every failure is logged and -/// dropped: the timeout fact itself is already durable, and the next tick -/// re-attempts anything that did not land. -async fn settle_abandoned_turn( - db: &DatabaseHandle, - project_id: ProjectId, - events_tx: &broadcast::Sender, - clock: &ActivityClock, - run_id: crew_protocol::RunId, -) { - let settled = db - .run_domain_op(Box::new(move |conn| { - Ok(DomainRepository::new(conn, project_id) - .settle_abandoned_turn(run_id)? - .map(|committed| { - embed_envelope( - json!({ "sequence": committed.sequence }), - &committed.envelope, - ) - }) - .unwrap_or(serde_json::Value::Null)) - })) - .await; - match settled { - Ok(mut value) if !value.is_null() => { - broadcast_committed(events_tx, &mut value); - tracing::info!( - run_id = %run_id, - "settled an abandoned turn as lost after inactivity" - ); - // The run is terminal now, so its clock has nothing left to - // measure. - clock.forget(&run_id); - // The marker described a pause that no longer exists; leaving - // it set would tell `run/get` an answer is still waiting to be - // collected on a run that has ended. - match db - .run_domain_op(Box::new(move |conn| { - DomainRepository::new(conn, project_id) - .set_run_flag(run_id, crate::domain::RunFlag::TurnSettled, false) - .map(|committed| { - embed_envelope( - json!({ "sequence": committed.sequence }), - &committed.envelope, - ) - }) - })) - .await - { - Ok(mut cleared) => { - broadcast_committed(events_tx, &mut cleared); - } - Err(err) => tracing::warn!( - run_id = %run_id, - error = %err, - "failed to clear turnSettled after settling an abandoned turn" - ), - } - } - Ok(_) => {} - Err(err) => tracing::warn!( - run_id = %run_id, - error = %err, - "failed to settle an abandoned turn" - ), - } -} diff --git a/crates/runtime/tests/orchestration_rpc.rs b/crates/runtime/tests/orchestration_rpc.rs index 18c84f46..eac615f6 100644 --- a/crates/runtime/tests/orchestration_rpc.rs +++ b/crates/runtime/tests/orchestration_rpc.rs @@ -26,7 +26,7 @@ use std::time::Duration; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::net::UnixStream; use tokio::net::unix::OwnedWriteHalf; -use tokio::sync::oneshot; +use tokio::sync::{broadcast, oneshot}; use tokio::task::JoinHandle; #[path = "support/spawn_evidence_adapter.rs"] @@ -5929,14 +5929,16 @@ impl RunDriver for SettlingRunDriver { } } -/// Submits a run against `adapter: "fake"`, returning its id. Mirrors the -/// other suites' helper; ids start at 2 because `initialize` consumed 1. -async fn submit_fake_run(client: &mut Client) -> String { +/// Submits a run against `adapter: "fake"`, owned by `owner` (which must +/// match `client`'s own connected instance id -- `task/upsert` refuses a +/// mismatch), returning its id. Mirrors the other suites' helper; ids +/// start at 2 because `initialize` consumed 1. +async fn submit_fake_run(client: &mut Client, owner: &str) -> String { let task = client .call( 2, "task/upsert", - json!({ "ownerClientInstanceId": "omp-1", "revision": 1 }), + json!({ "ownerClientInstanceId": owner, "revision": 1 }), ) .await; let task_id = task["result"]["taskId"].as_str().unwrap().to_string(); @@ -5984,7 +5986,7 @@ async fn run_finish_settles_a_settled_turn_as_succeeded_and_tears_down_the_worke }) .await; let mut client = omp_client(&harness, "omp-1").await; - let run_id = submit_fake_run(&mut client).await; + let run_id = submit_fake_run(&mut client, "omp-1").await; assert!( wait_for_state(&mut client, 5, &run_id, "waitingUser").await, "the seeded turn boundary must park the run first" @@ -6034,7 +6036,7 @@ async fn run_finish_can_settle_a_run_as_failed_when_the_leader_says_so() { }) .await; let mut client = omp_client(&harness, "omp-1").await; - let run_id = submit_fake_run(&mut client).await; + let run_id = submit_fake_run(&mut client, "omp-1").await; assert!(wait_for_state(&mut client, 5, &run_id, "waitingUser").await); let finish = client @@ -6060,7 +6062,7 @@ async fn run_finish_refuses_an_already_terminal_run() { }) .await; let mut client = omp_client(&harness, "omp-1").await; - let run_id = submit_fake_run(&mut client).await; + let run_id = submit_fake_run(&mut client, "omp-1").await; assert!(wait_for_state(&mut client, 5, &run_id, "waitingUser").await); let first = client @@ -6092,7 +6094,7 @@ async fn run_finish_refuses_an_unknown_outcome() { }) .await; let mut client = omp_client(&harness, "omp-1").await; - let run_id = submit_fake_run(&mut client).await; + let run_id = submit_fake_run(&mut client, "omp-1").await; assert!(wait_for_state(&mut client, 5, &run_id, "waitingUser").await); let finish = client @@ -6450,3 +6452,273 @@ async fn a_child_denial_reason_is_redacted_before_it_is_journaled() { .expect("a childWorkerRequestDenied event"); assert_reason_masked(denied); } + +// -------------------------------------------- leader-disconnect grace window + +/// The `workerTimeout` `kind` tag values journaled for `run_id` in a replay +/// response, in emission order. Mirrors `workspace_event_kinds_for_run`'s +/// shape for the sibling `workerTimeout` event type. +fn worker_timeout_kinds_for_run(replay: &Value, run_id: &str) -> Vec { + replay["result"] + .as_array() + .expect("events/replay returns an array") + .iter() + .map(|e| &e["event"]) + .filter(|e| e["type"] == "workerTimeout" && e["payload"]["runId"] == run_id) + .map(|e| e["payload"]["kind"].as_str().unwrap().to_string()) + .collect() +} + +/// A leader that disconnects entirely (drops its only connection) and never +/// comes back must have its parked run's vendor session torn down once the +/// grace window elapses -- the same `cancel_run(Worker)` call `run/cancel` +/// itself makes. +#[tokio::test] +async fn a_leader_disconnecting_stops_its_runs_workers() { + let driver = Arc::new(SettlingRunDriver::default()); + let harness = Harness::start(|c| { + c.run_driver = Some(Arc::clone(&driver) as Arc); + c.leader_disconnect_grace = Duration::from_millis(80); + }) + .await; + let mut client = omp_client(&harness, "leader-1").await; + let run_id = submit_fake_run(&mut client, "leader-1").await; + assert!( + wait_for_state(&mut client, 5, &run_id, "waitingUser").await, + "the seeded turn boundary must park the run first" + ); + + drop(client); // the leader disconnects entirely; nobody reconnects + + // Past the grace window, with margin for the timer's own wakeup. + tokio::time::sleep(Duration::from_millis(500)).await; + + let calls = driver.cancel_calls(); + assert_eq!( + calls.len(), + 1, + "the disconnect teardown must cancel the parked run's worker: {calls:?}" + ); + assert_eq!(calls[0].1, CancelScope::Worker); +} + +/// The same disconnect settles the run's own state through +/// `OrchestrationService::cancel_and_settle_run` -- the exact helper +/// `run/cancel` uses -- landing on `RunState::unrendered_verdict()` +/// (`"cancelled"`), never a literal decided at this call site. +#[tokio::test] +async fn a_disconnect_settles_owned_runs_via_the_shared_helper() { + let driver = Arc::new(SettlingRunDriver::default()); + let harness = Harness::start(|c| { + c.run_driver = Some(Arc::clone(&driver) as Arc); + c.leader_disconnect_grace = Duration::from_millis(80); + }) + .await; + let mut client = omp_client(&harness, "leader-2").await; + let run_id = submit_fake_run(&mut client, "leader-2").await; + assert!(wait_for_state(&mut client, 5, &run_id, "waitingUser").await); + + drop(client); + tokio::time::sleep(Duration::from_millis(500)).await; + + // `run/get` is not ownership-scoped -- a fresh, unrelated connection + // can read the settled state back. + let mut checker = omp_client(&harness, "checker").await; + let get = checker.call(2, "run/get", json!({ "runId": run_id })).await; + assert_eq!( + get["result"]["state"], "cancelled", + "the shared cancel_and_settle_run helper must settle the run: {get:?}" + ); +} + +/// A leader that disconnects and reconnects (same instance id) well within +/// the grace window must never have its run touched, and the grace-window +/// teardown must journal nothing -- an ordinary reconnect is indistinguishable +/// from a leader that was never gone. +#[tokio::test] +async fn a_reconnect_within_the_window_leaves_the_run_untouched_and_journals_nothing() { + let driver = Arc::new(SettlingRunDriver::default()); + let harness = Harness::start(|c| { + c.run_driver = Some(Arc::clone(&driver) as Arc); + c.leader_disconnect_grace = Duration::from_millis(150); + }) + .await; + let mut client = omp_client(&harness, "leader-3").await; + let run_id = submit_fake_run(&mut client, "leader-3").await; + assert!(wait_for_state(&mut client, 5, &run_id, "waitingUser").await); + + drop(client); // disconnects... + tokio::time::sleep(Duration::from_millis(30)).await; // ...well within the window... + let mut reconnected = omp_client(&harness, "leader-3").await; // ...and reconnects. + + // Past the ORIGINAL window, with margin. + tokio::time::sleep(Duration::from_millis(400)).await; + + assert!( + driver.cancel_calls().is_empty(), + "a reconnect within the window must never tear the worker down: {:?}", + driver.cancel_calls() + ); + + let get = reconnected + .call(5, "run/get", json!({ "runId": run_id })) + .await; + assert_eq!( + get["result"]["state"], "waitingUser", + "a reconnect within the window must leave the run untouched: {get:?}" + ); + + let replay = reconnected + .call(6, "events/replay", json!({ "afterSequence": 0 })) + .await; + assert!( + worker_timeout_kinds_for_run(&replay, &run_id) + .iter() + .all(|k| k != "leaderGone"), + "a reconnect within the window must journal nothing: {replay:?}" + ); +} + +/// A run owned by one leader is untouched by a different leader's +/// connection lifecycle entirely -- disconnecting `leader-4` must never +/// affect a run owned by `leader-5`, still live. +#[tokio::test] +async fn a_run_owned_by_another_live_client_is_not_touched() { + let driver = Arc::new(SettlingRunDriver::default()); + let harness = Harness::start(|c| { + c.run_driver = Some(Arc::clone(&driver) as Arc); + c.leader_disconnect_grace = Duration::from_millis(80); + }) + .await; + + let mut departing = omp_client(&harness, "leader-4").await; + let departing_run = submit_fake_run(&mut departing, "leader-4").await; + assert!(wait_for_state(&mut departing, 5, &departing_run, "waitingUser").await); + + let mut staying = omp_client(&harness, "leader-5").await; + let staying_run = submit_fake_run(&mut staying, "leader-5").await; + assert!(wait_for_state(&mut staying, 5, &staying_run, "waitingUser").await); + + drop(departing); // only leader-4 goes away + + tokio::time::sleep(Duration::from_millis(500)).await; + + let calls = driver.cancel_calls(); + assert_eq!( + calls + .iter() + .filter(|(id, _)| id.to_string() == departing_run) + .count(), + 1, + "the departed leader's run must be torn down: {calls:?}" + ); + assert_eq!( + calls + .iter() + .filter(|(id, _)| id.to_string() == staying_run) + .count(), + 0, + "a run owned by a still-connected leader must never be touched: {calls:?}" + ); + + let get = staying + .call(6, "run/get", json!({ "runId": staying_run })) + .await; + assert_eq!( + get["result"]["state"], "waitingUser", + "the still-connected leader's own run must be untouched: {get:?}" + ); +} + +/// A parked run whose leader is still connected must never be settled by +/// the inactivity sweep alone, however long it stays quiet -- only the +/// leader's connection actually being gone (`ipc::leader_registry`) may +/// settle it. This is the maintainer's ruling that removed +/// `settle_abandoned_turn` (see `timeout_sweep`'s own module doc). +#[tokio::test] +async fn a_parked_run_whose_leader_is_alive_is_never_settled_by_inactivity() { + let driver = Arc::new(SettlingRunDriver::default()); + let harness = Harness::start(|c| { + c.run_driver = Some(Arc::clone(&driver) as Arc); + }) + .await; + let mut client = omp_client(&harness, "leader-6").await; + let run_id_str = submit_fake_run(&mut client, "leader-6").await; + assert!(wait_for_state(&mut client, 5, &run_id_str, "waitingUser").await); + // `client` is never dropped: the leader stays connected for the whole test. + + let run_id = RunId::parse(&run_id_str).expect("run id round-trips"); + let clock = ActivityClock::new(); + // Stale activity: comfortably past even a millisecond-scale threshold. + clock.touch(&run_id, std::time::Instant::now() - Duration::from_secs(10)); + + let sweep_db = DatabaseHandle::start(harness.database.clone()) + .await + .expect("a second reader/writer handle onto the same WAL-mode database"); + let (events_tx, _events_rx) = broadcast::channel(16); + crew_runtime::timeout_sweep::sweep_once( + &sweep_db, + harness.project_id, + &events_tx, + &clock, + Duration::from_millis(1), + Duration::from_secs(999_999), + ) + .await; + + let get = client + .call(6, "run/get", json!({ "runId": run_id_str })) + .await; + assert_eq!( + get["result"]["state"], "waitingUser", + "inactivity alone must never settle a run whose leader is still connected: {get:?}" + ); + assert!( + driver.cancel_calls().is_empty(), + "no teardown may fire from inactivity alone: {:?}", + driver.cancel_calls() + ); +} + +/// The inactivity fact itself is unaffected by `settle_abandoned_turn`'s +/// removal: the sweep still journals `workerTimeout { kind: inactivity }` +/// exactly as before, for `run/timeoutAck` to react to -- only the +/// runtime-side settlement that used to follow it is gone. +#[tokio::test] +async fn an_inactivity_timeout_still_journals_the_fact() { + let driver = Arc::new(SettlingRunDriver::default()); + let harness = Harness::start(|c| { + c.run_driver = Some(Arc::clone(&driver) as Arc); + }) + .await; + let mut client = omp_client(&harness, "leader-7").await; + let run_id_str = submit_fake_run(&mut client, "leader-7").await; + assert!(wait_for_state(&mut client, 5, &run_id_str, "waitingUser").await); + + let run_id = RunId::parse(&run_id_str).expect("run id round-trips"); + let clock = ActivityClock::new(); + clock.touch(&run_id, std::time::Instant::now() - Duration::from_secs(10)); + + let sweep_db = DatabaseHandle::start(harness.database.clone()) + .await + .expect("a second reader/writer handle onto the same WAL-mode database"); + let (events_tx, _events_rx) = broadcast::channel(16); + crew_runtime::timeout_sweep::sweep_once( + &sweep_db, + harness.project_id, + &events_tx, + &clock, + Duration::from_millis(1), + Duration::from_secs(999_999), + ) + .await; + + let replay = client + .call(6, "events/replay", json!({ "afterSequence": 0 })) + .await; + assert_eq!( + worker_timeout_kinds_for_run(&replay, &run_id_str), + vec!["inactivity".to_string()], + "the inactivity fact must still be journaled: {replay:?}" + ); +} diff --git a/docs/adr/0027-turn-end-settles-a-run.md b/docs/adr/0027-turn-end-settles-a-run.md index ac4c72bd..c5c3ee89 100644 --- a/docs/adr/0027-turn-end-settles-a-run.md +++ b/docs/adr/0027-turn-end-settles-a-run.md @@ -1,6 +1,6 @@ # A run is a conversation the leader closes; a vendor's turn-end is durable evidence, not a terminal state -* Status: Accepted +* Status: Accepted; Decision point 3's backstop-settles-on-silence half superseded by [0036](0036-leader-disconnect-grace-window.md) -- its "never `succeeded`" half stands, unamended. * Date: 2026-08-29 * Amends: [0023](0023-run-state-edges-from-adapter-evidence.md) @@ -84,8 +84,10 @@ current bug under a nicer name — a leader that dies or loses interest leaves r Turn-end is journaled and broadcast as evidence, driving `working -> waitingUser` — an already-legal, already-modelled non-terminal state meaning "the runtime is not the blocker", which `next_hop` already routes through. `run/result` becomes readable for a run in `waitingUser` that has -at least one turn-end. The leader settles explicitly via `run/finish`; an inactivity backstop settles -an abandoned run without one. +at least one turn-end. The leader settles explicitly via `run/finish`; an inactivity backstop settled +an abandoned run without one until [0036](0036-leader-disconnect-grace-window.md) replaced that +backstop with a disconnect-grace-window teardown -- silence alone was never sufficient evidence that +a leader had given up, only its connection actually being gone is. Against it: it is two-phase rather than one change, and it borrows `waitingUser`, whose existing meaning in the monitor is "the worker asked a question" — so the two need to be distinguishable. diff --git a/docs/adr/0036-leader-disconnect-grace-window.md b/docs/adr/0036-leader-disconnect-grace-window.md new file mode 100644 index 00000000..f2221554 --- /dev/null +++ b/docs/adr/0036-leader-disconnect-grace-window.md @@ -0,0 +1,160 @@ +# A parked run is settled on its leader's connection actually being gone, never on silence alone + +* Status: Accepted +* Date: 2026-09-09 +* Amends: [0027](0027-turn-end-settles-a-run.md) + +## Context and Problem Statement + +ADR-0027's wave 3 gave the inactivity sweep a backstop: a run parked in `waitingUser` with +`turnSettled` set, quiet past the inactivity deadline, was settled to `lost`. That was written when +the only failure this daemon could observe was silence, and it conflated two genuinely different +situations under one signal: a worker gone quiet while its leader is still there, still watching, +still free to steer or finish it; and a leader that has itself vanished, leaving a parked run nobody +will ever finish or retry. Silence looks identical from the runtime's side in both cases, but only +the second one is actually abandoned. + +The maintainer ruled the backstop out on exactly that basis: silence alone is never evidence a leader +gave up, only the leader's own connection actually being gone is. Acting on that ruling required a +signal this daemon did not yet track — leader *connection* identity as its own liveness concern, +independent of the worker/vendor process activity `ActivityClock` already measures. It also surfaced +a second, un-mentioned failure mode once the first was fixed: an in-memory notion of "who is +connected" starts empty on every daemon restart regardless of what was true a moment before, so +naively treating "no live connection recorded yet" as "gone" would settle every parked run on every +restart — a regression far worse than the backstop this decision removes. + +## Decision Drivers + +* Never settle a run because it was quiet; settle it only because its owning leader is provably gone. +* Never strand a run forever either — an abandoned leader is not something a human will notice and + cancel by hand. +* Reuse `run_cancel`'s existing settle mechanism rather than invent a second one: the ordering and + broadcast invariants ADR-0020 requires were already solved once, and a second, parallel + implementation is a second place they could quietly diverge. +* A daemon restart is not a leader event. The daemon's own downtime must never be charged against a + leader's grace window, or a long-stopped daemon would settle every reconnecting leader's runs the + moment it came back up. +* More than one physical connection can legitimately share one leader instance id (the + `crew-extension` fallback id, used by any leader session with no real session id of its own), so + correctness cannot assume one instance id maps to one connection. + +## Considered Options + +* **Keep the inactivity backstop.** Rejected outright by the maintainer's ruling above — it conflates + worker silence with leader abandonment, which are not the same fact. +* **Track leader liveness as a boolean or a set of connected ids.** Fails the shared-instance-id case + (a second physical connection under the same id would overwrite or be indistinguishable from the + first) and both reconnect-race orderings (a fresh connect racing a stale disconnect can land either + order). +* **A bare refcount, re-checked at `count == 0` when a timer wakes.** Handles the shared-id and + race-ordering cases, but has a real bug: a leader that disconnects, reconnects, and disconnects + again — all within one grace window — can leave a *stale* timer (armed for the first disconnect) + waking after the window and reading `count == 0`, which is true, but for the *second* disconnect, + not the one the timer was spawned to watch. It would settle a leader that has only been gone a + fraction of the configured window. +* **A refcounted registry recording the instant of each true 1→0 transition.** Chosen — see below. + +## Decision Outcome + +Chosen option: a refcounted `LeaderRegistry`, keyed by leader instance id, living in +`ipc::server::Shared` beside the existing `active_connections` counter but populated from inside +`ipc::connection::handle` rather than `Server::admit` — the negotiated instance id does not exist +until the `initialize` handshake completes, deep inside `handle`, which `admit`'s spawn wrapper never +sees. + +Each entry is `{count, zero_since}`, not a bare count. `zero_since` records the instant of the +entry's most recent 1→0 transition and is cleared on any 0→1 — the property that closes the bare- +refcount bug above: a grace-window timer waking later can tell "the same zero it was spawned to +watch" apart from "a fresher zero from an intervening reconnect/disconnect cycle" (see +`ipc::leader_registry::Entry::zero_since`'s own doc comment for the worked timeline). The disconnect +grace window itself is 60 seconds (`LEADER_DISCONNECT_GRACE_WINDOW`): short enough that an abandoned +run does not tie up a concurrency slot indefinitely, long enough to absorb an ordinary restart or +reconnect blip — the case this must never fire for. The decrement that can bring an entry to zero is +a drop guard (`LeaderConnectionGuard`), not a statement paired by hand at each exit from `handle`'s +dispatch loop: that loop's decrement needs the instance id, which only exists once execution is +already inside it, so unlike `active_connections` it cannot be bookkept from the spawn wrapper, and a +future early `return` would otherwise leak the registration silently, failing this feature closed +with nothing to notice. The guard's own `on_gone` callback fires exactly on the drop that causes the +true 1→0 transition, arming the grace-window timer from inside `Drop::drop` itself, so arming is as +structural as the decrement it rides on. + +Teardown reuses `run_cancel`'s own mechanical shape rather than a second implementation of it, +factored into a shared `cancel_and_settle_run` helper parameterized on the target `RunState`: +`run_cancel` itself still asks for the literal `"cancelled"`, checked against the requesting +principal; the grace-window path asks for `RunState::unrendered_verdict()` and passes no principal, +bypassing the ownership check the same way the removed backstop did. `unrendered_verdict()` is the +one function that owns the word "cancelled" for an evidence-free settlement, so this path and the +process-exit-after-settled-turn path that already called it can never drift on what word they use. +Before settling, a `WorkerTimeout { kind: LeaderGone }` fact is journaled first — the same +report-before-decide shape the inactivity sweep already uses for `Inactivity`/`Total` — so the +journal always states the reason before the state changes because of it. + +Removed entirely: the inactivity sweep's `settle_abandoned_turn` path (ADR-0027 wave 3). A parked run +whose leader is still connected is never settled by the runtime now, however long it stays silent — +the inactivity fact is still journaled, for `run/timeoutAck` to react to, but it no longer drives a +state change on its own. `lost` returns to ADR-0023's original, narrower meaning: the evidence itself +is missing or ambiguous, never "the leader went quiet." + +**Daemon restart.** At startup, nothing has connected to the freshly constructed `LeaderRegistry` +yet, so every leader that owns a non-terminal run is — from this process's own perspective — +disconnected as of right now. Each such owner is seeded with `zero_since` set to the daemon's own +startup instant, and the same grace-window timer is armed for it as a live disconnect would get. A +leader that reconnects within the window clears it exactly as an ordinary reconnect does; one that +never returns gets its runs settled through the identical path a live disconnect uses. The clock +deliberately starts at the daemon's *own* startup instant, never at a run's last-seen activity: a +leader cannot reconnect to a daemon that is not running, so any time the daemon spent stopped is the +daemon's own absence, not the leader's, and must never be charged against the window — doing +otherwise would settle live, reconnecting leaders' runs on every restart that followed a stop longer +than the window, which is exactly the destructive direction this feature exists to avoid. + +**Shared-instance-id hazard.** The `"crew-extension"` fallback id — used by any leader session with +no real session id of its own — means more than one physical leader process can share a single +registry entry. Refcounting keeps the *mechanism* safe under that sharing: both connect/disconnect +orderings resolve correctly by construction (plain addition does not care which side's `+1`/`-1` is +observed first), and the entry only reads "gone" once every sharer has disconnected. It does not +resolve the pre-existing ownership ambiguity the shared id already carries elsewhere in this +protocol — which of several processes sharing the id actually "owns" a given run — and this decision +inherits that ambiguity rather than creating or fixing it. + +### Positive Consequences + +* A run is never settled on a worker's silence alone; only a leader's connection actually being gone, + for the full grace window, can do that — closing exactly the gap the maintainer's ruling named. +* An abandoned leader's runs are no longer stranded forever: they settle within one grace window of + the leader's last connection actually going away, restart included. +* The settle path shares its mechanical shape (and its ordering/broadcast guarantees) with + `run_cancel`, so a future change to that shape cannot fix one call site and silently miss the + other. +* `lost` regains ADR-0023's original single meaning, removing the two-meanings-under-one-flag + ambiguity ADR-0027 wave 3 introduced for it. + +### Negative Consequences + +* A leader whose connection drops for reasons other than actually leaving (a long GC pause, a flaky + local socket) and does not reconnect within 60 seconds has its parked runs settled the same as a + leader that genuinely gave up — the grace window trades a false negative (settling too early) for + eliminating the false positive (settling on silence) the backstop had; it does not eliminate every + failure mode. +* The in-memory registry adds real state to reason about across a restart, which is why the + restart-seeding half of this decision exists at all; a bug in that seeding has the same destructive + shape (settling live runs) as the failure mode this whole decision exists to prevent. +* The shared-instance-id ownership ambiguity named above is now load-bearing for a second mechanism + (grace-window teardown) beyond whatever already depended on it; a future fix to that ambiguity must + account for this decision's use of it too. + +## Links + +* Amends [0027](0027-turn-end-settles-a-run.md), whose Decision point 3 is split by this ADR: the + backstop-settles-on-silence half is superseded here; the "never `succeeded`" half is inherited + unamended (a grace-window settlement is `RunState::unrendered_verdict()`, never `succeeded`, + exactly as the backstop it replaces never was). +* Related: [ADR-0023](0023-run-state-edges-from-adapter-evidence.md) (`lost`'s original, narrower + meaning, restored here), [ADR-0016](0016-coordination-scope-tokens-bound-to-run-and-pid-ancestry.md) + (the same "reuse a settle mechanism rather than parallel it" reasoning, applied there to + coordination scope and here to run teardown), [ADR-0020](0020-per-mutation-event-broadcast-is-not-optional.md) + (the broadcast-in-the-same-call invariant this decision's teardown path inherits from + `cancel_and_settle_run`). +* Implementation: `crates/runtime/src/ipc/leader_registry.rs` (the registry and its drop guard), + `crates/runtime/src/service/orchestration.rs` (`settle_leader_gone`, + `seed_leader_registry_after_restart`, `cancel_and_settle_run`), `crates/runtime/src/timeout_sweep.rs` + (the backstop's removal). diff --git a/docs/adr/README.md b/docs/adr/README.md index 06e51be3..aeb0029f 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -69,7 +69,7 @@ rather than behind a number nobody can resolve. | [0024](0024-project-scoped-reads-are-open-ownership-gates-writes.md) | Project-scoped reads are open; ownership gates writes | Accepted | | [0025](0025-crew-v2-tui-control-plane.md) | Crew v2 TUI control plane | Accepted | | [0026](0026-headless-retirement.md) | Headless control plane retirement | Accepted | -| [0027](0027-turn-end-settles-a-run.md) | A run is a conversation the leader closes; a vendor's turn-end is durable evidence, not a terminal state | Accepted | +| [0027](0027-turn-end-settles-a-run.md) | A run is a conversation the leader closes; a vendor's turn-end is durable evidence, not a terminal state | Accepted, Decision point 3 partially superseded by [0036](0036-leader-disconnect-grace-window.md) | | [0028](0028-submit-prompt-is-journaled-redacted-run-intent.md) | The submit prompt is journaled, redacted, as durable run intent | Accepted | | [0029](0029-placement-follows-the-backend-embedded-deleted.md) | Display placement is the backend's natural form, and `Embedded` is deleted rather than deprecated | Accepted | | [0030](0030-paste-delivery-bounded-on-progress.md) | Paste delivery is bounded on progress, not on elapsed time | Accepted | @@ -77,6 +77,7 @@ rather than behind a number nobody can resolve. | [0032](0032-model-names-are-resolved-and-persisted-on-confirmation.md) | A worker's model name is resolved against the host's catalogue, and persisted only once confirmed | Accepted | | [0034](0034-dashboard-bearer-token-over-loopback.md) | A daemon-hosted web dashboard, authenticated by a per-run bearer token over loopback | Accepted | | [0035](0035-protocol-doc-comment-sigil-split.md) | Protocol doc comments split by sigil: `///` ships, `//` stays internal | Accepted | +| [0036](0036-leader-disconnect-grace-window.md) | A parked run is settled on its leader's connection actually being gone, never on silence alone | Accepted | ## When to add one diff --git a/docs/architecture.md b/docs/architecture.md index 0347360b..cd9738ac 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -766,9 +766,12 @@ sequenceDiagram A TUI vendor never exits, so `ProcessExited` alone would leave such a run non-terminal forever. `TurnEnded` is the vendor's own end-of-turn boundary, journaled as durable evidence and driving a **non-terminal** edge: the boundary says the *turn* ended, never that the *task* succeeded, so - only the leader closes a run (`run/finish`), with an inactivity backstop settling an abandoned one - to `lost`. A run parked this way carries the `turnSettled` flag, which is what lets a snapshot - reader tell "the answer is ready" from "the worker asked a question" — both are `waitingUser`. + only the leader closes a run (`run/finish`) — or, if the leader's own connection is gone for the + full disconnect grace window, the runtime settles it via `RunState::unrendered_verdict()` + ([ADR-0036](adr/0036-leader-disconnect-grace-window.md)); silence alone, with the leader still + connected, never settles a parked run. A run parked this way carries the `turnSettled` flag, which + is what lets a snapshot reader tell "the answer is ready" from "the worker asked a question" — both + are `waitingUser`. Two ordering properties hold. The state edge is committed durably before the signal it triggers, as it always was. And the concurrency slot is released on the **first of a turn boundary or a diff --git a/docs/future-features.md b/docs/future-features.md index 3c6c2fb0..997b98c4 100644 --- a/docs/future-features.md +++ b/docs/future-features.md @@ -258,8 +258,9 @@ fragile about input, opaque about completion, and unverifiable about delivery. E answer in the TUI path itself: prompts are delivered as one bracketed paste in paced chunks with a bounded write, so a vendor that stops reading fails loudly instead of truncating silently; a run reaches a settled state from the vendor's own end-of-turn boundary rather than only from process -exit, with `run/finish` for the leader and an inactivity backstop for an abandoned turn -([ADR-0027](adr/0027-turn-end-settles-a-run.md)); and post-submit verification compares what the +exit, with `run/finish` for the leader and a disconnect-grace-window teardown for a leader that is +actually gone, never merely quiet ([ADR-0027](adr/0027-turn-end-settles-a-run.md), +[ADR-0036](adr/0036-leader-disconnect-grace-window.md)); and post-submit verification compares what the vendor actually recorded against what was sent, so a truncating composer fails the run rather than producing a plausible-looking short answer. A headless plane would have to beat that, not merely match a PTY's old weaknesses. diff --git a/packages/extension/dist/index.js b/packages/extension/dist/index.js index c71c5d8a..d37d71c7 100644 --- a/packages/extension/dist/index.js +++ b/packages/extension/dist/index.js @@ -10179,10 +10179,19 @@ own (see that event's doc comment).`, }, TimeoutKind: { description: "Which liveness deadline a `workerTimeout` event reports.", - type: "string", - enum: [ - "inactivity", - "total" + oneOf: [ + { + type: "string", + enum: [ + "inactivity", + "total" + ] + }, + { + description: "The run's owning leader has held no live connection for the full\ndisconnect grace window. Distinct from `inactivity`/`total`,\nwhich describe the worker's own liveness -- this describes the\nleader's, and is the fact a grace-window teardown journals before\nthe run is settled.", + type: "string", + const: "leaderGone" + } ] }, DisplayConfig: { diff --git a/packages/protocol-ts/schema/crew.schema.json b/packages/protocol-ts/schema/crew.schema.json index d5d01d9f..834ff225 100644 --- a/packages/protocol-ts/schema/crew.schema.json +++ b/packages/protocol-ts/schema/crew.schema.json @@ -3124,10 +3124,19 @@ }, "TimeoutKind": { "description": "Which liveness deadline a `workerTimeout` event reports.", - "type": "string", - "enum": [ - "inactivity", - "total" + "oneOf": [ + { + "type": "string", + "enum": [ + "inactivity", + "total" + ] + }, + { + "description": "The run's owning leader has held no live connection for the full\ndisconnect grace window. Distinct from `inactivity`/`total`,\nwhich describe the worker's own liveness -- this describes the\nleader's, and is the fact a grace-window teardown journals before\nthe run is settled.", + "type": "string", + "const": "leaderGone" + } ] }, "DisplayConfig": { diff --git a/packages/protocol-ts/src/generated/TimeoutKind.ts b/packages/protocol-ts/src/generated/TimeoutKind.ts index dec6231c..1b711b82 100644 --- a/packages/protocol-ts/src/generated/TimeoutKind.ts +++ b/packages/protocol-ts/src/generated/TimeoutKind.ts @@ -3,4 +3,4 @@ /** * Which liveness deadline a `workerTimeout` event reports. */ -export type TimeoutKind = "inactivity" | "total"; +export type TimeoutKind = "inactivity" | "total" | "leaderGone";