feat(infra,app): store FS live-state + use cases (LS2)
Adapter d'infrastructure et use cases applicatifs du live-state agent. - `FsLiveStateStore` : persistance fichier avec écriture atomique (write-temp + rename) pour éviter tout snapshot partiel/corrompu. - Use cases `UpdateLiveState` / `GetLiveStateLean` + DTO « lean » (vue allégée, bornée pour l'injection/affichage). - Rétention bornée côté store (TTL + max_n) au-dessus des invariants domaine (keyed last-writer-wins, prune). - Garde-fou versionné : `.gitignore` exclut `.ideai/live-state.json` (snapshot runtime reconstruit, non versionné — contrairement à `.ideai/memory/`). cargo test -p infrastructure -p application : 0 échec (dont 5 tests live_state_store) ; cargo fmt --all --check : exit 0. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
3
.gitignore
vendored
3
.gitignore
vendored
@ -40,6 +40,9 @@ frontend/coverage/
|
||||
# Runtime file-protocol orchestration requests/responses — transient I/O, not
|
||||
# durable project state (curation .ideai §chantier secondaire).
|
||||
.ideai/requests/
|
||||
# Volatile agent live-state snapshot ("who is doing what right now", lot LS2):
|
||||
# rebuilt at runtime, keyed last-writer-wins — not versioned (unlike .ideai/memory/).
|
||||
.ideai/live-state.json
|
||||
|
||||
# ─── Editors / OS ───────────────────────────────────────────────────────────
|
||||
.idea/
|
||||
|
||||
@ -122,7 +122,8 @@ pub use window::{MoveTabToNewWindow, MoveTabToNewWindowInput, MoveTabToNewWindow
|
||||
pub use workstate::{
|
||||
AgentTicketState, AgentWorkState, AttachLiveAgent, AttachLiveAgentInput, AttachLiveAgentOutput,
|
||||
ConversationLogProvider, ConversationPreviewStatus, ConversationTurnWorkPreview,
|
||||
ConversationWorkSummary, GetProjectWorkState, GetProjectWorkStateInput, LiveWorkSession,
|
||||
ProjectWorkState, StopLiveAgent, StopLiveAgentInput, StopLiveAgentOutput, TicketWorkSource,
|
||||
TicketWorkStatus,
|
||||
ConversationWorkSummary, GetLiveStateLean, GetProjectWorkState, GetProjectWorkStateInput,
|
||||
LeanLiveEntry, LeanLiveState, LiveWorkSession, ProjectWorkState, StopLiveAgent,
|
||||
StopLiveAgentInput, StopLiveAgentOutput, TicketWorkSource, TicketWorkStatus, UpdateLiveState,
|
||||
UpdateLiveStateInput, LIVE_STATE_MAX_ENTRIES, LIVE_STATE_TTL_MS,
|
||||
};
|
||||
|
||||
343
crates/application/src/workstate/live.rs
Normal file
343
crates/application/src/workstate/live.rs
Normal file
@ -0,0 +1,343 @@
|
||||
//! Live-state write/read use cases (programme live-state, lot LS2).
|
||||
//!
|
||||
//! The **live-state** is the durable-but-volatile "who is doing what right now"
|
||||
//! projection: one [`LiveEntry`] per agent, persisted through the
|
||||
//! [`LiveStateStore`] port (keyed last-writer-wins, never a journal).
|
||||
//!
|
||||
//! This module owns the two use cases over that port:
|
||||
//! - [`UpdateLiveState`] — publish/replace an agent's current row (write side),
|
||||
//! stamping `updated_at_ms` from the injected [`Clock`] and enforcing the
|
||||
//! domain field bounds via [`LiveEntry::new`].
|
||||
//! - [`GetLiveStateLean`] — prune-on-read then return a **distilled** snapshot
|
||||
//! ([`LeanLiveState`]) for agent context injection and the future MCP tools.
|
||||
//!
|
||||
//! ## Boundary (programme invariant)
|
||||
//!
|
||||
//! The live-state stores **only what cannot be re-derived**: an agent's declared
|
||||
//! `intent`/`progress`, its current `ticket` and its `last_delegation`. Busy
|
||||
//! state, live sessions and the delegation queue are **computed elsewhere**
|
||||
//! ([`crate::workstate::GetProjectWorkState`]) and are deliberately *not* copied
|
||||
//! here. The lean DTO is also distinct from that rich, human-facing read model.
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
use domain::live_state::{LiveEntry, LiveState, WorkStatus};
|
||||
use domain::ports::{Clock, LiveStateStore};
|
||||
use domain::{AgentId, TicketId};
|
||||
|
||||
use crate::error::AppError;
|
||||
|
||||
/// Default TTL for a live-state row, in milliseconds: **6 hours**.
|
||||
///
|
||||
/// An entry not refreshed within this window is considered stale and pruned on
|
||||
/// the next read. Six hours comfortably spans a normal working session (an agent
|
||||
/// that has not published anything for that long is effectively offline), while
|
||||
/// still bounding the file so a forgotten/abandoned project does not accumulate
|
||||
/// dead rows indefinitely.
|
||||
pub const LIVE_STATE_TTL_MS: u64 = 6 * 60 * 60 * 1_000;
|
||||
|
||||
/// Default cardinality bound applied after the TTL sweep: keep at most this many
|
||||
/// most-recently-updated rows.
|
||||
///
|
||||
/// A project runs a handful of agents; `64` is a generous ceiling that still
|
||||
/// absorbs transient fan-out (an orchestrator delegating to many agents at once)
|
||||
/// while keeping the snapshot small and the injected context bounded.
|
||||
pub const LIVE_STATE_MAX_ENTRIES: usize = 64;
|
||||
|
||||
/// Input for [`UpdateLiveState::execute`] — the elements of one live-state
|
||||
/// transition. `updated_at_ms` is **not** part of the input: it is stamped by the
|
||||
/// use case from the injected [`Clock`].
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub struct UpdateLiveStateInput {
|
||||
/// The agent publishing the transition (the keyed identity).
|
||||
pub agent_id: AgentId,
|
||||
/// The ticket it is currently handling, if any.
|
||||
pub ticket: Option<TicketId>,
|
||||
/// Short description of the current intent (domain-bounded on construction).
|
||||
pub intent: String,
|
||||
/// Coarse status at a glance.
|
||||
pub status: WorkStatus,
|
||||
/// Optional finer-grained progress note (domain-bounded on construction).
|
||||
pub progress: Option<String>,
|
||||
/// The ticket of the most recent delegation this agent issued, if any.
|
||||
pub last_delegation: Option<TicketId>,
|
||||
}
|
||||
|
||||
/// Publishes (inserts or replaces) an agent's live-state row.
|
||||
pub struct UpdateLiveState {
|
||||
store: Arc<dyn LiveStateStore>,
|
||||
clock: Arc<dyn Clock>,
|
||||
}
|
||||
|
||||
impl UpdateLiveState {
|
||||
/// Builds the use case from the injected store and clock ports.
|
||||
#[must_use]
|
||||
pub fn new(store: Arc<dyn LiveStateStore>, clock: Arc<dyn Clock>) -> Self {
|
||||
Self { store, clock }
|
||||
}
|
||||
|
||||
/// Stamps `updated_at_ms` from the clock, builds a validated [`LiveEntry`]
|
||||
/// (applying the soft truncation / hard rejection bounds) and upserts it.
|
||||
///
|
||||
/// # Errors
|
||||
/// [`AppError::Invalid`] if a text field exceeds the domain hard threshold;
|
||||
/// [`AppError::Store`] on a persistence failure.
|
||||
pub async fn execute(&self, input: UpdateLiveStateInput) -> Result<(), AppError> {
|
||||
let now_ms = u64::try_from(self.clock.now_millis()).unwrap_or(0);
|
||||
let entry = LiveEntry::new(
|
||||
input.agent_id,
|
||||
input.ticket,
|
||||
input.intent,
|
||||
input.status,
|
||||
input.progress,
|
||||
input.last_delegation,
|
||||
now_ms,
|
||||
)
|
||||
.map_err(|e| AppError::Invalid(e.to_string()))?;
|
||||
self.store.upsert(entry).await?;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
/// A single distilled live-state row for injection / MCP — the **lean** DTO,
|
||||
/// deliberately narrower than the rich human read model (no `progress` free-text,
|
||||
/// no timestamp). All free-text is already domain-bounded.
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct LeanLiveEntry {
|
||||
/// The agent this row describes.
|
||||
pub agent_id: AgentId,
|
||||
/// Short description of what the agent is doing (domain-bounded).
|
||||
pub intent: String,
|
||||
/// Coarse status at a glance.
|
||||
pub status: WorkStatus,
|
||||
/// The ticket it is currently handling, if any.
|
||||
pub ticket: Option<TicketId>,
|
||||
/// The ticket of the most recent delegation it issued, if any.
|
||||
pub last_delegation: Option<TicketId>,
|
||||
}
|
||||
|
||||
impl From<LiveEntry> for LeanLiveEntry {
|
||||
fn from(e: LiveEntry) -> Self {
|
||||
Self {
|
||||
agent_id: e.agent_id,
|
||||
intent: e.intent,
|
||||
status: e.status,
|
||||
ticket: e.ticket,
|
||||
last_delegation: e.last_delegation,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The distilled live-state snapshot returned by [`GetLiveStateLean`].
|
||||
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct LeanLiveState {
|
||||
/// The current rows (already pruned), one per agent.
|
||||
pub entries: Vec<LeanLiveEntry>,
|
||||
}
|
||||
|
||||
impl From<LiveState> for LeanLiveState {
|
||||
fn from(state: LiveState) -> Self {
|
||||
Self {
|
||||
entries: state.entries.into_iter().map(LeanLiveEntry::from).collect(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Reads the live-state, pruning stale/excess rows first, and returns the lean
|
||||
/// snapshot.
|
||||
pub struct GetLiveStateLean {
|
||||
store: Arc<dyn LiveStateStore>,
|
||||
clock: Arc<dyn Clock>,
|
||||
}
|
||||
|
||||
impl GetLiveStateLean {
|
||||
/// Builds the use case from the injected store and clock ports.
|
||||
#[must_use]
|
||||
pub fn new(store: Arc<dyn LiveStateStore>, clock: Arc<dyn Clock>) -> Self {
|
||||
Self { store, clock }
|
||||
}
|
||||
|
||||
/// Applies the TTL + cardinality policy ([`LIVE_STATE_TTL_MS`],
|
||||
/// [`LIVE_STATE_MAX_ENTRIES`]) at read time, then returns the distilled
|
||||
/// snapshot.
|
||||
///
|
||||
/// # Errors
|
||||
/// [`AppError::Store`] on a persistence failure.
|
||||
pub async fn execute(&self) -> Result<LeanLiveState, AppError> {
|
||||
let now_ms = u64::try_from(self.clock.now_millis()).unwrap_or(0);
|
||||
self.store
|
||||
.prune(now_ms, LIVE_STATE_TTL_MS, LIVE_STATE_MAX_ENTRIES)
|
||||
.await?;
|
||||
let state = self.store.load().await?;
|
||||
Ok(LeanLiveState::from(state))
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
use std::sync::Mutex;
|
||||
|
||||
use domain::live_state::FIELD_MAX_BYTES;
|
||||
use domain::ports::StoreError;
|
||||
|
||||
/// In-memory [`LiveStateStore`] double mirroring the FS adapter's pure
|
||||
/// semantics (keyed upsert + prune), with 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(())
|
||||
}
|
||||
}
|
||||
|
||||
/// A fixed clock returning a preset epoch-ms value.
|
||||
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))
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn update_stamps_clock_and_persists() {
|
||||
let store = Arc::new(InMemoryLiveStateStore::default());
|
||||
let clock = Arc::new(FixedClock(4_242));
|
||||
let uc = UpdateLiveState::new(store.clone(), clock);
|
||||
|
||||
uc.execute(UpdateLiveStateInput {
|
||||
agent_id: aid(1),
|
||||
ticket: None,
|
||||
intent: "shipping".to_owned(),
|
||||
status: WorkStatus::Working,
|
||||
progress: None,
|
||||
last_delegation: None,
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let state = store.load().await.unwrap();
|
||||
assert_eq!(state.entries.len(), 1);
|
||||
assert_eq!(state.entries[0].intent, "shipping");
|
||||
assert_eq!(
|
||||
state.entries[0].updated_at_ms, 4_242,
|
||||
"updated_at_ms stamped from the clock"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn update_applies_domain_bounds_truncation() {
|
||||
let store = Arc::new(InMemoryLiveStateStore::default());
|
||||
let uc = UpdateLiveState::new(store.clone(), Arc::new(FixedClock(1)));
|
||||
|
||||
let long = "x".repeat(500);
|
||||
uc.execute(UpdateLiveStateInput {
|
||||
agent_id: aid(1),
|
||||
ticket: None,
|
||||
intent: long,
|
||||
status: WorkStatus::Working,
|
||||
progress: None,
|
||||
last_delegation: None,
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let state = store.load().await.unwrap();
|
||||
assert!(
|
||||
state.entries[0].intent.chars().count() <= 200,
|
||||
"soft bound truncated the over-long intent"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn update_rejects_oversize_field() {
|
||||
let store = Arc::new(InMemoryLiveStateStore::default());
|
||||
let uc = UpdateLiveState::new(store, Arc::new(FixedClock(1)));
|
||||
|
||||
let huge = "x".repeat(FIELD_MAX_BYTES + 1);
|
||||
let err = uc
|
||||
.execute(UpdateLiveStateInput {
|
||||
agent_id: aid(1),
|
||||
ticket: None,
|
||||
intent: huge,
|
||||
status: WorkStatus::Working,
|
||||
progress: None,
|
||||
last_delegation: None,
|
||||
})
|
||||
.await
|
||||
.expect_err("oversize intent must be rejected");
|
||||
assert!(matches!(err, AppError::Invalid(_)));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn get_lean_prunes_on_read_and_returns_lean_dto() {
|
||||
let store = Arc::new(InMemoryLiveStateStore::default());
|
||||
// One fresh row (now-ish) and one ancient row (well beyond the TTL).
|
||||
let now: i64 = 10 * 60 * 60 * 1_000; // 10h in ms
|
||||
store
|
||||
.upsert(
|
||||
LiveEntry::new(
|
||||
aid(1),
|
||||
None,
|
||||
"fresh",
|
||||
WorkStatus::Working,
|
||||
Some("detailed progress note".to_owned()),
|
||||
None,
|
||||
u64::try_from(now).unwrap(),
|
||||
)
|
||||
.unwrap(),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
store
|
||||
.upsert(LiveEntry::new(aid(2), None, "stale", WorkStatus::Idle, None, None, 0).unwrap())
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let lean = GetLiveStateLean::new(store.clone(), Arc::new(FixedClock(now)))
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// The ancient row (age 10h > 6h TTL) was pruned on read.
|
||||
assert_eq!(lean.entries.len(), 1);
|
||||
let entry = &lean.entries[0];
|
||||
assert_eq!(entry.agent_id, aid(1));
|
||||
assert_eq!(entry.intent, "fresh");
|
||||
assert_eq!(entry.status, WorkStatus::Working);
|
||||
|
||||
// Lean DTO is narrow: no `progress` free-text leaks through. The rich
|
||||
// serialized form must not carry it.
|
||||
let json = serde_json::to_string(&lean).unwrap();
|
||||
assert!(
|
||||
!json.contains("progress"),
|
||||
"lean DTO drops progress: {json}"
|
||||
);
|
||||
assert!(!json.contains("detailed progress note"));
|
||||
assert!(json.contains("\"agentId\""), "camelCase lean DTO");
|
||||
|
||||
// Prune persisted: the store no longer holds the stale row either.
|
||||
assert_eq!(store.load().await.unwrap().entries.len(), 1);
|
||||
}
|
||||
}
|
||||
@ -6,11 +6,16 @@
|
||||
//! durable projection.
|
||||
|
||||
mod actions;
|
||||
mod live;
|
||||
|
||||
pub use actions::{
|
||||
AttachLiveAgent, AttachLiveAgentInput, AttachLiveAgentOutput, StopLiveAgent,
|
||||
StopLiveAgentInput, StopLiveAgentOutput,
|
||||
};
|
||||
pub use live::{
|
||||
GetLiveStateLean, LeanLiveEntry, LeanLiveState, UpdateLiveState, UpdateLiveStateInput,
|
||||
LIVE_STATE_MAX_ENTRIES, LIVE_STATE_TTL_MS,
|
||||
};
|
||||
|
||||
use std::collections::{HashMap, HashSet};
|
||||
use std::sync::Arc;
|
||||
|
||||
@ -72,8 +72,8 @@ pub use store::{detect_ollama, HttpEmbedder, DEFAULT_LOCAL_EMBED_ENDPOINT};
|
||||
pub use store::{
|
||||
embedder_from_profile, index_token_size, onnx_model_is_cached, should_use_vector,
|
||||
AdaptiveMemoryRecall, EmbedderEnvProbe, FsEmbedderProfileStore, FsEmbedderPromptStore,
|
||||
FsMemoryStore, FsPermissionStore, FsProfileStore, FsProjectStore, FsSkillStore,
|
||||
FsTemplateStore, HashEmbedder, IdeaiContextStore, NaiveMemoryRecall, OnnxModelInfo,
|
||||
StubEmbedder, VectorMemoryRecall, DEFAULT_OLLAMA_BASE_URL, ONNX_CACHE_SUBDIR,
|
||||
FsLiveStateStore, FsMemoryStore, FsPermissionStore, FsProfileStore, FsProjectStore,
|
||||
FsSkillStore, FsTemplateStore, HashEmbedder, IdeaiContextStore, NaiveMemoryRecall,
|
||||
OnnxModelInfo, StubEmbedder, VectorMemoryRecall, DEFAULT_OLLAMA_BASE_URL, ONNX_CACHE_SUBDIR,
|
||||
RECOMMENDED_ONNX_MODELS, VECTOR_HTTP_ENABLED, VECTOR_ONNX_ENABLED,
|
||||
};
|
||||
|
||||
128
crates/infrastructure/src/store/live_state.rs
Normal file
128
crates/infrastructure/src/store/live_state.rs
Normal file
@ -0,0 +1,128 @@
|
||||
//! [`FsLiveStateStore`] — l'adapter `tokio::fs` du port [`LiveStateStore`]
|
||||
//! (programme live-state, lot LS2).
|
||||
//!
|
||||
//! Le **live-state** est la projection volatile « qui fait quoi en ce moment » :
|
||||
//! une entrée [`LiveEntry`](domain::live_state::LiveEntry) par agent, sémantique
|
||||
//! *keyed last-writer-wins* (jamais un journal). On en garde **un seul** fichier
|
||||
//! par projet, à la racine du `.ideai/` :
|
||||
//!
|
||||
//! ```text
|
||||
//! <project_root>/.ideai/live-state.json
|
||||
//! ```
|
||||
//!
|
||||
//! Le format est le JSON camelCase de [`LiveState`](domain::live_state::LiveState)
|
||||
//! (dérivé serde du domaine), lisible à l'œil.
|
||||
//!
|
||||
//! ## Robustesse
|
||||
//!
|
||||
//! - **Écriture atomique** : on écrit dans `live-state.json.tmp` puis on `rename`
|
||||
//! — un lecteur ne voit jamais de fichier à moitié écrit (le `rename` est
|
||||
//! atomique sur le FS). Même pattern que [`FsHandoffStore`](crate::FsHandoffStore).
|
||||
//! - **Fichier absent** ⇒ [`LiveState`] vide (jamais une erreur) : c'est l'état
|
||||
//! normal d'un projet qui n'a encore rien publié.
|
||||
//! - **Fichier présent mais illisible** (JSON corrompu/partiel) ⇒
|
||||
//! [`StoreError::Serialization`]. Choix **aligné** sur les autres stores FS
|
||||
//! (`FsHandoffStore`, l'`index.json` de `FsSkillStore`) : on ne panique jamais,
|
||||
//! mais on ne masque pas non plus une corruption — l'écriture atomique rend ce
|
||||
//! cas très improbable, et le fichier étant runtime/non-versionné, le supprimer
|
||||
//! suffit à repartir d'un état vide.
|
||||
//!
|
||||
//! Le project root est fourni au constructeur (le port [`LiveStateStore`] n'a pas
|
||||
//! de paramètre `root` par appel, contrairement à `SkillStore`/`MemoryStore`) ;
|
||||
//! une instance sert donc **un** projet.
|
||||
|
||||
use std::path::PathBuf;
|
||||
|
||||
use async_trait::async_trait;
|
||||
|
||||
use domain::live_state::{LiveEntry, LiveState};
|
||||
use domain::ports::{LiveStateStore, StoreError};
|
||||
use domain::project::ProjectPath;
|
||||
|
||||
/// Le dossier `.ideai/` dans une racine de projet.
|
||||
const IDEAI_DIR: &str = ".ideai";
|
||||
|
||||
/// Nom du fichier de live-state, par projet.
|
||||
const LIVE_STATE_FILE: &str = "live-state.json";
|
||||
|
||||
/// Nom du fichier temporaire d'écriture atomique (renommé sur `live-state.json`).
|
||||
const LIVE_STATE_TMP_FILE: &str = "live-state.json.tmp";
|
||||
|
||||
/// Adapter `tokio::fs` du store de live-state, un `live-state.json` par projet.
|
||||
///
|
||||
/// Même convention de construction que [`FsHandoffStore`](crate::FsHandoffStore) :
|
||||
/// le **project root** est fourni au constructeur, la base `<root>/.ideai` en
|
||||
/// dérive, et le fichier est créé paresseusement à la première écriture.
|
||||
pub struct FsLiveStateStore {
|
||||
/// Racine `<project_root>/.ideai`.
|
||||
dir: PathBuf,
|
||||
}
|
||||
|
||||
impl FsLiveStateStore {
|
||||
/// Construit l'adapter à partir du **project root**.
|
||||
#[must_use]
|
||||
pub fn new(root: &ProjectPath) -> Self {
|
||||
let dir = PathBuf::from(root.as_str()).join(IDEAI_DIR);
|
||||
Self { dir }
|
||||
}
|
||||
|
||||
/// `<root>/.ideai/live-state.json` — le fichier cible.
|
||||
fn path(&self) -> PathBuf {
|
||||
self.dir.join(LIVE_STATE_FILE)
|
||||
}
|
||||
|
||||
/// `<root>/.ideai/live-state.json.tmp` — le tmp d'écriture atomique.
|
||||
fn tmp_path(&self) -> PathBuf {
|
||||
self.dir.join(LIVE_STATE_TMP_FILE)
|
||||
}
|
||||
|
||||
/// Lit l'état courant : fichier absent ⇒ état vide ; JSON corrompu ⇒ erreur.
|
||||
async fn read_state(&self) -> Result<LiveState, StoreError> {
|
||||
match tokio::fs::read(self.path()).await {
|
||||
Ok(bytes) => {
|
||||
serde_json::from_slice(&bytes).map_err(|e| StoreError::Serialization(e.to_string()))
|
||||
}
|
||||
// Absent ⇒ état vide (jamais une erreur).
|
||||
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(LiveState::default()),
|
||||
Err(e) => Err(StoreError::Io(e.to_string())),
|
||||
}
|
||||
}
|
||||
|
||||
/// Réécrit l'état **atomiquement** : tmp puis `rename` sur la cible.
|
||||
async fn write_state(&self, state: &LiveState) -> Result<(), StoreError> {
|
||||
tokio::fs::create_dir_all(&self.dir)
|
||||
.await
|
||||
.map_err(|e| StoreError::Io(e.to_string()))?;
|
||||
|
||||
let bytes = serde_json::to_vec_pretty(state)
|
||||
.map_err(|e| StoreError::Serialization(e.to_string()))?;
|
||||
|
||||
let tmp = self.tmp_path();
|
||||
tokio::fs::write(&tmp, &bytes)
|
||||
.await
|
||||
.map_err(|e| StoreError::Io(e.to_string()))?;
|
||||
tokio::fs::rename(&tmp, self.path())
|
||||
.await
|
||||
.map_err(|e| StoreError::Io(e.to_string()))?;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl LiveStateStore for FsLiveStateStore {
|
||||
async fn load(&self) -> Result<LiveState, StoreError> {
|
||||
self.read_state().await
|
||||
}
|
||||
|
||||
async fn upsert(&self, entry: LiveEntry) -> Result<(), StoreError> {
|
||||
let mut state = self.read_state().await?;
|
||||
state.upsert(entry);
|
||||
self.write_state(&state).await
|
||||
}
|
||||
|
||||
async fn prune(&self, now_ms: u64, ttl_ms: u64, max_n: usize) -> Result<(), StoreError> {
|
||||
let mut state = self.read_state().await?;
|
||||
state.prune(now_ms, ttl_ms, max_n);
|
||||
self.write_state(&state).await
|
||||
}
|
||||
}
|
||||
@ -6,6 +6,7 @@
|
||||
|
||||
mod context;
|
||||
mod embedder;
|
||||
mod live_state;
|
||||
mod memory;
|
||||
mod permission;
|
||||
mod profile;
|
||||
@ -24,6 +25,7 @@ pub use embedder::{
|
||||
HashEmbedder, OnnxModelInfo, StubEmbedder, DEFAULT_OLLAMA_BASE_URL, ONNX_CACHE_SUBDIR,
|
||||
RECOMMENDED_ONNX_MODELS, VECTOR_HTTP_ENABLED, VECTOR_ONNX_ENABLED,
|
||||
};
|
||||
pub use live_state::FsLiveStateStore;
|
||||
pub use memory::{index_token_size, FsMemoryStore, NaiveMemoryRecall};
|
||||
pub use permission::FsPermissionStore;
|
||||
pub use profile::{FsEmbedderProfileStore, FsProfileStore};
|
||||
|
||||
157
crates/infrastructure/tests/live_state_store.rs
Normal file
157
crates/infrastructure/tests/live_state_store.rs
Normal file
@ -0,0 +1,157 @@
|
||||
//! LS2 integration tests for [`FsLiveStateStore`] against a **real** temp
|
||||
//! directory (programme live-state).
|
||||
//!
|
||||
//! These lock the durable behaviour an in-memory double cannot prove:
|
||||
//! - round-trip `upsert`/`load` survives a fresh instance ("restart");
|
||||
//! - a missing file => empty `LiveState` (never an error);
|
||||
//! - the write is atomic (no `.tmp` residue left behind);
|
||||
//! - `upsert` is keyed last-writer-wins, persisted (no duplicate per agent);
|
||||
//! - `prune` (TTL + max_n) is persisted.
|
||||
//!
|
||||
//! Convention: a hand-rolled [`TempDir`] over the OS temp dir (calqué sur
|
||||
//! `conversation_log.rs`) — absolute path for `ProjectPath::new`, cleaned on drop.
|
||||
|
||||
use std::path::PathBuf;
|
||||
|
||||
use domain::live_state::{LiveEntry, WorkStatus};
|
||||
use domain::ports::LiveStateStore;
|
||||
use domain::project::ProjectPath;
|
||||
use domain::AgentId;
|
||||
use infrastructure::FsLiveStateStore;
|
||||
use uuid::Uuid;
|
||||
|
||||
struct TempDir(PathBuf);
|
||||
impl TempDir {
|
||||
fn new() -> Self {
|
||||
let p = std::env::temp_dir().join(format!("idea-ls2-livestate-{}", Uuid::new_v4()));
|
||||
std::fs::create_dir_all(&p).unwrap();
|
||||
Self(p)
|
||||
}
|
||||
fn project_path(&self) -> ProjectPath {
|
||||
ProjectPath::new(self.0.to_string_lossy().into_owned()).unwrap()
|
||||
}
|
||||
fn file_path(&self) -> PathBuf {
|
||||
self.0.join(".ideai").join("live-state.json")
|
||||
}
|
||||
fn tmp_path(&self) -> PathBuf {
|
||||
self.0.join(".ideai").join("live-state.json.tmp")
|
||||
}
|
||||
}
|
||||
impl Drop for TempDir {
|
||||
fn drop(&mut self) {
|
||||
let _ = std::fs::remove_dir_all(&self.0);
|
||||
}
|
||||
}
|
||||
|
||||
fn aid(n: u128) -> AgentId {
|
||||
AgentId::from_uuid(Uuid::from_u128(n))
|
||||
}
|
||||
|
||||
fn entry(agent: u128, intent: &str, status: WorkStatus, updated_at_ms: u64) -> LiveEntry {
|
||||
LiveEntry::new(aid(agent), None, intent, status, None, None, updated_at_ms).unwrap()
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn missing_file_loads_empty_without_error() {
|
||||
let tmp = TempDir::new();
|
||||
let store = FsLiveStateStore::new(&tmp.project_path());
|
||||
|
||||
let state = store.load().await.unwrap();
|
||||
assert!(state.entries.is_empty(), "no file ⇒ empty live-state");
|
||||
// load() must not have created the file.
|
||||
assert!(!tmp.file_path().exists(), "load does not write");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn upsert_round_trips_across_a_fresh_instance() {
|
||||
let tmp = TempDir::new();
|
||||
{
|
||||
let store = FsLiveStateStore::new(&tmp.project_path());
|
||||
store
|
||||
.upsert(entry(1, "building", WorkStatus::Working, 100))
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
// A fresh instance on the same root relits the persisted row.
|
||||
let store = FsLiveStateStore::new(&tmp.project_path());
|
||||
let state = store.load().await.unwrap();
|
||||
assert_eq!(state.entries.len(), 1);
|
||||
assert_eq!(state.entries[0].intent, "building");
|
||||
assert_eq!(state.entries[0].status, WorkStatus::Working);
|
||||
|
||||
// Persisted JSON is camelCase.
|
||||
let raw = std::fs::read_to_string(tmp.file_path()).unwrap();
|
||||
assert!(raw.contains("\"agentId\""), "camelCase on disk: {raw}");
|
||||
assert!(raw.contains("\"updatedAtMs\""));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn write_is_atomic_no_tmp_residue() {
|
||||
let tmp = TempDir::new();
|
||||
let store = FsLiveStateStore::new(&tmp.project_path());
|
||||
store
|
||||
.upsert(entry(1, "a", WorkStatus::Working, 1))
|
||||
.await
|
||||
.unwrap();
|
||||
store
|
||||
.upsert(entry(2, "b", WorkStatus::Idle, 2))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert!(tmp.file_path().exists(), "final file present");
|
||||
assert!(
|
||||
!tmp.tmp_path().exists(),
|
||||
"tmp file must be renamed away, not left behind"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn upsert_is_last_writer_wins_persisted() {
|
||||
let tmp = TempDir::new();
|
||||
let store = FsLiveStateStore::new(&tmp.project_path());
|
||||
|
||||
store
|
||||
.upsert(entry(1, "first", WorkStatus::Working, 1))
|
||||
.await
|
||||
.unwrap();
|
||||
store
|
||||
.upsert(entry(1, "second", WorkStatus::Blocked, 2))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let state = store.load().await.unwrap();
|
||||
assert_eq!(state.entries.len(), 1, "same agent ⇒ one row, no duplicate");
|
||||
assert_eq!(state.entries[0].intent, "second");
|
||||
assert_eq!(state.entries[0].status, WorkStatus::Blocked);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn prune_ttl_and_max_n_persisted() {
|
||||
let tmp = TempDir::new();
|
||||
let store = FsLiveStateStore::new(&tmp.project_path());
|
||||
|
||||
// Three rows at t=10/20/30; now=50, ttl=35 ⇒ t=10 (age 40>35) dropped,
|
||||
// t=20 (age 30) and t=30 (age 20) kept.
|
||||
store
|
||||
.upsert(entry(1, "old", WorkStatus::Idle, 10))
|
||||
.await
|
||||
.unwrap();
|
||||
store
|
||||
.upsert(entry(2, "mid", WorkStatus::Working, 20))
|
||||
.await
|
||||
.unwrap();
|
||||
store
|
||||
.upsert(entry(3, "new", WorkStatus::Working, 30))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
store.prune(50, 35, 100).await.unwrap();
|
||||
let after_ttl = store.load().await.unwrap();
|
||||
assert_eq!(after_ttl.entries.len(), 2, "TTL dropped the oldest");
|
||||
|
||||
// Now bound to 1 ⇒ keep the most recent (t=30).
|
||||
store.prune(50, 35, 1).await.unwrap();
|
||||
let after_cap = store.load().await.unwrap();
|
||||
assert_eq!(after_cap.entries.len(), 1);
|
||||
assert_eq!(after_cap.entries[0].updated_at_ms, 30);
|
||||
}
|
||||
Reference in New Issue
Block a user