diff --git a/crates/application/src/conversation/mod.rs b/crates/application/src/conversation/mod.rs new file mode 100644 index 0000000..c2e1ee7 --- /dev/null +++ b/crates/application/src/conversation/mod.rs @@ -0,0 +1,18 @@ +//! Conversation persistence use cases (cadrage « persistance conversationnelle », +//! ARCHITECTURE §19, lot P6). +//! +//! Where [`crate::agent::lifecycle`] owns the *runtime* of an agent (spawn, PTY, +//! sessions), this module owns the **durable conversational memory** of a +//! conversation (paire) : the canonical append-only log and its incremental +//! handoff. The single use case here, [`RecordTurn`], materialises the +//! **end-of-turn checkpoint** (D19-5) by talking **only** to the three driven/driving +//! ports of §19 ([`domain::ConversationLog`], [`domain::HandoffStore`], +//! [`domain::HandoffSummarizer`]) — never to a concrete adapter. +//! +//! This is the **isolated logic** of P6 (lot P6a) : no wiring to the orchestrator, +//! the PTY or `app-tauri` (that is P6b). It is fully testable with in-memory fakes +//! of the three ports. + +mod record; + +pub use record::RecordTurn; diff --git a/crates/application/src/conversation/record.rs b/crates/application/src/conversation/record.rs new file mode 100644 index 0000000..b6d5109 --- /dev/null +++ b/crates/application/src/conversation/record.rs @@ -0,0 +1,100 @@ +//! [`RecordTurn`] — the end-of-turn checkpoint use case (ARCHITECTURE §19, lot P6a). + +use std::sync::Arc; + +use domain::{ConversationId, ConversationLog, ConversationTurn, HandoffStore, HandoffSummarizer}; + +use crate::error::AppError; + +/// Records one conversation turn at a checkpoint: appends it to the canonical +/// log, then advances the incremental handoff (ARCHITECTURE §19, décision D19-5). +/// +/// **Single Responsibility**: materialise *one* end-of-turn checkpoint. It is the +/// only place that couples the append-only log (source of truth, [`ConversationLog`]) +/// with the cumulative resume summary ([`HandoffStore`]) via the incremental +/// [`HandoffSummarizer`]. It consumes **exactly** those three ports (Interface +/// Segregation), injected as `Arc` at the composition root. +/// +/// ## Incremental by contract (no full re-read) +/// +/// Each call folds **only the new turn** into the previous handoff +/// (`fold(prev, &[turn])`), never re-reading the whole thread — exactly the seam +/// [`HandoffSummarizer`] was designed for (§19, lot P4). The resulting handoff's +/// [`Handoff::up_to`](domain::Handoff::up_to) is the recorded turn's id (guaranteed +/// by the summarizer, P4). +/// +/// ## Best-effort fold, honest `Result` +/// +/// The fold itself cannot fail (it returns no `Result`, P4 — a future LLM +/// summarizer falls back rather than erroring), so the only failures are the +/// store operations (`append` / `load` / `save`), which propagate as [`AppError`]. +/// This use case stays **honest** and surfaces them; the live wiring (P6b) decides +/// whether to swallow a handoff failure rather than block the turn. +/// +/// ## No debounce here (deferred) +/// +/// One `append` + one `save` per recorded turn. The debounce optimisation (coalesce +/// several rapid turns into a single handoff recompute) is intentionally **deferred** +/// — it belongs to the live wiring (P6b), not to this isolated logic. +pub struct RecordTurn { + log: Arc, + handoffs: Arc, + summarizer: Arc, +} + +impl RecordTurn { + /// Builds the use case from its three injected ports. + #[must_use] + pub fn new( + log: Arc, + handoffs: Arc, + summarizer: Arc, + ) -> Self { + Self { + log, + handoffs, + summarizer, + } + } + + /// Records `turn` in `conversation` at an end-of-turn checkpoint. + /// + /// Steps (ordering is contractual): + /// 1. **append** the turn to the canonical log (source of truth) ; + /// 2. **load** the previous handoff (`None` on a first checkpoint) ; + /// 3. **fold** `prev` with the single new turn (incremental, best-effort) ; + /// 4. **save** the advanced handoff. + /// + /// The append happens **first** so the durable source of truth is never behind + /// the derived handoff. The turn is cloned for the append because the fold (step + /// 3) borrows it as the incremental slice. + /// + /// # Errors + /// [`AppError::Store`] (or [`AppError::NotFound`] for a missing store item) when + /// the canonical log append, the handoff load, or the handoff save fails — every + /// [`domain::ports::StoreError`] is mapped through the existing `From` impl. The + /// fold never fails (P4). + pub async fn record( + &self, + conversation: ConversationId, + turn: ConversationTurn, + ) -> Result<(), AppError> { + // 1. Append to the canonical append-only log (source of truth, D19-1a). + self.log.append(conversation, turn.clone()).await?; + + // 2. Load the previous resume point (absence is never an error, P3). + let prev = self.handoffs.load(conversation).await?; + + // 3. Fold the previous handoff with *only* the new turn (incremental, P4): + // no full re-read of the log; best-effort, so no `Result` to propagate. + let handoff = self + .summarizer + .fold(prev, std::slice::from_ref(&turn)) + .await; + + // 4. Persist the advanced handoff (overwrites the previous resume point). + self.handoffs.save(conversation, handoff).await?; + + Ok(()) + } +} diff --git a/crates/application/src/lib.rs b/crates/application/src/lib.rs index f6cd7bf..042ce85 100644 --- a/crates/application/src/lib.rs +++ b/crates/application/src/lib.rs @@ -12,6 +12,7 @@ #![warn(missing_docs)] pub mod agent; +pub mod conversation; pub mod embedder; pub mod error; pub mod git; @@ -40,6 +41,7 @@ pub use agent::{ SaveProfileInput, SaveProfileOutput, StructuredSessionDescriptor, UpdateAgentContext, UpdateAgentContextInput, send_blocking, AGENT_MEMORY_RECALL_BUDGET, }; +pub use conversation::RecordTurn; pub use embedder::{ CheckEmbedderSuggestion, CheckEmbedderSuggestionInput, CheckEmbedderSuggestionOutput, DeleteEmbedderProfile, DeleteEmbedderProfileInput, DescribeEmbedderEngines, DismissChoice, diff --git a/crates/application/tests/conversation_record.rs b/crates/application/tests/conversation_record.rs new file mode 100644 index 0000000..c0642e8 --- /dev/null +++ b/crates/application/tests/conversation_record.rs @@ -0,0 +1,360 @@ +//! P6a tests for the [`RecordTurn`] use case (ARCHITECTURE §19) with in-memory +//! port fakes (no real store/FS). +//! +//! Coverage: +//! - single `record`: log contains the turn; handoff `up_to == turn.id`; summary +//! contains the turn text; +//! - incremental over two `record` calls: log order [t1, t2]; `up_to == t2.id`; +//! summary contains both texts; **proof of incrementality** — the summarizer +//! saw `prev = None` then `prev = Some(_)`, and `new_turns.len() == 1` on *both* +//! calls (never 2 → the log is never fully re-read); +//! - error propagation: a store whose `append` / `save` / `load` returns a +//! non-`NotFound` `StoreError` makes `record` return `AppError::Store(_)`; +//! a failing `append` performs **no** `save` (no side effect after failure); +//! - isolation by `ConversationId`. + +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; + +use async_trait::async_trait; +use domain::ports::StoreError; +use domain::{ + ConversationId, ConversationLog, ConversationTurn, Handoff, HandoffStore, HandoffSummarizer, + InputSource, TurnId, TurnRole, +}; + +use application::{AppError, RecordTurn}; + +// --------------------------------------------------------------------------- +// Deterministic constructors +// --------------------------------------------------------------------------- + +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 turn(conv: ConversationId, id: TurnId, text: &str) -> ConversationTurn { + ConversationTurn::new(id, conv, 1_000, InputSource::Human, TurnRole::Prompt, text) +} + +// --------------------------------------------------------------------------- +// Fakes — Vec/Map backed, Mutex never held across an `.await`. +// --------------------------------------------------------------------------- + +#[derive(Default)] +struct InMemoryConversationLog { + threads: Mutex>>, + /// When set, `append` returns this error instead of appending. + fail_append: Mutex>, +} + +impl InMemoryConversationLog { + fn failing_append(err: StoreError) -> Self { + Self { + threads: Mutex::new(HashMap::new()), + fail_append: Mutex::new(Some(err)), + } + } + + fn thread(&self, conv: ConversationId) -> Vec { + self.threads.lock().unwrap().get(&conv).cloned().unwrap_or_default() + } +} + +#[async_trait] +impl ConversationLog for InMemoryConversationLog { + async fn append( + &self, + conversation: ConversationId, + turn: ConversationTurn, + ) -> Result<(), StoreError> { + if let Some(err) = self.fail_append.lock().unwrap().clone() { + return Err(err); + } + self.threads + .lock() + .unwrap() + .entry(conversation) + .or_default() + .push(turn); + Ok(()) + } + + async fn read( + &self, + conversation: ConversationId, + since: Option, + ) -> Result, StoreError> { + let all = self.thread(conversation); + let out = match since { + None => all, + Some(cursor) => { + let pos = all.iter().position(|t| t.id == cursor); + match pos { + Some(i) => all[i + 1..].to_vec(), + None => all, + } + } + }; + Ok(out) + } + + async fn last( + &self, + conversation: ConversationId, + n: usize, + ) -> Result, StoreError> { + let all = self.thread(conversation); + let start = all.len().saturating_sub(n); + Ok(all[start..].to_vec()) + } +} + +#[derive(Default)] +struct InMemoryHandoffStore { + store: Mutex>, + fail_load: Mutex>, + fail_save: Mutex>, + /// Number of completed `save` calls (used to prove no save-after-append-fail). + saves: Mutex, +} + +impl InMemoryHandoffStore { + fn failing_load(err: StoreError) -> Self { + Self { + fail_load: Mutex::new(Some(err)), + ..Self::default() + } + } + + fn failing_save(err: StoreError) -> Self { + Self { + fail_save: Mutex::new(Some(err)), + ..Self::default() + } + } + + fn loaded(&self, conv: ConversationId) -> Option { + self.store.lock().unwrap().get(&conv).cloned() + } + + fn save_count(&self) -> usize { + *self.saves.lock().unwrap() + } +} + +#[async_trait] +impl HandoffStore for InMemoryHandoffStore { + async fn load(&self, conversation: ConversationId) -> Result, StoreError> { + if let Some(err) = self.fail_load.lock().unwrap().clone() { + return Err(err); + } + Ok(self.store.lock().unwrap().get(&conversation).cloned()) + } + + async fn save( + &self, + conversation: ConversationId, + handoff: Handoff, + ) -> Result<(), StoreError> { + if let Some(err) = self.fail_save.lock().unwrap().clone() { + return Err(err); + } + self.store.lock().unwrap().insert(conversation, handoff); + *self.saves.lock().unwrap() += 1; + Ok(()) + } +} + +/// Records, on every `fold` call, the arguments it received so the test can prove +/// incrementality. Deterministic: `up_to = last new turn id`, and +/// `summary_md = prev.summary_md + each new turn's text`. +#[derive(Default)] +struct RecordingSummarizer { + /// One entry per `fold` call: `(prev_was_some, new_turns_len)`. + calls: Mutex>, +} + +impl RecordingSummarizer { + fn calls(&self) -> Vec<(bool, usize)> { + self.calls.lock().unwrap().clone() + } +} + +#[async_trait] +impl HandoffSummarizer for RecordingSummarizer { + async fn fold(&self, prev: Option, new_turns: &[ConversationTurn]) -> Handoff { + self.calls + .lock() + .unwrap() + .push((prev.is_some(), new_turns.len())); + + let mut summary = prev.map(|h| h.summary_md).unwrap_or_default(); + for t in new_turns { + summary.push_str(&t.text); + } + let up_to = new_turns + .last() + .map(|t| t.id) + .expect("fold must receive at least one new turn in these tests"); + Handoff::new(summary, up_to, None) + } +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +#[tokio::test] +async fn single_record_appends_turn_and_advances_handoff() { + let log = Arc::new(InMemoryConversationLog::default()); + let handoffs = Arc::new(InMemoryHandoffStore::default()); + let summarizer = Arc::new(RecordingSummarizer::default()); + let uc = RecordTurn::new(log.clone(), handoffs.clone(), summarizer.clone()); + + let conv = conv_id(1); + let t1 = turn(conv, turn_id(10), "hello world"); + + uc.record(conv, t1.clone()).await.expect("record should succeed"); + + // Log contains the turn. + assert_eq!(log.thread(conv), vec![t1.clone()]); + + // Handoff advanced: up_to == turn id, summary contains the turn text. + let h = handoffs.loaded(conv).expect("handoff should be saved"); + assert_eq!(h.up_to, t1.id); + assert!(h.summary_md.contains("hello world"), "summary={:?}", h.summary_md); + + // First fold: prev = None, exactly one new turn. + assert_eq!(summarizer.calls(), vec![(false, 1)]); + assert_eq!(handoffs.save_count(), 1); +} + +#[tokio::test] +async fn two_records_fold_incrementally_without_full_reread() { + let log = Arc::new(InMemoryConversationLog::default()); + let handoffs = Arc::new(InMemoryHandoffStore::default()); + let summarizer = Arc::new(RecordingSummarizer::default()); + let uc = RecordTurn::new(log.clone(), handoffs.clone(), summarizer.clone()); + + let conv = conv_id(1); + let t1 = turn(conv, turn_id(10), "first-text"); + let t2 = turn(conv, turn_id(20), "second-text"); + + uc.record(conv, t1.clone()).await.expect("record t1"); + uc.record(conv, t2.clone()).await.expect("record t2"); + + // Log: [t1, t2] in order. + assert_eq!(log.thread(conv), vec![t1.clone(), t2.clone()]); + + // Handoff up_to == t2.id; summary contains both texts. + let h = handoffs.loaded(conv).expect("handoff saved"); + assert_eq!(h.up_to, t2.id); + assert!(h.summary_md.contains("first-text"), "summary={:?}", h.summary_md); + assert!(h.summary_md.contains("second-text"), "summary={:?}", h.summary_md); + + // Proof of incrementality: prev None then Some, and len == 1 on BOTH calls + // (never 2 → the whole log is never re-read into the fold). + assert_eq!(summarizer.calls(), vec![(false, 1), (true, 1)]); + assert_eq!(handoffs.save_count(), 2); +} + +#[tokio::test] +async fn append_error_propagates_as_store_and_does_not_save() { + let log = Arc::new(InMemoryConversationLog::failing_append(StoreError::Io( + "disk full".to_owned(), + ))); + let handoffs = Arc::new(InMemoryHandoffStore::default()); + let summarizer = Arc::new(RecordingSummarizer::default()); + let uc = RecordTurn::new(log.clone(), handoffs.clone(), summarizer.clone()); + + let conv = conv_id(1); + let err = uc + .record(conv, turn(conv, turn_id(10), "x")) + .await + .expect_err("append failure must surface"); + + assert!(matches!(err, AppError::Store(_)), "got {err:?}"); + + // No side effect after a failed append: handoff never saved, fold never called. + assert_eq!(handoffs.save_count(), 0); + assert!(handoffs.loaded(conv).is_none()); + assert!(summarizer.calls().is_empty()); +} + +#[tokio::test] +async fn load_error_propagates_as_store() { + let log = Arc::new(InMemoryConversationLog::default()); + let handoffs = Arc::new(InMemoryHandoffStore::failing_load(StoreError::Io( + "load boom".to_owned(), + ))); + let summarizer = Arc::new(RecordingSummarizer::default()); + let uc = RecordTurn::new(log.clone(), handoffs.clone(), summarizer.clone()); + + let conv = conv_id(1); + let err = uc + .record(conv, turn(conv, turn_id(10), "x")) + .await + .expect_err("load failure must surface"); + + assert!(matches!(err, AppError::Store(_)), "got {err:?}"); + // Append happened (source of truth first), but no save and no fold. + assert_eq!(log.thread(conv).len(), 1); + assert_eq!(handoffs.save_count(), 0); + assert!(summarizer.calls().is_empty()); +} + +#[tokio::test] +async fn save_error_propagates_as_store() { + let log = Arc::new(InMemoryConversationLog::default()); + let handoffs = Arc::new(InMemoryHandoffStore::failing_save(StoreError::Serialization( + "save boom".to_owned(), + ))); + let summarizer = Arc::new(RecordingSummarizer::default()); + let uc = RecordTurn::new(log.clone(), handoffs.clone(), summarizer.clone()); + + let conv = conv_id(1); + let err = uc + .record(conv, turn(conv, turn_id(10), "x")) + .await + .expect_err("save failure must surface"); + + assert!(matches!(err, AppError::Store(_)), "got {err:?}"); + // Append + fold happened, but the save itself failed → nothing stored. + assert_eq!(log.thread(conv).len(), 1); + assert_eq!(handoffs.save_count(), 0); + assert!(handoffs.loaded(conv).is_none()); + assert_eq!(summarizer.calls(), vec![(false, 1)]); +} + +#[tokio::test] +async fn records_are_isolated_per_conversation() { + let log = Arc::new(InMemoryConversationLog::default()); + let handoffs = Arc::new(InMemoryHandoffStore::default()); + let summarizer = Arc::new(RecordingSummarizer::default()); + let uc = RecordTurn::new(log.clone(), handoffs.clone(), summarizer.clone()); + + let a = conv_id(1); + let b = conv_id(2); + let ta = turn(a, turn_id(10), "alpha"); + let tb = turn(b, turn_id(20), "beta"); + + uc.record(a, ta.clone()).await.expect("record a"); + uc.record(b, tb.clone()).await.expect("record b"); + + // Logs are kept apart. + assert_eq!(log.thread(a), vec![ta.clone()]); + assert_eq!(log.thread(b), vec![tb.clone()]); + + // Handoffs are kept apart: each covers only its own conversation's turn. + let ha = handoffs.loaded(a).expect("handoff a"); + let hb = handoffs.loaded(b).expect("handoff b"); + assert_eq!(ha.up_to, ta.id); + assert!(ha.summary_md.contains("alpha") && !ha.summary_md.contains("beta")); + assert_eq!(hb.up_to, tb.id); + assert!(hb.summary_md.contains("beta") && !hb.summary_md.contains("alpha")); +}