nexo_core/
config_changes_store.rs

1//! Durable audit log of `ConfigTool` proposals + their lifecycle
2//! (proposed → applied | rolled_back | rejected | expired).
3//!
4//! Two consumers:
5//!   * `crates/core/src/agent/config_tool.rs` writes one row per
6//!     state transition (defense-in-depth: even if the staging
7//!     file disappears, the audit row survives).
8//!   * `crates/core/src/agent/config_changes_tail_tool.rs` is the
9//!     read-only LLM tool that lets a model post-mortem its own
10//!     mutation history.
11//!
12//! Uses idempotent-on-id semantics so a duplicate write is a no-op.
13
14use async_trait::async_trait;
15use serde::{Deserialize, Serialize};
16use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
17use sqlx::SqlitePool;
18use std::str::FromStr;
19
20#[derive(Debug, thiserror::Error)]
21pub enum ConfigChangesError {
22    #[error("sqlx error: {0}")]
23    Sqlx(#[from] sqlx::Error),
24    #[error("sqlx migrate error: {0}")]
25    Migrate(#[from] sqlx::migrate::MigrateError),
26    #[error("invalid n parameter: {0}")]
27    InvalidN(usize),
28}
29
30/// One row of the audit log. `status` is the transition that
31/// produced this row; the same `patch_id` may appear multiple
32/// times across statuses, but `(patch_id, status)` is unique.
33#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
34pub struct ConfigChangeRow {
35    pub patch_id: String,
36    pub binding_id: String,
37    pub agent_id: String,
38    /// `propose` | `apply` | `expire` | `reject`. Identifies the
39    /// op that fired this row (one row per transition).
40    pub op: String,
41    pub key: String,
42    /// YAML-rendered value. Caller is responsible for redacting
43    /// secrets BEFORE passing to `record` — store does not
44    /// inspect strings.
45    pub value: Option<String>,
46    /// `proposed` | `applied` | `rolled_back` | `rejected` |
47    /// `expired`. Pre-rendered so a SQL filter can pick rows of
48    /// interest without parsing op transitions.
49    pub status: String,
50    pub error: Option<String>,
51    pub created_at: i64,
52    pub applied_at: Option<i64>,
53}
54
55#[async_trait]
56pub trait ConfigChangesStore: Send + Sync + 'static {
57    /// Idempotent on `(patch_id, status)` — repeat insert with
58    /// the same key is a no-op. Caller passes one row per
59    /// state transition.
60    async fn record(&self, row: &ConfigChangeRow) -> Result<(), ConfigChangesError>;
61
62    /// Latest `n` rows ordered by `created_at DESC`. Capped at
63    /// 200 internally so a runaway tool call cannot pull the full
64    /// table into memory.
65    async fn tail(&self, n: usize) -> Result<Vec<ConfigChangeRow>, ConfigChangesError>;
66
67    /// Latest row for a given `patch_id`, regardless of status.
68    /// Returns `None` when no row exists yet (proposal never
69    /// recorded — only happens during a race window between
70    /// staging file write and the first `record`).
71    async fn get(&self, patch_id: &str) -> Result<Option<ConfigChangeRow>, ConfigChangesError>;
72}
73
74const MAX_TAIL_ROWS: usize = 200;
75
76const SCHEMA_SQL: &str = r#"
77CREATE TABLE IF NOT EXISTS config_changes (
78    patch_id    TEXT NOT NULL,
79    status      TEXT NOT NULL,
80    binding_id  TEXT NOT NULL,
81    agent_id    TEXT NOT NULL,
82    op          TEXT NOT NULL,
83    key         TEXT NOT NULL,
84    value       TEXT,
85    error       TEXT,
86    created_at  INTEGER NOT NULL,
87    applied_at  INTEGER,
88    PRIMARY KEY (patch_id, status)
89);
90CREATE INDEX IF NOT EXISTS idx_config_changes_created_at ON config_changes(created_at DESC);
91CREATE INDEX IF NOT EXISTS idx_config_changes_patch_id ON config_changes(patch_id);
92"#;
93
94pub struct SqliteConfigChangesStore {
95    pool: SqlitePool,
96}
97
98impl SqliteConfigChangesStore {
99    /// Open or create the SQLite store at `url`. Pass
100    /// `"sqlite::memory:"` for tests. Schema is idempotent.
101    pub async fn open(url: &str) -> Result<Self, ConfigChangesError> {
102        let opts = SqliteConnectOptions::from_str(url)?.create_if_missing(true);
103        let pool = SqlitePoolOptions::new()
104            .max_connections(2)
105            .connect_with(opts)
106            .await?;
107        sqlx::query(SCHEMA_SQL).execute(&pool).await?;
108        Ok(Self { pool })
109    }
110
111    pub async fn open_in_memory() -> Result<Self, ConfigChangesError> {
112        Self::open("sqlite::memory:").await
113    }
114}
115
116#[async_trait]
117impl ConfigChangesStore for SqliteConfigChangesStore {
118    async fn record(&self, row: &ConfigChangeRow) -> Result<(), ConfigChangesError> {
119        // ON CONFLICT DO NOTHING — the (patch_id, status) PK
120        // makes this idempotent across replays.
121        sqlx::query(
122            r#"INSERT INTO config_changes
123                (patch_id, status, binding_id, agent_id, op, key, value, error, created_at, applied_at)
124                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
125                ON CONFLICT(patch_id, status) DO NOTHING"#,
126        )
127        .bind(&row.patch_id)
128        .bind(&row.status)
129        .bind(&row.binding_id)
130        .bind(&row.agent_id)
131        .bind(&row.op)
132        .bind(&row.key)
133        .bind(&row.value)
134        .bind(&row.error)
135        .bind(row.created_at)
136        .bind(row.applied_at)
137        .execute(&self.pool)
138        .await?;
139        Ok(())
140    }
141
142    async fn tail(&self, n: usize) -> Result<Vec<ConfigChangeRow>, ConfigChangesError> {
143        let limit = n.clamp(1, MAX_TAIL_ROWS) as i64;
144        let rows = sqlx::query_as::<_, ConfigChangeRow>(
145            r#"SELECT patch_id, binding_id, agent_id, op, key, value, status, error,
146                      created_at, applied_at
147                 FROM config_changes
148                ORDER BY created_at DESC, rowid DESC
149                LIMIT ?"#,
150        )
151        .bind(limit)
152        .fetch_all(&self.pool)
153        .await?;
154        Ok(rows)
155    }
156
157    async fn get(&self, patch_id: &str) -> Result<Option<ConfigChangeRow>, ConfigChangesError> {
158        let row = sqlx::query_as::<_, ConfigChangeRow>(
159            r#"SELECT patch_id, binding_id, agent_id, op, key, value, status, error,
160                      created_at, applied_at
161                 FROM config_changes
162                WHERE patch_id = ?
163                ORDER BY created_at DESC, rowid DESC
164                LIMIT 1"#,
165        )
166        .bind(patch_id)
167        .fetch_optional(&self.pool)
168        .await?;
169        Ok(row)
170    }
171}
172
173impl<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow> for ConfigChangeRow {
174    fn from_row(row: &'r sqlx::sqlite::SqliteRow) -> Result<Self, sqlx::Error> {
175        use sqlx::Row;
176        Ok(Self {
177            patch_id: row.try_get("patch_id")?,
178            binding_id: row.try_get("binding_id")?,
179            agent_id: row.try_get("agent_id")?,
180            op: row.try_get("op")?,
181            key: row.try_get("key")?,
182            value: row.try_get("value")?,
183            status: row.try_get("status")?,
184            error: row.try_get("error")?,
185            created_at: row.try_get("created_at")?,
186            applied_at: row.try_get("applied_at")?,
187        })
188    }
189}
190
191#[cfg(test)]
192mod tests {
193    use super::*;
194
195    fn fixture(patch_id: &str, status: &str, created_at: i64) -> ConfigChangeRow {
196        ConfigChangeRow {
197            patch_id: patch_id.into(),
198            binding_id: "wa:default".into(),
199            agent_id: "cody".into(),
200            op: "propose".into(),
201            key: "model.model".into(),
202            value: Some("\"claude-opus-4-7\"".into()),
203            status: status.into(),
204            error: None,
205            created_at,
206            applied_at: None,
207        }
208    }
209
210    #[tokio::test]
211    async fn open_in_memory_creates_schema() {
212        let store = SqliteConfigChangesStore::open_in_memory().await.unwrap();
213        let tail = store.tail(10).await.unwrap();
214        assert!(tail.is_empty());
215    }
216
217    #[tokio::test]
218    async fn record_then_tail_returns_rows() {
219        let store = SqliteConfigChangesStore::open_in_memory().await.unwrap();
220        store
221            .record(&fixture("01J7AAA", "proposed", 100))
222            .await
223            .unwrap();
224        store
225            .record(&fixture("01J7BBB", "proposed", 200))
226            .await
227            .unwrap();
228        store
229            .record(&fixture("01J7CCC", "proposed", 300))
230            .await
231            .unwrap();
232        let tail = store.tail(10).await.unwrap();
233        assert_eq!(tail.len(), 3);
234        // Newest first.
235        assert_eq!(tail[0].patch_id, "01J7CCC");
236        assert_eq!(tail[1].patch_id, "01J7BBB");
237        assert_eq!(tail[2].patch_id, "01J7AAA");
238    }
239
240    #[tokio::test]
241    async fn idempotent_on_patch_id_status() {
242        let store = SqliteConfigChangesStore::open_in_memory().await.unwrap();
243        store
244            .record(&fixture("01J7AAA", "proposed", 100))
245            .await
246            .unwrap();
247        // Same (patch_id, status) — no-op.
248        store
249            .record(&fixture("01J7AAA", "proposed", 999))
250            .await
251            .unwrap();
252        let tail = store.tail(10).await.unwrap();
253        assert_eq!(tail.len(), 1);
254        // Original timestamp survives.
255        assert_eq!(tail[0].created_at, 100);
256    }
257
258    #[tokio::test]
259    async fn different_status_for_same_patch_id_appends_new_row() {
260        let store = SqliteConfigChangesStore::open_in_memory().await.unwrap();
261        store
262            .record(&fixture("01J7AAA", "proposed", 100))
263            .await
264            .unwrap();
265        store
266            .record(&fixture("01J7AAA", "applied", 200))
267            .await
268            .unwrap();
269        let tail = store.tail(10).await.unwrap();
270        assert_eq!(tail.len(), 2);
271    }
272
273    #[tokio::test]
274    async fn get_returns_latest_status_for_patch() {
275        let store = SqliteConfigChangesStore::open_in_memory().await.unwrap();
276        store
277            .record(&fixture("01J7AAA", "proposed", 100))
278            .await
279            .unwrap();
280        store
281            .record(&fixture("01J7AAA", "applied", 200))
282            .await
283            .unwrap();
284        let got = store.get("01J7AAA").await.unwrap().unwrap();
285        assert_eq!(got.status, "applied");
286        assert_eq!(got.created_at, 200);
287
288        let missing = store.get("01J7ZZZ").await.unwrap();
289        assert!(missing.is_none());
290    }
291
292    #[tokio::test]
293    async fn tail_caps_at_max() {
294        let store = SqliteConfigChangesStore::open_in_memory().await.unwrap();
295        for i in 0..250 {
296            let row = fixture(&format!("p{i:03}"), "proposed", i as i64);
297            store.record(&row).await.unwrap();
298        }
299        // Caller asks for 1000; we cap at 200.
300        let tail = store.tail(1000).await.unwrap();
301        assert_eq!(tail.len(), MAX_TAIL_ROWS);
302    }
303}
nexo_core/config_changes_store.rs

nexo_core/
config_changes_store.rs
