agentmux_srv/backend/storage/
managed.rs

1// Copyright 2025-2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! The "managed primitive" storage shape shared by standalone Skills and MCP
5//! Servers (v1 composable model, SPEC_V1_MCP_SKILLS_PRIMITIVES_2026_06_30.md;
6//! bundle-level refs per SPEC_BUNDLE_AS_CONTAINER_V2_2026_08_17.md).
7//!
8//! Both primitives are the same pattern: one catalog table whose rows are
9//! either global (visible to everyone) or private, an agent-level ref table
10//! binding rows to agents, and a bundle-level ref table binding rows to
11//! bundles. (Only skills use theirs now: a bundle carries no MCP servers,
12//! `SPEC_BUNDLE_CONTENTS_MEMORY_NOT_MCP_2026_10_07.md` §3.1.) Before this module `skills.rs` and `mcp_servers.rs` each carried
13//! their own copy of the eighteen list / get / delete / bind / unbind /
14//! upsert / access-check methods — identical except for table and column
15//! names and the noun in error messages
16//! (REPORT_DRY_AND_MODULARITY_AUDIT_2026_09_06.md §2.4). The per-primitive
17//! modules keep their public methods, response structs, doc comments and
18//! tests; only the bodies delegate here. Every SQL statement below is the
19//! one the hand-written methods used, with the names substituted.
20//!
21//! SQL is assembled with `format!` from the trait's `const` names only —
22//! never from runtime input — so the generated statements are exactly as
23//! fixed as the literals they replace. Values still go through `?N`
24//! parameters.
25//!
26//! ## `self` vs `catalog` (Phase 2 of SPEC_DURABLE_BINDINGS_2026_09_10.md)
27//!
28//! As of Phase 2's application-layer redirect (§5.4), `self` and the
29//! catalog table's store are no longer guaranteed to be the same physical
30//! SQLite file: `db_skills`/`db_mcp_servers` are authoritatively
31//! `identity_store` now, while `db_agent_skills_ref`/`db_bundle_skills_ref`
32//! (and the MCP equivalents) stay on `self` (`mstore`) until Phase 3/6
33//! promotes them too. Every method that used to join the catalog table
34//! against a ref table in one SQL statement now takes an explicit
35//! `catalog: &Store` parameter and does two queries composed in Rust
36//! instead — fetch the relevant ref ids from `self`, fetch the relevant
37//! catalog rows from `catalog`, combine and sort in memory. This is not a
38//! new pattern — `managed_bind_bundle`/`managed_upsert_unique_for_bundle`
39//! already took an `id_store: &Store` parameter for bundle-existence checks,
40//! for the identical reason. `managed_get` needs no such change — it is
41//! genuinely single-table, so callers invoke it directly on whichever store
42//! actually owns the table (`catalog` for a Skill/McpServer row).
43//!
44//! `managed_upsert_unique` additionally loses single-transaction atomicity
45//! by this same split: the catalog insert and the ref-table bind are now two
46//! separate connections, so they can't share a `rusqlite::Transaction`. This
47//! is a documented, accepted interim gap (§5.4) — narrowed by a best-effort
48//! compensating delete on the catalog row if the ref-bind fails, so a failed
49//! bind cannot leave a permanent, undiscoverable orphan catalog row behind.
50
51use std::collections::{HashMap, HashSet};
52
53use rusqlite::{params, params_from_iter, types::Value, Row};
54
55use super::error::StoreError;
56use super::store::Store;
57
58/// One managed primitive: its catalog row type plus the names that differ
59/// between primitives.
60pub trait ManagedResource: Sized {
61    /// Catalog table, e.g. `db_skills`.
62    const TABLE: &'static str;
63    /// The ref tables' column that points at `TABLE.id`, e.g. `skill_id`.
64    const REF_COL: &'static str;
65    /// Agent-level ref table `(agent_id, REF_COL)`, e.g. `db_agent_skills_ref`.
66    const AGENT_REF_TABLE: &'static str;
67    /// Bundle-level ref table `(bundle_id, REF_COL)`, e.g. `db_bundle_skills_ref`.
68    const BUNDLE_REF_TABLE: &'static str;
69    /// Every column of `TABLE`, in the order `from_row` reads them and
70    /// `values` produces them. Must include `id`, `name`, `is_global`,
71    /// `created_at` and `updated_at`.
72    const COLUMNS: &'static [&'static str];
73    /// Columns rewritten by `ON CONFLICT(id) DO UPDATE` — everything a caller
74    /// may edit; never `id`, `is_global` or `created_at`.
75    const UPDATE_COLUMNS: &'static [&'static str];
76    /// Noun in name-uniqueness errors: "`skill` name 'x' already bound…".
77    const NAME_NOUN: &'static str;
78    /// Noun with article in bundle errors: "cannot bind `an MCP server` to…".
79    const ARTICLE_NOUN: &'static str;
80    /// Noun in the cross-channel bind error: "cross-channel `MCP server` binding…".
81    const KIND: &'static str;
82
83    /// Read one row whose leading columns are `COLUMNS`, in order, from 0.
84    fn from_row(row: &Row<'_>) -> rusqlite::Result<Self>;
85    /// The row's values in `COLUMNS` order (`is_global` as 0/1).
86    fn values(&self) -> Vec<Value>;
87    fn id(&self) -> &str;
88    fn name(&self) -> &str;
89    /// Whether this row is a global (catalog-wide) row. Needed by
90    /// `managed_list`'s in-memory sort, now that combining a global set with
91    /// a private set can no longer happen inside one `ORDER BY` (Phase 2 of
92    /// SPEC_DURABLE_BINDINGS_2026_09_10.md).
93    fn is_global(&self) -> bool;
94    /// Last-updated timestamp, for the same in-memory sort as `is_global`.
95    fn updated_at(&self) -> i64;
96}
97
98/// Which ref table a scoped query is keyed on.
99#[derive(Clone, Copy, Debug, PartialEq, Eq)]
100pub enum Owner {
101    Agent,
102    Bundle,
103}
104
105impl Owner {
106    fn ref_table<R: ManagedResource>(self) -> &'static str {
107        match self {
108            Owner::Agent => R::AGENT_REF_TABLE,
109            Owner::Bundle => R::BUNDLE_REF_TABLE,
110        }
111    }
112    fn key_col(self) -> &'static str {
113        match self {
114            Owner::Agent => "agent_id",
115            Owner::Bundle => "bundle_id",
116        }
117    }
118    fn word(self) -> &'static str {
119        match self {
120            Owner::Agent => "agent",
121            Owner::Bundle => "bundle",
122        }
123    }
124}
125
126/// `?1, ?2, ... ?n`.
127fn placeholders(n: usize) -> String {
128    (1..=n).map(|i| format!("?{i}")).collect::<Vec<_>>().join(", ")
129}
130
131/// `?{start}, ?{start+1}, ... ?{start+n-1}` — for a `?N` list that must
132/// continue numbering after some earlier bound parameters (`managed_upsert_unique`'s
133/// dup-check binds `name`/`id` at `?1`/`?2` before any owner-bound ids).
134fn placeholders_from(start: usize, n: usize) -> String {
135    (start..start + n).map(|i| format!("?{i}")).collect::<Vec<_>>().join(", ")
136}
137
138fn update_set<R: ManagedResource>() -> String {
139    R::UPDATE_COLUMNS
140        .iter()
141        .map(|c| format!("{c}=excluded.{c}"))
142        .collect::<Vec<_>>()
143        .join(", ")
144}
145
146/// `row.values()` with `is_global` replaced when a method hard-codes it
147/// (the global upsert forces 1, the bundle-scoped upsert forces 0).
148fn values_with_global<R: ManagedResource>(row: &R, force_global: Option<bool>) -> Vec<Value> {
149    let mut v = row.values();
150    if let Some(g) = force_global {
151        let idx = R::COLUMNS
152            .iter()
153            .position(|c| *c == "is_global")
154            .expect("ManagedResource::COLUMNS must contain is_global");
155        v[idx] = Value::Integer(if g { 1 } else { 0 });
156    }
157    v
158}
159
160impl Store {
161    // ── Cross-store helpers (Phase 2 of SPEC_DURABLE_BINDINGS_2026_09_10.md) ──
162    // Small building blocks the rewritten methods below compose. Split out so
163    // each one locks exactly one connection for exactly as long as it needs
164    // to — important because `self` and `catalog` (or `id_store`) may be the
165    // SAME `Store` (test_state() aliases them; even in production `id_store`
166    // and `identity_store` can coincide with `mstore` in a degraded-mode
167    // fallback), and `Mutex<Connection>` is not reentrant. A method that
168    // locked `self` and then called into `catalog` while still holding that
169    // lock would deadlock the moment they aliased.
170
171    /// The ref ids `owner_id` holds for one primitive, on `self`.
172    fn managed_bound_ids<R: ManagedResource>(&self, owner: Owner, owner_id: &str) -> Result<HashSet<String>, StoreError> {
173        let conn = self.conn.lock().unwrap();
174        let sql = format!(
175            "SELECT {refc} FROM {reft} WHERE {key} = ?1",
176            refc = R::REF_COL,
177            reft = owner.ref_table::<R>(),
178            key = owner.key_col(),
179        );
180        let mut stmt = conn.prepare(&sql)?;
181        let rows = stmt.query_map(params![owner_id], |row| row.get::<_, String>(0))?;
182        let mut out = HashSet::new();
183        for r in rows {
184            out.insert(r?);
185        }
186        Ok(out)
187    }
188
189    /// `REF_COL -> count` across every agent-level ref row, on `self`.
190    fn managed_bound_counts<R: ManagedResource>(&self) -> Result<HashMap<String, i64>, StoreError> {
191        let conn = self.conn.lock().unwrap();
192        let sql = format!(
193            "SELECT {refc}, COUNT(*) FROM {reft} GROUP BY {refc}",
194            refc = R::REF_COL,
195            reft = R::AGENT_REF_TABLE,
196        );
197        let mut stmt = conn.prepare(&sql)?;
198        let rows = stmt.query_map([], |row| Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?)))?;
199        let mut out = HashMap::new();
200        for r in rows {
201            let (id, count) = r?;
202            out.insert(id, count);
203        }
204        Ok(out)
205    }
206
207    /// Every global row of a primitive, on `self` acting as the CATALOG
208    /// store — newest first.
209    fn managed_catalog_global<R: ManagedResource>(&self) -> Result<Vec<R>, StoreError> {
210        let sql = format!(
211            "SELECT {cols} FROM {table} WHERE is_global = 1 ORDER BY updated_at DESC",
212            cols = R::COLUMNS.join(", "),
213            table = R::TABLE,
214        );
215        let conn = self.conn.lock().unwrap();
216        let mut stmt = conn.prepare(&sql)?;
217        let rows = stmt.query_map([], |row| R::from_row(row))?;
218        let mut out = Vec::new();
219        for r in rows {
220            out.push(r?);
221        }
222        Ok(out)
223    }
224
225    /// Every global row, plus (if `ids` is non-empty) every row in `ids` —
226    /// global or private — on `self` acting as the CATALOG store. Unordered;
227    /// callers sort as needed.
228    fn managed_catalog_global_and_ids<R: ManagedResource>(&self, ids: &HashSet<String>) -> Result<Vec<R>, StoreError> {
229        let sql = if ids.is_empty() {
230            format!(
231                "SELECT {cols} FROM {table} WHERE is_global = 1",
232                cols = R::COLUMNS.join(", "),
233                table = R::TABLE,
234            )
235        } else {
236            format!(
237                "SELECT {cols} FROM {table} WHERE is_global = 1 OR id IN ({ph})",
238                cols = R::COLUMNS.join(", "),
239                table = R::TABLE,
240                ph = placeholders(ids.len()),
241            )
242        };
243        let conn = self.conn.lock().unwrap();
244        let mut stmt = conn.prepare(&sql)?;
245        let bind: Vec<Value> = ids.iter().map(|s| Value::Text(s.clone())).collect();
246        let rows = stmt.query_map(params_from_iter(bind), |row| R::from_row(row))?;
247        let mut out = Vec::new();
248        for r in rows {
249            out.push(r?);
250        }
251        Ok(out)
252    }
253
254    // ── Public-to-crate managed-primitive operations ──────────────────────
255
256    /// Rows visible to `owner_id` — its own referenced rows plus every global
257    /// row — each paired with whether this owner holds the ref. Global rows
258    /// first, newest first.
259    pub(super) fn managed_list<R: ManagedResource>(
260        &self,
261        catalog: &Store,
262        owner: Owner,
263        owner_id: &str,
264    ) -> Result<Vec<(R, bool)>, StoreError> {
265        let bound_ids = self.managed_bound_ids::<R>(owner, owner_id)?;
266        let mut rows = catalog.managed_catalog_global_and_ids::<R>(&bound_ids)?;
267        // A global row this owner ALSO directly refs is still "bound" — the
268        // flag reflects ownership of the ref row, not visibility.
269        rows.sort_by(|a, b| b.is_global().cmp(&a.is_global()).then_with(|| b.updated_at().cmp(&a.updated_at())));
270        Ok(rows
271            .into_iter()
272            .map(|row| {
273                let bound = bound_ids.contains(row.id());
274                (row, bound)
275            })
276            .collect())
277    }
278
279    /// Global rows only, each paired with whether `agent_id` holds the
280    /// agent-level ref. Newest first. Never returns private rows — the
281    /// caller-supplied `agent_id` is unverified on the catalog RPCs that
282    /// use this (reagentx P0 on PR #2329).
283    pub(super) fn managed_list_global_for_agent<R: ManagedResource>(
284        &self,
285        catalog: &Store,
286        agent_id: &str,
287    ) -> Result<Vec<(R, bool)>, StoreError> {
288        let globals = catalog.managed_catalog_global::<R>()?;
289        let bound_ids = self.managed_bound_ids::<R>(Owner::Agent, agent_id)?;
290        Ok(globals
291            .into_iter()
292            .map(|row| {
293                let bound = bound_ids.contains(row.id());
294                (row, bound)
295            })
296            .collect())
297    }
298
299    /// Global rows with how many agents hold an agent-level ref to each —
300    /// the catalog's "used by N agents" in Connectors and Memory. Newest first.
301    pub(super) fn managed_list_global<R: ManagedResource>(&self, catalog: &Store) -> Result<Vec<(R, i64)>, StoreError> {
302        let globals = catalog.managed_catalog_global::<R>()?;
303        let counts = self.managed_bound_counts::<R>()?;
304        Ok(globals
305            .into_iter()
306            .map(|row| {
307                let count = counts.get(row.id()).copied().unwrap_or(0);
308                (row, count)
309            })
310            .collect())
311    }
312
313    /// One row by id. Genuinely single-table — callers invoke this directly
314    /// on whichever store owns `R::TABLE` (`catalog` for Skill/McpServer).
315    pub(super) fn managed_get<R: ManagedResource>(&self, id: &str) -> Result<Option<R>, StoreError> {
316        let sql = format!(
317            "SELECT {cols} FROM {table} WHERE id = ?1",
318            cols = R::COLUMNS.join(", "),
319            table = R::TABLE,
320        );
321        let conn = self.conn.lock().unwrap();
322        match conn.query_row(&sql, params![id], |row| R::from_row(row)) {
323            Ok(r) => Ok(Some(r)),
324            Err(rusqlite::Error::QueryReturnedNoRows) => Ok(None),
325            Err(e) => Err(StoreError::Sqlite(e)),
326        }
327    }
328
329    /// Delete the catalog row on `catalog` FIRST, then purge `id`'s ref rows
330    /// on `self`. True if a catalog row was deleted.
331    ///
332    /// Order matters (Codex P1, PR #3183): with the ref-purge running first,
333    /// a catalog-delete failure AFTER the ref purge already committed (disk
334    /// full, the identity store locked, any I/O error) stranded the row —
335    /// still present in `catalog`, but no ref points at it and it is not
336    /// necessarily global, so no access-check path can find it again, and
337    /// the ordinary delete RPC's own pre-check (`skill_is_bound_to`) now
338    /// rejects a retry too, since the ref that check depends on is already
339    /// gone. Catalog-first fixes the failure mode entirely: if the catalog
340    /// delete itself errors, `?` returns before the ref purge ever runs, so
341    /// the row keeps its ref(s) and stays exactly as accessible/retryable as
342    /// before the call. If the catalog delete SUCCEEDS (including "0 rows,
343    /// already gone") and the SUBSEQUENT ref purge then fails or a process
344    /// crashes between the two, the worst case is a dangling ref pointing at
345    /// an id the identity store no longer holds — the same already-accepted,
346    /// gracefully-degrading gap `SPEC_DURABLE_BINDINGS_2026_09_10.md` §5.4
347    /// documents for cross-channel dangling refs (`skill_get` on a missing
348    /// id returns `None`, which every read path already handles), not a new
349    /// failure mode.
350    pub(super) fn managed_delete<R: ManagedResource>(&self, catalog: &Store, id: &str) -> Result<bool, StoreError> {
351        let deleted = {
352            let conn = catalog.conn.lock().unwrap();
353            let rows = conn.execute(&format!("DELETE FROM {} WHERE id = ?1", R::TABLE), params![id])?;
354            rows > 0
355        };
356        {
357            let conn = self.conn.lock().unwrap();
358            conn.execute(&format!("DELETE FROM {} WHERE {} = ?1", R::AGENT_REF_TABLE, R::REF_COL), params![id])?;
359            conn.execute(&format!("DELETE FROM {} WHERE {} = ?1", R::BUNDLE_REF_TABLE, R::REF_COL), params![id])?;
360        }
361        Ok(deleted)
362    }
363
364    /// Purge every bundle-level ref for `bundle_id`, for one resource kind.
365    ///
366    /// The mirror of `managed_delete`'s ref purge, from the other side: that
367    /// one runs when a catalog row goes away, this one when a bundle does.
368    /// Neither can rely on foreign keys: the ref tables declare none to the
369    /// catalog rows or to `db_bundles`. The bundle side *cannot* have an
370    /// enforcing FK at all, since `db_bundles` lives in a different physical
371    /// database from the ref tables (see the `db_bundle_skills_ref` comment
372    /// in migrations.rs).
373    ///
374    /// Returns the number of refs removed.
375    pub(super) fn managed_unbind_all_for_bundle<R: ManagedResource>(
376        &self,
377        bundle_id: &str,
378    ) -> Result<usize, StoreError> {
379        let conn = self.conn.lock().unwrap();
380        let rows = conn.execute(
381            &format!(
382                "DELETE FROM {} WHERE {} = ?1",
383                R::BUNDLE_REF_TABLE,
384                Owner::Bundle.key_col()
385            ),
386            params![bundle_id],
387        )?;
388        Ok(rows)
389    }
390
391    /// Insert the agent-level ref (idempotent). Errors if `agent_id` is not
392    /// a LOCAL agent: the ref table's `agent_id → db_agents` FK would
393    /// otherwise swallow the `INSERT OR IGNORE` silently for a cross-channel
394    /// agent, which is indistinguishable by row count from "already bound"
395    /// (reagentx P1 on PR #2315; REPORT_ARMORY_SKILLS_MARKDOWN_AND_BIND_BUG_2026_07_27.md).
396    ///
397    /// Checks `db_agents`, not `db_agent_definitions`: `R::AGENT_REF_TABLE`'s
398    /// FK targets `db_agents` as of Phase 3c (#3088), so checking the legacy
399    /// table here would reject a bind for any agent that exists ONLY as a
400    /// `db_agents` row (a template launch), even though the FK the INSERT
401    /// below actually depends on would accept it.
402    ///
403    /// Additionally errors if `id` has no row in `catalog` — Part A of
404    /// SPEC_DURABLE_BINDINGS_2026_09_10.md §5.4 drops the ref table's
405    /// `skill_id`/`mcp_id → db_skills`/`db_mcp_servers` FK entirely (it used
406    /// to target THIS channel's local, now-stale catalog copy), so the
407    /// existence check that FK used to provide is now explicit here instead
408    /// — mirroring `managed_bind_bundle`'s existing `id_store` check.
409    pub(super) fn managed_bind_agent<R: ManagedResource>(
410        &self,
411        catalog: &Store,
412        agent_id: &str,
413        id: &str,
414    ) -> Result<(), StoreError> {
415        let agent_exists: bool = {
416            let conn = self.conn.lock().unwrap();
417            conn.query_row(
418                "SELECT EXISTS(SELECT 1 FROM db_agents WHERE id = ?1)",
419                params![agent_id],
420                |row| row.get(0),
421            )?
422        };
423        if !agent_exists {
424            return Err(StoreError::Other(format!(
425                "agent {agent_id} not found in this channel's local registry — cross-channel {} binding is not supported",
426                R::KIND
427            )));
428        }
429        if catalog.managed_get::<R>(id)?.is_none() {
430            return Err(StoreError::Other(format!(
431                "{} {id} not found — cannot bind it to agent {agent_id}",
432                R::NAME_NOUN
433            )));
434        }
435        let conn = self.conn.lock().unwrap();
436        conn.execute(
437            &format!(
438                "INSERT OR IGNORE INTO {} (agent_id, {}) VALUES (?1, ?2)",
439                R::AGENT_REF_TABLE,
440                R::REF_COL
441            ),
442            params![agent_id, id],
443        )?;
444        Ok(())
445    }
446
447    /// Insert the bundle-level ref (idempotent). Bundle existence is checked
448    /// in `id_store`, not `self` — bundles are authoritatively written through
449    /// the shared store, and `self`'s local `db_bundles` copy is essentially
450    /// always empty in production (reagentx P0 on PR #2639).
451    ///
452    /// Also checks `id` exists in `catalog` — same Part A FK-removal reason
453    /// as `managed_bind_agent`'s new check (`db_bundle_skills_ref` also lost
454    /// its `skill_id` FK, and never had a `bundle_id` FK at all — see
455    /// `migrations.rs`'s own comment on these tables). Added now for the same
456    /// reason as the agent-level check, even though `bundle_skill_bind`'s
457    /// current call sites already pre-check existence via `skill_get` before
458    /// calling — see this crate's report for why
459    /// this is still worth doing as a second line of defense at the store
460    /// layer rather than trusting every call site forever.
461    pub(super) fn managed_bind_bundle<R: ManagedResource>(
462        &self,
463        catalog: &Store,
464        id_store: &Store,
465        bundle_id: &str,
466        id: &str,
467    ) -> Result<(), StoreError> {
468        if !id_store.bundle_exists(bundle_id)? {
469            return Err(StoreError::Other(format!(
470                "bundle {bundle_id} not found — cannot bind {} to a nonexistent bundle",
471                R::ARTICLE_NOUN
472            )));
473        }
474        if catalog.managed_get::<R>(id)?.is_none() {
475            return Err(StoreError::Other(format!(
476                "{} {id} not found — cannot bind it to bundle {bundle_id}",
477                R::NAME_NOUN
478            )));
479        }
480        let conn = self.conn.lock().unwrap();
481        conn.execute(
482            &format!(
483                "INSERT OR IGNORE INTO {} (bundle_id, {}) VALUES (?1, ?2)",
484                R::BUNDLE_REF_TABLE,
485                R::REF_COL
486            ),
487            params![bundle_id, id],
488        )?;
489        Ok(())
490    }
491
492    fn bundle_exists(&self, bundle_id: &str) -> Result<bool, StoreError> {
493        let conn = self.conn.lock().unwrap();
494        Ok(conn.query_row(
495            "SELECT EXISTS(SELECT 1 FROM db_bundles WHERE id = ?1)",
496            params![bundle_id],
497            |row| row.get(0),
498        )?)
499    }
500
501    /// Remove the owner-level ref. True if a row was removed.
502    pub(super) fn managed_unbind<R: ManagedResource>(
503        &self,
504        owner: Owner,
505        owner_id: &str,
506        id: &str,
507    ) -> Result<bool, StoreError> {
508        let conn = self.conn.lock().unwrap();
509        let rows = conn.execute(
510            &format!(
511                "DELETE FROM {} WHERE {} = ?1 AND {} = ?2",
512                owner.ref_table::<R>(),
513                owner.key_col(),
514                R::REF_COL
515            ),
516            params![owner_id, id],
517        )?;
518        Ok(rows > 0)
519    }
520
521    /// Upsert `row` into `catalog` enforcing owner-scoped name uniqueness (no
522    /// other row visible to the owner — bound or global — may share the
523    /// name), and bind it into `self` when `bind_new`.
524    ///
525    /// **No longer atomic** (Phase 2 of SPEC_DURABLE_BINDINGS_2026_09_10.md
526    /// §5.4, accepted interim gap): the catalog insert and the ref-table bind
527    /// are two separate connections now, so a narrow same-name race between
528    /// two concurrent upserts for a brand-new name is possible (single-owner,
529    /// single-submit UI flows are not a realistic multi-writer scenario — not
530    /// closed until Phase 3/6 restores single-connection atomicity). What
531    /// this DOES still guarantee: if the ref-bind fails after the catalog
532    /// insert already committed, a best-effort COMPENSATING DELETE removes
533    /// the just-inserted catalog row before the bind error propagates, so the
534    /// caller's failure and the durable state agree ("nothing was created")
535    /// instead of leaving a permanent, undiscoverable orphan catalog row (no
536    /// ordinary list/access/delete path can find a row that is neither bound
537    /// nor global). The compensating delete's own failure is logged, never
538    /// allowed to mask the original bind error.
539    pub(super) fn managed_upsert_unique<R: ManagedResource>(
540        &self,
541        catalog: &Store,
542        owner: Owner,
543        owner_id: &str,
544        row: &R,
545        bind_new: bool,
546        force_global: Option<bool>,
547    ) -> Result<(), StoreError> {
548        let bound_ids = self.managed_bound_ids::<R>(owner, owner_id)?;
549
550        let dup_sql = if bound_ids.is_empty() {
551            format!(
552                "SELECT COUNT(*) FROM {table} WHERE name = ?1 AND id <> ?2 AND is_global = 1",
553                table = R::TABLE,
554            )
555        } else {
556            format!(
557                "SELECT COUNT(*) FROM {table} WHERE name = ?1 AND id <> ?2 AND (is_global = 1 OR id IN ({ph}))",
558                table = R::TABLE,
559                ph = placeholders_from(3, bound_ids.len()),
560            )
561        };
562        let dup: i64 = {
563            let conn = catalog.conn.lock().unwrap();
564            let mut bind: Vec<Value> = vec![Value::Text(row.name().to_string()), Value::Text(row.id().to_string())];
565            bind.extend(bound_ids.iter().map(|s| Value::Text(s.clone())));
566            conn.query_row(&dup_sql, params_from_iter(bind), |r| r.get(0))?
567        };
568        if dup > 0 {
569            return Err(StoreError::Other(format!(
570                "{} name '{}' already bound to this {}",
571                R::NAME_NOUN,
572                row.name(),
573                owner.word()
574            )));
575        }
576
577        {
578            let conn = catalog.conn.lock().unwrap();
579            conn.execute(&insert_sql::<R>(), params_from_iter(values_with_global(row, force_global)))?;
580        }
581
582        if bind_new {
583            let bind_result = {
584                let conn = self.conn.lock().unwrap();
585                conn.execute(
586                    &format!(
587                        "INSERT OR IGNORE INTO {} ({}, {}) VALUES (?1, ?2)",
588                        owner.ref_table::<R>(),
589                        owner.key_col(),
590                        R::REF_COL
591                    ),
592                    params![owner_id, row.id()],
593                )
594            };
595            if let Err(bind_err) = bind_result {
596                let conn = catalog.conn.lock().unwrap();
597                if let Err(compensate_err) =
598                    conn.execute(&format!("DELETE FROM {} WHERE id = ?1", R::TABLE), params![row.id()])
599                {
600                    tracing::warn!(
601                        id = row.id(),
602                        bind_error = %bind_err,
603                        compensate_error = %compensate_err,
604                        "managed_upsert_unique: compensating delete failed after a bind error — a durable orphan catalog row may remain"
605                    );
606                }
607                return Err(bind_err.into());
608            }
609        }
610        Ok(())
611    }
612
613    /// Bundle-scoped upsert: create a NEW, PRIVATE (never global) row bound
614    /// directly to a bundle, enforcing bundle-scoped name uniqueness. Bundle
615    /// existence is checked in `id_store` (see `managed_bind_bundle`).
616    /// `managed_bind_bundle` may only bind EXISTING global rows — binding an
617    /// existing private row would let one bundle "steal" read access to
618    /// whatever entity that row belongs to — but globals already reach every
619    /// agent, so this is how a bundle gets a genuinely bundle-specific row
620    /// that has never belonged to anyone else (reagentx P1 on PR #2639).
621    pub(super) fn managed_upsert_unique_for_bundle<R: ManagedResource>(
622        &self,
623        catalog: &Store,
624        id_store: &Store,
625        bundle_id: &str,
626        row: &R,
627        bind_new: bool,
628    ) -> Result<(), StoreError> {
629        if !id_store.bundle_exists(bundle_id)? {
630            return Err(StoreError::Other(format!(
631                "bundle {bundle_id} not found — cannot create {} for a nonexistent bundle",
632                R::ARTICLE_NOUN
633            )));
634        }
635        self.managed_upsert_unique(catalog, Owner::Bundle, bundle_id, row, bind_new, Some(false))
636    }
637
638    /// Atomically upsert a GLOBAL row enforcing catalog-wide name uniqueness
639    /// among every global row (not just those visible to one owner). Forces
640    /// `is_global = 1` on INSERT; the caller is expected to have set it too.
641    ///
642    /// Pure single-table operation — callers invoke this directly on
643    /// whichever store owns `R::TABLE` (`identity_store` in production).
644    pub(super) fn managed_upsert_unique_global<R: ManagedResource>(&self, row: &R) -> Result<(), StoreError> {
645        let mut conn = self.conn.lock().unwrap();
646        let tx = conn.transaction()?;
647        let dup: i64 = tx.query_row(
648            &format!(
649                "SELECT COUNT(*) FROM {} WHERE name = ?1 AND id <> ?2 AND is_global = 1",
650                R::TABLE
651            ),
652            params![row.name(), row.id()],
653            |r| r.get(0),
654        )?;
655        if dup > 0 {
656            return Err(StoreError::Other(format!(
657                "a global {} named '{}' already exists",
658                R::NAME_NOUN,
659                row.name()
660            )));
661        }
662        tx.execute(&insert_sql::<R>(), params_from_iter(values_with_global(row, Some(true))))?;
663        tx.commit()?;
664        Ok(())
665    }
666
667    /// True when the row is global (in `catalog`) or the owner holds a ref
668    /// to it (in `self`) — the read/mutation access check.
669    pub(super) fn managed_is_accessible_to<R: ManagedResource>(
670        &self,
671        catalog: &Store,
672        owner: Owner,
673        owner_id: &str,
674        id: &str,
675    ) -> Result<bool, StoreError> {
676        let is_global: Option<bool> = {
677            let conn = catalog.conn.lock().unwrap();
678            match conn.query_row(
679                &format!("SELECT is_global FROM {} WHERE id = ?1", R::TABLE),
680                params![id],
681                |r| r.get::<_, i64>(0),
682            ) {
683                Ok(v) => Some(v != 0),
684                Err(rusqlite::Error::QueryReturnedNoRows) => None,
685                Err(e) => return Err(e.into()),
686            }
687        };
688        match is_global {
689            None => Ok(false),
690            Some(true) => Ok(true),
691            Some(false) => {
692                let conn = self.conn.lock().unwrap();
693                let count: i64 = conn.query_row(
694                    &format!(
695                        "SELECT COUNT(*) FROM {} WHERE {} = ?1 AND {} = ?2",
696                        owner.ref_table::<R>(),
697                        owner.key_col(),
698                        R::REF_COL
699                    ),
700                    params![owner_id, id],
701                    |r| r.get(0),
702                )?;
703                Ok(count > 0)
704            }
705        }
706    }
707
708    /// True when the owner holds a direct ref to the row (global rows are
709    /// not implied) — the ownership check for edit/delete. Pure ref-table
710    /// operation on `self`.
711    pub(super) fn managed_is_bound_to<R: ManagedResource>(
712        &self,
713        owner: Owner,
714        owner_id: &str,
715        id: &str,
716    ) -> Result<bool, StoreError> {
717        let conn = self.conn.lock().unwrap();
718        let count: i64 = conn.query_row(
719            &format!(
720                "SELECT COUNT(*) FROM {} WHERE {} = ?1 AND {} = ?2",
721                owner.ref_table::<R>(),
722                owner.key_col(),
723                R::REF_COL
724            ),
725            params![owner_id, id],
726            |r| r.get(0),
727        )?;
728        Ok(count > 0)
729    }
730
731    /// Union the rows referenced by `agent_id`'s bundles into `visible`, in
732    /// list order (its own bundle first, then its picks: `agent_bundle_chain`),
733    /// deduped by id, so the first bundle to name a row wins and a global row
734    /// already in `visible` isn't added twice. Composable model v2 — without
735    /// this, bundle-level refs would be exactly as inert at launch as the
736    /// bundle's old inline JSON columns were (GH issue #2024 item 3). Silent
737    /// on lookup failure. Skills only now: a bundle carries no MCP servers
738    /// (`SPEC_BUNDLE_CONTENTS_MEMORY_NOT_MCP_2026_10_07.md` §3.1).
739    pub(super) fn managed_union_bundle_refs<R: ManagedResource>(
740        &self,
741        catalog: &Store,
742        agent_id: &str,
743        visible: &mut Vec<R>,
744    ) {
745        for bundle_id in self.agent_bundle_chain(agent_id) {
746            for (row, _) in self
747                .managed_list::<R>(catalog, Owner::Bundle, &bundle_id)
748                .unwrap_or_default()
749            {
750                if !visible.iter().any(|s| s.id() == row.id()) {
751                    visible.push(row);
752                }
753            }
754        }
755    }
756}
757
758fn insert_sql<R: ManagedResource>() -> String {
759    format!(
760        "INSERT INTO {table} ({cols})
761         VALUES ({vals})
762         ON CONFLICT(id) DO UPDATE SET {set}",
763        table = R::TABLE,
764        cols = R::COLUMNS.join(", "),
765        vals = placeholders(R::COLUMNS.len()),
766        set = update_set::<R>(),
767    )
768}
769
770#[cfg(test)]
771mod tests {
772    use super::*;
773
774    struct Probe;
775    impl ManagedResource for Probe {
776        const TABLE: &'static str = "db_probe";
777        const REF_COL: &'static str = "probe_id";
778        const AGENT_REF_TABLE: &'static str = "db_agent_probe_ref";
779        const BUNDLE_REF_TABLE: &'static str = "db_bundle_probe_ref";
780        const COLUMNS: &'static [&'static str] = &["id", "name", "is_global", "created_at", "updated_at"];
781        const UPDATE_COLUMNS: &'static [&'static str] = &["name", "updated_at"];
782        const NAME_NOUN: &'static str = "probe";
783        const ARTICLE_NOUN: &'static str = "a probe";
784        const KIND: &'static str = "probe";
785        fn from_row(_row: &Row<'_>) -> rusqlite::Result<Self> {
786            Ok(Probe)
787        }
788        fn values(&self) -> Vec<Value> {
789            vec![
790                Value::Text("id".into()),
791                Value::Text("n".into()),
792                Value::Integer(0),
793                Value::Integer(1),
794                Value::Integer(2),
795            ]
796        }
797        fn id(&self) -> &str {
798            "id"
799        }
800        fn name(&self) -> &str {
801            "n"
802        }
803        fn is_global(&self) -> bool {
804            false
805        }
806        fn updated_at(&self) -> i64 {
807            2
808        }
809    }
810
811    #[test]
812    fn insert_sql_lists_every_column_once_and_updates_only_the_editable_ones() {
813        let sql = insert_sql::<Probe>();
814        assert!(sql.contains("INSERT INTO db_probe (id, name, is_global, created_at, updated_at)"));
815        assert!(sql.contains("VALUES (?1, ?2, ?3, ?4, ?5)"));
816        assert!(sql.contains("ON CONFLICT(id) DO UPDATE SET name=excluded.name, updated_at=excluded.updated_at"));
817        assert!(!sql.contains("is_global=excluded"), "is_global must never be rewritten on conflict");
818    }
819
820    #[test]
821    fn force_global_replaces_only_the_is_global_value() {
822        let forced = values_with_global(&Probe, Some(true));
823        assert_eq!(forced[2], Value::Integer(1));
824        assert_eq!(forced[0], Value::Text("id".into()));
825        let cleared = values_with_global(&Probe, Some(false));
826        assert_eq!(cleared[2], Value::Integer(0));
827        assert_eq!(values_with_global(&Probe, None), Probe.values());
828    }
829
830    #[test]
831    fn owner_selects_the_matching_ref_table_and_key() {
832        assert_eq!(Owner::Agent.ref_table::<Probe>(), "db_agent_probe_ref");
833        assert_eq!(Owner::Bundle.ref_table::<Probe>(), "db_bundle_probe_ref");
834        assert_eq!(Owner::Agent.key_col(), "agent_id");
835        assert_eq!(Owner::Bundle.key_col(), "bundle_id");
836    }
837
838    #[test]
839    fn placeholders_from_continues_numbering_past_earlier_bound_params() {
840        assert_eq!(placeholders_from(3, 0), "");
841        assert_eq!(placeholders_from(3, 1), "?3");
842        assert_eq!(placeholders_from(3, 3), "?3, ?4, ?5");
843    }
844
845    // ── Cross-store integration tests (Phase 2 of SPEC_DURABLE_BINDINGS_2026_09_10.md) ──
846    //
847    // Use the REAL `Skill` primitive against two genuinely separate
848    // `Store::open_in_memory()` instances — one acting as the ref-owning
849    // `mstore`, one as `catalog` (with the identity schema applied) — so a
850    // bug that only shows up when the two stores are NOT the same connection
851    // (e.g. accidentally querying `self` for a catalog row, or vice versa)
852    // cannot hide the way it would if both roles resolved to one aliased
853    // store (as `test_state()` does elsewhere in this codebase).
854
855    use crate::backend::storage::skills::Skill;
856
857    fn mstore() -> Store {
858        Store::open_in_memory().unwrap()
859    }
860
861    fn catalog_store() -> Store {
862        let store = Store::open_in_memory().unwrap();
863        store.apply_identity_schema_for_tests().unwrap();
864        store
865    }
866
867    fn skill(id: &str, name: &str, is_global: bool, updated_at: i64) -> Skill {
868        Skill {
869            id: id.to_string(),
870            name: name.to_string(),
871            trigger: String::new(),
872            skill_type: "prompt".to_string(),
873            description: String::new(),
874            content: "c".to_string(),
875            is_global,
876            created_at: updated_at,
877            updated_at,
878        }
879    }
880
881    fn insert_agent(store: &Store, id: &str) {
882        store
883            .conn()
884            .lock()
885            .unwrap()
886            .execute("INSERT INTO db_agents (id, name, provider) VALUES (?1, 'A', 'claude')", params![id])
887            .unwrap();
888    }
889
890    fn insert_bundle(store: &Store, id: &str) {
891        store
892            .conn()
893            .lock()
894            .unwrap()
895            .execute(
896                "INSERT INTO db_bundles (id, name, created_at, updated_at) VALUES (?1, ?1, 0, 0)",
897                params![id],
898            )
899            .unwrap();
900    }
901
902    #[test]
903    fn managed_list_combines_global_and_bound_private_rows_from_two_distinct_stores() {
904        let mstore = mstore();
905        let catalog = catalog_store();
906        insert_agent(&mstore, "agent-1");
907        catalog.managed_upsert_unique_global(&skill("global-1", "Global", true, 200)).unwrap();
908        catalog.managed_upsert_unique_global(&skill("global-2", "Unrelated Global", true, 100)).unwrap();
909        // A private row that exists in the catalog but is NOT referenced by
910        // agent-1 — must not appear.
911        {
912            let conn = catalog.conn().lock().unwrap();
913            conn.execute(
914                "INSERT INTO db_skills (id, name, is_global, created_at, updated_at) VALUES ('private-other', 'Someone Else', 0, 50, 50)",
915                [],
916            )
917            .unwrap();
918            conn.execute(
919                "INSERT INTO db_skills (id, name, is_global, created_at, updated_at) VALUES ('private-mine', 'Mine', 0, 150, 150)",
920                [],
921            )
922            .unwrap();
923        }
924        mstore.skill_bind(&catalog, "agent-1", "private-mine").unwrap();
925
926        let list = mstore.managed_list::<Skill>(&catalog, Owner::Agent, "agent-1").unwrap();
927        let ids: Vec<&str> = list.iter().map(|(s, _)| s.id.as_str()).collect();
928        assert_eq!(ids, vec!["global-1", "global-2", "private-mine"], "globals first (newest first), then the bound private row");
929        assert!(!list.iter().any(|(s, _)| s.id == "private-other"), "an unbound private row from another owner must not appear");
930
931        let bound: std::collections::HashMap<&str, bool> = list.iter().map(|(s, b)| (s.id.as_str(), *b)).collect();
932        assert!(!bound["global-1"], "a global the owner has not directly bound is visible but not 'bound'");
933        assert!(bound["private-mine"]);
934    }
935
936    #[test]
937    fn managed_list_global_reports_bound_count_from_the_mstore_side() {
938        let mstore = mstore();
939        let catalog = catalog_store();
940        insert_agent(&mstore, "agent-1");
941        insert_agent(&mstore, "agent-2");
942        catalog.managed_upsert_unique_global(&skill("global-1", "Global", true, 100)).unwrap();
943        mstore.skill_bind(&catalog, "agent-1", "global-1").unwrap();
944        mstore.skill_bind(&catalog, "agent-2", "global-1").unwrap();
945
946        let list = mstore.managed_list_global::<Skill>(&catalog).unwrap();
947        assert_eq!(list.len(), 1);
948        assert_eq!(list[0].1, 2, "bound_count must come from mstore's ref table, not catalog");
949    }
950
951    #[test]
952    fn managed_list_global_for_agent_never_returns_private_rows() {
953        let mstore = mstore();
954        let catalog = catalog_store();
955        insert_agent(&mstore, "agent-1");
956        catalog.managed_upsert_unique_global(&skill("global-1", "Global", true, 100)).unwrap();
957        {
958            let conn = catalog.conn().lock().unwrap();
959            conn.execute(
960                "INSERT INTO db_skills (id, name, is_global, created_at, updated_at) VALUES ('private-1', 'Private', 0, 50, 50)",
961                [],
962            )
963            .unwrap();
964        }
965        mstore.skill_bind(&catalog, "agent-1", "global-1").unwrap();
966
967        let list = mstore.managed_list_global_for_agent::<Skill>(&catalog, "agent-1").unwrap();
968        assert_eq!(list.len(), 1);
969        assert_eq!(list[0].0.id, "global-1");
970        assert!(list[0].1);
971    }
972
973    #[test]
974    fn managed_is_accessible_to_checks_globality_in_catalog_and_binding_in_mstore() {
975        let mstore = mstore();
976        let catalog = catalog_store();
977        insert_agent(&mstore, "agent-1");
978        catalog.managed_upsert_unique_global(&skill("global-1", "Global", true, 100)).unwrap();
979        {
980            let conn = catalog.conn().lock().unwrap();
981            conn.execute(
982                "INSERT INTO db_skills (id, name, is_global, created_at, updated_at) VALUES ('private-1', 'Private', 0, 50, 50)",
983                [],
984            )
985            .unwrap();
986        }
987
988        assert!(mstore.managed_is_accessible_to::<Skill>(&catalog, Owner::Agent, "agent-1", "global-1").unwrap());
989        assert!(!mstore.managed_is_accessible_to::<Skill>(&catalog, Owner::Agent, "agent-1", "private-1").unwrap());
990        mstore.skill_bind(&catalog, "agent-1", "private-1").unwrap();
991        assert!(mstore.managed_is_accessible_to::<Skill>(&catalog, Owner::Agent, "agent-1", "private-1").unwrap());
992        assert!(!mstore.managed_is_accessible_to::<Skill>(&catalog, Owner::Agent, "agent-1", "no-such-id").unwrap());
993    }
994
995    #[test]
996    fn managed_upsert_unique_detects_a_duplicate_name_in_the_catalog_store() {
997        let mstore = mstore();
998        let catalog = catalog_store();
999        insert_agent(&mstore, "agent-1");
1000        catalog.managed_upsert_unique_global(&skill("global-1", "Deploy", true, 100)).unwrap();
1001
1002        let result = mstore.managed_upsert_unique(&catalog, Owner::Agent, "agent-1", &skill("new-1", "Deploy", false, 200), true, None);
1003        assert!(result.is_err(), "a name already used by a visible global row must be rejected");
1004        // The catalog must not have gained a row from the rejected attempt.
1005        assert!(catalog.managed_get::<Skill>("new-1").unwrap().is_none());
1006    }
1007
1008    #[test]
1009    fn managed_upsert_unique_compensates_by_deleting_the_catalog_row_when_the_bind_fails() {
1010        let mstore = mstore();
1011        let catalog = catalog_store();
1012        insert_agent(&mstore, "agent-1");
1013        // Force the ref-insert to fail deterministically: drop the ref table
1014        // out from under it (a stand-in for the "lock contention / disk
1015        // full / a channel-local constraint" failures the doc comment
1016        // describes — `INSERT OR IGNORE` swallows an agent_id FK violation
1017        // silently rather than erroring, so a missing db_agents row can't be
1018        // used to exercise this path; a missing TABLE is a real, unignored
1019        // SQL error).
1020        mstore.conn().lock().unwrap().execute_batch("DROP TABLE db_agent_skills_ref;").unwrap();
1021
1022        let result = mstore.managed_upsert_unique(&catalog, Owner::Agent, "agent-1", &skill("new-1", "Deploy", false, 200), true, None);
1023        assert!(result.is_err(), "the bind must fail now that its table is gone");
1024        assert!(
1025            catalog.managed_get::<Skill>("new-1").unwrap().is_none(),
1026            "the catalog insert must be compensated away, leaving no orphan row"
1027        );
1028    }
1029
1030    #[test]
1031    fn managed_upsert_unique_leaves_the_catalog_row_when_bind_new_is_false() {
1032        let mstore = mstore();
1033        let catalog = catalog_store();
1034        mstore.managed_upsert_unique(&catalog, Owner::Agent, "agent-1", &skill("new-1", "Deploy", false, 200), false, None).unwrap();
1035        assert!(catalog.managed_get::<Skill>("new-1").unwrap().is_some());
1036    }
1037
1038    #[test]
1039    fn managed_delete_purges_mstore_refs_and_removes_the_catalog_row() {
1040        let mstore = mstore();
1041        let catalog = catalog_store();
1042        insert_agent(&mstore, "agent-1");
1043        insert_bundle(&mstore, "bundle-1");
1044        catalog.managed_upsert_unique_global(&skill("global-1", "Global", true, 100)).unwrap();
1045        mstore.skill_bind(&catalog, "agent-1", "global-1").unwrap();
1046        mstore.bundle_skill_bind(&catalog, &mstore, "bundle-1", "global-1").unwrap();
1047
1048        let deleted = mstore.managed_delete::<Skill>(&catalog, "global-1").unwrap();
1049        assert!(deleted);
1050        assert!(catalog.managed_get::<Skill>("global-1").unwrap().is_none());
1051        assert!(!mstore.skill_is_bound_to("agent-1", "global-1").unwrap());
1052        assert!(!mstore.bundle_skill_is_bound_to("bundle-1", "global-1").unwrap());
1053    }
1054
1055    #[test]
1056    fn managed_bind_agent_succeeds_when_the_catalog_row_exists_in_the_separate_store() {
1057        let mstore = mstore();
1058        let catalog = catalog_store();
1059        insert_agent(&mstore, "agent-1");
1060        catalog.managed_upsert_unique_global(&skill("global-1", "Global", true, 100)).unwrap();
1061
1062        mstore.skill_bind(&catalog, "agent-1", "global-1").unwrap();
1063        assert!(mstore.skill_is_bound_to("agent-1", "global-1").unwrap());
1064    }
1065
1066    #[test]
1067    fn managed_bind_agent_rejects_a_catalog_row_that_does_not_exist_anywhere() {
1068        let mstore = mstore();
1069        let catalog = catalog_store();
1070        insert_agent(&mstore, "agent-1");
1071
1072        let result = mstore.skill_bind(&catalog, "agent-1", "no-such-skill");
1073        assert!(result.is_err(), "binding an id with no row in catalog must error, not silently no-op");
1074        assert!(!mstore.skill_is_bound_to("agent-1", "no-such-skill").unwrap());
1075    }
1076}
agentmux_srv/backend/storage/managed.rs

agentmux_srv/backend/storage/
managed.rs
