merge(orchestrator,agent): ask_agent survit au restart de la cible same-project et cross-project — #289 (QA verte: orchestrator_service 87/87 dont 6 tests restart, application lib 171/171, backend 89+7 ignored, cargo check -p backend, fmt --check; branche fix/289-ask-agent-during-target-restart commit d6d9891; restants: rebuild AppImage + validation live ask_agent sur cible restartée same-project et cross-project; scripts/ hors périmètre; aucun push)
This commit is contained in:
197
.ideai/tickets/289/carnet.md
Normal file
197
.ideai/tickets/289/carnet.md
Normal file
@ -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.
|
||||||
17
.ideai/tickets/289/issue.md
Normal file
17
.ideai/tickets/289/issue.md
Normal file
@ -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.
|
||||||
@ -1,3 +1,3 @@
|
|||||||
{
|
{
|
||||||
"nextNumber": 289
|
"nextNumber": 290
|
||||||
}
|
}
|
||||||
@ -3949,6 +3949,20 @@
|
|||||||
"agent_id": "a6c6ea12-bfc6-4bdc-8031-324102dfa34d"
|
"agent_id": "a6c6ea12-bfc6-4bdc-8031-324102dfa34d"
|
||||||
},
|
},
|
||||||
"updatedAt": 1788898181382
|
"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
|
||||||
}
|
}
|
||||||
]
|
]
|
||||||
}
|
}
|
||||||
@ -168,8 +168,9 @@ pub use orchestrator::{
|
|||||||
AgentRestartScheduled, AgentSelfRestartScheduler, AgentWakeService, AskLivenessProbe,
|
AgentRestartScheduled, AgentSelfRestartScheduler, AgentWakeService, AskLivenessProbe,
|
||||||
ContextGuardUseCases, LiveStateProvider, LiveStateReadProvider, McpRuntimeProvider,
|
ContextGuardUseCases, LiveStateProvider, LiveStateReadProvider, McpRuntimeProvider,
|
||||||
OrchestratorOutcome, OrchestratorService, ProposeContext, ReadContext, ReadMemory,
|
OrchestratorOutcome, OrchestratorService, ProposeContext, ReadContext, ReadMemory,
|
||||||
RecordTurnProvider, WakeSessionProvider, WatchdogOutcome, WriteMemory,
|
RecordTurnProvider, RestartRendezvousCoordinator, RestartRendezvousState, WakeSessionProvider,
|
||||||
DEFAULT_RENDEZVOUS_CEILING, DEFAULT_RENDEZVOUS_WINDOW,
|
WatchdogOutcome, WriteMemory, DEFAULT_RENDEZVOUS_CEILING, DEFAULT_RENDEZVOUS_WINDOW,
|
||||||
|
SELF_RESTART_CONTINUATION_PROMPT,
|
||||||
};
|
};
|
||||||
pub use permission::{
|
pub use permission::{
|
||||||
GetProjectPermissions, GetProjectPermissionsInput, GetProjectPermissionsOutput,
|
GetProjectPermissions, GetProjectPermissionsInput, GetProjectPermissionsOutput,
|
||||||
|
|||||||
@ -21,6 +21,7 @@ pub use rendezvous::{
|
|||||||
pub use service::{
|
pub use service::{
|
||||||
AgentRestartScheduled, AgentSelfRestartScheduler, AskLivenessProbe, ContextGuardUseCases,
|
AgentRestartScheduled, AgentSelfRestartScheduler, AskLivenessProbe, ContextGuardUseCases,
|
||||||
LiveStateProvider, LiveStateReadProvider, McpRuntimeProvider, OrchestratorOutcome,
|
LiveStateProvider, LiveStateReadProvider, McpRuntimeProvider, OrchestratorOutcome,
|
||||||
OrchestratorService, RecordTurnProvider,
|
OrchestratorService, RecordTurnProvider, RestartRendezvousCoordinator, RestartRendezvousState,
|
||||||
|
SELF_RESTART_CONTINUATION_PROMPT,
|
||||||
};
|
};
|
||||||
pub use wake::{AgentWakeService, WakeSessionProvider};
|
pub use wake::{AgentWakeService, WakeSessionProvider};
|
||||||
|
|||||||
@ -19,7 +19,7 @@ use std::sync::atomic::{AtomicU64, Ordering as AtomicOrdering};
|
|||||||
use std::sync::{Arc, Mutex as StdMutex};
|
use std::sync::{Arc, Mutex as StdMutex};
|
||||||
use std::time::{Duration, Instant};
|
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::{ConversationParty, ConversationRegistry, SessionRef, WaitForGraph};
|
||||||
use domain::conversation_log::{ConversationTurn, TurnId, TurnRole};
|
use domain::conversation_log::{ConversationTurn, TurnId, TurnRole};
|
||||||
@ -82,6 +82,330 @@ pub trait AgentSelfRestartScheduler: Send + Sync {
|
|||||||
) -> Result<AgentRestartScheduled, AppError>;
|
) -> Result<AgentRestartScheduled, AppError>;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// 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<RestartRendezvousState>,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// 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<HashMap<RuntimeAgentKey, RestartRendezvousEntry>>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl RestartRendezvousCoordinator {
|
||||||
|
/// Creates an empty runtime-keyed coordinator.
|
||||||
|
#[must_use]
|
||||||
|
pub fn new() -> Self {
|
||||||
|
Self::default()
|
||||||
|
}
|
||||||
|
|
||||||
|
fn register(self: &Arc<Self>, 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<u64> {
|
||||||
|
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<RestartRendezvousCoordinator>,
|
||||||
|
key: RuntimeAgentKey,
|
||||||
|
rx: watch::Receiver<RestartRendezvousState>,
|
||||||
|
}
|
||||||
|
|
||||||
|
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
|
/// 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
|
/// 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.
|
/// /cols so the spawn never fails on a zero-sized terminal.
|
||||||
@ -706,6 +1030,8 @@ pub struct OrchestratorService {
|
|||||||
spawn_background_command: Option<Arc<SpawnBackgroundCommand>>,
|
spawn_background_command: Option<Arc<SpawnBackgroundCommand>>,
|
||||||
/// Detached self-restart coordinator, wired at the composition root.
|
/// Detached self-restart coordinator, wired at the composition root.
|
||||||
self_restart: Option<Arc<dyn AgentSelfRestartScheduler>>,
|
self_restart: Option<Arc<dyn AgentSelfRestartScheduler>>,
|
||||||
|
/// Stable control-plane handoff between an active ask and a detached restart.
|
||||||
|
restart_rendezvous: Arc<RestartRendezvousCoordinator>,
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Bundle des quatre use cases C7 sous [`domain::fileguard::FileGuard`], injectés
|
/// Bundle des quatre use cases C7 sous [`domain::fileguard::FileGuard`], injectés
|
||||||
@ -806,6 +1132,7 @@ impl OrchestratorService {
|
|||||||
session_limits: None,
|
session_limits: None,
|
||||||
spawn_background_command: None,
|
spawn_background_command: None,
|
||||||
self_restart: None,
|
self_restart: None,
|
||||||
|
restart_rendezvous: Arc::new(RestartRendezvousCoordinator::new()),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -837,6 +1164,16 @@ impl OrchestratorService {
|
|||||||
self
|
self
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[must_use]
|
||||||
|
/// Shares the runtime-keyed restart handoff with the detached restart adapter.
|
||||||
|
pub fn with_restart_rendezvous(
|
||||||
|
mut self,
|
||||||
|
coordinator: Arc<RestartRendezvousCoordinator>,
|
||||||
|
) -> Self {
|
||||||
|
self.restart_rendezvous = coordinator;
|
||||||
|
self
|
||||||
|
}
|
||||||
|
|
||||||
/// Branche le store des tâches de fond pour tracer les rendez-vous inter-agents
|
/// Branche le store des tâches de fond pour tracer les rendez-vous inter-agents
|
||||||
/// comme [`BackgroundTaskKind::HeadlessRendezvous`] (B6).
|
/// comme [`BackgroundTaskKind::HeadlessRendezvous`] (B6).
|
||||||
///
|
///
|
||||||
@ -2393,7 +2730,7 @@ impl OrchestratorService {
|
|||||||
conversation_id,
|
conversation_id,
|
||||||
requester,
|
requester,
|
||||||
task,
|
task,
|
||||||
session.as_ref(),
|
session,
|
||||||
)
|
)
|
||||||
.await
|
.await
|
||||||
}
|
}
|
||||||
@ -2419,7 +2756,7 @@ impl OrchestratorService {
|
|||||||
conversation_id: domain::conversation::ConversationId,
|
conversation_id: domain::conversation::ConversationId,
|
||||||
requester: Option<AgentId>,
|
requester: Option<AgentId>,
|
||||||
task: String,
|
task: String,
|
||||||
session: &dyn domain::ports::AgentSession,
|
session: Arc<dyn domain::ports::AgentSession>,
|
||||||
) -> Result<OrchestratorOutcome, AppError> {
|
) -> Result<OrchestratorOutcome, AppError> {
|
||||||
let (input, mailbox) = match (&self.input, &self.mailbox) {
|
let (input, mailbox) = match (&self.input, &self.mailbox) {
|
||||||
(Some(i), Some(m)) => (i, m),
|
(Some(i), Some(m)) => (i, m),
|
||||||
@ -2524,57 +2861,131 @@ impl OrchestratorService {
|
|||||||
ticket: ticket_id,
|
ticket: ticket_id,
|
||||||
});
|
});
|
||||||
let local_activity = Arc::new(AtomicU64::new(0));
|
let local_activity = Arc::new(AtomicU64::new(0));
|
||||||
let drain = drain_with_readiness_announcements_and_activity_outcome(
|
let mut restart = self.restart_rendezvous.register(agent_key);
|
||||||
session,
|
let structured = Arc::clone(
|
||||||
&task,
|
self.structured
|
||||||
None,
|
.as_ref()
|
||||||
input.as_ref(),
|
.expect("ask_structured requires the structured registry"),
|
||||||
agent_key,
|
|
||||||
announcement_publisher,
|
|
||||||
Some(Arc::clone(&local_activity)),
|
|
||||||
);
|
);
|
||||||
|
let wait_activity = Arc::clone(&local_activity);
|
||||||
|
|
||||||
// L'attente du rendez-vous structured, rendue comme `Result<String, AppError>`
|
// The concrete provider session may disappear under us when the target calls
|
||||||
// pour être enveloppée par le watchdog. Un flux clos sans `Final` correspond
|
// `idea_restart_self`. The logical rendezvous remains registered by
|
||||||
// au no-reply du chemin headless : la cible a rendu la main sans réponse
|
// RuntimeAgentKey; on the restart control signal we drop only the old drain,
|
||||||
// exploitable, on expose donc l'erreur métier retryable plutôt qu'un PROCESS.
|
// wait for the replacement session, then capture its continuation `Final`.
|
||||||
let wait = async {
|
let wait = async {
|
||||||
match drain.await {
|
let mut current_session = session;
|
||||||
Ok(TurnOutcome::Completed(content)) => Ok(content),
|
let mut prompt = task.clone();
|
||||||
Ok(TurnOutcome::RateLimited { resets_at_ms }) => {
|
loop {
|
||||||
let conversation_id = session.conversation_id();
|
let turn = {
|
||||||
let (conversation_id, resets_at_ms) = match (conversation_id, resets_at_ms) {
|
let drain = drain_with_readiness_announcements_and_activity_outcome(
|
||||||
(Some(conversation_id), resets_at_ms) => {
|
current_session.as_ref(),
|
||||||
(Some(conversation_id), resets_at_ms)
|
&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),
|
outcome = &mut drain => Some(outcome),
|
||||||
// Reset connu mais conversation moteur absente : ne pas armer
|
}
|
||||||
// une reprise non reprenable ; surfacer le fallback humain.
|
};
|
||||||
(None, Some(_)) => (None, None),
|
|
||||||
};
|
let control_active = !matches!(restart.state(), RestartRendezvousState::Idle);
|
||||||
if let (Some(service), Some(structured)) =
|
if let Some(outcome) = turn.filter(|_| !control_active) {
|
||||||
(&self.session_limits, &self.structured)
|
match outcome {
|
||||||
{
|
Ok(TurnOutcome::Completed(content)) => return Ok(content),
|
||||||
if let Some(node_id) =
|
Ok(TurnOutcome::RateLimited { resets_at_ms }) => {
|
||||||
structured.node_for_agent_in_project(project.id, &agent_id)
|
let conversation_id = current_session.conversation_id();
|
||||||
{
|
let (conversation_id, resets_at_ms) =
|
||||||
service.on_rate_limited(
|
match (conversation_id, resets_at_ms) {
|
||||||
project.id,
|
(Some(conversation_id), resets_at_ms) => {
|
||||||
agent_id,
|
(Some(conversation_id), resets_at_ms)
|
||||||
node_id,
|
}
|
||||||
conversation_id,
|
(None, None) => (None, None),
|
||||||
resets_at_ms,
|
// 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,
|
started,
|
||||||
local: local_activity,
|
local: local_activity,
|
||||||
},
|
},
|
||||||
|
Some(agent_key),
|
||||||
)
|
)
|
||||||
.await
|
.await
|
||||||
{
|
{
|
||||||
@ -3278,6 +3690,10 @@ impl OrchestratorService {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
let restart_cancelled = self
|
||||||
|
.restart_rendezvous
|
||||||
|
.cancel(runtime_key(project, agent_id));
|
||||||
|
|
||||||
let structured = self
|
let structured = self
|
||||||
.structured
|
.structured
|
||||||
.clone()
|
.clone()
|
||||||
@ -3317,6 +3733,12 @@ impl OrchestratorService {
|
|||||||
.await?
|
.await?
|
||||||
.is_empty()
|
.is_empty()
|
||||||
{
|
{
|
||||||
|
if restart_cancelled {
|
||||||
|
return Ok(OrchestratorOutcome {
|
||||||
|
detail: format!("cancelled restarting work for agent {label}"),
|
||||||
|
reply: None,
|
||||||
|
});
|
||||||
|
}
|
||||||
return Err(AppError::NotFound(format!(
|
return Err(AppError::NotFound(format!(
|
||||||
"running session for agent {label}"
|
"running session for agent {label}"
|
||||||
)));
|
)));
|
||||||
@ -3547,6 +3969,7 @@ impl OrchestratorService {
|
|||||||
agent_id: AgentId,
|
agent_id: AgentId,
|
||||||
target: &str,
|
target: &str,
|
||||||
activity: AskWatchdogActivity,
|
activity: AskWatchdogActivity,
|
||||||
|
restart_key: Option<RuntimeAgentKey>,
|
||||||
) -> WatchdogOutcome<R> {
|
) -> WatchdogOutcome<R> {
|
||||||
let AskWatchdogActivity {
|
let AskWatchdogActivity {
|
||||||
started,
|
started,
|
||||||
@ -3559,11 +3982,13 @@ impl OrchestratorService {
|
|||||||
#[derive(Default)]
|
#[derive(Default)]
|
||||||
struct CompositeActivity {
|
struct CompositeActivity {
|
||||||
local: u64,
|
local: u64,
|
||||||
|
restart: u64,
|
||||||
transcript: Option<u64>,
|
transcript: Option<u64>,
|
||||||
token: u64,
|
token: u64,
|
||||||
}
|
}
|
||||||
|
|
||||||
let probe = self.ask_liveness_probe.clone();
|
let probe = self.ask_liveness_probe.clone();
|
||||||
|
let restart_rendezvous = Arc::clone(&self.restart_rendezvous);
|
||||||
let root = project_root.clone();
|
let root = project_root.clone();
|
||||||
let composite = Arc::new(StdMutex::new(CompositeActivity::default()));
|
let composite = Arc::new(StdMutex::new(CompositeActivity::default()));
|
||||||
let probe_fn = move || {
|
let probe_fn = move || {
|
||||||
@ -3571,20 +3996,26 @@ impl OrchestratorService {
|
|||||||
let root = root.clone();
|
let root = root.clone();
|
||||||
let local_activity = Arc::clone(&local_activity);
|
let local_activity = Arc::clone(&local_activity);
|
||||||
let composite = Arc::clone(&composite);
|
let composite = Arc::clone(&composite);
|
||||||
|
let restart_rendezvous = Arc::clone(&restart_rendezvous);
|
||||||
async move {
|
async move {
|
||||||
let transcript = match &probe {
|
let transcript = match &probe {
|
||||||
Some(p) => p(root, agent_id).await,
|
Some(p) => p(root, agent_id).await,
|
||||||
None => None,
|
None => None,
|
||||||
};
|
};
|
||||||
let local = local_activity.load(AtomicOrdering::Relaxed);
|
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 mut state = composite.lock().unwrap();
|
||||||
let local_delta = local.saturating_sub(state.local);
|
let local_delta = local.saturating_sub(state.local);
|
||||||
|
let restart_delta = restart.saturating_sub(state.restart);
|
||||||
let transcript_delta = match (state.transcript, transcript) {
|
let transcript_delta = match (state.transcript, transcript) {
|
||||||
(Some(previous), Some(current)) => current.saturating_sub(previous),
|
(Some(previous), Some(current)) => current.saturating_sub(previous),
|
||||||
(None, Some(_)) => 1,
|
(None, Some(_)) => 1,
|
||||||
_ => 0,
|
_ => 0,
|
||||||
};
|
};
|
||||||
state.local = state.local.max(local);
|
state.local = state.local.max(local);
|
||||||
|
state.restart = state.restart.max(restart);
|
||||||
if let Some(current) = transcript {
|
if let Some(current) = transcript {
|
||||||
state.transcript =
|
state.transcript =
|
||||||
Some(state.transcript.map_or(current, |old| old.max(current)));
|
Some(state.transcript.map_or(current, |old| old.max(current)));
|
||||||
@ -3592,6 +4023,7 @@ impl OrchestratorService {
|
|||||||
state.token = state
|
state.token = state
|
||||||
.token
|
.token
|
||||||
.saturating_add(local_delta)
|
.saturating_add(local_delta)
|
||||||
|
.saturating_add(restart_delta)
|
||||||
.saturating_add(transcript_delta);
|
.saturating_add(transcript_delta);
|
||||||
Some(state.token)
|
Some(state.token)
|
||||||
}
|
}
|
||||||
@ -4057,4 +4489,34 @@ mod tests {
|
|||||||
"pas de cancel_head après disarm"
|
"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")));
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@ -2022,6 +2022,7 @@ struct AskFixture {
|
|||||||
/// assert the target's Working/Done transitions and the upsert count.
|
/// assert the target's Working/Done transitions and the upsert count.
|
||||||
live: Arc<RecordingLiveState>,
|
live: Arc<RecordingLiveState>,
|
||||||
structured: Arc<StructuredSessions>,
|
structured: Arc<StructuredSessions>,
|
||||||
|
restart_rendezvous: Arc<application::RestartRendezvousCoordinator>,
|
||||||
}
|
}
|
||||||
|
|
||||||
struct CompletionSession {
|
struct CompletionSession {
|
||||||
@ -2198,6 +2199,7 @@ fn ask_fixture_full(
|
|||||||
let bus = SpyBus::default();
|
let bus = SpyBus::default();
|
||||||
let mailbox = Arc::new(TestMailbox::new());
|
let mailbox = Arc::new(TestMailbox::new());
|
||||||
let structured = Arc::new(StructuredSessions::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());
|
let completion_factory = completion_factory_for_contexts(&contexts, mailbox.completions());
|
||||||
seed_structured_sessions(&structured, &contexts, mailbox.completions());
|
seed_structured_sessions(&structured, &contexts, mailbox.completions());
|
||||||
let memories = Arc::new(HarvestMemories {
|
let memories = Arc::new(HarvestMemories {
|
||||||
@ -2270,6 +2272,7 @@ fn ask_fixture_full(
|
|||||||
.with_conversations(conversations)
|
.with_conversations(conversations)
|
||||||
.with_events(Arc::new(bus.clone()))
|
.with_events(Arc::new(bus.clone()))
|
||||||
.with_structured(Arc::clone(&structured))
|
.with_structured(Arc::clone(&structured))
|
||||||
|
.with_restart_rendezvous(Arc::clone(&restart_rendezvous))
|
||||||
.with_memory_harvest(Arc::new(HarvestMemoryFromTurn::new(
|
.with_memory_harvest(Arc::new(HarvestMemoryFromTurn::new(
|
||||||
Arc::clone(&memories) as Arc<dyn MemoryStore>,
|
Arc::clone(&memories) as Arc<dyn MemoryStore>,
|
||||||
Arc::new(bus.clone()),
|
Arc::new(bus.clone()),
|
||||||
@ -2302,6 +2305,7 @@ fn ask_fixture_full(
|
|||||||
memories,
|
memories,
|
||||||
live,
|
live,
|
||||||
structured,
|
structured,
|
||||||
|
restart_rendezvous,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -4412,6 +4416,7 @@ fn ask_fixture_with_record(
|
|||||||
let bus = SpyBus::default();
|
let bus = SpyBus::default();
|
||||||
let mailbox = Arc::new(TestMailbox::new());
|
let mailbox = Arc::new(TestMailbox::new());
|
||||||
let structured = Arc::new(StructuredSessions::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());
|
let completion_factory = completion_factory_for_contexts(&contexts, mailbox.completions());
|
||||||
seed_structured_sessions(&structured, &contexts, mailbox.completions());
|
seed_structured_sessions(&structured, &contexts, mailbox.completions());
|
||||||
|
|
||||||
@ -4473,6 +4478,7 @@ fn ask_fixture_with_record(
|
|||||||
.with_conversations(conversations)
|
.with_conversations(conversations)
|
||||||
.with_events(Arc::new(bus.clone()))
|
.with_events(Arc::new(bus.clone()))
|
||||||
.with_structured(Arc::clone(&structured))
|
.with_structured(Arc::clone(&structured))
|
||||||
|
.with_restart_rendezvous(Arc::clone(&restart_rendezvous))
|
||||||
.with_record_turn(provider, Arc::new(FixedClock(clock_ms)) as Arc<dyn Clock>),
|
.with_record_turn(provider, Arc::new(FixedClock(clock_ms)) as Arc<dyn Clock>),
|
||||||
);
|
);
|
||||||
|
|
||||||
@ -4489,6 +4495,7 @@ fn ask_fixture_with_record(
|
|||||||
// Live-state likewise unwired here (RecordTurn-focused fixtures).
|
// Live-state likewise unwired here (RecordTurn-focused fixtures).
|
||||||
live: Arc::new(RecordingLiveState::default()),
|
live: Arc::new(RecordingLiveState::default()),
|
||||||
structured,
|
structured,
|
||||||
|
restart_rendezvous,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -4868,6 +4875,106 @@ struct FailingStructuredSession {
|
|||||||
message: String,
|
message: String,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
struct RestartingStructuredSession {
|
||||||
|
id: SessionId,
|
||||||
|
project_id: ProjectId,
|
||||||
|
agent_id: AgentId,
|
||||||
|
coordinator: Arc<application::RestartRendezvousCoordinator>,
|
||||||
|
structured: Arc<StructuredSessions>,
|
||||||
|
replacement: Arc<dyn AgentSession>,
|
||||||
|
delay: Duration,
|
||||||
|
failure: Option<String>,
|
||||||
|
began: Arc<std::sync::atomic::AtomicBool>,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[async_trait]
|
||||||
|
impl AgentSession for RestartingStructuredSession {
|
||||||
|
fn id(&self) -> SessionId {
|
||||||
|
self.id
|
||||||
|
}
|
||||||
|
|
||||||
|
fn conversation_id(&self) -> Option<String> {
|
||||||
|
Some("stable-engine-conversation".to_owned())
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn send(&self, _prompt: &str) -> Result<ReplyStream, AgentSessionError> {
|
||||||
|
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<Mutex<Vec<String>>>,
|
||||||
|
final_text: String,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[async_trait]
|
||||||
|
impl AgentSession for CapturingFinalSession {
|
||||||
|
fn id(&self) -> SessionId {
|
||||||
|
self.id
|
||||||
|
}
|
||||||
|
|
||||||
|
fn conversation_id(&self) -> Option<String> {
|
||||||
|
Some("stable-engine-conversation".to_owned())
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn send(&self, prompt: &str) -> Result<ReplyStream, AgentSessionError> {
|
||||||
|
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]
|
#[async_trait]
|
||||||
impl AgentSession for FailingStructuredSession {
|
impl AgentSession for FailingStructuredSession {
|
||||||
fn id(&self) -> SessionId {
|
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<Mutex<Vec<String>>>, Arc<std::sync::atomic::AtomicBool>) {
|
||||||
|
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<dyn AgentSession>;
|
||||||
|
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
|
/// 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
|
/// 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.
|
/// tour. Chaque événement non terminal doit réarmer la fenêtre jusqu'au `Final` tardif.
|
||||||
|
|||||||
@ -1037,9 +1037,6 @@ struct AppAgentResumer {
|
|||||||
contexts: ResumeContexts,
|
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 {
|
struct AppAgentSelfRestarter {
|
||||||
launch_agent: Arc<LaunchAgent>,
|
launch_agent: Arc<LaunchAgent>,
|
||||||
close_terminal: Arc<CloseTerminal>,
|
close_terminal: Arc<CloseTerminal>,
|
||||||
@ -1049,6 +1046,7 @@ struct AppAgentSelfRestarter {
|
|||||||
events: Arc<dyn EventBus>,
|
events: Arc<dyn EventBus>,
|
||||||
contexts: ResumeContexts,
|
contexts: ResumeContexts,
|
||||||
active: Arc<Mutex<HashSet<domain::RuntimeAgentKey>>>,
|
active: Arc<Mutex<HashSet<domain::RuntimeAgentKey>>>,
|
||||||
|
rendezvous: Arc<application::RestartRendezvousCoordinator>,
|
||||||
}
|
}
|
||||||
|
|
||||||
fn self_restart_launch_input(
|
fn self_restart_launch_input(
|
||||||
@ -1134,6 +1132,7 @@ impl AgentSelfRestartScheduler for AppAgentSelfRestarter {
|
|||||||
}
|
}
|
||||||
|
|
||||||
let restart_id = Uuid::new_v4().to_string();
|
let restart_id = Uuid::new_v4().to_string();
|
||||||
|
let delegated = self.rendezvous.begin(key, restart_id.clone());
|
||||||
self.events.publish(DomainEvent::AgentRestartRequested {
|
self.events.publish(DomainEvent::AgentRestartRequested {
|
||||||
agent_id: agent,
|
agent_id: agent,
|
||||||
restart_id: restart_id.clone(),
|
restart_id: restart_id.clone(),
|
||||||
@ -1148,6 +1147,7 @@ impl AgentSelfRestartScheduler for AppAgentSelfRestarter {
|
|||||||
let events = Arc::clone(&self.events);
|
let events = Arc::clone(&self.events);
|
||||||
let contexts = Arc::clone(&self.contexts);
|
let contexts = Arc::clone(&self.contexts);
|
||||||
let active = Arc::clone(&self.active);
|
let active = Arc::clone(&self.active);
|
||||||
|
let rendezvous = Arc::clone(&self.rendezvous);
|
||||||
let restart_id_task = restart_id.clone();
|
let restart_id_task = restart_id.clone();
|
||||||
let project = project.clone();
|
let project = project.clone();
|
||||||
spawn_detached(async move {
|
spawn_detached(async move {
|
||||||
@ -1176,7 +1176,9 @@ impl AgentSelfRestartScheduler for AppAgentSelfRestarter {
|
|||||||
});
|
});
|
||||||
|
|
||||||
publish_phase(AgentRestartPhase::Stopping);
|
publish_phase(AgentRestartPhase::Stopping);
|
||||||
input.preempt(key);
|
if !delegated {
|
||||||
|
input.preempt(key);
|
||||||
|
}
|
||||||
let stop_result = application::terminal::stop_live_agent_runtime_in_project(
|
let stop_result = application::terminal::stop_live_agent_runtime_in_project(
|
||||||
&project,
|
&project,
|
||||||
agent,
|
agent,
|
||||||
@ -1186,12 +1188,17 @@ impl AgentSelfRestartScheduler for AppAgentSelfRestarter {
|
|||||||
.await;
|
.await;
|
||||||
|
|
||||||
publish_phase(AgentRestartPhase::ReleasingInput);
|
publish_phase(AgentRestartPhase::ReleasingInput);
|
||||||
if let Some(ticket) = input.busy_state(key).ticket() {
|
if !delegated {
|
||||||
mailbox.cancel_head(key, ticket);
|
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| {
|
let fail = |phase, error: AppError| {
|
||||||
|
if delegated {
|
||||||
|
rendezvous.fail(key, &restart_id_task, error.to_string());
|
||||||
|
}
|
||||||
events.publish(DomainEvent::AgentRestartFailed {
|
events.publish(DomainEvent::AgentRestartFailed {
|
||||||
agent_id: agent,
|
agent_id: agent,
|
||||||
restart_id: restart_id_task.clone(),
|
restart_id: restart_id_task.clone(),
|
||||||
@ -1199,6 +1206,11 @@ impl AgentSelfRestartScheduler for AppAgentSelfRestarter {
|
|||||||
message: error.to_string(),
|
message: error.to_string(),
|
||||||
});
|
});
|
||||||
};
|
};
|
||||||
|
let finish_coordination = || {
|
||||||
|
if delegated {
|
||||||
|
rendezvous.finish(key, &restart_id_task);
|
||||||
|
}
|
||||||
|
};
|
||||||
match stop_result {
|
match stop_result {
|
||||||
Ok(stopped) if !stopped.is_empty() => {}
|
Ok(stopped) if !stopped.is_empty() => {}
|
||||||
Ok(_) => {
|
Ok(_) => {
|
||||||
@ -1206,16 +1218,28 @@ impl AgentSelfRestartScheduler for AppAgentSelfRestarter {
|
|||||||
AgentRestartPhase::Stopping,
|
AgentRestartPhase::Stopping,
|
||||||
AppError::NotFound(format!("running runtime for agent {agent}")),
|
AppError::NotFound(format!("running runtime for agent {agent}")),
|
||||||
);
|
);
|
||||||
|
finish_coordination();
|
||||||
active.lock().ok().map(|mut set| set.remove(&key));
|
active.lock().ok().map(|mut set| set.remove(&key));
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
Err(error) => {
|
Err(error) => {
|
||||||
fail(AgentRestartPhase::Stopping, error);
|
fail(AgentRestartPhase::Stopping, error);
|
||||||
|
finish_coordination();
|
||||||
active.lock().ok().map(|mut set| set.remove(&key));
|
active.lock().ok().map(|mut set| set.remove(&key));
|
||||||
return;
|
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);
|
publish_phase(AgentRestartPhase::Launching);
|
||||||
let mcp_runtime = crate::mcp_endpoint::idea_exe_path().map(|exe| McpRuntime {
|
let mcp_runtime = crate::mcp_endpoint::idea_exe_path().map(|exe| McpRuntime {
|
||||||
exe,
|
exe,
|
||||||
@ -1232,11 +1256,48 @@ impl AgentSelfRestartScheduler for AppAgentSelfRestarter {
|
|||||||
Ok(output) => output,
|
Ok(output) => output,
|
||||||
Err(error) => {
|
Err(error) => {
|
||||||
fail(AgentRestartPhase::Launching, error);
|
fail(AgentRestartPhase::Launching, error);
|
||||||
|
finish_coordination();
|
||||||
active.lock().ok().map(|mut set| set.remove(&key));
|
active.lock().ok().map(|mut set| set.remove(&key));
|
||||||
return;
|
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);
|
publish_phase(AgentRestartPhase::Resuming);
|
||||||
let resume_result = if let Some(session) = live
|
let resume_result = if let Some(session) = live
|
||||||
.structured
|
.structured
|
||||||
@ -1247,10 +1308,10 @@ impl AgentSelfRestartScheduler for AppAgentSelfRestarter {
|
|||||||
Arc::clone(&mailbox),
|
Arc::clone(&mailbox),
|
||||||
project.id,
|
project.id,
|
||||||
agent,
|
agent,
|
||||||
SELF_RESTART_PROMPT,
|
application::SELF_RESTART_CONTINUATION_PROMPT,
|
||||||
);
|
);
|
||||||
let result = session
|
let result = session
|
||||||
.send(SELF_RESTART_PROMPT)
|
.send(application::SELF_RESTART_CONTINUATION_PROMPT)
|
||||||
.await
|
.await
|
||||||
.map(|stream| stream.for_each(drop))
|
.map(|stream| stream.for_each(drop))
|
||||||
.map_err(|error| AppError::Process(error.to_string()));
|
.map_err(|error| AppError::Process(error.to_string()));
|
||||||
@ -1272,7 +1333,7 @@ impl AgentSelfRestartScheduler for AppAgentSelfRestarter {
|
|||||||
let ticket = domain::mailbox::Ticket::new(
|
let ticket = domain::mailbox::Ticket::new(
|
||||||
domain::mailbox::TicketId::new_random(),
|
domain::mailbox::TicketId::new_random(),
|
||||||
"IdeA",
|
"IdeA",
|
||||||
SELF_RESTART_PROMPT,
|
application::SELF_RESTART_CONTINUATION_PROMPT,
|
||||||
);
|
);
|
||||||
let _pending = input.enqueue(key, ticket);
|
let _pending = input.enqueue(key, ticket);
|
||||||
Ok(())
|
Ok(())
|
||||||
@ -3496,6 +3557,7 @@ impl BackendCore {
|
|||||||
Arc::clone(&events_port),
|
Arc::clone(&events_port),
|
||||||
resumer,
|
resumer,
|
||||||
));
|
));
|
||||||
|
let restart_rendezvous = Arc::new(application::RestartRendezvousCoordinator::new());
|
||||||
let self_restart = Arc::new(AppAgentSelfRestarter {
|
let self_restart = Arc::new(AppAgentSelfRestarter {
|
||||||
launch_agent: Arc::clone(&launch_agent),
|
launch_agent: Arc::clone(&launch_agent),
|
||||||
close_terminal: Arc::clone(&close_terminal),
|
close_terminal: Arc::clone(&close_terminal),
|
||||||
@ -3505,6 +3567,7 @@ impl BackendCore {
|
|||||||
events: Arc::clone(&events_port),
|
events: Arc::clone(&events_port),
|
||||||
contexts: Arc::clone(&resume_contexts),
|
contexts: Arc::clone(&resume_contexts),
|
||||||
active: Arc::new(Mutex::new(HashSet::new())),
|
active: Arc::new(Mutex::new(HashSet::new())),
|
||||||
|
rendezvous: Arc::clone(&restart_rendezvous),
|
||||||
}) as Arc<dyn AgentSelfRestartScheduler>;
|
}) as Arc<dyn AgentSelfRestartScheduler>;
|
||||||
// Drain du scheduler (§21.5-b) : à chaque réveil tiré, `TokioScheduler` pousse une
|
// Drain du scheduler (§21.5-b) : à chaque réveil tiré, `TokioScheduler` pousse une
|
||||||
// `ScheduledTask::ResumeAgent` ; on l'exécute via le service (relance + AgentResumed).
|
// `ScheduledTask::ResumeAgent` ; on l'exécute via le service (relance + AgentResumed).
|
||||||
@ -3576,6 +3639,7 @@ impl BackendCore {
|
|||||||
.with_conversations(Arc::clone(&conversation_registry))
|
.with_conversations(Arc::clone(&conversation_registry))
|
||||||
.with_events(Arc::clone(&events_port))
|
.with_events(Arc::clone(&events_port))
|
||||||
.with_self_restart(self_restart)
|
.with_self_restart(self_restart)
|
||||||
|
.with_restart_rendezvous(restart_rendezvous)
|
||||||
// Faits OS/runtime (exe $APPIMAGE/current_exe + endpoint loopback) pour
|
// Faits OS/runtime (exe $APPIMAGE/current_exe + endpoint loopback) pour
|
||||||
// que les (re)lancements issus du chemin `ask` écrivent la déclaration MCP
|
// 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.
|
// réelle ⇒ le pont MCP de la cible se spawne et `idea_reply` débloque le round-trip.
|
||||||
|
|||||||
Reference in New Issue
Block a user