feat(domain,infra,app): rotation/rétention log.jsonl + lecture paginée (LS6)
Backend uniquement (UI React repoussée à LS7) : - domain : port ConversationArchive + structs SegmentStats/PageCursor/PageDirection/ TurnSlice/RotationDecision/RotationThresholds, fn pures rotation_plan/clamp_page_limit + consts ; re-exports lib. - infrastructure : impl ConversationArchive pour FsConversationLog (stats/rotate/page + helpers), archive segmentée hors chemin chaud. - application : ReadConversationPage + DTO + ConversationArchiveProvider (conversation/paginate), RotateConversationLog (conversation/rotate), exports mod/lib. - app-tauri : AppConversationArchiveProvider + wiring (state), rotation détachée dans launch_agent + commande read_conversation_page (commands), DTOs (dto), commande enregistrée (generate_handler!). - tests (QA, verts) : conversation_log, conversation_rotate_paginate (nouveau), dto (module test). Pivots INV-LS6 et cohérence fold-après-rotation verts. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@ -31,15 +31,18 @@
|
||||
//! 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::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::{ConversationLog, ConversationTurn, TurnId};
|
||||
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;
|
||||
|
||||
@ -108,7 +111,7 @@ impl FsConversationLog {
|
||||
.clone()
|
||||
}
|
||||
|
||||
/// Lit et parse tout le fil de `conversation`, dans l'ordre d'ajout.
|
||||
/// 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
|
||||
@ -117,22 +120,155 @@ impl FsConversationLog {
|
||||
&self,
|
||||
conversation: ConversationId,
|
||||
) -> Result<Vec<ConversationTurn>, StoreError> {
|
||||
let bytes = match tokio::fs::read(self.log_path(conversation)).await {
|
||||
Ok(bytes) => bytes,
|
||||
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())),
|
||||
};
|
||||
// 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::<ConversationTurn>(line).ok())
|
||||
.collect();
|
||||
Ok(turns)
|
||||
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]
|
||||
@ -209,3 +345,137 @@ impl ConversationLog for FsConversationLog {
|
||||
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)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user