diff --git a/crates/app-tauri/src/state.rs b/crates/app-tauri/src/state.rs index a4b7100..45210a7 100644 --- a/crates/app-tauri/src/state.rs +++ b/crates/app-tauri/src/state.rs @@ -617,6 +617,13 @@ pub struct AppState { /// **remplace** son handle (l'ancien est droppé ⇒ polling arrêté) ; fermer/arrêter /// l'agent le retire. `Mutex` car launch/stop y accèdent concurremment. pub turn_watch_handles: Mutex>>, + /// Port `FileSystem` partagé, conservé pour bâtir la **sonde d'activité** du rendez-vous + /// `idea_ask_agent` (octets cumulés des transcripts de la cible). Même port que + /// l'inspecteur et le turn-watcher. + pub fs_port: Arc, + /// Répertoire home (`$HOME`) servant à dériver `/.claude/projects/` + /// pour la sonde d'activité du rendez-vous. + pub home_dir: String, } impl AppState { @@ -1028,7 +1035,7 @@ impl AppState { // — lit le même `/.claude/projects//` que l'inspecteur, via // le même `FileSystem`. Armé par agent supporté au lancement (cf. `arm_turn_watch`). let turn_watcher: Arc = Arc::new( - infrastructure::ClaudeTranscriptTurnWatcher::new(Arc::clone(&fs_port), home_dir), + infrastructure::ClaudeTranscriptTurnWatcher::new(Arc::clone(&fs_port), home_dir.clone()), ); let inspect_conversation = Arc::new(InspectConversation::new( Arc::clone(&contexts_port), @@ -1383,7 +1390,39 @@ impl AppState { // sa racine à la construction). Sans ça, l'outil renverrait « not configured ». .with_live_state_read(Arc::new(AppLiveStateLeanProvider { clock: Arc::clone(&clock) as Arc, - }) as Arc), + }) as Arc) + // Fenêtre d'inactivité réarmable (signe de vie) du rendez-vous délégué : la + // borne de tour dans le service n'est plus un timeout plat mais une fenêtre + // réarmée à chaque progrès observé de la cible (octets cumulés de son + // transcript Claude), sous le plafond ci-dessous. SANS cette sonde, la borne + // dégrade vers un timeout plat (fallback, zéro régression) et coupe un long + // tour unique à 600 s — c'est précisément la sonde qui rend le fix effectif. + // Keyée par (project_root, agent_id) : le run-dir transcript = `/.ideai/ + // run/`, encodé par Claude sous `/.claude/projects/...`. + .with_ask_liveness_probe({ + let fs = Arc::clone(&fs_port); + let home = home_dir.clone(); + Arc::new(move |root: domain::project::ProjectPath, agent_id| { + let fs = Arc::clone(&fs); + let home = home.clone(); + Box::pin(async move { + let run_dir = format!( + "{}/.ideai/run/{agent_id}", + root.as_str().trim_end_matches(['/', '\\']) + ); + let cwd = domain::project::ProjectPath::new(run_dir).ok()?; + infrastructure::transcript_activity_token(fs.as_ref(), &home, &cwd).await + }) as std::pin::Pin> + Send>> + }) as application::AskLivenessProbe + }) + // Plafond absolu du rendez-vous délégué (réglage projet via + // `IDEA_ASK_RENDEZVOUS_CEILING_MS`, défaut 4 h) : la fenêtre réarmée ne parque + // jamais un `ask` au-delà, même contre une cible perpétuellement active. + .with_ask_ceiling(application::resolve_rendezvous_ceiling( + std::env::var("IDEA_ASK_RENDEZVOUS_CEILING_MS") + .ok() + .and_then(|v| v.trim().parse::().ok()), + )), // NB (régression corrigée) : on ne câble PAS `.with_structured(...)` ici. // Décision produit lot B-2 (« Option 1 Terminal + MCP », cf. construction // de `LaunchAgent` plus haut) : la fabrique structurée est décâblée, donc @@ -1502,6 +1541,8 @@ impl AppState { turn_watcher, turn_watch_input: Arc::clone(&input_mediator), turn_watch_handles: Mutex::new(HashMap::new()), + fs_port: Arc::clone(&fs_port), + home_dir, move_tab, } } @@ -1595,11 +1636,50 @@ impl AppState { .ok() .and_then(|v| v.trim().parse::().ok()), ); + // Plafond absolu du rendez-vous (réglage projet optionnel via + // `IDEA_ASK_RENDEZVOUS_CEILING_MS`, défaut 4 h) : borne dure que la fenêtre + // d'inactivité réarmée ne dépasse jamais, même contre une cible perpétuellement active. + let ask_ceiling = infrastructure::resolve_ask_rendezvous_ceiling( + std::env::var("IDEA_ASK_RENDEZVOUS_CEILING_MS") + .ok() + .and_then(|v| v.trim().parse::().ok()), + ); + // Sonde d'activité de la cible (signe de vie du rendez-vous) : l'McpServer (infra) ne + // connaît la cible que par son NOM ; la composition root est la seule à savoir + // résoudre nom→AgentId puis dériver le run-dir transcript Claude. La sonde renvoie un + // jeton monotone = octets cumulés des `.jsonl` de la cible (croît même pendant un seul + // long tour sans `turn_duration`). `None` ⇒ cible/transcript introuvable ⇒ traité comme + // « pas de progrès » par le watchdog. + let service_for_probe = Arc::clone(&self.orchestrator_service); + let fs_for_probe = Arc::clone(&self.fs_port); + let home_for_probe = self.home_dir.clone(); + let project_for_probe = project.clone(); + let activity_probe: infrastructure::AskActivityProbe = Arc::new(move |target: String| { + let service = Arc::clone(&service_for_probe); + let fs = Arc::clone(&fs_for_probe); + let home = home_for_probe.clone(); + let project = project_for_probe.clone(); + Box::pin(async move { + let agent_id = service + .resolve_agent_id_by_name(&project, &target) + .await + .ok() + .flatten()?; + let run_dir = format!( + "{}/.ideai/run/{agent_id}", + project.root.as_str().trim_end_matches(['/', '\\']) + ); + let cwd = domain::project::ProjectPath::new(run_dir).ok()?; + infrastructure::transcript_activity_token(fs.as_ref(), &home, &cwd).await + }) + }); let handle = McpServerHandle::start( McpServer::new(Arc::clone(&self.orchestrator_service), project.clone()) .with_events(events) .with_ready_sink(ready_sink) - .with_ask_rendezvous_timeout(ask_timeout), + .with_ask_rendezvous_timeout(ask_timeout) + .with_ask_rendezvous_ceiling(ask_ceiling) + .with_activity_probe(activity_probe), endpoint, listener, project_id, diff --git a/crates/application/src/error.rs b/crates/application/src/error.rs index 5d1be6c..8ae5944 100644 --- a/crates/application/src/error.rs +++ b/crates/application/src/error.rs @@ -69,6 +69,21 @@ pub enum AppError { )] TargetReturnedNoReply(String), + /// The delegation target is **still actively working** (its transcript kept + /// growing across liveness probes) but the rendezvous **absolute ceiling** was + /// reached, so the synchronous wait was abandoned. **Distinct** from + /// [`Self::TargetReturnedNoReply`]: the target is *not* silent, so a blind retry + /// would stack a second heavy turn on a target already mid-task. **Not** a mute + /// timeout — the caller should check the target's in-progress work/branch (it may + /// still finish and call `idea_reply`) before re-soliciting lightly. Carries the + /// target's display name. + #[error( + "target {0} is still actively working but the rendezvous ceiling was reached; \ + do not retry blindly — check the target's in-progress work/branch (it may still \ + finish and call idea_reply), then re-solicit lightly if needed" + )] + TargetCeilingActive(String), + /// An unexpected internal error. #[error("internal error: {0}")] Internal(String), @@ -89,6 +104,7 @@ impl AppError { Self::Remote(_) => "REMOTE", Self::AgentAlreadyRunning { .. } => "AGENT_ALREADY_RUNNING", Self::TargetReturnedNoReply(_) => "TARGET_RETURNED_NO_REPLY", + Self::TargetCeilingActive(_) => "RENDEZVOUS_CEILING_ACTIVE", Self::Internal(_) => "INTERNAL", } } @@ -173,3 +189,31 @@ impl From for AppError { Self::Process(e.to_string()) } } + +#[cfg(test)] +mod tests { + use super::*; + + /// La cible-active-au-plafond porte un code **stable et distinct** du no-reply + /// (l'appelant peut brancher sans parser le message) et un message « ne retente + /// pas à l'aveugle » différent du retour-au-prompt-sans-reply. + #[test] + fn ceiling_active_code_is_stable_and_distinct() { + let ceiling = AppError::TargetCeilingActive("architect".to_owned()); + assert_eq!(ceiling.code(), "RENDEZVOUS_CEILING_ACTIVE"); + + let no_reply = AppError::TargetReturnedNoReply("architect".to_owned()); + assert_eq!(no_reply.code(), "TARGET_RETURNED_NO_REPLY"); + + // Deux issues TYPÉES distinctes (le cœur du cadrage Architect) : un plafond + // atteint sur cible active n'est jamais confondu avec un silence/no-reply. + assert_ne!(ceiling.code(), no_reply.code()); + // Et distinct d'un timeout muet (`Process` via AgentSessionError::Timeout). + assert_ne!( + ceiling.code(), + AppError::from(AgentSessionError::Timeout).code() + ); + // Le message guide explicitement vers « ne pas retenter à l'aveugle ». + assert!(ceiling.to_string().contains("do not retry blindly")); + } +} diff --git a/crates/application/src/lib.rs b/crates/application/src/lib.rs index c8a8aae..1c37995 100644 --- a/crates/application/src/lib.rs +++ b/crates/application/src/lib.rs @@ -85,9 +85,11 @@ pub use memory::{ UpdateMemory, UpdateMemoryInput, UpdateMemoryOutput, }; pub use orchestrator::{ - ContextGuardUseCases, LiveStateProvider, LiveStateReadProvider, McpRuntimeProvider, - OrchestratorOutcome, OrchestratorService, ProposeContext, ReadContext, ReadMemory, - RecordTurnProvider, WriteMemory, + resolve_rendezvous_ceiling, resolve_rendezvous_window, run_inactivity_watchdog, + AskLivenessProbe, ContextGuardUseCases, LiveStateProvider, LiveStateReadProvider, + McpRuntimeProvider, OrchestratorOutcome, OrchestratorService, ProposeContext, ReadContext, + ReadMemory, RecordTurnProvider, WatchdogOutcome, WriteMemory, DEFAULT_RENDEZVOUS_CEILING, + DEFAULT_RENDEZVOUS_WINDOW, }; pub use permission::{ GetProjectPermissions, GetProjectPermissionsInput, GetProjectPermissionsOutput, diff --git a/crates/application/src/orchestrator/mod.rs b/crates/application/src/orchestrator/mod.rs index 963fbed..a204525 100644 --- a/crates/application/src/orchestrator/mod.rs +++ b/crates/application/src/orchestrator/mod.rs @@ -5,13 +5,18 @@ //! ever spawning a process itself. See [`service::OrchestratorService`]. mod context_guard; +pub mod rendezvous; mod service; pub use context_guard::{ ProposeContext, ProposeContextInput, ProposeOutcome, ReadContext, ReadContextInput, ReadMemory, ReadMemoryInput, WriteMemory, WriteMemoryInput, }; -pub use service::{ - ContextGuardUseCases, LiveStateProvider, LiveStateReadProvider, McpRuntimeProvider, - OrchestratorOutcome, OrchestratorService, RecordTurnProvider, +pub use rendezvous::{ + resolve_rendezvous_ceiling, resolve_rendezvous_window, run_inactivity_watchdog, + WatchdogOutcome, DEFAULT_RENDEZVOUS_CEILING, DEFAULT_RENDEZVOUS_WINDOW, +}; +pub use service::{ + AskLivenessProbe, ContextGuardUseCases, LiveStateProvider, LiveStateReadProvider, + McpRuntimeProvider, OrchestratorOutcome, OrchestratorService, RecordTurnProvider, }; diff --git a/crates/application/src/orchestrator/rendezvous.rs b/crates/application/src/orchestrator/rendezvous.rs new file mode 100644 index 0000000..7411b1b --- /dev/null +++ b/crates/application/src/orchestrator/rendezvous.rs @@ -0,0 +1,294 @@ +//! 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 { + /// 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( + dispatch: impl Future, + 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 +where + F: FnMut() -> Fut, + Fut: Future>, +{ + 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) -> 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) -> 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 { + 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::(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); + } +} diff --git a/crates/application/src/orchestrator/service.rs b/crates/application/src/orchestrator/service.rs index 93baf2a..cb3edd8 100644 --- a/crates/application/src/orchestrator/service.rs +++ b/crates/application/src/orchestrator/service.rs @@ -39,6 +39,7 @@ use crate::agent::{ UpdateAgentContextInput, CODEX_SUBMIT_DELAY_MS, }; use crate::error::AppError; +use crate::orchestrator::rendezvous::{run_inactivity_watchdog, WatchdogOutcome}; use crate::orchestrator::{ ProposeContext, ProposeContextInput, ProposeOutcome, ReadContext, ReadContextInput, ReadMemory, ReadMemoryInput, WriteMemory, WriteMemoryInput, @@ -85,6 +86,35 @@ fn submit_config_for_profile(profile: &AgentProfile) -> SubmitConfig { /// ([`resolve_turn_timeout`]). const ASK_AGENT_TIMEOUT: Duration = Duration::from_secs(600); +/// **Sonde de vivacité** d'une cible de délégation, keyée par son [`AgentId`] : renvoie +/// un **jeton d'activité** monotone non-décroissant (en pratique la taille cumulée du +/// transcript de la cible). Un jeton strictement plus grand entre deux sondes ⇒ la cible +/// **progresse** (vivante et au travail, **même dans un seul long tour** qui n'émet aucun +/// événement intermédiaire — cf. [`rendezvous`](crate::orchestrator::rendezvous)). +/// `None` ⇒ vivacité non observable (cible/transcript introuvable) ⇒ traité comme +/// « pas de progrès ». +/// +/// **Pourquoi un port et pas le flux d'événements** : le flux de +/// [`domain::ports::AgentSession::send`] est **batché en fin de tour** (les `ReplyEvent` +/// arrivent tous à la fin, pas en continu), donc inutilisable comme signe de vie *pendant* +/// un long tour. La seule preuve de vie disponible est cette sonde externe (transcript). +/// Closure abstraite (frontière hexagonale) construite à la composition root — la seule +/// couche qui sait dériver le run-dir transcript d'un `AgentId` ; l'application reste pure. +pub type AskLivenessProbe = Arc< + dyn Fn( + domain::project::ProjectPath, + AgentId, + ) -> std::pin::Pin> + Send>> + + Send + + Sync, +>; + +/// Plafond **absolu** par défaut du rendez-vous délégué : la fenêtre d'inactivité +/// réarmée ne parque jamais un `ask` au-delà, même contre une cible perpétuellement +/// active. Généreux (4 h) pour qu'un tour unique lourd mais réel finisse bien avant. +/// Surchargeable via `IDEA_ASK_RENDEZVOUS_CEILING_MS` à la composition root. +const ASK_AGENT_CEILING: Duration = crate::orchestrator::rendezvous::DEFAULT_RENDEZVOUS_CEILING; + /// Borne d'attente **en file** pour acquérir le verrou de tour d'un agent (A0, /// cadrage v5 §4). /// @@ -362,6 +392,17 @@ pub struct OrchestratorService { /// Injecté via [`Self::with_live_state_read`] ; `None` ⇒ l'outil renvoie une erreur /// typée « not configured » (call sites/tests legacy restent verts). live_state_read: Option>, + /// **Sonde de vivacité** de la cible (signe de vie) pilotant la fenêtre d'inactivité + /// du rendez-vous délégué (cf. [`AskLivenessProbe`] et + /// [`rendezvous`](crate::orchestrator::rendezvous)). Injectée via + /// [`Self::with_ask_liveness_probe`]. `None` ⇒ la borne de tour reste un **timeout + /// plat** (fallback fenêtre plate = comportement historique, zéro régression). + ask_liveness_probe: Option, + /// Plafond **absolu** du rendez-vous délégué (cf. [`ASK_AGENT_CEILING`]) : la fenêtre + /// d'inactivité réarmée ne parque jamais un `ask` au-delà. Surchargeable via + /// [`Self::with_ask_ceiling`] (la composition root résout l'override d'env) ; les + /// tests le rétrécissent à quelques ms. Défaut = 4 h. + ask_ceiling: Duration, } /// Bundle des quatre use cases C7 sous [`domain::fileguard::FileGuard`], injectés @@ -429,9 +470,32 @@ impl OrchestratorService { read_skill: None, live_state: None, live_state_read: None, + ask_liveness_probe: None, + ask_ceiling: ASK_AGENT_CEILING, } } + /// Branche la **sonde de vivacité** (signe de vie) du rendez-vous délégué : la borne + /// de tour devient une **fenêtre d'inactivité** réarmée à chaque progrès observé, + /// sous le plafond [`Self::with_ask_ceiling`]. Builder additif (signature de + /// [`Self::new`] inchangée) ; sans sonde, la borne reste un timeout plat (fallback, + /// zéro régression). Cf. [`AskLivenessProbe`]. + #[must_use] + pub fn with_ask_liveness_probe(mut self, probe: AskLivenessProbe) -> Self { + self.ask_liveness_probe = Some(probe); + self + } + + /// Surcharge le **plafond absolu** du rendez-vous délégué (défaut [`ASK_AGENT_CEILING`] + /// = 4 h) au-delà duquel la fenêtre d'inactivité réarmée ne parque jamais un `ask`. + /// La composition root y applique l'override d'env ; les tests le rétrécissent. + /// Builder additif. + #[must_use] + pub fn with_ask_ceiling(mut self, ceiling: Duration) -> Self { + self.ask_ceiling = ceiling; + self + } + /// Branche le use case [`ReadSkill`] (`skill.read`/`idea_skill_read`). Builder /// additif (signature de [`Self::new`] inchangée) ; `None` ⇒ la commande /// renvoie une erreur typée « not configured ». @@ -1382,10 +1446,16 @@ impl OrchestratorService { // turn into the bound handle). The service no longer writes the PTY directly — // no ad-hoc `[IdeA · tâche …]` line here, no `\r` band-aid (cadrage C3 §5.1). - // 3. Attendre la réponse, bornée. Timeout/canal fermé ⇒ le garde retire le ticket - // (cible laissée vivante) et la ramène `Idle` au Drop ; renvoie une erreur typée. - match tokio::time::timeout(turn_timeout, pending).await { - Ok(Ok(TurnResolution::Replied(result))) => { + // 3. Attendre la réponse, bornée par la **fenêtre d'inactivité réarmable** + // (signe de vie = sonde de transcript) au lieu d'un timeout plat : un long + // tour unique qui progresse n'est plus coupé à `turn_timeout`. Vrai silence + // ⇒ timeout typé (comme avant) ; plafond atteint malgré progrès ⇒ erreur + // typée distincte. Canal fermé ⇒ le garde retire le ticket au Drop. + let verdict = self + .run_ask_with_watchdog(pending, turn_timeout, &project.root, agent_id, &target, started) + .await; + match verdict { + WatchdogOutcome::Resolved(Ok(TurnResolution::Replied(result))) => { // Checkpoint Response (P6b, best-effort) : persister la réponse AVANT de // déplacer `result` dans `reply_outcome`. Source = la **cible** (c'est // elle qui a rendu le tour) ; même conversation que le Prompt. @@ -1424,9 +1494,14 @@ impl OrchestratorService { // déjà retiré (`complete_without_reply`) et la cible déjà `Idle` (prompt-ready // `mark_idle`) ⇒ on **désarme** le garde (pas de `cancel_head`/`mark_idle` // redondant ni de beacon « busy-guard freed » trompeur) et on renvoie une - // erreur typée claire/retryable, en ~G au lieu du timeout long. - Ok(Ok(TurnResolution::ReturnedToPromptNoReply)) => { + // erreur typée claire/retryable, en ~G au lieu du timeout long. On repasse + // aussi la live-state de la cible à `Done` (best-effort) : le tour est conclu + // sans réponse, donc `idea_workstate_read` ne doit pas la voir `Working` + // (busy fantôme) comme sur la branche succès. + WatchdogOutcome::Resolved(Ok(TurnResolution::ReturnedToPromptNoReply)) => { busy_guard.disarm(); + self.mark_target_done_best_effort(&project.root, agent_id, ticket_id) + .await; crate::diag!( "[rendezvous] ask returned-to-prompt-no-reply: target={target} \ (agent {agent_id}) ticket={ticket_id} after_ms={}", @@ -1434,9 +1509,23 @@ impl OrchestratorService { ); Err(AppError::TargetReturnedNoReply(target)) } + // Plafond absolu atteint alors que la cible **progresse encore** (sonde de vie + // croissante) : verdict DISTINCT, non un faux timeout muet. Le garde fait + // `cancel_head` + `mark_idle` au Drop ; on réconcilie aussi la live-state à + // `Done` pour qu'un abandon ne laisse pas un busy fantôme. + WatchdogOutcome::CeilingActive => { + self.mark_target_done_best_effort(&project.root, agent_id, ticket_id) + .await; + crate::diag!( + "[rendezvous] ask CEILING (still active): target={target} \ + (agent {agent_id}) ticket={ticket_id} after_ms={}", + started.elapsed().as_millis(), + ); + Err(AppError::TargetCeilingActive(target)) + } // Erreur / timeout : on laisse le garde faire `cancel_head` + `mark_idle` au // Drop (retrait des `cancel_head` redondants — `cancel_head` reste idempotent). - Ok(Err(_cancelled)) => { + WatchdogOutcome::Resolved(Err(_cancelled)) => { crate::diag!( "[rendezvous] ask channel-closed: target={target} (agent {agent_id}) \ ticket={ticket_id} after_ms={}", @@ -1446,11 +1535,17 @@ impl OrchestratorService { "agent {target} : canal de réponse fermé avant un résultat" ))) } - Err(_elapsed) => { + WatchdogOutcome::NoReply => { + // Vrai silence (aucun progrès sur une fenêtre) : sémantique identique à + // l'ancien timeout plat. On réconcilie la live-state à `Done` pour qu'un + // abandon ne laisse pas la cible `Working` (busy fantôme) ; le garde fait + // `cancel_head` + `mark_idle` au Drop. + self.mark_target_done_best_effort(&project.root, agent_id, ticket_id) + .await; crate::diag!( "[rendezvous] ask TIMEOUT: target={target} (agent {agent_id}) \ ticket={ticket_id} after_ms={} (la cible n'a jamais appelé idea_reply \ - ni atteint son prompt-ready dans le turn_timeout)", + ni atteint son prompt-ready, et aucun progrès observé sur la fenêtre)", started.elapsed().as_millis(), ); Err(AppError::from(domain::ports::AgentSessionError::Timeout)) @@ -1552,74 +1647,90 @@ impl OrchestratorService { turn_timeout.as_millis(), ); - // Drainer le tour structuré (le `Final` ⇒ contenu + `mark_idle`), borné par le - // **même** garde-fou que le chemin PTY (profil ou défaut). On attend la - // **première** issue : le tour structuré OU un `idea_reply` explicite. - let drain = - drain_with_readiness(session, &task, Some(turn_timeout), input.as_ref(), agent_id); + // Drainer le tour structuré (le `Final` ⇒ contenu + `mark_idle`). Le drain lui-même + // est **non borné** (`None`) : la borne de tour est désormais portée par la + // **fenêtre d'inactivité réarmable** autour de l'attente (signe de vie = sonde de + // transcript), pas par un timeout plat interne — un long tour unique qui progresse + // n'est plus coupé à `turn_timeout`. On attend la **première** issue : le tour + // structuré OU un `idea_reply` explicite. + let drain = drain_with_readiness(session, &task, None, input.as_ref(), agent_id); - let result = tokio::select! { - biased; - // Issue déterministe : la session a rendu son `Final`. - drained = drain => match drained { - Ok(content) => { - crate::diag!( - "[rendezvous] ask resolved (structured/final): target={target} \ - (agent {agent_id}) ticket={ticket_id} after_ms={} reply_len={}", - started.elapsed().as_millis(), - content.len(), - ); - content - } - // Erreur de drain : le garde fait `cancel_head` + `mark_idle` au Drop. - Err(err) => { - crate::diag!( - "[rendezvous] ask drain-error (structured): target={target} \ - (agent {agent_id}) ticket={ticket_id} after_ms={} err={err}", - started.elapsed().as_millis(), - ); - return Err(AppError::from(err)); - } - }, - // Issue alternative : un `idea_reply` explicite a résolu le ticket d'abord. - replied = pending => match replied { - Ok(TurnResolution::Replied(content)) => { - // La session draine encore en arrière-plan ; on s'assure que la FIFO - // avance même si le `Final` n'est pas encore observé. - input.mark_idle(agent_id); - crate::diag!( - "[rendezvous] ask resolved (structured/reply): target={target} \ - (agent {agent_id}) ticket={ticket_id} after_ms={} reply_len={}", - started.elapsed().as_millis(), - content.len(), - ); - content - } - // Retour au prompt sans `idea_reply` : très rare sur le chemin structuré - // (la cible n'a pas de PTY/watcher), mais on traite la résolution comme - // sur le chemin PTY — erreur typée claire/retryable, garde désarmé. - Ok(TurnResolution::ReturnedToPromptNoReply) => { - input.mark_idle(agent_id); - busy_guard.disarm(); - crate::diag!( - "[rendezvous] ask returned-to-prompt-no-reply (structured): \ - target={target} (agent {agent_id}) ticket={ticket_id} after_ms={}", - started.elapsed().as_millis(), - ); - return Err(AppError::TargetReturnedNoReply(target.to_owned())); - } - // Canal fermé : le garde fait `cancel_head` + `mark_idle` au Drop. - Err(_cancelled) => { - crate::diag!( - "[rendezvous] ask channel-closed (structured): target={target} \ - (agent {agent_id}) ticket={ticket_id} after_ms={}", - started.elapsed().as_millis(), - ); - return Err(AppError::Process(format!( + // L'attente du rendez-vous (drain OU reply), rendue comme `Result` + // pour être enveloppée par le watchdog (au lieu de `return` directs dans le `select!`). + let wait = async { + tokio::select! { + biased; + drained = drain => match drained { + Ok(content) => Ok(content), + Err(err) => Err(AppError::from(err)), + }, + replied = pending => match replied { + Ok(TurnResolution::Replied(content)) => { + // La session draine encore en arrière-plan ; on fait avancer la FIFO. + input.mark_idle(agent_id); + Ok(content) + } + Ok(TurnResolution::ReturnedToPromptNoReply) => { + input.mark_idle(agent_id); + Err(AppError::TargetReturnedNoReply(target.to_owned())) + } + Err(_cancelled) => Err(AppError::Process(format!( "agent {target} : canal de réponse fermé avant un résultat" - ))); - } - }, + ))), + }, + } + }; + + // Borne par la fenêtre d'inactivité (réarmée sur signe de vie) sous plafond absolu. + let result = match self + .run_ask_with_watchdog(wait, turn_timeout, &project.root, agent_id, target, started) + .await + { + WatchdogOutcome::Resolved(Ok(content)) => { + crate::diag!( + "[rendezvous] ask resolved (structured): target={target} \ + (agent {agent_id}) ticket={ticket_id} after_ms={} reply_len={}", + started.elapsed().as_millis(), + content.len(), + ); + content + } + // Erreur de tour (drain-error, no-reply, canal fermé) : le garde fait + // `cancel_head` + `mark_idle` au Drop. On réconcilie la live-state à `Done` + // pour qu'un abandon ne laisse pas un busy fantôme. + WatchdogOutcome::Resolved(Err(err)) => { + self.mark_target_done_best_effort(&project.root, agent_id, ticket_id) + .await; + crate::diag!( + "[rendezvous] ask error (structured): target={target} (agent {agent_id}) \ + ticket={ticket_id} after_ms={} err={err}", + started.elapsed().as_millis(), + ); + return Err(err); + } + // Plafond absolu atteint alors que la cible progresse encore : verdict distinct. + WatchdogOutcome::CeilingActive => { + self.mark_target_done_best_effort(&project.root, agent_id, ticket_id) + .await; + crate::diag!( + "[rendezvous] ask CEILING (structured, still active): target={target} \ + (agent {agent_id}) ticket={ticket_id} after_ms={}", + started.elapsed().as_millis(), + ); + return Err(AppError::TargetCeilingActive(target.to_owned())); + } + // Vrai silence (aucun progrès sur la fenêtre) : timeout typé, identique à + // l'ancienne borne plate. Live-state réconciliée à `Done`. + WatchdogOutcome::NoReply => { + self.mark_target_done_best_effort(&project.root, agent_id, ticket_id) + .await; + crate::diag!( + "[rendezvous] ask TIMEOUT (structured): target={target} (agent {agent_id}) \ + ticket={ticket_id} after_ms={} (aucun progrès observé sur la fenêtre)", + started.elapsed().as_millis(), + ); + return Err(AppError::from(domain::ports::AgentSessionError::Timeout)); + } }; // Succès : désarmer le garde (le `mark_idle` propre est déjà porté par le `Final` @@ -1793,6 +1904,22 @@ impl OrchestratorService { } } + /// Résout l'`AgentId` d'un agent par son **nom d'affichage** (insensible à la casse) + /// dans le manifeste du projet, pour la composition root (qui ne connaît la cible d'un + /// `idea_ask_agent` que par son nom). Expose la résolution privée `find_agent_id_by_name` + /// afin que le filet serveur MCP puisse dériver le run-dir transcript de la cible (sonde + /// d'activité du rendez-vous). `None` si aucun agent ne porte ce nom. + /// + /// # Errors + /// Propage l'[`AppError`] du listing des agents (lecture du manifeste). + pub async fn resolve_agent_id_by_name( + &self, + project: &Project, + name: &str, + ) -> Result, AppError> { + self.find_agent_id_by_name(project, name).await + } + /// Déclare si une **cellule terminal frontend** est montée pour `agent` (write-portal /// actif). Pont entre le write-portal (adapter UI) et le médiateur : un agent avec /// cellule reçoit ses tours via l'événement `DelegationReady` (le front écrit) ; un @@ -2396,6 +2523,77 @@ impl OrchestratorService { resolve_turn_timeout(turn_timeout_ms) } + /// Borne un `wait` (l'attente synchrone du rendez-vous délégué) par la **fenêtre + /// d'inactivité réarmable** (cf. [`rendezvous`](crate::orchestrator::rendezvous)), + /// au lieu d'un timeout plat. La fenêtre vaut `window` (profil sinon défaut, la + /// **même** valeur que l'ancien timeout plat) ; à chaque expiration de fenêtre, la + /// sonde de vivacité de la cible est consultée : progrès ⇒ réarmement jusqu'au + /// plafond absolu [`Self::ask_ceiling`]. Verdict : + /// - [`WatchdogOutcome::Resolved`] : le `wait` a résolu (succès/erreur de tour) ⇒ rendu tel quel ; + /// - [`WatchdogOutcome::NoReply`] : vrai silence (aucun progrès sur une fenêtre) **ou** + /// fallback fenêtre plate (pas de sonde câblée) — sémantique **identique** à l'ancien + /// timeout de tour (l'appelant mappe vers le timeout typé) ; + /// - [`WatchdogOutcome::CeilingActive`] : progrès observé mais plafond atteint ⇒ verdict + /// distinct (l'appelant mappe vers [`AppError::TargetCeilingActive`]). + async fn run_ask_with_watchdog( + &self, + wait: impl std::future::Future, + window: Duration, + project_root: &domain::project::ProjectPath, + agent_id: AgentId, + target: &str, + started: Instant, + ) -> WatchdogOutcome { + let has_probe = self.ask_liveness_probe.is_some(); + let probe = self.ask_liveness_probe.clone(); + let root = project_root.clone(); + let probe_fn = move || { + let probe = probe.clone(); + let root = root.clone(); + async move { + match &probe { + Some(p) => p(root, agent_id).await, + None => None, + } + } + }; + let ext_target = target.to_owned(); + let exp_target = target.to_owned(); + let ceil_target = target.to_owned(); + let window_ms = window.as_millis(); + let ceiling_ms = self.ask_ceiling.as_millis(); + run_inactivity_watchdog( + wait, + window, + self.ask_ceiling, + started, + has_probe, + probe_fn, + move |elapsed| { + crate::diag!( + "[rendezvous] ask window extended (signe de vie): target={ext_target} \ + (agent {agent_id}) window_ms={window_ms} elapsed_ms={}", + elapsed.as_millis(), + ); + }, + move |elapsed| { + crate::diag!( + "[rendezvous] ask EXPIRED (no progress): target={exp_target} \ + (agent {agent_id}) window_ms={window_ms} elapsed_ms={}", + elapsed.as_millis(), + ); + }, + move |elapsed| { + crate::diag!( + "[rendezvous] ask CEILING active: target={ceil_target} (agent {agent_id}) \ + ceiling_ms={ceiling_ms} elapsed_ms={}", + elapsed.as_millis(), + ); + }, + ) + .await + } + /// Resolves a human-friendly profile reference (slug like `claude-code`, /// command like `claude`, or display name like `Claude Code`) to a configured /// [`ProfileId`]. Matching is universal — never hard-coded to one AI — by diff --git a/crates/infrastructure/src/inspector/claude_paths.rs b/crates/infrastructure/src/inspector/claude_paths.rs index 0371b12..66b271a 100644 --- a/crates/infrastructure/src/inspector/claude_paths.rs +++ b/crates/infrastructure/src/inspector/claude_paths.rs @@ -17,7 +17,7 @@ //! [`ClaudeTranscriptInspector`]: super::ClaudeTranscriptInspector //! [`ClaudeTranscriptTurnWatcher`]: super::ClaudeTranscriptTurnWatcher -use domain::ports::RemotePath; +use domain::ports::{FileSystem, RemotePath}; use domain::profile::{AgentProfile, ContextInjection}; use domain::project::ProjectPath; @@ -74,6 +74,38 @@ pub(crate) fn supports_claude(profile: &AgentProfile) -> bool { ) } +/// Cumulative byte size of **all** `.jsonl` transcripts in a Claude agent's per-run-dir +/// folder (`/.claude/projects//`) — a monotonic **activity token** for +/// the `idea_ask_agent` rendezvous inactivity watchdog (server-side safety net). +/// +/// Why bytes, not the `turn_duration` count: a target working in **one long turn** never +/// emits a `turn_duration` record until that turn ends, yet its transcript file keeps +/// **growing** as the model streams output. Summing transcript bytes therefore detects +/// "alive and working" even mid-turn — exactly the wedge the flat timeout mis-handled. +/// Best-effort and side-effect-free: a missing folder (cold start) or any unreadable file +/// contributes `0`; `None` only when the folder cannot be listed at all (treated by the +/// watchdog as "no observable progress"). Lives beside the path encoder so the transcript +/// layout knowledge stays in one tested seam. +pub async fn transcript_activity_token( + fs: &dyn FileSystem, + home_dir: &str, + cwd: &ProjectPath, +) -> Option { + let dir = claude_project_dir(home_dir, cwd); + let entries = fs.list(&dir).await.ok()?; + let mut total: u64 = 0; + for entry in entries { + if entry.is_dir || !entry.name.ends_with(".jsonl") { + continue; + } + let path = RemotePath::new(format!("{}/{}", dir.0, entry.name)); + if let Ok(bytes) = fs.read(&path).await { + total = total.saturating_add(bytes.len() as u64); + } + } + Some(total) +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/infrastructure/src/inspector/mod.rs b/crates/infrastructure/src/inspector/mod.rs index fc23515..68263f5 100644 --- a/crates/infrastructure/src/inspector/mod.rs +++ b/crates/infrastructure/src/inspector/mod.rs @@ -12,4 +12,5 @@ mod claude_paths; mod claude_turn_watcher; pub use claude::ClaudeTranscriptInspector; +pub use claude_paths::transcript_activity_token; pub use claude_turn_watcher::ClaudeTranscriptTurnWatcher; diff --git a/crates/infrastructure/src/lib.rs b/crates/infrastructure/src/lib.rs index 27ce230..efe9ac5 100644 --- a/crates/infrastructure/src/lib.rs +++ b/crates/infrastructure/src/lib.rs @@ -48,10 +48,11 @@ pub use fs::LocalFileSystem; pub use git::Git2Repository; pub use id::UuidGenerator; pub use input::{MediatedInbox, MillisClock, SystemMillisClock}; -pub use inspector::{ClaudeTranscriptInspector, ClaudeTranscriptTurnWatcher}; +pub use inspector::{transcript_activity_token, ClaudeTranscriptInspector, ClaudeTranscriptTurnWatcher}; pub use mailbox::InMemoryMailbox; pub use orchestrator::mcp::{ - resolve_ask_rendezvous_timeout, McpServer, MemoryTransport, StdioTransport, + resolve_ask_rendezvous_ceiling, resolve_ask_rendezvous_timeout, AskActivityProbe, McpServer, + MemoryTransport, StdioTransport, }; pub use orchestrator::{ process_request_file, FsOrchestratorWatcher, OrchestratorResponse, OrchestratorWatchHandle, diff --git a/crates/infrastructure/src/orchestrator/mcp/jsonrpc.rs b/crates/infrastructure/src/orchestrator/mcp/jsonrpc.rs index babf697..5031d3e 100644 --- a/crates/infrastructure/src/orchestrator/mcp/jsonrpc.rs +++ b/crates/infrastructure/src/orchestrator/mcp/jsonrpc.rs @@ -130,6 +130,16 @@ pub mod error_codes { /// may re-ask, reminding the target to answer via `idea_reply`. The accompanying /// `data` carries `{ "code": "TARGET_RETURNED_NO_REPLY", "retryable": true }`. pub const RENDEZVOUS_NO_REPLY: i32 = -32_001; + + /// **Server-defined** (JSON-RPC reserved range -32000..=-32099): the synchronous + /// `idea_ask_agent` rendezvous hit the **absolute ceiling** while the target was + /// **still actively working** (its transcript kept growing). **Distinct** from + /// [`RENDEZVOUS_NO_REPLY`]: the target is *not* silent -- it simply outran the hard + /// ceiling. It is therefore **non-retryable** (a blind re-ask would pile a second + /// heavy turn on a target already mid-task); the caller must inspect the target's + /// in-progress work/branch before re-soliciting lightly. The accompanying `data` + /// carries `{ "code": "RENDEZVOUS_CEILING_ACTIVE", "retryable": false }`. + pub const RENDEZVOUS_CEILING_ACTIVE: i32 = -32_002; } /// Errors a [`Transport`] may surface. diff --git a/crates/infrastructure/src/orchestrator/mcp/mod.rs b/crates/infrastructure/src/orchestrator/mcp/mod.rs index 86b8fa3..c775be4 100644 --- a/crates/infrastructure/src/orchestrator/mcp/mod.rs +++ b/crates/infrastructure/src/orchestrator/mcp/mod.rs @@ -36,6 +36,8 @@ pub mod transport; pub use jsonrpc::{ JsonRpcError, JsonRpcRequest, JsonRpcResponse, Transport, TransportError, JSONRPC_VERSION, }; -pub use server::{resolve_ask_rendezvous_timeout, McpServer}; +pub use server::{ + resolve_ask_rendezvous_ceiling, resolve_ask_rendezvous_timeout, AskActivityProbe, McpServer, +}; pub use tools::{catalogue, map_tool_call, tool_returns_reply, ToolDef, ToolMapError}; pub use transport::{MemoryTransport, StdioTransport}; diff --git a/crates/infrastructure/src/orchestrator/mcp/server.rs b/crates/infrastructure/src/orchestrator/mcp/server.rs index 999d38a..b743dd6 100644 --- a/crates/infrastructure/src/orchestrator/mcp/server.rs +++ b/crates/infrastructure/src/orchestrator/mcp/server.rs @@ -17,6 +17,8 @@ //! at the composition root (M3), exactly like the watcher — no application logic is //! duplicated here. +use std::future::Future; +use std::pin::Pin; use std::sync::Arc; use std::time::{Duration, Instant}; @@ -34,29 +36,51 @@ use super::tools::{self, ToolMapError}; /// The MCP protocol version this server speaks (advertised on `initialize`). const MCP_PROTOCOL_VERSION: &str = "2024-11-05"; -/// **Server-side safety net** bounding the synchronous `idea_ask_agent` rendezvous. +/// **Inactivity window (silence budget)** for the synchronous `idea_ask_agent` +/// rendezvous — the size of one *no-progress* probe window, NOT an absolute cap. /// /// `idea_ask_agent` is the only tool whose `tools/call` *blocks* awaiting another /// agent's `idea_reply` (a synchronous rendezvous, resolved deep in the application -/// layer). The application layer already bounds the rendezvous itself, but this -/// adapter adds its own **finite** outer bound so a wedged target can never park a -/// `tools/call` task forever: on expiry the call returns a clean JSON-RPC error -/// instead of hanging. +/// layer). Rather than a flat outer timeout (which wrongly fired on a single long +/// delegated turn that never returns to its prompt nor emits `turn_duration`), the +/// adapter runs an **inactivity watchdog**: it races the dispatch against this window +/// and, on each window expiry, probes whether the target made progress (its transcript +/// grew). Progress ⇒ the window is re-armed (extension) up to the absolute +/// [`ASK_RENDEZVOUS_CEILING`]; genuine silence ⇒ the no-reply error fires. 600 s by +/// default, overridable per project via `IDEA_ASK_RENDEZVOUS_TIMEOUT_MS`. Every other +/// tool (`idea_reply`, `idea_list_agents`, …) is left untouched — they don't rendezvous. /// -/// **Plancher universel FINI (backstop no-reply).** The real turn-end signal now exists -/// (the transcript `turn_duration` watcher), but only Claude emits it; this outer bound -/// stays the last resort for any silent target that never emits it (Codex & co) or that -/// wedges. It must therefore NEVER be (quasi-)infinite. 600 s by default, overridable per -/// project via `IDEA_ASK_RENDEZVOUS_TIMEOUT_MS`. Every other tool (`idea_reply`, -/// `idea_list_agents`, …) is left untouched — they don't rendezvous. +/// **Fallback (no activity probe).** When no probe is wired (legacy call sites, tests +/// that shrink the bound), this degrades to the previous *flat* timeout: the first +/// window expiry with no probe is treated as a no-reply expiry — zero regression. const ASK_RENDEZVOUS_TIMEOUT: Duration = Duration::from_secs(600); -/// Resolves the effective `idea_ask_agent` rendezvous bound from an optional -/// configured override in **milliseconds** (project/profile level), mirroring the -/// application's [`resolve_turn_timeout`](application) shape: `Some(ms)` ⇒ that bound, -/// `None`/absent ⇒ the finite [`ASK_RENDEZVOUS_TIMEOUT`] default (600 s). `Some(0)` is -/// treated as "no override" so a misconfigured zero can never collapse the safety net to -/// an instant expiry. Pure and unit-testable without wiring. +/// **Absolute ceiling (hard cap)** of the `idea_ask_agent` rendezvous, so an extending +/// inactivity window can never park a `tools/call` forever even against a target that +/// stays *busy* indefinitely. Reaching it while the target is still actively working +/// yields the distinct, **non-retryable** [`rendezvous_ceiling_active_error`] (never a +/// blind-retry suggestion). Generous by default (4 h), overridable per project via +/// `IDEA_ASK_RENDEZVOUS_CEILING_MS`. +const ASK_RENDEZVOUS_CEILING: Duration = Duration::from_secs(4 * 60 * 60); + +/// A monotonic **activity probe** of a target agent, keyed by its display name. +/// +/// Returns an opaque, monotonically non-decreasing **activity token** (in practice the +/// cumulative byte size of the target's transcript `.jsonl` files): a strictly larger +/// token between two probes means the target made progress (it is alive and working, +/// even within a single long turn that never emits `turn_duration`). `None` ⇒ the +/// target/transcript could not be resolved (cold start, unknown name) — treated as "no +/// observable progress" so a truly stuck call still expires. Async because resolving the +/// token reads the filesystem; built at the composition root (the only layer that knows +/// the name→run-dir mapping), keeping this adapter free of `AgentId`. +pub type AskActivityProbe = + Arc Pin> + Send>> + Send + Sync>; + +/// Resolves the effective `idea_ask_agent` **inactivity window** from an optional +/// configured override in **milliseconds**: `Some(ms)` ⇒ that window, `None`/absent ⇒ +/// the [`ASK_RENDEZVOUS_TIMEOUT`] default (600 s). `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_ask_rendezvous_timeout(override_ms: Option) -> Duration { match override_ms { @@ -65,6 +89,19 @@ pub fn resolve_ask_rendezvous_timeout(override_ms: Option) -> Duration { } } +/// Resolves the effective `idea_ask_agent` **absolute ceiling** from an optional +/// configured override in **milliseconds**: `Some(ms)` ⇒ that ceiling, `None`/absent ⇒ +/// the [`ASK_RENDEZVOUS_CEILING`] default (4 h). `Some(0)` is treated as "no override". +/// Pure and unit-testable without wiring. +#[must_use] +pub fn resolve_ask_rendezvous_ceiling(override_ms: Option) -> Duration { + match override_ms { + Some(ms) if ms > 0 => Duration::from_millis(u64::from(ms)), + _ => ASK_RENDEZVOUS_CEILING, + } +} + + /// The IdeA MCP server: an entry adapter over [`OrchestratorService::dispatch`]. /// /// Cheap to clone the dependencies it holds; one instance serves one project's @@ -93,11 +130,21 @@ pub struct McpServer { /// (fix race cold-launch, signal MCP). L'infra ne connaît pas `AgentId` : la /// composition root parse le `requester`. `None` ⇒ no-op. ready_sink: Option>, - /// Finite outer bound applied **only** to the `idea_ask_agent` rendezvous (see - /// [`ASK_RENDEZVOUS_TIMEOUT`]). Carried per instance so tests can shrink it to a - /// few milliseconds without polluting the public API; production code always uses - /// the default. + /// **Inactivity window (silence budget)** of the `idea_ask_agent` rendezvous (see + /// [`ASK_RENDEZVOUS_TIMEOUT`]). One probe window, re-armed on observed progress. + /// Carried per instance so tests can shrink it to a few milliseconds without + /// polluting the public API; production code uses the resolved default. ask_rendezvous_timeout: Duration, + /// **Absolute ceiling** of the `idea_ask_agent` rendezvous (see + /// [`ASK_RENDEZVOUS_CEILING`]): the extending inactivity window never parks a call + /// past this. Per instance so tests can shrink it; production uses the resolved + /// default (4 h). + ask_rendezvous_ceiling: Duration, + /// Optional monotonic **activity probe** of the target (see [`AskActivityProbe`]), + /// injected by the composition root. Drives window extension: progress between two + /// probes re-arms the window. `None` ⇒ no observability ⇒ the rendezvous degrades to + /// a single flat window (legacy behaviour, zero regression). + activity_probe: Option, } impl McpServer { @@ -113,6 +160,8 @@ impl McpServer { requester: String::new(), ready_sink: None, ask_rendezvous_timeout: ASK_RENDEZVOUS_TIMEOUT, + ask_rendezvous_ceiling: ASK_RENDEZVOUS_CEILING, + activity_probe: None, } } @@ -154,24 +203,50 @@ impl McpServer { requester: requester.into(), ready_sink: self.ready_sink.clone(), ask_rendezvous_timeout: self.ask_rendezvous_timeout, + ask_rendezvous_ceiling: self.ask_rendezvous_ceiling, + activity_probe: self.activity_probe.clone(), } } - /// Overrides the `idea_ask_agent` rendezvous safety-net bound (default - /// [`ASK_RENDEZVOUS_TIMEOUT`]). + /// Overrides the `idea_ask_agent` **inactivity window** (silence budget, default + /// [`ASK_RENDEZVOUS_TIMEOUT`] = 600 s). /// /// A genuine **configuration knob** (project/profile level, modelled on the /// application's `turn_timeout_ms` / `resolve_turn_timeout`): the composition root /// resolves an optional override with [`resolve_ask_rendezvous_timeout`] and applies - /// it here. Absent override ⇒ the generous 24 h default is preserved unchanged (zero - /// regression on legitimately long delegated turns). Tests also use it to shrink the - /// bound to milliseconds. + /// it here. This is the size of one *no-progress* window: as long as the activity + /// probe sees the target advancing, the window is re-armed up to the absolute + /// [`ASK_RENDEZVOUS_CEILING`]. Absent override ⇒ the 600 s default. Tests also use it + /// to shrink the window to milliseconds. #[must_use] pub fn with_ask_rendezvous_timeout(mut self, timeout: Duration) -> Self { self.ask_rendezvous_timeout = timeout; self } + /// Overrides the `idea_ask_agent` **absolute ceiling** (hard cap, default + /// [`ASK_RENDEZVOUS_CEILING`] = 4 h) past which the extending inactivity window never + /// parks a call, even against a perpetually-busy target. The composition root resolves + /// an optional override with [`resolve_ask_rendezvous_ceiling`]. Tests use it to shrink + /// the ceiling to milliseconds. Additive: callers that don't set it keep the default. + #[must_use] + pub fn with_ask_rendezvous_ceiling(mut self, ceiling: Duration) -> Self { + self.ask_rendezvous_ceiling = ceiling; + self + } + + /// Attaches the monotonic **activity probe** (see [`AskActivityProbe`]) driving the + /// inactivity-window extension: between two window expiries, a strictly larger token + /// means the target is alive and working, so the window is re-armed instead of + /// expiring. Built by the composition root (the only layer mapping a display name to a + /// run-dir transcript). Additive: without a probe the rendezvous degrades to a single + /// flat window (legacy behaviour, zero regression). + #[must_use] + pub fn with_activity_probe(mut self, probe: AskActivityProbe) -> Self { + self.activity_probe = Some(probe); + self + } + /// Serves JSON-RPC messages from `transport`, tagging every processed /// `tools/call` with `requester` as the delegating agent (cadrage v5 §1.4). /// @@ -406,43 +481,40 @@ impl McpServer { } }; - // `idea_ask_agent` is the only tool that blocks on a synchronous rendezvous - // (the target's `idea_reply`). Bound *only* that path with a finite outer - // timeout (server-side safety net, see [`ASK_RENDEZVOUS_TIMEOUT`]): on expiry - // return a clean JSON-RPC error instead of parking the task forever. Every - // other tool dispatches unbounded — they never rendezvous. + // `idea_ask_agent` is the only tool that blocks on a synchronous rendezvous (the + // target's `idea_reply`). Bound *only* that path with the **inactivity watchdog** + // (server-side safety net, see [`ASK_RENDEZVOUS_TIMEOUT`] / [`ASK_RENDEZVOUS_CEILING`]): + // race the dispatch against an inactivity window; on each window expiry, probe + // whether the target progressed (transcript grew). Progress ⇒ re-arm the window + // (extension) up to the absolute ceiling; genuine silence ⇒ the retryable no-reply + // error; ceiling reached while still active ⇒ the distinct non-retryable ceiling + // error. Every other tool dispatches unbounded — they never rendezvous. let dispatch = self.service.dispatch(&self.project, command); let result = if name == "idea_ask_agent" { - // Armed beacon : le rendezvous synchrone est désormais borné par le filet - // de sécurité serveur. Si « end » n'apparaît jamais après ce « armed », la - // cible n'a ni appelé `idea_reply` ni atteint son prompt-ready dans le délai. application::diag!( "[mcp] ask armed requester={requester_label} target={arg_target} \ - task_len={task_len} timeout_ms={}", + task_len={task_len} window_ms={} ceiling_ms={}", self.ask_rendezvous_timeout.as_millis(), + self.ask_rendezvous_ceiling.as_millis(), ); - match tokio::time::timeout(self.ask_rendezvous_timeout, dispatch).await { - Ok(result) => result, - Err(_elapsed) => { - // RISQUE #1 (Architect) : on `return` ici ⇒ le futur `dispatch` (déplacé - // dans `timeout`) est **droppé**, jamais détaché dans un `spawn`. C'est ce - // Drop qui libère le `BusyTurnGuard` RAII de `ask_agent` (cancel_head + - // mark_idle) et purge le `Busy`/ticket de la cible — aucune fuite `Busy`. - application::diag!( - "[mcp] ask EXPIRED requester={requester_label} target={arg_target} \ - timeout_ms={} elapsed_ms={}", - self.ask_rendezvous_timeout.as_millis(), - started.elapsed().as_millis(), - ); - // Filet serveur : ne plus renvoyer un INTERNAL_ERROR opaque, mais la même - // sémantique typée et **retryable** que `AppError::TargetReturnedNoReply`, - // sous un code JSON-RPC DISTINCT (RENDEZVOUS_NO_REPLY) + `data` structuré. - return Err(rendezvous_no_reply_error(&arg_target)); + match self + .run_ask_rendezvous(dispatch, &arg_target, &requester_label, started) + .await + { + // RISQUE #1 (Architect) : sur NoReply/CeilingActive, le futur `dispatch` a déjà + // été **droppé** dans `run_ask_rendezvous` (il le possède et le pin), ce Drop + // libère le `BusyTurnGuard` RAII et purge le `Busy`/ticket de la cible (aucune + // fuite `Busy`). + AskRendezvousOutcome::Resolved(result) => result, + AskRendezvousOutcome::NoReply => return Err(rendezvous_no_reply_error(&arg_target)), + AskRendezvousOutcome::CeilingActive => { + return Err(rendezvous_ceiling_active_error(&arg_target)) } } } else { dispatch.await }; + // Surface the processed delegation on the bus, tagged as the MCP door — the // twin of the file watcher's publish. `ok` mirrors the dispatch outcome; the // action is the tool name. No-op when no sink is attached. @@ -500,6 +572,77 @@ impl McpServer { }); } } + + /// Drives the `idea_ask_agent` synchronous rendezvous under the **inactivity + /// watchdog** (cf. [`ASK_RENDEZVOUS_TIMEOUT`] / [`ASK_RENDEZVOUS_CEILING`]). + /// + /// Thin adapter over the testable free function [`run_inactivity_watchdog`]: it + /// supplies this server's window, ceiling and (optional) activity probe, and maps the + /// watchdog verdict to an [`AskRendezvousOutcome`]. With no probe wired the watchdog + /// degrades to a single flat window (legacy behaviour, zero regression). + async fn run_ask_rendezvous( + &self, + dispatch: impl Future>, + target: &str, + requester_label: &str, + started: Instant, + ) -> AskRendezvousOutcome { + let probe = self.activity_probe.clone(); + let target_owned = target.to_owned(); + let probe_fn = move || { + let probe = probe.clone(); + let target = target_owned.clone(); + async move { + match &probe { + Some(p) => p(target).await, + None => None, + } + } + }; + // DRY : on réutilise l'**unique** implémentation du watchdog (couche application), + // la même que celle qui gouverne réellement le tour délégué dans le service. Ici + // c'est le filet de sécurité externe de l'adaptateur MCP. + let window_ms = self.ask_rendezvous_timeout.as_millis(); + let ceiling_ms = self.ask_rendezvous_ceiling.as_millis(); + let (t_ext, r_ext) = (target.to_owned(), requester_label.to_owned()); + let (t_exp, r_exp) = (target.to_owned(), requester_label.to_owned()); + let (t_ceil, r_ceil) = (target.to_owned(), requester_label.to_owned()); + match application::run_inactivity_watchdog( + dispatch, + self.ask_rendezvous_timeout, + self.ask_rendezvous_ceiling, + started, + self.activity_probe.is_some(), + probe_fn, + move |elapsed| { + application::diag!( + "[mcp] ask window extended (signe de vie) requester={r_ext} target={t_ext} \ + window_ms={window_ms} elapsed_ms={}", + elapsed.as_millis(), + ); + }, + move |elapsed| { + application::diag!( + "[mcp] ask EXPIRED (no progress) requester={r_exp} target={t_exp} \ + window_ms={window_ms} elapsed_ms={}", + elapsed.as_millis(), + ); + }, + move |elapsed| { + application::diag!( + "[mcp] ask CEILING active requester={r_ceil} target={t_ceil} \ + ceiling_ms={ceiling_ms} elapsed_ms={}", + elapsed.as_millis(), + ); + }, + ) + .await + { + application::WatchdogOutcome::Resolved(r) => AskRendezvousOutcome::Resolved(r), + application::WatchdogOutcome::NoReply => AskRendezvousOutcome::NoReply, + application::WatchdogOutcome::CeilingActive => AskRendezvousOutcome::CeilingActive, + } + } } /// Builds an MCP `tools/call` result with a single text content block. @@ -537,6 +680,44 @@ fn rendezvous_no_reply_error(target: &str) -> JsonRpcError { } } +/// Outcome of the inactivity-watchdog rendezvous ([`McpServer::run_ask_rendezvous`]), +/// folded into a `tools/call` result by the caller. +enum AskRendezvousOutcome { + /// The dispatch completed (success or a typed `AppError`) — returned unchanged. + Resolved(Result), + /// The inactivity window elapsed with no observable progress ⇒ the retryable + /// [`rendezvous_no_reply_error`]. Also the flat-window fallback when no probe is wired. + NoReply, + /// The absolute ceiling was reached while the target was still actively working ⇒ the + /// distinct, non-retryable [`rendezvous_ceiling_active_error`]. + CeilingActive, +} + +/// Builds the JSON-RPC error returned when the `idea_ask_agent` rendezvous reaches the +/// **absolute ceiling** while the target is **still actively working** (its transcript +/// kept growing across probes). +/// +/// **Distinct** from [`rendezvous_no_reply_error`]: the target is not silent, so a blind +/// retry would stack a second heavy turn on a target already mid-task. Carried under the +/// dedicated [`error_codes::RENDEZVOUS_CEILING_ACTIVE`] code with `data.retryable = false` +/// so a caller can branch without parsing the message. +fn rendezvous_ceiling_active_error(target: &str) -> JsonRpcError { + let message = format!( + "target {target} is still actively working but the rendezvous ceiling was reached; \ + do not retry blindly — check the target's in-progress work/branch (it may still \ + finish and call idea_reply), then re-solicit lightly if needed" + ); + JsonRpcError { + code: error_codes::RENDEZVOUS_CEILING_ACTIVE, + message, + data: Some(json!({ + "code": "RENDEZVOUS_CEILING_ACTIVE", + "retryable": false, + "target": target, + })), + } +} + /// Maps a [`ToolMapError`] to the right JSON-RPC error code. fn map_err_to_jsonrpc(err: ToolMapError) -> JsonRpcError { match err { @@ -548,3 +729,49 @@ fn map_err_to_jsonrpc(err: ToolMapError) -> JsonRpcError { } } } + +#[cfg(test)] +mod tests { + use super::*; + use std::time::Duration; + + // NOTE: the inactivity-watchdog **algorithm** now lives in (and is unit-tested by) + // the `application` crate (`application::run_inactivity_watchdog`), the single source + // of truth that also governs the delegated turn. This adapter only maps its verdict to + // an `AskRendezvousOutcome`, so we keep here only the adapter-specific concerns: the + // error builders and the override resolvers. + + /// The two error builders carry distinct JSON-RPC codes and the documented + /// `data.retryable` flags (no-reply ⇒ retryable; ceiling-active ⇒ not). + #[test] + fn error_builders_carry_distinct_codes_and_retryable_flags() { + let no_reply = rendezvous_no_reply_error("architect"); + assert_eq!(no_reply.code, error_codes::RENDEZVOUS_NO_REPLY); + assert_eq!(no_reply.data.as_ref().unwrap()["retryable"], serde_json::json!(true)); + assert_eq!( + no_reply.data.as_ref().unwrap()["code"], + serde_json::json!("TARGET_RETURNED_NO_REPLY") + ); + + let ceiling = rendezvous_ceiling_active_error("architect"); + assert_eq!(ceiling.code, error_codes::RENDEZVOUS_CEILING_ACTIVE); + assert_eq!(ceiling.data.as_ref().unwrap()["retryable"], serde_json::json!(false)); + assert_eq!( + ceiling.data.as_ref().unwrap()["code"], + serde_json::json!("RENDEZVOUS_CEILING_ACTIVE") + ); + assert_ne!(no_reply.code, ceiling.code); + } + + /// The resolvers honour an override and fall back to their finite defaults; `Some(0)` + /// is treated as "no override" (never an instant collapse). + #[test] + fn resolvers_apply_override_else_finite_default() { + assert_eq!(resolve_ask_rendezvous_timeout(Some(1234)), Duration::from_millis(1234)); + assert_eq!(resolve_ask_rendezvous_timeout(Some(0)), ASK_RENDEZVOUS_TIMEOUT); + assert_eq!(resolve_ask_rendezvous_timeout(None), ASK_RENDEZVOUS_TIMEOUT); + assert_eq!(resolve_ask_rendezvous_ceiling(Some(9999)), Duration::from_millis(9999)); + assert_eq!(resolve_ask_rendezvous_ceiling(Some(0)), ASK_RENDEZVOUS_CEILING); + assert_eq!(resolve_ask_rendezvous_ceiling(None), ASK_RENDEZVOUS_CEILING); + } +}