307 lines
13 KiB
Rust
307 lines
13 KiB
Rust
//! Inactivity watchdog for the synchronous inter-agent rendezvous
|
|
//! (`idea_ask_agent` ⇄ `idea_reply`).
|
|
//!
|
|
//! # Why this exists (root cause)
|
|
//!
|
|
//! A delegation target can legitimately work for a **very long time in a single
|
|
//! turn** (implement + `cargo test`/`clippy`), without ever returning to its prompt
|
|
//! nor emitting an intermediate signal the application can race against: the AI CLI
|
|
//! turn ([`domain::ports::AgentSession::send`]) only resolves **once the whole turn
|
|
//! is done** — its `ReplyEvent`s arrive as one batch at the end, not incrementally.
|
|
//! A **flat absolute timeout** around that wait therefore cuts a target that is
|
|
//! manifestly alive (its transcript keeps growing) at the cap, surfacing a generic,
|
|
//! non-actionable timeout.
|
|
//!
|
|
//! The fix is to bound the wait with an **inactivity window** re-armed on an observed
|
|
//! **sign of life**, under an **absolute ceiling**, with **distinct typed outcomes**:
|
|
//!
|
|
//! - progress observed under the ceiling ⇒ re-arm the window (no timeout);
|
|
//! - progress observed but the absolute ceiling is reached ⇒ a distinct
|
|
//! [`WatchdogOutcome::CeilingActive`] (the target is *active*, not silent — a blind
|
|
//! retry would stack a second heavy turn), never a mute timeout;
|
|
//! - genuine silence (no progress across a whole window) ⇒
|
|
//! [`WatchdogOutcome::NoReply`], exactly the previous timeout semantics.
|
|
//!
|
|
//! # Sign of life
|
|
//!
|
|
//! Liveness cannot come from the event stream (it is batched at end-of-turn, see
|
|
//! above), so it is supplied by an injected **monotonic activity probe** (in practice
|
|
//! the cumulative byte size of the target's transcript, which grows mid-turn). The
|
|
//! probe is a pure closure here — the algorithm stays free of infrastructure and is
|
|
//! unit-testable with a synthetic future + probe.
|
|
//!
|
|
//! # DRY
|
|
//!
|
|
//! This is the **single** implementation of that algorithm. The application layer
|
|
//! (`OrchestratorService`) governs the *delegated turn* with it (the seam that
|
|
//! actually cuts a long turn), and the infrastructure MCP adapter re-uses the very
|
|
//! same function as its outer safety net — no second, diverging copy.
|
|
|
|
use std::future::Future;
|
|
use std::time::{Duration, Instant};
|
|
|
|
/// Verdict of [`run_inactivity_watchdog`] over an arbitrary `dispatch` future,
|
|
/// decoupled from any concrete service/adapter so the algorithm is unit-testable
|
|
/// with a synthetic future + probe.
|
|
#[derive(Debug)]
|
|
pub enum WatchdogOutcome<T> {
|
|
/// The dispatch resolved before the watchdog tripped — carries its value.
|
|
Resolved(T),
|
|
/// Genuine silence (no observable progress across a window), or the flat-window
|
|
/// fallback when no probe is wired ⇒ the **retryable** no-reply / timeout verdict.
|
|
NoReply,
|
|
/// The absolute ceiling was reached **while still observing progress** ⇒ the
|
|
/// distinct, **non-retryable-blind** "target active, ceiling reached" verdict.
|
|
CeilingActive,
|
|
}
|
|
|
|
/// Races `dispatch` against an inactivity `window`, re-arming the window each time the
|
|
/// target shows **progress** (the monotonic token from `probe` strictly increased, or
|
|
/// became observable for the first time) until the dispatch resolves or the absolute
|
|
/// `ceiling` (measured from `started`) is reached.
|
|
///
|
|
/// - dispatch resolves first ⇒ [`WatchdogOutcome::Resolved`];
|
|
/// - window elapses with **no progress** (token unchanged/regressed/lost, or
|
|
/// `has_probe` is false ⇒ flat-window fallback) ⇒ [`WatchdogOutcome::NoReply`];
|
|
/// - window elapses **with progress** but `started.elapsed() >= ceiling` ⇒
|
|
/// [`WatchdogOutcome::CeilingActive`]; otherwise the window is re-armed.
|
|
///
|
|
/// `has_probe` distinguishes "no probe wired at all" (flat fallback: never progress,
|
|
/// = the previous flat-timeout behaviour ⇒ zero regression) from "probe wired but
|
|
/// currently returns `None`" (cannot prove liveness ⇒ treated as no progress).
|
|
///
|
|
/// `on_extended` / `on_expired` / `on_ceiling` are best-effort observation hooks
|
|
/// (diagnostics beacons); pass no-ops to ignore them.
|
|
#[allow(clippy::too_many_arguments)]
|
|
pub async fn run_inactivity_watchdog<T, F, Fut>(
|
|
dispatch: impl Future<Output = T>,
|
|
window: Duration,
|
|
ceiling: Duration,
|
|
started: Instant,
|
|
has_probe: bool,
|
|
mut probe: F,
|
|
mut on_extended: impl FnMut(Duration),
|
|
mut on_expired: impl FnMut(Duration),
|
|
mut on_ceiling: impl FnMut(Duration),
|
|
) -> WatchdogOutcome<T>
|
|
where
|
|
F: FnMut() -> Fut,
|
|
Fut: Future<Output = Option<u64>>,
|
|
{
|
|
tokio::pin!(dispatch);
|
|
// Baseline token before the first window; progress is judged relative to it.
|
|
let mut last_token = if has_probe { probe().await } else { None };
|
|
loop {
|
|
match tokio::time::timeout(window, &mut dispatch).await {
|
|
Ok(result) => return WatchdogOutcome::Resolved(result),
|
|
Err(_elapsed) => {
|
|
let elapsed = started.elapsed();
|
|
let now = if has_probe { probe().await } else { None };
|
|
// Progress iff a probe is wired AND the token strictly advanced (or
|
|
// became observable for the first time). No probe ⇒ never progress
|
|
// (flat-window fallback). Token lost ⇒ no proof of life ⇒ no progress.
|
|
let progressed = match (has_probe, last_token, now) {
|
|
(false, _, _) => false,
|
|
(true, Some(prev), Some(cur)) => cur > prev,
|
|
(true, None, Some(_)) => true,
|
|
(true, _, None) => false,
|
|
};
|
|
if progressed {
|
|
if elapsed >= ceiling {
|
|
on_ceiling(elapsed);
|
|
return WatchdogOutcome::CeilingActive;
|
|
}
|
|
on_extended(elapsed);
|
|
last_token = now;
|
|
continue;
|
|
}
|
|
on_expired(elapsed);
|
|
return WatchdogOutcome::NoReply;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Default inactivity window (silence budget) of the inter-agent rendezvous: the size
|
|
/// of **one** no-progress probe window, NOT an absolute cap. 600 s, matching the
|
|
/// historical flat cap so a genuinely silent target still expires in the same delay.
|
|
pub const DEFAULT_RENDEZVOUS_WINDOW: Duration = Duration::from_secs(600);
|
|
|
|
/// Default absolute ceiling of the inter-agent rendezvous: the extending inactivity
|
|
/// window never parks a call past this even against a perpetually-busy target.
|
|
/// Generous (4 h) so a heavy-but-real single turn finishes well within it.
|
|
pub const DEFAULT_RENDEZVOUS_CEILING: Duration = Duration::from_secs(4 * 60 * 60);
|
|
|
|
/// Resolves the effective rendezvous **inactivity window** from an optional override in
|
|
/// **milliseconds**: `Some(ms>0)` ⇒ that window, else the [`DEFAULT_RENDEZVOUS_WINDOW`].
|
|
/// `Some(0)` is treated as "no override" so a misconfigured zero can never collapse the
|
|
/// window to an instant expiry. Pure and unit-testable without wiring.
|
|
#[must_use]
|
|
pub fn resolve_rendezvous_window(override_ms: Option<u32>) -> Duration {
|
|
match override_ms {
|
|
Some(ms) if ms > 0 => Duration::from_millis(u64::from(ms)),
|
|
_ => DEFAULT_RENDEZVOUS_WINDOW,
|
|
}
|
|
}
|
|
|
|
/// Resolves the effective rendezvous **absolute ceiling** from an optional override in
|
|
/// **milliseconds**: `Some(ms>0)` ⇒ that ceiling, else [`DEFAULT_RENDEZVOUS_CEILING`].
|
|
/// `Some(0)` is treated as "no override". Pure and unit-testable without wiring.
|
|
#[must_use]
|
|
pub fn resolve_rendezvous_ceiling(override_ms: Option<u32>) -> Duration {
|
|
match override_ms {
|
|
Some(ms) if ms > 0 => Duration::from_millis(u64::from(ms)),
|
|
_ => DEFAULT_RENDEZVOUS_CEILING,
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use std::sync::atomic::{AtomicU64, Ordering};
|
|
use std::sync::Arc;
|
|
|
|
/// A `dispatch` future that never resolves — stands in for a target busy in one
|
|
/// long turn (or wedged) that has not yet produced a reply.
|
|
async fn never() -> Result<u32, u32> {
|
|
std::future::pending::<()>().await;
|
|
unreachable!()
|
|
}
|
|
|
|
const WINDOW: Duration = Duration::from_millis(30);
|
|
|
|
fn noop(_: Duration) {}
|
|
|
|
/// (c) Vrai silence : un probe est câblé mais son token n'avance jamais ⇒ le
|
|
/// watchdog expire en ~une fenêtre avec le verdict no-reply (pas d'extension),
|
|
/// soit exactement l'ancien timeout plat.
|
|
#[tokio::test]
|
|
async fn silent_target_expires_no_reply() {
|
|
let started = Instant::now();
|
|
let probe = || async { Some(42_u64) };
|
|
let outcome = run_inactivity_watchdog(
|
|
never(),
|
|
WINDOW,
|
|
Duration::from_secs(3600),
|
|
started,
|
|
true,
|
|
probe,
|
|
noop,
|
|
noop,
|
|
noop,
|
|
)
|
|
.await;
|
|
assert!(matches!(outcome, WatchdogOutcome::NoReply));
|
|
assert!(started.elapsed() < Duration::from_secs(1));
|
|
}
|
|
|
|
/// (d) Non-régression du fallback fenêtre plate : sans probe câblé, la première
|
|
/// expiration de fenêtre est un no-reply (= ancien timeout plat).
|
|
#[tokio::test]
|
|
async fn no_probe_falls_back_to_flat_window() {
|
|
let started = Instant::now();
|
|
let probe = || async { None };
|
|
let outcome = run_inactivity_watchdog(
|
|
never(),
|
|
WINDOW,
|
|
Duration::from_secs(3600),
|
|
started,
|
|
false,
|
|
probe,
|
|
noop,
|
|
noop,
|
|
noop,
|
|
)
|
|
.await;
|
|
assert!(matches!(outcome, WatchdogOutcome::NoReply));
|
|
}
|
|
|
|
/// (a) Cible qui progresse AU-DELÀ de l'ancien plafond : le probe avance à chaque
|
|
/// tick (extension) bien au-delà d'une fenêtre, puis la dispatch se résout ⇒ AUCUN
|
|
/// timeout prématuré, le résultat est rendu. C'est le défaut central corrigé.
|
|
#[tokio::test]
|
|
async fn progressing_target_extends_then_resolves() {
|
|
let started = Instant::now();
|
|
let tick = Arc::new(AtomicU64::new(0));
|
|
let tick2 = Arc::clone(&tick);
|
|
let probe = move || {
|
|
let tick = Arc::clone(&tick2);
|
|
async move { Some(tick.fetch_add(1, Ordering::SeqCst) + 1) }
|
|
};
|
|
// Résout après ~plusieurs fenêtres (110 ms ≫ 30 ms) : sans réarmement, un
|
|
// timeout plat de 30 ms aurait coupé. Le watchdog doit étendre et rendre 7.
|
|
let dispatch = async {
|
|
tokio::time::sleep(Duration::from_millis(110)).await;
|
|
Ok::<u32, u32>(7)
|
|
};
|
|
let outcome = run_inactivity_watchdog(
|
|
dispatch,
|
|
WINDOW,
|
|
Duration::from_secs(3600),
|
|
started,
|
|
true,
|
|
probe,
|
|
noop,
|
|
noop,
|
|
noop,
|
|
)
|
|
.await;
|
|
match outcome {
|
|
WatchdogOutcome::Resolved(Ok(v)) => assert_eq!(v, 7),
|
|
_ => panic!("expected Resolved(Ok(7)) — la fenêtre aurait dû être réarmée"),
|
|
}
|
|
}
|
|
|
|
/// (b) Plafond absolu atteint malgré progrès : le probe avance toujours (cible
|
|
/// active) mais le plafond est minuscule ⇒ verdict DISTINCT CeilingActive (pas un
|
|
/// faux no-reply muet), et l'appel termine (jamais parqué à l'infini).
|
|
#[tokio::test]
|
|
async fn progressing_target_hits_ceiling_distinctly() {
|
|
let started = Instant::now();
|
|
let tick = Arc::new(AtomicU64::new(0));
|
|
let tick2 = Arc::clone(&tick);
|
|
let probe = move || {
|
|
let tick = Arc::clone(&tick2);
|
|
async move { Some(tick.fetch_add(1, Ordering::SeqCst) + 1) }
|
|
};
|
|
let outcome = run_inactivity_watchdog(
|
|
never(),
|
|
WINDOW,
|
|
Duration::from_millis(45),
|
|
started,
|
|
true,
|
|
probe,
|
|
noop,
|
|
noop,
|
|
noop,
|
|
)
|
|
.await;
|
|
assert!(matches!(outcome, WatchdogOutcome::CeilingActive));
|
|
assert!(started.elapsed() < Duration::from_secs(1));
|
|
}
|
|
|
|
/// Les résolveurs honorent un override et retombent sur leurs défauts finis ;
|
|
/// `Some(0)` ⇒ "pas d'override" (jamais d'effondrement instantané).
|
|
#[test]
|
|
fn resolvers_apply_override_else_finite_default() {
|
|
assert_eq!(
|
|
resolve_rendezvous_window(Some(1234)),
|
|
Duration::from_millis(1234)
|
|
);
|
|
assert_eq!(
|
|
resolve_rendezvous_window(Some(0)),
|
|
DEFAULT_RENDEZVOUS_WINDOW
|
|
);
|
|
assert_eq!(resolve_rendezvous_window(None), DEFAULT_RENDEZVOUS_WINDOW);
|
|
assert_eq!(
|
|
resolve_rendezvous_ceiling(Some(9999)),
|
|
Duration::from_millis(9999)
|
|
);
|
|
assert_eq!(
|
|
resolve_rendezvous_ceiling(Some(0)),
|
|
DEFAULT_RENDEZVOUS_CEILING
|
|
);
|
|
assert_eq!(resolve_rendezvous_ceiling(None), DEFAULT_RENDEZVOUS_CEILING);
|
|
}
|
|
}
|