//! [`FsConversationLog`] — l'adapter `tokio::fs` du port [`ConversationLog`] //! (cadrage « persistance conversationnelle », lot P2). //! //! La **source de vérité durable** d'une conversation (ARCHITECTURE §19, D19-1a) //! est un log **append-only**, **un fichier JSONL par conversation (paire)** sous //! le project root : //! //! ```text //! /.ideai/conversations/ //! └── / //! └── log.jsonl # un ConversationTurn JSON par ligne, dans l'ordre d'ajout //! ``` //! //! Chaque ligne est un [`ConversationTurn`] sérialisé en JSON (`serde_json`), suivi //! d'un `\n`. Deux conversations sont **disjointes** : chacune a son propre dossier. //! //! ## Robustesse (survivre à un crash) //! //! Le log doit survivre à une **ligne tronquée** par un crash en plein milieu d'une //! écriture : à la relecture, une ligne **illisible/corrompue est silencieusement //! ignorée** (jamais de panic, jamais d'erreur dure). Un fichier **absent** est une //! conversation vide (un `Vec` vide, pas une erreur). Seules les vraies erreurs d'I/O //! (hors « absent », hors « ligne corrompue ») remontent en [`StoreError::Io`]. //! //! ## Concurrence //! //! L'écriture est **sérialisée par conversation** par un `Mutex` async dédié à //! chaque fichier (registre `paths → Arc>`), tenu le temps //! de l'`append`. Deux `append` sur des conversations **différentes** n'entrent //! jamais en contention sur la donnée (verrous distincts) ; ils ne se croisent que //! brièvement sur le registre. (P9 ajoutera un `FileGuard` plus large si un besoin //! réel d'arbitrage lecture/écriture émerge — ici on ne sur-conçoit pas.) use std::collections::HashMap; use std::path::PathBuf; use std::sync::{Arc, Mutex}; use async_trait::async_trait; use tokio::io::AsyncWriteExt; use domain::conversation::ConversationId; use domain::conversation_log::{ConversationLog, ConversationTurn, TurnId}; 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, TURN_LINE_MAX_CHARS, WINDOW}; /// Dossier `.ideai/` à la racine d'un project root. pub(crate) const IDEAI_DIR: &str = ".ideai"; /// Sous-dossier des logs de conversation dans `.ideai/`. pub(crate) const CONVERSATIONS_DIR: &str = "conversations"; /// Nom du fichier log JSONL, par conversation. const LOG_FILE: &str = "log.jsonl"; /// Adapter `tokio::fs` du log canonique append-only, un `log.jsonl` par conversation. /// /// Le **project root** est fourni au constructeur (comme [`crate::FsProjectStore`] et /// l'orchestrateur fichier reçoivent leur racine) : une instance sert toutes les /// conversations d'un même projet. Tous les chemins en dérivent. pub struct FsConversationLog { /// Racine `/.ideai/conversations`. base: PathBuf, /// Verrous d'écriture, un par fichier de conversation (sérialise les `append`). write_locks: Mutex>>>, } impl FsConversationLog { /// Construit l'adapter à partir du **project root**. /// /// La base `/.ideai/conversations` en est dérivée ; les dossiers de /// conversation sont créés paresseusement au premier `append`. #[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()) } /// `//log.jsonl` — le fichier log d'une conversation. fn log_path(&self, conversation: ConversationId) -> PathBuf { self.conversation_dir(conversation).join(LOG_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() } /// Lit et parse tout le fil de `conversation`, dans l'ordre d'ajout. /// /// Fichier absent ⇒ `Vec` vide. Une ligne illisible (UTF-8 invalide ou JSON /// corrompu) est **silencieusement ignorée** : le log survit à une ligne tronquée /// par un crash. Seule une vraie erreur d'I/O remonte. async fn read_all( &self, conversation: ConversationId, ) -> Result, StoreError> { let bytes = match tokio::fs::read(self.log_path(conversation)).await { Ok(bytes) => bytes, Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()), Err(e) => return Err(StoreError::Io(e.to_string())), }; // UTF-8 partiel (ex. fichier tronqué) : on décode en lossy plutôt que d'échouer ; // une ligne devenue invalide ne parsera simplement pas en JSON et sera ignorée. let text = String::from_utf8_lossy(&bytes); let turns = text .lines() .filter(|line| !line.trim().is_empty()) // Ligne corrompue/illisible ⇒ skip silencieux (jamais d'erreur dure). .filter_map(|line| serde_json::from_str::(line).ok()) .collect(); Ok(turns) } } #[async_trait] impl ConversationLog for FsConversationLog { async fn append( &self, conversation: ConversationId, turn: ConversationTurn, ) -> Result<(), StoreError> { // Sérialiser **avant** d'ouvrir le fichier : une erreur de sérialisation ne doit // pas laisser le fichier ouvert ni écrire de ligne partielle. let mut line = serde_json::to_string(&turn).map_err(|e| StoreError::Serialization(e.to_string()))?; line.push('\n'); // Écriture sérialisée par conversation : le verrou est tenu le temps de // create_dir_all + open(append) + write, donc deux `append` concurrents sur la // même conversation s'ordonnent (pas d'entrelacement de lignes). let lock = self.write_lock(conversation); let _guard = lock.lock().await; let dir = self.conversation_dir(conversation); tokio::fs::create_dir_all(&dir) .await .map_err(|e| StoreError::Io(e.to_string()))?; let mut file = tokio::fs::OpenOptions::new() .create(true) .append(true) .open(self.log_path(conversation)) .await .map_err(|e| StoreError::Io(e.to_string()))?; file.write_all(line.as_bytes()) .await .map_err(|e| StoreError::Io(e.to_string()))?; // Le `File` async de tokio met l'écriture en file vers une tâche blocante ; // droppé sans flush, l'écriture en vol du dernier `append` peut être jetée. // `sync_all` force le drainage **et** la durabilité crash promise par l'en-tête // du module (survivre à un crash en plein milieu d'une écriture). file.sync_all() .await .map_err(|e| StoreError::Io(e.to_string()))?; Ok(()) } async fn read( &self, conversation: ConversationId, since: Option, ) -> Result, StoreError> { let all = self.read_all(conversation).await?; let out = match since { None => all, // Curseur **exclusif** : tout ce qui suit strictement le tour `cursor`. // Curseur introuvable ⇒ rien (cohérent avec le double in-memory du port). Some(cursor) => match all.iter().position(|t| t.id == cursor) { Some(idx) => all[idx + 1..].to_vec(), None => Vec::new(), }, }; Ok(out) } async fn last( &self, conversation: ConversationId, n: usize, ) -> Result, StoreError> { if n == 0 { return Ok(Vec::new()); } let all = self.read_all(conversation).await?; let start = all.len().saturating_sub(n); Ok(all[start..].to_vec()) } }