fix: bring back an MCP connector whose process died
Nightly Build / build (push) Successful in 7m46s

A stdio connector *is* its child process, and nothing noticed when that
process went away. The handle stayed in the manager's map, so every later
tool call answered `MCP '<name>' disconnected: process exited with 139`,
and the connector's own background work stopped for good — until the user
happened to log in again.

The second half is the quiet one. A per-user connector is typically the
one that *pushes*: Gmail's poll thread produces the `event/new_email`
notifications that feed event triage. After a crash those simply stop,
with no call to fail and nothing in the UI to say so.

So a death is now reconciled, on the same terms as every other
reconciliation here: best-effort, bounded, settling at the next login if
it fails. `McpServerClient::is_alive` makes the death observable (the
read-loop clears the flag before failing the pending calls, so a caller
woken by the disconnect error finds a handle that admits it is dead), the
manager remembers the spec each server was started from, and one seam —
`restart_if_dead` — is driven from two places:

  - `call()`, which repairs the connector in time for the call that
    noticed it, so a crash costs one restart rather than a dead session;
  - a 10s sweep, which is the only thing that can bring back a connector
    nobody is calling.

The restart policy is a pure function so it can be tested without a DB
pool and a runtime. Backoff is enforced as a time gate, never a sleep: a
tool call that finds the gate shut fails immediately instead of parking a
waiting user behind a crash-loop, and the sweep retries later. Five
consecutive failures stop the attempts, and the reset window doubles as
the escape hatch — a box left running recovers from a transient outage
instead of staying dark.

`stop_server`/`stop_all` forget the spec, which is what keeps a stop a
stop: without it the sweep would resurrect a connector an admin had just
revoked, and a container remount would respawn into the container that
was being replaced.
This commit is contained in:
2026-08-07 12:27:20 +01:00
parent 31b4c76f51
commit c1177a934d
5 changed files with 346 additions and 0 deletions
+12
View File
@@ -273,6 +273,18 @@ pub enum McpCallResult {
pub trait McpServerClient: Send + Sync { pub trait McpServerClient: Send + Sync {
fn tools(&self) -> &[McpTool]; fn tools(&self) -> &[McpTool];
async fn call_tool(&self, name: &str, args: Value) -> anyhow::Result<McpCallResult>; async fn call_tool(&self, name: &str, args: Value) -> anyhow::Result<McpCallResult>;
/// Whether this connection is still usable.
///
/// A stdio server *is* its child process: once that exits, the handle stays in
/// the manager's map but every call on it fails with a disconnect error, so
/// something has to be able to ask. The default is `true` for HTTP/SSE, which
/// holds no process and no long-lived connection — a dead remote surfaces per
/// call, and answering `false` here would make the manager "restart" a server
/// that was never running.
fn is_alive(&self) -> bool {
true
}
} }
// ── Helpers ─────────────────────────────────────────────────────────────────── // ── Helpers ───────────────────────────────────────────────────────────────────
+12
View File
@@ -221,6 +221,10 @@ pub struct McpServer {
/// Capabilities the server advertised in its `InitializeResult`. Captured so a /// Capabilities the server advertised in its `InitializeResult`. Captured so a
/// future Tasks polling loop can gate on `tasks` support; unused for now. /// future Tasks polling loop can gate on `tasks` support; unused for now.
server_capabilities: Value, server_capabilities: Value,
/// Cleared by the read-loop the moment the child process is gone, so the
/// manager can tell "this handle is dead" from "this call failed". Shared with
/// that task, which is the only writer.
alive: Arc<std::sync::atomic::AtomicBool>,
} }
impl McpServer { impl McpServer {
@@ -324,6 +328,8 @@ impl McpServer {
Arc::new(Mutex::new(HashMap::new())); Arc::new(Mutex::new(HashMap::new()));
let pending_elicitations = Arc::new(AtomicUsize::new(0)); let pending_elicitations = Arc::new(AtomicUsize::new(0));
let alive = Arc::new(std::sync::atomic::AtomicBool::new(true));
let alive_bg = Arc::clone(&alive);
let pending_bg = pending.clone(); let pending_bg = pending.clone();
let server_name_bg = cfg.name.clone(); let server_name_bg = cfg.name.clone();
let notification_tx_bg = notification_tx; let notification_tx_bg = notification_tx;
@@ -381,6 +387,10 @@ impl McpServer {
), ),
_ => "process exited unexpectedly".into(), _ => "process exited unexpectedly".into(),
}; };
// Publish the death *before* failing the pending calls: a caller woken
// by the error below must find `is_alive() == false`, or it would
// conclude the call failed on a healthy server and not restart it.
alive_bg.store(false, Ordering::SeqCst);
let error_msg = format!("MCP '{}' disconnected: {exit_info}", server_name_bg); let error_msg = format!("MCP '{}' disconnected: {exit_info}", server_name_bg);
if let Some(tx) = &log_tx_bg { if let Some(tx) = &log_tx_bg {
let _ = tx.send(McpLogLine::lifecycle(server_name_bg.clone(), format!("disconnected: {exit_info}"))); let _ = tx.send(McpLogLine::lifecycle(server_name_bg.clone(), format!("disconnected: {exit_info}")));
@@ -401,6 +411,7 @@ impl McpServer {
tools: Vec::new(), tools: Vec::new(),
pending_elicitations, pending_elicitations,
server_capabilities: json!({}), server_capabilities: json!({}),
alive,
}; };
let init = server.request("initialize", json!({ let init = server.request("initialize", json!({
@@ -632,4 +643,5 @@ impl McpServer {
impl McpServerClient for McpServer { impl McpServerClient for McpServer {
fn tools(&self) -> &[McpTool] { self.tools() } fn tools(&self) -> &[McpTool] { self.tools() }
async fn call_tool(&self, name: &str, args: Value) -> Result<McpCallResult> { self.call_tool(name, args).await } async fn call_tool(&self, name: &str, args: Value) -> Result<McpCallResult> { self.call_tool(name, args).await }
fn is_alive(&self) -> bool { self.alive.load(Ordering::SeqCst) }
} }
+312
View File
@@ -37,6 +37,87 @@ pub use verify::{VerifyReport, VerifyTarget, apply_placeholders, run_verify};
const SERVER_START_TIMEOUT_SECS: u64 = 120; const SERVER_START_TIMEOUT_SECS: u64 = 120;
// ── Respawn policy ───────────────────────────────────────────────────────────
//
// A stdio connector *is* its child process, and that process can die under us —
// not only from a bug in the connector (the Gmail one segfaulted mid-call from a
// thread race on its HTTP connection) but from an OOM kill or a container blip.
// Nothing used to notice: the dead handle stayed in `servers`, every tool call on
// it answered "disconnected", and the connector's own background work (Gmail's
// new-mail polling) was silently over until the user next logged in.
//
// So a death is now reconciled, on the same terms as every other reconciliation
// in this codebase: best-effort, bounded, and settling at the next login if it
// fails. Two triggers share one seam (`restart_if_dead`) — a tool call, which
// repairs the connector in time for the call that noticed, and a periodic sweep,
// which is what gets a push-only connector back without anyone asking.
/// How often the sweep looks for a dead server. Sets the worst-case silence for a
/// connector whose only job is pushing notifications.
const RESPAWN_SWEEP_SECS: u64 = 10;
/// Consecutive restarts before the manager stops trying. A connector that dies
/// this many times in a row is broken in a way restarting does not fix, and the
/// spawn itself is not free (a container `docker exec`, an interpreter start).
const MAX_RESPAWN_ATTEMPTS: u32 = 5;
/// Base backoff between consecutive restarts; doubles per attempt up to
/// [`RESPAWN_BACKOFF_MAX_SECS`]. Enforced as a *time gate*, never as a sleep — a
/// tool call that finds the gate closed fails immediately rather than blocking a
/// user behind a crash-loop, and the sweep picks it up on a later tick.
const RESPAWN_BACKOFF_SECS: u64 = 2;
const RESPAWN_BACKOFF_MAX_SECS: u64 = 60;
/// Quiet period after which a server's attempt counter resets. Doubles as the
/// ceiling's escape hatch: a connector that exhausted its attempts is retried once
/// more after this long, so a box left running for weeks recovers from a transient
/// outage (a container recreated, a network gone and returned) instead of staying
/// dark until someone logs in again.
const RESPAWN_RESET_SECS: u64 = 300;
/// Per-server restart bookkeeping. Deliberately keyed on the last *attempt*, not on
/// uptime: what must be rate-limited is how often we spawn, and a server that dies
/// instantly on every try would otherwise never accumulate the uptime to be judged.
#[derive(Debug, Clone, Copy)]
struct RespawnState {
attempts: u32,
last_attempt: std::time::Instant,
}
/// How long to wait after the n-th consecutive restart before trying again.
fn respawn_backoff(attempts: u32) -> Duration {
let factor = 1u64.checked_shl(attempts.min(16)).unwrap_or(u64::MAX);
Duration::from_secs(
RESPAWN_BACKOFF_SECS.saturating_mul(factor).min(RESPAWN_BACKOFF_MAX_SECS)
)
}
/// The restart policy, as a pure decision: given what happened last time, may we
/// spawn now — and if so, which attempt is this?
///
/// `None` refuses. `Some(n)` allows and reports the attempt number to record.
/// Split out from [`McpManager::claim_respawn`] because the policy is the subtle
/// part (notably: the reset window doubles as the ceiling's escape hatch, so
/// "gave up" is never permanent), while a manager needs a DB pool and a running
/// tokio runtime just to be constructed.
fn respawn_decision(prev: Option<RespawnState>, now: std::time::Instant) -> Option<u32> {
let Some(st) = prev else {
return Some(1); // never tried before
};
let quiet = now.saturating_duration_since(st.last_attempt);
if quiet >= Duration::from_secs(RESPAWN_RESET_SECS) {
return Some(1); // long enough since we last tried: forgive the history
}
if st.attempts >= MAX_RESPAWN_ATTEMPTS {
return None; // crash-looping — wait out the reset window
}
if quiet < respawn_backoff(st.attempts) {
return None; // too soon after the last try
}
Some(st.attempts + 1)
}
// ── McpManager ─────────────────────────────────────────────────────────────── // ── McpManager ───────────────────────────────────────────────────────────────
pub struct McpManager { pub struct McpManager {
@@ -57,6 +138,21 @@ pub struct McpManager {
elicitation_handler: RwLock<Option<Arc<dyn ElicitationHandler>>>, elicitation_handler: RwLock<Option<Arc<dyn ElicitationHandler>>>,
/// Data root for persisting non-text tool-result media (`media_dir`). /// Data root for persisting non-text tool-result media (`media_dir`).
data_root: PathBuf, data_root: PathBuf,
/// The spec each running server was started from, kept so a dead one can be
/// respawned without going back to the DB — which the manager could not do
/// anyway for a per-user connector, whose row needs the OAuth credential
/// re-resolved against the registry (`user_row_spec_resolved`).
///
/// A spec carries the connector's live credential in `config.env` (an OAuth
/// refresh token, an API key). That is the same RAM the child process already
/// holds it in, on a per-user manager whose whole lifetime is one unlocked
/// session (§9) — it neither widens the blast radius nor outlives the DEK.
specs: RwLock<HashMap<String, McpServerSpec>>,
/// Restart bookkeeping per server. Empty until something actually dies.
respawns: RwLock<HashMap<String, RespawnState>>,
/// Serializes respawns across the whole manager. Async because a respawn spans
/// a process spawn and an `initialize` round-trip.
respawn_lock: tokio::sync::Mutex<()>,
} }
/// Whether a runtime's server-pushed notifications are persisted to `mcp_events`. /// Whether a runtime's server-pushed notifications are persisted to `mcp_events`.
@@ -115,6 +211,9 @@ impl McpManager {
log_tx, log_tx,
elicitation_handler: RwLock::new(None), elicitation_handler: RwLock::new(None),
data_root: data_root.into(), data_root: data_root.into(),
specs: RwLock::new(HashMap::new()),
respawns: RwLock::new(HashMap::new()),
respawn_lock: tokio::sync::Mutex::new(()),
} }
} }
@@ -222,9 +321,14 @@ impl McpManager {
{ {
let mut descs = self.descriptions.write().unwrap(); let mut descs = self.descriptions.write().unwrap();
let mut titles = self.titles.write().unwrap(); let mut titles = self.titles.write().unwrap();
let mut kept = self.specs.write().unwrap();
for spec in &specs { for spec in &specs {
descs.insert(spec.config.name.clone(), spec.description.clone()); descs.insert(spec.config.name.clone(), spec.description.clone());
titles.insert(spec.config.name.clone(), spec.tool_titles.clone()); titles.insert(spec.config.name.clone(), spec.tool_titles.clone());
// Remembered even if the start below fails: a server that never
// came up is exactly one the sweep should keep trying, under the
// same ceiling as one that came up and died.
kept.insert(spec.config.name.clone(), spec.clone());
} }
} }
if boot { if boot {
@@ -283,6 +387,7 @@ impl McpManager {
/// API) — this only touches the live connections. Returns the tool names. /// API) — this only touches the live connections. Returns the tool names.
pub async fn start_server(&self, spec: McpServerSpec) -> Result<Vec<String>> { pub async fn start_server(&self, spec: McpServerSpec) -> Result<Vec<String>> {
let name = spec.config.name.clone(); let name = spec.config.name.clone();
self.specs.write().unwrap().insert(name.clone(), spec.clone());
let client = tokio::time::timeout( let client = tokio::time::timeout(
Duration::from_secs(SERVER_START_TIMEOUT_SECS), Duration::from_secs(SERVER_START_TIMEOUT_SECS),
Self::start_one(&spec.config, Some(self.notification_tx.clone()), Some(self.log_tx.clone()), self.elicitation_handler()), Self::start_one(&spec.config, Some(self.notification_tx.clone()), Some(self.log_tx.clone()), self.elicitation_handler()),
@@ -312,6 +417,10 @@ impl McpManager {
self.errors.write().unwrap().remove(name); self.errors.write().unwrap().remove(name);
self.descriptions.write().unwrap().remove(name); self.descriptions.write().unwrap().remove(name);
self.titles.write().unwrap().remove(name); self.titles.write().unwrap().remove(name);
// Forgetting the spec is what makes a stop a stop: leaving it would let the
// sweep resurrect a connector the admin just revoked or deactivated.
self.specs.write().unwrap().remove(name);
self.respawns.write().unwrap().remove(name);
} }
/// Stops **every** running server (each dropped client → `kill_on_drop` kills /// Stops **every** running server (each dropped client → `kill_on_drop` kills
@@ -324,6 +433,12 @@ impl McpManager {
self.errors.write().unwrap().clear(); self.errors.write().unwrap().clear();
self.descriptions.write().unwrap().clear(); self.descriptions.write().unwrap().clear();
self.titles.write().unwrap().clear(); self.titles.write().unwrap().clear();
// Same reason as `stop_server`, and load-bearing for the container remount
// this is called from: the specs name a container that is being replaced, so
// respawning against them would spawn into the one that just went away.
// `connect_all` re-populates with specs built for the new container.
self.specs.write().unwrap().clear();
self.respawns.write().unwrap().clear();
} }
/// Whether a server by this name currently has a live connection in the /// Whether a server by this name currently has a live connection in the
@@ -333,6 +448,134 @@ impl McpManager {
self.servers.read().unwrap().contains_key(name) self.servers.read().unwrap().contains_key(name)
} }
// ── Respawn ──────────────────────────────────────────────────────────────
/// Whether `name` is a server we are supposed to be running but currently are
/// not: it has a remembered spec, and either no handle at all (a start that
/// failed) or a handle whose child process has exited.
///
/// A server with no spec is never "dead" — it was stopped on purpose, or is a
/// handle registered by some path that does not want supervision.
fn is_dead(&self, name: &str) -> bool {
let supervised = self.specs.read().unwrap().contains_key(name);
if !supervised {
return false;
}
match self.servers.read().unwrap().get(name) {
Some(s) => !s.is_alive(),
None => true,
}
}
/// The supervised servers that are currently down.
fn dead_servers(&self) -> Vec<String> {
// Names are collected before the `servers` lock is taken so the two guards
// never overlap — every other reader here takes them in this order too.
let names: Vec<String> = self.specs.read().unwrap().keys().cloned().collect();
let servers = self.servers.read().unwrap();
names.into_iter()
.filter(|n| servers.get(n).map(|s| !s.is_alive()).unwrap_or(true))
.collect()
}
/// Takes one unit of restart budget for `name`, or refuses.
///
/// Refusing never sleeps — see [`RESPAWN_BACKOFF_SECS`]. The caller either has a
/// user waiting (fail now, honestly) or is the sweep (come back in a few
/// seconds), and neither is improved by parking a task on a timer.
fn claim_respawn(&self, name: &str) -> bool {
let now = std::time::Instant::now();
let mut states = self.respawns.write().unwrap();
match respawn_decision(states.get(name).copied(), now) {
Some(attempts) => {
states.insert(name.to_string(), RespawnState { attempts, last_attempt: now });
true
}
None => false,
}
}
/// Restarts `name` if its process is gone. Returns whether it is alive on exit,
/// so a caller about to use the server knows whether the repair worked.
///
/// Best-effort by contract: a refusal (budget spent, spec forgotten, spawn
/// failed) leaves the manager exactly as it was and the caller's own error path
/// takes over. Nothing here is on an authorization path — the spec was already
/// access-checked when it was built, and `stop_server` drops it, so a revoked
/// connector has nothing to respawn from.
pub async fn restart_if_dead(&self, name: &str) -> bool {
if !self.is_dead(name) {
return true;
}
// One respawn at a time. A parallel tool batch can find the same server
// dead in several calls at once; without this each would spawn a child and
// all but the last would be orphaned in the map, still running.
let _guard = self.respawn_lock.lock().await;
// Re-check under the lock: whoever held it may have just fixed this one.
if !self.is_dead(name) {
return true;
}
if !self.claim_respawn(name) {
return false;
}
let spec = self.specs.read().unwrap().get(name).cloned();
let Some(spec) = spec else { return false };
warn!("MCP server '{name}' is not running — restarting it");
self.log_lifecycle(name, "process gone — restarting");
// Drop the dead handle first so its `kill_on_drop` reaps anything left of
// the old child before a new one claims the same connector directory.
self.servers.write().unwrap().remove(name);
match self.start_server(spec).await {
Ok(tools) => {
info!("MCP server '{name}' restarted — {} tool(s)", tools.len());
true
}
Err(e) => {
warn!("MCP server '{name}' restart failed: {e}");
self.errors.write().unwrap().insert(name.to_string(), e.to_string());
false
}
}
}
/// Starts the background sweep that restarts servers which died while nobody
/// was calling them.
///
/// Separate from a call-time repair because the two answer different failures.
/// A tool call repairs the connector it was about to use, which is enough for a
/// connector that only ever acts when asked. It is not enough for one that
/// *pushes* — Gmail's poll thread produces the `event/new_email` notifications
/// that feed event triage, and after a crash those simply stop, with no call to
/// notice and nothing in the UI to say so. The sweep is what bounds that
/// silence to [`RESPAWN_SWEEP_SECS`].
///
/// Holds a `Weak`: a per-user manager dies with its `UserContext` at logout, and
/// an `Arc` here would keep that runtime — and its `docker exec` children —
/// alive past the moment the user's key left RAM (§9).
pub fn spawn_respawn_sweep(
self: &Arc<Self>,
shutdown: CancellationToken,
) -> tokio::task::JoinHandle<()> {
let weak = Arc::downgrade(self);
tokio::spawn(async move {
let mut tick = tokio::time::interval(Duration::from_secs(RESPAWN_SWEEP_SECS));
tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
loop {
tokio::select! {
_ = shutdown.cancelled() => break,
_ = tick.tick() => {
let Some(mgr) = weak.upgrade() else { break };
for name in mgr.dead_servers() {
mgr.restart_if_dead(&name).await;
}
}
}
}
})
}
pub fn tools(&self) -> Vec<McpTool> { pub fn tools(&self) -> Vec<McpTool> {
self.servers.read().unwrap().values() self.servers.read().unwrap().values()
.flat_map(|s| s.tools().iter().cloned()) .flat_map(|s| s.tools().iter().cloned())
@@ -375,6 +618,11 @@ impl McpManager {
} }
pub async fn call(&self, server: &str, tool: &str, args: Value) -> Result<ToolResult> { pub async fn call(&self, server: &str, tool: &str, args: Value) -> Result<ToolResult> {
// Repair a connector that died since the last call, before using it. This is
// the path the user actually feels: a crashed server used to turn every
// subsequent tool call into "MCP '<name>' disconnected" until the next
// login. A no-op (and one map read) whenever the server is healthy.
self.restart_if_dead(server).await;
let s = self.servers.read().unwrap() let s = self.servers.read().unwrap()
.get(server) .get(server)
.cloned() .cloned()
@@ -463,6 +711,7 @@ impl McpManager {
/// "Available MCP servers" prompt section. Decouples [`McpManager`] from any DB /// "Available MCP servers" prompt section. Decouples [`McpManager`] from any DB
/// table — the global and per-user runtimes each build these from their own rows /// table — the global and per-user runtimes each build these from their own rows
/// (`global_row_spec` / `user_row_spec`). /// (`global_row_spec` / `user_row_spec`).
#[derive(Clone)]
pub struct McpServerSpec { pub struct McpServerSpec {
pub config: McpServerConfig, pub config: McpServerConfig,
pub description: Option<String>, pub description: Option<String>,
@@ -900,4 +1149,67 @@ mod tests {
assert_eq!(s, "a=VAL&b=VAL&c=CC"); assert_eq!(s, "a=VAL&b=VAL&c=CC");
assert!(!spent, "env satisfied the tokens, so api_key was not spent"); assert!(!spent, "env satisfied the tokens, so api_key was not spent");
} }
// ── Respawn policy ───────────────────────────────────────────────────────
use std::time::Instant;
/// A state whose last attempt was `secs` ago, as seen from `now`.
fn attempted(attempts: u32, secs: u64, now: Instant) -> Option<RespawnState> {
Some(RespawnState {
attempts,
last_attempt: now.checked_sub(Duration::from_secs(secs)).expect("test clock underflow"),
})
}
#[test]
fn first_death_restarts_immediately() {
// The common case, and the one the user feels: a connector that has been up
// for days segfaults once. There is nothing to back off from yet.
let now = Instant::now();
assert_eq!(respawn_decision(None, now), Some(1));
}
#[test]
fn a_second_death_waits_out_the_backoff() {
let now = Instant::now();
// One second after attempt #1 — inside the 2s window.
assert_eq!(respawn_decision(attempted(1, 1, now), now), None);
// Past it.
assert_eq!(respawn_decision(attempted(1, 5, now), now), Some(2));
}
#[test]
fn backoff_grows_then_stops_growing() {
assert_eq!(respawn_backoff(0), Duration::from_secs(2));
assert_eq!(respawn_backoff(3), Duration::from_secs(16));
// Capped, and no overflow panic for an attempt count that cannot occur but
// must not be able to crash the sweep if it ever did.
assert_eq!(respawn_backoff(40), Duration::from_secs(RESPAWN_BACKOFF_MAX_SECS));
}
#[test]
fn a_crash_loop_is_given_up_on() {
let now = Instant::now();
// Budget spent, and the backoff would otherwise have elapsed.
assert_eq!(respawn_decision(attempted(MAX_RESPAWN_ATTEMPTS, 120, now), now), None);
}
#[test]
fn giving_up_is_not_permanent() {
// The escape hatch: after a quiet window even an exhausted server is tried
// once more, so a box left running recovers from a transient outage instead
// of staying dark until the next login.
let now = Instant::now();
let quiet = RESPAWN_RESET_SECS + 1;
assert_eq!(respawn_decision(attempted(MAX_RESPAWN_ATTEMPTS, quiet, now), now), Some(1));
}
#[test]
fn recovery_forgives_the_attempt_history() {
// A server that died twice, was restarted, and then ran fine for an hour
// must not be judged on those two deaths when it eventually dies again.
let now = Instant::now();
assert_eq!(respawn_decision(attempted(2, 3600, now), now), Some(1));
}
} }
+4
View File
@@ -167,6 +167,10 @@ impl Integrations {
"data", "data",
crate::mcp::EventLog::Discard, crate::mcp::EventLog::Discard,
)); ));
// Supervise the global connectors: a crashed one is restarted rather than
// staying dead until the process does. Spawned post-construction because the
// sweep reacts through the `Arc` it is watching (see `spawn_respawn_sweep`).
mcp.spawn_respawn_sweep(rt.shutdown_token.clone());
let mut plugin_manager = PluginManager::new(Arc::clone(&rt.db)); let mut plugin_manager = PluginManager::new(Arc::clone(&rt.db));
for plugin in plugins { for plugin in plugins {
@@ -258,6 +258,12 @@ impl UserContextFactory {
// what event triage reads on their behalf. // what event triage reads on their behalf.
crate::mcp::EventLog::Persist, crate::mcp::EventLog::Persist,
)); ));
// Supervise this user's connectors. Matters more here than for the globals:
// a per-user connector is the one that pushes (Gmail's new-mail poll feeds
// event triage), so a crash is otherwise a silence nobody is told about.
// Bound to the user's own token, so a logout stops the sweep with everything
// else of theirs.
user_mcp.spawn_respawn_sweep(user_shutdown.clone());
// NOTE: per-user MCP elicitation (interactive connector login, §15) is // NOTE: per-user MCP elicitation (interactive connector login, §15) is
// deferred — api-key connectors don't need it. Wire the user's // deferred — api-key connectors don't need it. Wire the user's
// ElicitationBridge here when interactive auth lands. // ElicitationBridge here when interactive auth lands.