merge(workstate): intègre la réconciliation du live-state au reboot (ReconcileLiveState)
Intègre feature/reconcile-live-state : à l'ouverture d'un projet, les statuts d'agents restés `working` fantôme après un redémarrage sont réconciliés vers `idle` (use case ReconcileLiveState, best-effort, sans nouveau port). QA VERT : cargo test --workspace 1624 passed / 0 failed, clippy 0 erreur. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@ -19,6 +19,7 @@ use application::{
|
|||||||
ListResumableAgentsInput, ListSkillsInput, LiveSessions, LoadLayoutInput, McpRuntime,
|
ListResumableAgentsInput, ListSkillsInput, LiveSessions, LoadLayoutInput, McpRuntime,
|
||||||
MutateLayoutInput, OpenProjectInput, ReadAgentContextInput, ReadConversationPageInput,
|
MutateLayoutInput, OpenProjectInput, ReadAgentContextInput, ReadConversationPageInput,
|
||||||
ReadMemoryIndexInput, ReadProjectContextInput, RecallMemoryInput, ReconcileLayoutsInput,
|
ReadMemoryIndexInput, ReadProjectContextInput, RecallMemoryInput, ReconcileLayoutsInput,
|
||||||
|
ReconcileLiveStateInput,
|
||||||
RenameLayoutInput, ResolveAgentPermissionsInput, ResolveMemoryLinksInput,
|
RenameLayoutInput, ResolveAgentPermissionsInput, ResolveMemoryLinksInput,
|
||||||
RotateConversationLogInput, SetActiveLayoutInput, SnapshotRunningAgentsInput,
|
RotateConversationLogInput, SetActiveLayoutInput, SnapshotRunningAgentsInput,
|
||||||
StopLiveAgentInput, SyncAgentWithTemplateInput, UnassignSkillFromAgentInput,
|
StopLiveAgentInput, SyncAgentWithTemplateInput, UnassignSkillFromAgentInput,
|
||||||
@ -124,6 +125,16 @@ pub async fn open_project(
|
|||||||
.reconcile_layouts
|
.reconcile_layouts
|
||||||
.execute(ReconcileLayoutsInput { project_id: id })
|
.execute(ReconcileLayoutsInput { project_id: id })
|
||||||
.await;
|
.await;
|
||||||
|
// Réconcilie les lignes de live-state fantômes : un crash ne déclenche jamais
|
||||||
|
// `close_project`, donc l'ouverture est le seul filet fiable pour repasser en
|
||||||
|
// `idle` les agents `working`/`waiting`/`blocked` dont la session est morte.
|
||||||
|
// Acte système (appelle le port live-state directement, jamais via la surface
|
||||||
|
// MCP self-only). Best-effort : un échec ne bloque pas l'ouverture. Tourne
|
||||||
|
// AVANT toute relance d'agent (toute reprise réécrit ensuite en LWW).
|
||||||
|
let _ = state
|
||||||
|
.reconcile_live_state
|
||||||
|
.execute(ReconcileLiveStateInput { project_id: id })
|
||||||
|
.await;
|
||||||
// (Re)start the orchestrator watcher for this project (idempotent, §14.3).
|
// (Re)start the orchestrator watcher for this project (idempotent, §14.3).
|
||||||
state.ensure_orchestrator_watch(&output.project);
|
state.ensure_orchestrator_watch(&output.project);
|
||||||
state.reconcile_claude_run_dirs(&output.project).await;
|
state.reconcile_claude_run_dirs(&output.project).await;
|
||||||
|
|||||||
@ -27,7 +27,8 @@ use application::{
|
|||||||
McpRuntime, MoveTabToNewWindow, MutateLayout, OnnxModelView, OpenProject, OpenTerminal,
|
McpRuntime, MoveTabToNewWindow, MutateLayout, OnnxModelView, OpenProject, OpenTerminal,
|
||||||
OrchestratorService, PermissionProjectorRegistry, ProposeContext, ReadAgentContext,
|
OrchestratorService, PermissionProjectorRegistry, ProposeContext, ReadAgentContext,
|
||||||
ReadContext, ReadConversationPage, ReadMemory, ReadMemoryIndex, ReadProjectContext, ReadSkill,
|
ReadContext, ReadConversationPage, ReadMemory, ReadMemoryIndex, ReadProjectContext, ReadSkill,
|
||||||
RecallMemory, ReconcileLayouts, RecordTurn, RecordTurnProvider, ReferenceProfiles,
|
RecallMemory, ReconcileLayouts, ReconcileLiveState, ReconcileLiveStateInput, RecordTurn,
|
||||||
|
RecordTurnProvider, ReferenceProfiles,
|
||||||
RenameLayout, ResizeTerminal, ResolveAgentPermissions, ResolveMemoryLinks,
|
RenameLayout, ResizeTerminal, ResolveAgentPermissions, ResolveMemoryLinks,
|
||||||
RotateConversationLog, SaveEmbedderProfile, SaveProfile, SessionLimitService, SetActiveLayout,
|
RotateConversationLog, SaveEmbedderProfile, SaveProfile, SessionLimitService, SetActiveLayout,
|
||||||
SnapshotRunningAgents, StopLiveAgent, StructuredSessions, SuggestedThisSession,
|
SnapshotRunningAgents, StopLiveAgent, StructuredSessions, SuggestedThisSession,
|
||||||
@ -212,6 +213,43 @@ impl LiveStateReadProvider for AppLiveStateLeanProvider {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Provider par-root de la réconciliation du live-state au reboot (jumeau de
|
||||||
|
/// [`AppLiveStateProvider`] côté système). Résout le **project root** depuis le
|
||||||
|
/// `project_id` via le [`ProjectStore`], matérialise un [`FsLiveStateStore`] ciblant
|
||||||
|
/// `<root>/.ideai/live-state.json`, puis exécute le use case applicatif
|
||||||
|
/// [`ReconcileLiveState`] (qui croise chaque ligne avec la liveness réelle de
|
||||||
|
/// [`LiveSessions`] et ré-upsert les fantômes en `idle`).
|
||||||
|
///
|
||||||
|
/// **Acte système** : appelle le port [`domain::ports::LiveStateStore`]
|
||||||
|
/// **directement**, jamais via `OrchestratorCommand::SetWorkState` ⇒ le self-only de
|
||||||
|
/// `idea_workstate_set` reste préservé par construction. Best-effort : le call site
|
||||||
|
/// (`open_project`) ignore le résultat.
|
||||||
|
pub(crate) struct AppReconcileLiveState {
|
||||||
|
projects: Arc<dyn ProjectStore>,
|
||||||
|
registry: Arc<LiveSessions>,
|
||||||
|
clock: Arc<dyn Clock>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl AppReconcileLiveState {
|
||||||
|
/// Résout la racine du projet, lie le store par-root et lance la réconciliation.
|
||||||
|
///
|
||||||
|
/// # Errors
|
||||||
|
/// [`AppError`] si le projet est inconnu (registre) ou si le store échoue.
|
||||||
|
pub(crate) async fn execute(
|
||||||
|
&self,
|
||||||
|
input: ReconcileLiveStateInput,
|
||||||
|
) -> Result<(), AppError> {
|
||||||
|
let project = self.projects.load_project(input.project_id).await?;
|
||||||
|
let store = Arc::new(FsLiveStateStore::new(&project.root));
|
||||||
|
let uc = ReconcileLiveState::new(
|
||||||
|
store,
|
||||||
|
Arc::clone(&self.registry) as Arc<dyn LiveAgentRegistry>,
|
||||||
|
Arc::clone(&self.clock),
|
||||||
|
);
|
||||||
|
uc.execute(input).await
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// Implémente [`ProviderSessionProvider`](application::ProviderSessionProvider) (lot
|
/// Implémente [`ProviderSessionProvider`](application::ProviderSessionProvider) (lot
|
||||||
/// P8b) en matérialisant un [`FsProviderSessionStore`] ciblant le **project root** du
|
/// P8b) en matérialisant un [`FsProviderSessionStore`] ciblant le **project root** du
|
||||||
/// lancement en cours.
|
/// lancement en cours.
|
||||||
@ -383,6 +421,10 @@ pub struct AppState {
|
|||||||
/// Dé-doublonne, à l'ouverture, les feuilles d'agent en double d'un même
|
/// Dé-doublonne, à l'ouverture, les feuilles d'agent en double d'un même
|
||||||
/// agent dans `layouts.json` (R0c). Idempotent : no-op sans doublon.
|
/// agent dans `layouts.json` (R0c). Idempotent : no-op sans doublon.
|
||||||
pub reconcile_layouts: Arc<ReconcileLayouts>,
|
pub reconcile_layouts: Arc<ReconcileLayouts>,
|
||||||
|
/// Réconcilie, à l'ouverture, les lignes de live-state fantômes (agent
|
||||||
|
/// `working`/`waiting`/`blocked` dont la session n'est plus vivante) : les
|
||||||
|
/// repasse en `idle`. Acte système best-effort, jamais via la surface MCP.
|
||||||
|
pub(crate) reconcile_live_state: Arc<AppReconcileLiveState>,
|
||||||
/// Detect which candidate profiles' CLIs are installed (first-run).
|
/// Detect which candidate profiles' CLIs are installed (first-run).
|
||||||
pub detect_profiles: Arc<DetectProfiles>,
|
pub detect_profiles: Arc<DetectProfiles>,
|
||||||
/// List configured profiles.
|
/// List configured profiles.
|
||||||
@ -1169,6 +1211,16 @@ impl AppState {
|
|||||||
Arc::clone(&terminal_sessions),
|
Arc::clone(&terminal_sessions),
|
||||||
Arc::clone(&structured_sessions),
|
Arc::clone(&structured_sessions),
|
||||||
));
|
));
|
||||||
|
// Réconciliation du live-state au reboot : repasse en `idle` les lignes
|
||||||
|
// fantômes (working/waiting/blocked) dont la session n'est plus vivante,
|
||||||
|
// selon le MÊME registre de liveness que `GetProjectWorkState`. Provider
|
||||||
|
// par root (le store fixe sa racine à la construction). Câblé dans
|
||||||
|
// `open_project`, best-effort (cf. `reconcile_layouts`).
|
||||||
|
let reconcile_live_state = Arc::new(AppReconcileLiveState {
|
||||||
|
projects: Arc::clone(&store_port),
|
||||||
|
registry: Arc::clone(&live_sessions),
|
||||||
|
clock: Arc::clone(&clock) as Arc<dyn Clock>,
|
||||||
|
});
|
||||||
// Lot C — résumés de conversation best-effort : on câble les sources par
|
// Lot C — résumés de conversation best-effort : on câble les sources par
|
||||||
// project root (handoff = primaire, log = repli `last(_, 3)`). Lecture seule,
|
// project root (handoff = primaire, log = repli `last(_, 3)`). Lecture seule,
|
||||||
// aucune persistance ; un échec de preview ne bloque ni live/busy ni tickets.
|
// aucune persistance ; un échec de preview ne bloque ni live/busy ni tickets.
|
||||||
@ -1374,6 +1426,7 @@ impl AppState {
|
|||||||
set_active_layout,
|
set_active_layout,
|
||||||
snapshot_running_agents,
|
snapshot_running_agents,
|
||||||
reconcile_layouts,
|
reconcile_layouts,
|
||||||
|
reconcile_live_state,
|
||||||
detect_profiles,
|
detect_profiles,
|
||||||
list_profiles,
|
list_profiles,
|
||||||
save_profile,
|
save_profile,
|
||||||
|
|||||||
@ -128,7 +128,8 @@ pub use workstate::{
|
|||||||
AgentTicketState, AgentWorkState, AttachLiveAgent, AttachLiveAgentInput, AttachLiveAgentOutput,
|
AgentTicketState, AgentWorkState, AttachLiveAgent, AttachLiveAgentInput, AttachLiveAgentOutput,
|
||||||
ConversationLogProvider, ConversationPreviewStatus, ConversationTurnWorkPreview,
|
ConversationLogProvider, ConversationPreviewStatus, ConversationTurnWorkPreview,
|
||||||
ConversationWorkSummary, GetLiveStateLean, GetProjectWorkState, GetProjectWorkStateInput,
|
ConversationWorkSummary, GetLiveStateLean, GetProjectWorkState, GetProjectWorkStateInput,
|
||||||
LeanLiveEntry, LeanLiveState, LiveWorkSession, ProjectWorkState, StopLiveAgent,
|
LeanLiveEntry, LeanLiveState, LiveWorkSession, ProjectWorkState, ReconcileLiveState,
|
||||||
StopLiveAgentInput, StopLiveAgentOutput, TicketWorkSource, TicketWorkStatus, UpdateLiveState,
|
ReconcileLiveStateInput, StopLiveAgent, StopLiveAgentInput, StopLiveAgentOutput,
|
||||||
UpdateLiveStateInput, LIVE_STATE_MAX_ENTRIES, LIVE_STATE_TTL_MS,
|
TicketWorkSource, TicketWorkStatus, UpdateLiveState, UpdateLiveStateInput,
|
||||||
|
LIVE_STATE_MAX_ENTRIES, LIVE_STATE_TTL_MS,
|
||||||
};
|
};
|
||||||
|
|||||||
@ -7,11 +7,13 @@
|
|||||||
|
|
||||||
mod actions;
|
mod actions;
|
||||||
mod live;
|
mod live;
|
||||||
|
mod reconcile;
|
||||||
|
|
||||||
pub use actions::{
|
pub use actions::{
|
||||||
AttachLiveAgent, AttachLiveAgentInput, AttachLiveAgentOutput, StopLiveAgent,
|
AttachLiveAgent, AttachLiveAgentInput, AttachLiveAgentOutput, StopLiveAgent,
|
||||||
StopLiveAgentInput, StopLiveAgentOutput,
|
StopLiveAgentInput, StopLiveAgentOutput,
|
||||||
};
|
};
|
||||||
|
pub use reconcile::{ReconcileLiveState, ReconcileLiveStateInput};
|
||||||
pub use live::{
|
pub use live::{
|
||||||
GetLiveStateLean, LeanLiveEntry, LeanLiveState, UpdateLiveState, UpdateLiveStateInput,
|
GetLiveStateLean, LeanLiveEntry, LeanLiveState, UpdateLiveState, UpdateLiveStateInput,
|
||||||
LIVE_STATE_MAX_ENTRIES, LIVE_STATE_TTL_MS,
|
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");
|
||||||
|
}
|
||||||
|
}
|
||||||
@ -32,6 +32,12 @@ pub const FIELD_PREVIEW_MAX_CHARS: usize = 160;
|
|||||||
/// live state. Distinct semantics: soft = truncate, hard = reject.
|
/// live state. Distinct semantics: soft = truncate, hard = reject.
|
||||||
pub const FIELD_MAX_BYTES: usize = 2 * 1024;
|
pub const FIELD_MAX_BYTES: usize = 2 * 1024;
|
||||||
|
|
||||||
|
/// `progress` marker stamped on an orphan row reconciled at restart (see
|
||||||
|
/// [`LiveState::reconcile_orphans`]): it records *why* a `Working`/`Waiting`/
|
||||||
|
/// `Blocked` row was downgraded to `Idle` — its session was not running when the
|
||||||
|
/// project reopened. Well under both field bounds.
|
||||||
|
pub const STALE_AT_RESTART_MARKER: &str = "(stale — session not running at restart)";
|
||||||
|
|
||||||
/// What an agent is doing right now, at a glance.
|
/// What an agent is doing right now, at a glance.
|
||||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
|
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
|
||||||
#[serde(rename_all = "camelCase")]
|
#[serde(rename_all = "camelCase")]
|
||||||
@ -183,6 +189,46 @@ impl LiveState {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Computes the **reconciliation** of orphan rows at restart: rows whose
|
||||||
|
/// status implies a live session (`Working`/`Waiting`/`Blocked`) but whose
|
||||||
|
/// agent is **not** currently live according to `is_live`.
|
||||||
|
///
|
||||||
|
/// At reboot a crash never runs `close_project`, so `live-state.json` can keep
|
||||||
|
/// "ghost" rows for agents whose sessions are dead. This pure transform returns
|
||||||
|
/// the **rewritten** rows to upsert (it does **not** mutate `self`): each orphan
|
||||||
|
/// is downgraded to [`WorkStatus::Idle`], keeping its `intent` (trace), clearing
|
||||||
|
/// the now-dead `ticket` and `last_delegation`, stamping a `progress` stale
|
||||||
|
/// marker and a fresh `updated_at_ms = now_ms` (last-writer-wins).
|
||||||
|
///
|
||||||
|
/// Rows that are already `Idle`/`Done`, or whose agent is still live, are
|
||||||
|
/// **never** touched and are absent from the result. The caller persists each
|
||||||
|
/// returned entry through the keyed [`upsert`](LiveState::upsert) (best-effort).
|
||||||
|
#[must_use]
|
||||||
|
pub fn reconcile_orphans(
|
||||||
|
&self,
|
||||||
|
is_live: impl Fn(&AgentId) -> bool,
|
||||||
|
now_ms: u64,
|
||||||
|
) -> Vec<LiveEntry> {
|
||||||
|
self.entries
|
||||||
|
.iter()
|
||||||
|
.filter(|e| {
|
||||||
|
matches!(
|
||||||
|
e.status,
|
||||||
|
WorkStatus::Working | WorkStatus::Waiting | WorkStatus::Blocked
|
||||||
|
) && !is_live(&e.agent_id)
|
||||||
|
})
|
||||||
|
.map(|e| LiveEntry {
|
||||||
|
agent_id: e.agent_id,
|
||||||
|
ticket: None,
|
||||||
|
intent: e.intent.clone(),
|
||||||
|
status: WorkStatus::Idle,
|
||||||
|
progress: Some(STALE_AT_RESTART_MARKER.to_owned()),
|
||||||
|
last_delegation: None,
|
||||||
|
updated_at_ms: now_ms,
|
||||||
|
})
|
||||||
|
.collect()
|
||||||
|
}
|
||||||
|
|
||||||
/// Drops entries older than `ttl_ms` relative to `now_ms`, then bounds the
|
/// Drops entries older than `ttl_ms` relative to `now_ms`, then bounds the
|
||||||
/// result to `max_n`, keeping the most recently updated rows.
|
/// result to `max_n`, keeping the most recently updated rows.
|
||||||
///
|
///
|
||||||
@ -361,6 +407,94 @@ mod tests {
|
|||||||
assert_eq!(kept, vec![50, 40], "most recent first, oldest dropped");
|
assert_eq!(kept, vec![50, 40], "most recent first, oldest dropped");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn reconcile_orphans_downgrades_dead_active_rows_and_rewrites_fields() {
|
||||||
|
let mut state = LiveState::default();
|
||||||
|
// An orphan: Working but its agent is not live.
|
||||||
|
state.upsert(
|
||||||
|
LiveEntry::new(
|
||||||
|
aid(1),
|
||||||
|
Some(tid(7)),
|
||||||
|
"implementing lot 2",
|
||||||
|
WorkStatus::Working,
|
||||||
|
Some("half done".to_owned()),
|
||||||
|
Some(tid(8)),
|
||||||
|
100,
|
||||||
|
)
|
||||||
|
.unwrap(),
|
||||||
|
);
|
||||||
|
|
||||||
|
// Nothing is live ⇒ aid(1) is an orphan.
|
||||||
|
let rewritten = state.reconcile_orphans(|_| false, 999);
|
||||||
|
|
||||||
|
assert_eq!(rewritten.len(), 1, "the single active-but-dead row is an orphan");
|
||||||
|
let row = &rewritten[0];
|
||||||
|
assert_eq!(row.agent_id, aid(1));
|
||||||
|
assert_eq!(row.status, WorkStatus::Idle, "downgraded to idle");
|
||||||
|
assert_eq!(row.intent, "implementing lot 2", "intent kept as trace");
|
||||||
|
assert_eq!(row.ticket, None, "dead ticket cleared");
|
||||||
|
assert_eq!(row.last_delegation, None, "dead delegation cleared");
|
||||||
|
assert_eq!(row.progress.as_deref(), Some(STALE_AT_RESTART_MARKER));
|
||||||
|
assert_eq!(row.updated_at_ms, 999, "stamped with the fresh now");
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn reconcile_orphans_covers_working_waiting_blocked_only() {
|
||||||
|
let mut state = LiveState::default();
|
||||||
|
state.upsert(
|
||||||
|
LiveEntry::new(aid(1), None, "w", WorkStatus::Working, None, None, 1).unwrap(),
|
||||||
|
);
|
||||||
|
state.upsert(
|
||||||
|
LiveEntry::new(aid(2), None, "a", WorkStatus::Waiting, None, None, 1).unwrap(),
|
||||||
|
);
|
||||||
|
state.upsert(
|
||||||
|
LiveEntry::new(aid(3), None, "b", WorkStatus::Blocked, None, None, 1).unwrap(),
|
||||||
|
);
|
||||||
|
// idle/done are never orphans, even when the agent is dead.
|
||||||
|
state.upsert(LiveEntry::new(aid(4), None, "i", WorkStatus::Idle, None, None, 1).unwrap());
|
||||||
|
state.upsert(LiveEntry::new(aid(5), None, "d", WorkStatus::Done, None, None, 1).unwrap());
|
||||||
|
|
||||||
|
let rewritten = state.reconcile_orphans(|_| false, 50);
|
||||||
|
|
||||||
|
let mut ids: Vec<AgentId> = rewritten.iter().map(|e| e.agent_id).collect();
|
||||||
|
ids.sort();
|
||||||
|
assert_eq!(
|
||||||
|
ids,
|
||||||
|
vec![aid(1), aid(2), aid(3)],
|
||||||
|
"only working/waiting/blocked rows are reconciled; idle/done untouched"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn reconcile_orphans_spares_live_agents() {
|
||||||
|
let mut state = LiveState::default();
|
||||||
|
state.upsert(
|
||||||
|
LiveEntry::new(aid(1), None, "alive", WorkStatus::Working, None, None, 1).unwrap(),
|
||||||
|
);
|
||||||
|
state.upsert(
|
||||||
|
LiveEntry::new(aid(2), None, "dead", WorkStatus::Working, None, None, 1).unwrap(),
|
||||||
|
);
|
||||||
|
|
||||||
|
// aid(1) is still live ⇒ spared; only aid(2) is reconciled.
|
||||||
|
let rewritten = state.reconcile_orphans(|a| *a == aid(1), 50);
|
||||||
|
|
||||||
|
assert_eq!(rewritten.len(), 1);
|
||||||
|
assert_eq!(rewritten[0].agent_id, aid(2), "only the dead agent is reconciled");
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn reconcile_orphans_does_not_mutate_self() {
|
||||||
|
let mut state = LiveState::default();
|
||||||
|
state.upsert(
|
||||||
|
LiveEntry::new(aid(1), Some(tid(1)), "x", WorkStatus::Working, None, None, 1).unwrap(),
|
||||||
|
);
|
||||||
|
let before = state.clone();
|
||||||
|
|
||||||
|
let _ = state.reconcile_orphans(|_| false, 999);
|
||||||
|
|
||||||
|
assert_eq!(state, before, "reconcile_orphans is a pure read (returns rows to upsert)");
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn serde_round_trip_is_camel_case() {
|
fn serde_round_trip_is_camel_case() {
|
||||||
let mut state = LiveState::default();
|
let mut state = LiveState::default();
|
||||||
|
|||||||
Reference in New Issue
Block a user