//! 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")); }