feat(orchestrator): watchdog d'inactivité réarmable + plafond du rendez-vous inter-agents

Redessine la borne de fin de tour du rendez-vous `idea_ask_agent` ⇄ `idea_reply` :
au lieu d'un timeout plat (qui coupait à 600 s un seul long tour de la cible, cf.
T7), la borne devient une **fenêtre d'inactivité réarmée** à chaque progrès observé
de la cible, sous un **plafond absolu** (défaut 4 h, réglable via
`IDEA_ASK_RENDEZVOUS_CEILING_MS`).

- Sonde d'activité (`transcript_activity_token`, inspector) : jeton monotone =
  octets cumulés des `.jsonl` de la cible. Croît même pendant un seul long tour
  sans `turn_duration` ⇒ détecte « vivant et au travail » mi-tour. Best-effort,
  sans effet de bord ; folder absent/illisible ⇒ « pas de progrès ».
- Watchdog (`run_inactivity_watchdog`, nouveau module `orchestrator/rendezvous`) :
  fenêtre réarmable + plafond, fallback timeout plat si aucune sonde (zéro régression).
- Issue typée distincte `TargetCeilingActive` (code `RENDEZVOUS_CEILING_ACTIVE`) :
  une cible **active** stoppée au plafond n'est jamais confondue avec un
  `TargetReturnedNoReply` (silence) ; le message guide « ne pas retenter à l'aveugle ».
- Câblage composition-root (`state.rs`) : sonde résolue nom→AgentId→run-dir transcript,
  branchée sur le service et sur l'McpServer (`AskActivityProbe`, `with_ask_ceiling`).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-06-24 13:32:19 +02:00
parent 1efe2f11dc
commit 6050a6da5f
12 changed files with 1036 additions and 140 deletions

View File

@ -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<AgentSessionError> 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"));
}
}

View File

@ -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,

View File

@ -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,
};

View File

@ -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<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);
}
}

View File

@ -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<Box<dyn std::future::Future<Output = Option<u64>> + 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<Arc<dyn LiveStateReadProvider>>,
/// **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<AskLivenessProbe>,
/// 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<String, AppError>`
// 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<Option<domain::AgentId>, 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<R>(
&self,
wait: impl std::future::Future<Output = R>,
window: Duration,
project_root: &domain::project::ProjectPath,
agent_id: AgentId,
target: &str,
started: Instant,
) -> WatchdogOutcome<R> {
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