From 40ca3e522fa5b0632bfea43e100e1582da4f605b Mon Sep 17 00:00:00 2001 From: Blomios Date: Mon, 22 Jun 2026 12:58:40 +0200 Subject: [PATCH] =?UTF-8?q?feat(domain,infra,app):=20rotation/r=C3=A9tenti?= =?UTF-8?q?on=20log.jsonl=20+=20lecture=20pagin=C3=A9e=20(LS6)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- crates/app-tauri/src/commands.rs | 86 ++- crates/app-tauri/src/dto.rs | 249 +++++++- crates/app-tauri/src/lib.rs | 1 + crates/app-tauri/src/state.rs | 52 +- crates/application/src/conversation/mod.rs | 7 + .../application/src/conversation/paginate.rs | 148 +++++ crates/application/src/conversation/rotate.rs | 95 +++ crates/application/src/lib.rs | 5 +- .../tests/conversation_rotate_paginate.rs | 343 +++++++++++ crates/domain/src/conversation_log.rs | 283 +++++++++ crates/domain/src/lib.rs | 7 +- .../src/conversation_log/mod.rs | 302 +++++++++- .../infrastructure/tests/conversation_log.rs | 560 +++++++++++++++++- 13 files changed, 2095 insertions(+), 43 deletions(-) create mode 100644 crates/application/src/conversation/paginate.rs create mode 100644 crates/application/src/conversation/rotate.rs create mode 100644 crates/application/tests/conversation_rotate_paginate.rs diff --git a/crates/app-tauri/src/commands.rs b/crates/app-tauri/src/commands.rs index cd6c9f9..41b6b79 100644 --- a/crates/app-tauri/src/commands.rs +++ b/crates/app-tauri/src/commands.rs @@ -17,12 +17,13 @@ use application::{ GitInitInput, GitLogInput, GitStagePathInput, GitStatusInput, InspectConversationInput, LaunchAgentInput, ListAgentsInput, ListLayoutsInput, ListMemoriesInput, ListResumableAgentsInput, ListSkillsInput, LiveSessions, LoadLayoutInput, McpRuntime, - MutateLayoutInput, OpenProjectInput, ReadAgentContextInput, ReadMemoryIndexInput, - ReadProjectContextInput, RecallMemoryInput, ReconcileLayoutsInput, RenameLayoutInput, - ResolveAgentPermissionsInput, ResolveMemoryLinksInput, SetActiveLayoutInput, - SnapshotRunningAgentsInput, StopLiveAgentInput, SyncAgentWithTemplateInput, - UnassignSkillFromAgentInput, UpdateAgentContextInput, UpdateAgentPermissionsInput, - UpdateMemoryInput, UpdateProjectContextInput, UpdateProjectPermissionsInput, UpdateSkillInput, + MutateLayoutInput, OpenProjectInput, ReadAgentContextInput, ReadConversationPageInput, + ReadMemoryIndexInput, ReadProjectContextInput, RecallMemoryInput, ReconcileLayoutsInput, + RenameLayoutInput, ResolveAgentPermissionsInput, ResolveMemoryLinksInput, + RotateConversationLogInput, SetActiveLayoutInput, SnapshotRunningAgentsInput, + StopLiveAgentInput, SyncAgentWithTemplateInput, UnassignSkillFromAgentInput, + UpdateAgentContextInput, UpdateAgentPermissionsInput, UpdateMemoryInput, + UpdateProjectContextInput, UpdateProjectPermissionsInput, UpdateSkillInput, }; use domain::ports::PtyHandle; @@ -43,13 +44,14 @@ use crate::dto::{ InterruptAgentRequestDto, LaunchAgentRequestDto, LayoutDto, LayoutOperationDto, ListLayoutsDto, LiveAgentListDto, MemoryDto, MemoryIndexDto, MemoryLinksDto, MemoryListDto, OpenTerminalRequestDto, ProfileDto, ProfileListDto, ProjectDto, ProjectListDto, - ProjectPermissionsDto, ProjectWorkStateDto, ReadAgentContextResponseDto, ReattachChatDto, - ReattachResultDto, RecallMemoryRequestDto, RenameLayoutRequestDto, ReplyChunk, - ResizeTerminalRequestDto, ResolveAgentPermissionsRequestDto, ResumableAgentListDto, - SaveEmbedderProfileRequestDto, SaveProfileRequestDto, SetActiveLayoutRequestDto, SkillDto, - SkillListDto, StopLiveAgentRequestDto, StopLiveAgentResponseDto, - SyncAgentWithTemplateRequestDto, SyncResultDto, TemplateDto, TemplateListDto, - TerminalClosedDto, TerminalSessionDto, UnassignSkillRequestDto, UpdateAgentContextRequestDto, + ProjectPermissionsDto, ProjectWorkStateDto, ReadAgentContextResponseDto, + ReadConversationPageRequestDto, ReattachChatDto, ReattachResultDto, RecallMemoryRequestDto, + RenameLayoutRequestDto, ReplyChunk, ResizeTerminalRequestDto, + ResolveAgentPermissionsRequestDto, ResumableAgentListDto, SaveEmbedderProfileRequestDto, + SaveProfileRequestDto, SetActiveLayoutRequestDto, SkillDto, SkillListDto, + StopLiveAgentRequestDto, StopLiveAgentResponseDto, SyncAgentWithTemplateRequestDto, + SyncResultDto, TemplateDto, TemplateListDto, TerminalClosedDto, TerminalSessionDto, + TurnPageDto, UnassignSkillRequestDto, UpdateAgentContextRequestDto, UpdateAgentPermissionsRequestDto, UpdateMemoryRequestDto, UpdateProjectContextRequestDto, UpdateProjectPermissionsRequestDto, UpdateSkillRequestDto, UpdateTemplateRequestDto, WriteTerminalRequestDto, @@ -989,6 +991,39 @@ pub async fn get_project_work_state( .map_err(ErrorDto::from) } +/// `read_conversation_page` — human, paginated read of a conversation's **full** +/// transcript (lot LS6). Archive-aware (segments + active), text never truncated. +/// +/// # Errors +/// Returns an [`ErrorDto`] (`INVALID` for a malformed project/conversation id, +/// `NOT_FOUND` if the project is unknown, `STORE` on log I/O failure). +#[tauri::command] +pub async fn read_conversation_page( + request: ReadConversationPageRequestDto, + state: State<'_, AppState>, +) -> Result { + let project = resolve_project(&request.project_id, &state).await?; + let conversation = uuid::Uuid::parse_str(&request.conversation_id) + .map(domain::ConversationId::from_uuid) + .map_err(|_| ErrorDto { + code: "INVALID".to_owned(), + message: format!("invalid conversation id: {}", request.conversation_id), + })?; + let cursor = request.cursor(); + let limit = request.limit.unwrap_or(0); + state + .read_conversation_page + .execute(ReadConversationPageInput { + project_root: project.root, + conversation, + cursor, + limit, + }) + .await + .map(TurnPageDto::from) + .map_err(ErrorDto::from) +} + /// `list_live_agents` — list every agent that currently owns a live session /// (raw PTY **or** structured/chat) and the cell hosting each, so the UI can /// disable an agent already running in another cell (the "one live session per @@ -1222,6 +1257,9 @@ pub async fn launch_agent( // agent so an auto-resume (which only carries agent/node/conversation) can recompose a // full `LaunchAgentInput`. Cloned here — `project` is moved into the launch below. let resume_project = project.clone(); + // Lot LS6 — project root captured before `project` moves, for the off-hot-path log + // rotation triggered at thread (re)open below. + let rotation_root = project.root.clone(); let output = state .launch_agent @@ -1250,6 +1288,28 @@ pub async fn launch_agent( ); } + // Lot LS6 — rotation **best-effort** du log, déclenchée à la (re)ouverture du fil et + // **hors chemin chaud** : détachée (`tokio::spawn`) pour n'ajouter aucune latence au + // launch, erreurs **avalées** (la rotation ne doit jamais casser une reprise). Un + // `append` ne déclenche JAMAIS la rotation. Skippée si la cellule ne porte pas une + // `conversation_id` UUID (rien à roter). + if let Some(conversation) = request + .conversation_id + .as_deref() + .and_then(|raw| uuid::Uuid::parse_str(raw).ok()) + .map(domain::ConversationId::from_uuid) + { + let rotate = std::sync::Arc::clone(&state.rotate_conversation_log); + tokio::spawn(async move { + let _ = rotate + .execute(RotateConversationLogInput { + project_root: rotation_root, + conversation, + }) + .await; + }); + } + let session_id = output.session.id; // Host cell of the freshly (re)launched session — the pivot the level-2 tap reports // to the service alongside the agent. diff --git a/crates/app-tauri/src/dto.rs b/crates/app-tauri/src/dto.rs index 4d0a05e..27adf9c 100644 --- a/crates/app-tauri/src/dto.rs +++ b/crates/app-tauri/src/dto.rs @@ -13,8 +13,9 @@ use application::{ ConversationTurnWorkPreview, ConversationWorkSummary, CreateProjectInput, CreateProjectOutput, GitGraphOutput, HealthInput, HealthReport, LayoutKind, ListProjectsOutput, LiveSessionKind, OpenProjectOutput, ProjectWorkState, StopLiveAgentOutput, TicketWorkSource, TicketWorkStatus, + TurnPage, TurnSource, TurnView, }; -use domain::{AgentBusyState, Project, ProjectId, TurnRole}; +use domain::{AgentBusyState, PageCursor, PageDirection, Project, ProjectId, TurnRole}; /// Request DTO for the `health` command. #[derive(Debug, Clone, Default, Deserialize)] @@ -2599,3 +2600,249 @@ pub fn parse_memory_slug(raw: &str) -> Result { message: format!("invalid memory slug: {raw}"), }) } + +// ── Conversation pagination (lot LS6) ──────────────────────────────────────── + +/// Request DTO for `read_conversation_page` — a human, paginated transcript read. +/// +/// `anchor` is a turn id to paginate around (omit to start from a thread end); +/// `direction` is `"forward"` (towards newer) or `"backward"` (towards older, +/// default — the human view opens on the latest page); `limit` is clamped by the +/// domain to `[1, 200]` (omit/`0` ⇒ default 50). +#[derive(Debug, Clone, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ReadConversationPageRequestDto { + /// The owning project's id. + pub project_id: String, + /// The conversation (pair) id to read. + pub conversation_id: String, + /// Optional anchor turn id; omit to start from a thread end. + #[serde(default)] + pub anchor: Option, + /// Pagination direction (`"forward"` or `"backward"`); defaults to backward. + #[serde(default)] + pub direction: Option, + /// Requested page size (omit/`0` ⇒ default; clamped to `[1, 200]`). + #[serde(default)] + pub limit: Option, +} + +impl ReadConversationPageRequestDto { + /// Builds the domain [`PageCursor`] from the request (default direction backward, + /// an unparseable anchor degrades to `None` = a thread-end page). + #[must_use] + pub fn cursor(&self) -> PageCursor { + let direction = match self.direction.as_deref() { + Some(d) if d.eq_ignore_ascii_case("forward") => PageDirection::Forward, + _ => PageDirection::Backward, + }; + let anchor = self + .anchor + .as_deref() + .and_then(|raw| uuid::Uuid::parse_str(raw).ok()) + .map(domain::TurnId::from_uuid); + PageCursor { anchor, direction } + } +} + +/// Origin of a turn in the human transcript view (mirrors [`TurnSource`]). +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase", tag = "kind")] +pub enum TurnSourceDto { + /// The human operator. + Human, + /// Another agent (delegation via `idea_ask_agent`). + #[serde(rename_all = "camelCase")] + Agent { + /// The originating agent id. + agent_id: String, + }, +} + +impl From for TurnSourceDto { + fn from(source: TurnSource) -> Self { + match source { + TurnSource::Human => Self::Human, + TurnSource::Agent { agent_id } => Self::Agent { + agent_id: agent_id.to_string(), + }, + } + } +} + +/// One turn in the human transcript — **full text, never truncated**. +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct TurnViewDto { + /// Stable turn id (also the pagination anchor). + pub id: String, + /// Timestamp (epoch milliseconds). + pub at_ms: u64, + /// Turn nature (prompt, response, tool activity). + pub role: TurnRole, + /// Origin (human or delegating agent). + pub source: TurnSourceDto, + /// The turn's **complete** text (not truncated). + pub text: String, + /// Character length of the text. + pub text_len: usize, +} + +impl From for TurnViewDto { + fn from(turn: TurnView) -> Self { + Self { + id: turn.id.to_string(), + at_ms: turn.at_ms, + role: turn.role, + source: turn.source.into(), + text: turn.text, + text_len: turn.text_len, + } + } +} + +/// A page of the human transcript, oldest-to-newest. +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct TurnPageDto { + /// The page's turns, oldest to newest. + pub turns: Vec, + /// Whether more turns exist beyond the page in the travel direction. + pub has_more: bool, + /// Last turn id of the page (anchor for the next request), or `None` if empty. + pub next_anchor: Option, +} + +impl From for TurnPageDto { + fn from(page: TurnPage) -> Self { + Self { + turns: page.turns.into_iter().map(TurnViewDto::from).collect(), + has_more: page.has_more, + next_anchor: page.next_anchor.map(|id| id.to_string()), + } + } +} + +#[cfg(test)] +mod ls6_pagination_tests { + //! LS6 — DTO de la lecture humaine paginée : parsing du curseur + mapping riche + //! `TurnPage → TurnPageDto` (texte complet non borné, source Human/Agent, `has_more`, + //! `next_anchor`). C'est le « round-trip d'une page » testable sans `State` Tauri. + + use super::*; + use application::{TurnPage, TurnSource, TurnView}; + + fn tid(n: u128) -> domain::TurnId { + domain::TurnId::from_uuid(uuid::Uuid::from_u128(n)) + } + + #[test] + fn request_cursor_defaults_to_backward_and_parses_anchor() { + // Pas de direction ⇒ backward (la vue humaine ouvre sur la page la plus récente). + let req = ReadConversationPageRequestDto { + project_id: "p".into(), + conversation_id: "c".into(), + anchor: None, + direction: None, + limit: None, + }; + let cur = req.cursor(); + assert_eq!(cur.direction, PageDirection::Backward); + assert_eq!(cur.anchor, None); + + // Forward (insensible à la casse) + ancre UUID valide. + let u = uuid::Uuid::from_u128(9); + let req = ReadConversationPageRequestDto { + project_id: "p".into(), + conversation_id: "c".into(), + anchor: Some(u.to_string()), + direction: Some("Forward".into()), + limit: Some(10), + }; + let cur = req.cursor(); + assert_eq!(cur.direction, PageDirection::Forward); + assert_eq!(cur.anchor, Some(domain::TurnId::from_uuid(u))); + + // Ancre non-UUID ⇒ dégrade en None (page depuis un bout du fil), jamais d'erreur. + let req = ReadConversationPageRequestDto { + project_id: "p".into(), + conversation_id: "c".into(), + anchor: Some("not-a-uuid".into()), + direction: Some("backward".into()), + limit: None, + }; + assert_eq!(req.cursor().anchor, None); + } + + #[test] + fn turn_page_dto_round_trips_full_text_and_maps_sources() { + let big = "Z".repeat(40_000); + let page = TurnPage { + turns: vec![ + TurnView { + id: tid(1), + at_ms: 1_700_000_000_000, + role: TurnRole::Prompt, + source: TurnSource::Human, + text: big.clone(), + text_len: big.chars().count(), + }, + TurnView { + id: tid(2), + at_ms: 1_700_000_000_001, + role: TurnRole::Response, + source: TurnSource::Agent { + agent_id: domain::AgentId::from_uuid(uuid::Uuid::from_u128(7)), + }, + text: "réponse".to_owned(), + text_len: 7, + }, + ], + has_more: true, + next_anchor: Some(tid(2)), + }; + + let dto = TurnPageDto::from(page); + // Texte complet préservé (non borné) + text_len. + assert_eq!(dto.turns[0].text.chars().count(), 40_000); + assert_eq!(dto.turns[0].text_len, 40_000); + // next_anchor stringifié, has_more porté. + assert!(dto.has_more); + assert_eq!( + dto.next_anchor.as_deref(), + Some(tid(2).to_string().as_str()) + ); + + // Round-trip JSON (la « page » telle qu'envoyée au front) : camelCase + source. + let json = serde_json::to_value(&dto).unwrap(); + assert_eq!(json["hasMore"], serde_json::json!(true)); + assert_eq!(json["nextAnchor"], serde_json::json!(tid(2).to_string())); + assert_eq!( + json["turns"][0]["source"]["kind"], + serde_json::json!("human") + ); + assert_eq!( + json["turns"][1]["source"]["kind"], + serde_json::json!("agent") + ); + assert_eq!( + json["turns"][1]["source"]["agentId"], + serde_json::json!(uuid::Uuid::from_u128(7).to_string()) + ); + // Le texte intégral survit au passage JSON. + assert_eq!( + json["turns"][0]["text"].as_str().unwrap().chars().count(), + 40_000 + ); + } + + #[test] + fn empty_turn_page_dto_has_no_next_anchor() { + let dto = TurnPageDto::from(TurnPage { + turns: Vec::new(), + has_more: false, + next_anchor: None, + }); + assert!(dto.turns.is_empty() && !dto.has_more && dto.next_anchor.is_none()); + } +} diff --git a/crates/app-tauri/src/lib.rs b/crates/app-tauri/src/lib.rs index d541b4f..371fcfe 100644 --- a/crates/app-tauri/src/lib.rs +++ b/crates/app-tauri/src/lib.rs @@ -164,6 +164,7 @@ pub fn run() { commands::create_agent, commands::list_agents, commands::get_project_work_state, + commands::read_conversation_page, commands::list_live_agents, commands::attach_live_agent, commands::stop_live_agent, diff --git a/crates/app-tauri/src/state.rs b/crates/app-tauri/src/state.rs index 61875ea..b753015 100644 --- a/crates/app-tauri/src/state.rs +++ b/crates/app-tauri/src/state.rs @@ -26,14 +26,15 @@ use application::{ LiveSessions, LiveStateLeanProvider, LiveStateProvider, LiveStateReadProvider, LoadLayout, McpRuntime, MoveTabToNewWindow, MutateLayout, OnnxModelView, OpenProject, OpenTerminal, OrchestratorService, PermissionProjectorRegistry, ProposeContext, ReadAgentContext, - ReadContext, ReadMemory, ReadMemoryIndex, ReadProjectContext, ReadSkill, RecallMemory, - ReconcileLayouts, RecordTurn, RecordTurnProvider, ReferenceProfiles, RenameLayout, - ResizeTerminal, ResolveAgentPermissions, ResolveMemoryLinks, SaveEmbedderProfile, SaveProfile, - SessionLimitService, SetActiveLayout, SnapshotRunningAgents, StopLiveAgent, StructuredSessions, - SuggestedThisSession, SyncAgentWithTemplate, TerminalSessions, UnassignSkillFromAgent, - UpdateAgentContext, UpdateAgentPermissions, UpdateLiveState, UpdateMemory, - UpdateProjectContext, UpdateProjectPermissions, UpdateSkill, UpdateTemplate, WriteMemory, - WriteToTerminal, AGENT_MEMORY_RECALL_BUDGET, + ReadContext, ReadConversationPage, ReadMemory, ReadMemoryIndex, ReadProjectContext, ReadSkill, + RecallMemory, ReconcileLayouts, RecordTurn, RecordTurnProvider, ReferenceProfiles, + RenameLayout, ResizeTerminal, ResolveAgentPermissions, ResolveMemoryLinks, + RotateConversationLog, SaveEmbedderProfile, SaveProfile, SessionLimitService, SetActiveLayout, + SnapshotRunningAgents, StopLiveAgent, StructuredSessions, SuggestedThisSession, + SyncAgentWithTemplate, TerminalSessions, UnassignSkillFromAgent, UpdateAgentContext, + UpdateAgentPermissions, UpdateLiveState, UpdateMemory, UpdateProjectContext, + UpdateProjectPermissions, UpdateSkill, UpdateTemplate, WriteMemory, WriteToTerminal, + AGENT_MEMORY_RECALL_BUDGET, }; use domain::ports::{ AgentContextStore, AgentRuntime, AgentSessionFactory, Clock, Embedder, EmbedderEnvInspector, @@ -93,6 +94,24 @@ impl RecordTurnProvider for AppRecordTurnProvider { } } +/// Implémente [`ConversationArchiveProvider`](application::ConversationArchiveProvider) +/// (lot LS6) en matérialisant un [`FsConversationLog`] (qui implémente aussi +/// [`domain::ConversationArchive`]) ciblant le **project root** courant. +/// +/// Même raison d'être que [`AppRecordTurnProvider`] : les use cases d'archivage/pagination +/// sont uniques pour tous les projets, alors que les logs sont **par project root**. On +/// construit donc un archive frais par appel, ciblant le bon dossier. Sans état. +struct AppConversationArchiveProvider; + +impl application::ConversationArchiveProvider for AppConversationArchiveProvider { + fn conversation_archive_for( + &self, + root: &domain::project::ProjectPath, + ) -> Option> { + Some(Arc::new(FsConversationLog::new(root)) as Arc) + } +} + /// Implémente [`HandoffProvider`](application::HandoffProvider) (lot P7) en /// matérialisant un [`FsHandoffStore`] ciblant le **project root** du lancement en /// cours. @@ -414,6 +433,10 @@ pub struct AppState { pub list_resumable_agents: Arc, /// Read-only live/busy state for the project's manifest agents. pub get_project_work_state: Arc, + /// Human paginated read of a conversation's full transcript (lot LS6). + pub read_conversation_page: Arc, + /// Best-effort log rotation, triggered off the hot path at thread resume/open (lot LS6). + pub rotate_conversation_log: Arc, /// Rebinds an already-running agent's live session to a visible cell (Lot D). pub attach_live_agent: Arc, /// Tears down an already-running agent's live session by agent id (Lot D). @@ -1143,6 +1166,17 @@ impl AppState { as Arc, ), ); + // Lot LS6 — rotation (hors chemin chaud) + lecture humaine paginée. Tous deux + // composent le provider d'archive par root ; la rotation lit aussi le handoff + // (plancher `up_to`, INV-LS6). Aucune persistance déclenchée par un `append`. + let archive_provider = Arc::new(AppConversationArchiveProvider) + as Arc; + let read_conversation_page = + Arc::new(ReadConversationPage::new(Arc::clone(&archive_provider))); + let rotate_conversation_log = Arc::new(RotateConversationLog::new( + Arc::clone(&archive_provider), + Arc::new(AppHandoffProvider) as Arc, + )); // Lot D — actions contrôlées agent-level sur les sessions vivantes : attach // (rebind de la cellule-vue, zéro spawn) et stop (kill PTY via la primitive // `CloseTerminal` existante / shutdown structuré). Aucune création de session. @@ -1343,6 +1377,8 @@ impl AppState { change_agent_profile, list_resumable_agents, get_project_work_state, + read_conversation_page, + rotate_conversation_log, attach_live_agent, stop_live_agent, inspect_conversation, diff --git a/crates/application/src/conversation/mod.rs b/crates/application/src/conversation/mod.rs index c2e1ee7..fe4fab2 100644 --- a/crates/application/src/conversation/mod.rs +++ b/crates/application/src/conversation/mod.rs @@ -13,6 +13,13 @@ //! the PTY or `app-tauri` (that is P6b). It is fully testable with in-memory fakes //! of the three ports. +mod paginate; mod record; +mod rotate; +pub use paginate::{ + ConversationArchiveProvider, ReadConversationPage, ReadConversationPageInput, TurnPage, + TurnSource, TurnView, +}; pub use record::RecordTurn; +pub use rotate::{RotateConversationLog, RotateConversationLogInput}; diff --git a/crates/application/src/conversation/paginate.rs b/crates/application/src/conversation/paginate.rs new file mode 100644 index 0000000..4328d18 --- /dev/null +++ b/crates/application/src/conversation/paginate.rs @@ -0,0 +1,148 @@ +//! [`ReadConversationPage`] — lecture **humaine** paginée d'une conversation (lot LS6). +//! +//! Lecture archive-aware (segments d'archive + actif) exposée comme un **DTO humain +//! riche** : contrairement au handoff (résumé borné) ou au read-model work-state +//! (aperçus tronqués), la page humaine porte le **texte complet, non borné** de chaque +//! tour — c'est la vue « transcript » destinée à l'utilisateur (le React arrive en LS7). +//! +//! Hors chemin chaud : composé du seul port [`ConversationArchive`] (via un provider par +//! root), jamais d'`append`/de rotation ici. + +use std::sync::Arc; + +use domain::project::ProjectPath; +use domain::{AgentId, ConversationArchive, ConversationId, PageCursor, TurnId, TurnRole}; + +use crate::error::AppError; + +/// Fournit le [`ConversationArchive`] **lié au project root** courant (lot LS6), +/// calqué sur [`crate::HandoffProvider`]/[`crate::RecordTurnProvider`]. +/// +/// Les use cases d'archivage/pagination ([`crate::RotateConversationLog`], +/// [`ReadConversationPage`]) sont **uniques** pour tous les projets, alors que les logs +/// sont **par project root** (`/.ideai/conversations/`) et l'adapter `Fs*` fixe sa +/// racine à la construction. Ce port matérialise un archive ciblant le **bon** dossier à +/// chaque appel. `None` ⇒ pas d'archivage/pagination (best-effort absent). Implémenté +/// dans `app-tauri` (seul détenteur des adapters `Fs*`). +pub trait ConversationArchiveProvider: Send + Sync { + /// Construit le [`ConversationArchive`] dont la persistance cible `root`, ou `None`. + fn conversation_archive_for(&self, root: &ProjectPath) -> Option>; +} + +/// Origine d'un tour, pour la vue humaine paginée (miroir de [`domain::input::InputSource`]). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum TurnSource { + /// L'opérateur humain. + Human, + /// Un autre agent (délégation via `idea_ask_agent`). + Agent { + /// L'agent à l'origine du tour. + agent_id: AgentId, + }, +} + +impl From for TurnSource { + fn from(source: domain::input::InputSource) -> Self { + match source { + domain::input::InputSource::Human => Self::Human, + domain::input::InputSource::Agent { agent_id } => Self::Agent { agent_id }, + } + } +} + +/// Un tour projeté pour la vue humaine (lot LS6) — **texte complet, non borné**. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct TurnView { + /// Identifiant stable du tour (sert aussi d'ancre de pagination). + pub id: TurnId, + /// Horodatage (epoch millisecondes). + pub at_ms: u64, + /// Nature du tour (invite, réponse, activité outillée). + pub role: TurnRole, + /// Origine (humain ou agent délégant). + pub source: TurnSource, + /// Le **texte intégral** du tour (jamais tronqué — c'est la vue transcript). + pub text: String, + /// Longueur (en caractères) du texte, pour l'UI. + pub text_len: usize, +} + +impl From for TurnView { + fn from(turn: domain::ConversationTurn) -> Self { + let text_len = turn.text.chars().count(); + Self { + id: turn.id, + at_ms: turn.at_ms, + role: turn.role, + source: turn.source.into(), + text: turn.text, + text_len, + } + } +} + +/// Une page de tours pour la vue humaine (lot LS6). +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct TurnPage { + /// Les tours de la page, **du plus ancien au plus récent**. + pub turns: Vec, + /// `true` s'il existe d'autres tours au-delà de la page dans le sens de progression. + pub has_more: bool, + /// L'id du **dernier** tour de la page (ancre pour la requête suivante) ; `None` si + /// la page est vide. + pub next_anchor: Option, +} + +/// Entrée de [`ReadConversationPage::execute`]. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ReadConversationPageInput { + /// Le project root dont dériver l'archive. + pub project_root: ProjectPath, + /// La conversation (paire) à lire. + pub conversation: ConversationId, + /// Le curseur (ancre + sens) de pagination. + pub cursor: PageCursor, + /// La taille de page demandée (`0` ⇒ défaut ; clampée par le domaine). + pub limit: usize, +} + +/// Use case de lecture humaine paginée (lot LS6) — consomme le seul +/// [`ConversationArchive`] (résolu par root via le provider). +pub struct ReadConversationPage { + archives: Arc, +} + +impl ReadConversationPage { + /// Construit le use case à partir du provider d'archive par root. + #[must_use] + pub fn new(archives: Arc) -> Self { + Self { archives } + } + + /// Lit une page de `conversation`, mappée vers le DTO humain riche. + /// + /// Provider d'archive absent pour ce root ⇒ page **vide** (best-effort, jamais une + /// erreur dure). Le texte de chaque tour est renvoyé **intégralement**. + /// + /// # Errors + /// [`AppError::Store`] si la lecture du log échoue. + pub async fn execute(&self, input: ReadConversationPageInput) -> Result { + let Some(archive) = self.archives.conversation_archive_for(&input.project_root) else { + return Ok(TurnPage { + turns: Vec::new(), + has_more: false, + next_anchor: None, + }); + }; + let slice = archive + .page(input.conversation, input.cursor, input.limit) + .await?; + let next_anchor = slice.turns.last().map(|t| t.id); + let turns = slice.turns.into_iter().map(TurnView::from).collect(); + Ok(TurnPage { + turns, + has_more: slice.has_more, + next_anchor, + }) + } +} diff --git a/crates/application/src/conversation/rotate.rs b/crates/application/src/conversation/rotate.rs new file mode 100644 index 0000000..9b505cb --- /dev/null +++ b/crates/application/src/conversation/rotate.rs @@ -0,0 +1,95 @@ +//! [`RotateConversationLog`] — rotation **best-effort** du log d'une conversation (lot LS6). +//! +//! Déclenchée **hors chemin chaud** (à la reprise/ouverture d'un fil, jamais par un +//! `append`), elle borne le segment actif en archivant sa tête froide. L'invariant pivot +//! (INV-LS6) est garanti par le **plancher** : `keep_from = up_to` du handoff courant ⇒ +//! aucun tour `≥ up_to` n'est jamais archivé, donc `read(since = up_to)` (fold incrémental) +//! et la reprise restent corrects. +//! +//! Best-effort strict : pas de handoff (ou `up_to` nil) ⇒ skip ; toute erreur store est +//! typée et **avalée par l'appelant** (le câblage app-tauri), jamais propagée en échec de +//! reprise. + +use std::sync::Arc; + +use domain::project::ProjectPath; +use domain::{rotation_plan, ConversationId, RotationDecision, RotationThresholds}; + +use crate::conversation::ConversationArchiveProvider; +use crate::error::AppError; +use crate::HandoffProvider; + +/// Entrée de [`RotateConversationLog::execute`]. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RotateConversationLogInput { + /// Le project root dont dériver l'archive et le handoff. + pub project_root: ProjectPath, + /// La conversation (paire) à roter. + pub conversation: ConversationId, +} + +/// Use case de rotation du log (lot LS6) — consomme un [`ConversationArchive`] et le +/// [`HandoffStore`] (pour lire `floor = up_to`), tous deux résolus par root. +/// +/// [`ConversationArchive`]: domain::ConversationArchive +/// [`HandoffStore`]: domain::HandoffStore +pub struct RotateConversationLog { + archives: Arc, + handoffs: Arc, + thresholds: RotationThresholds, +} + +impl RotateConversationLog { + /// Construit le use case à partir des providers par root (seuils par défaut). + #[must_use] + pub fn new( + archives: Arc, + handoffs: Arc, + ) -> Self { + Self { + archives, + handoffs, + thresholds: RotationThresholds::defaults(), + } + } + + /// Applique la politique de rotation à `conversation`, si nécessaire. + /// + /// Étapes : résoudre l'archive + le handoff store du root ; lire le plancher + /// `floor = up_to` (pas de handoff ⇒ `None` ⇒ skip) ; lire les `stats` bon marché du + /// segment actif ; décider via [`rotation_plan`] ; archiver + /// (`rotate(keep_from = up_to)`) seulement si la décision est `Archive`. + /// + /// # Errors + /// [`AppError::Store`] si la lecture du handoff/stats ou la rotation échoue. L'appelant + /// (câblage) **avale** cette erreur : la rotation ne doit jamais casser une reprise. + pub async fn execute(&self, input: RotateConversationLogInput) -> Result<(), AppError> { + // Sans provider d'archive/handoff câblé pour ce root ⇒ rien à faire (best-effort). + let Some(archive) = self.archives.conversation_archive_for(&input.project_root) else { + return Ok(()); + }; + let Some(handoff_store) = self.handoffs.handoff_store_for(&input.project_root) else { + return Ok(()); + }; + + // Plancher = `up_to` du handoff courant (pas de handoff ⇒ None ⇒ skip plus bas). + let floor = handoff_store + .load(input.conversation) + .await? + .map(|handoff| handoff.up_to); + + let stats = archive.stats(input.conversation).await?; + match rotation_plan( + stats.active_turns, + stats.active_bytes, + floor, + self.thresholds, + ) { + RotationDecision::Skip => Ok(()), + RotationDecision::Archive { keep_from } => archive + .rotate(input.conversation, keep_from) + .await + .map_err(AppError::from), + } + } +} diff --git a/crates/application/src/lib.rs b/crates/application/src/lib.rs index c7b702e..ad6f406 100644 --- a/crates/application/src/lib.rs +++ b/crates/application/src/lib.rs @@ -47,7 +47,10 @@ pub use agent::{ StructuredSessionDescriptor, TurnOutcome, UpdateAgentContext, UpdateAgentContextInput, AGENT_MEMORY_RECALL_BUDGET, CODEX_SUBMIT_DELAY_MS, LIVE_STATE_INJECT_MAX, RESUME_PROMPT, }; -pub use conversation::RecordTurn; +pub use conversation::{ + ConversationArchiveProvider, ReadConversationPage, ReadConversationPageInput, RecordTurn, + RotateConversationLog, RotateConversationLogInput, TurnPage, TurnSource, TurnView, +}; pub use embedder::{ CheckEmbedderSuggestion, CheckEmbedderSuggestionInput, CheckEmbedderSuggestionOutput, DeleteEmbedderProfile, DeleteEmbedderProfileInput, DescribeEmbedderEngines, DismissChoice, diff --git a/crates/application/tests/conversation_rotate_paginate.rs b/crates/application/tests/conversation_rotate_paginate.rs new file mode 100644 index 0000000..44c824d --- /dev/null +++ b/crates/application/tests/conversation_rotate_paginate.rs @@ -0,0 +1,343 @@ +//! LS6 tests — `RotateConversationLog` (rotation best-effort, plancher = `up_to`) et +//! `ReadConversationPage` (DTO humain riche, texte complet, mapping source), avec des +//! fakes in-memory des ports (aucune I/O réelle ; celle-ci est couverte côté infra). + +use std::sync::{Arc, Mutex}; + +use async_trait::async_trait; +use domain::input::InputSource; +use domain::ports::StoreError; +use domain::project::ProjectPath; +use domain::{ + AgentId, ConversationArchive, ConversationId, ConversationTurn, Handoff, HandoffStore, + PageCursor, PageDirection, SegmentStats, TurnId, TurnRole, TurnSlice, +}; + +use application::{ + ConversationArchiveProvider, HandoffProvider, ReadConversationPage, ReadConversationPageInput, + RotateConversationLog, RotateConversationLogInput, TurnSource, +}; + +// --------------------------------------------------------------------------- +// Constructeurs déterministes +// --------------------------------------------------------------------------- + +fn conv_id(n: u128) -> ConversationId { + ConversationId::from_uuid(uuid::Uuid::from_u128(n)) +} +fn turn_id(n: u128) -> TurnId { + TurnId::from_uuid(uuid::Uuid::from_u128(n)) +} +fn agent_id(n: u128) -> AgentId { + AgentId::from_uuid(uuid::Uuid::from_u128(n)) +} +fn root() -> ProjectPath { + ProjectPath::new("/home/me/proj").unwrap() +} +fn turn(id: TurnId, source: InputSource, role: TurnRole, text: &str) -> ConversationTurn { + ConversationTurn::new(id, conv_id(1), 1_700_000_000_000, source, role, text) +} + +// --------------------------------------------------------------------------- +// Fakes des ports LS6 +// --------------------------------------------------------------------------- + +/// [`ConversationArchive`] in-memory : `stats` configurables, `page` servie depuis un +/// Vec, et **enregistrement** des appels `rotate` (pour prouver `keep_from = up_to`). +/// Optionnellement, force une erreur store pour prouver la propagation typée. +struct FakeArchive { + stats: Mutex, + turns: Mutex>, + rotate_calls: Mutex>, + fail: Mutex, +} +impl FakeArchive { + fn build(stats: SegmentStats, turns: Vec, fail: bool) -> Self { + Self { + stats: Mutex::new(stats), + turns: Mutex::new(turns), + rotate_calls: Mutex::new(Vec::new()), + fail: Mutex::new(fail), + } + } + fn with_stats(turns: usize, bytes: u64) -> Self { + Self::build( + SegmentStats { + active_turns: turns, + active_bytes: bytes, + }, + Vec::new(), + false, + ) + } + fn with_page(turns: Vec) -> Self { + Self::build( + SegmentStats { + active_turns: 0, + active_bytes: 0, + }, + turns, + false, + ) + } + fn failing() -> Self { + Self::build( + SegmentStats { + active_turns: 10_000, + active_bytes: 0, + }, + Vec::new(), + true, + ) + } + fn rotate_calls(&self) -> Vec { + self.rotate_calls.lock().unwrap().clone() + } +} +#[async_trait] +impl ConversationArchive for FakeArchive { + async fn stats(&self, _c: ConversationId) -> Result { + if *self.fail.lock().unwrap() { + return Err(StoreError::Io("forced".into())); + } + Ok(*self.stats.lock().unwrap()) + } + async fn rotate(&self, _c: ConversationId, keep_from: TurnId) -> Result<(), StoreError> { + if *self.fail.lock().unwrap() { + return Err(StoreError::Io("forced".into())); + } + self.rotate_calls.lock().unwrap().push(keep_from); + Ok(()) + } + async fn page( + &self, + _c: ConversationId, + cursor: PageCursor, + limit: usize, + ) -> Result { + if *self.fail.lock().unwrap() { + return Err(StoreError::Io("forced".into())); + } + // Pagination minimale Forward-depuis-début suffisante pour les tests DTO. + let all = self.turns.lock().unwrap().clone(); + let _ = cursor; + let end = limit.min(all.len()); + Ok(TurnSlice { + turns: all[..end].to_vec(), + has_more: end < all.len(), + }) + } +} + +struct FakeArchiveProvider(Arc); +impl ConversationArchiveProvider for FakeArchiveProvider { + fn conversation_archive_for( + &self, + _root: &ProjectPath, + ) -> Option> { + Some(Arc::clone(&self.0)) + } +} + +/// Provider qui ne fournit **aucune** archive (best-effort absent). +struct NoArchiveProvider; +impl ConversationArchiveProvider for NoArchiveProvider { + fn conversation_archive_for( + &self, + _root: &ProjectPath, + ) -> Option> { + None + } +} + +#[derive(Default)] +struct FakeHandoffStore(Mutex>); +#[async_trait] +impl HandoffStore for FakeHandoffStore { + async fn load(&self, _c: ConversationId) -> Result, StoreError> { + Ok(self.0.lock().unwrap().clone()) + } + async fn save(&self, _c: ConversationId, handoff: Handoff) -> Result<(), StoreError> { + *self.0.lock().unwrap() = Some(handoff); + Ok(()) + } +} +struct FakeHandoffProvider(Arc); +impl HandoffProvider for FakeHandoffProvider { + fn handoff_store_for(&self, _root: &ProjectPath) -> Option> { + Some(Arc::clone(&self.0)) + } +} + +fn rotate_uc( + archive: Arc, + handoff: Option, +) -> RotateConversationLog { + let store = Arc::new(FakeHandoffStore(Mutex::new(handoff))) as Arc; + RotateConversationLog::new( + Arc::new(FakeArchiveProvider(archive)), + Arc::new(FakeHandoffProvider(store)), + ) +} + +fn input() -> RotateConversationLogInput { + RotateConversationLogInput { + project_root: root(), + conversation: conv_id(1), + } +} + +// --------------------------------------------------------------------------- +// RotateConversationLog +// --------------------------------------------------------------------------- + +/// Plancher lu du handoff + stats au-dessus du seuil ⇒ `rotate(keep_from = up_to)`. +#[tokio::test] +async fn rotate_uses_handoff_up_to_as_floor_and_calls_rotate() { + // Stats au-dessus du seuil de tours (défaut 500). + let archive = Arc::new(FakeArchive::with_stats(10_000, 0)); + let handoff = Handoff::new("résumé", turn_id(42), Some("but".to_owned())); + let uc = rotate_uc( + archive.clone() as Arc, + Some(handoff), + ); + + uc.execute(input()).await.expect("rotate ok"); + + assert_eq!( + archive.rotate_calls(), + vec![turn_id(42)], + "rotate appelé avec keep_from = up_to du handoff" + ); +} + +/// Pas de handoff ⇒ `floor = None` ⇒ skip : `rotate` jamais appelé même au-dessus du seuil. +#[tokio::test] +async fn rotate_skips_without_handoff() { + let archive = Arc::new(FakeArchive::with_stats(10_000, u64::MAX)); + let uc = rotate_uc(archive.clone() as Arc, None); + + uc.execute(input()).await.expect("skip ok"); + + assert!( + archive.rotate_calls().is_empty(), + "sans handoff ⇒ aucun rotate" + ); +} + +/// Handoff présent mais stats SOUS les deux seuils ⇒ skip (pas de rotation). +#[tokio::test] +async fn rotate_skips_under_thresholds() { + let archive = Arc::new(FakeArchive::with_stats(10, 1_000)); + let handoff = Handoff::new("r", turn_id(5), None); + let uc = rotate_uc( + archive.clone() as Arc, + Some(handoff), + ); + + uc.execute(input()).await.expect("skip ok"); + + assert!(archive.rotate_calls().is_empty(), "sous les seuils ⇒ skip"); +} + +/// Provider d'archive absent ⇒ no-op silencieux (best-effort), jamais une erreur. +#[tokio::test] +async fn rotate_without_archive_provider_is_noop() { + let store = Arc::new(FakeHandoffStore(Mutex::new(Some(Handoff::new( + "r", + turn_id(5), + None, + ))))) as Arc; + let uc = RotateConversationLog::new( + Arc::new(NoArchiveProvider), + Arc::new(FakeHandoffProvider(store)), + ); + uc.execute(input()).await.expect("no provider ⇒ Ok(())"); +} + +/// Une erreur store est **propagée typée** (`AppError::Store`) par le use case ; le +/// **swallowing best-effort** est la responsabilité du câblage (cf. test app-tauri). +#[tokio::test] +async fn rotate_propagates_typed_store_error() { + let archive = Arc::new(FakeArchive::failing()); + let handoff = Handoff::new("r", turn_id(9), None); + let uc = rotate_uc(archive as Arc, Some(handoff)); + + let err = uc.execute(input()).await.expect_err("store error propagée"); + assert!( + matches!(err, application::AppError::Store(_)), + "erreur store typée, got {err:?}" + ); +} + +// --------------------------------------------------------------------------- +// ReadConversationPage — DTO humain riche +// --------------------------------------------------------------------------- + +/// La page mappe un DTO riche : texte **complet** (non borné), `text_len` correct, +/// mapping source Human/Agent, `has_more` et `next_anchor` (= dernier id de la page). +#[tokio::test] +async fn read_page_maps_rich_dto_with_full_text_and_sources() { + let big = "Z".repeat(30_000); + let turns = vec![ + turn(turn_id(1), InputSource::Human, TurnRole::Prompt, &big), + turn( + turn_id(2), + InputSource::agent(agent_id(7)), + TurnRole::Response, + "réponse agent", + ), + turn(turn_id(3), InputSource::Human, TurnRole::Prompt, "suite"), + ]; + let archive = Arc::new(FakeArchive::with_page(turns)); + let uc = ReadConversationPage::new(Arc::new(FakeArchiveProvider(archive))); + + // limit 2 ⇒ page [1,2], has_more (il reste le tour 3). + let page = uc + .execute(ReadConversationPageInput { + project_root: root(), + conversation: conv_id(1), + cursor: PageCursor { + anchor: None, + direction: PageDirection::Forward, + }, + limit: 2, + }) + .await + .expect("page ok"); + + assert_eq!(page.turns.len(), 2); + // Texte complet préservé + text_len. + assert_eq!(page.turns[0].text.chars().count(), 30_000); + assert_eq!(page.turns[0].text_len, 30_000); + // Mapping source : Human puis Agent{7}. + assert_eq!(page.turns[0].source, TurnSource::Human); + assert_eq!( + page.turns[1].source, + TurnSource::Agent { + agent_id: agent_id(7) + } + ); + // has_more + next_anchor = dernier id de la page (tour 2). + assert!(page.has_more, "il reste le tour 3"); + assert_eq!(page.next_anchor, Some(turn_id(2))); +} + +/// Provider d'archive absent ⇒ page **vide** (best-effort), `next_anchor = None`. +#[tokio::test] +async fn read_page_without_provider_is_empty() { + let uc = ReadConversationPage::new(Arc::new(NoArchiveProvider)); + let page = uc + .execute(ReadConversationPageInput { + project_root: root(), + conversation: conv_id(1), + cursor: PageCursor { + anchor: None, + direction: PageDirection::Forward, + }, + limit: 50, + }) + .await + .expect("page ok"); + assert!(page.turns.is_empty() && !page.has_more && page.next_anchor.is_none()); +} diff --git a/crates/domain/src/conversation_log.rs b/crates/domain/src/conversation_log.rs index 8124f85..d1bf88d 100644 --- a/crates/domain/src/conversation_log.rs +++ b/crates/domain/src/conversation_log.rs @@ -451,6 +451,208 @@ pub trait ProviderSessionStore: Send + Sync { ) -> Result<(), StoreError>; } +// ===================================================================================== +// Rotation & pagination du log (lot LS6) +// ===================================================================================== +// +// Le `log.jsonl` actif est **borné** par rotation : au-delà d'un seuil (tours OU octets) +// la **tête froide** est déplacée dans des **segments d'archive** (`log..jsonl`), pour +// que le segment actif (et donc tout `read`/`last`/append) reste petit. La lecture +// **humaine** paginée traverse archives + actif. +// +// ## Invariant pivot (INV-LS6) +// +// La rotation ne déplace/élague **JAMAIS** un tour d'id ≥ `up_to` du handoff courant : le +// tour `up_to` et tous ses postérieurs restent dans le segment **actif**. Sans handoff (ou +// `up_to` = sentinelle nil de LS5) ⇒ **aucune** rotation. C'est ce qui garantit que +// [`ConversationLog::read`] avec `since = up_to` (fold incrémental) et la reprise restent +// corrects après rotation. + +/// Seuil de rotation en **nombre de tours** du segment actif : au-delà, la tête froide +/// est archivée (lot LS6). +pub const ROTATE_AFTER_TURNS: usize = 500; + +/// Seuil de rotation en **octets** du segment actif (1 MiB) : au-delà, la tête froide est +/// archivée (lot LS6). L'un OU l'autre seuil déclenche. +pub const ROTATE_AFTER_BYTES: u64 = 1024 * 1024; + +/// Nombre maximum de **segments d'archive** conservés par conversation (lot LS6). Au-delà, +/// le segment d'archive **le plus ancien** est supprimé (backstop) ; jamais l'actif, +/// jamais un segment contenant un tour ≥ `up_to` (les archives n'en contiennent jamais). +pub const MAX_ARCHIVE_SEGMENTS: usize = 20; + +/// Taille de page par défaut de la lecture humaine paginée (lot LS6), appliquée quand +/// l'appelant ne précise pas de limite (`limit == 0`). +pub const PAGE_DEFAULT_LIMIT: usize = 50; + +/// Borne **dure** de la taille d'une page (lot LS6) : toute limite est clampée à au plus +/// cette valeur (anti-dump, la lecture humaine reste paginée). +pub const PAGE_MAX_LIMIT: usize = 200; + +/// Seuils de rotation passés à [`rotation_plan`] (lot LS6). Regroupés pour garder la +/// fonction pure simple et testable avec des seuils ad hoc. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct RotationThresholds { + /// Seuil en nombre de tours du segment actif (cf. [`ROTATE_AFTER_TURNS`]). + pub max_turns: usize, + /// Seuil en octets du segment actif (cf. [`ROTATE_AFTER_BYTES`]). + pub max_bytes: u64, +} + +impl RotationThresholds { + /// Les seuils par défaut du projet ([`ROTATE_AFTER_TURNS`] / [`ROTATE_AFTER_BYTES`]). + #[must_use] + pub const fn defaults() -> Self { + Self { + max_turns: ROTATE_AFTER_TURNS, + max_bytes: ROTATE_AFTER_BYTES, + } + } +} + +/// Statistiques **bon marché** du segment actif d'une conversation (lot LS6), servant de +/// **déclencheur** de rotation sans relire/élaguer quoi que ce soit. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct SegmentStats { + /// Nombre de tours (lignes valides) du segment actif. + pub active_turns: usize, + /// Taille en octets du segment actif. + pub active_bytes: u64, +} + +/// Sens de pagination d'une page humaine (lot LS6). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum PageDirection { + /// Vers les tours **plus récents** (postérieurs à l'ancre). + Forward, + /// Vers les tours **plus anciens** (antérieurs à l'ancre). + Backward, +} + +/// Curseur d'une lecture paginée (lot LS6). +/// +/// `anchor = None` ⇒ on part d'un **bout** du fil : `Forward` ⇒ le tout début (plus +/// anciens), `Backward` ⇒ la toute fin (plus récents). `anchor = Some(id)` ⇒ on pagine +/// **strictement** autour de ce tour, dans la `direction` donnée. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct PageCursor { + /// Le tour pivot, ou `None` pour partir d'un bout du fil. + pub anchor: Option, + /// Le sens de progression. + pub direction: PageDirection, +} + +/// Une tranche paginée de tours (lot LS6), **toujours en ordre chronologique croissant**. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct TurnSlice { + /// Les tours de la page, du plus ancien au plus récent. + pub turns: Vec, + /// `true` s'il existe d'autres tours **dans le sens de progression** au-delà de la page. + pub has_more: bool, +} + +/// La décision **pure** de rotation calculée par [`rotation_plan`] (lot LS6). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RotationDecision { + /// Ne rien faire (sous les seuils, ou pas de plancher `up_to`). + Skip, + /// Archiver la tête froide, en **gardant** tout à partir de `keep_from` dans l'actif. + Archive { + /// Le plancher : premier tour conservé dans l'actif (= `up_to` du handoff). Tous + /// les tours **strictement antérieurs** sont archivés ; `keep_from` et ses + /// postérieurs restent actifs (INV-LS6). + keep_from: TurnId, + }, +} + +/// Décide **purement** s'il faut archiver la tête froide du segment actif (lot LS6). +/// +/// - `floor = None` ⇒ [`RotationDecision::Skip`] : sans plancher `up_to` (pas de handoff), +/// on ne rote jamais (INV-LS6). De même si `floor` est la **sentinelle nil** de LS5 +/// (`up_to` d'un handoff vide) : rien à garder en aval de façon fiable ⇒ `Skip`. +/// - Sous **les deux** seuils ⇒ `Skip`. +/// - Au-dessus de **l'un** des seuils ⇒ [`RotationDecision::Archive`] avec +/// `keep_from = floor` : `keep_from` n'est **jamais** antérieur à `up_to` (il **est** +/// `up_to`), donc l'invariant pivot est respecté par construction. +/// +/// Pure et idempotente (mêmes entrées ⇒ même sortie ; aucune I/O). +#[must_use] +pub fn rotation_plan( + active_turns: usize, + active_bytes: u64, + floor: Option, + thresholds: RotationThresholds, +) -> RotationDecision { + // Pas de plancher, ou plancher = sentinelle nil (handoff vide LS5) ⇒ jamais de rotation. + let Some(keep_from) = floor else { + return RotationDecision::Skip; + }; + if keep_from.as_uuid().is_nil() { + return RotationDecision::Skip; + } + let over_turns = active_turns > thresholds.max_turns; + let over_bytes = active_bytes > thresholds.max_bytes; + if over_turns || over_bytes { + RotationDecision::Archive { keep_from } + } else { + RotationDecision::Skip + } +} + +/// Clampe une taille de page demandée dans `[1, PAGE_MAX_LIMIT]` (lot LS6), en appliquant +/// [`PAGE_DEFAULT_LIMIT`] quand l'appelant ne précise rien (`limit == 0`). Pure/testable. +#[must_use] +pub fn clamp_page_limit(limit: usize) -> usize { + let limit = if limit == 0 { + PAGE_DEFAULT_LIMIT + } else { + limit + }; + limit.clamp(1, PAGE_MAX_LIMIT) +} + +/// Le port d'**archivage & pagination** du log d'une conversation (port driven, lot LS6). +/// +/// Complète [`ConversationLog`] (append/read/last, **inchangé**) sans le modifier : la +/// rotation est une opération **hors chemin chaud** (jamais déclenchée par un `append`) et +/// la pagination est une lecture **humaine** archive-aware. Implémenté par le **même** +/// adapter FS que [`ConversationLog`] (il détient déjà les fichiers). +/// +/// `#[async_trait]` + erreur [`StoreError`] comme les ports voisins ; injecté en +/// `Arc` au composition root, gardé object-safe. +#[async_trait::async_trait] +pub trait ConversationArchive: Send + Sync { + /// Statistiques bon marché du segment actif (déclencheur de rotation). + /// + /// # Errors + /// [`StoreError`] en cas d'échec de lecture. + async fn stats(&self, conversation: ConversationId) -> Result; + + /// Archive la tête froide en **gardant** tout à partir de `keep_from` dans l'actif + /// (INV-LS6). Idempotente ; ne perd jamais de tour ; n'archive jamais un tour + /// `≥ keep_from`. No-op si `keep_from` est absent de l'actif (déjà roté/inconnu). + /// + /// # Errors + /// [`StoreError`] en cas d'échec d'I/O. + async fn rotate( + &self, + conversation: ConversationId, + keep_from: TurnId, + ) -> Result<(), StoreError>; + + /// Lit une **page humaine** (archive-aware), toujours en ordre chronologique croissant, + /// selon le `cursor` et la `limit` (clampée `[1, PAGE_MAX_LIMIT]`). + /// + /// # Errors + /// [`StoreError`] en cas d'échec de lecture. + async fn page( + &self, + conversation: ConversationId, + cursor: PageCursor, + limit: usize, + ) -> Result; +} + #[cfg(test)] mod tests { use super::*; @@ -707,6 +909,87 @@ mod tests { assert!(once.summary_md.chars().count() <= 200); } + // --------------------------------------------------------------------- + // rotation_plan + clamp_page_limit (lot LS6) — fonctions pures + // --------------------------------------------------------------------- + + #[test] + fn rotation_plan_skips_without_floor() { + // Largement au-dessus des seuils mais pas de plancher ⇒ jamais de rotation. + let d = rotation_plan(10_000, 10 << 20, None, RotationThresholds::defaults()); + assert_eq!(d, RotationDecision::Skip); + } + + #[test] + fn rotation_plan_skips_with_nil_sentinel_floor() { + let nil = TurnId::from_uuid(uuid::Uuid::nil()); + let d = rotation_plan(10_000, 10 << 20, Some(nil), RotationThresholds::defaults()); + assert_eq!( + d, + RotationDecision::Skip, + "sentinelle nil ⇒ pas de rotation" + ); + } + + #[test] + fn rotation_plan_skips_under_both_thresholds() { + let floor = turn_id(7); + let d = rotation_plan(10, 1_000, Some(floor), RotationThresholds::defaults()); + assert_eq!(d, RotationDecision::Skip); + } + + #[test] + fn rotation_plan_archives_over_either_threshold_keeping_floor() { + let floor = turn_id(7); + // Au-dessus du seuil de tours seulement. + assert_eq!( + rotation_plan( + ROTATE_AFTER_TURNS + 1, + 0, + Some(floor), + RotationThresholds::defaults() + ), + RotationDecision::Archive { keep_from: floor } + ); + // Au-dessus du seuil d'octets seulement. + assert_eq!( + rotation_plan( + 0, + ROTATE_AFTER_BYTES + 1, + Some(floor), + RotationThresholds::defaults() + ), + RotationDecision::Archive { keep_from: floor } + ); + } + + #[test] + fn rotation_plan_is_idempotent() { + let floor = turn_id(7); + let inputs = (ROTATE_AFTER_TURNS + 5, ROTATE_AFTER_BYTES + 5); + let once = rotation_plan( + inputs.0, + inputs.1, + Some(floor), + RotationThresholds::defaults(), + ); + let twice = rotation_plan( + inputs.0, + inputs.1, + Some(floor), + RotationThresholds::defaults(), + ); + assert_eq!(once, twice); + } + + #[test] + fn clamp_page_limit_applies_default_and_bounds() { + assert_eq!(clamp_page_limit(0), PAGE_DEFAULT_LIMIT, "0 ⇒ défaut"); + assert_eq!(clamp_page_limit(1), 1); + assert_eq!(clamp_page_limit(50), 50); + assert_eq!(clamp_page_limit(10_000), PAGE_MAX_LIMIT, "clampé au max"); + } + #[tokio::test] async fn append_then_read_all_preserves_insertion_order() { let log = InMemoryConversationLog::default(); diff --git a/crates/domain/src/lib.rs b/crates/domain/src/lib.rs index 81da3fa..8865fe0 100644 --- a/crates/domain/src/lib.rs +++ b/crates/domain/src/lib.rs @@ -103,8 +103,11 @@ pub use readiness::{ReadinessPolicy, ReadinessSignal}; pub use session_limit::{plan_resume, RateLimitSource, ResumePlan, SessionLimit}; pub use conversation_log::{ - bound_handoff_summary, ConversationLog, ConversationTurn, Handoff, HandoffStore, - HandoffSummarizer, ProviderSessionStore, TurnId, TurnRole, HANDOFF_SUMMARY_MAX_CHARS, + bound_handoff_summary, clamp_page_limit, rotation_plan, ConversationArchive, ConversationLog, + ConversationTurn, Handoff, HandoffStore, HandoffSummarizer, PageCursor, PageDirection, + ProviderSessionStore, RotationDecision, RotationThresholds, SegmentStats, TurnId, TurnRole, + TurnSlice, HANDOFF_SUMMARY_MAX_CHARS, MAX_ARCHIVE_SEGMENTS, PAGE_DEFAULT_LIMIT, PAGE_MAX_LIMIT, + ROTATE_AFTER_BYTES, ROTATE_AFTER_TURNS, }; pub use fileguard::{ diff --git a/crates/infrastructure/src/conversation_log/mod.rs b/crates/infrastructure/src/conversation_log/mod.rs index a469a57..85a2e10 100644 --- a/crates/infrastructure/src/conversation_log/mod.rs +++ b/crates/infrastructure/src/conversation_log/mod.rs @@ -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, 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) ────────────────────────────────────── + + /// `//log..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, 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::(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, StoreError> { + let mut out: Vec = Vec::new(); + let mut seen: HashSet = 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, 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::(line).ok()) + .collect(); + Ok(turns) +} + +/// Parse l'indice `k` d'un nom de fichier d'archive `log..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 { + 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::().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 { + // 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::(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 { + 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 `.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) +} diff --git a/crates/infrastructure/tests/conversation_log.rs b/crates/infrastructure/tests/conversation_log.rs index 1cfe9fd..11b76f7 100644 --- a/crates/infrastructure/tests/conversation_log.rs +++ b/crates/infrastructure/tests/conversation_log.rs @@ -18,8 +18,9 @@ use std::path::PathBuf; use domain::conversation::ConversationId; use domain::conversation_log::{ - bound_handoff_summary, ConversationLog, ConversationTurn, Handoff, HandoffStore, - HandoffSummarizer, ProviderSessionStore, TurnId, TurnRole, HANDOFF_SUMMARY_MAX_CHARS, + bound_handoff_summary, ConversationArchive, ConversationLog, ConversationTurn, Handoff, + HandoffStore, HandoffSummarizer, PageCursor, PageDirection, ProviderSessionStore, TurnId, + TurnRole, HANDOFF_SUMMARY_MAX_CHARS, MAX_ARCHIVE_SEGMENTS, ROTATE_AFTER_BYTES, }; use domain::input::InputSource; use domain::ports::StoreError; @@ -1292,3 +1293,558 @@ async fn provider_is_isolated_per_conversation() { "aucune fuite de c1 dans le providers.json de c2" ); } + +// =========================================================================== +// LS6 — rotation (ConversationArchive) + pagination humaine, I/O réelle. +// Réutilise le scaffolding L2 (TempDir absolu, constructeurs déterministes). +// Convention d'archive : `/log..jsonl` (k≥1, le plus ancien=1). +// =========================================================================== + +/// Liste, triés croissants, les indices `k` des segments d'archive `log..jsonl` +/// présents sur disque pour `c` (l'actif `log.jsonl` est exclu). +fn archive_segment_indices(tmp: &TempDir, c: ConversationId) -> Vec { + let dir = tmp.conversation_dir(c); + let mut out = Vec::new(); + if let Ok(rd) = std::fs::read_dir(&dir) { + for entry in rd.flatten() { + if let Some(name) = entry.file_name().to_str() { + if let Some(mid) = name + .strip_prefix("log.") + .and_then(|s| s.strip_suffix(".jsonl")) + { + if let Ok(k) = mid.parse::() { + out.push(k); + } + } + } + } + } + out.sort_unstable(); + out +} + +/// Écrit un segment d'archive brut `log..jsonl` (simulation de crash / dédoublonnage). +fn write_raw_archive_segment( + tmp: &TempDir, + c: ConversationId, + k: usize, + turns: &[ConversationTurn], +) { + let dir = tmp.conversation_dir(c); + std::fs::create_dir_all(&dir).unwrap(); + let mut buf = String::new(); + for t in turns { + buf.push_str(&serde_json::to_string(t).unwrap()); + buf.push('\n'); + } + std::fs::write(dir.join(format!("log.{k}.jsonl")), buf).unwrap(); +} + +/// Tous les `.tmp` restant dans le dossier de la conversation (doivent être absents). +fn leftover_tmp_files(tmp: &TempDir, c: ConversationId) -> Vec { + let dir = tmp.conversation_dir(c); + let mut out = Vec::new(); + if let Ok(rd) = std::fs::read_dir(&dir) { + for entry in rd.flatten() { + if let Some(name) = entry.file_name().to_str() { + if name.ends_with(".tmp") { + out.push(name.to_owned()); + } + } + } + } + out +} + +fn ids(turns: &[ConversationTurn]) -> Vec { + turns.iter().map(|t| t.id).collect() +} + +/// Enregistre `n` tours (id 1..=n) dans l'actif et renvoie le log. +async fn seed_turns(log: &FsConversationLog, c: ConversationId, n: u128) { + for i in 1..=n { + let role = if i % 2 == 1 { + TurnRole::Prompt + } else { + TurnRole::Response + }; + log.append(c, turn(c, turn_id(i), role, &format!("texte du tour {i}"))) + .await + .unwrap(); + } +} + +// --- INV-LS6 : le test pivot ------------------------------------------------- + +/// INV-LS6 (PIVOT) : après `rotate(keep_from=up_to)`, le curseur `up_to` reste +/// trouvable dans l'ACTIF, donc `ConversationLog::read(since=up_to)` renvoie toujours +/// le bon incrément, et `last(n)` fonctionne. C'est l'invariant qui garantit que la +/// rotation ne casse jamais le fold incrémental du handoff. +#[tokio::test] +async fn inv_ls6_read_since_up_to_survives_rotation() { + let tmp = TempDir::new(); + let log = FsConversationLog::new(&tmp.project_path()); + let c = conv_id(1); + seed_turns(&log, c, 10).await; + + // up_to = curseur du handoff (tour 5). On rote en gardant tout depuis 5. + let up_to = turn_id(5); + log.rotate(c, up_to).await.expect("rotate ok"); + + // Le curseur est dans l'actif ⇒ l'incrément strictement postérieur est intact. + let inc = log.read(c, Some(up_to)).await.unwrap(); + assert_eq!( + texts(&inc), + (6..=10) + .map(|i| format!("texte du tour {i}")) + .collect::>(), + "read(since=up_to) renvoie l'incrément 6..=10 après rotation" + ); + // last(n) lit bien la fin (archive-unaware mais l'actif contient la queue). + let last3 = log.last(c, 3).await.unwrap(); + assert_eq!(ids(&last3), vec![turn_id(8), turn_id(9), turn_id(10)]); +} + +// --- archive : déplacement de tête, aucune perte, idempotence ---------------- + +/// `rotate(keep_from=5)` déplace la tête froide (1..4) dans un segment, garde +/// `[5..10]` dans l'actif, sans perdre aucun tour (∪ actif+archives = tous). +#[tokio::test] +async fn rotate_moves_cold_head_to_a_segment_without_loss() { + let tmp = TempDir::new(); + let log = FsConversationLog::new(&tmp.project_path()); + let c = conv_id(1); + seed_turns(&log, c, 10).await; + + log.rotate(c, turn_id(5)).await.expect("rotate ok"); + + // Un seul segment d'archive créé (le plus ancien = index 1). + assert_eq!(archive_segment_indices(&tmp, c), vec![1]); + + // L'actif = [5..10] (keep_from inclus en tête). + let active = log.read(c, None).await.unwrap(); + assert_eq!(ids(&active), (5..=10).map(turn_id).collect::>()); + + // Aucun tour perdu : la page archive-aware réunit tout, 1..10 en ordre croissant. + let page = log + .page( + c, + PageCursor { + anchor: None, + direction: PageDirection::Forward, + }, + 200, + ) + .await + .unwrap(); + assert_eq!(ids(&page.turns), (1..=10).map(turn_id).collect::>()); + assert!(!page.has_more, "10 tours ≤ limit ⇒ pas d'autre page"); +} + +/// `rotate` est idempotente : un 2e appel avec le même `keep_from` (désormais en tête +/// de l'actif) est un no-op — aucun nouveau segment, actif inchangé. +#[tokio::test] +async fn rotate_is_idempotent() { + let tmp = TempDir::new(); + let log = FsConversationLog::new(&tmp.project_path()); + let c = conv_id(1); + seed_turns(&log, c, 10).await; + + log.rotate(c, turn_id(5)).await.expect("rotate 1"); + let after1 = log.read(c, None).await.unwrap(); + let segs1 = archive_segment_indices(&tmp, c); + + log.rotate(c, turn_id(5)).await.expect("rotate 2 (no-op)"); + let after2 = log.read(c, None).await.unwrap(); + let segs2 = archive_segment_indices(&tmp, c); + + assert_eq!(ids(&after1), ids(&after2), "actif inchangé au 2e rotate"); + assert_eq!(segs1, segs2, "aucun nouveau segment au 2e rotate"); + assert_eq!(segs2, vec![1], "toujours exactement un segment"); +} + +/// `keep_from` absent de l'actif (déjà roté / inconnu) ⇒ no-op, jamais de perte. +#[tokio::test] +async fn rotate_with_unknown_keep_from_is_noop() { + let tmp = TempDir::new(); + let log = FsConversationLog::new(&tmp.project_path()); + let c = conv_id(1); + seed_turns(&log, c, 5).await; + + log.rotate(c, turn_id(999)).await.expect("rotate noop"); + assert!(archive_segment_indices(&tmp, c).is_empty(), "aucun segment"); + let active = log.read(c, None).await.unwrap(); + assert_eq!(ids(&active), (1..=5).map(turn_id).collect::>()); +} + +/// Réécriture de l'actif **atomique** : après rotation, aucun `.tmp` ne traîne et la +/// queue active est intégralement présente (jamais perdue). +#[tokio::test] +async fn rotate_leaves_no_tmp_and_preserves_active_queue() { + let tmp = TempDir::new(); + let log = FsConversationLog::new(&tmp.project_path()); + let c = conv_id(1); + seed_turns(&log, c, 8).await; + + log.rotate(c, turn_id(4)).await.expect("rotate ok"); + + assert!( + leftover_tmp_files(&tmp, c).is_empty(), + "aucun .tmp résiduel après écriture atomique : {:?}", + leftover_tmp_files(&tmp, c) + ); + let active = log.read(c, None).await.unwrap(); + assert_eq!( + ids(&active), + (4..=8).map(turn_id).collect::>(), + "queue active [4..8] intégralement préservée" + ); +} + +/// Tête dupliquée (simulation de crash : archive écrite mais actif non swappé) ⇒ la +/// lecture paginée **dédoublonne par TurnId** (chaque tour une seule fois). +#[tokio::test] +async fn page_dedups_duplicated_head_after_crash() { + let tmp = TempDir::new(); + let log = FsConversationLog::new(&tmp.project_path()); + let c = conv_id(1); + seed_turns(&log, c, 6).await; // actif = [1..6] + + // Crash simulé : un segment d'archive contenant la tête (1..3) est écrit, mais + // l'actif n'a PAS été tronqué (il contient toujours 1..6). + let active_all = log.read(c, None).await.unwrap(); + let head: Vec = active_all[..3].to_vec(); + write_raw_archive_segment(&tmp, c, 1, &head); + + // La page réunit archive(1..3) + actif(1..6) MAIS dédoublonne ⇒ 1..6, pas de doublon. + let page = log + .page( + c, + PageCursor { + anchor: None, + direction: PageDirection::Forward, + }, + 200, + ) + .await + .unwrap(); + assert_eq!( + ids(&page.turns), + (1..=6).map(turn_id).collect::>(), + "dédoublonnage par TurnId : chaque tour une seule fois" + ); +} + +/// `append` ne déclenche **jamais** la rotation : grossir l'actif au-delà du seuil +/// d'octets sans appeler `rotate` ne crée aucun segment d'archive. +#[tokio::test] +async fn append_never_triggers_rotation_even_over_threshold() { + let tmp = TempDir::new(); + let log = FsConversationLog::new(&tmp.project_path()); + let c = conv_id(1); + + // Dépasser le seuil d'octets (1 MiB) : ~16 tours de 80 000 caractères. + let big = "m".repeat(80_000); + for i in 1..=16u128 { + log.append(c, turn(c, turn_id(i), TurnRole::Response, &big)) + .await + .unwrap(); + } + let stats = log.stats(c).await.unwrap(); + assert!( + stats.active_bytes > ROTATE_AFTER_BYTES, + "pré-condition : actif au-dessus du seuil d'octets ({})", + stats.active_bytes + ); + // Pourtant : aucun segment d'archive (la rotation est hors-chemin, jamais sur append). + assert!( + archive_segment_indices(&tmp, c).is_empty(), + "append ne crée jamais d'archive : {:?}", + archive_segment_indices(&tmp, c) + ); +} + +// --- pagination humaine ------------------------------------------------------ + +/// Page vide pour une conversation vide. +#[tokio::test] +async fn page_of_empty_conversation_is_empty() { + let tmp = TempDir::new(); + let log = FsConversationLog::new(&tmp.project_path()); + let c = conv_id(1); + let page = log + .page( + c, + PageCursor { + anchor: None, + direction: PageDirection::Forward, + }, + 50, + ) + .await + .unwrap(); + assert!(page.turns.is_empty() && !page.has_more); +} + +/// Forward depuis le début, Backward depuis la fin, ancré dans les deux sens ; +/// `has_more` correct ; ordre **chronologique croissant** dans chaque page. +#[tokio::test] +async fn page_forward_backward_anchored_and_has_more() { + let tmp = TempDir::new(); + let log = FsConversationLog::new(&tmp.project_path()); + let c = conv_id(1); + seed_turns(&log, c, 10).await; + + // Forward depuis le début, limit 4 ⇒ [1..4], has_more. + let p = log + .page( + c, + PageCursor { + anchor: None, + direction: PageDirection::Forward, + }, + 4, + ) + .await + .unwrap(); + assert_eq!(ids(&p.turns), (1..=4).map(turn_id).collect::>()); + assert!(p.has_more); + + // Backward depuis la fin, limit 4 ⇒ [7..10] (ordre croissant !), has_more. + let p = log + .page( + c, + PageCursor { + anchor: None, + direction: PageDirection::Backward, + }, + 4, + ) + .await + .unwrap(); + assert_eq!(ids(&p.turns), (7..=10).map(turn_id).collect::>()); + assert!(p.has_more); + + // Ancré Forward après le tour 4, limit 3 ⇒ [5,6,7], has_more. + let p = log + .page( + c, + PageCursor { + anchor: Some(turn_id(4)), + direction: PageDirection::Forward, + }, + 3, + ) + .await + .unwrap(); + assert_eq!(ids(&p.turns), vec![turn_id(5), turn_id(6), turn_id(7)]); + assert!(p.has_more); + + // Ancré Backward avant le tour 4, limit 10 ⇒ [1,2,3], plus rien avant. + let p = log + .page( + c, + PageCursor { + anchor: Some(turn_id(4)), + direction: PageDirection::Backward, + }, + 10, + ) + .await + .unwrap(); + assert_eq!(ids(&p.turns), vec![turn_id(1), turn_id(2), turn_id(3)]); + assert!(!p.has_more, "rien avant le tour 1"); + + // Dernière page Forward (fin atteinte) ⇒ has_more = false. + let p = log + .page( + c, + PageCursor { + anchor: Some(turn_id(7)), + direction: PageDirection::Forward, + }, + 50, + ) + .await + .unwrap(); + assert_eq!(ids(&p.turns), vec![turn_id(8), turn_id(9), turn_id(10)]); + assert!(!p.has_more); +} + +/// La pagination est **archive-aware** : après rotation, une page peut chevaucher +/// plusieurs segments (archive + actif) et reste en ordre croissant continu. +#[tokio::test] +async fn page_spans_multiple_segments_after_rotation() { + let tmp = TempDir::new(); + let log = FsConversationLog::new(&tmp.project_path()); + let c = conv_id(1); + seed_turns(&log, c, 12).await; + log.rotate(c, turn_id(7)).await.expect("rotate"); // archive [1..6], actif [7..12] + + // Une page ancrée Forward autour de la frontière archive/actif renvoie un span continu. + let p = log + .page( + c, + PageCursor { + anchor: Some(turn_id(5)), + direction: PageDirection::Forward, + }, + 4, + ) + .await + .unwrap(); + // Tours 6 (archive) puis 7,8,9 (actif) : continuité par-delà la frontière de segment. + assert_eq!( + ids(&p.turns), + vec![turn_id(6), turn_id(7), turn_id(8), turn_id(9)] + ); + assert!(p.has_more); +} + +/// Le **texte complet** d'un tour est préservé (non borné) à la lecture paginée — la +/// borne LS5 ne s'applique qu'au handoff, jamais au log canonique. +#[tokio::test] +async fn page_preserves_full_unbounded_turn_text() { + let tmp = TempDir::new(); + let log = FsConversationLog::new(&tmp.project_path()); + let c = conv_id(1); + let big = "Z".repeat(50_000); + log.append(c, turn(c, turn_id(1), TurnRole::Response, &big)) + .await + .unwrap(); + + let p = log + .page( + c, + PageCursor { + anchor: None, + direction: PageDirection::Forward, + }, + 50, + ) + .await + .unwrap(); + assert_eq!(p.turns.len(), 1); + assert_eq!( + p.turns[0].text.chars().count(), + 50_000, + "texte complet préservé (non borné) à la pagination" + ); +} + +// --- backstop MAX_ARCHIVE_SEGMENTS ------------------------------------------ + +/// Au-delà de [`MAX_ARCHIVE_SEGMENTS`] segments, le(s) plus ANCIEN(s) sont supprimés ; +/// l'actif n'est jamais touché et aucun tour `≥ keep_from` n'est archivé/perdu. +#[tokio::test] +async fn rotation_enforces_max_archive_segments_dropping_oldest() { + let tmp = TempDir::new(); + let log = FsConversationLog::new(&tmp.project_path()); + let c = conv_id(1); + let total = 40u128; + seed_turns(&log, c, total).await; + + // Roter `MAX_ARCHIVE_SEGMENTS + 3` fois, en avançant keep_from d'un tour à chaque fois. + // Chaque rotation déplace une tête d'un tour ⇒ crée un segment. + let rotations = (MAX_ARCHIVE_SEGMENTS + 3) as u128; + for k in 2..=(1 + rotations) { + log.rotate(c, turn_id(k)).await.expect("rotate ok"); + } + + // Backstop : jamais plus de MAX_ARCHIVE_SEGMENTS segments sur disque. + let segs = archive_segment_indices(&tmp, c); + assert_eq!( + segs.len(), + MAX_ARCHIVE_SEGMENTS, + "plafonné à MAX_ARCHIVE_SEGMENTS, obtenu {segs:?}" + ); + + // L'actif n'a pas été touché par le backstop : il commence au dernier keep_from. + let last_keep = 1 + rotations; // = turn_id du dernier keep_from + let active = log.read(c, None).await.unwrap(); + assert_eq!( + active.first().map(|t| t.id), + Some(turn_id(last_keep)), + "l'actif commence à keep_from (jamais archivé/supprimé)" + ); + assert_eq!( + active.last().map(|t| t.id), + Some(turn_id(total)), + "la fin de l'actif est intacte" + ); + // read(since=last_keep) — le curseur survit (INV-LS6) même après backstop. + let inc = log.read(c, Some(turn_id(last_keep))).await.unwrap(); + assert_eq!( + inc.first().map(|t| t.id), + Some(turn_id(last_keep + 1)), + "l'incrément après le dernier curseur reste correct" + ); +} + +// --- cohérence fold-après-rotation (le must-have) ---------------------------- + +/// COHÉRENCE (must-have) : beaucoup de tours → `rotate(keep_from = up_to)` → encore des +/// tours → un `fold` ultérieur **étend correctement** le handoff. On lit l'actif depuis +/// `up_to` (INV-LS6 : toujours trouvable après rotation), on replie l'incrément, et +/// `up_to` AVANCE. Prouve que la rotation ne casse jamais le fold incrémental réel. +#[tokio::test] +async fn fold_after_rotation_extends_handoff_correctly() { + let tmp = TempDir::new(); + let log = FsConversationLog::new(&tmp.project_path()); + let summarizer = HeuristicHandoffSummarizer::new(); + let c = conv_id(1); + + // 1. Tours 1..7 enregistrés PUIS repliés ⇒ handoff.up_to = 7. + seed_turns(&log, c, 7).await; + let first_batch = log.read(c, None).await.unwrap(); + let handoff = summarizer.fold(None, &first_batch).await; + let handoff = bound_handoff_summary(handoff, HANDOFF_SUMMARY_MAX_CHARS); + assert_eq!(handoff.up_to, turn_id(7), "curseur = dernier tour replié"); + + // 2. Plus de tours arrivent (8..15), non encore repliés, écrits dans l'actif. + for i in 8..=15u128 { + let role = if i % 2 == 1 { + TurnRole::Prompt + } else { + TurnRole::Response + }; + log.append(c, turn(c, turn_id(i), role, &format!("texte du tour {i}"))) + .await + .unwrap(); + } + + // 3. Rotation au plancher = up_to du handoff (7) : tête froide 1..6 archivée. + log.rotate(c, handoff.up_to).await.expect("rotate ok"); + // L'actif commence bien au curseur (INV-LS6) ⇒ le curseur reste trouvable. + assert_eq!( + ids(&log.read(c, None).await.unwrap()), + (7..=15).map(turn_id).collect::>(), + "actif = [up_to..fin] après rotation" + ); + + // 4. Fold ultérieur : on lit l'incrément depuis up_to (8..15) DANS L'ACTIF, on replie. + let increment = log.read(c, Some(handoff.up_to)).await.unwrap(); + assert_eq!( + ids(&increment), + (8..=15).map(turn_id).collect::>(), + "read(since=up_to) renvoie l'incrément 8..15 malgré la rotation" + ); + let advanced = summarizer.fold(Some(handoff.clone()), &increment).await; + let advanced = bound_handoff_summary(advanced, HANDOFF_SUMMARY_MAX_CHARS); + + // up_to a AVANCÉ jusqu'au dernier tour, et le résumé intègre les tours récents. + assert_eq!( + advanced.up_to, + turn_id(15), + "le curseur avance après le fold" + ); + assert!( + advanced.summary_md.contains("texte du tour 15"), + "le fold intègre le tour le plus récent : {}", + advanced.summary_md + ); + assert_ne!( + advanced.up_to, handoff.up_to, + "la rotation ne fige pas le curseur du handoff" + ); +}