fix: second-pass multi-tenant audit — scope SSE broadcasts, DB queries, dead handlers

Second audit pass applying learned patterns across the codebase:

- OAuth callback SSE broadcasts now use broadcast_for_user (lines 773, 912)
- jobs_list_handler uses list_agent_jobs_for_user instead of fetching
  all users' jobs and filtering in Rust
- list_agent_jobs_for_user added to Database trait + postgres + libsql
- Dead handler files (extensions.rs, static_files.rs) hardened with
  AuthenticatedUser to prevent auth regression if migrated

Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]>

src/channels/web/handlers/extensions.rs

@@ -82,6 +82,7 @@ pub async fn extensions_list_handler(
 
 pub async fn extensions_tools_handler(
     State(state): State<Arc<GatewayState>>,
+    AuthenticatedUser(_user): AuthenticatedUser,
 ) -> Result<Json<ToolListResponse>, (StatusCode, String)> {
     let registry = state.tool_registry.as_ref().ok_or((
         StatusCode::SERVICE_UNAVAILABLE,

src/channels/web/handlers/jobs.rs

@@ -52,13 +52,10 @@ pub async fn jobs_list_handler(
         }
     }
 
-    // Fetch agent (non-sandbox) jobs from database, deduplicating by ID.
-    match store.list_agent_jobs().await {
+    // Fetch agent (non-sandbox) jobs scoped to this user, deduplicating by ID.
+    match store.list_agent_jobs_for_user(&user.user_id).await {
         Ok(agent_jobs) => {
             for j in &agent_jobs {
-                if j.user_id != user.user_id {
-                    continue;
-                }
                 if seen_ids.contains(&j.id) {
                     continue;
                 }

src/channels/web/handlers/static_files.rs

@@ -7,6 +7,7 @@ use axum::{
 };
 
 use crate::bootstrap::ironclaw_base_dir;
+use crate::channels::web::auth::AuthenticatedUser;
 use crate::channels::web::types::*;
 
 // --- Static file handlers ---
@@ -113,6 +114,7 @@ use crate::channels::web::server::GatewayState;
 
 pub async fn logs_events_handler(
     State(state): State<Arc<GatewayState>>,
+    AuthenticatedUser(_user): AuthenticatedUser,
 ) -> Result<
     Sse<impl futures::Stream<Item = Result<Event, Infallible>> + Send + 'static>,
     (StatusCode, String),
@@ -152,6 +154,7 @@ pub async fn logs_events_handler(
 
 pub async fn gateway_status_handler(
     State(state): State<Arc<GatewayState>>,
+    AuthenticatedUser(_user): AuthenticatedUser,
 ) -> Json<GatewayStatusResponse> {
     let sse_connections = state.sse.connection_count();
     let ws_connections = state

src/channels/web/server.rs

@@ -770,7 +770,7 @@ async fn oauth_callback_handler(
         );
         // Notify UI so auth card can show error instead of staying stuck
         if let Some(ref sse) = flow.sse_manager {
-            sse.broadcast(SseEvent::AuthCompleted {
+            sse.broadcast_for_user(&flow.user_id, SseEvent::AuthCompleted {
                 extension_name: flow.extension_name.clone(),
                 success: false,
                 message: "OAuth flow expired. Please try again.".to_string(),
@@ -909,7 +909,7 @@ async fn oauth_callback_handler(
 
     // Broadcast SSE event to notify the web UI
     if let Some(ref sse) = flow.sse_manager {
-        sse.broadcast(SseEvent::AuthCompleted {
+        sse.broadcast_for_user(&flow.user_id, SseEvent::AuthCompleted {
             extension_name: flow.extension_name,
             success,
             message: final_message.clone(),

src/db/libsql/jobs.rs

@@ -230,6 +230,49 @@ impl JobStore for LibSqlBackend {
         Ok(jobs)
     }
 
+    async fn list_agent_jobs_for_user(
+        &self,
+        user_id: &str,
+    ) -> Result<Vec<AgentJobRecord>, DatabaseError> {
+        let conn = self.connect().await?;
+        let mut rows = conn
+            .query(
+                r#"
+                SELECT id, title, status, user_id, failure_reason,
+                       created_at, started_at, completed_at
+                FROM agent_jobs WHERE source = 'direct' AND user_id = ?1
+                ORDER BY created_at DESC
+                "#,
+                params![user_id],
+            )
+            .await
+            .map_err(|e| DatabaseError::Query(e.to_string()))?;
+
+        let mut jobs = Vec::new();
+        while let Some(row) = rows
+            .next()
+            .await
+            .map_err(|e| DatabaseError::Query(e.to_string()))?
+        {
+            let id_str = get_text(&row, 0);
+            let Ok(id) = id_str.parse() else {
+                tracing::warn!("Skipping agent job with invalid UUID: {}", id_str);
+                continue;
+            };
+            jobs.push(AgentJobRecord {
+                id,
+                title: get_text(&row, 1),
+                status: get_text(&row, 2),
+                user_id: get_text(&row, 3),
+                failure_reason: get_opt_text(&row, 4),
+                created_at: get_ts(&row, 5),
+                started_at: get_opt_ts(&row, 6),
+                completed_at: get_opt_ts(&row, 7),
+            });
+        }
+        Ok(jobs)
+    }
+
     async fn get_agent_job_failure_reason(
         &self,
         id: Uuid,

src/db/mod.rs

@@ -409,6 +409,10 @@ pub trait JobStore: Send + Sync {
     async fn mark_job_stuck(&self, id: Uuid) -> Result<(), DatabaseError>;
     async fn get_stuck_jobs(&self) -> Result<Vec<Uuid>, DatabaseError>;
     async fn list_agent_jobs(&self) -> Result<Vec<AgentJobRecord>, DatabaseError>;
+    async fn list_agent_jobs_for_user(
+        &self,
+        user_id: &str,
+    ) -> Result<Vec<AgentJobRecord>, DatabaseError>;
     async fn agent_job_summary(&self) -> Result<AgentJobSummary, DatabaseError>;
     async fn agent_job_summary_for_user(
         &self,

src/db/postgres.rs

@@ -249,6 +249,13 @@ impl JobStore for PgBackend {
         self.store.list_agent_jobs().await
     }
 
+    async fn list_agent_jobs_for_user(
+        &self,
+        user_id: &str,
+    ) -> Result<Vec<AgentJobRecord>, DatabaseError> {
+        self.store.list_agent_jobs_for_user(user_id).await
+    }
+
     async fn agent_job_summary(&self) -> Result<AgentJobSummary, DatabaseError> {
         self.store.agent_job_summary().await
     }

src/history/store.rs

@@ -842,6 +842,38 @@ impl Store {
             .collect())
     }
 
+    pub async fn list_agent_jobs_for_user(
+        &self,
+        user_id: &str,
+    ) -> Result<Vec<AgentJobRecord>, DatabaseError> {
+        let conn = self.conn().await?;
+        let rows = conn
+            .query(
+                r#"
+                SELECT id, title, status, user_id, failure_reason,
+                       created_at, started_at, completed_at
+                FROM agent_jobs WHERE source = 'direct' AND user_id = $1
+                ORDER BY created_at DESC
+                "#,
+                &[&user_id],
+            )
+            .await?;
+
+        Ok(rows
+            .iter()
+            .map(|r| AgentJobRecord {
+                id: r.get("id"),
+                title: r.get("title"),
+                status: r.get("status"),
+                user_id: r.get::<_, Option<String>>("user_id").unwrap_or_default(),
+                created_at: r.get("created_at"),
+                started_at: r.get("started_at"),
+                completed_at: r.get("completed_at"),
+                failure_reason: r.get("failure_reason"),
+            })
+            .collect())
+    }
+
     /// Get the failure reason for a single agent job.
     pub async fn get_agent_job_failure_reason(
         &self,