fix(orchestrator,agent): rendezvous cross-project réarmé par jeton d'activité live model-agnostique — #287 (le watchdog d'inactivité du rendez-vous expirait à 600s sur un tour structured actif car seul le transcript Claude servait de signe de vie; nouveau drain_with_readiness_announcements_and_activity_outcome exposant un jeton monotone AtomicU64 incrémenté par chaque ReplyEvent non terminal du tap send_with_tap, passé au watchdog via AskWatchdogActivity pour réarmer la fenêtre quelle que soit l'adapter; la sonde transcript devient fallback externe; QA verte: orchestrator_service 81/81, rendezvous 5/5, mcp_server 40/40, backend 87+7 ignored, cargo fmt --check; scripts/ hors périmètre)

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::{

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<AnnouncementPublisher>,
 ) -> Result<TurnOutcome, AgentSessionError> {
-    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<Duration>,
+    mediator: &dyn InputMediator,
+    agent: RuntimeAgentKey,
+    announcements: Option<AnnouncementPublisher>,
+    activity: Option<Arc<AtomicU64>>,
+) -> Result<TurnOutcome, AgentSessionError> {
     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 {

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<AtomicU64>,
+}
+
 /// 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<String, AppError>`
@@ -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<R> {
-        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<u64>,
+            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!(

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<OrchestratorService>,
 }
 
+struct TappedDelayedFinalSession {
+    id: SessionId,
+}
+
+#[async_trait]
+impl AgentSession for TappedDelayedFinalSession {
+    fn id(&self) -> SessionId {
+        self.id
+    }
+
+    fn conversation_id(&self) -> Option<String> {
+        Some(format!("engine-{}", self.id))
+    }
+
+    async fn send(&self, _prompt: &str) -> Result<ReplyStream, AgentSessionError> {
+        std::future::pending().await
+    }
+
+    async fn send_with_tap(
+        &self,
+        _prompt: &str,
+        tap: std::sync::mpsc::Sender<ReplyEvent>,
+    ) -> Result<ReplyStream, AgentSessionError> {
+        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<String> {
+        Some(format!("engine-{}", self.id))
+    }
+
+    async fn send(&self, _prompt: &str) -> Result<ReplyStream, AgentSessionError> {
+        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<application::AskLivenessProbe>,
+) -> 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

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 = `<root>/.ideai/
             // run/<agent_id>`, encodé par Claude sous `<home>/.claude/projects/...`.
             .with_ask_liveness_probe({