diff --git a/backend/.sqlx/query-127d1d6a9cec24f34ff91718cf62befb986da3c1081b08a3a650435c53a88df5.json b/backend/.sqlx/query-127d1d6a9cec24f34ff91718cf62befb986da3c1081b08a3a650435c53a88df5.json new file mode 100644 index 0000000000..1bb018341b --- /dev/null +++ b/backend/.sqlx/query-127d1d6a9cec24f34ff91718cf62befb986da3c1081b08a3a650435c53a88df5.json @@ -0,0 +1,12 @@ +{ + "db_name": "PostgreSQL", + "query": "WITH a AS (DELETE FROM slack_answered_message WHERE created_at < now() - interval '1 day')\n DELETE FROM slack_thread_agent WHERE created_at < now() - interval '90 days'", + "describe": { + "columns": [], + "parameters": { + "Left": [] + }, + "nullable": [] + }, + "hash": "127d1d6a9cec24f34ff91718cf62befb986da3c1081b08a3a650435c53a88df5" +} diff --git a/backend/.sqlx/query-8133bd7a4f8d7e81ff3420b4c7784192f5f8f5d2bfa4581e1075378f061982a8.json b/backend/.sqlx/query-35b5261c148af3bf68de2772847125dcfc4f0f8a22b5eeb927f3484d7534826c.json similarity index 58% rename from backend/.sqlx/query-8133bd7a4f8d7e81ff3420b4c7784192f5f8f5d2bfa4581e1075378f061982a8.json rename to backend/.sqlx/query-35b5261c148af3bf68de2772847125dcfc4f0f8a22b5eeb927f3484d7534826c.json index 5196e5be1b..f9ae6174ad 100644 --- a/backend/.sqlx/query-8133bd7a4f8d7e81ff3420b4c7784192f5f8f5d2bfa4581e1075378f061982a8.json +++ b/backend/.sqlx/query-35b5261c148af3bf68de2772847125dcfc4f0f8a22b5eeb927f3484d7534826c.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT channel_id, channel_name, agent_path FROM slack_channel_agent\n WHERE workspace_id = $1 ORDER BY channel_name", + "query": "SELECT c.channel_id, c.channel_name, c.agent_path FROM slack_channel_agent c\n JOIN resource r ON r.workspace_id = c.workspace_id AND r.path = c.agent_path\n WHERE c.workspace_id = $1 AND c.agent_path = $2 ORDER BY c.channel_name", "describe": { "columns": [ { @@ -21,6 +21,7 @@ ], "parameters": { "Left": [ + "Text", "Text" ] }, @@ -30,5 +31,5 @@ false ] }, - "hash": "8133bd7a4f8d7e81ff3420b4c7784192f5f8f5d2bfa4581e1075378f061982a8" + "hash": "35b5261c148af3bf68de2772847125dcfc4f0f8a22b5eeb927f3484d7534826c" } diff --git a/backend/.sqlx/query-80ef811675f39e6ab14071c5370dd9fde961177a086a931ecb4c6bb72cf70120.json b/backend/.sqlx/query-538edc1ca25281f2b2b52f55ef787f4b71f41914694d090a5f10ff6478627bb4.json similarity index 61% rename from backend/.sqlx/query-80ef811675f39e6ab14071c5370dd9fde961177a086a931ecb4c6bb72cf70120.json rename to backend/.sqlx/query-538edc1ca25281f2b2b52f55ef787f4b71f41914694d090a5f10ff6478627bb4.json index 5c48e8f618..d4ed33fbc4 100644 --- a/backend/.sqlx/query-80ef811675f39e6ab14071c5370dd9fde961177a086a931ecb4c6bb72cf70120.json +++ b/backend/.sqlx/query-538edc1ca25281f2b2b52f55ef787f4b71f41914694d090a5f10ff6478627bb4.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT channel_id, channel_name, agent_path FROM slack_channel_agent\n WHERE workspace_id = $1 AND agent_path = $2 ORDER BY channel_name", + "query": "SELECT c.channel_id, c.channel_name, c.agent_path FROM slack_channel_agent c\n JOIN resource r ON r.workspace_id = c.workspace_id AND r.path = c.agent_path\n WHERE c.workspace_id = $1 ORDER BY c.channel_name", "describe": { "columns": [ { @@ -21,7 +21,6 @@ ], "parameters": { "Left": [ - "Text", "Text" ] }, @@ -31,5 +30,5 @@ false ] }, - "hash": "80ef811675f39e6ab14071c5370dd9fde961177a086a931ecb4c6bb72cf70120" + "hash": "538edc1ca25281f2b2b52f55ef787f4b71f41914694d090a5f10ff6478627bb4" } diff --git a/backend/.sqlx/query-cad02f11ce7127c1e74eface0cb01939a9effee10fb7664d08cf39da5d2e13d7.json b/backend/.sqlx/query-cad02f11ce7127c1e74eface0cb01939a9effee10fb7664d08cf39da5d2e13d7.json deleted file mode 100644 index fd9662c731..0000000000 --- a/backend/.sqlx/query-cad02f11ce7127c1e74eface0cb01939a9effee10fb7664d08cf39da5d2e13d7.json +++ /dev/null @@ -1,12 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "DELETE FROM slack_answered_message WHERE created_at < now() - interval '1 day'", - "describe": { - "columns": [], - "parameters": { - "Left": [] - }, - "nullable": [] - }, - "hash": "cad02f11ce7127c1e74eface0cb01939a9effee10fb7664d08cf39da5d2e13d7" -} diff --git a/backend/migrations/20261008145656_slack_agents.up.sql b/backend/migrations/20261008145656_slack_agents.up.sql index b42ba8aa58..889f17c202 100644 --- a/backend/migrations/20261008145656_slack_agents.up.sql +++ b/backend/migrations/20261008145656_slack_agents.up.sql @@ -28,6 +28,7 @@ CREATE TABLE slack_thread_agent ( ); CREATE INDEX index_slack_thread_agent_workspace ON slack_thread_agent (workspace_id, agent_path); +CREATE INDEX index_slack_thread_agent_created_at ON slack_thread_agent (created_at); -- The Slack messages an agent has taken up. Slack resends an event it did not see acknowledged, -- and posts a direct message mentioning the bot as two events; only the first is answered. diff --git a/backend/windmill-api/src/agent_runs.rs b/backend/windmill-api/src/agent_runs.rs index 9f52b90d53..c2974647ef 100644 --- a/backend/windmill-api/src/agent_runs.rs +++ b/backend/windmill-api/src/agent_runs.rs @@ -42,9 +42,6 @@ pub(crate) async fn run_agent( Query(run_query): Query, Json(args): Json>>, ) -> Result<(StatusCode, String)> { - #[cfg(feature = "enterprise")] - crate::jobs::check_license_key_valid().await?; - let path = path.to_path(); check_scopes(&authed, || format!("jobs:run:agents:{path}"))?; @@ -62,6 +59,9 @@ pub(crate) async fn run_agent_as( run_query: &RunJobQuery, args: &HashMap>, ) -> Result { + #[cfg(feature = "enterprise")] + crate::jobs::check_license_key_valid().await?; + // Read through the caller's own permissions: reading the agent is what allows running it. let mut tx = user_db.clone().begin(authed).await?; let value = sqlx::query_scalar!( diff --git a/backend/windmill-api/src/slack_agents.rs b/backend/windmill-api/src/slack_agents.rs index 1f6714eb29..0366303eea 100644 --- a/backend/windmill-api/src/slack_agents.rs +++ b/backend/windmill-api/src/slack_agents.rs @@ -22,8 +22,8 @@ use windmill_common::{ db::UserDB, error::{Error, JsonResult, Result}, oauth2::WORKSPACE_SLACK_BOT_TOKEN_PATH, - utils::{require_admin, StripPath}, - variables::get_secret_value_as_admin, + utils::{calculate_hash, require_admin, StripPath}, + variables::{get_secret_value_as_admin, get_workspace_key}, BASE_URL, }; @@ -58,19 +58,23 @@ struct ChannelAgent { agent_path: String, } +/// Listed through the caller's permissions: an agent they cannot read stays unnamed. async fn list_channel_agents( - _authed: ApiAuthed, - Extension(db): Extension, + authed: ApiAuthed, + Extension(user_db): Extension, Path(w_id): Path, ) -> JsonResult> { + let mut tx = user_db.begin(&authed).await?; let rows = sqlx::query_as!( ChannelAgent, - "SELECT channel_id, channel_name, agent_path FROM slack_channel_agent - WHERE workspace_id = $1 ORDER BY channel_name", + "SELECT c.channel_id, c.channel_name, c.agent_path FROM slack_channel_agent c + JOIN resource r ON r.workspace_id = c.workspace_id AND r.path = c.agent_path + WHERE c.workspace_id = $1 ORDER BY c.channel_name", w_id ) - .fetch_all(&db) + .fetch_all(&mut *tx) .await?; + tx.commit().await?; Ok(Json(rows)) } @@ -82,8 +86,9 @@ struct AgentSlack { } async fn list_agent_channels( - _authed: ApiAuthed, + authed: ApiAuthed, Extension(db): Extension, + Extension(user_db): Extension, Path((w_id, path)): Path<(String, StripPath)>, ) -> JsonResult { let slack_team_name = sqlx::query_scalar!( @@ -94,15 +99,18 @@ async fn list_agent_channels( .fetch_optional(&db) .await? .flatten(); + let mut tx = user_db.begin(&authed).await?; let channels = sqlx::query_as!( ChannelAgent, - "SELECT channel_id, channel_name, agent_path FROM slack_channel_agent - WHERE workspace_id = $1 AND agent_path = $2 ORDER BY channel_name", + "SELECT c.channel_id, c.channel_name, c.agent_path FROM slack_channel_agent c + JOIN resource r ON r.workspace_id = c.workspace_id AND r.path = c.agent_path + WHERE c.workspace_id = $1 AND c.agent_path = $2 ORDER BY c.channel_name", w_id, path.to_path() ) - .fetch_all(&db) + .fetch_all(&mut *tx) .await?; + tx.commit().await?; Ok(Json(AgentSlack { slack_team_name, channels })) } @@ -304,6 +312,12 @@ enum Route { /// Whether an agent answers this message; if so, the answer is under way when this returns. /// `false` leaves the message to the workspace's command script. pub async fn try_answer(db: &DB, msg: SlackMessage) -> Result { + // Without it the Slack endpoints take unsigned requests, and the sender's `user_id` would + // pick whose permissions the agent runs with. + #[cfg(feature = "oauth2")] + if crate::SLACK_SIGNING_SECRET.is_none() { + return Ok(false); + } let text = decode_entities(&msg.text); let (name, question) = split_agent_name(&text); let thread_root = msg.thread_ts.clone().or_else(|| msg.ts.clone()); @@ -370,11 +384,15 @@ pub async fn try_answer(db: &DB, msg: SlackMessage) -> Result { if !first { return Ok(true); } - sqlx::query!( - "DELETE FROM slack_answered_message WHERE created_at < now() - interval '1 day'" - ) - .execute(db) - .await?; + } + // Recorded before answering, so a follow-up sent while the first answer is on its way + // already finds its agent. A named agent is recorded once the sender's name resolves. + if let (Route::Agent(path), Some(root)) = (&route, &thread_root) { + if let Err(e) = + record_thread_agent(db, &msg.team_id, &msg.channel_id, root, &w_id, path).await + { + tracing::warn!("Could not record the Slack thread's agent: {e:#}"); + } } if !matches!(route, Route::Help) { @@ -385,6 +403,7 @@ pub async fn try_answer(db: &DB, msg: SlackMessage) -> Result { // ponytail: the reply waits in this API process, so a restart mid-run loses the reply (not // the run, which still shows in Windmill); a job-completion hook would make it durable. tokio::spawn(async move { + prune(&db).await; if let Err(e) = answer(&db, &w_id, route, &question, msg).await { tracing::error!("Slack agent answer failed: {e:#}"); } @@ -392,6 +411,41 @@ pub async fn try_answer(db: &DB, msg: SlackMessage) -> Result { Ok(true) } +async fn record_thread_agent( + db: &DB, + team_id: &str, + channel_id: &str, + thread_ts: &str, + w_id: &str, + agent_path: &str, +) -> Result<()> { + sqlx::query!( + "INSERT INTO slack_thread_agent (slack_team_id, channel_id, thread_ts, workspace_id, agent_path) + VALUES ($1, $2, $3, $4, $5) ON CONFLICT DO NOTHING", + team_id, + channel_id, + thread_ts, + w_id, + agent_path + ) + .execute(db) + .await?; + Ok(()) +} + +/// Best effort: a message already marked answered must still be answered if this fails. +async fn prune(db: &DB) { + let res = sqlx::query!( + "WITH a AS (DELETE FROM slack_answered_message WHERE created_at < now() - interval '1 day') + DELETE FROM slack_thread_agent WHERE created_at < now() - interval '90 days'" + ) + .execute(db) + .await; + if let Err(e) = res { + tracing::warn!("Could not prune Slack agent routing: {e:#}"); + } +} + /// `+name rest` or `~name rest` → (name, rest). fn split_agent_name(text: &str) -> (Option<&str>, &str) { let text = text.trim(); @@ -605,23 +659,16 @@ async fn answer( .await; } - sqlx::query!( - "INSERT INTO slack_thread_agent (slack_team_id, channel_id, thread_ts, workspace_id, agent_path) - VALUES ($1, $2, $3, $4, $5) ON CONFLICT DO NOTHING", - msg.team_id, - channel, - thread_root, - w_id, - agent_path - ) - .execute(db) - .await?; + record_thread_agent(db, &msg.team_id, &channel, &thread_root, w_id, &agent_path).await?; + // Keyed with the workspace's secret: any member could rebuild the plain thread and user ids + // and load this person's conversation, tool results included, through the run API. + let workspace_key = get_workspace_key(w_id, db).await?; let run_query = RunJobQuery { - memory_id: Some(format!( - "slack:{}:{}:{}:{}", + memory_id: Some(calculate_hash(&format!( + "{workspace_key}:slack:{}:{}:{}:{}", msg.team_id, channel, thread_root, msg.user_id - )), + ))), ..Default::default() }; let args: HashMap> = [(