a2a_rs/domain/
conversation.rs

1//! A conversation: the durable, ordered record of what was said in one context.
2//!
3//! A `context_id` groups the tasks of one conversation, and each task records a
4//! message per status transition. Read back in order, that is the conversation —
5//! the same thing an agent framework calls a session or a thread. These types
6//! name the pieces an agent needs to rebuild it: a sequence number, an optional
7//! summary of the part already compacted, and the messages after that summary.
8//!
9//! Pure data. Loading and appending are the [`AsyncConversationStore`] port's
10//! job.
11//!
12//! [`AsyncConversationStore`]: crate::port::AsyncConversationStore
13
14use serde::{Deserialize, Serialize};
15
16use crate::domain::core::Message;
17
18/// Position of a message within a conversation.
19///
20/// Monotonic and dense enough to compare: `a < b` means `a` was recorded first.
21/// Values are opaque — nothing outside a store should construct one except from
22/// a [`Digest`] watermark or [`Seq::START`].
23#[derive(
24    Debug, Clone, Copy, Default, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize,
25)]
26#[serde(transparent)]
27pub struct Seq(u64);
28
29impl Seq {
30    /// Before every recorded message. Loading from here loads everything.
31    pub const START: Seq = Seq(0);
32
33    /// Build a sequence number from a store's own ordering key.
34    pub fn new(value: u64) -> Self {
35        Self(value)
36    }
37
38    /// The raw ordering key, for a store that has to persist it.
39    pub fn get(self) -> u64 {
40        self.0
41    }
42}
43
44impl std::fmt::Display for Seq {
45    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
46        write!(f, "{}", self.0)
47    }
48}
49
50/// One recorded message, with its position.
51#[derive(Debug, Clone)]
52pub struct SequencedMessage {
53    /// Where this sits in the conversation.
54    pub seq: Seq,
55    /// What was said.
56    pub message: Message,
57}
58
59/// A summary standing in for everything up to and including a watermark.
60///
61/// Digests are appended, never updated. Two turns of the same conversation can
62/// decide to compact at once; both rows land, the one with the highest
63/// [`covers_through`](Self::covers_through) wins on load, and the other is
64/// wasted work rather than a corrupted transcript.
65#[derive(Debug, Clone)]
66pub struct Digest {
67    /// Everything at or below this sequence number is represented by
68    /// [`summary`](Self::summary) and need not be loaded.
69    pub covers_through: Seq,
70    /// The summary text, as written by a model.
71    pub summary: String,
72    /// How many messages it replaced. Reporting only.
73    pub replaced_messages: u32,
74    /// Which model wrote it, so a summary produced by a weak model is
75    /// identifiable after the fact.
76    pub model: String,
77}
78
79/// A conversation as loaded for a prompt: the summary of what came before, and
80/// everything recorded since.
81///
82/// Both halves come from one read. Fetching them separately would let a digest
83/// written in between leave either a gap (messages the summary does not cover
84/// and the tail no longer includes) or duplicates (messages present in both).
85#[derive(Debug, Clone, Default)]
86pub struct Conversation {
87    /// The newest digest, when the conversation has been compacted.
88    pub digest: Option<Digest>,
89    /// Messages after the digest's watermark, oldest first.
90    pub tail: Vec<SequencedMessage>,
91}
92
93impl Conversation {
94    /// Whether anything was said in this context at all.
95    pub fn is_empty(&self) -> bool {
96        self.digest.is_none() && self.tail.is_empty()
97    }
98
99    /// The summary text, if this conversation has been compacted.
100    pub fn summary(&self) -> Option<&str> {
101        self.digest.as_ref().map(|digest| digest.summary.as_str())
102    }
103
104    /// The highest sequence number loaded, or the digest watermark when the
105    /// tail is empty. This is what a new digest should cover through.
106    pub fn watermark(&self) -> Seq {
107        self.tail
108            .last()
109            .map(|message| message.seq)
110            .or_else(|| self.digest.as_ref().map(|digest| digest.covers_through))
111            .unwrap_or(Seq::START)
112    }
113}
114
115#[cfg(test)]
116mod tests {
117    use super::*;
118
119    fn sequenced(seq: u64) -> SequencedMessage {
120        SequencedMessage {
121            seq: Seq::new(seq),
122            message: Message::default(),
123        }
124    }
125
126    #[test]
127    fn an_empty_conversation_watermarks_at_the_start() {
128        assert_eq!(Conversation::default().watermark(), Seq::START);
129        assert!(Conversation::default().is_empty());
130    }
131
132    #[test]
133    fn the_watermark_follows_the_newest_message() {
134        let conversation = Conversation {
135            digest: None,
136            tail: vec![sequenced(4), sequenced(9)],
137        };
138        assert_eq!(conversation.watermark(), Seq::new(9));
139    }
140
141    /// A conversation compacted to its very end has no tail, and its watermark
142    /// is the digest's. Reporting `START` here would have the next compaction
143    /// re-summarize from the beginning.
144    #[test]
145    fn a_fully_compacted_conversation_keeps_the_digest_watermark() {
146        let conversation = Conversation {
147            digest: Some(Digest {
148                covers_through: Seq::new(12),
149                summary: "they talked".to_string(),
150                replaced_messages: 6,
151                model: "test".to_string(),
152            }),
153            tail: Vec::new(),
154        };
155        assert_eq!(conversation.watermark(), Seq::new(12));
156        assert!(!conversation.is_empty());
157    }
158}
a2a_rs/domain/conversation.rs

a2a_rs/domain/
conversation.rs
