agentmux_srv/migrations/
m0036_drop_bundle_mcp.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Bundles stop carrying MCP servers: they belong to Connectors
5//! (`docs/specs/SPEC_BUNDLE_CONTENTS_MEMORY_NOT_MCP_2026_10_07.md` §3.1, D3).
6//!
7//! For this channel:
8//! - logs each agent that loses a server it reached only through a bundle,
9//!   naming the agent and the server, so the loss shows in the srv log;
10//! - deletes the servers private to a bundle from the identity store: not
11//!   global, referenced by a bundle here, and bound to no agent here. An
12//!   agent can't bind another entity's private server (`mcp.catalog.bind`
13//!   takes global ones only), so no other channel's agent holds one either;
14//! - deletes every `db_bundle_mcp_ref` row;
15//! - clears the shared store's inline `db_bundles.mcp_servers` to `[]`.
16//!
17//! An agent's own servers (`db_agent_mcp_ref`) and the global ones are left
18//! alone. `db_bundle_mcp_ref` stays defined, because an older build on the
19//! same channel expects it; dropping it is a later cleanup.
20//!
21//! Idempotent: a second run finds no refs and nothing inline. Fails (to be
22//! retried on the next boot) when the shared store can't be opened or the
23//! inline column can't be cleared, since those hold server configs. Never
24//! fails over the identity store: when it can't be opened, the private
25//! servers are left (unreachable, since their refs go) and that is logged.
26
27use std::collections::{BTreeMap, BTreeSet};
28use std::sync::Arc;
29
30use rusqlite::params;
31
32use crate::backend::storage::store::Store;
33use crate::registry;
34
35use super::{Migration, MigrationContext, MigrationError, MigrationScope};
36
37pub struct M0036DropBundleMcp;
38
39impl Migration for M0036DropBundleMcp {
40    fn id(&self) -> &'static str { "0036_drop_bundle_mcp" }
41    fn scope(&self) -> MigrationScope { MigrationScope::Channel }
42    fn description(&self) -> &'static str {
43        "Take MCP servers out of bundles; agents keep their own and the global ones"
44    }
45
46    fn up(&self, ctx: &MigrationContext) -> Result<(), MigrationError> {
47        if !ctx.channel_store_path.exists() {
48            return Ok(());
49        }
50        let mstore = Arc::new(
51            Store::open(&ctx.channel_store_path)
52                .map_err(|e| MigrationError(format!("drop_bundle_mcp: open mstore: {e}")))?,
53        );
54        // The shared store itself, not m0021's best-effort fallback to the
55        // channel store: clearing that (usually empty) local copy would record
56        // this as done with the shared configs still stored. Unavailable means
57        // fail, and retry on the next boot.
58        let bundle_store = Store::open_shared(&ctx.shared_store_path)
59            .map_err(|e| MigrationError(format!("drop_bundle_mcp: open shared store: {e}")))?;
60        let identity_store = match open_identity_store() {
61            Ok(store) => Some(store),
62            Err(e) => {
63                tracing::warn!(error = %e, "drop_bundle_mcp: identity store unavailable, bundle-private servers left in place");
64                None
65            }
66        };
67        drop_bundle_mcp(&mstore, identity_store.as_ref(), &bundle_store).map_err(MigrationError)
68    }
69}
70
71fn open_identity_store() -> Result<Store, String> {
72    let path = registry::resolve_identity_store_path()
73        .ok_or_else(|| "could not resolve identity store path".to_string())?;
74    if !path.exists() {
75        return Err(format!("{} does not exist", path.display()));
76    }
77    Store::open_identity_store(&path).map_err(|e| format!("open identity store: {e}"))
78}
79
80/// What one run removed, for the log and the tests.
81#[derive(Debug, Default, PartialEq, Eq)]
82struct Dropped {
83    refs: usize,
84    private_servers: usize,
85    inline_bundles: usize,
86}
87
88fn drop_bundle_mcp(mstore: &Store, identity_store: Option<&Store>, bundle_store: &Store) -> Result<(), String> {
89    let refs = bundle_mcp_refs(mstore).map_err(|e| format!("drop_bundle_mcp: read refs: {e}"))?;
90    let mut dropped = Dropped::default();
91
92    if !refs.is_empty() {
93        log_agents_losing_servers(mstore, identity_store, &refs);
94
95        if let Some(identity_store) = identity_store {
96            let agent_bound = agent_bound_mcp_ids(mstore).map_err(|e| format!("drop_bundle_mcp: read agent refs: {e}"))?;
97            let referenced: BTreeSet<&str> = refs.values().flatten().map(String::as_str).collect();
98            for id in referenced {
99                if agent_bound.contains(id) {
100                    continue;
101                }
102                let deleted = identity_store
103                    .conn()
104                    .lock()
105                    .unwrap()
106                    .execute("DELETE FROM db_mcp_servers WHERE id = ?1 AND is_global = 0", params![id])
107                    .map_err(|e| format!("drop_bundle_mcp: delete server {id}: {e}"))?;
108                dropped.private_servers += deleted;
109            }
110        }
111
112        dropped.refs = mstore
113            .conn()
114            .lock()
115            .unwrap()
116            .execute("DELETE FROM db_bundle_mcp_ref", [])
117            .map_err(|e| format!("drop_bundle_mcp: delete refs: {e}"))?;
118    }
119
120    // The inline column, which nothing has read since the ref tables became
121    // authoritative, but which still holds server configs, credentials
122    // included. A failure fails the migration, so it's retried on the next
123    // boot rather than recorded as done with the configs still stored.
124    dropped.inline_bundles = bundle_store
125        .conn()
126        .lock()
127        .unwrap()
128        .execute("UPDATE db_bundles SET mcp_servers = '[]' WHERE mcp_servers NOT IN ('', '[]')", [])
129        .map_err(|e| format!("drop_bundle_mcp: clear inline mcp_servers: {e}"))?;
130
131    tracing::info!(
132        refs = dropped.refs,
133        private_servers = dropped.private_servers,
134        inline_bundles = dropped.inline_bundles,
135        "drop_bundle_mcp: complete"
136    );
137    Ok(())
138}
139
140/// Every bundle's MCP refs on this channel, bundle id to server ids.
141fn bundle_mcp_refs(mstore: &Store) -> rusqlite::Result<BTreeMap<String, Vec<String>>> {
142    let conn = mstore.conn().lock().unwrap();
143    let mut stmt = conn.prepare("SELECT bundle_id, mcp_id FROM db_bundle_mcp_ref ORDER BY bundle_id, mcp_id")?;
144    let rows = stmt.query_map([], |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?)))?;
145    let mut out: BTreeMap<String, Vec<String>> = BTreeMap::new();
146    for row in rows {
147        let (bundle_id, mcp_id) = row?;
148        out.entry(bundle_id).or_default().push(mcp_id);
149    }
150    Ok(out)
151}
152
153fn agent_bound_mcp_ids(mstore: &Store) -> rusqlite::Result<BTreeSet<String>> {
154    let conn = mstore.conn().lock().unwrap();
155    let mut stmt = conn.prepare("SELECT DISTINCT mcp_id FROM db_agent_mcp_ref")?;
156    let ids = stmt.query_map([], |r| r.get::<_, String>(0))?;
157    ids.collect()
158}
159
160/// One log line per agent that reached a server only through its bundles: not
161/// a global one, and not one bound to the agent itself.
162fn log_agents_losing_servers(
163    mstore: &Store,
164    identity_store: Option<&Store>,
165    refs: &BTreeMap<String, Vec<String>>,
166) {
167    let Ok(defs) = mstore.agent_def_list() else {
168        return;
169    };
170    let server = |id: &str| identity_store.and_then(|s| s.mcp_server_get(id).ok().flatten());
171    for def in defs {
172        let own: BTreeSet<String> = {
173            let conn = mstore.conn().lock().unwrap();
174            let mut stmt = match conn.prepare("SELECT mcp_id FROM db_agent_mcp_ref WHERE agent_id = ?1") {
175                Ok(stmt) => stmt,
176                Err(_) => continue,
177            };
178            let ids = stmt.query_map(params![def.id], |r| r.get::<_, String>(0));
179            match ids {
180                Ok(ids) => ids.flatten().collect(),
181                Err(_) => continue,
182            }
183        };
184        let mut lost: BTreeSet<String> = BTreeSet::new();
185        for bundle_id in mstore.agent_bundle_chain(&def.id) {
186            for id in refs.get(&bundle_id).into_iter().flatten() {
187                if own.contains(id) {
188                    continue;
189                }
190                match server(id) {
191                    Some(s) if s.is_global => {}
192                    Some(s) => {
193                        lost.insert(s.name);
194                    }
195                    // Unknown without the identity store; name it by id.
196                    None => {
197                        lost.insert(id.clone());
198                    }
199                }
200            }
201        }
202        if !lost.is_empty() {
203            tracing::warn!(
204                agent_id = %def.id,
205                agent = %def.name,
206                servers = %lost.into_iter().collect::<Vec<_>>().join(", "),
207                "drop_bundle_mcp: agent no longer gets these MCP servers from its bundles; bind them in Connectors"
208            );
209        }
210    }
211}
212
213#[cfg(test)]
214mod tests {
215    use super::*;
216    use crate::backend::storage::agents::test_agent_def;
217    use crate::backend::storage::store::Bundle;
218    use crate::backend::storage::McpServer;
219
220    fn bundle(id: &str, inline_mcp: &str) -> Bundle {
221        Bundle {
222            id: id.to_string(),
223            name: id.to_string(),
224            description: String::new(),
225            is_blank: false,
226            is_global: false,
227            provider: String::new(),
228            model: String::new(),
229            instructions: String::new(),
230            instructions_by_provider: "{}".to_string(),
231            context_files: "[]".to_string(),
232            mcp_servers: inline_mcp.to_string(),
233            skills: "[]".to_string(),
234            sort_order: 0,
235            created_at: 0,
236            updated_at: 0,
237            is_system: false,
238        }
239    }
240
241    fn server(id: &str, is_global: bool) -> McpServer {
242        McpServer {
243            id: id.to_string(),
244            name: format!("name-{id}"),
245            transport: "stdio".to_string(),
246            config: "{}".to_string(),
247            is_global,
248            created_at: 1,
249            updated_at: 1,
250        }
251    }
252
253    fn exec(store: &Store, sql: &str, args: &[&str]) {
254        store.conn().lock().unwrap().execute(sql, rusqlite::params_from_iter(args)).unwrap();
255    }
256
257    fn count(store: &Store, sql: &str) -> i64 {
258        store.conn().lock().unwrap().query_row(sql, [], |r| r.get(0)).unwrap()
259    }
260
261    #[test]
262    fn bundle_refs_and_bundle_private_servers_go_and_everything_else_stays() {
263        let mstore = Store::open_in_memory().unwrap();
264        let identity = Store::open_in_memory().unwrap();
265        let shared = Store::open_in_memory().unwrap();
266
267        shared.bundle_upsert(&bundle("b1", "[]")).unwrap();
268        // What an older build left inline; bundle_upsert itself always stores "[]".
269        exec(&shared, r#"UPDATE db_bundles SET mcp_servers = '[{"name":"inline","env":{"T":"x"}}]' WHERE id = 'b1'"#, &[]);
270        let mut def = test_agent_def("agent-1", "Agent", "claude", "agent", 1, "");
271        def.memory_id = "b1".to_string();
272        mstore.agent_def_insert(&mut def).unwrap();
273
274        // b1 refs a private server, a global one, and a private one agent-1 also binds itself.
275        for (id, global) in [("private", false), ("global", true), ("shared-with-agent", false), ("agent-own", false)] {
276            identity.mcp_server_insert_raw(&server(id, global)).unwrap();
277        }
278        for id in ["private", "global", "shared-with-agent"] {
279            exec(&mstore, "INSERT INTO db_bundle_mcp_ref (bundle_id, mcp_id) VALUES ('b1', ?1)", &[id]);
280        }
281        for id in ["shared-with-agent", "agent-own"] {
282            exec(&mstore, "INSERT INTO db_agent_mcp_ref (agent_id, mcp_id) VALUES ('agent-1', ?1)", &[id]);
283        }
284
285        drop_bundle_mcp(&mstore, Some(&identity), &shared).unwrap();
286        // Idempotent.
287        drop_bundle_mcp(&mstore, Some(&identity), &shared).unwrap();
288
289        assert_eq!(count(&mstore, "SELECT COUNT(*) FROM db_bundle_mcp_ref"), 0);
290        assert_eq!(count(&mstore, "SELECT COUNT(*) FROM db_agent_mcp_ref"), 2, "the agent's own binds stay");
291        assert!(identity.mcp_server_get("private").unwrap().is_none(), "a bundle-private server goes");
292        for kept in ["global", "shared-with-agent", "agent-own"] {
293            assert!(identity.mcp_server_get(kept).unwrap().is_some(), "{kept} stays");
294        }
295        assert_eq!(count(&shared, "SELECT COUNT(*) FROM db_bundles WHERE mcp_servers != '[]'"), 0, "the inline column is cleared");
296    }
297
298    #[test]
299    fn without_the_identity_store_the_refs_still_go() {
300        let mstore = Store::open_in_memory().unwrap();
301        let shared = Store::open_in_memory().unwrap();
302        exec(&mstore, "INSERT INTO db_bundle_mcp_ref (bundle_id, mcp_id) VALUES ('b1', 'x')", &[]);
303        drop_bundle_mcp(&mstore, None, &shared).unwrap();
304        assert_eq!(count(&mstore, "SELECT COUNT(*) FROM db_bundle_mcp_ref"), 0);
305    }
306
307    /// A failed clear of the inline column fails the run, so the migration
308    /// isn't recorded as applied with server configs still stored.
309    #[test]
310    fn a_failed_inline_clear_fails_the_migration() {
311        let mstore = Store::open_in_memory().unwrap();
312        let shared = Store::open_in_memory().unwrap();
313        exec(&shared, "DROP TABLE db_bundles", &[]);
314        let err = drop_bundle_mcp(&mstore, None, &shared).unwrap_err();
315        assert!(err.contains("clear inline mcp_servers"), "{err}");
316    }
317
318    /// No fallback to the channel store's local copy: an unopenable shared
319    /// store fails the run, to be retried.
320    #[test]
321    fn an_unopenable_shared_store_fails_the_migration() {
322        let dir = tempfile::tempdir().unwrap();
323        let channel = dir.path().join("objects.db");
324        Store::open(&channel).unwrap();
325        let ctx = MigrationContext {
326            home: std::env::temp_dir(),
327            data_dir: std::env::temp_dir(),
328            // A directory, which SQLite can't open as a database.
329            shared_store_path: dir.path().to_path_buf(),
330            channel_store_path: channel,
331        };
332        let err = M0036DropBundleMcp.up(&ctx).unwrap_err();
333        assert!(err.0.contains("open shared store"), "{err}");
334    }
335
336    #[test]
337    fn a_missing_channel_store_is_a_no_op() {
338        let dir = tempfile::tempdir().unwrap();
339        let ctx = MigrationContext {
340            home: std::env::temp_dir(),
341            data_dir: std::env::temp_dir(),
342            shared_store_path: dir.path().join("shared.db"),
343            channel_store_path: dir.path().join("absent.db"),
344        };
345        M0036DropBundleMcp.up(&ctx).unwrap();
346    }
347}
agentmux_srv/migrations/m0036_drop_bundle_mcp.rs

agentmux_srv/migrations/
m0036_drop_bundle_mcp.rs
