From e265f38429f5408b4915e7e789152a8bd7b63dd0 Mon Sep 17 00:00:00 2001 From: Nikhil Sinha Date: Wed, 28 Oct 2026 22:07:42 +0530 Subject: [PATCH 1/2] enable mutation tools --- src/handlers/http/alerts.rs | 49 +++++--- src/handlers/http/targets.rs | 30 ++++- src/handlers/http/users/dashboards.rs | 154 +++++++++++++++-------- src/tool_catalog.rs | 133 ++++++++------------ src/tool_executor.rs | 172 ++++++++++++++++++++++++-- 5 files changed, 375 insertions(+), 163 deletions(-) diff --git a/src/handlers/http/alerts.rs b/src/handlers/http/alerts.rs index 975eb479c..0e544ce6b 100644 --- a/src/handlers/http/alerts.rs +++ b/src/handlers/http/alerts.rs @@ -387,6 +387,15 @@ pub async fn delete(req: HttpRequest, alert_id: Path) -> Result, +) -> Result { let guard = ALERTS.write().await; let alerts = if let Some(alerts) = guard.as_ref() { alerts @@ -394,26 +403,23 @@ pub async fn delete(req: HttpRequest, alert_id: Path) -> Result, +) -> Result { // Get alerts manager reference without holding the global lock let alerts = { let guard = ALERTS.read().await; @@ -596,15 +614,15 @@ pub async fn modify_alert( }; // Validate and prepare the new alert - let alert = alerts.get_alert_by_id(alert_id, &tenant_id).await?; - user_auth_for_alert_config(&session_key, &alert.to_alert_config()).await?; + let alert = alerts.get_alert_by_id(alert_id, tenant_id).await?; + user_auth_for_alert_config(session_key, &alert.to_alert_config()).await?; let mut new_config = alert_request.into(tenant_id.clone()).await?; if &new_config.alert_type != alert.get_alert_type() { return Err(AlertError::InvalidAlertModifyRequest); } - user_auth_for_alert_config(&session_key, &new_config).await?; + user_auth_for_alert_config(session_key, &new_config).await?; // Calculate notification config let eval_freq = new_config.get_eval_frequency(); @@ -639,18 +657,18 @@ pub async fn modify_alert( } }; - new_alert.validate(&session_key).await?; + new_alert.validate(session_key).await?; // Perform I/O operations PARSEABLE .metastore - .put_alert(&new_alert.to_alert_config(), &tenant_id) + .put_alert(&new_alert.to_alert_config(), tenant_id) .await?; let is_disabled = new_alert.get_state().eq(&AlertState::Disabled); // Now perform the atomic operations alerts.delete_task(alert_id).await?; - alerts.delete(alert_id, &tenant_id).await?; + alerts.delete(alert_id, tenant_id).await?; alerts.update(&*new_alert).await; // only restart the task if the state was not set to disabled @@ -658,8 +676,7 @@ pub async fn modify_alert( alerts.start_task(new_alert.clone_box()).await?; } - let config = new_alert.to_alert_config().to_response(); - Ok(web::Json(config)) + Ok(new_alert.to_alert_config().to_response()) } // PUT /alerts/{alert_id}/evaluate_alert diff --git a/src/handlers/http/targets.rs b/src/handlers/http/targets.rs index ae9d312ea..286bc1ff8 100644 --- a/src/handlers/http/targets.rs +++ b/src/handlers/http/targets.rs @@ -136,12 +136,23 @@ pub async fn get_internal( pub async fn update( req: HttpRequest, target_id: Path, - Json(mut target): Json, + Json(target): Json, ) -> Result { let target_id = target_id.into_inner(); let tenant_id = get_tenant_id_from_request(&req); + Ok(web::Json( + update_internal(target_id, target, &tenant_id).await?, + )) +} + +/// Shared target update implementation for HTTP handlers and in-process callers. +pub async fn update_internal( + target_id: Ulid, + mut target: Target, + tenant_id: &Option, +) -> Result { // if target_id does not exist, error - let old_target = TARGETS.get_target_by_id(&target_id, &tenant_id).await?; + let old_target = TARGETS.get_target_by_id(&target_id, tenant_id).await?; // do not allow modifying name if old_target.name != target.name { @@ -152,22 +163,29 @@ pub async fn update( // esnure that the supplied target id is assigned to the target config target.id = target_id; - target.tenant = tenant_id; + target.tenant.clone_from(tenant_id); target.validate_outbound_policy().await?; // should check for duplicacy and liveness (??) // add to the map TARGETS.update(target.clone()).await?; - Ok(web::Json(target.mask())) + Ok(serde_json::to_value(target.mask())?) } // DELETE /targets/{target_id} pub async fn delete(req: HttpRequest, target_id: Path) -> Result { let target_id = target_id.into_inner(); let tenant_id = get_tenant_id_from_request(&req); - let target = TARGETS.delete(&target_id, &tenant_id).await?; + Ok(web::Json(delete_internal(target_id, &tenant_id).await?)) +} - Ok(web::Json(target.mask())) +/// Shared target deletion implementation for HTTP handlers and in-process callers. +pub async fn delete_internal( + target_id: Ulid, + tenant_id: &Option, +) -> Result { + let target = TARGETS.delete(&target_id, tenant_id).await?; + Ok(serde_json::to_value(target.mask())?) } #[cfg(test)] diff --git a/src/handlers/http/users/dashboards.rs b/src/handlers/http/users/dashboards.rs index d7addb905..7beaf1eef 100644 --- a/src/handlers/http/users/dashboards.rs +++ b/src/handlers/http/users/dashboards.rs @@ -105,19 +105,28 @@ pub async fn get_dashboard_internal( pub async fn create_dashboard( req: HttpRequest, - Json(mut dashboard): Json, + Json(dashboard): Json, ) -> Result { + let (user_id, tenant_id) = get_user_and_tenant_from_request(&req)?; + let dashboard = create_dashboard_internal(&user_id, dashboard, &tenant_id).await?; + Ok((web::Json(dashboard), StatusCode::OK)) +} + +pub async fn create_dashboard_internal( + user_id: &str, + mut dashboard: Dashboard, + tenant_id: &Option, +) -> Result { if dashboard.title.is_empty() { return Err(DashboardError::Metadata("Title must be provided")); } - let (user_id, tenant_id) = get_user_and_tenant_from_request(&req)?; - dashboard.tenant_id.clone_from(&tenant_id); - let user_id = get_hash(&user_id); + dashboard.tenant_id.clone_from(tenant_id); + let user_id = get_hash(user_id); DASHBOARDS - .create(&user_id, &mut dashboard, &tenant_id) + .create(&user_id, &mut dashboard, tenant_id) .await?; - Ok((web::Json(dashboard), StatusCode::OK)) + Ok(dashboard) } pub async fn update_dashboard( @@ -126,15 +135,8 @@ pub async fn update_dashboard( dashboard: Option>, ) -> Result { let (user_id, tenant_id) = get_user_and_tenant_from_request(&req)?; - let user_id = get_hash(&user_id); let dashboard_id = validate_dashboard_id(dashboard_id.into_inner())?; let is_admin = is_admin(&req).map_err(|e| DashboardError::Custom(e.to_string()))?; - let mut existing_dashboard = DASHBOARDS - .get_dashboard_by_user(dashboard_id, &user_id, is_admin, &tenant_id) - .await - .ok_or(DashboardError::Metadata( - "Dashboard does not exist or user is not authorized", - ))?; let query_map = web::Query::>::from_query(req.query_string()) .map_err(|_| DashboardError::InvalidQueryParameter)?; @@ -149,7 +151,9 @@ pub async fn update_dashboard( )); } - let mut final_dashboard = if has_query_params { + let final_dashboard = if has_query_params { + let mut existing_dashboard = + get_dashboard_for_user_internal(&user_id, dashboard_id, is_admin, &tenant_id).await?; // Apply partial updates from query parameters if let Some(is_favorite) = query_map.get("isFavorite") { existing_dashboard.is_favorite = Some(is_favorite == "true"); @@ -176,35 +180,63 @@ pub async fn update_dashboard( } existing_dashboard } else { - let dashboard = dashboard + dashboard .ok_or(DashboardError::Metadata("Request body is required"))? - .into_inner(); - if let Some(tiles) = &dashboard.tiles { - if tiles.iter().any(|tile| tile.tile_id.is_nil()) { - return Err(DashboardError::Metadata("Tile ID must be provided")); - } - - // Check if tile_id are unique - let unique_tiles: Vec<_> = tiles - .iter() - .map(|tile| tile.tile_id) - .collect::>() - .into_iter() - .collect(); + .into_inner() + }; + let final_dashboard = update_dashboard_internal( + &user_id, + dashboard_id, + final_dashboard, + is_admin, + &tenant_id, + ) + .await?; - if unique_tiles.len() != tiles.len() { - return Err(DashboardError::Metadata("Tile IDs must be unique")); - } - } + Ok((web::Json(final_dashboard), StatusCode::OK)) +} - dashboard - }; +pub async fn get_dashboard_for_user_internal( + user_id: &str, + dashboard_id: ulid::Ulid, + is_admin: bool, + tenant_id: &Option, +) -> Result { + let user_id = get_hash(user_id); + DASHBOARDS + .get_dashboard_by_user(dashboard_id, &user_id, is_admin, tenant_id) + .await + .ok_or(DashboardError::Metadata( + "Dashboard does not exist or user is not authorized", + )) +} +pub async fn update_dashboard_internal( + user_id: &str, + dashboard_id: ulid::Ulid, + mut dashboard: Dashboard, + is_admin: bool, + tenant_id: &Option, +) -> Result { + // Authorize ownership before validating and persisting the replacement. + get_dashboard_for_user_internal(user_id, dashboard_id, is_admin, tenant_id).await?; + if let Some(tiles) = &dashboard.tiles { + if tiles.iter().any(|tile| tile.tile_id.is_nil()) { + return Err(DashboardError::Metadata("Tile ID must be provided")); + } + let unique_tiles = tiles + .iter() + .map(|tile| tile.tile_id) + .collect::>(); + if unique_tiles.len() != tiles.len() { + return Err(DashboardError::Metadata("Tile IDs must be unique")); + } + } + let user_id = get_hash(user_id); DASHBOARDS - .update(&user_id, dashboard_id, &mut final_dashboard, &tenant_id) + .update(&user_id, dashboard_id, &mut dashboard, tenant_id) .await?; - - Ok((web::Json(final_dashboard), StatusCode::OK)) + Ok(dashboard) } pub async fn delete_dashboard( @@ -212,36 +244,50 @@ pub async fn delete_dashboard( dashboard_id: Path, ) -> Result { let (user_id, tenant_id) = get_user_and_tenant_from_request(&req)?; - let user_id = get_hash(&user_id); let is_admin = is_admin(&req).map_err(|e| DashboardError::Custom(e.to_string()))?; let dashboard_id = validate_dashboard_id(dashboard_id.into_inner())?; - - DASHBOARDS - .delete_dashboard(&user_id, dashboard_id, is_admin, &tenant_id) - .await?; + delete_dashboard_internal(&user_id, dashboard_id, is_admin, &tenant_id).await?; Ok(HttpResponse::Ok().finish()) } +pub async fn delete_dashboard_internal( + user_id: &str, + dashboard_id: ulid::Ulid, + is_admin: bool, + tenant_id: &Option, +) -> Result<(), DashboardError> { + DASHBOARDS + .delete_dashboard(&get_hash(user_id), dashboard_id, is_admin, tenant_id) + .await +} + pub async fn add_tile( req: HttpRequest, dashboard_id: Path, Json(tile): Json, ) -> Result { - if tile.tile_id.is_nil() { - return Err(DashboardError::Metadata("Tile ID must be provided")); - } - let (user_id, tenant_id) = get_user_and_tenant_from_request(&req)?; - let user_id = get_hash(&user_id); let dashboard_id = validate_dashboard_id(dashboard_id.into_inner())?; let is_admin = is_admin(&req).map_err(|e| DashboardError::Custom(e.to_string()))?; + let dashboard = add_tile_internal(&user_id, dashboard_id, tile, is_admin, &tenant_id).await?; + Ok((web::Json(dashboard), StatusCode::OK)) +} - let mut dashboard = DASHBOARDS - .get_dashboard_by_user(dashboard_id, &user_id, is_admin, &tenant_id) - .await - .ok_or(DashboardError::Unauthorized)?; +pub async fn add_tile_internal( + user_id: &str, + dashboard_id: ulid::Ulid, + tile: Tile, + is_admin: bool, + tenant_id: &Option, +) -> Result { + if tile.tile_id.is_nil() { + return Err(DashboardError::Metadata("Tile ID must be provided")); + } + + let mut dashboard = + get_dashboard_for_user_internal(user_id, dashboard_id, is_admin, tenant_id).await?; let tiles = dashboard.tiles.get_or_insert_with(Vec::new); @@ -251,11 +297,11 @@ pub async fn add_tile( } tiles.push(tile); + let user_id = get_hash(user_id); DASHBOARDS - .update(&user_id, dashboard_id, &mut dashboard, &tenant_id) + .update(&user_id, dashboard_id, &mut dashboard, tenant_id) .await?; - - Ok((web::Json(dashboard), StatusCode::OK)) + Ok(dashboard) } pub async fn list_tags(req: HttpRequest) -> Result { diff --git a/src/tool_catalog.rs b/src/tool_catalog.rs index e5a878738..61a15dba0 100644 --- a/src/tool_catalog.rs +++ b/src/tool_catalog.rs @@ -323,9 +323,9 @@ fn target_schema() -> Value { ) } -/// Read-only tools implemented by Parseable OSS and safe for generic external -/// invocation without a confirmation flow. -pub fn oss_read_tool_specs() -> Vec { +/// Canonical OSS catalog used by internal agents, external MCP clients, and +/// Enterprise. Tool effects distinguish reads from confirmed actions. +pub fn oss_tool_specs() -> Vec { vec![ ToolSpec::new( "list_datasets", @@ -678,14 +678,6 @@ pub fn oss_read_tool_specs() -> Vec { &[], ), ), - ] -} - -/// Effectful OSS tools retained in the canonical registry for internal agents. -/// Generic external invocation must not use this catalog until confirmation is -/// server-verifiable. -pub fn oss_action_tool_specs() -> Vec { - vec![ ToolSpec::new( "enable_alert", "Enable alert", @@ -725,15 +717,13 @@ pub fn oss_action_tool_specs() -> Vec { ToolSpec::new( "update_alert", "Update alert", - "Update an alert using an explicit validated patch or complete replacement after confirmation.", + "Replace an alert with a complete validated specification after confirmation.", object_schema( json!({ "alertId": string_property("Alert ID to update."), - "patch": { "type": "object", "additionalProperties": true, "description": "Explicit camelCase alert patch." }, - "replacement": { "type": "object", "additionalProperties": true, "description": "Optional complete validated alert replacement." }, - "idempotencyKey": string_property("Server-scoped idempotency key.") + "spec": { "type": "object", "additionalProperties": true, "description": "Complete camelCase alert specification." } }), - &["alertId", "idempotencyKey"], + &["alertId", "spec"], ), ), ToolSpec::new( @@ -741,25 +731,20 @@ pub fn oss_action_tool_specs() -> Vec { "Delete alert", "Delete one alert after explicit confirmation.", object_schema( - json!({ - "alertId": string_property("Alert ID to delete."), - "idempotencyKey": string_property("Server-scoped idempotency key.") - }), - &["alertId", "idempotencyKey"], + json!({ "alertId": string_property("Alert ID to delete.") }), + &["alertId"], ), ), ToolSpec::new( "update_alert_target", "Update alert target", - "Update an alert target using an explicit patch or replacement after confirmation.", + "Replace an alert target with a complete specification after confirmation.", object_schema( json!({ "targetId": string_property("Alert-target ID."), - "patch": { "type": "object", "additionalProperties": true }, - "replacement": { "type": "object", "additionalProperties": true }, - "idempotencyKey": string_property("Server-scoped idempotency key.") + "spec": { "type": "object", "additionalProperties": true } }), - &["targetId", "idempotencyKey"], + &["targetId", "spec"], ), ), ToolSpec::new( @@ -767,11 +752,8 @@ pub fn oss_action_tool_specs() -> Vec { "Delete alert target", "Delete one alert target after explicit confirmation.", object_schema( - json!({ - "targetId": string_property("Alert-target ID."), - "idempotencyKey": string_property("Server-scoped idempotency key.") - }), - &["targetId", "idempotencyKey"], + json!({ "targetId": string_property("Alert-target ID.") }), + &["targetId"], ), ), ToolSpec::new( @@ -781,10 +763,9 @@ pub fn oss_action_tool_specs() -> Vec { object_schema( json!({ "dashboardId": string_property("Dashboard ID."), - "tile": { "type": "object", "additionalProperties": true }, - "idempotencyKey": string_property("Server-scoped idempotency key.") + "tile": { "type": "object", "additionalProperties": true } }), - &["dashboardId", "tile", "idempotencyKey"], + &["dashboardId", "tile"], ), ), ToolSpec::new( @@ -792,25 +773,20 @@ pub fn oss_action_tool_specs() -> Vec { "Create dashboard", "Persist a validated dashboard after confirmation.", object_schema( - json!({ - "dashboard": { "type": "object", "additionalProperties": true }, - "idempotencyKey": string_property("Server-scoped idempotency key.") - }), - &["dashboard", "idempotencyKey"], + json!({ "dashboard": { "type": "object", "additionalProperties": true } }), + &["dashboard"], ), ), ToolSpec::new( "update_dashboard", "Update dashboard", - "Update a dashboard using an explicit patch or validated replacement after confirmation.", + "Replace a dashboard with a complete validated dashboard after confirmation.", object_schema( json!({ "dashboardId": string_property("Dashboard ID."), - "patch": { "type": "object", "additionalProperties": true }, - "replacement": { "type": "object", "additionalProperties": true }, - "idempotencyKey": string_property("Server-scoped idempotency key.") + "dashboard": { "type": "object", "additionalProperties": true } }), - &["dashboardId", "idempotencyKey"], + &["dashboardId", "dashboard"], ), ), ToolSpec::new( @@ -818,31 +794,17 @@ pub fn oss_action_tool_specs() -> Vec { "Delete dashboard", "Delete one dashboard after explicit confirmation.", object_schema( - json!({ - "dashboardId": string_property("Dashboard ID."), - "idempotencyKey": string_property("Server-scoped idempotency key.") - }), - &["dashboardId", "idempotencyKey"], + json!({ "dashboardId": string_property("Dashboard ID.") }), + &["dashboardId"], ), ), ] } -/// Canonical OSS catalog used by internal agents and imported by Enterprise. -pub fn oss_tool_specs() -> Vec { - let mut tools = oss_read_tool_specs(); - tools.extend(oss_action_tool_specs()); - tools -} - pub fn is_oss_tool(name: &str) -> bool { oss_tool_specs().iter().any(|tool| tool.name == name) } -pub fn is_oss_read_tool(name: &str) -> bool { - oss_read_tool_specs().iter().any(|tool| tool.name == name) -} - pub fn registry_http_response(request: &HttpRequest, tools: Vec) -> HttpResponse { let payload = json!({ "tools": tools }); let body = serde_json::to_string(&payload).expect("tool registry is JSON serializable"); @@ -869,7 +831,7 @@ pub fn registry_http_response(request: &HttpRequest, tools: Vec) -> HttpR /// `GET /api/prism/v1/llm/tools/list` pub async fn list_tool_registry(request: HttpRequest) -> HttpResponse { - let tools = oss_read_tool_specs() + let tools = oss_tool_specs() .into_iter() .map(|tool| { json!({ @@ -891,10 +853,7 @@ mod tests { use actix_web::{App, http::header, test, web}; use serde_json::Value; - use super::{ - ToolCallRequest, ToolCallResult, ToolEffect, is_oss_read_tool, list_tool_registry, - oss_action_tool_specs, oss_read_tool_specs, oss_tool_specs, - }; + use super::{ToolCallRequest, ToolCallResult, ToolEffect, list_tool_registry, oss_tool_specs}; #[actix_web::test] async fn oss_catalog_has_unique_names() { @@ -918,27 +877,47 @@ mod tests { assert_eq!( response["tools"].as_array().unwrap().len(), - oss_read_tool_specs().len() + oss_tool_specs().len() ); assert_eq!(response["tools"][0]["name"], "list_datasets"); assert_eq!(response["tools"][0]["inputSchema"]["type"], "object"); - assert!(oss_read_tool_specs().iter().all(|tool| matches!( - tool.effect(), - ToolEffect::ReadOnly | ToolEffect::ExpensiveRead - ))); assert!( response["tools"] .as_array() .unwrap() .iter() - .all(|tool| tool["name"] != "create_alert") + .any(|tool| { tool["name"] == "create_alert" }) + ); + assert!( + response["tools"] + .as_array() + .unwrap() + .iter() + .any(|tool| { tool["name"] == "update_alert" }) ); } #[actix_web::test] async fn read_and_action_catalogs_are_structurally_separated() { - let read_tools = oss_read_tool_specs(); - let action_tools = oss_action_tool_specs(); + let tools = oss_tool_specs(); + let read_tools = tools + .iter() + .filter(|tool| { + matches!( + tool.effect(), + ToolEffect::ReadOnly | ToolEffect::ExpensiveRead + ) + }) + .collect::>(); + let action_tools = tools + .iter() + .filter(|tool| { + matches!( + tool.effect(), + ToolEffect::Mutation | ToolEffect::ExternalSideEffect + ) + }) + .collect::>(); assert_eq!(read_tools.len(), 37); assert_eq!(action_tools.len(), 13); @@ -950,8 +929,6 @@ mod tests { tool.effect(), ToolEffect::Mutation | ToolEffect::ExternalSideEffect ))); - assert!(read_tools.iter().all(|tool| is_oss_read_tool(tool.name))); - assert!(action_tools.iter().all(|tool| !is_oss_read_tool(tool.name))); } #[actix_web::test] @@ -992,7 +969,7 @@ mod tests { #[actix_web::test] async fn oss_catalog_exposes_list_log_errors() { - let tool = oss_read_tool_specs() + let tool = oss_tool_specs() .into_iter() .find(|tool| tool.name == "list_log_errors") .expect("list_log_errors must be available in OSS"); @@ -1009,7 +986,7 @@ mod tests { #[actix_web::test] async fn oss_catalog_exposes_list_trace_errors() { - let tool = oss_read_tool_specs() + let tool = oss_tool_specs() .into_iter() .find(|tool| tool.name == "list_trace_errors") .expect("list_trace_errors must be available in OSS"); @@ -1026,7 +1003,7 @@ mod tests { #[actix_web::test] async fn oss_catalog_exposes_trace_activity() { - let tool = oss_read_tool_specs() + let tool = oss_tool_specs() .into_iter() .find(|tool| tool.name == "get_trace_activity") .expect("get_trace_activity must be available in OSS"); diff --git a/src/tool_executor.rs b/src/tool_executor.rs index dc71c78e6..ead5a176c 100644 --- a/src/tool_executor.rs +++ b/src/tool_executor.rs @@ -26,8 +26,11 @@ use crate::{ parseable::PARSEABLE, prism::logstream::get_prism_logstream_info, rbac::{Response, Users, map::SessionKey, role::Action}, - tool_catalog::{ToolCallRequest, ToolCallResult, oss_read_tool_specs, oss_tool_specs}, - utils::{actix::extract_session_key_from_req, get_tenant_id_from_request, time::TimeRange}, + tool_catalog::{ToolCallRequest, ToolCallResult, oss_tool_specs}, + utils::{ + actix::extract_session_key_from_req, get_tenant_id_from_request, is_admin_for_session, + time::TimeRange, + }, }; fn required_string<'a>(arguments: &'a Value, name: &str) -> Result<&'a str, String> { @@ -38,6 +41,16 @@ fn required_string<'a>(arguments: &'a Value, name: &str) -> Result<&'a str, Stri .ok_or_else(|| format!("missing required string argument `{name}`")) } +fn session_user_id(session_key: &SessionKey) -> Result { + match session_key { + SessionKey::BasicAuth { username, .. } => Ok(username.clone()), + SessionKey::SessionId(_) => Users + .get_userid_from_session(session_key) + .map(|(user_id, _)| user_id) + .ok_or_else(|| "user session is not active".to_owned()), + } +} + fn bounded_integer(arguments: &Value, name: &str, default: u64, min: u64, max: u64) -> u64 { arguments .get(name) @@ -1146,6 +1159,33 @@ async fn execute_alert_tool( serde_json::to_value(response).map_err(|error| error.to_string()) } "create_alert" => create_alert(arguments, context).await, + "update_alert" => { + authorize(context.session_key, Action::PutAlert, None, None)?; + let id = Ulid::from_string(required_string(arguments, "alertId")?) + .map_err(|error| error.to_string())?; + let request = serde_json::from_value( + arguments + .get("spec") + .cloned() + .ok_or_else(|| "missing required object argument `spec`".to_owned())?, + ) + .map_err(|error| format!("invalid updated alert: {error}"))?; + serde_json::to_value( + alerts::modify_alert_internal(context.session_key, id, request, context.tenant_id) + .await + .map_err(|error| error.to_string())?, + ) + .map_err(|error| error.to_string()) + } + "delete_alert" => { + authorize(context.session_key, Action::DeleteAlert, None, None)?; + let id = Ulid::from_string(required_string(arguments, "alertId")?) + .map_err(|error| error.to_string())?; + let message = alerts::delete_internal(context.session_key, id, context.tenant_id) + .await + .map_err(|error| error.to_string())?; + Ok(json!({ "message": message })) + } _ => unreachable!("alert dispatcher only receives known tools"), } } @@ -1203,10 +1243,118 @@ async fn execute_alert_target_tool( .await .map_err(|error| error.to_string()) } + "update_alert_target" => { + authorize(context.session_key, Action::PutAlert, None, None)?; + let id = Ulid::from_string(required_string(arguments, "targetId")?) + .map_err(|error| error.to_string())?; + let target = serde_json::from_value( + arguments + .get("spec") + .cloned() + .ok_or_else(|| "missing required object argument `spec`".to_owned())?, + ) + .map_err(|error| format!("invalid updated alert target: {error}"))?; + targets::update_internal(id, target, context.tenant_id) + .await + .map_err(|error| error.to_string()) + } + "delete_alert_target" => { + authorize(context.session_key, Action::DeleteAlert, None, None)?; + let id = Ulid::from_string(required_string(arguments, "targetId")?) + .map_err(|error| error.to_string())?; + targets::delete_internal(id, context.tenant_id) + .await + .map_err(|error| error.to_string()) + } _ => unreachable!("alert target dispatcher only receives known tools"), } } +async fn execute_dashboard_action_tool( + name: &str, + arguments: &Value, + context: OssToolContext<'_>, +) -> Result { + let user_id = session_user_id(context.session_key)?; + let is_admin = is_admin_for_session(context.session_key); + match name { + "create_dashboard" => { + authorize(context.session_key, Action::CreateDashboard, None, None)?; + let dashboard = serde_json::from_value( + arguments + .get("dashboard") + .cloned() + .ok_or_else(|| "missing required object argument `dashboard`".to_owned())?, + ) + .map_err(|error| format!("invalid dashboard: {error}"))?; + serde_json::to_value( + dashboards::create_dashboard_internal(&user_id, dashboard, context.tenant_id) + .await + .map_err(|error| error.to_string())?, + ) + .map_err(|error| error.to_string()) + } + "update_dashboard" => { + authorize(context.session_key, Action::CreateDashboard, None, None)?; + let id = crate::users::dashboards::validate_dashboard_id( + required_string(arguments, "dashboardId")?.to_owned(), + ) + .map_err(|error| error.to_string())?; + let dashboard = serde_json::from_value( + arguments + .get("dashboard") + .cloned() + .ok_or_else(|| "missing required object argument `dashboard`".to_owned())?, + ) + .map_err(|error| format!("invalid updated dashboard: {error}"))?; + serde_json::to_value( + dashboards::update_dashboard_internal( + &user_id, + id, + dashboard, + is_admin, + context.tenant_id, + ) + .await + .map_err(|error| error.to_string())?, + ) + .map_err(|error| error.to_string()) + } + "save_tile" => { + authorize(context.session_key, Action::CreateDashboard, None, None)?; + let id = crate::users::dashboards::validate_dashboard_id( + required_string(arguments, "dashboardId")?.to_owned(), + ) + .map_err(|error| error.to_string())?; + let tile = serde_json::from_value( + arguments + .get("tile") + .cloned() + .ok_or_else(|| "missing required object argument `tile`".to_owned())?, + ) + .map_err(|error| format!("invalid tile: {error}"))?; + serde_json::to_value( + dashboards::add_tile_internal(&user_id, id, tile, is_admin, context.tenant_id) + .await + .map_err(|error| error.to_string())?, + ) + .map_err(|error| error.to_string()) + } + "delete_dashboard" => { + authorize(context.session_key, Action::DeleteDashboard, None, None)?; + let id = crate::users::dashboards::validate_dashboard_id( + required_string(arguments, "dashboardId")?.to_owned(), + ) + .map_err(|error| error.to_string())?; + dashboards::delete_dashboard_internal(&user_id, id, is_admin, context.tenant_id) + .await + .map_err(|error| error.to_string())?; + Ok(json!({ "deleted": true, "dashboardId": id })) + } + _ => unreachable!("dashboard action dispatcher only receives known tools"), + } +} + async fn execute_trace_tool( name: &str, arguments: &Value, @@ -1680,10 +1828,14 @@ async fn execute_oss_tool_inner( "list_log_errors" => list_log_errors(arguments, context).await, "query_sql" | "explain_query" => execute_sql_tool(name, arguments, context).await, "list_alerts" | "get_alert" | "list_alert_tags" | "enable_alert" | "disable_alert" - | "evaluate_alert" | "create_alert" => execute_alert_tool(name, arguments, context).await, - "list_alert_targets" | "get_alert_target" | "create_alert_target" => { - execute_alert_target_tool(name, arguments, context).await + | "evaluate_alert" | "create_alert" | "update_alert" | "delete_alert" => { + execute_alert_tool(name, arguments, context).await } + "list_alert_targets" + | "get_alert_target" + | "create_alert_target" + | "update_alert_target" + | "delete_alert_target" => execute_alert_target_tool(name, arguments, context).await, "get_traces" | "get_trace" | "list_trace_errors" => { execute_trace_tool(name, arguments, context).await } @@ -1700,15 +1852,17 @@ async fn execute_oss_tool_inner( | "list_dashboard_tags" | "get_dashboard" | "list_dashboards" => execute_saved_object_tool(name, arguments, context).await, + "save_tile" | "create_dashboard" | "update_dashboard" | "delete_dashboard" => { + execute_dashboard_action_tool(name, arguments, context).await + } _ => Err(format!( "tool `{name}` is registered but does not yet have an OSS internal executor" )), } } -/// Execute a registered OSS tool in-process. Enterprise uses this only after -/// applying its own agent policy and action-confirmation checks, so catalog -/// validation and the underlying platform implementation remain canonical. +/// Execute a registered OSS tool in-process. Agent callers apply their own +/// policies; external MCP callers are treated as client-confirmed actions. pub async fn execute_oss_tool( name: &str, arguments: &Value, @@ -1759,7 +1913,7 @@ pub async fn call_tool_registry( "Your session has expired or is no longer valid. Please re-authenticate to access this resource.", )); } - let Some(spec) = oss_read_tool_specs() + let Some(spec) = crate::tool_catalog::oss_tool_specs() .into_iter() .find(|tool| tool.name == call.name) else { From d62bd457b7fa070304f196067b633e2062901081 Mon Sep 17 00:00:00 2001 From: Nikhil Sinha Date: Wed, 28 Oct 2026 23:34:42 +0530 Subject: [PATCH 2/2] deepsource fix --- src/tool_executor.rs | 316 +++++++++++++++++++++++-------------------- 1 file changed, 169 insertions(+), 147 deletions(-) diff --git a/src/tool_executor.rs b/src/tool_executor.rs index ead5a176c..2eee2153e 100644 --- a/src/tool_executor.rs +++ b/src/tool_executor.rs @@ -1110,84 +1110,70 @@ async fn execute_alert_tool( context: OssToolContext<'_>, ) -> Result { match name { - "list_alerts" => { - authorize(context.session_key, Action::GetAlert, None, None)?; - serde_json::to_value( - alerts::list_internal(context.session_key.clone(), &HashMap::new()) - .await - .map_err(|error| error.to_string())?, - ) - .map_err(|error| error.to_string()) - } - "get_alert" => { - authorize(context.session_key, Action::GetAlert, None, None)?; - let id = Ulid::from_string(required_string(arguments, "alert_id")?) - .map_err(|error| error.to_string())?; - serde_json::to_value( - alerts::get_internal(context.session_key, id, context.tenant_id) - .await - .map_err(|error| error.to_string())?, - ) - .map_err(|error| error.to_string()) - } - "list_alert_tags" => { - authorize(context.session_key, Action::GetAlert, None, None)?; - serde_json::to_value( - alerts::list_tags_internal(context.tenant_id) - .await - .map_err(|error| error.to_string())?, - ) - .map_err(|error| error.to_string()) - } + "list_alerts" => list_alerts(context).await, + "get_alert" => get_alert(arguments, context).await, + "list_alert_tags" => list_alert_tags(context).await, "enable_alert" | "disable_alert" | "evaluate_alert" => { - authorize(context.session_key, Action::PutAlert, None, None)?; - let id = Ulid::from_string(required_string(arguments, "alert_id")?) - .map_err(|error| error.to_string())?; - let response = match name { - "enable_alert" => { - alerts::enable_alert_internal(context.session_key, id, context.tenant_id).await - } - "disable_alert" => { - alerts::disable_alert_internal(context.session_key, id, context.tenant_id).await - } - _ => { - alerts::evaluate_alert_internal(context.session_key, id, context.tenant_id) - .await - } - } - .map_err(|error| error.to_string())?; - serde_json::to_value(response).map_err(|error| error.to_string()) + change_alert_state(name, arguments, context).await } "create_alert" => create_alert(arguments, context).await, - "update_alert" => { - authorize(context.session_key, Action::PutAlert, None, None)?; - let id = Ulid::from_string(required_string(arguments, "alertId")?) - .map_err(|error| error.to_string())?; - let request = serde_json::from_value( - arguments - .get("spec") - .cloned() - .ok_or_else(|| "missing required object argument `spec`".to_owned())?, - ) - .map_err(|error| format!("invalid updated alert: {error}"))?; - serde_json::to_value( - alerts::modify_alert_internal(context.session_key, id, request, context.tenant_id) - .await - .map_err(|error| error.to_string())?, - ) - .map_err(|error| error.to_string()) + "update_alert" => update_alert(arguments, context).await, + "delete_alert" => delete_alert(arguments, context).await, + _ => unreachable!("alert dispatcher only receives known tools"), + } +} + +async fn list_alerts(context: OssToolContext<'_>) -> Result { + authorize(context.session_key, Action::GetAlert, None, None)?; + serde_json::to_value( + alerts::list_internal(context.session_key.clone(), &HashMap::new()) + .await + .map_err(|error| error.to_string())?, + ) + .map_err(|error| error.to_string()) +} + +async fn get_alert(arguments: &Value, context: OssToolContext<'_>) -> Result { + authorize(context.session_key, Action::GetAlert, None, None)?; + let id = Ulid::from_string(required_string(arguments, "alert_id")?) + .map_err(|error| error.to_string())?; + serde_json::to_value( + alerts::get_internal(context.session_key, id, context.tenant_id) + .await + .map_err(|error| error.to_string())?, + ) + .map_err(|error| error.to_string()) +} + +async fn list_alert_tags(context: OssToolContext<'_>) -> Result { + authorize(context.session_key, Action::GetAlert, None, None)?; + serde_json::to_value( + alerts::list_tags_internal(context.tenant_id) + .await + .map_err(|error| error.to_string())?, + ) + .map_err(|error| error.to_string()) +} + +async fn change_alert_state( + name: &str, + arguments: &Value, + context: OssToolContext<'_>, +) -> Result { + authorize(context.session_key, Action::PutAlert, None, None)?; + let id = Ulid::from_string(required_string(arguments, "alert_id")?) + .map_err(|error| error.to_string())?; + let response = match name { + "enable_alert" => { + alerts::enable_alert_internal(context.session_key, id, context.tenant_id).await } - "delete_alert" => { - authorize(context.session_key, Action::DeleteAlert, None, None)?; - let id = Ulid::from_string(required_string(arguments, "alertId")?) - .map_err(|error| error.to_string())?; - let message = alerts::delete_internal(context.session_key, id, context.tenant_id) - .await - .map_err(|error| error.to_string())?; - Ok(json!({ "message": message })) + "disable_alert" => { + alerts::disable_alert_internal(context.session_key, id, context.tenant_id).await } - _ => unreachable!("alert dispatcher only receives known tools"), + _ => alerts::evaluate_alert_internal(context.session_key, id, context.tenant_id).await, } + .map_err(|error| error.to_string())?; + serde_json::to_value(response).map_err(|error| error.to_string()) } async fn create_alert(arguments: &Value, context: OssToolContext<'_>) -> Result { @@ -1207,6 +1193,35 @@ async fn create_alert(arguments: &Value, context: OssToolContext<'_>) -> Result< .map_err(|error| error.to_string()) } +async fn update_alert(arguments: &Value, context: OssToolContext<'_>) -> Result { + authorize(context.session_key, Action::PutAlert, None, None)?; + let id = Ulid::from_string(required_string(arguments, "alertId")?) + .map_err(|error| error.to_string())?; + let request = serde_json::from_value( + arguments + .get("spec") + .cloned() + .ok_or_else(|| "missing required object argument `spec`".to_owned())?, + ) + .map_err(|error| format!("invalid updated alert: {error}"))?; + serde_json::to_value( + alerts::modify_alert_internal(context.session_key, id, request, context.tenant_id) + .await + .map_err(|error| error.to_string())?, + ) + .map_err(|error| error.to_string()) +} + +async fn delete_alert(arguments: &Value, context: OssToolContext<'_>) -> Result { + authorize(context.session_key, Action::DeleteAlert, None, None)?; + let id = Ulid::from_string(required_string(arguments, "alertId")?) + .map_err(|error| error.to_string())?; + let message = alerts::delete_internal(context.session_key, id, context.tenant_id) + .await + .map_err(|error| error.to_string())?; + Ok(json!({ "message": message })) +} + async fn execute_alert_target_tool( name: &str, arguments: &Value, @@ -1275,86 +1290,93 @@ async fn execute_dashboard_action_tool( arguments: &Value, context: OssToolContext<'_>, ) -> Result { - let user_id = session_user_id(context.session_key)?; - let is_admin = is_admin_for_session(context.session_key); match name { - "create_dashboard" => { - authorize(context.session_key, Action::CreateDashboard, None, None)?; - let dashboard = serde_json::from_value( - arguments - .get("dashboard") - .cloned() - .ok_or_else(|| "missing required object argument `dashboard`".to_owned())?, - ) - .map_err(|error| format!("invalid dashboard: {error}"))?; - serde_json::to_value( - dashboards::create_dashboard_internal(&user_id, dashboard, context.tenant_id) - .await - .map_err(|error| error.to_string())?, - ) - .map_err(|error| error.to_string()) - } - "update_dashboard" => { - authorize(context.session_key, Action::CreateDashboard, None, None)?; - let id = crate::users::dashboards::validate_dashboard_id( - required_string(arguments, "dashboardId")?.to_owned(), - ) - .map_err(|error| error.to_string())?; - let dashboard = serde_json::from_value( - arguments - .get("dashboard") - .cloned() - .ok_or_else(|| "missing required object argument `dashboard`".to_owned())?, - ) - .map_err(|error| format!("invalid updated dashboard: {error}"))?; - serde_json::to_value( - dashboards::update_dashboard_internal( - &user_id, - id, - dashboard, - is_admin, - context.tenant_id, - ) - .await - .map_err(|error| error.to_string())?, - ) - .map_err(|error| error.to_string()) - } - "save_tile" => { - authorize(context.session_key, Action::CreateDashboard, None, None)?; - let id = crate::users::dashboards::validate_dashboard_id( - required_string(arguments, "dashboardId")?.to_owned(), - ) - .map_err(|error| error.to_string())?; - let tile = serde_json::from_value( - arguments - .get("tile") - .cloned() - .ok_or_else(|| "missing required object argument `tile`".to_owned())?, - ) - .map_err(|error| format!("invalid tile: {error}"))?; - serde_json::to_value( - dashboards::add_tile_internal(&user_id, id, tile, is_admin, context.tenant_id) - .await - .map_err(|error| error.to_string())?, - ) - .map_err(|error| error.to_string()) - } - "delete_dashboard" => { - authorize(context.session_key, Action::DeleteDashboard, None, None)?; - let id = crate::users::dashboards::validate_dashboard_id( - required_string(arguments, "dashboardId")?.to_owned(), - ) - .map_err(|error| error.to_string())?; - dashboards::delete_dashboard_internal(&user_id, id, is_admin, context.tenant_id) - .await - .map_err(|error| error.to_string())?; - Ok(json!({ "deleted": true, "dashboardId": id })) - } + "create_dashboard" => create_dashboard(arguments, context).await, + "update_dashboard" => update_dashboard(arguments, context).await, + "save_tile" => save_tile(arguments, context).await, + "delete_dashboard" => delete_dashboard(arguments, context).await, _ => unreachable!("dashboard action dispatcher only receives known tools"), } } +async fn create_dashboard(arguments: &Value, context: OssToolContext<'_>) -> Result { + authorize(context.session_key, Action::CreateDashboard, None, None)?; + let user_id = session_user_id(context.session_key)?; + let dashboard = serde_json::from_value( + arguments + .get("dashboard") + .cloned() + .ok_or_else(|| "missing required object argument `dashboard`".to_owned())?, + ) + .map_err(|error| format!("invalid dashboard: {error}"))?; + serde_json::to_value( + dashboards::create_dashboard_internal(&user_id, dashboard, context.tenant_id) + .await + .map_err(|error| error.to_string())?, + ) + .map_err(|error| error.to_string()) +} + +async fn update_dashboard(arguments: &Value, context: OssToolContext<'_>) -> Result { + authorize(context.session_key, Action::CreateDashboard, None, None)?; + let user_id = session_user_id(context.session_key)?; + let is_admin = is_admin_for_session(context.session_key); + let id = crate::users::dashboards::validate_dashboard_id( + required_string(arguments, "dashboardId")?.to_owned(), + ) + .map_err(|error| error.to_string())?; + let dashboard = serde_json::from_value( + arguments + .get("dashboard") + .cloned() + .ok_or_else(|| "missing required object argument `dashboard`".to_owned())?, + ) + .map_err(|error| format!("invalid updated dashboard: {error}"))?; + serde_json::to_value( + dashboards::update_dashboard_internal(&user_id, id, dashboard, is_admin, context.tenant_id) + .await + .map_err(|error| error.to_string())?, + ) + .map_err(|error| error.to_string()) +} + +async fn save_tile(arguments: &Value, context: OssToolContext<'_>) -> Result { + authorize(context.session_key, Action::CreateDashboard, None, None)?; + let user_id = session_user_id(context.session_key)?; + let is_admin = is_admin_for_session(context.session_key); + let id = crate::users::dashboards::validate_dashboard_id( + required_string(arguments, "dashboardId")?.to_owned(), + ) + .map_err(|error| error.to_string())?; + let tile = serde_json::from_value( + arguments + .get("tile") + .cloned() + .ok_or_else(|| "missing required object argument `tile`".to_owned())?, + ) + .map_err(|error| format!("invalid tile: {error}"))?; + serde_json::to_value( + dashboards::add_tile_internal(&user_id, id, tile, is_admin, context.tenant_id) + .await + .map_err(|error| error.to_string())?, + ) + .map_err(|error| error.to_string()) +} + +async fn delete_dashboard(arguments: &Value, context: OssToolContext<'_>) -> Result { + authorize(context.session_key, Action::DeleteDashboard, None, None)?; + let user_id = session_user_id(context.session_key)?; + let is_admin = is_admin_for_session(context.session_key); + let id = crate::users::dashboards::validate_dashboard_id( + required_string(arguments, "dashboardId")?.to_owned(), + ) + .map_err(|error| error.to_string())?; + dashboards::delete_dashboard_internal(&user_id, id, is_admin, context.tenant_id) + .await + .map_err(|error| error.to_string())?; + Ok(json!({ "deleted": true, "dashboardId": id })) +} + async fn execute_trace_tool( name: &str, arguments: &Value,