feat: complete multi-tenant isolation — phases 2–4 (#1614)

* feat: complete multi-tenant isolation — per-user budgets, model selection, heartbeat cycling Finishes the remaining isolation work from phases 2–4 of #59: Phase 2 (DB scoping): Fix /status and /list commands to use _for_user DB variants instead of global queries that leaked cross-user job data. Phase 3 (Runtime isolation): Per-user workspace in routine engine's spawn_fire so lightweight routines run in the correct user context. Per-user daily cost tracking in CostGuard with configurable budget via MAX_COST_PER_USER_PER_DAY_CENTS. Multi-user heartbeat that cycles through all users with routines, auto-detected from GATEWAY_USER_TOKENS. Phase 4 (Provider/tools): Per-user model selection via preferred_model setting — looked up from SettingsStore on first iteration, threaded through ReasoningContext.model_override to CompletionRequest. Works with providers that support per-request model overrides (NearAI). Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]> * fix: use selected_model setting key to match /model command persistence The dispatcher was reading "preferred_model" but the /model command (merged from staging) persists to "selected_model". Since set_setting is already per-user scoped, using the same key makes /model work as the per-user model override in multi-tenant mode. Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]> * fix: heartbeat hygiene, /model multi-tenant guard, RigAdapter model override Three follow-up fixes for multi-tenant isolation: 1. Multi-user heartbeat now runs memory hygiene per user before each heartbeat check, matching single-user heartbeat behavior. 2. /model command in multi-tenant mode only persists to per-user settings (selected_model) without calling set_model() on the shared LlmProvider. The per-request model_override in the dispatcher reads from the same setting. Added multi_tenant flag to AgentConfig (auto-detected from GATEWAY_USER_TOKENS). 3. RigAdapter now supports per-request model overrides by injecting the model name into rig-core's additional_params. OpenAI/Anthropic/Ollama API servers use last-key-wins for duplicate JSON keys, so the override takes effect via serde's flatten serialization order. Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]> * fix: address PR review — cost model attribution, heartbeat concurrency, pruning Fixes from review comments on #1614: - Cost tracking now uses the override model name (not active_model_name) when a per-user model override is active, for accurate attribution. - Multi-user heartbeat runs per-user checks concurrently via JoinSet instead of sequentially, preventing one slow user from blocking others. - Per-user failure counts tracked independently; users exceeding max_failures are skipped (matching single-user semantics). - per_user_daily_cost HashMap pruned on day rollover to prevent unbounded growth in long-lived deployments. - Doc comment fixed: says "routines" not "active routines". Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]> * fix: /status ownership, model persistence scoping, heartbeat robustness Addresses second round of PR review on #1614: - /status <job_id> DB path now validates job.user_id == requesting user before returning data (was missing ownership check, security fix). - persist_selected_model takes user_id param instead of owner_id, and skips .env/TOML writes in multi-tenant mode (these are shared global files). handle_system_command now receives user_id from caller. - JoinSet collection handles Err(JoinError) explicitly instead of silently dropping panicked tasks. - Notification forwarder extracts owner_id from response metadata in multi-tenant mode for per-user routing instead of broadcasting to the agent owner. Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]> * fix: cost pricing, fire_manual workspace, heartbeat concurrency cap Round 3 review fixes: - Cost tracking passes None for cost_per_token when model override is active, letting CostGuard look up pricing by model name instead of using the default provider's rates (serrrfirat). - fire_manual() now uses per-user workspace, matching spawn_fire() pattern (serrrfirat). - Removed MULTI_TENANT env var — multi-tenant mode is auto-detected solely from GATEWAY_USER_TOKENS presence (serrrfirat + Copilot). - Multi-user heartbeat capped at 8 concurrent tasks to avoid flooding the LLM provider (serrrfirat + Copilot). - Fixed inject_model_override doc comment accuracy (Copilot). - Added comment explaining multi-tenant notification routing priority (Copilot). Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]> * feat: user-scoped webhook endpoint for multi-tenant isolation Adds POST /api/webhooks/u/{user_id}/{path} — a user-scoped webhook endpoint that filters the routine lookup by user_id, preventing cross-user webhook triggering when paths collide. The existing /api/webhooks/{path} endpoint remains unchanged for backward compatibility in single-user deployments. Changes: - get_webhook_routine_by_path gains user_id: Option<&str> param - Both postgres and libsql implementations add AND user_id = ? filter when user_id is provided - New webhook_trigger_user_scoped_handler extracts (user_id, path) from URL and passes to shared fire_webhook_inner logic - Route registered on public router (webhooks are called by external services that can't send bearer tokens) Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]> * feat: add TenantCtx for compile-time tenant isolation Implements zmanian's architectural proposal from #1614 review: two-tier scoped database access (TenantScope/AdminScope) so handler code cannot accidentally bypass tenant scoping. TenantScope (default): wraps user_id + Arc<dyn Database>, auto-binds user_id on every operation. ID-based lookups return None for cross- tenant resources. No escape hatch — forgetting to scope is a compile error. AdminScope (explicit opt-in): cross-tenant access for system-level components (heartbeat, routine engine, self-repair, scheduler, worker). TenantCtx bundles TenantScope + workspace + cost guard + per-user rate limiting. Constructed once per request in handle_message, threaded through all command handlers and ChatDelegate. Key changes: - New src/tenant.rs (~920 lines): TenantScope, AdminScope, TenantCtx, TenantRateState, TenantRateRegistry - All command handlers: user_id: &str → ctx: &TenantCtx - ChatDelegate: cost check/record/settings via self.tenant - System components: store field changed to AdminScope - Config: TENANT_MAX_LLM_CONCURRENT, TENANT_MAX_JOBS_CONCURRENT env vars - Fixes bug: /status <job_id> cross-tenant leak (now auto-filtered) Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]> --------- Co-authored-by: Claude Opus 4.6 (1M context) <[email protected]>
2026-08-31 08:39:24 +00:00 · 2026-03-25 17:24:48 -07:00
parent 86d1143064
commit 4c043bf057
30 changed files with 1825 additions and 174 deletions
@@ -13,7 +13,7 @@ use futures::StreamExt;
 use uuid::Uuid;

 use crate::agent::context_monitor::ContextMonitor;
-use crate::agent::heartbeat::spawn_heartbeat;
+use crate::agent::heartbeat::{spawn_heartbeat, spawn_multi_user_heartbeat};
 use crate::agent::routine_engine::{RoutineEngine, spawn_cron_ticker};
 use crate::agent::self_repair::{DefaultSelfRepair, RepairResult, SelfRepair};
 use crate::agent::session::ThreadState;
@@ -182,6 +182,8 @@ pub struct AgentDeps {
    /// Resolved LLM backend identifier (e.g., "nearai", "openai", "groq").
    /// Used by `/model` persistence to determine which env var to update.
    pub llm_backend: String,
+    /// Per-tenant rate limiting registry (lazily creates rate state per user).
+    pub tenant_rates: Arc<crate::tenant::TenantRateRegistry>,
 }

 /// The main agent that coordinates all components.
@@ -244,7 +246,10 @@ impl Agent {
            SchedulerDeps {
                tools: deps.tools.clone(),
                extension_manager: deps.extension_manager.clone(),
-                store: deps.store.clone(),
+                store: deps
+                    .store
+                    .as_ref()
+                    .map(|db| crate::tenant::AdminScope::new(Arc::clone(db))),
                hooks: deps.hooks.clone(),
            },
        );
@@ -325,6 +330,50 @@ impl Agent {
        &self.deps.cost_guard
    }

+    /// Build a tenant-scoped execution context for the given user.
+    ///
+    /// This is the standard entry point for per-user operations. The returned
+    /// [`TenantCtx`] provides a [`TenantScope`] that auto-binds `user_id` on
+    /// every database operation and a per-user rate limiter.
+    pub(super) async fn tenant_ctx(&self, user_id: &str) -> crate::tenant::TenantCtx {
+        let rate = self.deps.tenant_rates.get_or_create(user_id).await;
+
+        let store = self
+            .deps
+            .store
+            .as_ref()
+            .map(|db| crate::tenant::TenantScope::new(user_id, Arc::clone(db)));
+
+        // Reuse the owner workspace if user matches, otherwise create per-user.
+        let workspace = match &self.deps.workspace {
+            Some(ws) if ws.user_id() == user_id => Some(Arc::clone(ws)),
+            _ => self
+                .deps
+                .store
+                .as_ref()
+                .map(|db| Arc::new(Workspace::new_with_db(user_id, Arc::clone(db)))),
+        };
+
+        crate::tenant::TenantCtx::new(
+            user_id,
+            store,
+            workspace,
+            Arc::clone(&self.deps.cost_guard),
+            rate,
+        )
+    }
+
+    /// Get an admin-scoped database accessor for cross-tenant operations.
+    ///
+    /// Only for system-level components (heartbeat, routine engine, self-repair,
+    /// scheduler). Handler code should use [`tenant_ctx()`](Self::tenant_ctx) instead.
+    pub(super) fn admin_store(&self) -> Option<crate::tenant::AdminScope> {
+        self.deps
+            .store
+            .as_ref()
+            .map(|db| crate::tenant::AdminScope::new(Arc::clone(db)))
+    }
+
    pub(super) fn skill_registry(&self) -> Option<&Arc<std::sync::RwLock<SkillRegistry>>> {
        self.deps.skill_registry.as_ref()
    }
@@ -410,8 +459,8 @@ impl Agent {
            self.config.stuck_threshold,
            self.config.max_repair_attempts,
        );
-        if let Some(ref store) = self.deps.store {
-            self_repair = self_repair.with_store(Arc::clone(store));
+        if let Some(admin) = self.admin_store() {
+            self_repair = self_repair.with_store(admin);
        }
        if let Some(ref builder) = self.deps.builder {
            self_repair = self_repair.with_builder(Arc::clone(builder), Arc::clone(self.tools()));
@@ -518,6 +567,7 @@ impl Agent {
                        .with_interval(std::time::Duration::from_secs(hb_config.interval_secs));
                    config.quiet_hours_start = hb_config.quiet_hours_start;
                    config.quiet_hours_end = hb_config.quiet_hours_end;
+                    config.multi_tenant = hb_config.multi_tenant;
                    config.timezone = hb_config
                        .timezone
                        .clone()
@@ -547,30 +597,52 @@ 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.
+                            // This intentionally overrides the configured notify_target
+                            // because each user's heartbeat should notify that user.
+                            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
+                                            );
+                                        }
                                    }
                                }
                            }
@@ -583,14 +655,29 @@ impl Agent {
                        .map(|h| h.to_workspace_config())
                        .unwrap_or_default();

-                    Some(spawn_heartbeat(
-                        config,
-                        hygiene,
-                        workspace.clone(),
-                        self.cheap_llm().clone(),
-                        Some(notify_tx),
-                        self.store().map(Arc::clone),
-                    ))
+                    if config.multi_tenant {
+                        if let Some(admin) = self.admin_store() {
+                            Some(spawn_multi_user_heartbeat(
+                                config,
+                                hygiene,
+                                self.cheap_llm().clone(),
+                                Some(notify_tx),
+                                admin,
+                            ))
+                        } else {
+                            tracing::warn!("Multi-tenant heartbeat requires a database store");
+                            None
+                        }
+                    } else {
+                        Some(spawn_heartbeat(
+                            config,
+                            hygiene,
+                            workspace.clone(),
+                            self.cheap_llm().clone(),
+                            Some(notify_tx),
+                            self.admin_store(),
+                        ))
+                    }
                } else {
                    tracing::warn!("Heartbeat enabled but no workspace available");
                    None
@@ -612,7 +699,7 @@ impl Agent {

                    let engine = Arc::new(RoutineEngine::new(
                        rt_config.clone(),
-                        Arc::clone(store),
+                        crate::tenant::AdminScope::new(Arc::clone(store)),
                        self.llm().clone(),
                        Arc::clone(workspace),
                        notify_tx,
@@ -1173,13 +1260,22 @@ impl Agent {
            }
        }

+        // Build per-tenant execution context once; threaded through all handlers.
+        let tenant = self.tenant_ctx(&message.user_id).await;
+
        let session_for_empty_exit = Arc::clone(&session);

        // Process based on submission type
        let result = match submission {
            Submission::UserInput { content } => {
                let mut result = self
-                    .process_user_input(message, session.clone(), thread_id, &content)
+                    .process_user_input(
+                        message,
+                        tenant.clone(),
+                        session.clone(),
+                        thread_id,
+                        &content,
+                    )
                    .await;

                // Drain any messages queued during processing.
@@ -1246,7 +1342,13 @@ impl Agent {
                    let mut queued_msg = message.clone();
                    queued_msg.attachments.clear();
                    result = self
-                        .process_user_input(&queued_msg, session.clone(), thread_id, &next_content)
+                        .process_user_input(
+                            &queued_msg,
+                            tenant.clone(),
+                            session.clone(),
+                            thread_id,
+                            &next_content,
+                        )
                        .await;

                    // If processing failed, re-queue the drained content so it
@@ -1294,7 +1396,7 @@ impl Agent {
                    };
                }
                // 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, &tenant)
                    .await
            }
            Submission::Undo => self.process_undo(session, thread_id).await,
@@ -1307,12 +1409,9 @@ impl Agent {
            Submission::Summarize => self.process_summarize(session, thread_id).await,
            Submission::Suggest => self.process_suggest(session, thread_id).await,
            Submission::JobStatus { job_id } => {
-                self.process_job_status(&message.user_id, job_id.as_deref())
-                    .await
-            }
-            Submission::JobCancel { job_id } => {
-                self.process_job_cancel(&message.user_id, &job_id).await
+                self.process_job_status(&tenant, job_id.as_deref()).await
            }
+            Submission::JobCancel { job_id } => self.process_job_cancel(&tenant, &job_id).await,
            Submission::Quit => return Ok(None),
            Submission::SwitchThread { thread_id: target } => {
                self.process_switch_thread(message, target).await
@@ -33,6 +33,7 @@ impl Agent {
        &self,
        intent: MessageIntent,
        message: &IncomingMessage,
+        tenant: &crate::tenant::TenantCtx,
    ) -> Result<SubmissionResult, Error> {
        // Send thinking status for non-trivial operations
        if let MessageIntent::CreateJob { .. } = &intent {
@@ -52,24 +53,18 @@ impl Agent {
                description,
                category,
            } => {
-                self.handle_create_job(&message.user_id, title, description, category)
+                self.handle_create_job(tenant, title, description, category)
                    .await?
            }
            MessageIntent::CheckJobStatus { job_id } => {
-                self.handle_check_status(&message.user_id, job_id).await?
-            }
-            MessageIntent::CancelJob { job_id } => {
-                self.handle_cancel_job(&message.user_id, &job_id).await?
-            }
-            MessageIntent::ListJobs { filter } => {
-                self.handle_list_jobs(&message.user_id, filter).await?
-            }
-            MessageIntent::HelpJob { job_id } => {
-                self.handle_help_job(&message.user_id, &job_id).await?
+                self.handle_check_status(tenant, job_id).await?
            }
+            MessageIntent::CancelJob { job_id } => self.handle_cancel_job(tenant, &job_id).await?,
+            MessageIntent::ListJobs { filter } => self.handle_list_jobs(tenant, filter).await?,
+            MessageIntent::HelpJob { job_id } => self.handle_help_job(tenant, &job_id).await?,
            MessageIntent::Command { command, args } => {
                match self
-                    .handle_command(&command, &args, &message.channel)
+                    .handle_command(&command, &args, &message.channel, tenant)
                    .await?
                {
                    Some(s) => s,
@@ -83,14 +78,14 @@ impl Agent {

    async fn handle_create_job(
        &self,
-        user_id: &str,
+        tenant: &crate::tenant::TenantCtx,
        title: String,
        description: String,
        category: Option<String>,
    ) -> Result<String, Error> {
        let job_id = self
            .scheduler
-            .dispatch_job(user_id, &title, &description, None)
+            .dispatch_job(tenant.user_id(), &title, &description, None)
            .await?;

        // Set the dedicated category field (not stored in metadata)
@@ -113,7 +108,7 @@ impl Agent {

    async fn handle_check_status(
        &self,
-        user_id: &str,
+        tenant: &crate::tenant::TenantCtx,
        job_id: Option<String>,
    ) -> Result<String, Error> {
        match job_id {
@@ -122,7 +117,8 @@ impl Agent {
                    .map_err(|_| crate::error::JobError::NotFound { id: Uuid::nil() })?;

                // Try DB first for persistent state, fall back to ContextManager.
-                if let Some(store) = self.store()
+                // TenantScope.get_job() auto-filters by ownership — no manual check needed.
+                if let Some(store) = tenant.store()
                    && let Ok(Some(ctx)) = store.get_job(uuid).await
                {
                    return Ok(format!(
@@ -138,7 +134,7 @@ impl Agent {
                }

                let ctx = self.context_manager.get_context(uuid).await?;
-                if ctx.user_id != user_id {
+                if ctx.user_id != tenant.user_id() {
                    return Err(crate::error::JobError::NotFound { id: uuid }.into());
                }

@@ -155,7 +151,8 @@ impl Agent {
            }
            None => {
                // Show summary from DB for consistency with Jobs tab.
-                if let Some(store) = self.store() {
+                // TenantScope methods auto-scope to user — no user_id parameter needed.
+                if let Some(store) = tenant.store() {
                    let mut total = 0;
                    let mut in_progress = 0;
                    let mut completed = 0;
@@ -183,7 +180,7 @@ impl Agent {
                }

                // Fallback to ContextManager if no DB.
-                let summary = self.context_manager.summary_for(user_id).await;
+                let summary = self.context_manager.summary_for(tenant.user_id()).await;
                Ok(format!(
                    "Jobs summary: Total: {} In Progress: {} Completed: {} Failed: {} Stuck: {}",
                    summary.total,
@@ -196,19 +193,24 @@ impl Agent {
        }
    }

-    async fn handle_cancel_job(&self, user_id: &str, job_id: &str) -> Result<String, Error> {
+    async fn handle_cancel_job(
+        &self,
+        tenant: &crate::tenant::TenantCtx,
+        job_id: &str,
+    ) -> Result<String, Error> {
        let uuid = Uuid::parse_str(job_id)
            .map_err(|_| crate::error::JobError::NotFound { id: Uuid::nil() })?;

        let ctx = self.context_manager.get_context(uuid).await?;
-        if ctx.user_id != user_id {
+        if ctx.user_id != tenant.user_id() {
            return Err(crate::error::JobError::NotFound { id: uuid }.into());
        }

        self.scheduler.stop(uuid).await?;

        // Also update DB so the Jobs tab reflects cancellation immediately.
-        if let Some(store) = self.store()
+        // Use TenantScope — ownership already verified above.
+        if let Some(store) = tenant.store()
            && let Err(e) = store
                .update_job_status(uuid, JobState::Cancelled, Some("Cancelled by user"))
                .await
@@ -221,11 +223,12 @@ impl Agent {

    async fn handle_list_jobs(
        &self,
-        user_id: &str,
+        tenant: &crate::tenant::TenantCtx,
        _filter: Option<String>,
    ) -> Result<String, Error> {
        // List from DB for consistency with Jobs tab.
-        if let Some(store) = self.store() {
+        // TenantScope methods auto-scope to user.
+        if let Some(store) = tenant.store() {
            let agent_jobs = match store.list_agent_jobs().await {
                Ok(jobs) => jobs,
                Err(e) => {
@@ -256,7 +259,7 @@ impl Agent {
        }

        // Fallback to ContextManager if no DB.
-        let jobs = self.context_manager.all_jobs_for(user_id).await;
+        let jobs = self.context_manager.all_jobs_for(tenant.user_id()).await;
        if jobs.is_empty() {
            return Ok("No jobs found.".to_string());
        }
@@ -270,12 +273,16 @@ impl Agent {
        Ok(output)
    }

-    async fn handle_help_job(&self, user_id: &str, job_id: &str) -> Result<String, Error> {
+    async fn handle_help_job(
+        &self,
+        tenant: &crate::tenant::TenantCtx,
+        job_id: &str,
+    ) -> Result<String, Error> {
        let uuid = Uuid::parse_str(job_id)
            .map_err(|_| crate::error::JobError::NotFound { id: Uuid::nil() })?;

        let ctx = self.context_manager.get_context(uuid).await?;
-        if ctx.user_id != user_id {
+        if ctx.user_id != tenant.user_id() {
            return Err(crate::error::JobError::NotFound { id: uuid }.into());
        }

@@ -308,11 +315,11 @@ impl Agent {
    /// Show job status inline — either all jobs (no id) or a specific job.
    pub(super) async fn process_job_status(
        &self,
-        user_id: &str,
+        tenant: &crate::tenant::TenantCtx,
        job_id: Option<&str>,
    ) -> Result<SubmissionResult, Error> {
        match self
-            .handle_check_status(user_id, job_id.map(|s| s.to_string()))
+            .handle_check_status(tenant, job_id.map(|s| s.to_string()))
            .await
        {
            Ok(text) => Ok(SubmissionResult::response(text)),
@@ -323,10 +330,10 @@ impl Agent {
    /// Cancel a job by ID.
    pub(super) async fn process_job_cancel(
        &self,
-        user_id: &str,
+        tenant: &crate::tenant::TenantCtx,
        job_id: &str,
    ) -> Result<SubmissionResult, Error> {
-        match self.handle_cancel_job(user_id, job_id).await {
+        match self.handle_cancel_job(tenant, job_id).await {
            Ok(text) => Ok(SubmissionResult::response(text)),
            Err(e) => Ok(SubmissionResult::error(format!("Cancel error: {}", e))),
        }
@@ -559,6 +566,7 @@ impl Agent {
        command: &str,
        args: &[String],
        channel: &str,
+        tenant: &crate::tenant::TenantCtx,
    ) -> Result<SubmissionResult, Error> {
        match command {
            "help" => Ok(SubmissionResult::response(concat!(
@@ -752,19 +760,32 @@ impl Agent {
                        }
                    }

-                    match self.llm().set_model(requested) {
-                        Ok(()) => {
-                            // Persist the model choice so it survives restarts.
-                            self.persist_selected_model(requested).await;
-                            Ok(SubmissionResult::response(format!(
-                                "Switched model to: {}",
-                                requested
-                            )))
+                    if self.config.multi_tenant {
+                        // 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(tenant, requested).await;
+                        Ok(SubmissionResult::response(format!(
+                            "Model preference set to: {} (per-user)",
+                            requested
+                        )))
+                    } else {
+                        match self.llm().set_model(requested) {
+                            Ok(()) => {
+                                // Persist the model choice so it survives restarts.
+                                self.persist_selected_model(tenant, requested).await;
+                                Ok(SubmissionResult::response(format!(
+                                    "Switched model to: {}",
+                                    requested
+                                )))
+                            }
+                            Err(e) => Ok(SubmissionResult::error(format!(
+                                "Failed to switch model: {}",
+                                e
+                            ))),
                        }
-                        Err(e) => Ok(SubmissionResult::error(format!(
-                            "Failed to switch model: {}",
-                            e
-                        ))),
                    }
                }
            }
@@ -906,10 +927,14 @@ impl Agent {
        command: &str,
        args: &[String],
        channel: &str,
+        tenant: &crate::tenant::TenantCtx,
    ) -> Result<Option<String>, 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, tenant)
+            .await?
+        {
            SubmissionResult::Response { content } => Ok(Some(content)),
            SubmissionResult::Ok { message } => Ok(message),
            SubmissionResult::Error { message } => Ok(Some(format!("Error: {}", message))),
@@ -921,23 +946,33 @@ 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.
-        if let Some(store) = self.store() {
+    ///
+    /// 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, tenant: &crate::tenant::TenantCtx, model: &str) {
+        // 1. Persist to DB if available (per-user scoped via TenantScope).
+        if let Some(store) = tenant.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("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 = tenant.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 || {
@@ -21,6 +21,9 @@ pub struct CostGuardConfig {
    pub max_cost_per_day_cents: Option<u64>,
    /// Maximum LLM calls per hour. None = unlimited.
    pub max_actions_per_hour: Option<u64>,
+    /// Maximum spend per user per day in cents. None = unlimited.
+    /// Applied independently per user alongside the global budget.
+    pub max_cost_per_user_per_day_cents: Option<u64>,
 }

 /// Error returned when a cost limit is exceeded.
@@ -30,6 +33,12 @@ pub enum CostLimitExceeded {
    DailyBudget { spent_cents: u64, limit_cents: u64 },
    /// Hourly action rate limit reached.
    HourlyRate { actions: u64, limit: u64 },
+    /// Per-user daily spending cap reached.
+    UserDailyBudget {
+        user_id: String,
+        spent_cents: u64,
+        limit_cents: u64,
+    },
 }

 impl std::fmt::Display for CostLimitExceeded {
@@ -49,6 +58,17 @@ impl std::fmt::Display for CostLimitExceeded {
                "Hourly action limit exceeded: {} actions of {} allowed per hour",
                actions, limit
            ),
+            Self::UserDailyBudget {
+                user_id,
+                spent_cents,
+                limit_cents,
+            } => write!(
+                f,
+                "User '{}' daily cost limit exceeded: spent ${:.2} of ${:.2} allowed",
+                user_id,
+                *spent_cents as f64 / 100.0,
+                *limit_cents as f64 / 100.0
+            ),
        }
    }
 }
@@ -78,6 +98,9 @@ pub struct CostGuard {

    /// Per-model token usage since startup.
    model_tokens: Mutex<HashMap<String, ModelTokens>>,
+
+    /// Per-user daily cost tracking. Each entry resets independently at midnight UTC.
+    per_user_daily_cost: Mutex<HashMap<String, DailyCost>>,
 }

 struct DailyCost {
@@ -97,6 +120,7 @@ impl CostGuard {
            action_window: Mutex::new(VecDeque::new()),
            budget_exceeded: AtomicBool::new(false),
            model_tokens: Mutex::new(HashMap::new()),
+            per_user_daily_cost: Mutex::new(HashMap::new()),
        }
    }

@@ -203,6 +227,11 @@ impl CostGuard {
                daily.reset_date = today;
                self.budget_exceeded.store(false, Ordering::Relaxed);
                tracing::info!("Cost guard: daily counter reset for {}", today);
+
+                // Prune per-user entries from previous days to prevent
+                // unbounded HashMap growth in long-lived deployments.
+                let mut per_user = self.per_user_daily_cost.lock().await;
+                per_user.retain(|_, entry| entry.reset_date == today);
            }
            daily.total += cost;

@@ -248,6 +277,85 @@ impl CostGuard {
        cost
    }

+    /// Record an LLM call with per-user attribution.
+    ///
+    /// Delegates to `record_llm_call` for global tracking, then additionally
+    /// records the cost against the user's daily budget.
+    #[allow(clippy::too_many_arguments)]
+    pub async fn record_llm_call_for_user(
+        &self,
+        user_id: &str,
+        model: &str,
+        input_tokens: u32,
+        output_tokens: u32,
+        cache_read_input_tokens: u32,
+        cache_creation_input_tokens: u32,
+        cache_read_discount: Decimal,
+        cache_write_multiplier: Decimal,
+        cost_per_token: Option<(Decimal, Decimal)>,
+    ) -> Decimal {
+        let cost = self
+            .record_llm_call(
+                model,
+                input_tokens,
+                output_tokens,
+                cache_read_input_tokens,
+                cache_creation_input_tokens,
+                cache_read_discount,
+                cache_write_multiplier,
+                cost_per_token,
+            )
+            .await;
+
+        // Track per-user daily cost
+        {
+            let today = chrono::Utc::now().date_naive();
+            let mut per_user = self.per_user_daily_cost.lock().await;
+            let entry = per_user
+                .entry(user_id.to_string())
+                .or_insert_with(|| DailyCost {
+                    total: Decimal::ZERO,
+                    reset_date: today,
+                });
+            if today != entry.reset_date {
+                entry.total = Decimal::ZERO;
+                entry.reset_date = today;
+            }
+            entry.total += cost;
+        }
+
+        cost
+    }
+
+    /// Check whether the next action is allowed for a specific user.
+    ///
+    /// Checks the global limits first (via `check_allowed`), then additionally
+    /// checks the per-user daily budget if configured.
+    pub async fn check_allowed_for_user(&self, user_id: &str) -> Result<(), CostLimitExceeded> {
+        // Check global limits first
+        self.check_allowed().await?;
+
+        // Check per-user daily budget
+        if let Some(limit_cents) = self.config.max_cost_per_user_per_day_cents {
+            let today = chrono::Utc::now().date_naive();
+            let per_user = self.per_user_daily_cost.lock().await;
+            if let Some(entry) = per_user.get(user_id)
+                && entry.reset_date == today
+            {
+                let spent_cents = to_cents(entry.total);
+                if spent_cents >= limit_cents {
+                    return Err(CostLimitExceeded::UserDailyBudget {
+                        user_id: user_id.to_string(),
+                        spent_cents,
+                        limit_cents,
+                    });
+                }
+            }
+        }
+
+        Ok(())
+    }
+
    /// Current daily spend in USD (as Decimal).
    pub async fn daily_spend(&self) -> Decimal {
        let daily = self.daily_cost.lock().await;
@@ -259,6 +367,16 @@ impl CostGuard {
        }
    }

+    /// Current daily spend for a specific user in USD (as Decimal).
+    pub async fn daily_spend_for_user(&self, user_id: &str) -> Decimal {
+        let today = chrono::Utc::now().date_naive();
+        let per_user = self.per_user_daily_cost.lock().await;
+        match per_user.get(user_id) {
+            Some(entry) if entry.reset_date == today => entry.total,
+            _ => Decimal::ZERO,
+        }
+    }
+
    /// Number of actions in the current hourly window.
    pub async fn actions_this_hour(&self) -> u64 {
        let mut window = self.action_window.lock().await;
@@ -314,7 +432,7 @@ mod tests {
    async fn test_daily_budget_enforcement() {
        let guard = CostGuard::new(CostGuardConfig {
            max_cost_per_day_cents: Some(1), // $0.01 limit
-            max_actions_per_hour: None,
+            ..CostGuardConfig::default()
        });

        // First call allowed
@@ -350,8 +468,8 @@ mod tests {
    #[tokio::test]
    async fn test_hourly_rate_enforcement() {
        let guard = CostGuard::new(CostGuardConfig {
-            max_cost_per_day_cents: None,
            max_actions_per_hour: Some(3),
+            ..CostGuardConfig::default()
        });

        // First 3 actions allowed
@@ -633,8 +751,8 @@ mod tests {
        // A fresh CostGuard with rate limits should not panic even if
        // checked_sub returns None (simulating short uptime).
        let guard = CostGuard::new(CostGuardConfig {
-            max_cost_per_day_cents: None,
            max_actions_per_hour: Some(100),
+            ..CostGuardConfig::default()
        });

        // These must not panic regardless of system uptime
@@ -656,4 +774,119 @@ mod tests {
        let result = Instant::now().checked_sub(std::time::Duration::MAX);
        assert!(result.is_none());
    }
+
+    #[tokio::test]
+    async fn test_per_user_daily_budget_enforcement() {
+        let guard = CostGuard::new(CostGuardConfig {
+            max_cost_per_day_cents: None,
+            max_actions_per_hour: None,
+            max_cost_per_user_per_day_cents: Some(1), // $0.01 per user
+        });
+
+        // Both users initially allowed
+        assert!(guard.check_allowed_for_user("alice").await.is_ok());
+        assert!(guard.check_allowed_for_user("bob").await.is_ok());
+
+        // Alice makes an expensive call
+        guard
+            .record_llm_call_for_user(
+                "alice",
+                "gpt-4o",
+                10_000,
+                10_000,
+                0,
+                0,
+                Decimal::ONE,
+                Decimal::ONE,
+                None,
+            )
+            .await;
+
+        // Alice should be blocked, Bob should still be allowed
+        let result = guard.check_allowed_for_user("alice").await;
+        assert!(result.is_err());
+        match result.unwrap_err() {
+            CostLimitExceeded::UserDailyBudget {
+                user_id,
+                limit_cents,
+                ..
+            } => {
+                assert_eq!(user_id, "alice");
+                assert_eq!(limit_cents, 1);
+            }
+            other => panic!("Expected UserDailyBudget, got {:?}", other),
+        }
+        assert!(guard.check_allowed_for_user("bob").await.is_ok());
+    }
+
+    #[tokio::test]
+    async fn test_per_user_daily_spend_tracking() {
+        let guard = CostGuard::new(CostGuardConfig::default());
+
+        assert_eq!(guard.daily_spend_for_user("alice").await, Decimal::ZERO);
+        assert_eq!(guard.daily_spend_for_user("bob").await, Decimal::ZERO);
+
+        let cost = guard
+            .record_llm_call_for_user(
+                "alice",
+                "gpt-4o",
+                1000,
+                500,
+                0,
+                0,
+                Decimal::ONE,
+                Decimal::ONE,
+                None,
+            )
+            .await;
+
+        assert_eq!(guard.daily_spend_for_user("alice").await, cost);
+        assert_eq!(guard.daily_spend_for_user("bob").await, Decimal::ZERO);
+        // Global spend should also be tracked
+        assert_eq!(guard.daily_spend().await, cost);
+    }
+
+    #[tokio::test]
+    async fn test_per_user_budget_independent_of_global() {
+        let guard = CostGuard::new(CostGuardConfig {
+            max_cost_per_day_cents: Some(100_000), // $1000 global limit
+            max_actions_per_hour: None,
+            max_cost_per_user_per_day_cents: Some(1), // $0.01 per user
+        });
+
+        // User hits their personal limit
+        guard
+            .record_llm_call_for_user(
+                "alice",
+                "gpt-4o",
+                10_000,
+                10_000,
+                0,
+                0,
+                Decimal::ONE,
+                Decimal::ONE,
+                None,
+            )
+            .await;
+
+        // Alice blocked by per-user limit, not global
+        assert!(guard.check_allowed_for_user("alice").await.is_err());
+        // Global limit is far from reached
+        assert!(guard.check_allowed().await.is_ok());
+        // Bob is unaffected
+        assert!(guard.check_allowed_for_user("bob").await.is_ok());
+    }
+
+    #[test]
+    fn test_user_cost_limit_display() {
+        let limit = CostLimitExceeded::UserDailyBudget {
+            user_id: "alice".to_string(),
+            spent_cents: 150,
+            limit_cents: 100,
+        };
+        let msg = limit.to_string();
+        assert!(msg.contains("alice"));
+        assert!(msg.contains("$1.50"));
+        assert!(msg.contains("$1.00"));
+    }
 }
@@ -42,6 +42,7 @@ impl Agent {
    pub(super) async fn run_agentic_loop(
        &self,
        message: &IncomingMessage,
+        tenant: crate::tenant::TenantCtx,
        session: Arc<Mutex<Session>>,
        thread_id: Uuid,
        initial_messages: Vec<ChatMessage>,
@@ -168,6 +169,7 @@ impl Agent {

        let delegate = ChatDelegate {
            agent: self,
+            tenant,
            session: session.clone(),
            thread_id,
            message,
@@ -240,6 +242,7 @@ impl Agent {
 /// auth intercept, and cost tracking.
 struct ChatDelegate<'a> {
    agent: &'a Agent,
+    tenant: crate::tenant::TenantCtx,
    session: Arc<Mutex<Session>>,
    thread_id: Uuid,
    message: &'a IncomingMessage,
@@ -336,8 +339,8 @@ impl<'a> LoopDelegate for ChatDelegate<'a> {
        reason_ctx: &mut ReasoningContext,
        iteration: usize,
    ) -> Result<crate::llm::RespondOutput, Error> {
-        // Enforce cost guardrails before the LLM call
-        if let Err(limit) = self.agent.cost_guard().check_allowed().await {
+        // Enforce cost guardrails before the LLM call (global + per-user)
+        if let Err(limit) = self.tenant.check_cost_allowed().await {
            return Err(crate::error::LlmError::InvalidResponse {
                provider: "agent".to_string(),
                reason: limit.to_string(),
@@ -345,6 +348,21 @@ impl<'a> LoopDelegate for ChatDelegate<'a> {
            .into());
        }

+        // Apply per-user model override from settings (first iteration only
+        // to avoid repeated DB lookups within the same agentic loop).
+        // Uses "selected_model" — the same key the /model command persists to
+        // via SettingsStore (per-user scoped via TenantScope).
+        if iteration == 0
+            && let Some(store) = self.tenant.store()
+            && let Ok(Some(value)) = store.get_setting("selected_model").await
+            && let Some(model) = value.as_str()
+        {
+            let model = model.trim();
+            if !model.is_empty() {
+                reason_ctx.model_override = Some(model.to_string());
+            }
+        }
+
        let output = match reasoning.respond_with_tools(reason_ctx).await {
            Ok(output) => output,
            Err(crate::error::LlmError::ContextLengthExceeded { used, limit }) => {
@@ -379,13 +397,22 @@ impl<'a> LoopDelegate for ChatDelegate<'a> {
            Err(e) => return Err(e.into()),
        };

-        // Record cost and track token usage
-        let model_name = self.agent.llm().active_model_name();
+        // Record cost and track token usage (global + per-user).
+        // When a model override is active, use the override name for attribution
+        // and let CostGuard look up pricing via costs::model_cost() instead of
+        // using the default provider's cost_per_token (which reflects the wrong model).
+        let (model_name, cost_per_token) = if let Some(ref ovr) = reason_ctx.model_override {
+            (ovr.clone(), None)
+        } else {
+            (
+                self.agent.llm().active_model_name(),
+                Some(self.agent.llm().cost_per_token()),
+            )
+        };
        let read_discount = self.agent.llm().cache_read_discount();
        let write_multiplier = self.agent.llm().cache_write_multiplier();
        let call_cost = self
-            .agent
-            .cost_guard()
+            .tenant
            .record_llm_call(
                &model_name,
                output.usage.input_tokens,
@@ -394,7 +421,7 @@ impl<'a> LoopDelegate for ChatDelegate<'a> {
                output.usage.cache_creation_input_tokens,
                read_discount,
                write_multiplier,
-                Some(self.agent.llm().cost_per_token()),
+                cost_per_token,
            )
            .await;
        tracing::debug!(
@@ -1305,6 +1332,7 @@ mod tests {
            sandbox_readiness: crate::agent::routine_engine::SandboxReadiness::DisabledByConfig,
            builder: None,
            llm_backend: "nearai".to_string(),
+            tenant_rates: Arc::new(crate::tenant::TenantRateRegistry::new(4, 3)),
        };

        Agent::new(
@@ -1320,10 +1348,14 @@ mod tests {
                allow_local_tools: false,
                max_cost_per_day_cents: None,
                max_actions_per_hour: None,
+                max_cost_per_user_per_day_cents: None,
                max_tool_iterations: 50,
                auto_approve_tools: false,
                default_timezone: "UTC".to_string(),
                max_tokens_per_job: 0,
+                multi_tenant: false,
+                max_llm_concurrent_per_user: None,
+                max_jobs_concurrent_per_user: None,
            },
            deps,
            Arc::new(ChannelManager::new()),
@@ -2181,6 +2213,7 @@ mod tests {
            sandbox_readiness: crate::agent::routine_engine::SandboxReadiness::DisabledByConfig,
            builder: None,
            llm_backend: "nearai".to_string(),
+            tenant_rates: Arc::new(crate::tenant::TenantRateRegistry::new(4, 3)),
        };

        Agent::new(
@@ -2196,10 +2229,14 @@ mod tests {
                allow_local_tools: false,
                max_cost_per_day_cents: None,
                max_actions_per_hour: None,
+                max_cost_per_user_per_day_cents: None,
                max_tool_iterations,
                auto_approve_tools: true,
                default_timezone: "UTC".to_string(),
                max_tokens_per_job: 0,
+                multi_tenant: false,
+                max_llm_concurrent_per_user: None,
+                max_jobs_concurrent_per_user: None,
            },
            deps,
            Arc::new(ChannelManager::new()),
@@ -2234,13 +2271,14 @@ mod tests {

        let message = IncomingMessage::new("test", "test-user", "do something");
        let initial_messages = vec![ChatMessage::user("do something")];
+        let tenant = agent.tenant_ctx("test-user").await;

        // The dispatcher must terminate within 5 seconds. If there is an
        // infinite loop bug (e.g., index not advancing on tool failure), the
        // timeout will fire and the test will fail.
        let result = tokio::time::timeout(
            Duration::from_secs(5),
-            agent.run_agentic_loop(&message, session, thread_id, initial_messages),
+            agent.run_agentic_loop(&message, tenant, session, thread_id, initial_messages),
        )
        .await;

@@ -2302,6 +2340,7 @@ mod tests {
                sandbox_readiness: crate::agent::routine_engine::SandboxReadiness::DisabledByConfig,
                builder: None,
                llm_backend: "nearai".to_string(),
+                tenant_rates: Arc::new(crate::tenant::TenantRateRegistry::new(4, 3)),
            };

            Agent::new(
@@ -2317,10 +2356,14 @@ mod tests {
                    allow_local_tools: false,
                    max_cost_per_day_cents: None,
                    max_actions_per_hour: None,
+                    max_cost_per_user_per_day_cents: None,
                    max_tool_iterations: max_iter,
                    auto_approve_tools: true,
                    default_timezone: "UTC".to_string(),
                    max_tokens_per_job: 0,
+                    multi_tenant: false,
+                    max_llm_concurrent_per_user: None,
+                    max_jobs_concurrent_per_user: None,
                },
                deps,
                Arc::new(ChannelManager::new()),
@@ -2340,13 +2383,14 @@ mod tests {

        let message = IncomingMessage::new("test", "test-user", "keep calling tools");
        let initial_messages = vec![ChatMessage::user("keep calling tools")];
+        let tenant = agent.tenant_ctx("test-user").await;

        // Even with an LLM that always wants to call tools, the dispatcher
        // must terminate within the timeout thanks to force_text at
        // max_tool_iterations.
        let result = tokio::time::timeout(
            Duration::from_secs(5),
-            agent.run_agentic_loop(&message, session, thread_id, initial_messages),
+            agent.run_agentic_loop(&message, tenant, session, thread_id, initial_messages),
        )
        .await;

@@ -31,8 +31,8 @@ use chrono_tz::Tz;
 use tokio::sync::mpsc;

 use crate::channels::OutgoingResponse;
-use crate::db::Database;
 use crate::llm::{ChatMessage, CompletionRequest, LlmProvider, Reasoning};
+use crate::tenant::AdminScope;
 use crate::workspace::Workspace;
 use crate::workspace::hygiene::HygieneConfig;

@@ -57,6 +57,9 @@ pub struct HeartbeatConfig {
    pub quiet_hours_end: Option<u32>,
    /// Timezone for fire_at and quiet hours evaluation (IANA name).
    pub timezone: Option<String>,
+    /// When true, cycle through all users with routines instead of
+    /// running heartbeat for a single user. Requires a database store.
+    pub multi_tenant: bool,
 }

 impl Default for HeartbeatConfig {
@@ -71,6 +74,7 @@ impl Default for HeartbeatConfig {
            quiet_hours_start: None,
            quiet_hours_end: None,
            timezone: None,
+            multi_tenant: false,
        }
    }
 }
@@ -178,7 +182,7 @@ pub struct HeartbeatRunner {
    workspace: Arc<Workspace>,
    llm: Arc<dyn LlmProvider>,
    response_tx: Option<mpsc::Sender<OutgoingResponse>>,
-    store: Option<Arc<dyn Database>>,
+    store: Option<AdminScope>,
    consecutive_failures: u32,
 }

@@ -207,8 +211,8 @@ impl HeartbeatRunner {
        self
    }

-    /// Set the database store for persistent heartbeat conversations.
-    pub fn with_store(mut self, store: Arc<dyn Database>) -> Self {
+    /// Set the admin-scoped database store for persistent heartbeat conversations.
+    pub fn with_store(mut self, store: AdminScope) -> Self {
        self.store = Some(store);
        self
    }
@@ -396,7 +400,7 @@ impl HeartbeatRunner {
    }

    /// Send a notification about heartbeat findings.
-    async fn send_notification(&self, message: &str) {
+    pub(crate) async fn send_notification(&self, message: &str) {
        let Some(ref tx) = self.response_tx else {
            tracing::debug!("No response channel configured for heartbeat notifications");
            return;
@@ -493,7 +497,7 @@ pub fn spawn_heartbeat(
    workspace: Arc<Workspace>,
    llm: Arc<dyn LlmProvider>,
    response_tx: Option<mpsc::Sender<OutgoingResponse>>,
-    store: Option<Arc<dyn Database>>,
+    store: Option<AdminScope>,
 ) -> tokio::task::JoinHandle<()> {
    let mut runner = HeartbeatRunner::new(config, hygiene_config, workspace, llm);
    if let Some(tx) = response_tx {
@@ -508,6 +512,179 @@ pub fn spawn_heartbeat(
    })
 }

+/// Spawn a multi-user heartbeat runner that cycles through all users that
+/// own routines (enabled or not). Each tick, it queries the DB for distinct
+/// user_ids, creates a per-user workspace, and runs a heartbeat check for
+/// each user concurrently. Per-user failure counts are tracked independently.
+pub fn spawn_multi_user_heartbeat(
+    config: HeartbeatConfig,
+    hygiene_config: HygieneConfig,
+    llm: Arc<dyn LlmProvider>,
+    response_tx: Option<mpsc::Sender<OutgoingResponse>>,
+    store: AdminScope,
+) -> tokio::task::JoinHandle<()> {
+    tokio::spawn(async move {
+        if !config.enabled {
+            tracing::info!("Multi-user heartbeat is disabled");
+            return;
+        }
+
+        let mut tick_interval = if config.fire_at.is_none() {
+            let mut iv = tokio::time::interval(config.interval);
+            iv.tick().await; // skip immediate tick
+            Some(iv)
+        } else {
+            None
+        };
+
+        // Track consecutive failures per user so we can disable heartbeat
+        // for persistently-failing users (same semantics as single-user mode).
+        let mut user_failures: std::collections::HashMap<String, u32> =
+            std::collections::HashMap::new();
+
+        tracing::info!("Starting multi-user heartbeat loop");
+
+        loop {
+            if let Some(fire_at) = config.fire_at {
+                let sleep_dur = duration_until_next_fire(fire_at, config.resolved_tz());
+                tokio::time::sleep(sleep_dur).await;
+            } else if let Some(ref mut iv) = tick_interval {
+                iv.tick().await;
+            }
+
+            if config.is_quiet_hours() {
+                continue;
+            }
+
+            // Get distinct user_ids from routines
+            let user_ids = match store.list_all_routines().await {
+                Ok(routines) => {
+                    let mut ids: Vec<String> = routines
+                        .iter()
+                        .map(|r| r.user_id.clone())
+                        .collect::<std::collections::HashSet<_>>()
+                        .into_iter()
+                        .collect();
+                    ids.sort();
+                    ids
+                }
+                Err(e) => {
+                    tracing::error!("Multi-user heartbeat: failed to list routines: {}", e);
+                    continue;
+                }
+            };
+
+            // Run user heartbeats concurrently so one slow LLM call doesn't
+            // block others. Cap concurrency to avoid flooding the LLM provider.
+            const MAX_CONCURRENT_HEARTBEATS: usize = 8;
+            let mut join_set = tokio::task::JoinSet::new();
+
+            for user_id in &user_ids {
+                // Skip users that have exceeded max_failures
+                let failures = user_failures.get(user_id).copied().unwrap_or(0);
+                if failures >= config.max_failures {
+                    continue;
+                }
+
+                let workspace = Arc::new(Workspace::new_with_db(user_id, Arc::clone(store.db())));
+
+                // Run memory hygiene per user (same as single-user heartbeat).
+                let hygiene_ws = Arc::clone(&workspace);
+                let hygiene_cfg = hygiene_config.clone();
+                let hygiene_user = user_id.clone();
+                tokio::spawn(async move {
+                    let report =
+                        crate::workspace::hygiene::run_if_due(&hygiene_ws, &hygiene_cfg).await;
+                    if report.had_work() {
+                        tracing::info!(
+                            user_id = hygiene_user,
+                            daily_logs_deleted = report.daily_logs_deleted,
+                            conversation_docs_deleted = report.conversation_docs_deleted,
+                            "multi-user heartbeat: memory hygiene deleted stale documents"
+                        );
+                    }
+                });
+
+                // Drain completed tasks to stay within the concurrency cap.
+                while join_set.len() >= MAX_CONCURRENT_HEARTBEATS {
+                    if let Some(join_result) = join_set.join_next().await {
+                        collect_heartbeat_result(join_result, &mut user_failures, &config);
+                    }
+                }
+
+                let uid = user_id.clone();
+                let cfg = config.clone();
+                let hyg = hygiene_config.clone();
+                let llm_clone = llm.clone();
+                let tx = response_tx.clone();
+                let admin = store.clone();
+
+                join_set.spawn(async move {
+                    let mut runner = HeartbeatRunner::new(cfg, hyg, workspace, llm_clone);
+                    if let Some(tx) = tx {
+                        runner = runner.with_response_channel(tx);
+                    }
+                    runner = runner.with_store(admin);
+
+                    let result = runner.check_heartbeat().await;
+                    if let HeartbeatResult::NeedsAttention(msg) = &result {
+                        runner.send_notification(msg).await;
+                    }
+                    (uid, result)
+                });
+            }
+
+            // Collect remaining results and update failure counts
+            while let Some(join_result) = join_set.join_next().await {
+                collect_heartbeat_result(join_result, &mut user_failures, &config);
+            }
+        }
+    })
+}
+
+/// Process a single JoinSet result from the multi-user heartbeat loop.
+fn collect_heartbeat_result(
+    join_result: Result<(String, HeartbeatResult), tokio::task::JoinError>,
+    user_failures: &mut std::collections::HashMap<String, u32>,
+    config: &HeartbeatConfig,
+) {
+    let (uid, result) = match join_result {
+        Ok(pair) => pair,
+        Err(e) => {
+            tracing::error!("Multi-user heartbeat task panicked: {}", e);
+            return;
+        }
+    };
+    match result {
+        HeartbeatResult::Ok => {
+            tracing::trace!(user_id = uid, "Multi-user heartbeat OK");
+            user_failures.remove(&uid);
+        }
+        HeartbeatResult::NeedsAttention(_) => {
+            tracing::info!(user_id = uid, "Multi-user heartbeat needs attention");
+            user_failures.remove(&uid);
+        }
+        HeartbeatResult::Skipped => {}
+        HeartbeatResult::Failed(err) => {
+            let count = user_failures.entry(uid.clone()).or_insert(0);
+            *count += 1;
+            tracing::error!(
+                user_id = uid,
+                consecutive_failures = *count,
+                "Multi-user heartbeat failed: {}",
+                err
+            );
+            if *count >= config.max_failures {
+                tracing::error!(
+                    user_id = uid,
+                    "Multi-user heartbeat disabled for user after {} consecutive failures",
+                    count
+                );
+            }
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
    use super::*;
@@ -726,7 +903,7 @@ mod tests {
            Arc<crate::workspace::Workspace>,
            Arc<dyn crate::llm::LlmProvider>,
            Option<tokio::sync::mpsc::Sender<crate::channels::OutgoingResponse>>,
-            Option<Arc<dyn crate::db::Database>>,
+            Option<AdminScope>,
        ) -> tokio::task::JoinHandle<()> = spawn_heartbeat;
        let _ = _fn_ptr;
    }
@@ -36,7 +36,9 @@ pub(crate) use agent_loop::truncate_for_preview;
 pub use agent_loop::{Agent, AgentDeps};
 pub use compaction::{CompactionResult, ContextCompactor};
 pub use context_monitor::{CompactionStrategy, ContextBreakdown, ContextMonitor};
-pub use heartbeat::{HeartbeatConfig, HeartbeatResult, HeartbeatRunner, spawn_heartbeat};
+pub use heartbeat::{
+    HeartbeatConfig, HeartbeatResult, HeartbeatRunner, spawn_heartbeat, spawn_multi_user_heartbeat,
+};
 pub use router::{MessageIntent, Router};
 pub use routine::{Routine, RoutineAction, RoutineRun, Trigger};
 pub use routine_engine::{RoutineEngine, SandboxReadiness};
@@ -28,12 +28,12 @@ use crate::agent::routine::{
 use crate::channels::{IncomingMessage, OutgoingResponse};
 use crate::config::RoutineConfig;
 use crate::context::{JobContext, JobState};
-use crate::db::Database;
 use crate::error::RoutineError;
 use crate::extensions::ExtensionManager;
 use crate::llm::{
    ChatMessage, CompletionRequest, FinishReason, LlmProvider, ToolCall, ToolCompletionRequest,
 };
+use crate::tenant::AdminScope;
 use crate::tools::{
    ToolError, ToolRegistry, autonomous_allowed_tool_names, autonomous_unavailable_message,
    prepare_tool_params,
@@ -99,7 +99,7 @@ pub(crate) fn routine_matches_message(routine: &Routine, message: &IncomingMessa
 /// The routine execution engine.
 pub struct RoutineEngine {
    config: RoutineConfig,
-    store: Arc<dyn Database>,
+    store: AdminScope,
    llm: Arc<dyn LlmProvider>,
    workspace: Arc<Workspace>,
    /// Sender for notifications (routed to channel manager).
@@ -128,7 +128,7 @@ impl RoutineEngine {
    #[allow(clippy::too_many_arguments)]
    pub fn new(
        config: RoutineConfig,
-        store: Arc<dyn Database>,
+        store: AdminScope,
        llm: Arc<dyn LlmProvider>,
        workspace: Arc<Workspace>,
        notify_tx: mpsc::Sender<OutgoingResponse>,
@@ -782,12 +782,22 @@ impl RoutineEngine {
            });
        }

+        // Per-user workspace (same pattern as spawn_fire).
+        let routine_workspace = if routine.user_id == self.workspace.user_id() {
+            self.workspace.clone()
+        } else {
+            Arc::new(Workspace::new_with_db(
+                &routine.user_id,
+                Arc::clone(self.store.db()),
+            ))
+        };
+
        // Execute inline for manual triggers (caller wants to wait)
        let engine = EngineContext {
            config: self.config.clone(),
            store: self.store.clone(),
            llm: self.llm.clone(),
-            workspace: self.workspace.clone(),
+            workspace: routine_workspace,
            notify_tx: self.notify_tx.clone(),
            running_count: self.running_count.clone(),
            scheduler: self.scheduler.clone(),
@@ -910,11 +920,23 @@ impl RoutineEngine {
            created_at: Utc::now(),
        };

+        // Use per-user workspace so each routine executes in the correct
+        // user's context. Fall back to the engine-wide workspace when the
+        // routine belongs to the same user (avoids unnecessary allocation).
+        let routine_workspace = if routine.user_id == self.workspace.user_id() {
+            self.workspace.clone()
+        } else {
+            Arc::new(Workspace::new_with_db(
+                &routine.user_id,
+                Arc::clone(self.store.db()),
+            ))
+        };
+
        let engine = EngineContext {
            config: self.config.clone(),
            store: self.store.clone(),
            llm: self.llm.clone(),
-            workspace: self.workspace.clone(),
+            workspace: routine_workspace,
            notify_tx: self.notify_tx.clone(),
            running_count: self.running_count.clone(),
            scheduler: self.scheduler.clone(),
@@ -967,7 +989,7 @@ impl RoutineEngine {
 /// an active state (Pending/InProgress/Stuck). Maps the final `JobState` to
 /// a `RunStatus` for the routine run.
 struct FullJobWatcher {
-    store: Arc<dyn Database>,
+    store: AdminScope,
    job_id: Uuid,
    routine_name: String,
 }
@@ -978,7 +1000,7 @@ impl FullJobWatcher {
    /// Safety ceiling: 24 hours, derived from POLL_INTERVAL.
    const MAX_POLLS: u32 = (24 * 60 * 60) / Self::POLL_INTERVAL.as_secs() as u32;

-    fn new(store: Arc<dyn Database>, job_id: Uuid, routine_name: String) -> Self {
+    fn new(store: AdminScope, job_id: Uuid, routine_name: String) -> Self {
        Self {
            store,
            job_id,
@@ -1050,7 +1072,7 @@ impl FullJobWatcher {
 /// Shared context passed to the execution function.
 struct EngineContext {
    config: RoutineConfig,
-    store: Arc<dyn Database>,
+    store: AdminScope,
    llm: Arc<dyn LlmProvider>,
    workspace: Arc<Workspace>,
    notify_tx: mpsc::Sender<OutgoingResponse>,
@@ -11,12 +11,12 @@ use uuid::Uuid;
 use crate::agent::task::{Task, TaskContext, TaskOutput};
 use crate::config::AgentConfig;
 use crate::context::{ContextManager, JobContext, JobState};
-use crate::db::Database;
 use crate::error::{Error, JobError};
 use crate::extensions::ExtensionManager;
 use crate::hooks::HookRegistry;
 use crate::llm::LlmProvider;
 use crate::safety::SafetyLayer;
+use crate::tenant::AdminScope;
 use crate::tools::{
    ApprovalContext, ToolRegistry, autonomous_allowed_tool_names, autonomous_unavailable_error,
    prepare_tool_params,
@@ -52,7 +52,7 @@ struct ScheduledSubtask {
 pub struct SchedulerDeps {
    pub tools: Arc<ToolRegistry>,
    pub extension_manager: Option<Arc<ExtensionManager>>,
-    pub store: Option<Arc<dyn Database>>,
+    pub store: Option<AdminScope>,
    pub hooks: Arc<HookRegistry>,
 }

@@ -64,7 +64,7 @@ pub struct Scheduler {
    safety: Arc<SafetyLayer>,
    tools: Arc<ToolRegistry>,
    extension_manager: Option<Arc<ExtensionManager>>,
-    store: Option<Arc<dyn Database>>,
+    store: Option<AdminScope>,
    hooks: Arc<HookRegistry>,
    /// SSE manager for live job event streaming.
    sse_tx: Option<Arc<crate::channels::web::sse::SseManager>>,
@@ -780,10 +780,14 @@ mod tests {
            allow_local_tools: true,
            max_cost_per_day_cents: None,
            max_actions_per_hour: None,
+            max_cost_per_user_per_day_cents: None,
            max_tool_iterations: 10,
            auto_approve_tools: true,
            default_timezone: "UTC".to_string(),
            max_tokens_per_job,
+            multi_tenant: false,
+            max_llm_concurrent_per_user: None,
+            max_jobs_concurrent_per_user: None,
        };
        let cm = Arc::new(ContextManager::new(5));
        let llm: Arc<dyn LlmProvider> = Arc::new(StubLlm);
@@ -8,8 +8,8 @@ use chrono::{DateTime, Utc};
 use uuid::Uuid;

 use crate::context::{ContextManager, JobState};
-use crate::db::Database;
 use crate::error::RepairError;
+use crate::tenant::AdminScope;
 use crate::tools::{BuildRequirement, Language, SoftwareBuilder, SoftwareType, ToolRegistry};

 /// A job that has been detected as stuck.
@@ -69,7 +69,7 @@ pub struct DefaultSelfRepair {
    /// Jobs in `InProgress` longer than this are treated as stuck.
    stuck_threshold: Duration,
    max_repair_attempts: u32,
-    store: Option<Arc<dyn Database>>,
+    store: Option<AdminScope>,
    builder: Option<Arc<dyn SoftwareBuilder>>,
    tools: Option<Arc<ToolRegistry>>,
 }
@@ -91,8 +91,8 @@ impl DefaultSelfRepair {
        }
    }

-    /// Add a Store for tool failure tracking.
-    pub fn with_store(mut self, store: Arc<dyn Database>) -> Self {
+    /// Add an admin-scoped store for tool failure tracking.
+    pub fn with_store(mut self, store: AdminScope) -> Self {
        self.store = Some(store);
        self
    }
@@ -806,7 +806,7 @@ mod tests {
        // Create self-repair with zero threshold (detect immediately),
        // wired with store, builder, and tools.
        let repair = DefaultSelfRepair::new(Arc::clone(&cm), Duration::from_secs(0), 3)
-            .with_store(Arc::clone(&db))
+            .with_store(crate::tenant::AdminScope::new(Arc::clone(&db)))
            .with_builder(
                Arc::clone(&builder) as Arc<dyn crate::tools::SoftwareBuilder>,
                tools,
@@ -175,6 +175,7 @@ impl Agent {
    pub(super) async fn process_user_input(
        &self,
        message: &IncomingMessage,
+        tenant: crate::tenant::TenantCtx,
        session: Arc<Mutex<Session>>,
        thread_id: Uuid,
        content: &str,
@@ -351,7 +352,7 @@ impl Agent {

        if let Some(intent) = self.router.route_command(&temp_message) {
            // Explicit command like /status, /job, /list - handle directly
-            return self.handle_job_or_command(intent, message).await;
+            return self.handle_job_or_command(intent, message, &tenant).await;
        }

        // Natural language goes through the agentic loop
@@ -462,7 +463,7 @@ impl Agent {

        // Run the agentic tool execution loop
        let result = self
-            .run_agentic_loop(message, session.clone(), thread_id, turn_messages)
+            .run_agentic_loop(message, tenant, session.clone(), thread_id, turn_messages)
            .await;

        // Re-acquire lock and check if interrupted
@@ -1473,7 +1474,13 @@ impl Agent {

            // Continue the agentic loop (a tool was already executed this turn)
            let result = self
-                .run_agentic_loop(message, session.clone(), thread_id, context_messages)
+                .run_agentic_loop(
+                    message,
+                    self.tenant_ctx(&message.user_id).await,
+                    session.clone(),
+                    thread_id,
+                    context_messages,
+                )
                .await;

            // Handle the result