agentd/store/
mod.rs

1// SPDX-License-Identifier: Apache-2.0
2//! The **state store** contract and adapters (RFC 0025 §2, §4).
3//!
4//! agentd's durability rests on four operations —
5//! `put(key, seq, envelope) / get(key[, seq]) / list(prefix) / delete(key)` —
6//! implemented by an adapter chosen in `store.kind`: [`mcp`] (any MCP server's
7//! tools, mapped), [`http`] (plain HTTP), [`file`] (the local filesystem, RFC
8//! 0033 — durable for one host, single-writer), or [`memory`] (in-process;
9//! tests and dev). `put` is a **compare-and-set on `seq`**: the stored seq
10//! must be lower, else `Conflict` — the split-brain guard every caller treats
11//! as fatal.
12//! agentd links no database client and defines no schema beyond the
13//! [`Envelope`].
14
15pub mod file;
16pub mod http;
17pub mod mapping;
18pub mod mcp;
19pub mod memory;
20
21use serde::{Deserialize, Serialize};
22use serde_json::Value;
23use std::sync::Arc;
24use std::time::Duration;
25
26/// The envelope major this build writes and accepts (RFC 0025 §3.2).
27pub const ENVELOPE_VERSION: u32 = 2;
28
29/// A versioned store record: `state` is the kind-specific payload; a tombstone
30/// carries `state: null`.
31#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
32pub struct Envelope {
33    pub v: u32,
34    pub kind: String,
35    pub id: String,
36    pub seq: u64,
37    /// Unix ms.
38    pub ts: u64,
39    pub instance: String,
40    #[serde(default, skip_serializing_if = "Option::is_none")]
41    pub hash: Option<String>,
42    pub state: Value,
43}
44
45impl Envelope {
46    pub fn new(
47        kind: &str,
48        id: &str,
49        seq: u64,
50        instance: &str,
51        hash: Option<String>,
52        state: Value,
53    ) -> Envelope {
54        Envelope {
55            v: ENVELOPE_VERSION,
56            kind: kind.to_string(),
57            id: id.to_string(),
58            seq,
59            ts: now_ms(),
60            instance: instance.to_string(),
61            hash,
62            state,
63        }
64    }
65
66    pub fn is_tombstone(&self) -> bool {
67        self.state.is_null()
68    }
69
70    pub fn to_value(&self) -> Value {
71        serde_json::to_value(self).unwrap_or(Value::Null)
72    }
73
74    /// Parse a stored value; refuses an unknown envelope major.
75    pub fn from_value(v: Value) -> Result<Envelope, StoreError> {
76        let env: Envelope = serde_json::from_value(v)
77            .map_err(|e| StoreError::Corrupt(format!("envelope does not parse: {e}")))?;
78        if env.v != ENVELOPE_VERSION {
79            return Err(StoreError::Corrupt(format!(
80                "envelope version {} is not supported (this build writes {})",
81                env.v, ENVELOPE_VERSION
82            )));
83        }
84        Ok(env)
85    }
86}
87
88/// `<prefix>/<instance>/<kind>/<id>` (RFC 0025 §3.1).
89pub fn key(prefix: &str, instance: &str, kind: &str, id: &str) -> String {
90    format!("{prefix}/{instance}/{kind}/{id}")
91}
92
93/// Split a key produced by [`key`] back into `(kind, id)` for the given
94/// prefix/instance; `None` for a foreign key.
95pub fn parse_key<'a>(prefix: &str, instance: &str, k: &'a str) -> Option<(&'a str, &'a str)> {
96    let rest = k.strip_prefix(&format!("{prefix}/{instance}/"))?;
97    let (kind, id) = rest.split_once('/')?;
98    Some((kind, id))
99}
100
101/// The outcome of a `put`.
102#[derive(Debug, Clone, PartialEq, Eq)]
103pub enum PutOutcome {
104    Ok,
105    /// Another writer owns the key (a stored seq ≥ ours). Fatal for the writer.
106    Conflict {
107        latest_seq: Option<u64>,
108    },
109}
110
111/// A `list` entry.
112#[derive(Debug, Clone, PartialEq, Eq)]
113pub struct KeySeq {
114    pub key: String,
115    pub seq: Option<u64>,
116}
117
118#[derive(Debug, Clone, PartialEq, Eq)]
119pub enum StoreError {
120    /// Transport / server failure (retryable at the caller's discretion).
121    Io(String),
122    /// The adapter does not implement this optional operation.
123    Unsupported(&'static str),
124    /// A mapping template or extraction failed (a config problem).
125    Mapping(String),
126    /// A stored record is unreadable.
127    Corrupt(String),
128    /// A `put` conflict surfaced as an error by a caller that treats it as fatal.
129    Conflict(String),
130}
131
132impl std::fmt::Display for StoreError {
133    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
134        match self {
135            StoreError::Io(m) => write!(f, "store i/o: {m}"),
136            StoreError::Unsupported(op) => {
137                write!(f, "store: {op} is not supported by this adapter")
138            }
139            StoreError::Mapping(m) => write!(f, "store mapping: {m}"),
140            StoreError::Corrupt(m) => write!(f, "store record: {m}"),
141            StoreError::Conflict(m) => write!(f, "store conflict: {m}"),
142        }
143    }
144}
145
146impl std::error::Error for StoreError {}
147
148/// The four-operation contract (RFC 0025 §2). Implementations are `Send +
149/// Sync` so the runtime's executor pool can call them; every operation is
150/// bounded by the adapter's timeout.
151pub trait Store: Send + Sync {
152    /// Compare-and-set write: `seq` must exceed the stored one.
153    fn put(&self, key: &str, seq: u64, envelope: &Value) -> Result<PutOutcome, StoreError>;
154    /// The latest record (or the pinned `seq` if the store keeps history).
155    fn get(&self, key: &str, seq: Option<u64>) -> Result<Option<Value>, StoreError>;
156    /// Keys under `prefix` (optional; `Unsupported` is a legal answer).
157    fn list(&self, prefix: &str) -> Result<Vec<KeySeq>, StoreError>;
158    /// Remove a key (optional; `Unsupported` ⇒ callers tombstone via `put`).
159    fn delete(&self, key: &str) -> Result<(), StoreError>;
160    /// The adapter kind (`mcp` / `http` / `file` / `memory`), for status and
161    /// metrics.
162    fn kind(&self) -> &'static str;
163}
164
165/// A shared store handle.
166pub type SharedStore = Arc<dyn Store>;
167
168/// The store timeout class: the management timeout (RFC 0016 §10) unless the
169/// settings say otherwise.
170pub fn default_timeout() -> Duration {
171    crate::obs::health::management_timeout()
172}
173
174/// Bounded retry on `Io` errors (never on `Conflict`/`Mapping`/`Corrupt`).
175pub fn with_retry<T>(
176    mut op: impl FnMut() -> Result<T, StoreError>,
177    attempts: u32,
178) -> Result<T, StoreError> {
179    let mut last = None;
180    for n in 0..attempts.max(1) {
181        match op() {
182            Err(StoreError::Io(m)) => {
183                last = Some(StoreError::Io(m));
184                if n + 1 < attempts {
185                    std::thread::sleep(Duration::from_millis(50 * (n as u64 + 1)));
186                }
187            }
188            other => return other,
189        }
190    }
191    Err(last.unwrap_or(StoreError::Io("no attempts".into())))
192}
193
194pub(crate) fn now_ms() -> u64 {
195    std::time::SystemTime::now()
196        .duration_since(std::time::UNIX_EPOCH)
197        .map(|d| d.as_millis() as u64)
198        .unwrap_or(0)
199}
200
201/// Build the store an instance is configured with (RFC 0030 §3.5). `servers`
202/// resolves the `mcp` adapter's coordination server by name; `kind: none`
203/// yields no store (the caller decides whether that is allowed).
204pub fn open(
205    settings: &crate::config::v2::Store,
206    servers: &dyn Fn(&str) -> Option<Arc<dyn mcp::McpCall>>,
207) -> Result<Option<SharedStore>, StoreError> {
208    use crate::config::v2::StoreKind;
209    let timeout = settings
210        .timeout
211        .map(|d| d.0)
212        .unwrap_or_else(default_timeout);
213    match settings.kind {
214        StoreKind::None => Ok(None),
215        StoreKind::Memory => Ok(Some(Arc::new(memory::MemoryStore::new()))),
216        StoreKind::Mcp => {
217            let cfg = settings.mcp.as_ref().ok_or_else(|| {
218                StoreError::Mapping("store.kind is mcp but store.mcp is not set".into())
219            })?;
220            let client = servers(&cfg.server).ok_or_else(|| {
221                StoreError::Mapping(format!(
222                    "store.mcp.server '{}' is not a connected MCP server",
223                    cfg.server
224                ))
225            })?;
226            Ok(Some(Arc::new(mcp::McpStore::new(
227                client,
228                cfg.clone(),
229                timeout,
230            ))))
231        }
232        StoreKind::File => {
233            // The root is resolved by the config module (RFC 0033 §4) so the
234            // startup log, `--capabilities` and this open all name the same
235            // directory. `store.file` may be absent entirely — the chain then
236            // runs on the environment alone.
237            let root = crate::config::v2::file_store_root(settings);
238            // `open` takes the exclusive instance lock, and a held lock arrives
239            // as `Io` (RFC 0033 §4.1) carrying the holder's pid. Name the
240            // adapter in front of it: the operator reads this at exit, where
241            // "store i/o: …" alone would not say which store or which path.
242            let store = file::FileStore::open(&root).map_err(|e| match e {
243                StoreError::Io(m) => StoreError::Io(format!("store.file: {m}")),
244                other => other,
245            })?;
246            Ok(Some(Arc::new(store)))
247        }
248        StoreKind::Http => {
249            let cfg = settings.http.as_ref().ok_or_else(|| {
250                StoreError::Mapping("store.kind is http but store.http is not set".into())
251            })?;
252            Ok(Some(Arc::new(http::HttpStore::new(cfg.clone(), timeout)?)))
253        }
254    }
255}
256
257#[cfg(test)]
258mod tests {
259    use super::*;
260    use serde_json::json;
261
262    #[test]
263    fn envelope_round_trips_and_refuses_unknown_major() {
264        let e = Envelope::new("run", "01J", 3, "inst", Some("abc".into()), json!({"a": 1}));
265        let v = e.to_value();
266        assert_eq!(v["v"], json!(2));
267        assert_eq!(v["seq"], json!(3));
268        let back = Envelope::from_value(v).unwrap();
269        assert_eq!(back, e);
270        let mut bad = e.to_value();
271        bad["v"] = json!(9);
272        assert!(matches!(
273            Envelope::from_value(bad),
274            Err(StoreError::Corrupt(_))
275        ));
276        assert!(!e.is_tombstone());
277        assert!(Envelope::new("run", "x", 1, "i", None, Value::Null).is_tombstone());
278    }
279
280    #[test]
281    fn keys_compose_and_parse() {
282        let k = key("agentd", "inst-0", "run", "01J");
283        assert_eq!(k, "agentd/inst-0/run/01J");
284        assert_eq!(parse_key("agentd", "inst-0", &k), Some(("run", "01J")));
285        assert_eq!(parse_key("agentd", "other", &k), None);
286        // Ids may contain slashes (a2a contextId…): kind is the first segment.
287        assert_eq!(
288            parse_key("agentd", "i", "agentd/i/task/a/b"),
289            Some(("task", "a/b"))
290        );
291    }
292
293    #[test]
294    fn retry_only_on_io() {
295        let mut n = 0;
296        let r: Result<(), StoreError> = with_retry(
297            || {
298                n += 1;
299                Err(StoreError::Io("down".into()))
300            },
301            3,
302        );
303        assert!(matches!(r, Err(StoreError::Io(_))));
304        assert_eq!(n, 3);
305        let mut m = 0;
306        let r: Result<(), StoreError> = with_retry(
307            || {
308                m += 1;
309                Err(StoreError::Mapping("bad".into()))
310            },
311            3,
312        );
313        assert!(matches!(r, Err(StoreError::Mapping(_))));
314        assert_eq!(m, 1);
315    }
316}
agentd/store/mod.rs

agentd/store/
mod.rs
