feat(workstate): réconcilie le live-state au reboot (use case ReconcileLiveState)
Au redémarrage, les agents qui étaient `working` lors de la fermeture restaient figés en statut fantôme `working` alors qu'aucun tour n'est plus en cours (bug reproductible en live : QA et DevBackend eux-mêmes sont restés `working` fantôme après leurs tâches). On réconcilie l'état persistant à l'ouverture du projet : les statuts orphelins sont ramenés à la cible `idle`. - domain/live_state.rs : const STALE_AT_RESTART_MARKER + `reconcile_orphans` (+ tests). - application/workstate/reconcile.rs (nouveau) : use case ReconcileLiveState (pas de nouveau port), best-effort, no-op si store vide (+ tests, dont empty_store_is_a_noop_at_boot). - application/workstate/mod.rs + lib.rs : module + re-export. - app-tauri/state.rs : provider par-root + champ AppState + wiring. - app-tauri/commands.rs : hook best-effort dans open_project. QA VERT par commande réelle : cargo test --workspace 1624 passed / 0 failed, cargo clippy --workspace --all-targets 0 erreur sur le code ajouté. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@ -128,7 +128,8 @@ pub use workstate::{
|
||||
AgentTicketState, AgentWorkState, AttachLiveAgent, AttachLiveAgentInput, AttachLiveAgentOutput,
|
||||
ConversationLogProvider, ConversationPreviewStatus, ConversationTurnWorkPreview,
|
||||
ConversationWorkSummary, GetLiveStateLean, GetProjectWorkState, GetProjectWorkStateInput,
|
||||
LeanLiveEntry, LeanLiveState, LiveWorkSession, ProjectWorkState, StopLiveAgent,
|
||||
StopLiveAgentInput, StopLiveAgentOutput, TicketWorkSource, TicketWorkStatus, UpdateLiveState,
|
||||
UpdateLiveStateInput, LIVE_STATE_MAX_ENTRIES, LIVE_STATE_TTL_MS,
|
||||
LeanLiveEntry, LeanLiveState, LiveWorkSession, ProjectWorkState, ReconcileLiveState,
|
||||
ReconcileLiveStateInput, StopLiveAgent, StopLiveAgentInput, StopLiveAgentOutput,
|
||||
TicketWorkSource, TicketWorkStatus, UpdateLiveState, UpdateLiveStateInput,
|
||||
LIVE_STATE_MAX_ENTRIES, LIVE_STATE_TTL_MS,
|
||||
};
|
||||
|
||||
@ -7,11 +7,13 @@
|
||||
|
||||
mod actions;
|
||||
mod live;
|
||||
mod reconcile;
|
||||
|
||||
pub use actions::{
|
||||
AttachLiveAgent, AttachLiveAgentInput, AttachLiveAgentOutput, StopLiveAgent,
|
||||
StopLiveAgentInput, StopLiveAgentOutput,
|
||||
};
|
||||
pub use reconcile::{ReconcileLiveState, ReconcileLiveStateInput};
|
||||
pub use live::{
|
||||
GetLiveStateLean, LeanLiveEntry, LeanLiveState, UpdateLiveState, UpdateLiveStateInput,
|
||||
LIVE_STATE_MAX_ENTRIES, LIVE_STATE_TTL_MS,
|
||||
|
||||
230
crates/application/src/workstate/reconcile.rs
Normal file
230
crates/application/src/workstate/reconcile.rs
Normal file
@ -0,0 +1,230 @@
|
||||
//! [`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<dyn LiveStateStore>,
|
||||
registry: Arc<dyn LiveAgentRegistry>,
|
||||
clock: Arc<dyn Clock>,
|
||||
}
|
||||
|
||||
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<dyn LiveStateStore>,
|
||||
registry: Arc<dyn LiveAgentRegistry>,
|
||||
clock: Arc<dyn Clock>,
|
||||
) -> 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<LiveState>,
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl LiveStateStore for InMemoryLiveStateStore {
|
||||
async fn load(&self) -> Result<LiveState, StoreError> {
|
||||
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<AgentId>,
|
||||
}
|
||||
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<InMemoryLiveStateStore>, live: HashSet<AgentId>, 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");
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user