Merge branch 'staging' into feat/wechat-integration

src/agent/agent_loop.rs

@@ -13,9 +13,10 @@ 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;
 use crate::agent::session_manager::SessionManager;
 use crate::agent::submission::{Submission, SubmissionParser, SubmissionResult};
 use crate::agent::{HeartbeatConfig as AgentHeartbeatConfig, Router, Scheduler, SchedulerDeps};
@@ -84,6 +85,15 @@ fn resolve_owner_scope_notification_user(
     trimmed_option(explicit_user).or_else(|| trimmed_option(owner_fallback))
 }
 
+fn is_single_message_repl(message: &IncomingMessage) -> bool {
+    message.channel == "repl"
+        && message
+            .metadata
+            .get("single_message_mode")
+            .and_then(|value| value.as_bool())
+            .unwrap_or(false)
+}
+
 async fn resolve_channel_notification_user(
     extension_manager: Option<&Arc<ExtensionManager>>,
     channel: Option<&str>,
@@ -172,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.
@@ -234,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(),
             },
         );
@@ -315,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()
     }
@@ -400,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()));
@@ -508,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()
@@ -537,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
+                                            );
+                                        }
                                     }
                                 }
                             }
@@ -573,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
@@ -602,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,
@@ -1140,9 +1237,14 @@ impl Agent {
             && let Submission::UserInput { ref content } = submission
             && let Some(engine) = self.routine_engine().await
         {
+            let single_message_repl = is_single_message_repl(message);
             // Use post-hook content so that BeforeInbound hooks that rewrite
             // input are respected by event trigger matching.
-            let fired = engine.check_event_triggers(message, content).await;
+            let fired = if single_message_repl {
+                engine.check_event_triggers_and_wait(message, content).await
+            } else {
+                engine.check_event_triggers(message, content).await
+            };
             if fired > 0 {
                 tracing::debug!(
                     channel = %message.channel,
@@ -1150,15 +1252,30 @@ impl Agent {
                     fired,
                     "Consumed inbound user message with matching event-triggered routine(s)"
                 );
-                return Ok(Some(String::new()));
+                return if single_message_repl {
+                    Ok(None)
+                } else {
+                    Ok(Some(String::new()))
+                };
             }
         }
 
+        // 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.
@@ -1225,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
@@ -1250,8 +1373,30 @@ impl Agent {
                     command,
                     message.channel
                 );
+                // /reasoning is special-cased here (not in handle_system_command)
+                // because it needs the session + thread_id to read turn reasoning
+                // data, which handle_system_command's signature doesn't provide.
+                if command == "reasoning" {
+                    let result = self
+                        .handle_reasoning_command(&args, &session, thread_id)
+                        .await;
+                    return match result {
+                        SubmissionResult::Response { content } => Ok(Some(content)),
+                        SubmissionResult::Ok { message } => Ok(message),
+                        SubmissionResult::Error { message } => {
+                            Ok(Some(format!("Error: {}", message)))
+                        }
+                        _ => {
+                            if is_single_message_repl(message) {
+                                Ok(None)
+                            } else {
+                                Ok(Some(String::new()))
+                            }
+                        }
+                    };
+                }
                 // 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,
@@ -1264,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
@@ -1309,7 +1451,26 @@ impl Agent {
                     Ok(Some(content))
                 }
             }
-            SubmissionResult::Ok { message } => Ok(message),
+            SubmissionResult::Ok {
+                message: output_message,
+            } => {
+                let should_exit =
+                    if output_message.as_deref() == Some("") && is_single_message_repl(message) {
+                        let sess = session_for_empty_exit.lock().await;
+                        sess.threads
+                            .get(&thread_id)
+                            .map(|thread| thread.state != ThreadState::AwaitingApproval)
+                            .unwrap_or(true)
+                    } else {
+                        false
+                    };
+
+                if should_exit {
+                    Ok(None)
+                } else {
+                    Ok(output_message)
+                }
+            }
             SubmissionResult::Error { message } => Ok(Some(format!("Error: {}", message))),
             SubmissionResult::Interrupted => Ok(Some("Interrupted.".into())),
             SubmissionResult::NeedApproval { .. } => {
@@ -1325,7 +1486,7 @@ impl Agent {
 #[cfg(test)]
 mod tests {
     use super::{
-        chat_tool_execution_metadata, resolve_routine_notification_user,
+        chat_tool_execution_metadata, is_single_message_repl, resolve_routine_notification_user,
         should_fallback_routine_notification, truncate_for_preview,
     };
     use crate::channels::IncomingMessage;
@@ -1487,4 +1648,17 @@ mod tests {
 
         assert!(should_fallback_routine_notification(&error)); // safety: test-only assertion
     }
+
+    #[test]
+    fn single_message_repl_detection_requires_repl_channel_and_metadata_flag() {
+        let repl = IncomingMessage::new("repl", "owner-scope", "hello")
+            .with_metadata(serde_json::json!({ "single_message_mode": true }));
+        let gateway = IncomingMessage::new("gateway", "owner-scope", "hello")
+            .with_metadata(serde_json::json!({ "single_message_mode": true }));
+        let plain_repl = IncomingMessage::new("repl", "owner-scope", "hello");
+
+        assert!(is_single_message_repl(&repl)); // safety: test-only assertion
+        assert!(!is_single_message_repl(&gateway)); // safety: test-only assertion
+        assert!(!is_single_message_repl(&plain_repl)); // safety: test-only assertion
+    }
 }

src/agent/agentic_loop.rs

@@ -10,7 +10,7 @@ use std::borrow::Cow;
 
 use crate::agent::session::PendingApproval;
 use crate::error::Error;
-use crate::llm::{ChatMessage, Reasoning, ReasoningContext, RespondResult};
+use crate::llm::{ChatMessage, FinishReason, Reasoning, ReasoningContext, RespondResult};
 
 /// Signal from the delegate indicating how the loop should proceed.
 pub enum LoopSignal {
@@ -134,6 +134,9 @@ pub async fn run_agentic_loop(
     config: &AgenticLoopConfig,
 ) -> Result<LoopOutcome, Error> {
     let mut consecutive_tool_intent_nudges: u32 = 0;
+    // Accumulates across all iterations (not reset by text responses) so
+    // non-consecutive truncations still escalate to force_text.
+    let mut truncation_count: u32 = 0;
 
     for iteration in 1..=config.max_iterations {
         // Check for external signals (stop, cancellation, user messages)
@@ -215,7 +218,35 @@ pub async fn run_agentic_loop(
                 tool_calls,
                 content,
             } => {
+                // If the response was truncated, tool call parameters are likely
+                // incomplete. Discard them and tell the LLM to try a different
+                // approach rather than executing malformed tool calls.
+                if output.finish_reason == FinishReason::Length {
+                    truncation_count += 1;
+                    let names: Vec<&str> = tool_calls.iter().map(|tc| tc.name.as_str()).collect();
+                    tracing::warn!(
+                        iteration,
+                        tools = ?names,
+                        truncation_count,
+                        "Discarding truncated tool calls (finish_reason=Length)"
+                    );
+                    if let Some(ref text) = content {
+                        reason_ctx.messages.push(ChatMessage::assistant(text));
+                    }
+                    reason_ctx
+                        .messages
+                        .push(ChatMessage::user(crate::llm::TRUNCATED_TOOL_CALL_NOTICE));
+                    // After repeated truncations, force text-only mode so the LLM
+                    // stops attempting tool calls it can't fit in the output budget.
+                    if truncation_count >= 3 {
+                        reason_ctx.force_text = true;
+                    }
+                    delegate.after_iteration(iteration).await;
+                    continue;
+                }
+
                 consecutive_tool_intent_nudges = 0;
+                truncation_count = 0;
 
                 if let Some(outcome) = delegate
                     .execute_tool_calls(tool_calls, content, reason_ctx)
@@ -271,6 +302,7 @@ mod tests {
         RespondOutput {
             result: RespondResult::Text(text.to_string()),
             usage: zero_usage(),
+            finish_reason: FinishReason::Stop,
         }
     }
 
@@ -281,6 +313,7 @@ mod tests {
                 content: None,
             },
             usage: zero_usage(),
+            finish_reason: FinishReason::ToolUse,
         }
     }
 
@@ -414,6 +447,7 @@ mod tests {
             id: "call_1".to_string(),
             name: "echo".to_string(),
             arguments: serde_json::json!({}),
+            reasoning: None,
         };
         let delegate = MockDelegate::new(vec![
             tool_calls_output(vec![tool_call]),
@@ -621,4 +655,95 @@ mod tests {
         let result = truncate_for_preview("café", 4);
         assert_eq!(result, "caf...");
     }
+
+    #[tokio::test]
+    async fn test_truncated_tool_calls_discarded_on_length() {
+        let truncated_tool_call = ToolCall {
+            id: "call_1".to_string(),
+            name: "memory_write".to_string(),
+            arguments: serde_json::json!({}), // empty — truncated
+            reasoning: None,
+        };
+        let truncated_output = RespondOutput {
+            result: RespondResult::ToolCalls {
+                tool_calls: vec![truncated_tool_call],
+                content: Some("I'll write the report.".to_string()),
+            },
+            usage: zero_usage(),
+            finish_reason: FinishReason::Length, // response was truncated
+        };
+        let delegate = MockDelegate::new(vec![truncated_output, text_output("Summarized it.")]);
+        let reasoning = stub_reasoning();
+        let mut ctx = ReasoningContext::new();
+        let config = AgenticLoopConfig {
+            max_iterations: 5,
+            ..Default::default()
+        };
+
+        let outcome = run_agentic_loop(&delegate, &reasoning, &mut ctx, &config)
+            .await
+            .unwrap();
+
+        // Tool calls should NOT have been executed
+        assert_eq!(delegate.tool_exec_count.load(Ordering::SeqCst), 0);
+        // The loop should have continued and returned the text response
+        assert!(matches!(outcome, LoopOutcome::Response(ref t) if t == "Summarized it."));
+        // A truncation notice should have been injected into context
+        assert!(
+            ctx.messages
+                .iter()
+                .any(|m| m.role == crate::llm::Role::User && m.content.contains("truncated")),
+            "Should inject truncation notice into context"
+        );
+        // The partial assistant content should have been preserved
+        assert!(
+            ctx.messages
+                .iter()
+                .any(|m| m.role == crate::llm::Role::Assistant
+                    && m.content.contains("write the report")),
+            "Should preserve partial assistant content"
+        );
+    }
+
+    #[tokio::test]
+    async fn test_repeated_truncations_force_text_mode() {
+        let make_truncated = || RespondOutput {
+            result: RespondResult::ToolCalls {
+                tool_calls: vec![ToolCall {
+                    id: "call_1".to_string(),
+                    name: "memory_write".to_string(),
+                    arguments: serde_json::json!({}),
+                    reasoning: None,
+                }],
+                content: None,
+            },
+            usage: zero_usage(),
+            finish_reason: FinishReason::Length,
+        };
+        // Three truncated responses, then a text response
+        let delegate = MockDelegate::new(vec![
+            make_truncated(),
+            make_truncated(),
+            make_truncated(),
+            text_output("Gave up on tool calls."),
+        ]);
+        let reasoning = stub_reasoning();
+        let mut ctx = ReasoningContext::new();
+        let config = AgenticLoopConfig {
+            max_iterations: 5,
+            ..Default::default()
+        };
+
+        let outcome = run_agentic_loop(&delegate, &reasoning, &mut ctx, &config)
+            .await
+            .unwrap();
+
+        assert!(matches!(outcome, LoopOutcome::Response(_)));
+        assert_eq!(delegate.tool_exec_count.load(Ordering::SeqCst), 0);
+        // After 3 truncations, force_text should be set
+        assert!(
+            ctx.force_text,
+            "Should escalate to force_text after repeated truncations"
+        );
+    }
 }

src/agent/commands.rs

@@ -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))),
         }
@@ -465,12 +472,101 @@ impl Agent {
         }
     }
 
+    /// Handle `/reasoning [N|all]` — show reasoning history for the active thread.
+    pub(super) async fn handle_reasoning_command(
+        &self,
+        args: &[String],
+        session: &Arc<Mutex<Session>>,
+        thread_id: Uuid,
+    ) -> SubmissionResult {
+        // Clone the turn data we need, then drop the session lock.
+        let turns_snapshot: Vec<(
+            usize,
+            Option<String>,
+            Vec<crate::agent::session::TurnToolCall>,
+        )>;
+        {
+            let sess = session.lock().await;
+            let thread = match sess.threads.get(&thread_id) {
+                Some(t) => t,
+                None => return SubmissionResult::error("No active thread."),
+            };
+
+            if thread.turns.is_empty() {
+                return SubmissionResult::ok_with_message("No turns yet.");
+            }
+
+            // Parse argument: default=last turn, "all"=all turns, N=specific turn (1-based).
+            let selected: Vec<&crate::agent::session::Turn> = match args.first().map(|s| s.as_str())
+            {
+                Some("all") => thread.turns.iter().collect(),
+                Some(n) => match n.parse::<usize>() {
+                    Ok(0) => return SubmissionResult::error("Turn numbers start at 1."),
+                    Ok(num) if num > thread.turns.len() => {
+                        return SubmissionResult::error(format!(
+                            "Turn {} does not exist (max: {}).",
+                            num,
+                            thread.turns.len()
+                        ));
+                    }
+                    Ok(num) => vec![&thread.turns[num - 1]],
+                    Err(_) => return SubmissionResult::error("Usage: /reasoning [N|all]"),
+                },
+                None => {
+                    // Default: last turn that has tool calls
+                    match thread.turns.iter().rev().find(|t| !t.tool_calls.is_empty()) {
+                        Some(t) => vec![t],
+                        None => {
+                            return SubmissionResult::ok_with_message("No turns with tool calls.");
+                        }
+                    }
+                }
+            };
+
+            turns_snapshot = selected
+                .into_iter()
+                .map(|t| (t.turn_number, t.narrative.clone(), t.tool_calls.clone()))
+                .collect();
+        }
+        // Session lock is now dropped — format output without holding it.
+
+        let mut output = String::new();
+        for (turn_number, narrative, tool_calls) in &turns_snapshot {
+            output.push_str(&format!("--- Turn {} ---\n", turn_number + 1));
+            if let Some(narrative) = narrative {
+                output.push_str(&format!("Reasoning: {}\n", narrative));
+            }
+            if tool_calls.is_empty() {
+                output.push_str("  (no tool calls)\n");
+            } else {
+                for tc in tool_calls {
+                    let status = if tc.error.is_some() {
+                        "error"
+                    } else if tc.result.is_some() {
+                        "ok"
+                    } else {
+                        "pending"
+                    };
+                    output.push_str(&format!("  {} [{}]", tc.name, status));
+                    if let Some(ref rationale) = tc.rationale {
+                        output.push_str(&format!(" — {}", rationale));
+                    }
+                    output.push('\n');
+                }
+            }
+            output.push('\n');
+        }
+
+        SubmissionResult::response(output.trim_end())
+    }
+
     /// Handle system commands that bypass thread-state checks entirely.
     pub(super) async fn handle_system_command(
         &self,
         command: &str,
         args: &[String],
         channel: &str,
+        tenant: &crate::tenant::TenantCtx,
     ) -> Result<SubmissionResult, Error> {
         match command {
             "help" => Ok(SubmissionResult::response(concat!(
@@ -480,6 +576,7 @@ impl Agent {
                 "  /version          Show version info\n",
                 "  /tools            List available tools\n",
                 "  /debug            Toggle debug mode\n",
+                "  /reasoning [N|all] Show agent reasoning for turns\n",
                 "  /ping             Connectivity check\n",
                 "\n",
                 "Jobs:\n",
@@ -663,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
-                        ))),
                     }
                 }
             }
@@ -817,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))),
@@ -832,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 || {

src/agent/cost_guard.rs

@@ -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"));
+    }
 }

src/agent/dispatcher.rs

@@ -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>,
@@ -63,7 +64,12 @@ impl Agent {
         );
 
         let system_prompt = if let Some(ws) = self.workspace() {
-            match ws
+            let scoped_workspace = if ws.user_id() == message.user_id {
+                Arc::clone(ws)
+            } else {
+                Arc::new(ws.scoped_to_user(&message.user_id))
+            };
+            match scoped_workspace
                 .system_prompt_for_context_tz(is_group_chat, user_tz)
                 .await
             {
@@ -163,6 +169,7 @@ impl Agent {
 
         let delegate = ChatDelegate {
             agent: self,
+            tenant,
             session: session.clone(),
             thread_id,
             message,
@@ -235,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,
@@ -297,6 +305,8 @@ impl<'a> LoopDelegate for ChatDelegate<'a> {
         };
         // Update context for this iteration
         reason_ctx.available_tools = tool_defs;
+        // Preserve force_text if already set (e.g. by truncation escalation).
+        let force_text = force_text || reason_ctx.force_text;
         reason_ctx.system_prompt = Some(if force_text {
             self.cached_prompt_no_tools.clone()
         } else {
@@ -330,8 +340,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(),
@@ -339,6 +349,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 }) => {
@@ -373,13 +398,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,
@@ -388,7 +422,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!(
@@ -419,6 +453,19 @@ impl<'a> LoopDelegate for ChatDelegate<'a> {
         content: Option<String>,
         reason_ctx: &mut ReasoningContext,
     ) -> Result<Option<LoopOutcome>, Error> {
+        // Extract and sanitize the narrative before consuming `content`.
+        let narrative = content
+            .as_deref()
+            .filter(|c| !c.trim().is_empty())
+            .map(|c| {
+                let sanitized = self
+                    .agent
+                    .safety()
+                    .sanitize_tool_output("agent_narrative", c);
+                sanitized.content
+            })
+            .filter(|c| !c.trim().is_empty());
+
         // Add the assistant message with tool_calls to context.
         // OpenAI protocol requires this before tool-result messages.
         reason_ctx
@@ -439,6 +486,41 @@ impl<'a> LoopDelegate for ChatDelegate<'a> {
             )
             .await;
 
+        // Build per-tool decisions for the reasoning update.
+        // Sanitize each rationale through SafetyLayer (parity with JobDelegate).
+        let decisions: Vec<crate::channels::ToolDecision> = tool_calls
+            .iter()
+            .filter_map(|tc| {
+                tc.reasoning.as_ref().map(|r| {
+                    let sanitized = self
+                        .agent
+                        .safety()
+                        .sanitize_tool_output("tool_rationale", r)
+                        .content;
+                    crate::channels::ToolDecision {
+                        tool_name: tc.name.clone(),
+                        rationale: sanitized,
+                    }
+                })
+            })
+            .collect();
+
+        // Emit reasoning update to channels.
+        if narrative.is_some() || !decisions.is_empty() {
+            let _ = self
+                .agent
+                .channels
+                .send_status(
+                    &self.message.channel,
+                    StatusUpdate::ReasoningUpdate {
+                        narrative: narrative.clone().unwrap_or_default(),
+                        decisions: decisions.clone(),
+                    },
+                    &self.message.metadata,
+                )
+                .await;
+        }
+
         // Record tool calls in the thread with sensitive params redacted.
         {
             let mut redacted_args: Vec<serde_json::Value> = Vec::with_capacity(tool_calls.len());
@@ -454,8 +536,23 @@ impl<'a> LoopDelegate for ChatDelegate<'a> {
             if let Some(thread) = sess.threads.get_mut(&self.thread_id)
                 && let Some(turn) = thread.last_turn_mut()
             {
+                // Set turn-level narrative.
+                if turn.narrative.is_none() {
+                    turn.narrative = narrative;
+                }
                 for (tc, safe_args) in tool_calls.iter().zip(redacted_args) {
-                    turn.record_tool_call(&tc.name, safe_args);
+                    let sanitized_rationale = tc.reasoning.as_ref().map(|r| {
+                        self.agent
+                            .safety()
+                            .sanitize_tool_output("tool_rationale", r)
+                            .content
+                    });
+                    turn.record_tool_call_with_reasoning(
+                        &tc.name,
+                        safe_args,
+                        sanitized_rationale,
+                        Some(tc.id.clone()),
+                    );
                 }
             }
         }
@@ -725,7 +822,7 @@ impl<'a> LoopDelegate for ChatDelegate<'a> {
                         if let Some(thread) = sess.threads.get_mut(&self.thread_id)
                             && let Some(turn) = thread.last_turn_mut()
                         {
-                            turn.record_tool_error(error_msg.clone());
+                            turn.record_tool_error_for(&tc.id, error_msg.clone());
                         }
                     }
                     reason_ctx
@@ -851,16 +948,19 @@ impl<'a> LoopDelegate for ChatDelegate<'a> {
                         Err(e) => format!("Tool '{}' failed: {}", tc.name, e),
                     };
 
-                    // Record sanitized result in thread
+                    // Record sanitized result in thread (identity-based matching).
                     {
                         let mut sess = self.session.lock().await;
                         if let Some(thread) = sess.threads.get_mut(&self.thread_id)
                             && let Some(turn) = thread.last_turn_mut()
                         {
                             if is_tool_error {
-                                turn.record_tool_error(result_content.clone());
+                                turn.record_tool_error_for(&tc.id, result_content.clone());
                             } else {
-                                turn.record_tool_result(serde_json::json!(result_content));
+                                turn.record_tool_result_for(
+                                    &tc.id,
+                                    serde_json::json!(result_content),
+                                );
                             }
                         }
                     }
@@ -1233,6 +1333,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(
@@ -1248,10 +1349,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()),
@@ -1461,11 +1566,13 @@ mod tests {
                     id: "call_2".to_string(),
                     name: "http".to_string(),
                     arguments: serde_json::json!({"url": "https://example.com"}),
+                    reasoning: None,
                 },
                 ToolCall {
                     id: "call_3".to_string(),
                     name: "echo".to_string(),
                     arguments: serde_json::json!({"message": "done"}),
+                    reasoning: None,
                 },
             ],
             user_timezone: None,
@@ -1651,6 +1758,7 @@ mod tests {
                     id: "call_1".to_string(),
                     name: "echo".to_string(),
                     arguments: serde_json::json!({"message": "hi"}),
+                    reasoning: None,
                 }],
             ),
             ChatMessage::tool_result("call_1", "echo", "hi"),
@@ -1743,11 +1851,13 @@ mod tests {
                         id: "c1".to_string(),
                         name: "http".to_string(),
                         arguments: serde_json::json!({}),
+                        reasoning: None,
                     },
                     ToolCall {
                         id: "c2".to_string(),
                         name: "echo".to_string(),
                         arguments: serde_json::json!({}),
+                        reasoning: None,
                     },
                 ],
             ),
@@ -1781,6 +1891,7 @@ mod tests {
                     id: "c1".to_string(),
                     name: "echo".to_string(),
                     arguments: serde_json::json!({}),
+                    reasoning: None,
                 }],
             ),
             ChatMessage::tool_result("c1", "echo", "done"),
@@ -1911,6 +2022,7 @@ mod tests {
                     id: crate::llm::generate_tool_call_id(0, 0),
                     name: "echo".to_string(),
                     arguments: serde_json::json!({"message": "looping"}),
+                    reasoning: None,
                 }],
                 input_tokens: 0,
                 output_tokens: 5,
@@ -2064,6 +2176,7 @@ mod tests {
                     id: crate::llm::generate_tool_call_id(0, 0),
                     name: "nonexistent_tool".to_string(),
                     arguments: serde_json::json!({}),
+                    reasoning: None,
                 }],
                 input_tokens: 0,
                 output_tokens: 5,
@@ -2101,6 +2214,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(
@@ -2116,10 +2230,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()),
@@ -2154,13 +2272,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;
 
@@ -2222,6 +2341,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(
@@ -2237,10 +2357,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()),
@@ -2260,13 +2384,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;
 

src/agent/heartbeat.rs

@@ -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;
     }

src/agent/job_monitor.rs

@@ -21,8 +21,8 @@ use tokio::task::JoinHandle;
 use uuid::Uuid;
 
 use crate::channels::IncomingMessage;
-use crate::channels::web::types::SseEvent;
 use crate::context::{ContextManager, JobState};
+use ironclaw_common::AppEvent;
 
 /// Route context for forwarding job monitor events back to the user's channel.
 #[derive(Debug, Clone)]
@@ -36,15 +36,15 @@ pub struct JobMonitorRoute {
 /// injects assistant messages into the agent loop.
 ///
 /// The monitor forwards:
-/// - `SseEvent::JobMessage` (assistant role): injected as incoming messages so
+/// - `AppEvent::JobMessage` (assistant role): injected as incoming messages so
 ///   the main agent can read and relay to the user.
-/// - `SseEvent::JobResult`: injected as a completion notice, then the task exits.
+/// - `AppEvent::JobResult`: injected as a completion notice, then the task exits.
 ///
 /// Tool use/result and status events are intentionally skipped (too noisy for
 /// the main agent's context window).
 pub fn spawn_job_monitor(
     job_id: Uuid,
-    event_rx: broadcast::Receiver<(Uuid, String, SseEvent)>,
+    event_rx: broadcast::Receiver<(Uuid, String, AppEvent)>,
     inject_tx: mpsc::Sender<IncomingMessage>,
     route: JobMonitorRoute,
 ) -> JoinHandle<()> {
@@ -56,7 +56,7 @@ pub fn spawn_job_monitor(
 /// jobs don't stay `InProgress` forever in the `ContextManager`.
 pub fn spawn_job_monitor_with_context(
     job_id: Uuid,
-    mut event_rx: broadcast::Receiver<(Uuid, String, SseEvent)>,
+    mut event_rx: broadcast::Receiver<(Uuid, String, AppEvent)>,
     inject_tx: mpsc::Sender<IncomingMessage>,
     route: JobMonitorRoute,
     context_manager: Option<Arc<ContextManager>>,
@@ -74,7 +74,7 @@ pub fn spawn_job_monitor_with_context(
                     }
 
                     match event {
-                        SseEvent::JobMessage { role, content, .. } if role == "assistant" => {
+                        AppEvent::JobMessage { role, content, .. } if role == "assistant" => {
                             let mut msg = IncomingMessage::new(
                                 route.channel.clone(),
                                 route.user_id.clone(),
@@ -92,7 +92,7 @@ pub fn spawn_job_monitor_with_context(
                                 break;
                             }
                         }
-                        SseEvent::JobResult { status, .. } => {
+                        AppEvent::JobResult { status, .. } => {
                             // Transition in-memory state so the job frees its
                             // max_jobs slot and query tools show the final state.
                             if let Some(ref cm) = context_manager {
@@ -162,7 +162,7 @@ pub fn spawn_job_monitor_with_context(
 /// inject messages into) but we still need to free the `max_jobs` slot.
 pub fn spawn_completion_watcher(
     job_id: Uuid,
-    mut event_rx: broadcast::Receiver<(Uuid, String, SseEvent)>,
+    mut event_rx: broadcast::Receiver<(Uuid, String, AppEvent)>,
     context_manager: Arc<ContextManager>,
 ) -> JoinHandle<()> {
     let short_id = job_id.to_string()[..8].to_string();
@@ -170,7 +170,7 @@ pub fn spawn_completion_watcher(
     tokio::spawn(async move {
         loop {
             match event_rx.recv().await {
-                Ok((ev_job_id, _user_id, SseEvent::JobResult { status, .. }))
+                Ok((ev_job_id, _user_id, AppEvent::JobResult { status, .. }))
                     if ev_job_id == job_id =>
                 {
                     let target = if status == "completed" {
@@ -229,7 +229,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_monitor_forwards_assistant_messages() {
-        let (event_tx, _) = broadcast::channel::<(Uuid, String, SseEvent)>(16);
+        let (event_tx, _) = broadcast::channel::<(Uuid, String, AppEvent)>(16);
         let (inject_tx, mut inject_rx) = mpsc::channel::<IncomingMessage>(16);
 
         let job_id = Uuid::new_v4();
@@ -240,7 +240,7 @@ mod tests {
             .send((
                 job_id,
                 "test-user".to_string(),
-                SseEvent::JobMessage {
+                AppEvent::JobMessage {
                     job_id: job_id.to_string(),
                     role: "assistant".to_string(),
                     content: "I found a bug".to_string(),
@@ -262,7 +262,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_monitor_ignores_other_jobs() {
-        let (event_tx, _) = broadcast::channel::<(Uuid, String, SseEvent)>(16);
+        let (event_tx, _) = broadcast::channel::<(Uuid, String, AppEvent)>(16);
         let (inject_tx, mut inject_rx) = mpsc::channel::<IncomingMessage>(16);
 
         let job_id = Uuid::new_v4();
@@ -274,7 +274,7 @@ mod tests {
             .send((
                 other_job_id,
                 "test-user".to_string(),
-                SseEvent::JobMessage {
+                AppEvent::JobMessage {
                     job_id: other_job_id.to_string(),
                     role: "assistant".to_string(),
                     content: "wrong job".to_string(),
@@ -293,7 +293,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_monitor_exits_on_job_result() {
-        let (event_tx, _) = broadcast::channel::<(Uuid, String, SseEvent)>(16);
+        let (event_tx, _) = broadcast::channel::<(Uuid, String, AppEvent)>(16);
         let (inject_tx, mut inject_rx) = mpsc::channel::<IncomingMessage>(16);
 
         let job_id = Uuid::new_v4();
@@ -304,7 +304,7 @@ mod tests {
             .send((
                 job_id,
                 "test-user".to_string(),
-                SseEvent::JobResult {
+                AppEvent::JobResult {
                     job_id: job_id.to_string(),
                     status: "completed".to_string(),
                     session_id: None,
@@ -329,7 +329,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_monitor_skips_tool_events() {
-        let (event_tx, _) = broadcast::channel::<(Uuid, String, SseEvent)>(16);
+        let (event_tx, _) = broadcast::channel::<(Uuid, String, AppEvent)>(16);
         let (inject_tx, mut inject_rx) = mpsc::channel::<IncomingMessage>(16);
 
         let job_id = Uuid::new_v4();
@@ -340,7 +340,7 @@ mod tests {
             .send((
                 job_id,
                 "test-user".to_string(),
-                SseEvent::JobToolUse {
+                AppEvent::JobToolUse {
                     job_id: job_id.to_string(),
                     tool_name: "shell".to_string(),
                     input: serde_json::json!({"command": "ls"}),
@@ -353,7 +353,7 @@ mod tests {
             .send((
                 job_id,
                 "test-user".to_string(),
-                SseEvent::JobMessage {
+                AppEvent::JobMessage {
                     job_id: job_id.to_string(),
                     role: "user".to_string(),
                     content: "user prompt".to_string(),
@@ -409,7 +409,7 @@ mod tests {
             .await
             .unwrap();
 
-        let (event_tx, _) = broadcast::channel::<(Uuid, String, SseEvent)>(16);
+        let (event_tx, _) = broadcast::channel::<(Uuid, String, AppEvent)>(16);
         let (inject_tx, mut inject_rx) = mpsc::channel::<IncomingMessage>(16);
 
         let handle = spawn_job_monitor_with_context(
@@ -425,7 +425,7 @@ mod tests {
             .send((
                 job_id,
                 "test-user".to_string(),
-                SseEvent::JobResult {
+                AppEvent::JobResult {
                     job_id: job_id.to_string(),
                     status: "completed".to_string(),
                     session_id: None,
@@ -458,7 +458,7 @@ mod tests {
             .await
             .unwrap();
 
-        let (event_tx, _) = broadcast::channel::<(Uuid, String, SseEvent)>(16);
+        let (event_tx, _) = broadcast::channel::<(Uuid, String, AppEvent)>(16);
         let (inject_tx, mut inject_rx) = mpsc::channel::<IncomingMessage>(16);
 
         let handle = spawn_job_monitor_with_context(
@@ -474,7 +474,7 @@ mod tests {
             .send((
                 job_id,
                 "test-user".to_string(),
-                SseEvent::JobResult {
+                AppEvent::JobResult {
                     job_id: job_id.to_string(),
                     status: "failed".to_string(),
                     session_id: None,
@@ -507,14 +507,14 @@ mod tests {
             .await
             .unwrap();
 
-        let (event_tx, _) = broadcast::channel::<(Uuid, String, SseEvent)>(16);
+        let (event_tx, _) = broadcast::channel::<(Uuid, String, AppEvent)>(16);
         let handle = spawn_completion_watcher(job_id, event_tx.subscribe(), Arc::clone(&cm));
 
         event_tx
             .send((
                 job_id,
                 "test-user".to_string(),
-                SseEvent::JobResult {
+                AppEvent::JobResult {
                     job_id: job_id.to_string(),
                     status: "completed".to_string(),
                     session_id: None,

src/agent/mod.rs

@@ -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};

src/agent/routine_engine.rs

@@ -18,6 +18,7 @@ use std::time::Duration;
 use chrono::Utc;
 use regex::Regex;
 use tokio::sync::{RwLock, mpsc};
+use tokio::task::JoinHandle;
 use uuid::Uuid;
 
 use crate::agent::Scheduler;
@@ -27,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,
@@ -45,6 +46,11 @@ enum EventMatcher {
     System { routine: Routine },
 }
 
+struct TriggeredRoutine {
+    routine: Routine,
+    detail: String,
+}
+
 /// Distinguishes why sandbox is unavailable so error messages are accurate.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum SandboxReadiness {
@@ -93,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).
@@ -122,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>,
@@ -202,6 +208,44 @@ impl RoutineEngine {
 
     /// Check incoming message against event triggers. Returns number of routines fired.
     pub async fn check_event_triggers(&self, message: &IncomingMessage, content: &str) -> usize {
+        let triggered = self.matching_event_triggers(message, content).await;
+        let fired = triggered.len();
+        for triggered in triggered {
+            std::mem::drop(self.spawn_fire(triggered.routine, "event", Some(triggered.detail)));
+        }
+        fired
+    }
+
+    /// Fire matching event-triggered routines and wait for them to complete.
+    ///
+    /// Used by single-message REPL mode so the process does not exit before
+    /// background event-triggered routines finish.
+    pub async fn check_event_triggers_and_wait(
+        &self,
+        message: &IncomingMessage,
+        content: &str,
+    ) -> usize {
+        let triggered = self.matching_event_triggers(message, content).await;
+        let fired = triggered.len();
+        let handles: Vec<JoinHandle<()>> = triggered
+            .into_iter()
+            .map(|triggered| self.spawn_fire(triggered.routine, "event", Some(triggered.detail)))
+            .collect();
+
+        for handle in handles {
+            if let Err(e) = handle.await {
+                tracing::warn!(error = %e, "Event-triggered routine task failed");
+            }
+        }
+
+        fired
+    }
+
+    async fn matching_event_triggers(
+        &self,
+        message: &IncomingMessage,
+        content: &str,
+    ) -> Vec<TriggeredRoutine> {
         let cache = self.event_cache.read().await;
 
         // Early return if there are no message matchers at all.
@@ -209,10 +253,9 @@ impl RoutineEngine {
             .iter()
             .any(|m| matches!(m, EventMatcher::Message { .. }))
         {
-            return 0;
+            return Vec::new();
         }
-
-        let mut fired = 0;
+        let mut triggered = Vec::new();
 
         // Collect routine IDs for batch query
         let routine_ids: Vec<Uuid> = cache
@@ -224,13 +267,13 @@ impl RoutineEngine {
             .collect();
 
         if routine_ids.is_empty() {
-            return 0;
+            return Vec::new();
         }
 
         // Single batch query instead of N queries
         let concurrent_counts = match self.batch_concurrent_counts(&routine_ids).await {
             Some(counts) => counts,
-            None => return 0,
+            None => return Vec::new(),
         };
 
         for matcher in cache.iter() {
@@ -285,11 +328,13 @@ impl RoutineEngine {
             }
 
             let detail = truncate(content, 200);
-            self.spawn_fire(routine.clone(), "event", Some(detail));
-            fired += 1;
+            triggered.push(TriggeredRoutine {
+                routine: routine.clone(),
+                detail,
+            });
         }
 
-        fired
+        triggered
     }
 
     /// Emit a structured event to system-event routines.
@@ -737,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(),
@@ -845,7 +900,12 @@ impl RoutineEngine {
     }
 
     /// Spawn a fire in a background task.
-    fn spawn_fire(&self, routine: Routine, trigger_type: &str, trigger_detail: Option<String>) {
+    fn spawn_fire(
+        &self,
+        routine: Routine,
+        trigger_type: &str,
+        trigger_detail: Option<String>,
+    ) -> JoinHandle<()> {
         let run = RoutineRun {
             id: Uuid::new_v4(),
             routine_id: routine.id,
@@ -860,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(),
@@ -882,7 +954,7 @@ impl RoutineEngine {
                 return;
             }
             execute_routine(engine, routine, run).await;
-        });
+        })
     }
 
     fn check_cooldown(&self, routine: &Routine) -> bool {
@@ -917,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,
 }
@@ -928,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,
@@ -1000,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>,
@@ -1541,7 +1613,10 @@ async fn execute_lightweight_with_tools(
         let force_text = iteration >= max_iterations;
 
         if force_text {
-            // Final iteration: no tools, just get text response
+            // Final iteration: no tools, just get text response.
+            // Claude 4.6 rejects assistant prefill; NEAR AI rejects any non-user-ending
+            // conversation. Ensure the last message is user-role.
+            crate::util::ensure_ends_with_user_message(&mut messages);
             let request = CompletionRequest::new(messages)
                 .with_max_tokens(effective_max_tokens)
                 .with_temperature(0.3);

src/agent/scheduler.rs

@@ -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);

src/agent/self_repair.rs

@@ -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,

src/agent/session.rs

@@ -16,8 +16,8 @@ use chrono::{DateTime, TimeDelta, Utc};
 use serde::{Deserialize, Serialize};
 use uuid::Uuid;
 
-use crate::channels::web::util::truncate_preview;
 use crate::llm::{ChatMessage, ToolCall, generate_tool_call_id};
+use ironclaw_common::truncate_preview;
 
 /// A session containing one or more threads.
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -449,6 +449,7 @@ impl Thread {
                         id: call_id.clone(),
                         name: tc.name.clone(),
                         arguments: tc.parameters.clone(),
+                        reasoning: None,
                     })
                     .collect();
 
@@ -522,7 +523,12 @@ impl Thread {
                             && let Some(ref tcs) = assistant_msg.tool_calls
                         {
                             for tc in tcs {
-                                turn.record_tool_call(&tc.name, tc.arguments.clone());
+                                turn.record_tool_call_with_reasoning(
+                                    &tc.name,
+                                    tc.arguments.clone(),
+                                    tc.reasoning.clone(),
+                                    Some(tc.id.clone()),
+                                );
                             }
                         }
 
@@ -602,6 +608,10 @@ pub struct Turn {
     pub completed_at: Option<DateTime<Utc>>,
     /// Error message (if failed).
     pub error: Option<String>,
+    /// Agent's reasoning narrative for this turn.
+    /// Cleaned via `clean_response` and sanitized through `SafetyLayer` before storage.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub narrative: Option<String>,
     /// Transient image content parts for multimodal LLM input.
     /// Not serialized — images are only needed for the current LLM call.
     /// The text description in `user_input` persists for compaction/context.
@@ -621,6 +631,7 @@ impl Turn {
             started_at: Utc::now(),
             completed_at: None,
             error: None,
+            narrative: None,
             image_content_parts: Vec::new(),
         }
     }
@@ -656,6 +667,26 @@ impl Turn {
             parameters: params,
             result: None,
             error: None,
+            rationale: None,
+            tool_call_id: None,
+        });
+    }
+
+    /// Record a tool call with reasoning context.
+    pub fn record_tool_call_with_reasoning(
+        &mut self,
+        name: impl Into<String>,
+        params: serde_json::Value,
+        rationale: Option<String>,
+        tool_call_id: Option<String>,
+    ) {
+        self.tool_calls.push(TurnToolCall {
+            name: name.into(),
+            parameters: params,
+            result: None,
+            error: None,
+            rationale,
+            tool_call_id,
         });
     }
 
@@ -672,6 +703,60 @@ impl Turn {
             call.error = Some(error.into());
         }
     }
+
+    /// Record a tool result by tool_call_id, with fallback to first pending call.
+    pub fn record_tool_result_for(&mut self, tool_call_id: &str, result: serde_json::Value) {
+        if let Some(call) = self
+            .tool_calls
+            .iter_mut()
+            .find(|c| c.tool_call_id.as_deref() == Some(tool_call_id))
+        {
+            call.result = Some(result);
+        } else if let Some(call) = self
+            .tool_calls
+            .iter_mut()
+            .find(|c| c.result.is_none() && c.error.is_none())
+        {
+            tracing::debug!(
+                tool_call_id = %tool_call_id,
+                fallback_tool = %call.name,
+                "tool_call_id not found, falling back to first pending call"
+            );
+            call.result = Some(result);
+        } else {
+            tracing::warn!(
+                tool_call_id = %tool_call_id,
+                "Tool result dropped: no matching or pending tool call"
+            );
+        }
+    }
+
+    /// Record a tool error by tool_call_id, with fallback to first pending call.
+    pub fn record_tool_error_for(&mut self, tool_call_id: &str, error: impl Into<String>) {
+        if let Some(call) = self
+            .tool_calls
+            .iter_mut()
+            .find(|c| c.tool_call_id.as_deref() == Some(tool_call_id))
+        {
+            call.error = Some(error.into());
+        } else if let Some(call) = self
+            .tool_calls
+            .iter_mut()
+            .find(|c| c.result.is_none() && c.error.is_none())
+        {
+            tracing::debug!(
+                tool_call_id = %tool_call_id,
+                fallback_tool = %call.name,
+                "tool_call_id not found, falling back to first pending call"
+            );
+            call.error = Some(error.into());
+        } else {
+            tracing::warn!(
+                tool_call_id = %tool_call_id,
+                "Tool error dropped: no matching or pending tool call"
+            );
+        }
+    }
 }
 
 /// Record of a tool call made during a turn.
@@ -685,6 +770,12 @@ pub struct TurnToolCall {
     pub result: Option<serde_json::Value>,
     /// Error from the tool (if failed).
     pub error: Option<String>,
+    /// Agent's reasoning for choosing this tool.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub rationale: Option<String>,
+    /// The tool_call_id from the LLM, for identity-based result matching.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub tool_call_id: Option<String>,
 }
 
 #[cfg(test)]
@@ -1309,6 +1400,7 @@ mod tests {
             id: "call_0".to_string(),
             name: "search".to_string(),
             arguments: serde_json::json!({"q": "test"}),
+            reasoning: None,
         };
         let messages = vec![
             ChatMessage::user("Find test"),
@@ -1339,6 +1431,7 @@ mod tests {
             id: "call_0".to_string(),
             name: "http".to_string(),
             arguments: serde_json::json!({}),
+            reasoning: None,
         };
         let messages = vec![
             ChatMessage::user("Fetch URL"),
@@ -1404,11 +1497,13 @@ mod tests {
             id: "call_a".to_string(),
             name: "search".to_string(),
             arguments: serde_json::json!({"q": "data"}),
+            reasoning: None,
         };
         let tc2 = ToolCall {
             id: "call_b".to_string(),
             name: "write".to_string(),
             arguments: serde_json::json!({"path": "out.txt"}),
+            reasoning: None,
         };
         let messages = vec![
             ChatMessage::user("Find and save"),
@@ -1620,4 +1715,100 @@ mod tests {
         let merged = thread.drain_pending_messages().unwrap();
         assert_eq!(merged, "failed batch\nnew msg");
     }
+
+    #[test]
+    fn test_record_tool_result_for_by_id() {
+        let mut turn = Turn::new(0, "test");
+        turn.record_tool_call_with_reasoning(
+            "tool_a",
+            serde_json::json!({}),
+            None,
+            Some("id_a".into()),
+        );
+        turn.record_tool_call_with_reasoning(
+            "tool_b",
+            serde_json::json!({}),
+            None,
+            Some("id_b".into()),
+        );
+
+        // Record result for second tool by ID
+        turn.record_tool_result_for("id_b", serde_json::json!("result_b"));
+        assert!(turn.tool_calls[0].result.is_none());
+        assert_eq!(
+            turn.tool_calls[1].result.as_ref().unwrap(),
+            &serde_json::json!("result_b")
+        );
+    }
+
+    #[test]
+    fn test_record_tool_error_for_by_id() {
+        let mut turn = Turn::new(0, "test");
+        turn.record_tool_call_with_reasoning(
+            "tool_a",
+            serde_json::json!({}),
+            None,
+            Some("id_a".into()),
+        );
+        turn.record_tool_call_with_reasoning(
+            "tool_b",
+            serde_json::json!({}),
+            None,
+            Some("id_b".into()),
+        );
+
+        turn.record_tool_error_for("id_a", "failed");
+        assert_eq!(turn.tool_calls[0].error.as_deref(), Some("failed"));
+        assert!(turn.tool_calls[1].error.is_none());
+    }
+
+    #[test]
+    fn test_record_tool_result_for_fallback_to_pending() {
+        let mut turn = Turn::new(0, "test");
+        turn.record_tool_call_with_reasoning(
+            "tool_a",
+            serde_json::json!({}),
+            None,
+            Some("id_a".into()),
+        );
+        turn.record_tool_call_with_reasoning(
+            "tool_b",
+            serde_json::json!({}),
+            None,
+            Some("id_b".into()),
+        );
+
+        // First tool already has a result
+        turn.tool_calls[0].result = Some(serde_json::json!("done"));
+
+        // Unknown ID should fall back to first pending (tool_b)
+        turn.record_tool_result_for("unknown_id", serde_json::json!("fallback"));
+        assert_eq!(
+            turn.tool_calls[0].result.as_ref().unwrap(),
+            &serde_json::json!("done")
+        );
+        assert_eq!(
+            turn.tool_calls[1].result.as_ref().unwrap(),
+            &serde_json::json!("fallback")
+        );
+    }
+
+    #[test]
+    fn test_record_tool_result_for_no_pending_is_noop() {
+        let mut turn = Turn::new(0, "test");
+        turn.record_tool_call_with_reasoning(
+            "tool_a",
+            serde_json::json!({}),
+            None,
+            Some("id_a".into()),
+        );
+        turn.tool_calls[0].result = Some(serde_json::json!("done"));
+
+        // No pending calls, unknown ID — should be a no-op
+        turn.record_tool_result_for("unknown_id", serde_json::json!("lost"));
+        assert_eq!(
+            turn.tool_calls[0].result.as_ref().unwrap(),
+            &serde_json::json!("done")
+        );
+    }
 }

src/agent/submission.rs

@@ -92,6 +92,17 @@ impl SubmissionParser {
                 args: vec![],
             };
         }
+        if lower == "/reasoning" || lower.starts_with("/reasoning ") {
+            let args: Vec<String> = trimmed
+                .split_whitespace()
+                .skip(1)
+                .map(|s| s.to_string())
+                .collect();
+            return Submission::SystemCommand {
+                command: "reasoning".to_string(),
+                args,
+            };
+        }
         if lower == "/restart" {
             tracing::debug!("[SubmissionParser::parse] Recognized /restart command");
             return Submission::SystemCommand {

src/agent/thread_ops.rs

@@ -16,12 +16,12 @@ use crate::agent::dispatcher::{
 };
 use crate::agent::session::{MAX_PENDING_MESSAGES, PendingApproval, Session, ThreadState};
 use crate::agent::submission::SubmissionResult;
-use crate::channels::web::util::truncate_preview;
 use crate::channels::{IncomingMessage, StatusUpdate};
 use crate::context::JobContext;
 use crate::error::Error;
 use crate::llm::{ChatMessage, ToolCall};
 use crate::tools::redact_params;
+use ironclaw_common::truncate_preview;
 
 const FORGED_THREAD_ID_ERROR: &str = "Invalid or unauthorized thread ID.";
 
@@ -226,6 +226,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
@@ -513,10 +514,10 @@ impl Agent {
                 };
 
                 thread.complete_turn(&response);
-                let (turn_number, tool_calls) = thread
+                let (turn_number, tool_calls, narrative) = thread
                     .turns
                     .last()
-                    .map(|t| (t.turn_number, t.tool_calls.clone()))
+                    .map(|t| (t.turn_number, t.tool_calls.clone(), t.narrative.clone()))
                     .unwrap_or_default();
                 let _ = self
                     .channels
@@ -534,6 +535,7 @@ impl Agent {
                     &message.user_id,
                     turn_number,
                     &tool_calls,
+                    narrative.as_deref(),
                 )
                 .await;
                 self.persist_assistant_response(
@@ -725,7 +727,9 @@ impl Agent {
     ///
     /// Stored between the user and assistant messages so that
     /// `build_turns_from_db_messages` can reconstruct the tool call history.
-    /// Content is a JSON array of tool call summaries.
+    /// Content is a JSON object: `{ "calls": [...], "narrative": "..." }`.
+    /// The `calls` array contains tool call summaries with optional `rationale`
+    /// and `tool_call_id` fields. Legacy rows may be plain JSON arrays.
     pub(super) async fn persist_tool_calls(
         &self,
         thread_id: Uuid,
@@ -733,6 +737,7 @@ impl Agent {
         user_id: &str,
         turn_number: usize,
         tool_calls: &[crate::agent::session::TurnToolCall],
+        narrative: Option<&str>,
     ) {
         if tool_calls.is_empty() {
             return;
@@ -767,11 +772,30 @@ impl Agent {
                 if let Some(ref error) = tc.error {
                     obj["error"] = serde_json::Value::String(truncate_preview(error, 200));
                 }
+                if let Some(ref rationale) = tc.rationale {
+                    obj["rationale"] = serde_json::Value::String(truncate_preview(rationale, 500));
+                }
+                if let Some(ref tool_call_id) = tc.tool_call_id {
+                    obj["tool_call_id"] =
+                        serde_json::Value::String(truncate_preview(tool_call_id, 128));
+                }
                 obj
             })
             .collect();
 
-        let content = match serde_json::to_string(&summaries) {
+        // Wrap in an object with optional narrative so it can be reconstructed.
+        // safety: no byte-index slicing here; comment describes JSON shape
+        let wrapper = if let Some(n) = narrative {
+            serde_json::json!({
+                "narrative": truncate_preview(n, 1000),
+                "calls": summaries,
+            })
+        } else {
+            serde_json::json!({
+                "calls": summaries,
+            })
+        };
+        let content = match serde_json::to_string(&wrapper) {
             Ok(c) => c,
             Err(e) => {
                 tracing::warn!("Failed to serialize tool calls: {}", e);
@@ -1104,9 +1128,12 @@ impl Agent {
                     && let Some(turn) = thread.last_turn_mut()
                 {
                     if is_tool_error {
-                        turn.record_tool_error(result_content.clone());
+                        turn.record_tool_error_for(&pending.tool_call_id, result_content.clone());
                     } else {
-                        turn.record_tool_result(serde_json::json!(result_content));
+                        turn.record_tool_result_for(
+                            &pending.tool_call_id,
+                            serde_json::json!(result_content),
+                        );
                     }
                 }
             }
@@ -1358,9 +1385,12 @@ impl Agent {
                         && let Some(turn) = thread.last_turn_mut()
                     {
                         if is_deferred_error {
-                            turn.record_tool_error(deferred_content.clone());
+                            turn.record_tool_error_for(&tc.id, deferred_content.clone());
                         } else {
-                            turn.record_tool_result(serde_json::json!(deferred_content));
+                            turn.record_tool_result_for(
+                                &tc.id,
+                                serde_json::json!(deferred_content),
+                            );
                         }
                     }
                 }
@@ -1444,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
@@ -1459,10 +1495,10 @@ impl Agent {
                     let (response, suggestions) =
                         crate::agent::dispatcher::extract_suggestions(&response);
                     thread.complete_turn(&response);
-                    let (turn_number, tool_calls) = thread
+                    let (turn_number, tool_calls, narrative) = thread
                         .turns
                         .last()
-                        .map(|t| (t.turn_number, t.tool_calls.clone()))
+                        .map(|t| (t.turn_number, t.tool_calls.clone(), t.narrative.clone()))
                         .unwrap_or_default();
                     // User message already persisted at turn start; save tool calls then assistant response
                     self.persist_tool_calls(
@@ -1471,6 +1507,7 @@ impl Agent {
                         &message.user_id,
                         turn_number,
                         &tool_calls,
+                        narrative.as_deref(),
                     )
                     .await;
                     self.persist_assistant_response(
@@ -1816,7 +1853,20 @@ fn rebuild_chat_messages_from_db(
             "assistant" => result.push(ChatMessage::assistant(&msg.content)),
             "tool_calls" => {
                 // Try to parse the enriched JSON and rebuild tool messages.
-                if let Ok(calls) = serde_json::from_str::<Vec<serde_json::Value>>(&msg.content) {
+                // Supports two formats:
+                // - Old: plain JSON array of tool call summaries
+                // - New: wrapped object { "calls": [...], "narrative": "..." }
+                let calls: Vec<serde_json::Value> =
+                    match serde_json::from_str::<serde_json::Value>(&msg.content) {
+                        Ok(serde_json::Value::Array(arr)) => arr,
+                        Ok(serde_json::Value::Object(obj)) => obj
+                            .get("calls")
+                            .and_then(|v| v.as_array())
+                            .cloned()
+                            .unwrap_or_default(),
+                        _ => Vec::new(),
+                    };
+                {
                     if calls.is_empty() {
                         continue;
                     }
@@ -1839,6 +1889,10 @@ fn rebuild_chat_messages_from_db(
                                     .get("parameters")
                                     .cloned()
                                     .unwrap_or(serde_json::json!({})),
+                                reasoning: c
+                                    .get("rationale")
+                                    .and_then(|v| v.as_str())
+                                    .map(String::from),
                             })
                             .collect();