diff --git a/src/channels/web/handlers/extensions.rs b/src/channels/web/handlers/extensions.rs index f855aa43..d705591e 100644 --- a/src/channels/web/handlers/extensions.rs +++ b/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>, + AuthenticatedUser(_user): AuthenticatedUser, ) -> Result, (StatusCode, String)> { let registry = state.tool_registry.as_ref().ok_or(( StatusCode::SERVICE_UNAVAILABLE, diff --git a/src/channels/web/handlers/jobs.rs b/src/channels/web/handlers/jobs.rs index c5d5d142..6901d943 100644 --- a/src/channels/web/handlers/jobs.rs +++ b/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; } diff --git a/src/channels/web/handlers/static_files.rs b/src/channels/web/handlers/static_files.rs index c198d95e..effc7037 100644 --- a/src/channels/web/handlers/static_files.rs +++ b/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>, + AuthenticatedUser(_user): AuthenticatedUser, ) -> Result< Sse> + Send + 'static>, (StatusCode, String), @@ -152,6 +154,7 @@ pub async fn logs_events_handler( pub async fn gateway_status_handler( State(state): State>, + AuthenticatedUser(_user): AuthenticatedUser, ) -> Json { let sse_connections = state.sse.connection_count(); let ws_connections = state diff --git a/src/channels/web/server.rs b/src/channels/web/server.rs index fc00cbff..bed844cd 100644 --- a/src/channels/web/server.rs +++ b/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(), diff --git a/src/db/libsql/jobs.rs b/src/db/libsql/jobs.rs index 1ddaab83..297a9282 100644 --- a/src/db/libsql/jobs.rs +++ b/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, 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, diff --git a/src/db/mod.rs b/src/db/mod.rs index 2fc69f1f..c0594bda 100644 --- a/src/db/mod.rs +++ b/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, DatabaseError>; async fn list_agent_jobs(&self) -> Result, DatabaseError>; + async fn list_agent_jobs_for_user( + &self, + user_id: &str, + ) -> Result, DatabaseError>; async fn agent_job_summary(&self) -> Result; async fn agent_job_summary_for_user( &self, diff --git a/src/db/postgres.rs b/src/db/postgres.rs index fba28b03..a2c686d3 100644 --- a/src/db/postgres.rs +++ b/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, DatabaseError> { + self.store.list_agent_jobs_for_user(user_id).await + } + async fn agent_job_summary(&self) -> Result { self.store.agent_job_summary().await } diff --git a/src/history/store.rs b/src/history/store.rs index ec8e33ce..d6570b3c 100644 --- a/src/history/store.rs +++ b/src/history/store.rs @@ -842,6 +842,38 @@ impl Store { .collect()) } + pub async fn list_agent_jobs_for_user( + &self, + user_id: &str, + ) -> Result, 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>("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,