IdeA/crates/infrastructure/src/conversation_log/mod.rs

//! [`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
//! <project_root>/.ideai/conversations/
//! └── <conversationId>/
//!     └── 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<tokio::sync::Mutex<()>>`), 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, HashSet};
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};

use async_trait::async_trait;
use tokio::io::AsyncWriteExt;

use domain::conversation::ConversationId;
use domain::conversation_log::{
    clamp_page_limit, ConversationArchive, ConversationLog, ConversationTurn, PageCursor,
    PageDirection, SegmentStats, TurnId, TurnSlice, MAX_ARCHIVE_SEGMENTS,
};
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 `<project_root>/.ideai/conversations`.
    base: PathBuf,
    /// Verrous d'écriture, un par fichier de conversation (sérialise les `append`).
    write_locks: Mutex<HashMap<ConversationId, Arc<tokio::sync::Mutex<()>>>>,
}

impl FsConversationLog {
    /// Construit l'adapter à partir du **project root**.
    ///
    /// La base `<root>/.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()),
        }
    }

    /// `<base>/<conversationId>` — le dossier d'une conversation.
    fn conversation_dir(&self, conversation: ConversationId) -> PathBuf {
        self.base.join(conversation.to_string())
    }

    /// `<base>/<conversationId>/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<tokio::sync::Mutex<()>> {
        self.write_locks
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .entry(conversation)
            .or_default()
            .clone()
    }

    /// Lit et parse tout le **segment actif** 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<Vec<ConversationTurn>, StoreError> {
        read_segment(&self.log_path(conversation)).await
    }

    // ── Rotation & pagination (lot LS6) ──────────────────────────────────────

    /// `<base>/<conversationId>/log.<k>.jsonl` — un **segment d'archive** (`k ≥ 1`,
    /// `log.1.jsonl` = le plus ancien, indices croissants = plus récents).
    fn archive_path(&self, conversation: ConversationId, index: usize) -> PathBuf {
        self.conversation_dir(conversation)
            .join(format!("log.{index}.jsonl"))
    }

    /// Liste les **indices** des segments d'archive existants, triés **croissants**
    /// (du plus ancien au plus récent). Dossier absent ⇒ `Vec` vide.
    async fn archive_indices(
        &self,
        conversation: ConversationId,
    ) -> Result<Vec<usize>, StoreError> {
        let dir = self.conversation_dir(conversation);
        let mut entries = match tokio::fs::read_dir(&dir).await {
            Ok(entries) => entries,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
            Err(e) => return Err(StoreError::Io(e.to_string())),
        };
        let mut indices = Vec::new();
        while let Some(entry) = entries
            .next_entry()
            .await
            .map_err(|e| StoreError::Io(e.to_string()))?
        {
            if let Some(name) = entry.file_name().to_str() {
                if let Some(index) = parse_archive_index(name) {
                    indices.push(index);
                }
            }
        }
        indices.sort_unstable();
        Ok(indices)
    }

    /// Lit **tous** les segments d'une conversation, ordre chronologique croissant
    /// (archives du plus ancien au plus récent, puis l'actif), avec **dédoublonnage par
    /// [`TurnId`]** (première occurrence gagne) — défense contre une tête dupliquée laissée
    /// par un crash entre l'écriture d'archive et le swap de l'actif.
    async fn read_all_segments(
        &self,
        conversation: ConversationId,
    ) -> Result<Vec<ConversationTurn>, StoreError> {
        let mut out: Vec<ConversationTurn> = Vec::new();
        let mut seen: HashSet<TurnId> = HashSet::new();
        for index in self.archive_indices(conversation).await? {
            for turn in read_segment(&self.archive_path(conversation, index)).await? {
                if seen.insert(turn.id) {
                    out.push(turn);
                }
            }
        }
        for turn in read_segment(&self.log_path(conversation)).await? {
            if seen.insert(turn.id) {
                out.push(turn);
            }
        }
        Ok(out)
    }

    /// Backstop [`MAX_ARCHIVE_SEGMENTS`] : supprime **uniquement** le ou les segments
    /// d'archive **les plus anciens** au-delà du plafond. Ne touche jamais l'actif ; les
    /// archives ne contiennent par construction que des tours `< keep_from` (`< up_to`),
    /// donc jamais un tour `≥ up_to`.
    async fn enforce_archive_backstop(
        &self,
        conversation: ConversationId,
    ) -> Result<(), StoreError> {
        let indices = self.archive_indices(conversation).await?;
        if indices.len() <= MAX_ARCHIVE_SEGMENTS {
            return Ok(());
        }
        let to_drop = indices.len() - MAX_ARCHIVE_SEGMENTS;
        for &index in indices.iter().take(to_drop) {
            // Suppression du plus ancien ; absent (course) ⇒ on ignore.
            match tokio::fs::remove_file(self.archive_path(conversation, index)).await {
                Ok(()) => {}
                Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
                Err(e) => return Err(StoreError::Io(e.to_string())),
            }
        }
        Ok(())
    }
}

/// Lit et parse un **segment** (fichier JSONL) donné, dans l'ordre d'ajout.
///
/// Fichier absent ⇒ `Vec` vide. Ligne illisible (UTF-8 invalide / JSON corrompu) ⇒
/// **ignorée** (survit à une ligne tronquée par un crash). Seule une vraie I/O remonte.
async fn read_segment(path: &Path) -> Result<Vec<ConversationTurn>, StoreError> {
    let bytes = match tokio::fs::read(path).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())),
    };
    let text = String::from_utf8_lossy(&bytes);
    let turns = text
        .lines()
        .filter(|line| !line.trim().is_empty())
        .filter_map(|line| serde_json::from_str::<ConversationTurn>(line).ok())
        .collect();
    Ok(turns)
}

/// Parse l'indice `k` d'un nom de fichier d'archive `log.<k>.jsonl` (`k` entier `≥ 1`).
/// Renvoie `None` pour l'actif (`log.jsonl`), un `.tmp`, ou tout autre nom.
fn parse_archive_index(file_name: &str) -> Option<usize> {
    let middle = file_name.strip_prefix("log.")?.strip_suffix(".jsonl")?;
    // `log.jsonl` ⇒ middle == "" ⇒ pas un segment d'archive ; un `.tmp` n'a pas ce suffixe.
    middle.parse::<usize>().ok().filter(|&k| k >= 1)
}

/// Sérialise les `turns` en JSONL puis les écrit **atomiquement** dans `final_path` via
/// un fichier temporaire `tmp_path` (`write` + `sync_all` + `rename`).
///
/// Le `rename` au sein du même dossier est atomique : un lecteur voit soit l'ancien
/// contenu complet, soit le nouveau, jamais un fichier à moitié écrit.
async fn write_segment_atomic(
    tmp_path: &Path,
    final_path: &Path,
    turns: &[ConversationTurn],
) -> Result<(), StoreError> {
    let mut buf = String::new();
    for turn in turns {
        let line =
            serde_json::to_string(turn).map_err(|e| StoreError::Serialization(e.to_string()))?;
        buf.push_str(&line);
        buf.push('\n');
    }
    let mut file = tokio::fs::File::create(tmp_path)
        .await
        .map_err(|e| StoreError::Io(e.to_string()))?;
    file.write_all(buf.as_bytes())
        .await
        .map_err(|e| StoreError::Io(e.to_string()))?;
    // Durabilité : forcer le drainage avant le rename (cf. en-tête module).
    file.sync_all()
        .await
        .map_err(|e| StoreError::Io(e.to_string()))?;
    drop(file);
    tokio::fs::rename(tmp_path, final_path)
        .await
        .map_err(|e| StoreError::Io(e.to_string()))?;
    Ok(())
}

#[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<TurnId>,
    ) -> Result<Vec<ConversationTurn>, 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<Vec<ConversationTurn>, 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())
    }
}

#[async_trait]
impl ConversationArchive for FsConversationLog {
    async fn stats(&self, conversation: ConversationId) -> Result<SegmentStats, StoreError> {
        // Bon marché : une seule lecture du segment actif ; bytes = taille du fichier,
        // turns = lignes valides. Fichier absent ⇒ `{0, 0}`.
        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(SegmentStats {
                    active_turns: 0,
                    active_bytes: 0,
                })
            }
            Err(e) => return Err(StoreError::Io(e.to_string())),
        };
        let active_bytes = bytes.len() as u64;
        let text = String::from_utf8_lossy(&bytes);
        let active_turns = text
            .lines()
            .filter(|line| !line.trim().is_empty())
            .filter(|line| serde_json::from_str::<ConversationTurn>(line).is_ok())
            .count();
        Ok(SegmentStats {
            active_turns,
            active_bytes,
        })
    }

    async fn rotate(
        &self,
        conversation: ConversationId,
        keep_from: TurnId,
    ) -> Result<(), StoreError> {
        // Verrou d'écriture de la conversation (le **même** que `append`) : tenu brièvement,
        // hors chemin chaud. Sérialise rotation et appends.
        let lock = self.write_lock(conversation);
        let _guard = lock.lock().await;

        let active = self.read_all(conversation).await?;
        // INV-LS6 : on ne garde QUE si `keep_from` est dans l'actif. Absent ⇒ déjà roté ou
        // inconnu ⇒ no-op (idempotent, jamais de perte). Position 0 ⇒ tête vide ⇒ no-op.
        let Some(idx) = active.iter().position(|t| t.id == keep_from) else {
            return Ok(());
        };
        if idx == 0 {
            return Ok(());
        }
        let head = &active[..idx];
        let queue = &active[idx..];

        // 1. Écrire la tête froide dans un **nouveau** segment d'archive (le plus récent),
        //    atomiquement. `head` ne contient que des tours `< keep_from` (INV-LS6).
        let next_index = self
            .archive_indices(conversation)
            .await?
            .last()
            .map_or(1, |max| max + 1);
        let archive_path = self.archive_path(conversation, next_index);
        let archive_tmp = with_tmp_suffix(&archive_path);
        write_segment_atomic(&archive_tmp, &archive_path, head).await?;

        // 2. Réécrire l'actif atomiquement **EN DERNIER** = `[keep_from..fin]`. Point de
        //    commit : un crash entre (1) et (2) laisse l'actif intact (tête dupliquée dans
        //    l'archive), jamais de queue perdue — la lecture dédoublonne par TurnId.
        let active_path = self.log_path(conversation);
        let active_tmp = with_tmp_suffix(&active_path);
        write_segment_atomic(&active_tmp, &active_path, queue).await?;

        // 3. Backstop : élaguer le(s) segment(s) d'archive le(s) plus ancien(s) au-delà du
        //    plafond (jamais l'actif, jamais un segment `≥ up_to`).
        self.enforce_archive_backstop(conversation).await?;
        Ok(())
    }

    async fn page(
        &self,
        conversation: ConversationId,
        cursor: PageCursor,
        limit: usize,
    ) -> Result<TurnSlice, StoreError> {
        let limit = clamp_page_limit(limit);
        // Lecture archive-aware + dédoublonnage par TurnId (frontières de segment).
        let all = self.read_all_segments(conversation).await?;
        let len = all.len();

        // Calcul de la fenêtre `[start, end)` (toujours en ordre croissant) + `has_more`
        // dans le sens de progression. Page vide pour un fil vide ou une ancre inconnue.
        let (start, end, has_more) = match cursor.anchor {
            None => match cursor.direction {
                // Début du fil (plus anciens).
                PageDirection::Forward => {
                    let end = limit.min(len);
                    (0, end, end < len)
                }
                // Fin du fil (plus récents).
                PageDirection::Backward => {
                    let start = len.saturating_sub(limit);
                    (start, len, start > 0)
                }
            },
            Some(anchor) => match all.iter().position(|t| t.id == anchor) {
                // Ancre inconnue ⇒ page vide (jamais d'erreur).
                None => (0, 0, false),
                Some(pos) => match cursor.direction {
                    // Strictement après l'ancre, jusqu'à `limit`.
                    PageDirection::Forward => {
                        let start = pos + 1;
                        let end = (start + limit).min(len);
                        (start, end, end < len)
                    }
                    // Les `limit` tours strictement avant l'ancre (les plus proches).
                    PageDirection::Backward => {
                        let end = pos;
                        let start = end.saturating_sub(limit);
                        (start, end, start > 0)
                    }
                },
            },
        };

        Ok(TurnSlice {
            turns: all[start..end].to_vec(),
            has_more,
        })
    }
}

/// Dérive le chemin temporaire `<path>.tmp` d'un fichier de segment (écriture atomique).
fn with_tmp_suffix(path: &Path) -> PathBuf {
    let mut name = path.as_os_str().to_owned();
    name.push(".tmp");
    PathBuf::from(name)
}