From 77169101cbc0575261eac4dff1094c89a7d32348 Mon Sep 17 00:00:00 2001 From: Blomios Date: Tue, 8 Sep 2026 20:29:48 +0200 Subject: [PATCH] =?UTF-8?q?fix(orchestrator,agent):=20rendezvous=20cross-p?= =?UTF-8?q?roject=20r=C3=A9arm=C3=A9=20par=20jeton=20d'activit=C3=A9=20liv?= =?UTF-8?q?e=20model-agnostique=20=E2=80=94=20#287=20(le=20watchdog=20d'in?= =?UTF-8?q?activit=C3=A9=20du=20rendez-vous=20expirait=20=C3=A0=20600s=20s?= =?UTF-8?q?ur=20un=20tour=20structured=20actif=20car=20seul=20le=20transcr?= =?UTF-8?q?ipt=20Claude=20servait=20de=20signe=20de=20vie;=20nouveau=20dra?= =?UTF-8?q?in=5Fwith=5Freadiness=5Fannouncements=5Fand=5Factivity=5Foutcom?= =?UTF-8?q?e=20exposant=20un=20jeton=20monotone=20AtomicU64=20incr=C3=A9me?= =?UTF-8?q?nt=C3=A9=20par=20chaque=20ReplyEvent=20non=20terminal=20du=20ta?= =?UTF-8?q?p=20send=5Fwith=5Ftap,=20pass=C3=A9=20au=20watchdog=20via=20Ask?= =?UTF-8?q?WatchdogActivity=20pour=20r=C3=A9armer=20la=20fen=C3=AAtre=20qu?= =?UTF-8?q?elle=20que=20soit=20l'adapter;=20la=20sonde=20transcript=20devi?= =?UTF-8?q?ent=20fallback=20externe;=20QA=20verte:=20orchestrator=5Fservic?= =?UTF-8?q?e=2081/81,=20rendezvous=205/5,=20mcp=5Fserver=2040/40,=20backen?= =?UTF-8?q?d=2087+7=20ignored,=20cargo=20fmt=20--check;=20scripts/=20hors?= =?UTF-8?q?=20p=C3=A9rim=C3=A8tre)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- crates/application/src/agent/mod.rs | 3 +- crates/application/src/agent/structured.rs | 47 ++++- .../application/src/orchestrator/service.rs | 84 +++++++-- .../application/tests/orchestrator_service.rs | 165 +++++++++++++++++- crates/backend/src/lib.rs | 6 +- 5 files changed, 277 insertions(+), 28 deletions(-) diff --git a/crates/application/src/agent/mod.rs b/crates/application/src/agent/mod.rs index 6d839c7e..ec7b9643 100644 --- a/crates/application/src/agent/mod.rs +++ b/crates/application/src/agent/mod.rs @@ -23,7 +23,8 @@ pub use session_limit::{AgentResumer, SessionLimitService, RESUME_PROMPT}; pub use structured::{ drain_reply_stream_with_readiness, drain_with_readiness, drain_with_readiness_and_announcements, drain_with_readiness_and_announcements_outcome, - drain_with_readiness_outcome, send_blocking, AnnouncementPublisher, TurnOutcome, + drain_with_readiness_announcements_and_activity_outcome, drain_with_readiness_outcome, + send_blocking, AnnouncementPublisher, TurnOutcome, }; pub use catalogue::{ diff --git a/crates/application/src/agent/structured.rs b/crates/application/src/agent/structured.rs index 98d0af06..fcfeea09 100644 --- a/crates/application/src/agent/structured.rs +++ b/crates/application/src/agent/structured.rs @@ -12,6 +12,7 @@ //! DRY : un seul chemin de lecture (le flux). `send_blocking` n'est qu'un *consom- //! mateur* du même flux que la cellule chat utilise pour le rendu incrémental. +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; use std::time::{Duration, SystemTime, UNIX_EPOCH}; @@ -203,15 +204,49 @@ pub async fn drain_with_readiness_and_announcements_outcome( agent: RuntimeAgentKey, announcements: Option, ) -> Result { - let Some(publisher) = announcements else { + if announcements.is_none() { return drain_with_readiness_outcome(session, prompt, timeout, mediator, agent).await; - }; + } + drain_with_readiness_announcements_and_activity_outcome( + session, + prompt, + timeout, + mediator, + agent, + announcements, + None, + ) + .await +} +/// Variante du drain d'annonces qui expose aussi un jeton monotone d'activité live. +/// +/// Le jeton avance pour chaque [`ReplyEvent`] non terminal reçu via +/// [`AgentSession::send_with_tap`], donc pendant que `send_with_tap` travaille encore +/// et avant que son flux batch final soit disponible. Il permet au watchdog du +/// rendez-vous de distinguer un tour structured actif d'un vrai silence, quel que +/// soit le provider. `Final` reste exclusivement traité par le drain normal. +pub async fn drain_with_readiness_announcements_and_activity_outcome( + session: &dyn AgentSession, + prompt: &str, + timeout: Option, + mediator: &dyn InputMediator, + agent: RuntimeAgentKey, + announcements: Option, + activity: Option>, +) -> Result { let (tap_tx, tap_rx) = std::sync::mpsc::channel(); - let live_publisher = publisher.clone(); + let live_publisher = announcements.clone(); let live_pump = std::thread::spawn(move || { for event in tap_rx { - live_publisher.publish_event(&event); + if !matches!(event, ReplyEvent::Final { .. }) { + if let Some(activity) = &activity { + activity.fetch_add(1, Ordering::Relaxed); + } + } + if let Some(publisher) = &live_publisher { + publisher.publish_event(&event); + } } }); @@ -230,7 +265,9 @@ pub async fn drain_with_readiness_and_announcements_outcome( if !matches!(event, ReplyEvent::Final { .. }) { mediator.mark_alive(agent); } - publisher.publish_event(event); + if let Some(publisher) = &announcements { + publisher.publish_event(event); + } }, |signal| { if signal == ReadinessSignal::TurnEnded { diff --git a/crates/application/src/orchestrator/service.rs b/crates/application/src/orchestrator/service.rs index ea4f6b3d..80ae5693 100644 --- a/crates/application/src/orchestrator/service.rs +++ b/crates/application/src/orchestrator/service.rs @@ -15,6 +15,7 @@ use std::collections::HashMap; use std::path::{Component, Path, PathBuf}; +use std::sync::atomic::{AtomicU64, Ordering as AtomicOrdering}; use std::sync::{Arc, Mutex as StdMutex}; use std::time::{Duration, Instant}; @@ -40,10 +41,10 @@ use crate::conversation::RecordTurn; use crate::memory::HarvestMemoryFromTurn; use crate::agent::{ - drain_with_readiness_and_announcements_outcome, AnnouncementPublisher, CreateAgentFromScratch, - CreateAgentInput, LaunchAgent, LaunchAgentInput, ListAgents, ListAgentsInput, - ManagedTempExposureRequest, McpRuntime, ReattachDecision, SessionLimitService, TurnOutcome, - UpdateAgentContext, UpdateAgentContextInput, CODEX_SUBMIT_DELAY_MS, + drain_with_readiness_announcements_and_activity_outcome, AnnouncementPublisher, + CreateAgentFromScratch, CreateAgentInput, LaunchAgent, LaunchAgentInput, ListAgents, + ListAgentsInput, ManagedTempExposureRequest, McpRuntime, ReattachDecision, SessionLimitService, + TurnOutcome, UpdateAgentContext, UpdateAgentContextInput, CODEX_SUBMIT_DELAY_MS, }; use crate::background::{SpawnBackgroundCommand, SpawnBackgroundCommandInput}; use crate::error::AppError; @@ -209,12 +210,11 @@ const ASK_AGENT_TIMEOUT: Duration = Duration::from_secs(600); /// `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. +/// Sur le chemin structured, cette sonde externe complète le tap live +/// [`domain::ports::AgentSession::send_with_tap`] : elle reste utile lorsque l'adapter +/// ne produit aucun événement intermédiaire. 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, @@ -224,6 +224,11 @@ pub type AskLivenessProbe = Arc< + Sync, >; +struct AskWatchdogActivity { + started: Instant, + local: Arc, +} + /// 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. @@ -2470,13 +2475,15 @@ impl OrchestratorService { target: agent_id, ticket: ticket_id, }); - let drain = drain_with_readiness_and_announcements_outcome( + let local_activity = Arc::new(AtomicU64::new(0)); + let drain = drain_with_readiness_announcements_and_activity_outcome( session, &task, None, input.as_ref(), agent_key, announcement_publisher, + Some(Arc::clone(&local_activity)), ); // L'attente du rendez-vous structured, rendue comme `Result` @@ -2525,7 +2532,17 @@ impl OrchestratorService { // 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) + .run_ask_with_watchdog( + wait, + turn_timeout, + &project.root, + agent_id, + target, + AskWatchdogActivity { + started, + local: local_activity, + }, + ) .await { WatchdogOutcome::Resolved(Ok(content)) => { @@ -3481,19 +3498,54 @@ impl OrchestratorService { project_root: &domain::project::ProjectPath, agent_id: AgentId, target: &str, - started: Instant, + activity: AskWatchdogActivity, ) -> WatchdogOutcome { - let has_probe = self.ask_liveness_probe.is_some(); + let AskWatchdogActivity { + started, + local: local_activity, + } = activity; + // Le tap structured est toujours observable ; le transcript reste un fallback + // utile pour les providers qui ne produisent aucun événement live pendant un + // long tour. Les deux sources sont fusionnées en un jeton monotone cumulatif : + // la disparition/régression de l'une ne peut jamais masquer le progrès de l'autre. + #[derive(Default)] + struct CompositeActivity { + local: u64, + transcript: Option, + token: u64, + } + let probe = self.ask_liveness_probe.clone(); let root = project_root.clone(); + let composite = Arc::new(StdMutex::new(CompositeActivity::default())); let probe_fn = move || { let probe = probe.clone(); let root = root.clone(); + let local_activity = Arc::clone(&local_activity); + let composite = Arc::clone(&composite); async move { - match &probe { + let transcript = match &probe { Some(p) => p(root, agent_id).await, None => None, + }; + let local = local_activity.load(AtomicOrdering::Relaxed); + let mut state = composite.lock().unwrap(); + let local_delta = local.saturating_sub(state.local); + let transcript_delta = match (state.transcript, transcript) { + (Some(previous), Some(current)) => current.saturating_sub(previous), + (None, Some(_)) => 1, + _ => 0, + }; + state.local = state.local.max(local); + if let Some(current) = transcript { + state.transcript = + Some(state.transcript.map_or(current, |old| old.max(current))); } + state.token = state + .token + .saturating_add(local_delta) + .saturating_add(transcript_delta); + Some(state.token) } }; let ext_target = target.to_owned(); @@ -3506,7 +3558,7 @@ impl OrchestratorService { window, self.ask_ceiling, started, - has_probe, + true, probe_fn, move |elapsed| { crate::diag!( diff --git a/crates/application/tests/orchestrator_service.rs b/crates/application/tests/orchestrator_service.rs index 1fbd0561..1dc13ee5 100644 --- a/crates/application/tests/orchestrator_service.rs +++ b/crates/application/tests/orchestrator_service.rs @@ -29,8 +29,8 @@ use domain::ports::{ SpawnSpec, StoreError, }; use domain::profile::{ - AgentProfile, ContextInjection, McpCapability, McpConfigStrategy, McpTransport, - StructuredAdapter, + AgentProfile, ContextInjection, LivenessStrategy, McpCapability, McpConfigStrategy, + McpTransport, StructuredAdapter, }; use domain::project::{Project, ProjectPath}; use domain::remote::RemoteRef; @@ -4892,6 +4892,81 @@ struct BackgroundThenFinalSession { service: Arc, } +struct TappedDelayedFinalSession { + id: SessionId, +} + +#[async_trait] +impl AgentSession for TappedDelayedFinalSession { + fn id(&self) -> SessionId { + self.id + } + + fn conversation_id(&self) -> Option { + Some(format!("engine-{}", self.id)) + } + + async fn send(&self, _prompt: &str) -> Result { + std::future::pending().await + } + + async fn send_with_tap( + &self, + _prompt: &str, + tap: std::sync::mpsc::Sender, + ) -> Result { + for event in [ + ReplyEvent::Heartbeat, + ReplyEvent::Progress { + progress: domain::ports::ReplyProgress::new( + domain::ports::ReplyProgressSource::ProviderNative, + domain::ports::ReplyProgressKind::Message, + domain::ports::ReplyProgressStage::Delta, + "working", + ), + }, + ReplyEvent::Heartbeat, + ] { + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + tap.send(event).expect("tap receiver remains live"); + } + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + Ok(Box::new( + vec![ReplyEvent::Final { + content: "late final".to_owned(), + }] + .into_iter(), + )) + } + + async fn shutdown(&self) -> Result<(), AgentSessionError> { + Ok(()) + } +} + +struct SilentStructuredSession { + id: SessionId, +} + +#[async_trait] +impl AgentSession for SilentStructuredSession { + fn id(&self) -> SessionId { + self.id + } + + fn conversation_id(&self) -> Option { + Some(format!("engine-{}", self.id)) + } + + async fn send(&self, _prompt: &str) -> Result { + std::future::pending().await + } + + async fn shutdown(&self) -> Result<(), AgentSessionError> { + Ok(()) + } +} + #[async_trait] impl AgentSession for BackgroundThenFinalSession { fn id(&self) -> SessionId { @@ -4992,7 +5067,16 @@ fn structured_ask_fixture_with_background( contexts: FakeContexts, enable_background: bool, ) -> StructuredAskFixture { - let profiles = Arc::new(FakeProfiles::new(vec![claude_profile()])); + structured_ask_fixture_with_profile(contexts, enable_background, claude_profile(), None) +} + +fn structured_ask_fixture_with_profile( + contexts: FakeContexts, + enable_background: bool, + profile: AgentProfile, + liveness_probe: Option, +) -> StructuredAskFixture { + let profiles = Arc::new(FakeProfiles::new(vec![profile])); let sessions = Arc::new(TerminalSessions::new()); let pty = FakePty::new(sid(777)); let bus = SpyBus::default(); @@ -5079,6 +5163,10 @@ fn structured_ask_fixture_with_background( // insérée par `launch_structured`. .with_structured(Arc::clone(&structured)); + if let Some(probe) = liveness_probe { + service = service.with_ask_liveness_probe(probe); + } + if enable_background { service = service .with_background_tasks( @@ -5241,6 +5329,77 @@ async fn ask_warm_structured_target_reuses_session_without_relaunch() { ); } +/// Régression #287 : la sonde transcript peut être stable (notamment inexistante pour +/// un provider non-Claude), tandis que `send_with_tap` publie réellement l'activité du +/// tour. Chaque événement non terminal doit réarmer la fenêtre jusqu'au `Final` tardif. +#[tokio::test] +async fn ask_structured_rearms_inactivity_window_from_send_with_tap() { + let agent = scratch_agent(aid(1), "architect", "agents/architect.md"); + let profile = claude_profile() + .with_liveness(LivenessStrategy::new(None, Some(30)).expect("valid short test window")); + let stable_probe: application::AskLivenessProbe = + Arc::new(|_root, _agent_id| Box::pin(async { Some(42) })); + let fx = structured_ask_fixture_with_profile( + FakeContexts::with_agent(&agent, "# persona"), + false, + profile, + Some(stable_probe), + ); + fx.structured.insert_in_project( + project_id(), + Arc::new(TappedDelayedFinalSession { id: sid(9900) }), + aid(1), + nid(701), + ); + + let started = std::time::Instant::now(); + let out = fx + .service + .dispatch(&project(), cmd(ASK_JSON)) + .await + .expect("tap activity must keep the rendezvous alive until Final"); + + assert_eq!(out.reply.as_deref(), Some("late final")); + assert!( + started.elapsed() >= std::time::Duration::from_millis(70), + "the Final must arrive after multiple 30 ms inactivity windows" + ); +} + +/// Sans événement tap, sans progrès transcript et sans `Final`, le même chemin doit +/// conserver le timeout d'inactivité au bout d'une fenêtre complète. +#[tokio::test] +async fn ask_structured_without_tap_probe_progress_or_final_times_out() { + let agent = scratch_agent(aid(1), "architect", "agents/architect.md"); + let profile = claude_profile() + .with_liveness(LivenessStrategy::new(None, Some(30)).expect("valid short test window")); + let stable_probe: application::AskLivenessProbe = + Arc::new(|_root, _agent_id| Box::pin(async { Some(42) })); + let fx = structured_ask_fixture_with_profile( + FakeContexts::with_agent(&agent, "# persona"), + false, + profile, + Some(stable_probe), + ); + fx.structured.insert_in_project( + project_id(), + Arc::new(SilentStructuredSession { id: sid(9901) }), + aid(1), + nid(702), + ); + + let started = std::time::Instant::now(); + let err = fx + .service + .dispatch(&project(), cmd(ASK_JSON)) + .await + .expect_err("a fully silent structured turn must time out"); + + assert_eq!(err.code(), "PROCESS", "got {err:?}"); + assert!(err.to_string().to_ascii_lowercase().contains("timed out")); + assert!(started.elapsed() < std::time::Duration::from_secs(1)); +} + /// Régression live : une cible structurée peut déjà être vivante en PTY parce qu'elle /// a été ouverte depuis la surface humaine/menu historique. Le chemin `ask` headless /// doit alors créer une session structurée **sans fermer** ce PTY visible : les deux diff --git a/crates/backend/src/lib.rs b/crates/backend/src/lib.rs index 6c70e368..0827a400 100644 --- a/crates/backend/src/lib.rs +++ b/crates/backend/src/lib.rs @@ -3346,9 +3346,9 @@ impl BackendCore { // 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. + // transcript Claude), sous le plafond ci-dessous. Cette sonde est le fallback + // externe du tap live model-agnostique alimenté par `AgentSession::send_with_tap` ; + // l'une ou l'autre source suffit désormais à réarmer la fenêtre. // Keyée par (project_root, agent_id) : le run-dir transcript = `/.ideai/ // run/`, encodé par Claude sous `/.claude/projects/...`. .with_ask_liveness_probe({