//! [`ReconcileLiveState`] — réconcilie, à l'**ouverture** d'un projet, les lignes //! de live-state **fantômes** : un agent marqué `working`/`waiting`/`blocked` //! alors que sa session n'est plus vivante (un crash ne déclenche jamais //! `close_project`, donc `open` est le seul filet fiable). //! //! C'est la **jumelle** de [`super::live::UpdateLiveState`] côté système : là où //! `UpdateLiveState` publie une transition déclarée par un agent, cette //! réconciliation est un **acte système** du composition root qui appelle le port //! [`LiveStateStore`] **directement** — jamais via `OrchestratorCommand::SetWorkState`. //! Le self-only de l'outil MCP `idea_workstate_set` (un agent ne peut écrire que sa //! propre ligne) est un invariant **de la surface MCP**, pas du store : il reste //! préservé par construction, exactement comme `snapshot_running_agents` / //! `reconcile_layouts`. //! //! Le use case est un mince orchestrateur au-dessus de : //! - le store du live-state ([`LiveStateStore`], par project root), //! - le registre des sessions vivantes ([`LiveAgentRegistry`] — la *même* vérité //! que `GetProjectWorkState`, dépendance restreinte par ISP : on n'a besoin que //! de « cet agent est-il vivant ? »), //! - l'opération **pure** du domaine [`LiveState::reconcile_orphans`] (qui porte la //! politique d'orphelin et la réécriture exacte des champs), //! - l'horloge ([`Clock`]) pour estampiller `updated_at_ms` (last-writer-wins). //! //! **Best-effort** : appelé après `reconcile_layouts` dans la chaîne `open_project`, //! un échec ne doit jamais bloquer l'ouverture (le call site ignore le résultat). //! **Race-safe** : tourne **avant** toute relance d'agent ; toute reprise ultérieure //! réécrit la ligne avec un `updated_at_ms` plus frais (LWW). use std::sync::Arc; use domain::ports::{Clock, LiveStateStore}; use domain::ProjectId; use crate::error::AppError; use crate::terminal::LiveAgentRegistry; /// Input de [`ReconcileLiveState::execute`]. /// /// `project_id` identifie le projet **au point d'appel** ; le [`LiveStateStore`] /// par-root est lié par le provider du composition root (cf. `AppLiveStateProvider`), /// si bien que le flux pur ci-dessous n'a plus de résolution à faire. #[derive(Debug, Clone, PartialEq, Eq)] pub struct ReconcileLiveStateInput { /// Le projet dont le live-state doit être réconcilié. pub project_id: ProjectId, } /// Réconcilie les lignes de live-state fantômes d'un projet à l'ouverture. pub struct ReconcileLiveState { store: Arc, registry: Arc, clock: Arc, } impl ReconcileLiveState { /// Construit le use case à partir de ses ports injectés (store par-root, /// registre de liveness, horloge). #[must_use] pub fn new( store: Arc, registry: Arc, clock: Arc, ) -> Self { Self { store, registry, clock, } } /// Charge le live-state, calcule les orphelins via la transformation pure du /// domaine ([`LiveState::reconcile_orphans`]) en croisant chaque ligne avec la /// liveness réelle, puis ré-upsert chaque ligne réécrite (keyed LWW). /// /// `_input` ne sert qu'à l'identité du projet au point d'appel : la racine est /// déjà liée dans le store injecté. /// /// # Errors /// [`AppError::Store`] sur défaillance de chargement ou d'upsert du store. pub async fn execute(&self, _input: ReconcileLiveStateInput) -> Result<(), AppError> { let state = self.store.load().await?; let now_ms = u64::try_from(self.clock.now_millis()).unwrap_or(0); let reconciled = state.reconcile_orphans(|a| self.registry.is_agent_live(a), now_ms); for entry in reconciled { self.store.upsert(entry).await?; } Ok(()) } } #[cfg(test)] mod tests { use super::*; use std::collections::HashSet; use std::sync::Mutex; use domain::live_state::{LiveEntry, LiveState, WorkStatus, STALE_AT_RESTART_MARKER}; use domain::ports::StoreError; use domain::{AgentId, NodeId}; /// In-memory [`LiveStateStore`] double (keyed upsert + prune), no I/O. #[derive(Default)] struct InMemoryLiveStateStore { state: Mutex, } #[async_trait::async_trait] impl LiveStateStore for InMemoryLiveStateStore { async fn load(&self) -> Result { Ok(self.state.lock().unwrap().clone()) } async fn upsert(&self, entry: LiveEntry) -> Result<(), StoreError> { self.state.lock().unwrap().upsert(entry); Ok(()) } async fn prune(&self, now_ms: u64, ttl_ms: u64, max_n: usize) -> Result<(), StoreError> { self.state.lock().unwrap().prune(now_ms, ttl_ms, max_n); Ok(()) } } /// Liveness fake: only the agents in the set are considered live. struct FakeRegistry { live: HashSet, } impl LiveAgentRegistry for FakeRegistry { fn is_agent_live(&self, agent_id: &AgentId) -> bool { self.live.contains(agent_id) } fn is_node_live(&self, _node_id: &NodeId) -> bool { false } } struct FixedClock(i64); impl Clock for FixedClock { fn now_millis(&self) -> i64 { self.0 } } fn aid(n: u128) -> AgentId { AgentId::from_uuid(uuid::Uuid::from_u128(n)) } fn pid() -> ProjectId { ProjectId::from_uuid(uuid::Uuid::from_u128(42)) } async fn run(store: Arc, live: HashSet, now: i64) { let uc = ReconcileLiveState::new( store, Arc::new(FakeRegistry { live }), Arc::new(FixedClock(now)), ); uc.execute(ReconcileLiveStateInput { project_id: pid() }) .await .unwrap(); } #[tokio::test] async fn reconciles_orphan_with_dead_session() { let store = Arc::new(InMemoryLiveStateStore::default()); store .upsert( LiveEntry::new(aid(1), None, "work", WorkStatus::Working, None, None, 1).unwrap(), ) .await .unwrap(); // No agent live, now = 999. run(store.clone(), HashSet::new(), 999).await; let after = store.load().await.unwrap(); assert_eq!(after.entries.len(), 1, "still one keyed row (upsert, not append)"); let row = &after.entries[0]; assert_eq!(row.status, WorkStatus::Idle, "orphan downgraded to idle"); assert_eq!(row.progress.as_deref(), Some(STALE_AT_RESTART_MARKER)); assert_eq!(row.updated_at_ms, 999, "stamped from the clock"); } #[tokio::test] async fn spares_a_live_agent() { let store = Arc::new(InMemoryLiveStateStore::default()); store .upsert( LiveEntry::new(aid(1), None, "work", WorkStatus::Working, None, None, 7).unwrap(), ) .await .unwrap(); // aid(1) is still live ⇒ untouched. run(store.clone(), HashSet::from([aid(1)]), 999).await; let row = &store.load().await.unwrap().entries[0]; assert_eq!(row.status, WorkStatus::Working, "live agent left as-is"); assert_eq!(row.updated_at_ms, 7, "not restamped"); } #[tokio::test] async fn empty_store_is_a_noop_at_boot() { // First boot / crashed before any publish ⇒ the live-state file is absent // and `load()` yields an empty state. Reconciliation must be a clean no-op: // no upsert, no panic — `open_project` must never choke on a missing file. let store = Arc::new(InMemoryLiveStateStore::default()); run(store.clone(), HashSet::new(), 999).await; assert!( store.load().await.unwrap().entries.is_empty(), "nothing to reconcile ⇒ store stays empty" ); } #[tokio::test] async fn leaves_idle_row_unchanged() { let store = Arc::new(InMemoryLiveStateStore::default()); store .upsert(LiveEntry::new(aid(1), None, "rest", WorkStatus::Idle, None, None, 3).unwrap()) .await .unwrap(); run(store.clone(), HashSet::new(), 999).await; let row = &store.load().await.unwrap().entries[0]; assert_eq!(row.status, WorkStatus::Idle); assert_eq!(row.updated_at_ms, 3, "idle row never reconciled, not restamped"); } }