feat: multi-tenant auth with per-user workspace isolation

src/agent/agent_loop.rs

@@ -157,8 +157,8 @@ pub struct AgentDeps {
     pub hooks: Arc<HookRegistry>,
     /// Cost enforcement guardrails (daily budget, hourly rate limits).
     pub cost_guard: Arc<crate::agent::cost_guard::CostGuard>,
-    /// SSE broadcast sender for live job event streaming to the web gateway.
-    pub sse_tx: Option<tokio::sync::broadcast::Sender<crate::channels::web::types::SseEvent>>,
+    /// SSE manager for live job event streaming to the web gateway.
+    pub sse_tx: Option<Arc<crate::channels::web::sse::SseManager>>,
     /// HTTP interceptor for trace recording/replay.
     pub http_interceptor: Option<Arc<dyn crate::llm::recording::HttpInterceptor>>,
     /// Audio transcription middleware for voice messages.
@@ -235,8 +235,8 @@ impl Agent {
                 hooks: deps.hooks.clone(),
             },
         );
-        if let Some(ref tx) = deps.sse_tx {
-            scheduler.set_sse_sender(tx.clone());
+        if let Some(ref sse) = deps.sse_tx {
+            scheduler.set_sse_sender(Arc::clone(sse));
         }
         if let Some(ref interceptor) = deps.http_interceptor {
             scheduler.set_http_interceptor(Arc::clone(interceptor));

src/agent/job_monitor.rs

@@ -44,7 +44,7 @@ pub struct JobMonitorRoute {
 /// the main agent's context window).
 pub fn spawn_job_monitor(
     job_id: Uuid,
-    event_rx: broadcast::Receiver<(Uuid, SseEvent)>,
+    event_rx: broadcast::Receiver<(Uuid, String, SseEvent)>,
     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, SseEvent)>,
+    mut event_rx: broadcast::Receiver<(Uuid, String, SseEvent)>,
     inject_tx: mpsc::Sender<IncomingMessage>,
     route: JobMonitorRoute,
     context_manager: Option<Arc<ContextManager>>,
@@ -68,7 +68,7 @@ pub fn spawn_job_monitor_with_context(
 
         loop {
             match event_rx.recv().await {
-                Ok((ev_job_id, event)) => {
+                Ok((ev_job_id, _user_id, event)) => {
                     if ev_job_id != job_id {
                         continue;
                     }
@@ -162,15 +162,17 @@ 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, SseEvent)>,
+    mut event_rx: broadcast::Receiver<(Uuid, String, SseEvent)>,
     context_manager: Arc<ContextManager>,
 ) -> JoinHandle<()> {
     let short_id = job_id.to_string()[..8].to_string();
 
     tokio::spawn(async move {
         loop {
             match event_rx.recv().await {
-                Ok((ev_job_id, SseEvent::JobResult { status, .. })) if ev_job_id == job_id => {
+                Ok((ev_job_id, _user_id, SseEvent::JobResult { status, .. }))
+                    if ev_job_id == job_id =>
+                {
                     let target = if status == "completed" {
                         JobState::Completed
                     } else {
@@ -227,7 +229,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_monitor_forwards_assistant_messages() {
-        let (event_tx, _) = broadcast::channel::<(Uuid, SseEvent)>(16);
+        let (event_tx, _) = broadcast::channel::<(Uuid, String, SseEvent)>(16);
         let (inject_tx, mut inject_rx) = mpsc::channel::<IncomingMessage>(16);
 
         let job_id = Uuid::new_v4();
@@ -237,6 +239,7 @@ mod tests {
         event_tx
             .send((
                 job_id,
+                "test-user".to_string(),
                 SseEvent::JobMessage {
                     job_id: job_id.to_string(),
                     role: "assistant".to_string(),
@@ -259,7 +262,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_monitor_ignores_other_jobs() {
-        let (event_tx, _) = broadcast::channel::<(Uuid, SseEvent)>(16);
+        let (event_tx, _) = broadcast::channel::<(Uuid, String, SseEvent)>(16);
         let (inject_tx, mut inject_rx) = mpsc::channel::<IncomingMessage>(16);
 
         let job_id = Uuid::new_v4();
@@ -270,6 +273,7 @@ mod tests {
         event_tx
             .send((
                 other_job_id,
+                "test-user".to_string(),
                 SseEvent::JobMessage {
                     job_id: other_job_id.to_string(),
                     role: "assistant".to_string(),
@@ -289,7 +293,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_monitor_exits_on_job_result() {
-        let (event_tx, _) = broadcast::channel::<(Uuid, SseEvent)>(16);
+        let (event_tx, _) = broadcast::channel::<(Uuid, String, SseEvent)>(16);
         let (inject_tx, mut inject_rx) = mpsc::channel::<IncomingMessage>(16);
 
         let job_id = Uuid::new_v4();
@@ -299,6 +303,7 @@ mod tests {
         event_tx
             .send((
                 job_id,
+                "test-user".to_string(),
                 SseEvent::JobResult {
                     job_id: job_id.to_string(),
                     status: "completed".to_string(),
@@ -324,7 +329,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_monitor_skips_tool_events() {
-        let (event_tx, _) = broadcast::channel::<(Uuid, SseEvent)>(16);
+        let (event_tx, _) = broadcast::channel::<(Uuid, String, SseEvent)>(16);
         let (inject_tx, mut inject_rx) = mpsc::channel::<IncomingMessage>(16);
 
         let job_id = Uuid::new_v4();
@@ -334,6 +339,7 @@ mod tests {
         event_tx
             .send((
                 job_id,
+                "test-user".to_string(),
                 SseEvent::JobToolUse {
                     job_id: job_id.to_string(),
                     tool_name: "shell".to_string(),
@@ -346,6 +352,7 @@ mod tests {
         event_tx
             .send((
                 job_id,
+                "test-user".to_string(),
                 SseEvent::JobMessage {
                     job_id: job_id.to_string(),
                     role: "user".to_string(),
@@ -402,7 +409,7 @@ mod tests {
             .await
             .unwrap();
 
-        let (event_tx, _) = broadcast::channel::<(Uuid, SseEvent)>(16);
+        let (event_tx, _) = broadcast::channel::<(Uuid, String, SseEvent)>(16);
         let (inject_tx, mut inject_rx) = mpsc::channel::<IncomingMessage>(16);
 
         let handle = spawn_job_monitor_with_context(
@@ -417,6 +424,7 @@ mod tests {
         event_tx
             .send((
                 job_id,
+                "test-user".to_string(),
                 SseEvent::JobResult {
                     job_id: job_id.to_string(),
                     status: "completed".to_string(),
@@ -450,7 +458,7 @@ mod tests {
             .await
             .unwrap();
 
-        let (event_tx, _) = broadcast::channel::<(Uuid, SseEvent)>(16);
+        let (event_tx, _) = broadcast::channel::<(Uuid, String, SseEvent)>(16);
         let (inject_tx, mut inject_rx) = mpsc::channel::<IncomingMessage>(16);
 
         let handle = spawn_job_monitor_with_context(
@@ -465,6 +473,7 @@ mod tests {
         event_tx
             .send((
                 job_id,
+                "test-user".to_string(),
                 SseEvent::JobResult {
                     job_id: job_id.to_string(),
                     status: "failed".to_string(),
@@ -498,12 +507,13 @@ mod tests {
             .await
             .unwrap();
 
-        let (event_tx, _) = broadcast::channel::<(Uuid, SseEvent)>(16);
+        let (event_tx, _) = broadcast::channel::<(Uuid, String, SseEvent)>(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 {
                     job_id: job_id.to_string(),
                     status: "completed".to_string(),

src/agent/scheduler.rs

@@ -9,7 +9,6 @@ use tokio::task::JoinHandle;
 use uuid::Uuid;
 
 use crate::agent::task::{Task, TaskContext, TaskOutput};
-use crate::channels::web::types::SseEvent;
 use crate::config::AgentConfig;
 use crate::context::{ContextManager, JobContext, JobState};
 use crate::db::Database;
@@ -67,8 +66,8 @@ pub struct Scheduler {
     extension_manager: Option<Arc<ExtensionManager>>,
     store: Option<Arc<dyn Database>>,
     hooks: Arc<HookRegistry>,
-    /// SSE broadcast sender for live job event streaming.
-    sse_tx: Option<tokio::sync::broadcast::Sender<SseEvent>>,
+    /// SSE manager for live job event streaming.
+    sse_tx: Option<Arc<crate::channels::web::sse::SseManager>>,
     /// HTTP interceptor for trace recording/replay (propagated to workers).
     http_interceptor: Option<Arc<dyn crate::llm::recording::HttpInterceptor>>,
     /// Running jobs (main LLM-driven jobs).
@@ -102,9 +101,9 @@ impl Scheduler {
         }
     }
 
-    /// Set the SSE broadcast sender for live job event streaming.
-    pub fn set_sse_sender(&mut self, tx: tokio::sync::broadcast::Sender<SseEvent>) {
-        self.sse_tx = Some(tx);
+    /// Set the SSE manager for live job event streaming.
+    pub fn set_sse_sender(&mut self, sse: Arc<crate::channels::web::sse::SseManager>) {
+        self.sse_tx = Some(sse);
     }
 
     /// Set the HTTP interceptor for trace recording/replay.

src/agent/thread_ops.rs

@@ -1646,7 +1646,7 @@ impl Agent {
         };
 
         match ext_mgr
-            .configure_token(&pending.extension_name, token)
+            .configure_token(&pending.extension_name, token, &message.user_id)
             .await
         {
             Ok(result) if result.activated => {

src/app.rs

@@ -327,7 +327,7 @@ impl AppBuilder {
                 .with_search_config(&self.config.search);
 
             if let Some(ref emb) = embeddings {
-                ws = ws.with_embeddings_cached(emb.clone(), emb_cache_config);
+                ws = ws.with_embeddings_cached(emb.clone(), emb_cache_config.clone());
             }
 
             // Wire workspace-level settings (read scopes, memory layers)
@@ -341,7 +341,35 @@ impl AppBuilder {
             }
             ws = ws.with_memory_layers(self.config.workspace.memory_layers.clone());
             let ws = Arc::new(ws);
-            tools.register_memory_tools(Arc::clone(&ws));
+
+            // Detect multi-tenant mode: when GATEWAY_USER_TOKENS is configured,
+            // each authenticated user needs their own workspace scope. Use
+            // WorkspacePool (which implements WorkspaceResolver) to create
+            // per-user workspaces on demand instead of sharing the startup
+            // workspace across all users.
+            let is_multi_tenant = self
+                .config
+                .channels
+                .gateway
+                .as_ref()
+                .is_some_and(|gw| gw.user_tokens.is_some());
+
+            if is_multi_tenant {
+                let pool = Arc::new(crate::channels::web::server::WorkspacePool::new(
+                    Arc::clone(db),
+                    embeddings.clone(),
+                    emb_cache_config,
+                    self.config.search.clone(),
+                    self.config.workspace.clone(),
+                ));
+                tools.register_memory_tools_with_resolver(pool);
+                tracing::info!(
+                    "Memory tools configured with per-user workspace resolver (multi-tenant mode)"
+                );
+            } else {
+                tools.register_memory_tools(Arc::clone(&ws));
+            }
+
             Some(ws)
         } else {
             None