feat(persistence): P6a — use case RecordTurn (checkpoint fin de tour, isolé)
Logique du checkpoint conversationnel isolée et testable, sans câblage live :
append du tour au log canonique → load handoff → fold incrémental (tour neuf
seul) → save. Mapping StoreError→AppError via From existant. Debounce repoussé.
- application/src/conversation/{mod,record}.rs : RecordTurn::record
- tests : 6 cas (incrémentalité prouvée via summarizer-espion, propagation
d'erreur sans effet de bord, isolation par conversation) ; suite verte
Câblage sur les seams réels (reply/ask) + composition root = P6b.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
360
crates/application/tests/conversation_record.rs
Normal file
360
crates/application/tests/conversation_record.rs
Normal file
@ -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<HashMap<ConversationId, Vec<ConversationTurn>>>,
|
||||
/// When set, `append` returns this error instead of appending.
|
||||
fail_append: Mutex<Option<StoreError>>,
|
||||
}
|
||||
|
||||
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<ConversationTurn> {
|
||||
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<TurnId>,
|
||||
) -> Result<Vec<ConversationTurn>, 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<Vec<ConversationTurn>, StoreError> {
|
||||
let all = self.thread(conversation);
|
||||
let start = all.len().saturating_sub(n);
|
||||
Ok(all[start..].to_vec())
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct InMemoryHandoffStore {
|
||||
store: Mutex<HashMap<ConversationId, Handoff>>,
|
||||
fail_load: Mutex<Option<StoreError>>,
|
||||
fail_save: Mutex<Option<StoreError>>,
|
||||
/// Number of completed `save` calls (used to prove no save-after-append-fail).
|
||||
saves: Mutex<usize>,
|
||||
}
|
||||
|
||||
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<Handoff> {
|
||||
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<Option<Handoff>, 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<Vec<(bool, usize)>>,
|
||||
}
|
||||
|
||||
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<Handoff>, 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"));
|
||||
}
|
||||
Reference in New Issue
Block a user