agent_engine/memory_backend/
forum.rs1use super::{error, MemoryBinding, USER_SCOPE};
3use crate::Result;
4use agent_core::memory::forum::{self, Envelope, Page, Post, PostRequest, Read, Receipt, Status};
5use serde_json::json;
6
7impl MemoryBinding {
8 fn require_forum(&self) -> Result<()> {
9 if !self.is_axel() || self.scope()?.key() == USER_SCOPE {
10 return Err(error("project forum requires selected Axel repository memory; unavailable for legacy/user scope; no fallback"));
11 }
12 self.forum_author().validate().map_err(error)
13 }
14 pub async fn forum_post(&self, post: Post) -> Result<Receipt> {
15 self.require_forum()?;
16 let expected = Envelope::new(self.scope()?.key(), self.forum_author().clone(), &post)
17 .map_err(error)?;
18 let request = PostRequest {
19 author: self.forum_author().clone(),
20 post,
21 };
22 let value = self.call("forum_post",serde_json::to_value(request).map_err(|_|error("forum serialization failed"))?).await
23 .map_err(|cause|error(format!("forum post has no success receipt for {}: {cause}. Do not claim publication. If the outcome is uncertain, reconcile first; any exact retry must keep the original request key, payload and host author",expected.id())))?;
24 let receipt: Receipt = serde_json::from_value(value).map_err(|_| {
25 error(format!(
26 "invalid forum acknowledgement; commit unknown for {}",
27 expected.id()
28 ))
29 })?;
30 if receipt.id != expected.id()
31 || receipt.thread_id != expected.thread_id
32 || receipt.digest != expected.digest
33 || (receipt.status != Status::Tombstoned && receipt.timestamp_ms.is_none())
34 || receipt.timestamp_ms.is_some_and(|n| n > i64::MAX as u64)
35 {
36 return Err(error(format!(
37 "forum acknowledgement mismatch; commit unknown for {}",
38 expected.id()
39 )));
40 }
41 Ok(receipt)
42 }
43 pub async fn forum_read(&self, query: Read) -> Result<Page> {
44 self.require_forum()?;
45 query.validate().map_err(error)?;
46 let value = self
47 .call(
48 "forum_read",
49 serde_json::to_value(&query)
50 .map_err(|_| error("forum query serialization failed"))?,
51 )
52 .await?;
53 if serde_json::to_vec(&value)
54 .map_err(|_| error("forum result serialization failed"))?
55 .len()
56 > forum::PAGE_BYTES
57 {
58 return Err(error("forum page exceeds byte limit"));
59 }
60 let page: Page = serde_json::from_value(value).map_err(|_| error("invalid forum page"))?;
61 let projects = page
62 .entries
63 .iter()
64 .map(|e| e.envelope.project.as_str())
65 .collect::<Vec<_>>();
66 let members = self.authorized_projects(&projects).await?;
67 validate_page(&page, &query, &members)?;
68 Ok(page)
69 }
70 pub async fn forum_forget(&self, id: &str) -> Result<()> {
71 self.require_forum()?;
72 if !forum::valid_id(id) {
73 return Err(error("forum forget requires an exact msg- ID"));
74 }
75 match self.call("forum_forget", json!({"id":id})).await? {
76 serde_json::Value::Bool(true) => Ok(()),
77 serde_json::Value::Bool(false) => Err(error("forum post unavailable in this project")),
78 _ => Err(error(
79 "invalid forum forget acknowledgement; outcome unknown",
80 )),
81 }
82 }
83}
84fn validate_page(page: &Page, query: &Read, members: &[String]) -> Result<()> {
85 if page.entries.len() > query.limit
86 || page.entries.iter().map(|e| e.body.len()).sum::<usize>() > forum::PAGE_BODY_BYTES
87 {
88 return Err(error("forum page exceeds entry/body bound"));
89 }
90 let mut previous = query.after.clone();
91 for entry in &page.entries {
92 entry.validate().map_err(error)?;
93 let cursor = entry.cursor();
94 if !members.contains(&entry.envelope.project)
95 || previous.as_ref().is_some_and(|p| cursor <= *p)
96 || match &query.thread_id {
97 Some(id) => entry.envelope.thread_id != *id || entry.truncated,
98 None => !entry.envelope.is_root(),
99 }
100 {
101 return Err(error("forum page violates scope/order/thread constraints"));
102 }
103 previous = Some(cursor);
104 }
105 if let Some(next) = &page.next {
106 next.validate().map_err(error)?;
107 if page.entries.last().map(|e| e.cursor()).as_ref() != Some(next) {
108 return Err(error("forum cursor skips unreturned entries"));
109 }
110 }
111 Ok(())
112}
113
114#[cfg(test)]
115mod tests {
116 use super::*;
117 use forum::{Author, Entry};
118 fn entry() -> Entry {
119 let post = Post {
120 request_key: "one".into(),
121 thread_id: None,
122 reply_to: None,
123 title: "Finding".into(),
124 body: "synthetic note".into(),
125 retention_days: 30,
126 };
127 let envelope = Envelope::new("p1111111111111111", Author::fresh(), &post).unwrap();
128 Entry {
129 id: envelope.id(),
130 timestamp_ms: 1,
131 envelope,
132 body: post.body,
133 truncated: false,
134 }
135 }
136 #[test]
137 fn validates_scope_order_digest_and_cursor() {
138 let row = entry();
139 let scope = vec![row.envelope.project.clone()];
140 let q = Read::default();
141 let mut p = Page {
142 entries: vec![row.clone()],
143 next: Some(row.cursor()),
144 };
145 validate_page(&p, &q, &scope).unwrap();
146 assert!(validate_page(&p, &q, &["p2222222222222222".into()]).is_err());
147 p.entries[0].body.push('!');
148 assert!(validate_page(&p, &q, &scope).is_err());
149 p.entries[0] = row.clone();
150 p.entries.push(row.clone());
151 assert!(validate_page(&p, &q, &scope).is_err());
152 p.entries.pop();
153 p.next.as_mut().unwrap().timestamp_ms += 1;
154 assert!(validate_page(&p, &q, &scope).is_err());
155 }
156 #[test]
157 fn rejects_truncated_thread_reads_and_invalid_root_listing() {
158 let row = entry();
159 let scope = vec![row.envelope.project.clone()];
160 let q = Read {
161 thread_id: Some(row.id.clone()),
162 ..Default::default()
163 };
164 let mut p = Page {
165 entries: vec![row],
166 next: None,
167 };
168 validate_page(&p, &q, &scope).unwrap();
169 p.entries[0].truncated = true;
170 assert!(validate_page(&p, &q, &scope).is_err());
171 }
172}
173
174#[cfg(all(test, unix))]
175#[path = "forum_integration_tests.rs"]
176mod integration_tests;