a2a_rs/port/
conversation_store.rs

1//! Durable conversation memory for a context.
2
3use async_trait::async_trait;
4
5use crate::domain::{A2AError, ContextId, Conversation, Digest};
6
7/// Reads and compacts the conversation recorded against one context.
8///
9/// Two methods, one capability. Splitting the tail and the digest into separate
10/// ports reads tidier and is wrong: a digest written between two reads leaves
11/// either a gap (messages the summary does not cover and the tail no longer
12/// includes) or duplicates, and nothing in two separate signatures says the
13/// reads have to agree. One method makes that boundary the implementation's
14/// problem, which is where it can actually be solved.
15///
16/// ## The caller argument
17///
18/// Every method takes the authenticated principal's id, or `None` where the
19/// agent runs without an authenticator. This is not ceremony: a `context_id`
20/// only groups tasks today, but a store that hands back a conversation turns it
21/// into a capability — whoever holds one can read what was said in it.
22/// Ownership is claimed on first write and enforced on every read, and a
23/// mismatch is [`A2AError::ContextAccessDenied`].
24#[async_trait]
25pub trait AsyncConversationStore: Send + Sync {
26    /// Load the newest digest and the messages recorded after its watermark.
27    ///
28    /// `limit` caps the tail, keeping the **newest** messages — a conversation
29    /// too long to load whole is more usefully truncated at its start, and the
30    /// part before it is what compaction summarizes. `None` loads everything
31    /// after the watermark.
32    ///
33    /// An unknown context is an empty [`Conversation`], not an error: the first
34    /// turn of a conversation asks for history that does not exist yet.
35    async fn load(
36        &self,
37        context_id: &ContextId,
38        caller: Option<&str>,
39        limit: Option<u32>,
40    ) -> Result<Conversation, A2AError>;
41
42    /// Append a digest covering everything through
43    /// [`Digest::covers_through`], claiming the context for `caller` if it is
44    /// new.
45    ///
46    /// Appends. Nothing is deleted and no earlier digest is replaced, so a
47    /// concurrent compaction of the same conversation costs duplicated work
48    /// rather than a lost or doubled transcript.
49    async fn compact(
50        &self,
51        context_id: &ContextId,
52        caller: Option<&str>,
53        digest: Digest,
54    ) -> Result<(), A2AError>;
55}
56
57/// Conveniences over [`AsyncConversationStore`].
58///
59/// Blanket-implemented, so they ride along on `Arc<dyn AsyncConversationStore>`
60/// too. `?Sized` is what makes that work.
61#[async_trait]
62pub trait AsyncConversationStoreExt: AsyncConversationStore {
63    /// Load a conversation, keeping at most `keep` of the most recent messages.
64    async fn load_recent(
65        &self,
66        context_id: &ContextId,
67        caller: Option<&str>,
68        keep: u32,
69    ) -> Result<Conversation, A2AError> {
70        self.load(context_id, caller, Some(keep)).await
71    }
72
73    /// Summarize everything loaded so far under one digest.
74    ///
75    /// Takes the watermark from the conversation itself, which is the common
76    /// case and the easy one to get wrong: a watermark computed from anything
77    /// other than what was actually summarized either re-summarizes messages or
78    /// drops them.
79    async fn compact_through(
80        &self,
81        context_id: &ContextId,
82        caller: Option<&str>,
83        conversation: &Conversation,
84        summary: String,
85        model: String,
86    ) -> Result<(), A2AError> {
87        let digest = Digest {
88            covers_through: conversation.watermark(),
89            summary,
90            replaced_messages: conversation.tail.len() as u32,
91            model,
92        };
93        self.compact(context_id, caller, digest).await
94    }
95}
96
97impl<T: AsyncConversationStore + ?Sized> AsyncConversationStoreExt for T {}
98
99/// A store that remembers nothing.
100///
101/// The adapter for `mode = "none"`: every load is an empty conversation and
102/// every compaction is a no-op. It exists so "this agent does not carry history"
103/// is a wired-up choice rather than an absent collaborator, which is what lets
104/// the handler take one code path either way.
105#[derive(Debug, Clone, Copy, Default)]
106pub struct NoConversationMemory;
107
108#[async_trait]
109impl AsyncConversationStore for NoConversationMemory {
110    async fn load(
111        &self,
112        _context_id: &ContextId,
113        _caller: Option<&str>,
114        _limit: Option<u32>,
115    ) -> Result<Conversation, A2AError> {
116        Ok(Conversation::default())
117    }
118
119    async fn compact(
120        &self,
121        _context_id: &ContextId,
122        _caller: Option<&str>,
123        _digest: Digest,
124    ) -> Result<(), A2AError> {
125        Ok(())
126    }
127}
128
129#[cfg(test)]
130mod tests {
131    use super::*;
132    use std::str::FromStr;
133
134    #[tokio::test]
135    async fn the_no_memory_store_reports_an_empty_conversation() {
136        let store = NoConversationMemory;
137        let context = ContextId::from_str("ctx-1").unwrap();
138
139        let conversation = store.load(&context, None, None).await.unwrap();
140        assert!(conversation.is_empty());
141
142        // And compacting it is not an error, so a handler needs no branch.
143        store
144            .compact_through(
145                &context,
146                None,
147                &conversation,
148                "nothing happened".to_string(),
149                "test".to_string(),
150            )
151            .await
152            .unwrap();
153    }
154}
a2a_rs/port/conversation_store.rs

a2a_rs/port/
conversation_store.rs
