merge(live-state): intègre LS6 (rotation/rétention log.jsonl + lecture paginée)
Incrément backend cohérent et vert : archive segmentée hors chemin chaud, port ConversationArchive, lecture paginée riche + commande Tauri read_conversation_page. Tests verts domain/infrastructure/application/app-tauri. LS7 (UX React) continue sur la feature. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@ -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<TurnPageDto, ErrorDto> {
|
||||
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.
|
||||
|
||||
@ -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<MemorySlug, ErrorDto> {
|
||||
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<String>,
|
||||
/// Pagination direction (`"forward"` or `"backward"`); defaults to backward.
|
||||
#[serde(default)]
|
||||
pub direction: Option<String>,
|
||||
/// Requested page size (omit/`0` ⇒ default; clamped to `[1, 200]`).
|
||||
#[serde(default)]
|
||||
pub limit: Option<usize>,
|
||||
}
|
||||
|
||||
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<TurnSource> 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<TurnView> 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<TurnViewDto>,
|
||||
/// 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<String>,
|
||||
}
|
||||
|
||||
impl From<TurnPage> 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());
|
||||
}
|
||||
}
|
||||
|
||||
@ -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,
|
||||
|
||||
@ -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<Arc<dyn domain::ConversationArchive>> {
|
||||
Some(Arc::new(FsConversationLog::new(root)) as Arc<dyn domain::ConversationArchive>)
|
||||
}
|
||||
}
|
||||
|
||||
/// 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<ListResumableAgents>,
|
||||
/// Read-only live/busy state for the project's manifest agents.
|
||||
pub get_project_work_state: Arc<GetProjectWorkState>,
|
||||
/// Human paginated read of a conversation's full transcript (lot LS6).
|
||||
pub read_conversation_page: Arc<ReadConversationPage>,
|
||||
/// Best-effort log rotation, triggered off the hot path at thread resume/open (lot LS6).
|
||||
pub rotate_conversation_log: Arc<RotateConversationLog>,
|
||||
/// Rebinds an already-running agent's live session to a visible cell (Lot D).
|
||||
pub attach_live_agent: Arc<AttachLiveAgent>,
|
||||
/// Tears down an already-running agent's live session by agent id (Lot D).
|
||||
@ -1143,6 +1166,17 @@ impl AppState {
|
||||
as Arc<dyn application::ConversationLogProvider>,
|
||||
),
|
||||
);
|
||||
// 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<dyn application::ConversationArchiveProvider>;
|
||||
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<dyn application::HandoffProvider>,
|
||||
));
|
||||
// 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,
|
||||
|
||||
@ -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};
|
||||
|
||||
148
crates/application/src/conversation/paginate.rs
Normal file
148
crates/application/src/conversation/paginate.rs
Normal file
@ -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** (`<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<Arc<dyn ConversationArchive>>;
|
||||
}
|
||||
|
||||
/// 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<domain::input::InputSource> 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<domain::ConversationTurn> 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<TurnView>,
|
||||
/// `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<TurnId>,
|
||||
}
|
||||
|
||||
/// 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<dyn ConversationArchiveProvider>,
|
||||
}
|
||||
|
||||
impl ReadConversationPage {
|
||||
/// Construit le use case à partir du provider d'archive par root.
|
||||
#[must_use]
|
||||
pub fn new(archives: Arc<dyn ConversationArchiveProvider>) -> 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<TurnPage, AppError> {
|
||||
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,
|
||||
})
|
||||
}
|
||||
}
|
||||
95
crates/application/src/conversation/rotate.rs
Normal file
95
crates/application/src/conversation/rotate.rs
Normal file
@ -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<dyn ConversationArchiveProvider>,
|
||||
handoffs: Arc<dyn HandoffProvider>,
|
||||
thresholds: RotationThresholds,
|
||||
}
|
||||
|
||||
impl RotateConversationLog {
|
||||
/// Construit le use case à partir des providers par root (seuils par défaut).
|
||||
#[must_use]
|
||||
pub fn new(
|
||||
archives: Arc<dyn ConversationArchiveProvider>,
|
||||
handoffs: Arc<dyn HandoffProvider>,
|
||||
) -> 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),
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -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,
|
||||
|
||||
343
crates/application/tests/conversation_rotate_paginate.rs
Normal file
343
crates/application/tests/conversation_rotate_paginate.rs
Normal file
@ -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<SegmentStats>,
|
||||
turns: Mutex<Vec<ConversationTurn>>,
|
||||
rotate_calls: Mutex<Vec<TurnId>>,
|
||||
fail: Mutex<bool>,
|
||||
}
|
||||
impl FakeArchive {
|
||||
fn build(stats: SegmentStats, turns: Vec<ConversationTurn>, 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<ConversationTurn>) -> 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<TurnId> {
|
||||
self.rotate_calls.lock().unwrap().clone()
|
||||
}
|
||||
}
|
||||
#[async_trait]
|
||||
impl ConversationArchive for FakeArchive {
|
||||
async fn stats(&self, _c: ConversationId) -> Result<SegmentStats, StoreError> {
|
||||
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<TurnSlice, StoreError> {
|
||||
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<dyn ConversationArchive>);
|
||||
impl ConversationArchiveProvider for FakeArchiveProvider {
|
||||
fn conversation_archive_for(
|
||||
&self,
|
||||
_root: &ProjectPath,
|
||||
) -> Option<Arc<dyn ConversationArchive>> {
|
||||
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<Arc<dyn ConversationArchive>> {
|
||||
None
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct FakeHandoffStore(Mutex<Option<Handoff>>);
|
||||
#[async_trait]
|
||||
impl HandoffStore for FakeHandoffStore {
|
||||
async fn load(&self, _c: ConversationId) -> Result<Option<Handoff>, 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<dyn HandoffStore>);
|
||||
impl HandoffProvider for FakeHandoffProvider {
|
||||
fn handoff_store_for(&self, _root: &ProjectPath) -> Option<Arc<dyn HandoffStore>> {
|
||||
Some(Arc::clone(&self.0))
|
||||
}
|
||||
}
|
||||
|
||||
fn rotate_uc(
|
||||
archive: Arc<dyn ConversationArchive>,
|
||||
handoff: Option<Handoff>,
|
||||
) -> RotateConversationLog {
|
||||
let store = Arc::new(FakeHandoffStore(Mutex::new(handoff))) as Arc<dyn HandoffStore>;
|
||||
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<dyn ConversationArchive>,
|
||||
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<dyn ConversationArchive>, 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<dyn ConversationArchive>,
|
||||
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<dyn HandoffStore>;
|
||||
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<dyn ConversationArchive>, 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());
|
||||
}
|
||||
@ -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.<k>.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<TurnId>,
|
||||
/// 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<ConversationTurn>,
|
||||
/// `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<TurnId>,
|
||||
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<dyn ConversationArchive>` 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<SegmentStats, StoreError>;
|
||||
|
||||
/// 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<TurnSlice, StoreError>;
|
||||
}
|
||||
|
||||
#[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();
|
||||
|
||||
@ -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::{
|
||||
|
||||
@ -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)
|
||||
}
|
||||
|
||||
@ -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 : `<conversationId>/log.<k>.jsonl` (k≥1, le plus ancien=1).
|
||||
// ===========================================================================
|
||||
|
||||
/// Liste, triés croissants, les indices `k` des segments d'archive `log.<k>.jsonl`
|
||||
/// présents sur disque pour `c` (l'actif `log.jsonl` est exclu).
|
||||
fn archive_segment_indices(tmp: &TempDir, c: ConversationId) -> Vec<usize> {
|
||||
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::<usize>() {
|
||||
out.push(k);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
out.sort_unstable();
|
||||
out
|
||||
}
|
||||
|
||||
/// Écrit un segment d'archive brut `log.<k>.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<String> {
|
||||
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<TurnId> {
|
||||
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::<Vec<_>>(),
|
||||
"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::<Vec<_>>());
|
||||
|
||||
// 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::<Vec<_>>());
|
||||
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::<Vec<_>>());
|
||||
}
|
||||
|
||||
/// 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::<Vec<_>>(),
|
||||
"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<ConversationTurn> = 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::<Vec<_>>(),
|
||||
"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::<Vec<_>>());
|
||||
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::<Vec<_>>());
|
||||
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::<Vec<_>>(),
|
||||
"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::<Vec<_>>(),
|
||||
"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"
|
||||
);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user