From f3046f3dd8bf1d747b11fb43fe3030bd2fa3056b Mon Sep 17 00:00:00 2001 From: Blomios Date: Fri, 12 Jun 2026 13:14:34 +0200 Subject: [PATCH] =?UTF-8?q?feat(persistence):=20P5=20=E2=80=94=20ProviderS?= =?UTF-8?q?essionStore=20(resumable=5Fid=20par=20provider)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Range le resumable_id du moteur par (conversation, provider) dans providers.json, support de reprise non exclusif (le handoff reste la source de fidélité au swap cross-profile) — §19.2/§19.3. - domaine : port ProviderSessionStore (get/set par provider) - infra : FsProviderSessionStore — providers.json map plate, set en read-modify-write atomique (tmp+rename) sérialisé par conversation (coexistence multi-providers garantie), get absent ⇒ None, corrompu ⇒ StoreError::Serialization Tests : 6 cas P5 (coexistence claude+codex, 12 set concurrents sans perte, corruption, isolation) ; cible conversation_log 30 verts, suites complètes sans régression. Co-Authored-By: Claude Opus 4.8 --- crates/domain/src/conversation_log.rs | 48 ++++ crates/domain/src/lib.rs | 3 +- .../src/conversation_log/mod.rs | 2 + .../src/conversation_log/providers.rs | 174 +++++++++++++ crates/infrastructure/src/lib.rs | 4 +- .../infrastructure/tests/conversation_log.rs | 240 +++++++++++++++++- 6 files changed, 467 insertions(+), 4 deletions(-) create mode 100644 crates/infrastructure/src/conversation_log/providers.rs diff --git a/crates/domain/src/conversation_log.rs b/crates/domain/src/conversation_log.rs index 7987706..410ac2c 100644 --- a/crates/domain/src/conversation_log.rs +++ b/crates/domain/src/conversation_log.rs @@ -278,6 +278,54 @@ pub trait HandoffSummarizer: Send + Sync { async fn fold(&self, prev: Option, new_turns: &[ConversationTurn]) -> Handoff; } +/// Le store des `resumable_id` CLI, rangés **par (conversation, provider)** (port +/// driven, lot P5). +/// +/// Le `resumable_id` est l'identifiant **propre au moteur** (le `--resume`/`--continue` +/// d'une CLI : `"claude"`, `"codex"`…) qui permet de **réattacher** la session native du +/// provider à la reprise (ARCHITECTURE §19.2/§19.3). Contrairement au [`Handoff`] +/// (résumé **portable**, indépendant du provider, source de vérité de la continuité), +/// ce store garde l'optimisation **non portable** : chaque provider a **son** id, et +/// plusieurs providers peuvent coexister pour une même conversation (après un swap +/// Claude→Codex, on garde l'id de chacun). La clé est donc bien le couple +/// `(conversation, provider_id)`. +/// +/// `#[async_trait]` et erreur [`StoreError`] comme les ports voisins +/// ([`ConversationLog`], [`HandoffStore`]) : injecté en `Arc` +/// au composition root, gardé object-safe. La persistance réelle (`providers.json`) est +/// une affaire d'infrastructure (l'adapter `FsProviderSessionStore`, lot P5). +#[async_trait::async_trait] +pub trait ProviderSessionStore: Send + Sync { + /// Charge le `resumable_id` du provider `provider_id` pour `conversation`. + /// + /// Renvoie `Ok(Some(id))` si un id a été rangé pour ce couple, `Ok(None)` si le + /// provider (ou la conversation) n'a encore rien rangé. L'absence n'est **jamais** + /// une erreur. + /// + /// # Errors + /// [`StoreError`] en cas d'échec de lecture ou de désérialisation. + async fn get( + &self, + conversation: ConversationId, + provider_id: &str, + ) -> Result, StoreError>; + + /// Range (en écrasant) le `resumable_id` du provider `provider_id` pour + /// `conversation`. + /// + /// Les autres providers déjà rangés pour la même conversation **coexistent** : seul + /// l'enregistrement de `provider_id` est créé/écrasé. + /// + /// # Errors + /// [`StoreError`] en cas d'échec d'écriture ou de sérialisation. + async fn set( + &self, + conversation: ConversationId, + provider_id: &str, + resumable_id: &str, + ) -> Result<(), StoreError>; +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/domain/src/lib.rs b/crates/domain/src/lib.rs index 38e4a71..393672e 100644 --- a/crates/domain/src/lib.rs +++ b/crates/domain/src/lib.rs @@ -87,7 +87,8 @@ pub use conversation::{ pub use input::{AgentBusyState, InputMediator, InputSource}; pub use conversation_log::{ - ConversationLog, ConversationTurn, Handoff, HandoffStore, HandoffSummarizer, TurnId, TurnRole, + ConversationLog, ConversationTurn, Handoff, HandoffStore, HandoffSummarizer, + ProviderSessionStore, TurnId, TurnRole, }; pub use fileguard::{ diff --git a/crates/infrastructure/src/conversation_log/mod.rs b/crates/infrastructure/src/conversation_log/mod.rs index 633cef6..b9dbcc1 100644 --- a/crates/infrastructure/src/conversation_log/mod.rs +++ b/crates/infrastructure/src/conversation_log/mod.rs @@ -44,9 +44,11 @@ use domain::ports::StoreError; use domain::project::ProjectPath; mod handoff; +mod providers; mod summarizer; pub use handoff::FsHandoffStore; +pub use providers::FsProviderSessionStore; pub use summarizer::{HeuristicHandoffSummarizer, WINDOW}; /// Dossier `.ideai/` à la racine d'un project root. diff --git a/crates/infrastructure/src/conversation_log/providers.rs b/crates/infrastructure/src/conversation_log/providers.rs new file mode 100644 index 0000000..6aa8a2d --- /dev/null +++ b/crates/infrastructure/src/conversation_log/providers.rs @@ -0,0 +1,174 @@ +//! [`FsProviderSessionStore`] — l'adapter `tokio::fs` du port [`ProviderSessionStore`] +//! (cadrage « persistance conversationnelle », lot P5). +//! +//! Le `resumable_id` **propre au moteur** (le `--resume`/`--continue` d'une CLI) est +//! rangé **par (conversation, provider)** (ARCHITECTURE §19.2/§19.3) dans un seul fichier +//! JSON par conversation : une **map `providerId → resumableId`** où plusieurs providers +//! coexistent (après un swap Claude→Codex, on garde l'id de chacun) : +//! +//! ```text +//! /.ideai/conversations/ +//! └── / +//! ├── log.jsonl # le log append-only (lot P2) +//! ├── handoff.md # le dernier point de reprise (lot P3) +//! └── providers.json # { "": "", ... } (ce module) +//! ``` +//! +//! ## Format `providers.json` +//! +//! Un objet JSON plat `{ "claude": "abc-123", "codex": "def-456" }`. C'est l'unité +//! écrite/relue : pas de wrapper, la clé est l'identifiant de profil/moteur. +//! +//! ## Robustesse +//! +//! - **Écriture atomique** : on écrit dans `providers.json.tmp` puis on `rename` — +//! jamais de fichier à moitié écrit visible (même convention que [`super::FsHandoffStore`]). +//! - **Lecture-modification-écriture** : `set` charge la map existante, insère/écrase la +//! seule clé `provider_id`, puis réécrit — les autres providers ne sont jamais perdus. +//! L'opération est **sérialisée par conversation** (un `Mutex` async par fichier, comme +//! [`super::FsConversationLog`]) pour qu'un `set` concurrent ne perde pas l'écriture de +//! l'autre (read-modify-write atomique du point de vue applicatif). +//! - **Fichier ou clé absent** ⇒ `get` renvoie `Ok(None)` (jamais une erreur) : un +//! provider sans session rangée est l'état normal au premier tour. +//! - **Fichier présent mais illisible** (JSON corrompu) ⇒ [`StoreError::Serialization`]. +//! Contrairement au log JSONL (où une ligne corrompue est sautée), il n'y a qu'un +//! enregistrement : on ne peut pas « sauter », c'est une vraie erreur. + +use std::collections::HashMap; +use std::path::PathBuf; +use std::sync::{Arc, Mutex}; + +use async_trait::async_trait; + +use domain::conversation::ConversationId; +use domain::conversation_log::ProviderSessionStore; +use domain::ports::StoreError; +use domain::project::ProjectPath; + +use super::{CONVERSATIONS_DIR, IDEAI_DIR}; + +/// Nom du fichier des sessions par provider, par conversation. +const PROVIDERS_FILE: &str = "providers.json"; + +/// Nom du fichier temporaire d'écriture atomique (renommé sur `providers.json`). +const PROVIDERS_TMP_FILE: &str = "providers.json.tmp"; + +/// La map persistée : `providerId → resumableId`. +type ProviderMap = HashMap; + +/// Adapter `tokio::fs` du store des `resumable_id`, un `providers.json` par conversation. +/// +/// Même convention de construction que [`super::FsConversationLog`] / [`super::FsHandoffStore`] : +/// le **project root** est fourni au constructeur, la base `/.ideai/conversations` +/// en dérive, et chaque conversation a son sous-dossier `/`. +pub struct FsProviderSessionStore { + /// Racine `/.ideai/conversations`. + base: PathBuf, + /// Verrous d'écriture, un par fichier de conversation (sérialise le read-modify-write). + write_locks: Mutex>>>, +} + +impl FsProviderSessionStore { + /// Construit l'adapter à partir du **project root**. + /// + /// La base `/.ideai/conversations` en est dérivée ; le dossier de conversation + /// est créé paresseusement au premier `set`. + #[must_use] + pub fn new(root: &ProjectPath) -> Self { + let base = PathBuf::from(root.as_str()) + .join(IDEAI_DIR) + .join(CONVERSATIONS_DIR); + Self { + base, + write_locks: Mutex::new(HashMap::new()), + } + } + + /// `/` — le dossier d'une conversation. + fn conversation_dir(&self, conversation: ConversationId) -> PathBuf { + self.base.join(conversation.to_string()) + } + + /// `//providers.json` — le fichier des sessions par provider. + fn providers_path(&self, conversation: ConversationId) -> PathBuf { + self.conversation_dir(conversation).join(PROVIDERS_FILE) + } + + /// `//providers.json.tmp` — le fichier temporaire d'écriture. + fn providers_tmp_path(&self, conversation: ConversationId) -> PathBuf { + self.conversation_dir(conversation).join(PROVIDERS_TMP_FILE) + } + + /// Renvoie (en le créant au besoin) le verrou d'écriture de `conversation`. + fn write_lock(&self, conversation: ConversationId) -> Arc> { + self.write_locks + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .entry(conversation) + .or_default() + .clone() + } + + /// Charge la map `providerId → resumableId` de `conversation`. + /// + /// Fichier absent ⇒ map vide (jamais une erreur). JSON illisible ⇒ + /// [`StoreError::Serialization`]. + async fn load_map(&self, conversation: ConversationId) -> Result { + let bytes = match tokio::fs::read(self.providers_path(conversation)).await { + Ok(bytes) => bytes, + Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(ProviderMap::new()), + Err(e) => return Err(StoreError::Io(e.to_string())), + }; + serde_json::from_slice(&bytes) + .map_err(|e| StoreError::Serialization(format!("providers.json: {e}"))) + } +} + +#[async_trait] +impl ProviderSessionStore for FsProviderSessionStore { + async fn get( + &self, + conversation: ConversationId, + provider_id: &str, + ) -> Result, StoreError> { + let map = self.load_map(conversation).await?; + Ok(map.get(provider_id).cloned()) + } + + async fn set( + &self, + conversation: ConversationId, + provider_id: &str, + resumable_id: &str, + ) -> Result<(), StoreError> { + // Read-modify-write sérialisé par conversation : le verrou est tenu le temps du + // load + insert + write atomique, donc deux `set` concurrents (providers + // différents ou non) s'ordonnent sans s'écraser. + let lock = self.write_lock(conversation); + let _guard = lock.lock().await; + + let mut map = self.load_map(conversation).await?; + map.insert(provider_id.to_string(), resumable_id.to_string()); + + // Sérialiser **avant** toute I/O d'écriture : une erreur de sérialisation ne doit + // pas laisser de fichier tmp partiel. + let body = + serde_json::to_vec(&map).map_err(|e| StoreError::Serialization(e.to_string()))?; + + let dir = self.conversation_dir(conversation); + tokio::fs::create_dir_all(&dir) + .await + .map_err(|e| StoreError::Io(e.to_string()))?; + + // Écriture atomique : écrire le tmp puis `rename` sur la cible. Un lecteur ne voit + // jamais de fichier à moitié écrit (le rename est atomique sur le FS). + let tmp = self.providers_tmp_path(conversation); + tokio::fs::write(&tmp, &body) + .await + .map_err(|e| StoreError::Io(e.to_string()))?; + tokio::fs::rename(&tmp, self.providers_path(conversation)) + .await + .map_err(|e| StoreError::Io(e.to_string()))?; + Ok(()) + } +} diff --git a/crates/infrastructure/src/lib.rs b/crates/infrastructure/src/lib.rs index 9cafcde..b458629 100644 --- a/crates/infrastructure/src/lib.rs +++ b/crates/infrastructure/src/lib.rs @@ -33,7 +33,9 @@ pub mod store; pub use clock::SystemClock; pub use conversation::InMemoryConversationRegistry; -pub use conversation_log::{FsConversationLog, FsHandoffStore, HeuristicHandoffSummarizer, WINDOW}; +pub use conversation_log::{ + FsConversationLog, FsHandoffStore, FsProviderSessionStore, HeuristicHandoffSummarizer, WINDOW, +}; pub use eventbus::TokioBroadcastEventBus; pub use fileguard::RwFileGuard; pub use input::{MediatedInbox, MillisClock, SystemMillisClock}; diff --git a/crates/infrastructure/tests/conversation_log.rs b/crates/infrastructure/tests/conversation_log.rs index 6e5b08e..37ef810 100644 --- a/crates/infrastructure/tests/conversation_log.rs +++ b/crates/infrastructure/tests/conversation_log.rs @@ -18,12 +18,15 @@ use std::path::PathBuf; use domain::conversation::ConversationId; use domain::conversation_log::{ - ConversationLog, ConversationTurn, Handoff, HandoffStore, HandoffSummarizer, TurnId, TurnRole, + ConversationLog, ConversationTurn, Handoff, HandoffStore, HandoffSummarizer, + ProviderSessionStore, TurnId, TurnRole, }; use domain::input::InputSource; use domain::ports::StoreError; use domain::project::ProjectPath; -use infrastructure::{FsConversationLog, FsHandoffStore, HeuristicHandoffSummarizer, WINDOW}; +use infrastructure::{ + FsConversationLog, FsHandoffStore, FsProviderSessionStore, HeuristicHandoffSummarizer, WINDOW, +}; use uuid::Uuid; // --------------------------------------------------------------------------- @@ -62,6 +65,14 @@ impl TempDir { fn handoff_tmp_path(&self, conversation: ConversationId) -> PathBuf { self.conversation_dir(conversation).join("handoff.md.tmp") } + /// `/.ideai/conversations//providers.json` — the per-provider sessions (P5). + fn providers_path(&self, conversation: ConversationId) -> PathBuf { + self.conversation_dir(conversation).join("providers.json") + } + /// `/.ideai/conversations//providers.json.tmp` — the atomic-write tmp (P5). + fn providers_tmp_path(&self, conversation: ConversationId) -> PathBuf { + self.conversation_dir(conversation).join("providers.json.tmp") + } } impl Drop for TempDir { fn drop(&mut self) { @@ -761,3 +772,228 @@ async fn fold_flattens_multiline_and_marker_like_text_into_one_line() { assert_eq!(lines2[2], "- **Response:** suite"); assert_eq!(h2.up_to, turn_id(3)); } + +// =========================================================================== +// P5 — FsProviderSessionStore (providers.json): le `resumable_id` rangé par +// (conversation, provider). Mêmes helpers L2 que P2/P3/P4 — `TempDir` maison +// (chemin absolu pour `ProjectPath::new`), constructeurs déterministes +// `conv_id`/`turn_id`, convention de dossier `/.ideai/conversations//`. +// +// Format verrouillé : objet JSON plat `{ "": "", ... }`. +// Contrat : get/set par provider ; coexistence multi-providers sans perte ; +// absent ⇒ Ok(None) ; corrompu ⇒ Err(Serialization) ; read-modify-write atomique +// (tmp+rename, mutex par conversation) ; isolation par ConversationId. +// =========================================================================== + +/// Écrit un `providers.json` brut (corruption volontaire), créant le dossier au besoin. +fn write_raw_providers(tmp: &TempDir, c: ConversationId, content: &str) { + std::fs::create_dir_all(tmp.conversation_dir(c)).unwrap(); + std::fs::write(tmp.providers_path(c), content).unwrap(); +} + +// --------------------------------------------------------------------------- +// get/set par provider + round-trip disque via une nouvelle instance +// --------------------------------------------------------------------------- + +#[tokio::test] +async fn provider_set_then_get_round_trips_across_a_fresh_instance() { + let tmp = TempDir::new(); + let c = conv_id(1); + + // Première instance : set puis get immédiat. + { + let store = FsProviderSessionStore::new(&tmp.project_path()); + store.set(c, "claude", "a").await.unwrap(); + assert_eq!( + store.get(c, "claude").await.unwrap(), + Some("a".to_string()), + "get doit rendre la valeur posée" + ); + } + + // Le fichier existe réellement sur disque, au bon emplacement. + assert!( + tmp.providers_path(c).exists(), + "providers.json doit exister après set" + ); + + // Une nouvelle instance sur le même root relit la persistance (aucun cache mémoire). + let reborn = FsProviderSessionStore::new(&tmp.project_path()); + assert_eq!( + reborn.get(c, "claude").await.unwrap(), + Some("a".to_string()), + "persistance réelle : relue via une instance neuve" + ); +} + +// --------------------------------------------------------------------------- +// coexistence multi-providers : set claude + codex => les deux ; re-set claude +// n'efface pas codex ; sur disque les deux clés sont présentes. +// --------------------------------------------------------------------------- + +#[tokio::test] +async fn provider_set_keeps_other_providers_and_disk_holds_both_keys() { + let tmp = TempDir::new(); + let store = FsProviderSessionStore::new(&tmp.project_path()); + let c = conv_id(2); + + store.set(c, "claude", "a").await.unwrap(); + store.set(c, "codex", "b").await.unwrap(); + + // Les deux providers rendent leur valeur. + assert_eq!(store.get(c, "claude").await.unwrap(), Some("a".to_string())); + assert_eq!(store.get(c, "codex").await.unwrap(), Some("b".to_string())); + + // Re-set claude (read-modify-write) ne doit PAS effacer codex. + store.set(c, "claude", "a2").await.unwrap(); + assert_eq!( + store.get(c, "claude").await.unwrap(), + Some("a2".to_string()), + "claude mis à jour" + ); + assert_eq!( + store.get(c, "codex").await.unwrap(), + Some("b".to_string()), + "codex préservé après re-set de claude" + ); + + // Sur disque, le providers.json contient bien les deux clés (objet JSON plat). + let raw = std::fs::read_to_string(tmp.providers_path(c)).unwrap(); + let map: std::collections::HashMap = serde_json::from_str(&raw).unwrap(); + assert_eq!(map.get("claude").map(String::as_str), Some("a2")); + assert_eq!(map.get("codex").map(String::as_str), Some("b")); + assert_eq!(map.len(), 2, "exactement les deux providers, got: {raw:?}"); +} + +// --------------------------------------------------------------------------- +// absent ⇒ Ok(None) (conversation/provider jamais écrit ; dossier inexistant) +// --------------------------------------------------------------------------- + +#[tokio::test] +async fn provider_get_absent_yields_none() { + let tmp = TempDir::new(); + let store = FsProviderSessionStore::new(&tmp.project_path()); + + // Conversation jamais écrite : pas de dossier .ideai du tout. + assert_eq!( + store.get(conv_id(99), "claude").await.unwrap(), + None, + "conversation inexistante ⇒ None (pas d'erreur)" + ); + + // Conversation avec un provider posé, mais provider demandé absent ⇒ None. + let c = conv_id(98); + store.set(c, "claude", "a").await.unwrap(); + assert_eq!( + store.get(c, "codex").await.unwrap(), + None, + "clé provider absente ⇒ None" + ); +} + +// --------------------------------------------------------------------------- +// corrompu ⇒ Err(Serialization), aucun panic +// --------------------------------------------------------------------------- + +#[tokio::test] +async fn provider_corrupted_file_yields_serialization_error_no_panic() { + let tmp = TempDir::new(); + let store = FsProviderSessionStore::new(&tmp.project_path()); + + // Cas de contenus non-désérialisables en map. + let cases: [(u128, &str, &str); 3] = [ + (201, "{ ceci n'est pas du json", "JSON syntaxiquement invalide"), + (202, "[\"pas\", \"un\", \"objet\"]", "JSON valide mais pas un objet map"), + (203, "{\"claude\": 123}", "valeur non-String"), + ]; + + for (n, content, label) in cases { + let c = conv_id(n); + write_raw_providers(&tmp, c, content); + + match store.get(c, "claude").await { + Err(StoreError::Serialization(msg)) => { + assert!( + msg.starts_with("providers.json:"), + "cas «{label}» : message préfixé `providers.json:` attendu, got: {msg:?}" + ); + } + other => panic!("cas «{label}» : Err(Serialization) attendu, got: {other:?}"), + } + } +} + +// --------------------------------------------------------------------------- +// atomicité / concurrence : N set concurrents (providers distincts, même conv) +// => la map finale contient les N entrées (aucune perte) ; pas de .tmp résiduel. +// --------------------------------------------------------------------------- + +#[tokio::test] +async fn provider_concurrent_sets_keep_all_entries_no_tmp_left() { + use std::sync::Arc; + + let tmp = TempDir::new(); + let store = Arc::new(FsProviderSessionStore::new(&tmp.project_path())); + let c = conv_id(42); + + const N: usize = 12; + let mut handles = Vec::new(); + for i in 0..N { + let s = Arc::clone(&store); + handles.push(tokio::spawn(async move { + let provider = format!("provider-{i}"); + let id = format!("id-{i}"); + s.set(c, &provider, &id).await.unwrap(); + })); + } + for h in handles { + h.await.unwrap(); + } + + // Les N entrées sont toutes présentes : aucun set concurrent n'en a écrasé un autre. + for i in 0..N { + assert_eq!( + store.get(c, &format!("provider-{i}")).await.unwrap(), + Some(format!("id-{i}")), + "provider-{i} perdu sous concurrence" + ); + } + + // Le fichier sur disque contient bien les N clés. + let raw = std::fs::read_to_string(tmp.providers_path(c)).unwrap(); + let map: std::collections::HashMap = serde_json::from_str(&raw).unwrap(); + assert_eq!(map.len(), N, "map finale = N entrées, got: {raw:?}"); + + // Aucun fichier temporaire résiduel après les renames atomiques. + assert!( + !tmp.providers_tmp_path(c).exists(), + "providers.json.tmp ne doit pas subsister après les set" + ); +} + +// --------------------------------------------------------------------------- +// isolation entre deux ConversationId : pas de fuite croisée +// --------------------------------------------------------------------------- + +#[tokio::test] +async fn provider_is_isolated_per_conversation() { + let tmp = TempDir::new(); + let store = FsProviderSessionStore::new(&tmp.project_path()); + let c1 = conv_id(10); + let c2 = conv_id(20); + + store.set(c1, "claude", "c1-id").await.unwrap(); + store.set(c2, "claude", "c2-id").await.unwrap(); + + // Chacune relit la sienne. + assert_eq!(store.get(c1, "claude").await.unwrap(), Some("c1-id".to_string())); + assert_eq!(store.get(c2, "claude").await.unwrap(), Some("c2-id".to_string())); + + // Deux fichiers distincts sur disque, sans fuite de contenu. + let p1 = tmp.providers_path(c1); + let p2 = tmp.providers_path(c2); + assert_ne!(p1, p2, "deux conversations ⇒ deux providers.json distincts"); + assert!(p1.exists() && p2.exists()); + let raw2 = std::fs::read_to_string(&p2).unwrap(); + assert!(!raw2.contains("c1-id"), "aucune fuite de c1 dans le providers.json de c2"); +}