agentmux_srv/registry/
def_store.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! `DefinitionStore` — file-per-definition CRUD on `<root>/<id>.json`
5//! with a sibling `retired/<id>.json` tombstone tree.
6//!
7//! Sibling of `store.rs` (the named-*instance* `Registry`). Same on-disk
8//! discipline: atomic rename for cross-process safety, a forward-compat
9//! merge that preserves unknown fields, refuses schema downgrades, and
10//! never clobbers an unparseable file. Holds the GLOBAL agent-definition
11//! roster so any channel/version sees the same agents (P0.2 of
12//! `docs/specs/SPEC_CROSS_CHANNEL_AGENT_PERSISTENCE_2026-06-13.md`).
13//!
14//! The forward-compat merge helpers below intentionally parallel the
15//! private ones in `store.rs`. They are kept separate (not shared) so
16//! this addition is fully isolated from the in-production instance
17//! registry; a later cleanup can extract a generic file-store both use.
18
19use std::path::{Path, PathBuf};
20use std::sync::Mutex;
21
22use serde_json::Value;
23use thiserror::Error;
24
25use super::atomic::{rename_atomic, write_atomic};
26use super::def_schema::{validate, DefValidationError, DefinitionRecord, DEF_MAX_SUPPORTED_SCHEMA};
27
28#[derive(Debug, Error)]
29pub enum DefStoreError {
30    #[error("io: {0}")]
31    Io(#[from] std::io::Error),
32    #[error("json: {0}")]
33    Json(#[from] serde_json::Error),
34    #[error("validation: {0}")]
35    Validation(#[from] DefValidationError),
36}
37
38pub struct DefinitionStore {
39    root: PathBuf,
40    write_lock: Mutex<()>,
41}
42
43impl DefinitionStore {
44    /// Open or create the store rooted at `root`. Ensures the active dir
45    /// and `retired/` subdir both exist.
46    pub fn open(root: PathBuf) -> Result<Self, DefStoreError> {
47        std::fs::create_dir_all(&root)?;
48        std::fs::create_dir_all(root.join("retired"))?;
49        Ok(Self {
50            root,
51            write_lock: Mutex::new(()),
52        })
53    }
54
55    pub fn root(&self) -> &Path {
56        &self.root
57    }
58
59    fn active_path(&self, id: &str) -> PathBuf {
60        self.root.join(format!("{id}.json"))
61    }
62
63    fn retired_path(&self, id: &str) -> PathBuf {
64        self.root.join("retired").join(format!("{id}.json"))
65    }
66
67    /// Insert or update a definition. Preserves unknown top-level + `data`
68    /// fields on an existing file (forward-compat); refuses to overwrite a
69    /// corrupt or higher-schema file.
70    pub fn upsert(&self, rec: &DefinitionRecord) -> Result<(), DefStoreError> {
71        let _g = self.write_lock.lock().unwrap_or_else(|e| e.into_inner());
72        let path = self.active_path(&rec.data.id);
73        let bytes = match std::fs::read(&path) {
74            Ok(existing) => match merge_for_write(&existing, rec)? {
75                Some(b) => b,
76                None => return Ok(()),
77            },
78            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
79                // Don't resurrect a tombstoned definition: a missing active
80                // file plus an existing `retired/<id>.json` means the agent
81                // was deliberately deleted (possibly in another channel). A
82                // stale cross-channel mirror calling `upsert` must NOT recreate
83                // it; the caller must `unretire` first to bring it back.
84                // (codex P1 on #1384.)
85                if self.retired_path(&rec.data.id).exists() {
86                    return Ok(());
87                }
88                to_pretty(rec)?
89            }
90            Err(e) => return Err(e.into()),
91        };
92        write_atomic(&path, &bytes)?;
93        Ok(())
94    }
95
96    /// Soft-delete → move into `retired/` (tombstone). Idempotent. The
97    /// tombstone prevents another channel's migration from resurrecting an
98    /// agent the user deliberately deleted.
99    pub fn retire(&self, id: &str) -> Result<(), DefStoreError> {
100        let _g = self.write_lock.lock().unwrap_or_else(|e| e.into_inner());
101        let from = self.active_path(id);
102        if !from.exists() {
103            return Ok(());
104        }
105        rename_atomic(&from, &self.retired_path(id))?;
106        Ok(())
107    }
108
109    /// Move a record back from `retired/` to active. Idempotent.
110    pub fn unretire(&self, id: &str) -> Result<(), DefStoreError> {
111        let _g = self.write_lock.lock().unwrap_or_else(|e| e.into_inner());
112        let from = self.retired_path(id);
113        if !from.exists() {
114            return Ok(());
115        }
116        rename_atomic(&from, &self.active_path(id))?;
117        Ok(())
118    }
119
120    /// Hard-delete (drops both active and retired files). Mirrors SQLite
121    /// `agent_def_delete`. Use [`Self::retire`] when a resurrection-proof
122    /// tombstone is wanted instead.
123    pub fn hard_delete(&self, id: &str) -> Result<(), DefStoreError> {
124        let _g = self.write_lock.lock().unwrap_or_else(|e| e.into_inner());
125        for p in [self.active_path(id), self.retired_path(id)] {
126            match std::fs::remove_file(&p) {
127                Ok(()) => {}
128                Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
129                Err(e) => return Err(e.into()),
130            }
131        }
132        Ok(())
133    }
134
135    pub fn exists(&self, id: &str) -> bool {
136        self.active_path(id).exists()
137    }
138
139    pub fn exists_anywhere(&self, id: &str) -> bool {
140        self.active_path(id).exists() || self.retired_path(id).exists()
141    }
142
143    /// Read a single active definition record by id. `None` if absent;
144    /// invalid/corrupt files surface as an error (caller logs + falls back).
145    pub fn get(&self, id: &str) -> Result<Option<DefinitionRecord>, DefStoreError> {
146        match std::fs::read(self.active_path(id)) {
147            Ok(bytes) => {
148                let rec: DefinitionRecord = serde_json::from_slice(&bytes)?;
149                validate(id, &rec)?;
150                Ok(Some(rec))
151            }
152            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
153            Err(e) => Err(e.into()),
154        }
155    }
156
157    /// Read every valid active definition. Invalid files are skipped +
158    /// logged (left on disk for ops triage).
159    pub fn list_active(&self) -> Result<Vec<DefinitionRecord>, DefStoreError> {
160        let mut out = Vec::new();
161        for entry in std::fs::read_dir(&self.root)? {
162            let entry = entry?;
163            let path = entry.path();
164            if !path.is_file() {
165                continue;
166            }
167            let Some(stem) = path.file_stem().and_then(|s| s.to_str()) else {
168                continue;
169            };
170            if path.extension().and_then(|s| s.to_str()) != Some("json") {
171                continue;
172            }
173            match read_and_validate(&path, stem) {
174                Ok(rec) => out.push(rec),
175                Err(e) => {
176                    tracing::warn!(
177                        file = %path.display(),
178                        error = %e,
179                        "def registry: skipping invalid record"
180                    );
181                }
182            }
183        }
184        Ok(out)
185    }
186}
187
188fn read_and_validate(path: &Path, stem: &str) -> Result<DefinitionRecord, DefStoreError> {
189    let bytes = std::fs::read(path)?;
190    let rec: DefinitionRecord = serde_json::from_slice(&bytes)?;
191    validate(stem, &rec)?;
192    Ok(rec)
193}
194
195fn to_pretty(rec: &DefinitionRecord) -> Result<Vec<u8>, serde_json::Error> {
196    let mut bytes = serde_json::to_vec_pretty(rec)?;
197    bytes.push(b'\n');
198    Ok(bytes)
199}
200
201/// Merge an in-memory record into an on-disk file's raw JSON, preserving
202/// fields beyond this binary's struct shape. Returns `Ok(None)` (skip,
203/// leave file intact) on corrupt JSON, missing `schema_version`, or
204/// `schema_version` above `DEF_MAX_SUPPORTED_SCHEMA`. Parallels
205/// `store.rs::merge_for_write` (see module docs).
206fn merge_for_write(
207    existing: &[u8],
208    rec: &DefinitionRecord,
209) -> Result<Option<Vec<u8>>, DefStoreError> {
210    let on_disk: Value = match serde_json::from_slice(existing) {
211        Ok(v) => v,
212        Err(e) => {
213            tracing::warn!(
214                error = %e,
215                "def registry: existing file is unparseable JSON — refusing to overwrite"
216            );
217            return Ok(None);
218        }
219    };
220    // `try_from` so a number above u32::MAX is treated as unparseable
221    // rather than wrapping past the forward-compat guard.
222    let on_disk_version = on_disk
223        .get("schema_version")
224        .and_then(|v| v.as_u64())
225        .and_then(|v| u32::try_from(v).ok());
226    match on_disk_version {
227        Some(v) if v > DEF_MAX_SUPPORTED_SCHEMA => {
228            tracing::warn!(
229                on_disk = v,
230                writer_max = DEF_MAX_SUPPORTED_SCHEMA,
231                "def registry: on-disk schema_version > writer max — refusing downgrade"
232            );
233            return Ok(None);
234        }
235        Some(_) => {}
236        None => {
237            tracing::warn!("def registry: existing file lacks schema_version — refusing to overwrite");
238            return Ok(None);
239        }
240    }
241    let mut merged = on_disk;
242    let updates = serde_json::to_value(rec)?;
243    merge_known(&mut merged, &updates);
244    let mut bytes = serde_json::to_vec_pretty(&merged)?;
245    bytes.push(b'\n');
246    Ok(Some(bytes))
247}
248
249/// Overwrite `target`'s top-level keys with `updates`' keys, preserving
250/// keys absent from `updates`; recurses into `data`.
251fn merge_known(target: &mut Value, updates: &Value) {
252    let (Some(t), Some(u)) = (target.as_object_mut(), updates.as_object()) else {
253        *target = updates.clone();
254        return;
255    };
256    for (k, v) in u {
257        if k == "data" {
258            if let Some(t_data) = t.get_mut("data") {
259                merge_known(t_data, v);
260                continue;
261            }
262        }
263        t.insert(k.clone(), v.clone());
264    }
265}
266
267#[cfg(test)]
268mod tests {
269    use super::*;
270    use crate::registry::DefinitionRecordV1;
271
272    fn store() -> (tempfile::TempDir, DefinitionStore) {
273        let tmp = tempfile::tempdir().unwrap();
274        let s = DefinitionStore::open(tmp.path().join("definitions")).unwrap();
275        (tmp, s)
276    }
277
278    fn rec(id: &str, name: &str) -> DefinitionRecord {
279        DefinitionRecord {
280            schema_version: 1,
281            data: DefinitionRecordV1 {
282                id: id.to_string(),
283                slug: id.to_string(),
284                name: name.to_string(),
285                icon: "✦".to_string(),
286                provider: "claude".to_string(),
287                description: String::new(),
288                working_directory: String::new(),
289                shell: String::new(),
290                provider_flags: String::new(),
291                auto_start: 0,
292                restart_on_crash: 0,
293                idle_timeout_minutes: 0,
294                created_at: 1,
295                agent_type: "host".to_string(),
296                environment: String::new(),
297                agent_bus_id: String::new(),
298                is_seeded: 0,
299                accounts: String::new(),
300                parent_id: String::new(),
301                branch_label: String::new(),
302                updated_at: 1,
303                user_hidden: 0,
304                container_image: String::new(),
305                container_volumes: "[]".to_string(),
306                container_name: String::new(),
307                use_ambient_login: 0,
308                auto_continue_enabled: 0,
309                content: Vec::new(),
310                skills: Vec::new(),
311            },
312        }
313    }
314
315    #[test]
316    fn upsert_then_list() {
317        let (_t, s) = store();
318        s.upsert(&rec("aaa", "Demo")).unwrap();
319        let listed = s.list_active().unwrap();
320        assert_eq!(listed.len(), 1);
321        assert_eq!(listed[0].data.name, "Demo");
322    }
323
324    #[test]
325    fn upsert_update_replaces_known_fields() {
326        let (_t, s) = store();
327        s.upsert(&rec("aaa", "Demo")).unwrap();
328        s.upsert(&rec("aaa", "Renamed")).unwrap();
329        let listed = s.list_active().unwrap();
330        assert_eq!(listed.len(), 1);
331        assert_eq!(listed[0].data.name, "Renamed");
332    }
333
334    #[test]
335    fn retire_then_unretire_round_trips() {
336        let (_t, s) = store();
337        s.upsert(&rec("aaa", "Demo")).unwrap();
338        s.retire("aaa").unwrap();
339        assert!(s.list_active().unwrap().is_empty());
340        assert!(s.exists_anywhere("aaa"));
341        s.unretire("aaa").unwrap();
342        assert_eq!(s.list_active().unwrap().len(), 1);
343    }
344
345    #[test]
346    fn hard_delete_removes_both_paths() {
347        let (_t, s) = store();
348        s.upsert(&rec("aaa", "Demo")).unwrap();
349        s.retire("aaa").unwrap();
350        s.upsert(&rec("bbb", "Demo2")).unwrap();
351        s.hard_delete("aaa").unwrap();
352        s.hard_delete("bbb").unwrap();
353        assert!(!s.exists_anywhere("aaa"));
354        assert!(!s.exists_anywhere("bbb"));
355    }
356
357    #[test]
358    fn unknown_field_survives_older_writer_update() {
359        let (_t, s) = store();
360        // Write a record carrying a future field directly to disk.
361        let path = s.root().join("aaa.json");
362        let raw = serde_json::json!({
363            "schema_version": 1,
364            "data": {
365                "id": "aaa", "name": "Demo", "provider": "claude",
366                "tags": ["keep-me"]
367            }
368        });
369        std::fs::write(&path, serde_json::to_vec_pretty(&raw).unwrap()).unwrap();
370        // An older binary updates a known field.
371        s.upsert(&rec("aaa", "Renamed")).unwrap();
372        let on_disk: Value = serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap();
373        assert_eq!(on_disk.pointer("/data/tags"), Some(&serde_json::json!(["keep-me"])));
374        assert_eq!(on_disk.pointer("/data/name"), Some(&serde_json::json!("Renamed")));
375    }
376
377    #[test]
378    fn refuses_to_downgrade_higher_schema_on_disk() {
379        let (_t, s) = store();
380        let path = s.root().join("aaa.json");
381        let v999 = serde_json::json!({
382            "schema_version": 999,
383            "data": { "id": "aaa", "future_only": "field" }
384        });
385        std::fs::write(&path, serde_json::to_vec_pretty(&v999).unwrap()).unwrap();
386        s.upsert(&rec("aaa", "Renamed")).unwrap();
387        let after: Value = serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap();
388        assert_eq!(after.pointer("/schema_version"), Some(&serde_json::json!(999)));
389        assert_eq!(after.pointer("/data/future_only"), Some(&serde_json::json!("field")));
390    }
391
392    #[test]
393    fn upsert_does_not_resurrect_a_tombstoned_definition() {
394        let (_t, s) = store();
395        s.upsert(&rec("aaa", "Demo")).unwrap();
396        s.retire("aaa").unwrap();
397        // A stale cross-channel mirror upserts the same id.
398        s.upsert(&rec("aaa", "Resurrected")).unwrap();
399        // Tombstone respected: still not active, not resurrected.
400        assert!(
401            s.list_active().unwrap().is_empty(),
402            "tombstoned definition must not be resurrected by upsert"
403        );
404        assert!(s.exists_anywhere("aaa"), "tombstone file still present");
405        // An explicit unretire is required to bring it back.
406        s.unretire("aaa").unwrap();
407        assert_eq!(s.list_active().unwrap().len(), 1);
408    }
409}
agentmux_srv/registry/def_store.rs

agentmux_srv/registry/
def_store.rs
