diff --git a/.ideai/tickets/289/carnet.md b/.ideai/tickets/289/carnet.md new file mode 100644 index 00000000..2b4cbeda --- /dev/null +++ b/.ideai/tickets/289/carnet.md @@ -0,0 +1,197 @@ +--- +issueRef: "#289" +version: 4 +updatedBy: {"kind":"agent","agent_id":"a6c6ea12-bfc6-4bdc-8031-324102dfa34d"} +updatedAt: 1788900709736 +--- +## 2026-09-08 — Main + +Demande utilisateur : garantir que si une cible appelée via `ask_agent` se restart, le demandeur attend la fin du restart et reçoit le travail post-restart rattaché au même agent, aussi bien intra-projet qu'en cross-project. + +Cadrage Architect : +- Identité stable par `RuntimeAgentKey` `(projectId, agentId)`. +- Le `conversation_id` logique et le verrou de tour doivent survivre au restart. +- Un restart accepté est une activité de contrôle du rendez-vous : ne pas classifier la fermeture de l'ancienne session comme no-reply/timeout d'inactivité. +- Cross-project : conserver la corrélation demandeur `(requesterProjectId, requesterAgentId, correlationId)` jusqu'au résultat post-restart. +- Cancel pendant restart doit cascader et nettoyer rendez-vous/workstate/registre cross-project. +- Échec de restart => erreur explicite, pas silence. + +Branche Git retenue : `fix/289-ask-agent-during-target-restart` depuis `develop` @ `db91b00`. + +Prochaines étapes : DevBackend implémente le contrat backend et tests ; QA valide same-project, cross-project, watchdog, failure/cancel. + +## 2026-09-08 — DevBackend + +Correctif backend implémenté sur `fix/289-ask-agent-during-target-restart`. + +Fichiers modifiés : +- `crates/application/src/orchestrator/service.rs` +- `crates/application/src/orchestrator/mod.rs` +- `crates/application/src/lib.rs` +- `crates/application/tests/orchestrator_service.rs` +- `crates/backend/src/lib.rs` + +Comportement livré : +- coordination stable par `RuntimeAgentKey (projectId, agentId)` entre le rendez-vous applicatif et le scheduler de self-restart ; +- conservation du verrou de tour, du ticket logique et du `conversation_id` pendant le remplacement de la session concrète ; +- abandon du drain de l'ancienne session sans no-reply, puis capture du `Final` du tour de continuation sur la session relancée ; +- activité de contrôle restart intégrée au watchdog d'inactivité ; +- corrélation cross-project conservée jusqu'au résultat post-restart ; +- échec de restart remonté explicitement ; cancel pendant la fenêtre sans runtime réveille l'ask, nettoie la corrélation et empêche la relance de continuation ; +- le scheduler standalone conserve son comportement historique lorsque aucun ask n'est actif. + +Tests exécutés : +- `cargo test -p application --test orchestrator_service` — 85 passés ; +- `cargo test -p application --lib` — 171 passés ; +- `cargo test -p backend --lib` — 89 passés, 7 ignorés (tests socket nécessitant une permission locale) ; +- `cargo check -p backend` — OK ; +- `cargo fmt --all -- --check` — OK. + +Tests de régression ajoutés : same-project Final post-restart, cross-project + nettoyage corrélation, faux no-reply interdit, failure explicite, cancel pendant le gap de restart, watchdog réarmé par activité de contrôle. + +Limites : pas de test E2E avec un vrai processus fournisseur/AppImage dans ce lot ; la validation est faite avec les ports/sessions structurés et le scheduler backend compilé. Aucun fichier frontend modifié. Aucun commit effectué par DevBackend. + +## 2026-09-08 — QA + +Verdict : VERT pour le ticket #289 sur `fix/289-ask-agent-during-target-restart`. + +Tests QA ajoutés dans `crates/application/tests/orchestrator_service.rs` : +- `ask_target_restart_cleans_busy_workstate_and_replaces_structured_registry_entry` : succès same-project post-restart, cible revenue `Idle`, FIFO vide, live-state `Done`, ancienne session structurée absente, session de remplacement enregistrée sur le même agent logique. +- `same_project_stop_during_restart_releases_ask_and_leaves_no_replacement_session` : cancel/stop same-project pendant le gap de restart, ask réveillé en erreur explicite de cancel, busy/FIFO/live-state nettoyés, aucune session de remplacement enregistrée. + +Commandes exécutées et sorties pertinentes : + +```bash +git status --short --branch +``` + +```text +## fix/289-ask-agent-during-target-restart + M .ideai/tickets/counter.json + M .ideai/tickets/index.json + M crates/application/src/lib.rs + M crates/application/src/orchestrator/mod.rs + M crates/application/src/orchestrator/service.rs + M crates/application/tests/orchestrator_service.rs + M crates/backend/src/lib.rs +?? .ideai/tickets/289/ +?? scripts/ +``` + +```bash +cargo test -p application --test orchestrator_service ask_target_restart_keeps_logical_turn_and_returns_post_restart_final -- --nocapture +``` + +```text +test ask_target_restart_keeps_logical_turn_and_returns_post_restart_final ... ok +test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 84 filtered out; finished in 0.02s +``` + +```bash +cargo test -p application --test orchestrator_service cross_project_ask_preserves_correlation_across_target_restart -- --nocapture +``` + +```text +test cross_project_ask_preserves_correlation_across_target_restart ... ok +test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 84 filtered out; finished in 0.02s +``` + +```bash +cargo test -p application --test orchestrator_service ask_target_restart_failure_is_explicit_not_no_reply -- --nocapture +``` + +```text +test ask_target_restart_failure_is_explicit_not_no_reply ... ok +test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 84 filtered out; finished in 0.01s +``` + +```bash +cargo test -p application --test orchestrator_service cross_project_cancel_during_restart_releases_ask_and_correlation -- --nocapture +``` + +```text +test cross_project_cancel_during_restart_releases_ask_and_correlation ... ok +test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 84 filtered out; finished in 0.00s +``` + +```bash +cargo test -p application --lib restart_control_activity_rearms_inactivity_watchdog -- --nocapture +``` + +```text +test orchestrator::service::tests::restart_control_activity_rearms_inactivity_watchdog ... ok +test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 170 filtered out; finished in 0.05s +``` + +```bash +cargo test -p application --test orchestrator_service ask_target_restart_cleans_busy_workstate_and_replaces_structured_registry_entry -- --nocapture +``` + +```text +test ask_target_restart_cleans_busy_workstate_and_replaces_structured_registry_entry ... ok +test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 86 filtered out; finished in 0.02s +``` + +```bash +cargo test -p application --test orchestrator_service same_project_stop_during_restart_releases_ask_and_leaves_no_replacement_session -- --nocapture +``` + +```text +test same_project_stop_during_restart_releases_ask_and_leaves_no_replacement_session ... ok +test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 86 filtered out; finished in 0.00s +``` + +```bash +cargo fmt --all -- --check +``` + +```text +OK (aucune sortie) +``` + +```bash +cargo test -p application --test orchestrator_service +``` + +```text +running 87 tests +test result: ok. 87 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.09s +``` + +```bash +cargo test -p application --lib +``` + +```text +running 171 tests +test result: ok. 171 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.15s +``` + +```bash +cargo test -p backend --lib +``` + +```text +running 96 tests +test result: ok. 89 passed; 0 failed; 7 ignored; 0 measured; 0 filtered out; finished in 0.04s +``` + +```bash +cargo check -p backend +``` + +```text +Finished `dev` profile [unoptimized + debuginfo] target(s) in 3.37s +``` + +Couverture validée : +- same-project : Final récupéré après restart et test QA de nettoyage busy/workstate/registre. +- cross-project : corrélation conservée jusqu'au Final post-restart puis supprimée. +- pas de faux no-reply/timeout pendant restart : failure restart reste `PROCESS` explicite, watchdog réarmé par activité de contrôle restart. +- failure restart : erreur explicite contenant la cause provider. +- cancel pendant restart : cross-project et same-project couverts. +- nettoyage busy/workstate/registry : FIFO vide, `AgentBusyState::Idle`, live-state `Done`, registre structuré remplacé après succès et non recréé après cancel. + +Risques résiduels : +- Pas d'E2E AppImage / vrai provider relancé : la preuve porte sur les ports applicatifs, les sessions structurées de test et le câblage backend compilé. +- Les 7 tests backend ignorés demandent une permission de bind socket locale ; ils restent hors preuve dans cet environnement. diff --git a/.ideai/tickets/289/issue.md b/.ideai/tickets/289/issue.md new file mode 100644 index 00000000..c70ca12b --- /dev/null +++ b/.ideai/tickets/289/issue.md @@ -0,0 +1,17 @@ +--- +id: "ae7f4f57-087e-463e-9390-33727e751d4f" +number: 289 +title: "Garantir ask_agent pendant restart de la cible, same-project et cross-project" +status: "qa" +priority: "high" +sprint: null +links: [] +agentRefs: [] +attachments: [] +createdBy: {"kind":"agent","agent_id":"a6c6ea12-bfc6-4bdc-8031-324102dfa34d"} +updatedBy: {"kind":"agent","agent_id":"a6c6ea12-bfc6-4bdc-8031-324102dfa34d"} +createdAt: 1788899102915 +updatedAt: 1788900574986 +version: 3 +--- +Vérifier et corriger le comportement suivant : lorsqu'un agent cible a été appelé via ask_agent puis se restart, le demandeur doit attendre la fin du restart et le travail repris/après restart doit rester rattaché à cette même cible. Le contrat doit couvrir les appels intra-projet et cross-project, sans perte de résultat ni attribution à un mauvais agent. \ No newline at end of file diff --git a/.ideai/tickets/counter.json b/.ideai/tickets/counter.json index 3c9828a5..194068b8 100644 --- a/.ideai/tickets/counter.json +++ b/.ideai/tickets/counter.json @@ -1,3 +1,3 @@ { - "nextNumber": 289 + "nextNumber": 290 } \ No newline at end of file diff --git a/.ideai/tickets/index.json b/.ideai/tickets/index.json index d785c482..24a24db9 100644 --- a/.ideai/tickets/index.json +++ b/.ideai/tickets/index.json @@ -3949,6 +3949,20 @@ "agent_id": "a6c6ea12-bfc6-4bdc-8031-324102dfa34d" }, "updatedAt": 1788898181382 + }, + { + "issueRef": "#289", + "path": "289", + "title": "Garantir ask_agent pendant restart de la cible, same-project et cross-project", + "status": "qa", + "priority": "high", + "sprint": null, + "assignedAgentIds": [], + "createdBy": { + "kind": "agent", + "agent_id": "a6c6ea12-bfc6-4bdc-8031-324102dfa34d" + }, + "updatedAt": 1788900574986 } ] } \ No newline at end of file diff --git a/crates/application/src/lib.rs b/crates/application/src/lib.rs index 1d2d164d..5cd85674 100644 --- a/crates/application/src/lib.rs +++ b/crates/application/src/lib.rs @@ -168,8 +168,9 @@ pub use orchestrator::{ AgentRestartScheduled, AgentSelfRestartScheduler, AgentWakeService, AskLivenessProbe, ContextGuardUseCases, LiveStateProvider, LiveStateReadProvider, McpRuntimeProvider, OrchestratorOutcome, OrchestratorService, ProposeContext, ReadContext, ReadMemory, - RecordTurnProvider, WakeSessionProvider, WatchdogOutcome, WriteMemory, - DEFAULT_RENDEZVOUS_CEILING, DEFAULT_RENDEZVOUS_WINDOW, + RecordTurnProvider, RestartRendezvousCoordinator, RestartRendezvousState, WakeSessionProvider, + WatchdogOutcome, WriteMemory, DEFAULT_RENDEZVOUS_CEILING, DEFAULT_RENDEZVOUS_WINDOW, + SELF_RESTART_CONTINUATION_PROMPT, }; pub use permission::{ GetProjectPermissions, GetProjectPermissionsInput, GetProjectPermissionsOutput, diff --git a/crates/application/src/orchestrator/mod.rs b/crates/application/src/orchestrator/mod.rs index 9d808012..1b8b9ca0 100644 --- a/crates/application/src/orchestrator/mod.rs +++ b/crates/application/src/orchestrator/mod.rs @@ -21,6 +21,7 @@ pub use rendezvous::{ pub use service::{ AgentRestartScheduled, AgentSelfRestartScheduler, AskLivenessProbe, ContextGuardUseCases, LiveStateProvider, LiveStateReadProvider, McpRuntimeProvider, OrchestratorOutcome, - OrchestratorService, RecordTurnProvider, + OrchestratorService, RecordTurnProvider, RestartRendezvousCoordinator, RestartRendezvousState, + SELF_RESTART_CONTINUATION_PROMPT, }; pub use wake::{AgentWakeService, WakeSessionProvider}; diff --git a/crates/application/src/orchestrator/service.rs b/crates/application/src/orchestrator/service.rs index d64ea14e..3eeb2d27 100644 --- a/crates/application/src/orchestrator/service.rs +++ b/crates/application/src/orchestrator/service.rs @@ -19,7 +19,7 @@ use std::sync::atomic::{AtomicU64, Ordering as AtomicOrdering}; use std::sync::{Arc, Mutex as StdMutex}; use std::time::{Duration, Instant}; -use tokio::sync::Mutex as AsyncMutex; +use tokio::sync::{watch, Mutex as AsyncMutex}; use domain::conversation::{ConversationParty, ConversationRegistry, SessionRef, WaitForGraph}; use domain::conversation_log::{ConversationTurn, TurnId, TurnRole}; @@ -82,6 +82,330 @@ pub trait AgentSelfRestartScheduler: Send + Sync { ) -> Result; } +/// Internal continuation prompt used when a self-restart is handed back to the +/// synchronous inter-agent rendezvous that was driving the old runtime. +pub const SELF_RESTART_CONTINUATION_PROMPT: &str = + "IdeA vient de te redémarrer. Reprends immédiatement le travail en cours là où tu l'avais laissé."; + +/// Control-plane state of a self-restart affecting an active logical rendezvous. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum RestartRendezvousState { + /// No restart currently affects the logical turn. + Idle, + /// The old runtime is being replaced. + Restarting { + /// Stable restart correlation id. + restart_id: String, + }, + /// The replacement structured session is available. + Ready { + /// Stable restart correlation id. + restart_id: String, + }, + /// Runtime replacement failed. + Failed { + /// Stable restart correlation id. + restart_id: String, + /// Explicit adapter failure. + message: String, + }, + /// The owning logical rendezvous was cancelled. + Cancelled { + /// Stable restart correlation id. + restart_id: String, + }, +} + +struct RestartRendezvousEntry { + active_waiters: usize, + activity: u64, + state: RestartRendezvousState, + tx: watch::Sender, +} + +/// Stable, runtime-keyed seam shared by the orchestrator and the detached restart +/// adapter. It keeps a logical ask alive while its concrete provider session is +/// replaced. +#[derive(Default)] +pub struct RestartRendezvousCoordinator { + entries: StdMutex>, +} + +impl RestartRendezvousCoordinator { + /// Creates an empty runtime-keyed coordinator. + #[must_use] + pub fn new() -> Self { + Self::default() + } + + fn register(self: &Arc, key: RuntimeAgentKey) -> RestartRendezvousRegistration { + let mut entries = self + .entries + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let entry = entries.entry(key).or_insert_with(|| { + let state = RestartRendezvousState::Idle; + let (tx, _) = watch::channel(state.clone()); + RestartRendezvousEntry { + active_waiters: 0, + activity: 0, + state, + tx, + } + }); + entry.active_waiters = entry.active_waiters.saturating_add(1); + RestartRendezvousRegistration { + coordinator: Arc::clone(self), + key, + rx: entry.tx.subscribe(), + } + } + + /// Announces a restart synchronously. Returns whether an active ask owns the + /// continuation and therefore must receive the post-restart `Final`. + pub fn begin(&self, key: RuntimeAgentKey, restart_id: String) -> bool { + let mut entries = self + .entries + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let Some(entry) = entries.get_mut(&key) else { + return false; + }; + if entry.active_waiters == 0 { + return false; + } + entry.activity = entry.activity.saturating_add(1); + entry.state = RestartRendezvousState::Restarting { restart_id }; + entry.tx.send_replace(entry.state.clone()); + true + } + + /// Publishes that the replacement structured session can accept continuation. + pub fn ready(&self, key: RuntimeAgentKey, restart_id: &str) { + self.transition(key, restart_id, |restart_id| { + RestartRendezvousState::Ready { restart_id } + }); + } + + fn resume_taken(&self, key: RuntimeAgentKey, restart_id: &str) { + let mut entries = self + .entries + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let Some(entry) = entries.get_mut(&key) else { + return; + }; + if !matches!( + &entry.state, + RestartRendezvousState::Ready { restart_id: current } if current == restart_id + ) { + return; + } + entry.state = RestartRendezvousState::Idle; + entry.activity = entry.activity.saturating_add(1); + entry.tx.send_replace(entry.state.clone()); + } + + /// Publishes an explicit restart failure to the awaiting logical rendezvous. + pub fn fail(&self, key: RuntimeAgentKey, restart_id: &str, message: String) { + self.transition(key, restart_id, |restart_id| { + RestartRendezvousState::Failed { + restart_id, + message, + } + }); + } + + /// Cancels the restart-aware continuation for a target being explicitly stopped. + /// Returns whether a restart handoff was active. + pub fn cancel(&self, key: RuntimeAgentKey) -> bool { + let mut entries = self + .entries + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let Some(entry) = entries.get_mut(&key) else { + return false; + }; + let restart_id = match &entry.state { + RestartRendezvousState::Restarting { restart_id } + | RestartRendezvousState::Ready { restart_id } => restart_id.clone(), + _ => return false, + }; + entry.state = RestartRendezvousState::Cancelled { restart_id }; + entry.activity = entry.activity.saturating_add(1); + entry.tx.send_replace(entry.state.clone()); + true + } + + fn transition( + &self, + key: RuntimeAgentKey, + restart_id: &str, + next: impl FnOnce(String) -> RestartRendezvousState, + ) { + let mut entries = self + .entries + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let Some(entry) = entries.get_mut(&key) else { + return; + }; + let matches = match &entry.state { + RestartRendezvousState::Restarting { + restart_id: current, + } + | RestartRendezvousState::Ready { + restart_id: current, + } => current == restart_id, + _ => false, + }; + if !matches { + return; + } + entry.activity = entry.activity.saturating_add(1); + entry.state = next(restart_id.to_owned()); + entry.tx.send_replace(entry.state.clone()); + } + + #[must_use] + /// Reports whether the owning rendezvous was cancelled during this restart. + pub fn is_cancelled(&self, key: RuntimeAgentKey, restart_id: &str) -> bool { + self.entries + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .get(&key) + .is_some_and(|entry| { + matches!( + &entry.state, + RestartRendezvousState::Cancelled { restart_id: current } + if current == restart_id + ) + }) + } + + /// Waits until the active ask has accepted the replacement session handoff. + pub async fn wait_for_handoff( + &self, + key: RuntimeAgentKey, + restart_id: &str, + ) -> Result<(), String> { + let mut rx = { + let entries = self + .entries + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + entries + .get(&key) + .ok_or_else(|| "restart rendezvous disappeared".to_owned())? + .tx + .subscribe() + }; + loop { + let state = rx.borrow_and_update().clone(); + match state { + RestartRendezvousState::Idle => return Ok(()), + RestartRendezvousState::Cancelled { + restart_id: current, + } if current == restart_id => { + return Err("restart rendezvous was cancelled".to_owned()); + } + RestartRendezvousState::Failed { + restart_id: current, + message, + } if current == restart_id => return Err(message), + _ => {} + } + rx.changed() + .await + .map_err(|_| "restart rendezvous channel closed".to_owned())?; + } + } + + /// Releases terminal coordination state after the detached scheduler observed it. + pub fn finish(&self, key: RuntimeAgentKey, restart_id: &str) { + let mut entries = self + .entries + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let remove = entries.get(&key).is_some_and(|entry| { + entry.active_waiters == 0 + && matches!( + &entry.state, + RestartRendezvousState::Cancelled { restart_id: current } + | RestartRendezvousState::Failed { restart_id: current, .. } + if current == restart_id + ) + }); + if remove { + entries.remove(&key); + } + } + + /// A restarting target is control-plane activity on every watchdog probe, even + /// when provider/transcript activity is temporarily impossible during teardown. + fn activity_token(&self, key: RuntimeAgentKey) -> Option { + let mut entries = self + .entries + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let entry = entries.get_mut(&key)?; + if matches!( + entry.state, + RestartRendezvousState::Restarting { .. } | RestartRendezvousState::Ready { .. } + ) { + entry.activity = entry.activity.saturating_add(1); + } + Some(entry.activity) + } +} + +struct RestartRendezvousRegistration { + coordinator: Arc, + key: RuntimeAgentKey, + rx: watch::Receiver, +} + +impl RestartRendezvousRegistration { + fn state(&mut self) -> RestartRendezvousState { + self.rx.borrow_and_update().clone() + } + + async fn changed(&mut self) -> Result<(), watch::error::RecvError> { + self.rx.changed().await + } +} + +impl Drop for RestartRendezvousRegistration { + fn drop(&mut self) { + let mut entries = self + .coordinator + .entries + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let Some(entry) = entries.get_mut(&self.key) else { + return; + }; + entry.active_waiters = entry.active_waiters.saturating_sub(1); + if entry.active_waiters == 0 { + let restart_id = match &entry.state { + RestartRendezvousState::Restarting { restart_id } + | RestartRendezvousState::Ready { restart_id } => Some(restart_id.clone()), + _ => None, + }; + if let Some(restart_id) = restart_id { + entry.state = RestartRendezvousState::Cancelled { restart_id }; + entry.activity = entry.activity.saturating_add(1); + entry.tx.send_replace(entry.state.clone()); + } else if matches!( + entry.state, + RestartRendezvousState::Idle | RestartRendezvousState::Failed { .. } + ) { + entries.remove(&self.key); + } + } + } +} + /// Default terminal geometry for an orchestrator-launched agent cell. The UI /// resizes the PTY to the real cell size on attach; these are sane starting rows /// /cols so the spawn never fails on a zero-sized terminal. @@ -706,6 +1030,8 @@ pub struct OrchestratorService { spawn_background_command: Option>, /// Detached self-restart coordinator, wired at the composition root. self_restart: Option>, + /// Stable control-plane handoff between an active ask and a detached restart. + restart_rendezvous: Arc, } /// Bundle des quatre use cases C7 sous [`domain::fileguard::FileGuard`], injectés @@ -806,6 +1132,7 @@ impl OrchestratorService { session_limits: None, spawn_background_command: None, self_restart: None, + restart_rendezvous: Arc::new(RestartRendezvousCoordinator::new()), } } @@ -837,6 +1164,16 @@ impl OrchestratorService { self } + #[must_use] + /// Shares the runtime-keyed restart handoff with the detached restart adapter. + pub fn with_restart_rendezvous( + mut self, + coordinator: Arc, + ) -> Self { + self.restart_rendezvous = coordinator; + self + } + /// Branche le store des tâches de fond pour tracer les rendez-vous inter-agents /// comme [`BackgroundTaskKind::HeadlessRendezvous`] (B6). /// @@ -2393,7 +2730,7 @@ impl OrchestratorService { conversation_id, requester, task, - session.as_ref(), + session, ) .await } @@ -2419,7 +2756,7 @@ impl OrchestratorService { conversation_id: domain::conversation::ConversationId, requester: Option, task: String, - session: &dyn domain::ports::AgentSession, + session: Arc, ) -> Result { let (input, mailbox) = match (&self.input, &self.mailbox) { (Some(i), Some(m)) => (i, m), @@ -2524,57 +2861,131 @@ impl OrchestratorService { ticket: ticket_id, }); 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)), + let mut restart = self.restart_rendezvous.register(agent_key); + let structured = Arc::clone( + self.structured + .as_ref() + .expect("ask_structured requires the structured registry"), ); + let wait_activity = Arc::clone(&local_activity); - // L'attente du rendez-vous structured, rendue comme `Result` - // pour être enveloppée par le watchdog. Un flux clos sans `Final` correspond - // au no-reply du chemin headless : la cible a rendu la main sans réponse - // exploitable, on expose donc l'erreur métier retryable plutôt qu'un PROCESS. + // The concrete provider session may disappear under us when the target calls + // `idea_restart_self`. The logical rendezvous remains registered by + // RuntimeAgentKey; on the restart control signal we drop only the old drain, + // wait for the replacement session, then capture its continuation `Final`. let wait = async { - match drain.await { - Ok(TurnOutcome::Completed(content)) => Ok(content), - Ok(TurnOutcome::RateLimited { resets_at_ms }) => { - let conversation_id = session.conversation_id(); - let (conversation_id, resets_at_ms) = match (conversation_id, resets_at_ms) { - (Some(conversation_id), resets_at_ms) => { - (Some(conversation_id), resets_at_ms) + let mut current_session = session; + let mut prompt = task.clone(); + loop { + let turn = { + let drain = drain_with_readiness_announcements_and_activity_outcome( + current_session.as_ref(), + &prompt, + None, + input.as_ref(), + agent_key, + announcement_publisher.clone(), + Some(Arc::clone(&wait_activity)), + ); + tokio::pin!(drain); + tokio::select! { + biased; + changed = restart.changed() => { + changed.map_err(|_| AppError::Process( + format!("restart coordination closed for target {target}") + ))?; + None } - (None, None) => (None, None), - // Reset connu mais conversation moteur absente : ne pas armer - // une reprise non reprenable ; surfacer le fallback humain. - (None, Some(_)) => (None, None), - }; - if let (Some(service), Some(structured)) = - (&self.session_limits, &self.structured) - { - if let Some(node_id) = - structured.node_for_agent_in_project(project.id, &agent_id) - { - service.on_rate_limited( - project.id, - agent_id, - node_id, - conversation_id, - resets_at_ms, - ); + outcome = &mut drain => Some(outcome), + } + }; + + let control_active = !matches!(restart.state(), RestartRendezvousState::Idle); + if let Some(outcome) = turn.filter(|_| !control_active) { + match outcome { + Ok(TurnOutcome::Completed(content)) => return Ok(content), + Ok(TurnOutcome::RateLimited { resets_at_ms }) => { + let conversation_id = current_session.conversation_id(); + let (conversation_id, resets_at_ms) = + match (conversation_id, resets_at_ms) { + (Some(conversation_id), resets_at_ms) => { + (Some(conversation_id), resets_at_ms) + } + (None, None) => (None, None), + // Reset connu mais conversation moteur absente : ne pas armer + // une reprise non reprenable ; surfacer le fallback humain. + (None, Some(_)) => (None, None), + }; + if let (Some(service), Some(structured)) = + (&self.session_limits, &self.structured) + { + if let Some(node_id) = + structured.node_for_agent_in_project(project.id, &agent_id) + { + service.on_rate_limited( + project.id, + agent_id, + node_id, + conversation_id, + resets_at_ms, + ); + } + } + return Err(AppError::Process( + "le tour s'est clos en limite de session, sans contenu Final" + .to_owned(), + )); + } + Err(err) if structured_no_reply_error(&err) => { + return Err(AppError::TargetReturnedNoReply(target.to_owned())); + } + Err(err) => return Err(AppError::from(err)), + } + } + + loop { + match restart.state() { + RestartRendezvousState::Restarting { .. } => { + restart.changed().await.map_err(|_| { + AppError::Process(format!( + "restart coordination closed for target {target}" + )) + })?; + } + RestartRendezvousState::Ready { restart_id } => { + current_session = structured + .session_for_agent_in_project(project.id, &agent_id) + .ok_or_else(|| { + AppError::Process(format!( + "restart completed without a live structured session for target {target}" + )) + })?; + self.bind_conversation_session(conversation_id, current_session.id()); + self.restart_rendezvous.resume_taken(agent_key, &restart_id); + let _ = restart.state(); + prompt = SELF_RESTART_CONTINUATION_PROMPT.to_owned(); + wait_activity.fetch_add(1, AtomicOrdering::Relaxed); + break; + } + RestartRendezvousState::Failed { message, .. } => { + return Err(AppError::Process(format!( + "restart failed for target {target}: {message}" + ))); + } + RestartRendezvousState::Cancelled { .. } => { + return Err(AppError::Process(format!( + "restart cancelled for target {target}" + ))); + } + RestartRendezvousState::Idle => { + restart.changed().await.map_err(|_| { + AppError::Process(format!( + "restart coordination closed for target {target}" + )) + })?; } } - Err(AppError::Process( - "le tour s'est clos en limite de session, sans contenu Final".to_owned(), - )) } - Err(err) if structured_no_reply_error(&err) => { - Err(AppError::TargetReturnedNoReply(target.to_owned())) - } - Err(err) => Err(AppError::from(err)), } }; @@ -2590,6 +3001,7 @@ impl OrchestratorService { started, local: local_activity, }, + Some(agent_key), ) .await { @@ -3278,6 +3690,10 @@ impl OrchestratorService { } } + let restart_cancelled = self + .restart_rendezvous + .cancel(runtime_key(project, agent_id)); + let structured = self .structured .clone() @@ -3317,6 +3733,12 @@ impl OrchestratorService { .await? .is_empty() { + if restart_cancelled { + return Ok(OrchestratorOutcome { + detail: format!("cancelled restarting work for agent {label}"), + reply: None, + }); + } return Err(AppError::NotFound(format!( "running session for agent {label}" ))); @@ -3547,6 +3969,7 @@ impl OrchestratorService { agent_id: AgentId, target: &str, activity: AskWatchdogActivity, + restart_key: Option, ) -> WatchdogOutcome { let AskWatchdogActivity { started, @@ -3559,11 +3982,13 @@ impl OrchestratorService { #[derive(Default)] struct CompositeActivity { local: u64, + restart: u64, transcript: Option, token: u64, } let probe = self.ask_liveness_probe.clone(); + let restart_rendezvous = Arc::clone(&self.restart_rendezvous); let root = project_root.clone(); let composite = Arc::new(StdMutex::new(CompositeActivity::default())); let probe_fn = move || { @@ -3571,20 +3996,26 @@ impl OrchestratorService { let root = root.clone(); let local_activity = Arc::clone(&local_activity); let composite = Arc::clone(&composite); + let restart_rendezvous = Arc::clone(&restart_rendezvous); async move { let transcript = match &probe { Some(p) => p(root, agent_id).await, None => None, }; let local = local_activity.load(AtomicOrdering::Relaxed); + let restart = restart_key + .and_then(|key| restart_rendezvous.activity_token(key)) + .unwrap_or(0); let mut state = composite.lock().unwrap(); let local_delta = local.saturating_sub(state.local); + let restart_delta = restart.saturating_sub(state.restart); 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); + state.restart = state.restart.max(restart); if let Some(current) = transcript { state.transcript = Some(state.transcript.map_or(current, |old| old.max(current))); @@ -3592,6 +4023,7 @@ impl OrchestratorService { state.token = state .token .saturating_add(local_delta) + .saturating_add(restart_delta) .saturating_add(transcript_delta); Some(state.token) } @@ -4057,4 +4489,34 @@ mod tests { "pas de cancel_head après disarm" ); } + + #[tokio::test] + async fn restart_control_activity_rearms_inactivity_watchdog() { + let coordinator = Arc::new(RestartRendezvousCoordinator::new()); + let key = rkey(9); + let _registration = coordinator.register(key); + assert!(coordinator.begin(key, "restart-watchdog".to_owned())); + let probe = Arc::clone(&coordinator); + + let outcome = run_inactivity_watchdog( + async { + tokio::time::sleep(Duration::from_millis(45)).await; + "continued" + }, + Duration::from_millis(10), + Duration::from_millis(100), + Instant::now(), + true, + move || { + let probe = Arc::clone(&probe); + async move { probe.activity_token(key) } + }, + |_| {}, + |_| {}, + |_| {}, + ) + .await; + + assert!(matches!(outcome, WatchdogOutcome::Resolved("continued"))); + } } diff --git a/crates/application/tests/orchestrator_service.rs b/crates/application/tests/orchestrator_service.rs index 1dc13ee5..df58098e 100644 --- a/crates/application/tests/orchestrator_service.rs +++ b/crates/application/tests/orchestrator_service.rs @@ -2022,6 +2022,7 @@ struct AskFixture { /// assert the target's Working/Done transitions and the upsert count. live: Arc, structured: Arc, + restart_rendezvous: Arc, } struct CompletionSession { @@ -2198,6 +2199,7 @@ fn ask_fixture_full( let bus = SpyBus::default(); let mailbox = Arc::new(TestMailbox::new()); let structured = Arc::new(StructuredSessions::new()); + let restart_rendezvous = Arc::new(application::RestartRendezvousCoordinator::new()); let completion_factory = completion_factory_for_contexts(&contexts, mailbox.completions()); seed_structured_sessions(&structured, &contexts, mailbox.completions()); let memories = Arc::new(HarvestMemories { @@ -2270,6 +2272,7 @@ fn ask_fixture_full( .with_conversations(conversations) .with_events(Arc::new(bus.clone())) .with_structured(Arc::clone(&structured)) + .with_restart_rendezvous(Arc::clone(&restart_rendezvous)) .with_memory_harvest(Arc::new(HarvestMemoryFromTurn::new( Arc::clone(&memories) as Arc, Arc::new(bus.clone()), @@ -2302,6 +2305,7 @@ fn ask_fixture_full( memories, live, structured, + restart_rendezvous, } } @@ -4412,6 +4416,7 @@ fn ask_fixture_with_record( let bus = SpyBus::default(); let mailbox = Arc::new(TestMailbox::new()); let structured = Arc::new(StructuredSessions::new()); + let restart_rendezvous = Arc::new(application::RestartRendezvousCoordinator::new()); let completion_factory = completion_factory_for_contexts(&contexts, mailbox.completions()); seed_structured_sessions(&structured, &contexts, mailbox.completions()); @@ -4473,6 +4478,7 @@ fn ask_fixture_with_record( .with_conversations(conversations) .with_events(Arc::new(bus.clone())) .with_structured(Arc::clone(&structured)) + .with_restart_rendezvous(Arc::clone(&restart_rendezvous)) .with_record_turn(provider, Arc::new(FixedClock(clock_ms)) as Arc), ); @@ -4489,6 +4495,7 @@ fn ask_fixture_with_record( // Live-state likewise unwired here (RecordTurn-focused fixtures). live: Arc::new(RecordingLiveState::default()), structured, + restart_rendezvous, } } @@ -4868,6 +4875,106 @@ struct FailingStructuredSession { message: String, } +struct RestartingStructuredSession { + id: SessionId, + project_id: ProjectId, + agent_id: AgentId, + coordinator: Arc, + structured: Arc, + replacement: Arc, + delay: Duration, + failure: Option, + began: Arc, +} + +#[async_trait] +impl AgentSession for RestartingStructuredSession { + fn id(&self) -> SessionId { + self.id + } + + fn conversation_id(&self) -> Option { + Some("stable-engine-conversation".to_owned()) + } + + async fn send(&self, _prompt: &str) -> Result { + let key = domain::RuntimeAgentKey::new(self.project_id, self.agent_id); + let restart_id = format!("restart-{}", self.id); + assert!( + self.coordinator.begin(key, restart_id.clone()), + "the logical rendezvous must be registered before provider send" + ); + self.began.store(true, Ordering::SeqCst); + let coordinator = Arc::clone(&self.coordinator); + let structured = Arc::clone(&self.structured); + let replacement = Arc::clone(&self.replacement); + let replacement_project_id = self.project_id; + let old_id = self.id; + let delay = self.delay; + let failure = self.failure.clone(); + structured.remove(&old_id); + tokio::spawn(async move { + tokio::time::sleep(delay).await; + if let Some(message) = failure { + coordinator.fail(key, &restart_id, message); + return; + } + if coordinator.is_cancelled(key, &restart_id) { + return; + } + structured.insert_in_project( + replacement_project_id, + replacement, + key.agent_id, + nid(799), + ); + coordinator.ready(key, &restart_id); + }); + Err(AgentSessionError::Io( + "old provider session closed during restart".to_owned(), + )) + } + + async fn shutdown(&self) -> Result<(), AgentSessionError> { + Ok(()) + } + + async fn cancel_current_turn(&self) -> Result<(), AgentSessionError> { + Ok(()) + } +} + +struct CapturingFinalSession { + id: SessionId, + prompts: Arc>>, + final_text: String, +} + +#[async_trait] +impl AgentSession for CapturingFinalSession { + fn id(&self) -> SessionId { + self.id + } + + fn conversation_id(&self) -> Option { + Some("stable-engine-conversation".to_owned()) + } + + async fn send(&self, prompt: &str) -> Result { + self.prompts.lock().unwrap().push(prompt.to_owned()); + Ok(Box::new( + vec![ReplyEvent::Final { + content: self.final_text.clone(), + }] + .into_iter(), + )) + } + + async fn shutdown(&self) -> Result<(), AgentSessionError> { + Ok(()) + } +} + #[async_trait] impl AgentSession for FailingStructuredSession { fn id(&self) -> SessionId { @@ -5329,6 +5436,286 @@ async fn ask_warm_structured_target_reuses_session_without_relaunch() { ); } +fn install_restarting_session( + fx: &AskFixture, + project_id: ProjectId, + agent_id: AgentId, + delay: Duration, + failure: Option<&str>, +) -> (Arc>>, Arc) { + if let Some(existing) = fx + .structured + .session_for_agent_in_project(project_id, &agent_id) + { + fx.structured.remove(&existing.id()); + } + let prompts = Arc::new(Mutex::new(Vec::new())); + let began = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let replacement = Arc::new(CapturingFinalSession { + id: sid(9951), + prompts: Arc::clone(&prompts), + final_text: "final after restart".to_owned(), + }) as Arc; + fx.structured.insert_in_project( + project_id, + Arc::new(RestartingStructuredSession { + id: sid(9950), + project_id, + agent_id, + coordinator: Arc::clone(&fx.restart_rendezvous), + structured: Arc::clone(&fx.structured), + replacement, + delay, + failure: failure.map(str::to_owned), + began: Arc::clone(&began), + }), + agent_id, + nid(798), + ); + (prompts, began) +} + +#[tokio::test] +async fn ask_target_restart_keeps_logical_turn_and_returns_post_restart_final() { + let target = scratch_agent(aid(1), "architect", "agents/architect.md"); + let fx = ask_fixture(FakeContexts::with_agent(&target, "# persona")); + let (prompts, _) = install_restarting_session( + &fx, + project_id(), + target.id, + Duration::from_millis(20), + None, + ); + + let out = fx + .service + .dispatch(&project(), cmd(ASK_JSON)) + .await + .expect("restart-aware ask must return the replacement session Final"); + + assert_eq!(out.reply.as_deref(), Some("final after restart")); + assert_eq!( + prompts.lock().unwrap().as_slice(), + [application::SELF_RESTART_CONTINUATION_PROMPT], + "the replacement runtime receives one continuation turn, not the original task twice" + ); + assert_eq!( + fx.mailbox.pending(&target.id), + 0, + "logical ticket is retired" + ); +} + +#[tokio::test] +async fn ask_target_restart_cleans_busy_workstate_and_replaces_structured_registry_entry() { + let target = scratch_agent(aid(1), "architect", "agents/architect.md"); + let fx = ask_fixture(FakeContexts::with_agent(&target, "# persona")); + install_restarting_session( + &fx, + project_id(), + target.id, + Duration::from_millis(20), + None, + ); + + let out = fx + .service + .dispatch(&project(), cmd(ASK_JSON)) + .await + .expect("restart-aware ask must complete"); + + assert_eq!(out.reply.as_deref(), Some("final after restart")); + assert_eq!( + fx.mediator.busy_state(rkey(1)), + AgentBusyState::Idle, + "the logical target is not left busy after the post-restart Final" + ); + assert_eq!(fx.mailbox.pending(&target.id), 0, "FIFO is empty"); + let entry = fx + .live + .entry_for(target.id) + .expect("target live-state entry is written"); + assert_eq!(entry.status, WorkStatus::Done); + assert_eq!(entry.last_delegation, entry.ticket); + assert!( + fx.structured.remove(&sid(9950)).is_none(), + "old provider session was removed during restart" + ); + let current = fx + .structured + .session_for_agent_in_project(project_id(), &target.id) + .expect("replacement session remains registered for the same logical agent"); + assert_eq!(current.id(), sid(9951)); +} + +#[tokio::test] +async fn cross_project_ask_preserves_correlation_across_target_restart() { + let target = scratch_agent(aid(1), "architect", "agents/architect.md").with_contactable(true); + let requester = scratch_agent(aid(99), "requester", "agents/requester.md"); + let contexts = FakeContexts::with_agent(&target, "# target"); + contexts.add_agent(&requester, "# requester"); + let fx = ask_fixture(contexts); + let target_project = other_project(); + let (prompts, _) = install_restarting_session( + &fx, + target_project.id, + target.id, + Duration::from_millis(20), + None, + ); + + let out = fx + .service + .ask_project_agent_for_mcp( + target_project.id, + target.id, + "cross-project work".to_owned(), + project_id(), + requester.id, + "corr-restart".to_owned(), + ) + .await + .expect("cross-project rendezvous survives target restart"); + + assert_eq!(out.reply.as_deref(), Some("final after restart")); + assert_eq!( + prompts.lock().unwrap().as_slice(), + [application::SELF_RESTART_CONTINUATION_PROMPT] + ); + let cleaned = fx + .service + .cancel_project_agent_for_mcp( + target_project.id, + target.id, + project_id(), + requester.id, + "corr-restart".to_owned(), + ) + .await + .expect_err("completed cross-project correlation must be removed"); + assert_eq!(cleaned.code(), "NOT_FOUND"); +} + +#[tokio::test] +async fn ask_target_restart_failure_is_explicit_not_no_reply() { + let target = scratch_agent(aid(1), "architect", "agents/architect.md"); + let fx = ask_fixture(FakeContexts::with_agent(&target, "# persona")); + install_restarting_session( + &fx, + project_id(), + target.id, + Duration::from_millis(10), + Some("replacement launch exploded"), + ); + + let err = fx + .service + .dispatch(&project(), cmd(ASK_JSON)) + .await + .expect_err("restart failure must terminate the awaiting ask explicitly"); + + assert_eq!(err.code(), "PROCESS"); + assert!(err.to_string().contains("replacement launch exploded")); + assert!(!err.to_string().contains("structured Final response")); + assert_eq!(fx.mailbox.pending(&target.id), 0); +} + +#[tokio::test] +async fn cross_project_cancel_during_restart_releases_ask_and_correlation() { + let target = scratch_agent(aid(1), "architect", "agents/architect.md").with_contactable(true); + let requester = scratch_agent(aid(99), "requester", "agents/requester.md"); + let contexts = FakeContexts::with_agent(&target, "# target"); + contexts.add_agent(&requester, "# requester"); + let fx = ask_fixture(contexts); + let target_project = other_project(); + let (_, began) = install_restarting_session( + &fx, + target_project.id, + target.id, + Duration::from_secs(5), + None, + ); + let service = Arc::clone(&fx.service); + let ask = tokio::spawn(async move { + service + .ask_project_agent_for_mcp( + target_project.id, + target.id, + "cross-project work".to_owned(), + project_id(), + requester.id, + "corr-cancel-restart".to_owned(), + ) + .await + }); + await_until(|| began.load(Ordering::SeqCst)).await; + + fx.service + .cancel_project_agent_for_mcp( + other_project().id, + aid(1), + project_id(), + aid(99), + "corr-cancel-restart".to_owned(), + ) + .await + .expect("cancel routes through the stable cross-project correlation"); + let err = ask + .await + .expect("ask task joins") + .expect_err("cancelled restart-aware ask must fail"); + assert!(err.to_string().to_ascii_lowercase().contains("cancel")); + assert_eq!(fx.mailbox.pending(&aid(1)), 0); +} + +#[tokio::test] +async fn same_project_stop_during_restart_releases_ask_and_leaves_no_replacement_session() { + let target = scratch_agent(aid(1), "architect", "agents/architect.md"); + let fx = ask_fixture(FakeContexts::with_agent(&target, "# persona")); + let (_, began) = + install_restarting_session(&fx, project_id(), target.id, Duration::from_secs(5), None); + let service = Arc::clone(&fx.service); + let ask = tokio::spawn(async move { service.dispatch(&project(), cmd(ASK_JSON)).await }); + await_until(|| began.load(Ordering::SeqCst)).await; + + let stopped = fx + .service + .dispatch( + &project(), + cmd(r#"{ "action":"stop_agent", "name":"architect" }"#), + ) + .await + .expect("same-project stop cancels the restart-aware logical turn"); + + assert_eq!( + stopped.detail, + "cancelled restarting work for agent architect" + ); + let err = ask + .await + .expect("ask task joins") + .expect_err("cancelled restart-aware ask must fail"); + assert!(err.to_string().to_ascii_lowercase().contains("cancel")); + assert_eq!( + fx.mediator.busy_state(rkey(1)), + AgentBusyState::Idle, + "cancel cleanup frees the target busy state" + ); + assert_eq!(fx.mailbox.pending(&target.id), 0, "FIFO is empty"); + await_until(|| { + fx.live + .entry_for(target.id) + .is_some_and(|entry| entry.status == WorkStatus::Done) + }) + .await; + assert!( + fx.structured + .session_for_agent_in_project(project_id(), &target.id) + .is_none(), + "cancel during restart prevents a replacement runtime from being registered" + ); +} + /// 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. diff --git a/crates/backend/src/lib.rs b/crates/backend/src/lib.rs index 797a2d50..f39fb79b 100644 --- a/crates/backend/src/lib.rs +++ b/crates/backend/src/lib.rs @@ -1037,9 +1037,6 @@ struct AppAgentResumer { contexts: ResumeContexts, } -const SELF_RESTART_PROMPT: &str = - "IdeA vient de te redémarrer. Reprends immédiatement le travail en cours là où tu l'avais laissé."; - struct AppAgentSelfRestarter { launch_agent: Arc, close_terminal: Arc, @@ -1049,6 +1046,7 @@ struct AppAgentSelfRestarter { events: Arc, contexts: ResumeContexts, active: Arc>>, + rendezvous: Arc, } fn self_restart_launch_input( @@ -1134,6 +1132,7 @@ impl AgentSelfRestartScheduler for AppAgentSelfRestarter { } let restart_id = Uuid::new_v4().to_string(); + let delegated = self.rendezvous.begin(key, restart_id.clone()); self.events.publish(DomainEvent::AgentRestartRequested { agent_id: agent, restart_id: restart_id.clone(), @@ -1148,6 +1147,7 @@ impl AgentSelfRestartScheduler for AppAgentSelfRestarter { let events = Arc::clone(&self.events); let contexts = Arc::clone(&self.contexts); let active = Arc::clone(&self.active); + let rendezvous = Arc::clone(&self.rendezvous); let restart_id_task = restart_id.clone(); let project = project.clone(); spawn_detached(async move { @@ -1176,7 +1176,9 @@ impl AgentSelfRestartScheduler for AppAgentSelfRestarter { }); publish_phase(AgentRestartPhase::Stopping); - input.preempt(key); + if !delegated { + input.preempt(key); + } let stop_result = application::terminal::stop_live_agent_runtime_in_project( &project, agent, @@ -1186,12 +1188,17 @@ impl AgentSelfRestartScheduler for AppAgentSelfRestarter { .await; publish_phase(AgentRestartPhase::ReleasingInput); - if let Some(ticket) = input.busy_state(key).ticket() { - mailbox.cancel_head(key, ticket); + if !delegated { + if let Some(ticket) = input.busy_state(key).ticket() { + mailbox.cancel_head(key, ticket); + } + input.mark_idle(key); } - input.mark_idle(key); let fail = |phase, error: AppError| { + if delegated { + rendezvous.fail(key, &restart_id_task, error.to_string()); + } events.publish(DomainEvent::AgentRestartFailed { agent_id: agent, restart_id: restart_id_task.clone(), @@ -1199,6 +1206,11 @@ impl AgentSelfRestartScheduler for AppAgentSelfRestarter { message: error.to_string(), }); }; + let finish_coordination = || { + if delegated { + rendezvous.finish(key, &restart_id_task); + } + }; match stop_result { Ok(stopped) if !stopped.is_empty() => {} Ok(_) => { @@ -1206,16 +1218,28 @@ impl AgentSelfRestartScheduler for AppAgentSelfRestarter { AgentRestartPhase::Stopping, AppError::NotFound(format!("running runtime for agent {agent}")), ); + finish_coordination(); active.lock().ok().map(|mut set| set.remove(&key)); return; } Err(error) => { fail(AgentRestartPhase::Stopping, error); + finish_coordination(); active.lock().ok().map(|mut set| set.remove(&key)); return; } } + if delegated && rendezvous.is_cancelled(key, &restart_id_task) { + fail( + AgentRestartPhase::ReleasingInput, + AppError::Process("restart cancelled with its active rendezvous".to_owned()), + ); + finish_coordination(); + active.lock().ok().map(|mut set| set.remove(&key)); + return; + } + publish_phase(AgentRestartPhase::Launching); let mcp_runtime = crate::mcp_endpoint::idea_exe_path().map(|exe| McpRuntime { exe, @@ -1232,11 +1256,48 @@ impl AgentSelfRestartScheduler for AppAgentSelfRestarter { Ok(output) => output, Err(error) => { fail(AgentRestartPhase::Launching, error); + finish_coordination(); active.lock().ok().map(|mut set| set.remove(&key)); return; } }; + if delegated { + if rendezvous.is_cancelled(key, &restart_id_task) { + let _ = application::terminal::stop_live_agent_runtime_in_project( + &project, + agent, + live.as_ref(), + close_terminal.as_ref(), + ) + .await; + fail( + AgentRestartPhase::Launching, + AppError::Process( + "restart cancelled before the continuation turn".to_owned(), + ), + ); + finish_coordination(); + active.lock().ok().map(|mut set| set.remove(&key)); + return; + } + publish_phase(AgentRestartPhase::Resuming); + rendezvous.ready(key, &restart_id_task); + if let Err(message) = rendezvous.wait_for_handoff(key, &restart_id_task).await { + fail(AgentRestartPhase::Resuming, AppError::Process(message)); + finish_coordination(); + active.lock().ok().map(|mut set| set.remove(&key)); + return; + } + events.publish(DomainEvent::AgentRestartCompleted { + agent_id: agent, + restart_id: restart_id_task.clone(), + }); + finish_coordination(); + active.lock().ok().map(|mut set| set.remove(&key)); + return; + } + publish_phase(AgentRestartPhase::Resuming); let resume_result = if let Some(session) = live .structured @@ -1247,10 +1308,10 @@ impl AgentSelfRestartScheduler for AppAgentSelfRestarter { Arc::clone(&mailbox), project.id, agent, - SELF_RESTART_PROMPT, + application::SELF_RESTART_CONTINUATION_PROMPT, ); let result = session - .send(SELF_RESTART_PROMPT) + .send(application::SELF_RESTART_CONTINUATION_PROMPT) .await .map(|stream| stream.for_each(drop)) .map_err(|error| AppError::Process(error.to_string())); @@ -1272,7 +1333,7 @@ impl AgentSelfRestartScheduler for AppAgentSelfRestarter { let ticket = domain::mailbox::Ticket::new( domain::mailbox::TicketId::new_random(), "IdeA", - SELF_RESTART_PROMPT, + application::SELF_RESTART_CONTINUATION_PROMPT, ); let _pending = input.enqueue(key, ticket); Ok(()) @@ -3496,6 +3557,7 @@ impl BackendCore { Arc::clone(&events_port), resumer, )); + let restart_rendezvous = Arc::new(application::RestartRendezvousCoordinator::new()); let self_restart = Arc::new(AppAgentSelfRestarter { launch_agent: Arc::clone(&launch_agent), close_terminal: Arc::clone(&close_terminal), @@ -3505,6 +3567,7 @@ impl BackendCore { events: Arc::clone(&events_port), contexts: Arc::clone(&resume_contexts), active: Arc::new(Mutex::new(HashSet::new())), + rendezvous: Arc::clone(&restart_rendezvous), }) as Arc; // Drain du scheduler (§21.5-b) : à chaque réveil tiré, `TokioScheduler` pousse une // `ScheduledTask::ResumeAgent` ; on l'exécute via le service (relance + AgentResumed). @@ -3576,6 +3639,7 @@ impl BackendCore { .with_conversations(Arc::clone(&conversation_registry)) .with_events(Arc::clone(&events_port)) .with_self_restart(self_restart) + .with_restart_rendezvous(restart_rendezvous) // Faits OS/runtime (exe $APPIMAGE/current_exe + endpoint loopback) pour // que les (re)lancements issus du chemin `ask` écrivent la déclaration MCP // réelle ⇒ le pont MCP de la cible se spawne et `idea_reply` débloque le round-trip.