diff --git a/src/agent/agent_loop.rs b/src/agent/agent_loop.rs index 0b78d7e7..05f3a81a 100644 --- a/src/agent/agent_loop.rs +++ b/src/agent/agent_loop.rs @@ -538,30 +538,50 @@ impl Agent { .await; let notify_user = heartbeat_notify_user; let channels = self.channels.clone(); + let is_multi_tenant = hb_config.multi_tenant; tokio::spawn(async move { while let Some(response) = notify_rx.recv().await { + // In multi-tenant mode, extract the owning user_id from + // the response metadata so notifications reach the + // correct user rather than the agent's owner. + let effective_user = if is_multi_tenant { + response + .metadata + .get("owner_id") + .and_then(|v| v.as_str()) + .map(String::from) + } else { + None + }; + // Try the configured channel first, fall back to // broadcasting on all channels. - let targeted_ok = if let Some(ref channel) = notify_channel - && let Some(ref user) = notify_target - { - channels - .broadcast(channel, user, response.clone()) - .await - .is_ok() + let targeted_ok = if let Some(ref channel) = notify_channel { + let target = effective_user.as_deref().or(notify_target.as_deref()); + if let Some(user) = target { + channels + .broadcast(channel, user, response.clone()) + .await + .is_ok() + } else { + false + } } else { false }; - if !targeted_ok && let Some(ref user) = notify_user { - let results = channels.broadcast_all(user, response).await; - for (ch, result) in results { - if let Err(e) = result { - tracing::warn!( - "Failed to broadcast heartbeat to {}: {}", - ch, - e - ); + if !targeted_ok { + let fallback = effective_user.as_deref().or(notify_user.as_deref()); + if let Some(user) = fallback { + let results = channels.broadcast_all(user, response).await; + for (ch, result) in results { + if let Err(e) = result { + tracing::warn!( + "Failed to broadcast heartbeat to {}: {}", + ch, + e + ); + } } } } @@ -1266,7 +1286,7 @@ impl Agent { message.channel ); // Authorization checks (including restart channel check) are enforced in handle_system_command - self.handle_system_command(&command, &args, &message.channel) + self.handle_system_command(&command, &args, &message.channel, &message.user_id) .await } Submission::Undo => self.process_undo(session, thread_id).await, diff --git a/src/agent/commands.rs b/src/agent/commands.rs index 4e5b681c..8e3bad41 100644 --- a/src/agent/commands.rs +++ b/src/agent/commands.rs @@ -69,7 +69,7 @@ impl Agent { } MessageIntent::Command { command, args } => { match self - .handle_command(&command, &args, &message.channel) + .handle_command(&command, &args, &message.channel, &message.user_id) .await? { Some(s) => s, @@ -125,6 +125,10 @@ impl Agent { if let Some(store) = self.store() && let Ok(Some(ctx)) = store.get_job(uuid).await { + // Ownership check: ensure the job belongs to the requesting user. + if ctx.user_id != user_id { + return Err(crate::error::JobError::NotFound { id: uuid }.into()); + } return Ok(format!( "Job: {}\nStatus: {:?}\nCreated: {}\nStarted: {}\nActual cost: {}", ctx.title, @@ -471,6 +475,7 @@ impl Agent { command: &str, args: &[String], channel: &str, + user_id: &str, ) -> Result { match command { "help" => Ok(SubmissionResult::response(concat!( @@ -664,12 +669,12 @@ impl Agent { } if self.config.multi_tenant { - // Multi-tenant: only persist to per-user settings. + // Multi-tenant: only persist to per-user DB settings. // Do NOT call set_model() on the shared provider — that // would change the default for all users. The per-request // model_override in the dispatcher reads from the same // "selected_model" setting and applies it per-user. - self.persist_selected_model(requested).await; + self.persist_selected_model(user_id, requested).await; Ok(SubmissionResult::response(format!( "Model preference set to: {} (per-user)", requested @@ -678,7 +683,7 @@ impl Agent { match self.llm().set_model(requested) { Ok(()) => { // Persist the model choice so it survives restarts. - self.persist_selected_model(requested).await; + self.persist_selected_model(user_id, requested).await; Ok(SubmissionResult::response(format!( "Switched model to: {}", requested @@ -830,10 +835,14 @@ impl Agent { command: &str, args: &[String], channel: &str, + user_id: &str, ) -> Result, Error> { // System commands are now handled directly via Submission::SystemCommand, // but the router may still send us unknown /commands. - match self.handle_system_command(command, args, channel).await? { + match self + .handle_system_command(command, args, channel, user_id) + .await? + { SubmissionResult::Response { content } => Ok(Some(content)), SubmissionResult::Ok { message } => Ok(message), SubmissionResult::Error { message } => Ok(Some(format!("Error: {}", message))), @@ -845,23 +854,29 @@ impl Agent { /// /// Best-effort: logs warnings on failure but does not propagate errors, /// since the in-memory model switch already succeeded. - async fn persist_selected_model(&self, model: &str) { - // 1. Persist to DB if available. + /// + /// In multi-tenant mode, only the per-user DB setting is written — global + /// .env and TOML files are shared across users and must not be mutated. + async fn persist_selected_model(&self, user_id: &str, model: &str) { + // 1. Persist to DB if available (per-user scoped). if let Some(store) = self.store() { let value = serde_json::Value::String(model.to_string()); - if let Err(e) = store - .set_setting(self.owner_id(), "selected_model", &value) - .await - { + if let Err(e) = store.set_setting(user_id, "selected_model", &value).await { tracing::warn!("Failed to persist model to DB: {}", e); } else { - tracing::debug!("Persisted selected_model to DB: {}", model); + tracing::debug!(user_id, "Persisted selected_model to DB: {}", model); } } else { tracing::warn!("No database store available — model choice will not persist to DB"); } - // 2. Update .env and TOML config file (sync I/O in spawn_blocking). + // 2. In multi-tenant mode, skip .env/TOML writes — these are global + // files shared by all users. The per-user DB setting is sufficient. + if self.config.multi_tenant { + return; + } + + // 3. Update .env and TOML config file (sync I/O in spawn_blocking). let model_owned = model.to_string(); let backend = self.deps.llm_backend.clone(); if let Err(e) = tokio::task::spawn_blocking(move || { diff --git a/src/agent/heartbeat.rs b/src/agent/heartbeat.rs index 3f3dfab5..32f79444 100644 --- a/src/agent/heartbeat.rs +++ b/src/agent/heartbeat.rs @@ -627,7 +627,14 @@ pub fn spawn_multi_user_heartbeat( } // Collect results and update failure counts - while let Some(Ok((uid, result))) = join_set.join_next().await { + while let Some(join_result) = join_set.join_next().await { + let (uid, result) = match join_result { + Ok(pair) => pair, + Err(e) => { + tracing::error!("Multi-user heartbeat task panicked: {}", e); + continue; + } + }; match result { HeartbeatResult::Ok => { tracing::trace!(user_id = uid, "Multi-user heartbeat OK");