diff --git a/backend/.sqlx/query-34db62cf16e12ab9720e724b1f4b0031934f2b04f03162254f4f6df30e393085.json b/backend/.sqlx/query-34db62cf16e12ab9720e724b1f4b0031934f2b04f03162254f4f6df30e393085.json new file mode 100644 index 0000000000..2fa1db0d9e --- /dev/null +++ b/backend/.sqlx/query-34db62cf16e12ab9720e724b1f4b0031934f2b04f03162254f4f6df30e393085.json @@ -0,0 +1,29 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT username, email FROM usr\n WHERE workspace_id = $1 AND lower(email) = lower($2) AND NOT disabled", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "username", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "email", + "type_info": "Varchar" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + false, + false + ] + }, + "hash": "34db62cf16e12ab9720e724b1f4b0031934f2b04f03162254f4f6df30e393085" +} diff --git a/backend/.sqlx/query-4e14b95c57c6f987f020376f76ad4dd0ff0d043b3eff62863c8f8db89ad34a4e.json b/backend/.sqlx/query-4e14b95c57c6f987f020376f76ad4dd0ff0d043b3eff62863c8f8db89ad34a4e.json new file mode 100644 index 0000000000..1e83c28a49 --- /dev/null +++ b/backend/.sqlx/query-4e14b95c57c6f987f020376f76ad4dd0ff0d043b3eff62863c8f8db89ad34a4e.json @@ -0,0 +1,29 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT result AS \"result: sqlx::types::Json\",\n status = 'success' AS \"success!\"\n FROM v2_job_completed WHERE id = $1 AND workspace_id = $2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "result: sqlx::types::Json", + "type_info": "Jsonb" + }, + { + "ordinal": 1, + "name": "success!", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Uuid", + "Text" + ] + }, + "nullable": [ + true, + null + ] + }, + "hash": "4e14b95c57c6f987f020376f76ad4dd0ff0d043b3eff62863c8f8db89ad34a4e" +} diff --git a/backend/.sqlx/query-53736152fcb882b62cb1c117b61b031ca6b839023dbaa18dc601927962e6cd57.json b/backend/.sqlx/query-53736152fcb882b62cb1c117b61b031ca6b839023dbaa18dc601927962e6cd57.json new file mode 100644 index 0000000000..6176980276 --- /dev/null +++ b/backend/.sqlx/query-53736152fcb882b62cb1c117b61b031ca6b839023dbaa18dc601927962e6cd57.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT path FROM resource\n WHERE workspace_id = $1 AND resource_type = 'ai_agent'\n AND (path = $2 OR right(path, length($2) + 1) = '/' || $2)\n ORDER BY path LIMIT 6", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "path", + "type_info": "Varchar" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + false + ] + }, + "hash": "53736152fcb882b62cb1c117b61b031ca6b839023dbaa18dc601927962e6cd57" +} diff --git a/backend/.sqlx/query-538001cb77b4fe44269bff8da944b281961ac0dee681d93b8ec84d41a083b004.json b/backend/.sqlx/query-538001cb77b4fe44269bff8da944b281961ac0dee681d93b8ec84d41a083b004.json new file mode 100644 index 0000000000..6b45591c39 --- /dev/null +++ b/backend/.sqlx/query-538001cb77b4fe44269bff8da944b281961ac0dee681d93b8ec84d41a083b004.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM slack_channel_agent WHERE workspace_id = $1 AND channel_id = $2", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [] + }, + "hash": "538001cb77b4fe44269bff8da944b281961ac0dee681d93b8ec84d41a083b004" +} diff --git a/backend/.sqlx/query-62cf55f694fcf8f87ecf037cc3850e005838e5d02067460335a9ba02ac9b1898.json b/backend/.sqlx/query-62cf55f694fcf8f87ecf037cc3850e005838e5d02067460335a9ba02ac9b1898.json new file mode 100644 index 0000000000..db4f9f4b1e --- /dev/null +++ b/backend/.sqlx/query-62cf55f694fcf8f87ecf037cc3850e005838e5d02067460335a9ba02ac9b1898.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT path FROM resource WHERE workspace_id = $1 AND resource_type = 'ai_agent'\n ORDER BY path LIMIT 11", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "path", + "type_info": "Varchar" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + false + ] + }, + "hash": "62cf55f694fcf8f87ecf037cc3850e005838e5d02067460335a9ba02ac9b1898" +} diff --git a/backend/.sqlx/query-7c13828708fd11748db936bd8b8219ecdcd3fa38e1f3bc33b6f5c93a4d6fd58b.json b/backend/.sqlx/query-7c13828708fd11748db936bd8b8219ecdcd3fa38e1f3bc33b6f5c93a4d6fd58b.json new file mode 100644 index 0000000000..d5f6ae64ef --- /dev/null +++ b/backend/.sqlx/query-7c13828708fd11748db936bd8b8219ecdcd3fa38e1f3bc33b6f5c93a4d6fd58b.json @@ -0,0 +1,26 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO slack_channel_agent\n (slack_team_id, channel_id, channel_name, workspace_id, agent_path)\n VALUES ($1, $2, $3, $4, $5)\n ON CONFLICT (slack_team_id, channel_id) DO UPDATE\n SET channel_name = EXCLUDED.channel_name, agent_path = EXCLUDED.agent_path\n WHERE slack_channel_agent.workspace_id = EXCLUDED.workspace_id\n RETURNING channel_id", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "channel_id", + "type_info": "Varchar" + } + ], + "parameters": { + "Left": [ + "Varchar", + "Varchar", + "Varchar", + "Varchar", + "Varchar" + ] + }, + "nullable": [ + false + ] + }, + "hash": "7c13828708fd11748db936bd8b8219ecdcd3fa38e1f3bc33b6f5c93a4d6fd58b" +} diff --git a/backend/.sqlx/query-80ef811675f39e6ab14071c5370dd9fde961177a086a931ecb4c6bb72cf70120.json b/backend/.sqlx/query-80ef811675f39e6ab14071c5370dd9fde961177a086a931ecb4c6bb72cf70120.json new file mode 100644 index 0000000000..5c48e8f618 --- /dev/null +++ b/backend/.sqlx/query-80ef811675f39e6ab14071c5370dd9fde961177a086a931ecb4c6bb72cf70120.json @@ -0,0 +1,35 @@ +{ + "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", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "channel_id", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "channel_name", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "agent_path", + "type_info": "Varchar" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + false, + false, + false + ] + }, + "hash": "80ef811675f39e6ab14071c5370dd9fde961177a086a931ecb4c6bb72cf70120" +} diff --git a/backend/.sqlx/query-8133bd7a4f8d7e81ff3420b4c7784192f5f8f5d2bfa4581e1075378f061982a8.json b/backend/.sqlx/query-8133bd7a4f8d7e81ff3420b4c7784192f5f8f5d2bfa4581e1075378f061982a8.json new file mode 100644 index 0000000000..5196e5be1b --- /dev/null +++ b/backend/.sqlx/query-8133bd7a4f8d7e81ff3420b4c7784192f5f8f5d2bfa4581e1075378f061982a8.json @@ -0,0 +1,34 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT channel_id, channel_name, agent_path FROM slack_channel_agent\n WHERE workspace_id = $1 ORDER BY channel_name", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "channel_id", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "channel_name", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "agent_path", + "type_info": "Varchar" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + false, + false, + false + ] + }, + "hash": "8133bd7a4f8d7e81ff3420b4c7784192f5f8f5d2bfa4581e1075378f061982a8" +} diff --git a/backend/.sqlx/query-94cdc3cb0127c53d3fd255bc6b11879bb510b4e91fed87a875d1930082f94577.json b/backend/.sqlx/query-94cdc3cb0127c53d3fd255bc6b11879bb510b4e91fed87a875d1930082f94577.json new file mode 100644 index 0000000000..cff9f2b006 --- /dev/null +++ b/backend/.sqlx/query-94cdc3cb0127c53d3fd255bc6b11879bb510b4e91fed87a875d1930082f94577.json @@ -0,0 +1,30 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT workspace_id, agent_path FROM slack_thread_agent\n WHERE slack_team_id = $1 AND channel_id = $2 AND thread_ts = $3", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "workspace_id", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "agent_path", + "type_info": "Varchar" + } + ], + "parameters": { + "Left": [ + "Text", + "Text", + "Text" + ] + }, + "nullable": [ + false, + false + ] + }, + "hash": "94cdc3cb0127c53d3fd255bc6b11879bb510b4e91fed87a875d1930082f94577" +} diff --git a/backend/.sqlx/query-a4cc1e5add2876756c464e448a2b47323a96f46102be72dcc72ae6bf8bab89f5.json b/backend/.sqlx/query-a4cc1e5add2876756c464e448a2b47323a96f46102be72dcc72ae6bf8bab89f5.json new file mode 100644 index 0000000000..6607b2a7de --- /dev/null +++ b/backend/.sqlx/query-a4cc1e5add2876756c464e448a2b47323a96f46102be72dcc72ae6bf8bab89f5.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM slack_channel_agent WHERE workspace_id = $1", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [] + }, + "hash": "a4cc1e5add2876756c464e448a2b47323a96f46102be72dcc72ae6bf8bab89f5" +} diff --git a/backend/.sqlx/query-b57ef57031250f3ae74965f081cf99e9acee76a0e55522e2ee128fe75bfb4170.json b/backend/.sqlx/query-b57ef57031250f3ae74965f081cf99e9acee76a0e55522e2ee128fe75bfb4170.json new file mode 100644 index 0000000000..60f50b402e --- /dev/null +++ b/backend/.sqlx/query-b57ef57031250f3ae74965f081cf99e9acee76a0e55522e2ee128fe75bfb4170.json @@ -0,0 +1,29 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT workspace_id, agent_path FROM slack_channel_agent\n WHERE slack_team_id = $1 AND channel_id = $2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "workspace_id", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "agent_path", + "type_info": "Varchar" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + false, + false + ] + }, + "hash": "b57ef57031250f3ae74965f081cf99e9acee76a0e55522e2ee128fe75bfb4170" +} diff --git a/backend/.sqlx/query-c9a8aca5b40f9393ccab04edd10cb507b5e8dc0cca2ddd424e09d784e3e22be5.json b/backend/.sqlx/query-c9a8aca5b40f9393ccab04edd10cb507b5e8dc0cca2ddd424e09d784e3e22be5.json new file mode 100644 index 0000000000..9b906031dc --- /dev/null +++ b/backend/.sqlx/query-c9a8aca5b40f9393ccab04edd10cb507b5e8dc0cca2ddd424e09d784e3e22be5.json @@ -0,0 +1,18 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO slack_thread_agent (slack_team_id, channel_id, thread_ts, workspace_id, agent_path)\n VALUES ($1, $2, $3, $4, $5) ON CONFLICT DO NOTHING", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + "Varchar", + "Varchar", + "Varchar", + "Varchar" + ] + }, + "nullable": [] + }, + "hash": "c9a8aca5b40f9393ccab04edd10cb507b5e8dc0cca2ddd424e09d784e3e22be5" +} diff --git a/backend/.sqlx/query-de3f99442e6f95013089f4c764b5981af2dd4c3ded4b1e167456e819fa0234eb.json b/backend/.sqlx/query-de3f99442e6f95013089f4c764b5981af2dd4c3ded4b1e167456e819fa0234eb.json new file mode 100644 index 0000000000..914294e195 --- /dev/null +++ b/backend/.sqlx/query-de3f99442e6f95013089f4c764b5981af2dd4c3ded4b1e167456e819fa0234eb.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT EXISTS(SELECT 1 FROM workspace_settings\n WHERE slack_team_id = $1 AND slack_command_script IS NOT NULL)", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "exists", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "de3f99442e6f95013089f4c764b5981af2dd4c3ded4b1e167456e819fa0234eb" +} diff --git a/backend/.sqlx/query-e9397e8f75e0108ab208de40d7c0fd674b457415389d5d4ef90472badc93dd35.json b/backend/.sqlx/query-e9397e8f75e0108ab208de40d7c0fd674b457415389d5d4ef90472badc93dd35.json new file mode 100644 index 0000000000..48592ca7e3 --- /dev/null +++ b/backend/.sqlx/query-e9397e8f75e0108ab208de40d7c0fd674b457415389d5d4ef90472badc93dd35.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT EXISTS(SELECT 1 FROM resource\n WHERE workspace_id = $1 AND path = $2 AND resource_type = 'ai_agent')", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "exists", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "e9397e8f75e0108ab208de40d7c0fd674b457415389d5d4ef90472badc93dd35" +} diff --git a/backend/.sqlx/query-f14b48a16ca770a062043a034a9d174c959438b5c6886935a2895879dbf72e3e.json b/backend/.sqlx/query-f14b48a16ca770a062043a034a9d174c959438b5c6886935a2895879dbf72e3e.json new file mode 100644 index 0000000000..4c11f9a737 --- /dev/null +++ b/backend/.sqlx/query-f14b48a16ca770a062043a034a9d174c959438b5c6886935a2895879dbf72e3e.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT COALESCE(slack_name, slack_team_id) FROM workspace_settings\n WHERE workspace_id = $1 AND slack_team_id IS NOT NULL", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "coalesce", + "type_info": "Varchar" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "f14b48a16ca770a062043a034a9d174c959438b5c6886935a2895879dbf72e3e" +} diff --git a/backend/.sqlx/query-f4726ac9311b2206088e5cfce32b47532839f91d2ecc462a5971f8de2194a9bf.json b/backend/.sqlx/query-f4726ac9311b2206088e5cfce32b47532839f91d2ecc462a5971f8de2194a9bf.json new file mode 100644 index 0000000000..e893140464 --- /dev/null +++ b/backend/.sqlx/query-f4726ac9311b2206088e5cfce32b47532839f91d2ecc462a5971f8de2194a9bf.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT slack_team_id FROM workspace_settings WHERE workspace_id = $1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "slack_team_id", + "type_info": "Varchar" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + true + ] + }, + "hash": "f4726ac9311b2206088e5cfce32b47532839f91d2ecc462a5971f8de2194a9bf" +} diff --git a/backend/.sqlx/query-f91c3e12549a3e037e8e784ad6ed439d762d17acb70793e9b26454572a156397.json b/backend/.sqlx/query-f91c3e12549a3e037e8e784ad6ed439d762d17acb70793e9b26454572a156397.json new file mode 100644 index 0000000000..30ec07404b --- /dev/null +++ b/backend/.sqlx/query-f91c3e12549a3e037e8e784ad6ed439d762d17acb70793e9b26454572a156397.json @@ -0,0 +1,28 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT workspace_id, slack_command_script IS NOT NULL AS \"has_script!\"\n FROM workspace_settings WHERE slack_team_id = $1\n ORDER BY slack_command_script IS NOT NULL DESC LIMIT 2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "workspace_id", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "has_script!", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + false, + null + ] + }, + "hash": "f91c3e12549a3e037e8e784ad6ed439d762d17acb70793e9b26454572a156397" +} diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 2de431e127..bc9011deb4 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -ab8476e5feedc98b727f7d83acffa1807b3b9c5b +c65ce0d02a906acf9c76dbe1c785e82a1cb65a46 diff --git a/backend/migrations/20261008145656_slack_agents.down.sql b/backend/migrations/20261008145656_slack_agents.down.sql new file mode 100644 index 0000000000..6009d448a6 --- /dev/null +++ b/backend/migrations/20261008145656_slack_agents.down.sql @@ -0,0 +1,2 @@ +DROP TABLE IF EXISTS slack_thread_agent; +DROP TABLE IF EXISTS slack_channel_agent; diff --git a/backend/migrations/20261008145656_slack_agents.up.sql b/backend/migrations/20261008145656_slack_agents.up.sql new file mode 100644 index 0000000000..f744d8dedf --- /dev/null +++ b/backend/migrations/20261008145656_slack_agents.up.sql @@ -0,0 +1,33 @@ +-- The agent that answers in a Slack channel when a message names none. Keyed by the Slack team +-- rather than the workspace: one channel has one default, whichever workspace set it. The agent +-- key cascades, so a renamed agent keeps its channels and a deleted one releases them. +CREATE TABLE slack_channel_agent ( + slack_team_id VARCHAR(255) NOT NULL, + channel_id VARCHAR(255) NOT NULL, + channel_name VARCHAR(255) NOT NULL, + workspace_id VARCHAR(50) NOT NULL, + agent_path VARCHAR(255) NOT NULL, + PRIMARY KEY (slack_team_id, channel_id), + FOREIGN KEY (workspace_id, agent_path) REFERENCES resource (workspace_id, path) + ON DELETE CASCADE ON UPDATE CASCADE +); + +CREATE INDEX index_slack_channel_agent_workspace ON slack_channel_agent (workspace_id, agent_path); + +-- The agent a Slack thread was started with, so a follow-up that names no agent goes to it. +CREATE TABLE slack_thread_agent ( + slack_team_id VARCHAR(255) NOT NULL, + channel_id VARCHAR(255) NOT NULL, + thread_ts VARCHAR(255) NOT NULL, + workspace_id VARCHAR(50) NOT NULL, + agent_path VARCHAR(255) NOT NULL, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (slack_team_id, channel_id, thread_ts), + FOREIGN KEY (workspace_id, agent_path) REFERENCES resource (workspace_id, path) + ON DELETE CASCADE ON UPDATE CASCADE +); + +GRANT ALL ON slack_channel_agent TO windmill_admin; +GRANT ALL ON slack_channel_agent TO windmill_user; +GRANT ALL ON slack_thread_agent TO windmill_admin; +GRANT ALL ON slack_thread_agent TO windmill_user; diff --git a/backend/summarized_schema.txt b/backend/summarized_schema.txt index 284a822b47..bf28c5f797 100644 --- a/backend/summarized_schema.txt +++ b/backend/summarized_schema.txt @@ -186,6 +186,10 @@ schedule: workspace_id(char), path(char), edited_by(char), edited_at(ts), schedu script: workspace_id(char), hash(bigint), path(char), parent_hashes(bigint[]), summary(text), description(text), content(text), created_by(char), created_at(ts), archived(bool), schema(json), deleted(bool), is_template(bool), extra_perms(jsonb), lock(text), lock_error_logs(text), language(script_lang), kind(script_kind), tag(char), draft_only(bool), envs(char), concurrent_limit(int), concurrency_time_window_s(int), cache_ttl(int), dedicated_worker(bool), ws_error_handler_muted(bool), priority(smallint), timeout(int), delete_after_use(bool), restart_unless_cancelled(bool), concurrency_key(char), visible_to_runner_only(bool), auto_kind(varchar), codebase(char), has_preprocessor(bool), schema_validation(bool), assets(jsonb), debounce_key(char), debounce_delay_s(int), cache_ignore_s3_path(bool), runnable_settings_handle(bigint), labels(text[]), on_behalf_of(varchar), on_behalf_of_email(text), job_token_scopes(text[]) FK: (workspace_id) -> workspace(id) skip_workspace_diff_tally: workspace_id(char), added_at(ts) +slack_channel_agent: slack_team_id(char), channel_id(char), channel_name(char), workspace_id(char), agent_path(char) + FK: (workspace_id, agent_path) -> resource(workspace_id, path) +slack_thread_agent: slack_team_id(char), channel_id(char), thread_ts(char), workspace_id(char), agent_path(char), created_at(timestamptz) + FK: (workspace_id, agent_path) -> resource(workspace_id, path) sqs_trigger: path(char), queue_url(char), aws_resource_path(char), message_attributes(text[]), script_path(char), is_flow(bool), workspace_id(char), edited_by(char), email(char), edited_at(ts), extra_perms(jsonb), error(text), server_id(char), last_server_ping(ts), aws_auth_resource_type(aws_auth_resource_type), error_handler_path(char), error_handler_args(jsonb), retry(jsonb), mode(trigger_mode), labels(text[]) FK: (workspace_id) -> workspace(id) trashbin: id(bigint), workspace_id(char), item_kind(char), item_path(char), item_data(jsonb), deleted_by(char), deleted_at(ts), expires_at(ts) diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 5e635e39bd..08da137ed6 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -13351,6 +13351,130 @@ paths: items: $ref: "#/components/schemas/FlowConversation" + /w/{workspace}/slack_agents/channels: + get: + summary: list the Slack channels whose default agent this workspace set + operationId: listSlackChannelAgents + tags: + - slack_agents + parameters: + - $ref: "#/components/parameters/WorkspaceId" + responses: + "200": + description: channels and their default agent + content: + application/json: + schema: + type: array + items: + $ref: "#/components/schemas/SlackChannelAgent" + post: + summary: make an agent the default in a Slack channel (admin only) + operationId: setSlackChannelAgent + tags: + - slack_agents + parameters: + - $ref: "#/components/parameters/WorkspaceId" + requestBody: + required: true + content: + application/json: + schema: + $ref: "#/components/schemas/SlackChannelAgent" + responses: + "200": + description: default agent set + content: + text/plain: + schema: + type: string + + /w/{workspace}/slack_agents/channels/{channel_id}: + delete: + summary: remove a Slack channel's default agent (admin only) + operationId: removeSlackChannelAgent + tags: + - slack_agents + parameters: + - $ref: "#/components/parameters/WorkspaceId" + - name: channel_id + in: path + required: true + schema: + type: string + responses: + "200": + description: default agent removed + content: + text/plain: + schema: + type: string + + /w/{workspace}/slack_agents/available_channels: + get: + summary: list the connected Slack workspace's public channels (admin only) + operationId: listAvailableSlackChannels + tags: + - slack_agents + parameters: + - $ref: "#/components/parameters/WorkspaceId" + - name: cursor + description: the next_cursor of the previous page + in: query + schema: + type: string + responses: + "200": + description: a page of public channels + content: + application/json: + schema: + type: object + properties: + channels: + type: array + items: + type: object + properties: + id: + type: string + name: + type: string + required: + - id + - name + next_cursor: + type: string + required: + - channels + + /w/{workspace}/slack_agents/agent_channels/{path}: + get: + summary: the Slack channels where an agent is the default, and whether Slack is connected + operationId: getAgentSlackChannels + tags: + - slack_agents + parameters: + - $ref: "#/components/parameters/WorkspaceId" + - $ref: "#/components/parameters/Path" + responses: + "200": + description: the agent's Slack channels + content: + application/json: + schema: + type: object + properties: + slack_team_name: + type: string + description: the connected Slack workspace; absent when Slack is not connected + channels: + type: array + items: + $ref: "#/components/schemas/SlackChannelAgent" + required: + - channels + /w/{workspace}/flow_conversations/update/{conversation_id}: post: summary: rename flow conversation @@ -29355,6 +29479,20 @@ components: definition_changed: type: boolean + SlackChannelAgent: + type: object + properties: + channel_id: + type: string + channel_name: + type: string + agent_path: + type: string + required: + - channel_id + - channel_name + - agent_path + FlowConversation: type: object required: diff --git a/backend/windmill-api/src/agent_runs.rs b/backend/windmill-api/src/agent_runs.rs index 76f729e7e0..9f52b90d53 100644 --- a/backend/windmill-api/src/agent_runs.rs +++ b/backend/windmill-api/src/agent_runs.rs @@ -48,8 +48,22 @@ pub(crate) async fn run_agent( let path = path.to_path(); check_scopes(&authed, || format!("jobs:run:agents:{path}"))?; + let uuid = run_agent_as(&authed, &db, &user_db, &w_id, path, &run_query, &args).await?; + Ok((StatusCode::CREATED, uuid.to_string())) +} + +/// [`run_agent`] for a caller already authenticated, such as a Slack user matched to a member. +pub(crate) async fn run_agent_as( + authed: &ApiAuthed, + db: &DB, + user_db: &UserDB, + w_id: &str, + path: &str, + run_query: &RunJobQuery, + args: &HashMap>, +) -> Result { // 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 mut tx = user_db.clone().begin(authed).await?; let value = sqlx::query_scalar!( "SELECT value AS \"value: sqlx::types::Json\" FROM resource WHERE workspace_id = $1 AND path = $2 AND resource_type = 'ai_agent'", @@ -84,17 +98,17 @@ pub(crate) async fn run_agent( )?; let push_authed = authed.clone().into(); - let scope_ceiling = windmill_api_auth::caller_scope_ceiling(&db, &authed).await?; + let scope_ceiling = windmill_api_auth::caller_scope_ceiling(db, authed).await?; let (uuid, mut tx) = push( - &db, + db, PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into()), - &w_id, + w_id, JobPayload::RawFlow { value: flow_value, path: Some(run_path.clone()), restarted_from: None, }, - PushArgs::from(&args), + PushArgs::from(args), authed.display_username(), &authed.email, username_to_permissioned_as(&authed.username), @@ -123,27 +137,27 @@ pub(crate) async fn run_agent( ) .await?; - if let Some(memory_id) = run_query.memory_key(&w_id, &run_path) { + if let Some(memory_id) = run_query.memory_key(w_id, &run_path) { set_flow_memory_id(&mut tx, uuid, memory_id).await?; } if chat { // A real conversation, not a test one: this is the agent as deployed, not a draft. handle_chat_conversation_messages( &mut tx, - &authed, - &w_id, + authed, + w_id, &run_path, - &run_query, + run_query, args.get("user_message"), uuid, false, - &args, + args, ) .await?; } tx.commit().await?; - Ok((StatusCode::CREATED, uuid.to_string())) + Ok(uuid) } /// The agent as a one-step flow, validated by deserializing through `FlowValue` rather than diff --git a/backend/windmill-api/src/lib.rs b/backend/windmill-api/src/lib.rs index 1cd2b4101c..bde592fba4 100644 --- a/backend/windmill-api/src/lib.rs +++ b/backend/windmill-api/src/lib.rs @@ -161,6 +161,9 @@ mod scim_oss; mod scripts; mod secret_backend_ext; mod service_logs; +// Answering in Slack is only reached from the EE Slack event handlers. +#[cfg_attr(not(all(feature = "oauth2", feature = "private")), allow(dead_code))] +mod slack_agents; mod slack_approvals; #[cfg(all(feature = "smtp", feature = "private"))] pub mod smtp_server_ee; @@ -251,7 +254,7 @@ lazy_static::lazy_static! { // Must be a subset of the bot scopes declared in the Slack app manifest. Default matches // Windmill's recommended manifest at docs.windmill.dev/docs/misc/setup_oauth. pub static ref SLACK_OAUTH_SCOPES: String = std::env::var("SLACK_OAUTH_SCOPES") - .unwrap_or_else(|_| "commands,chat:write,chat:write.public,channels:join,files:write,app_mentions:read,im:history,im:read".to_string()); + .unwrap_or_else(|_| "commands,chat:write,chat:write.public,channels:join,files:write,app_mentions:read,im:history,im:read,users:read,users:read.email,channels:read".to_string()); } // Compliance with cloud events spec. @@ -678,6 +681,7 @@ pub async fn run_server( .nest("/embeddings", embeddings::workspaced_service()) .nest("/favorites", favorite::workspaced_service()) .nest("/flows", flows::workspaced_service()) + .nest("/slack_agents", slack_agents::workspaced_service()) .nest("/runnables", runnables::workspaced_service()) .nest( "/workspace_dependencies", diff --git a/backend/windmill-api/src/slack_agents.rs b/backend/windmill-api/src/slack_agents.rs new file mode 100644 index 0000000000..2f2775629a --- /dev/null +++ b/backend/windmill-api/src/slack_agents.rs @@ -0,0 +1,808 @@ +//! AI agents answering in Slack. +//! +//! A message reaches an agent when it names one (`+name` or `~name`), when its thread was started +//! with one, or when its channel has a default agent. Anything else goes to the workspace's +//! command script as before, or, where there is none, is answered with how to reach an agent. +//! The agent runs as the workspace member whose email the Slack user +//! has, so a Slack user who is not a member gets nothing, and what the agent may read or call is +//! that member's. One Slack thread is one conversation per person: a run's tool results are its +//! runner's to see, and a shared conversation would hand them to whoever replies next. + +use std::collections::HashMap; + +use axum::{ + extract::{Path, Query}, + routing::{delete, get}, + Extension, Json, Router, +}; +use serde::{Deserialize, Serialize}; +use serde_json::value::RawValue; +use uuid::Uuid; +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, + BASE_URL, +}; + +use crate::{ + agent_runs::run_agent_as, + db::{ApiAuthed, DB}, + jobs::RunJobQuery, + users::fetch_api_authed, + HTTP_CLIENT, +}; + +/// How long the reply waits for the agent before pointing at the run instead. +const MAX_WAIT_SECS: u64 = 3600; +/// Under the 12,000 characters Slack takes in a markdown block. +const MAX_REPLY_CHARS: usize = 11_000; + +pub fn workspaced_service() -> Router { + Router::new() + .route( + "/channels", + get(list_channel_agents).post(set_channel_agent), + ) + .route("/channels/{channel_id}", delete(remove_channel_agent)) + .route("/available_channels", get(list_available_channels)) + .route("/agent_channels/{*path}", get(list_agent_channels)) +} + +#[derive(Serialize)] +struct ChannelAgent { + channel_id: String, + channel_name: String, + agent_path: String, +} + +async fn list_channel_agents( + _authed: ApiAuthed, + Extension(db): Extension, + Path(w_id): Path, +) -> JsonResult> { + 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", + w_id + ) + .fetch_all(&db) + .await?; + Ok(Json(rows)) +} + +#[derive(Serialize)] +struct AgentSlack { + /// The connected Slack workspace's name; `None` when Slack is not connected. + slack_team_name: Option, + channels: Vec, +} + +async fn list_agent_channels( + _authed: ApiAuthed, + Extension(db): Extension, + Path((w_id, path)): Path<(String, StripPath)>, +) -> JsonResult { + let slack_team_name = sqlx::query_scalar!( + "SELECT COALESCE(slack_name, slack_team_id) FROM workspace_settings + WHERE workspace_id = $1 AND slack_team_id IS NOT NULL", + w_id + ) + .fetch_optional(&db) + .await? + .flatten(); + 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", + w_id, + path.to_path() + ) + .fetch_all(&db) + .await?; + Ok(Json(AgentSlack { slack_team_name, channels })) +} + +#[derive(Deserialize)] +struct SetChannelAgent { + channel_id: String, + channel_name: String, + agent_path: String, +} + +async fn set_channel_agent( + authed: ApiAuthed, + Extension(db): Extension, + Path(w_id): Path, + Json(body): Json, +) -> Result { + require_admin(authed.is_admin, &authed.username)?; + let team_id = connected_team(&db, &w_id).await?; + + let is_agent = sqlx::query_scalar!( + "SELECT EXISTS(SELECT 1 FROM resource + WHERE workspace_id = $1 AND path = $2 AND resource_type = 'ai_agent')", + w_id, + body.agent_path + ) + .fetch_one(&db) + .await? + .unwrap_or(false); + if !is_agent { + return Err(Error::BadRequest(format!( + "{} is not an agent", + body.agent_path + ))); + } + + // A channel belongs to the Slack team, which several workspaces may share: one workspace + // must not take over a channel another one set. + let taken = sqlx::query!( + "INSERT INTO slack_channel_agent + (slack_team_id, channel_id, channel_name, workspace_id, agent_path) + VALUES ($1, $2, $3, $4, $5) + ON CONFLICT (slack_team_id, channel_id) DO UPDATE + SET channel_name = EXCLUDED.channel_name, agent_path = EXCLUDED.agent_path + WHERE slack_channel_agent.workspace_id = EXCLUDED.workspace_id + RETURNING channel_id", + team_id, + body.channel_id, + body.channel_name, + w_id, + body.agent_path + ) + .fetch_optional(&db) + .await? + .is_none(); + if taken { + return Err(Error::BadRequest(format!( + "#{} already has a default agent set from another workspace", + body.channel_name + ))); + } + Ok(format!( + "{} answers in #{}", + body.agent_path, body.channel_name + )) +} + +async fn remove_channel_agent( + authed: ApiAuthed, + Extension(db): Extension, + Path((w_id, channel_id)): Path<(String, String)>, +) -> Result { + require_admin(authed.is_admin, &authed.username)?; + sqlx::query!( + "DELETE FROM slack_channel_agent WHERE workspace_id = $1 AND channel_id = $2", + w_id, + channel_id + ) + .execute(&db) + .await?; + Ok("Removed".to_string()) +} + +#[derive(Serialize)] +struct SlackChannel { + id: String, + name: String, +} + +#[derive(Deserialize)] +struct ChannelsQuery { + cursor: Option, +} + +#[derive(Serialize)] +struct SlackChannels { + channels: Vec, + next_cursor: Option, +} + +/// The team's public channels, a page at a time, for picking one. +async fn list_available_channels( + authed: ApiAuthed, + Extension(db): Extension, + Path(w_id): Path, + Query(q): Query, +) -> JsonResult { + require_admin(authed.is_admin, &authed.username)?; + let token = bot_token(&db, &w_id).await?; + let mut form = vec![ + ("types", "public_channel".to_string()), + ("exclude_archived", "true".to_string()), + ("limit", "200".to_string()), + ]; + if let Some(cursor) = q.cursor.filter(|c| !c.is_empty()) { + form.push(("cursor", cursor)); + } + let res = slack_call(&token, "conversations.list", &form) + .await + .map_err(|e| Error::BadRequest(e.to_string()))?; + let channels = res["channels"] + .as_array() + .map(|cs| { + cs.iter() + .filter_map(|c| { + Some(SlackChannel { + id: c["id"].as_str()?.to_string(), + name: c["name"].as_str()?.to_string(), + }) + }) + .collect() + }) + .unwrap_or_default(); + let next_cursor = res["response_metadata"]["next_cursor"] + .as_str() + .filter(|c| !c.is_empty()) + .map(str::to_string); + Ok(Json(SlackChannels { channels, next_cursor })) +} + +async fn connected_team(db: &DB, w_id: &str) -> Result { + sqlx::query_scalar!( + "SELECT slack_team_id FROM workspace_settings WHERE workspace_id = $1", + w_id + ) + .fetch_optional(db) + .await? + .flatten() + .ok_or_else(|| Error::BadRequest("Slack is not connected to this workspace".to_string())) +} + +async fn bot_token(db: &DB, w_id: &str) -> Result { + get_secret_value_as_admin(db, w_id, WORKSPACE_SLACK_BOT_TOKEN_PATH).await +} + +/// A Slack Web API call, form-encoded since read methods such as `users.info` take no JSON. +/// Slack answers 200 with `ok: false` on failure; that is returned as the error. +async fn slack_call( + token: &str, + method: &str, + form: &[(&str, String)], +) -> anyhow::Result { + let res: serde_json::Value = HTTP_CLIENT + .post(format!("https://slack.com/api/{method}")) + .bearer_auth(token) + .form(form) + .send() + .await? + .json() + .await?; + if res["ok"].as_bool() != Some(true) { + let error = res["error"].as_str().unwrap_or("unknown_error"); + let needed = res["needed"].as_str().unwrap_or_default(); + anyhow::bail!("{method}: {error} {needed}"); + } + Ok(res) +} + +/// A message addressed to Windmill, from a mention, a direct message or `/windmill`. +pub struct SlackMessage { + pub team_id: String, + pub channel_id: String, + pub user_id: String, + /// With the bot mention stripped. + pub text: String, + /// The message's own ts; `None` for a slash command, which has no message to reply under. + pub ts: Option, + pub thread_ts: Option, + /// Slack resent the event because the first delivery was not acknowledged in time. + pub is_retry: bool, +} + +enum Route { + Named(String), + Agent(String), + /// Nothing answers here: say how to reach an agent. + Help, +} + +/// 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 { + 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()); + + let (w_id, route, source) = if let Some(name) = name { + let Some(w_id) = workspace_for_team(db, &msg.team_id).await? else { + return Ok(false); + }; + (w_id, Route::Named(name.to_string()), "mention") + } else if let Some(r) = match &thread_root { + Some(root) => sqlx::query!( + "SELECT workspace_id, agent_path FROM slack_thread_agent + WHERE slack_team_id = $1 AND channel_id = $2 AND thread_ts = $3", + msg.team_id, + msg.channel_id, + root + ) + .fetch_optional(db) + .await? + .map(|r| (r.workspace_id, r.agent_path)), + None => None, + } { + (r.0, Route::Agent(r.1), "thread") + } else if let Some(r) = sqlx::query!( + "SELECT workspace_id, agent_path FROM slack_channel_agent + WHERE slack_team_id = $1 AND channel_id = $2", + msg.team_id, + msg.channel_id + ) + .fetch_optional(db) + .await? + { + ( + r.workspace_id, + Route::Agent(r.agent_path), + "channel_default", + ) + } else if let Some(w_id) = workspace_without_command_script(db, &msg.team_id).await? { + (w_id, Route::Help, "help") + } else { + return Ok(false); + }; + + // The first delivery is already being answered. + if msg.is_retry { + return Ok(true); + } + + if !matches!(route, Route::Help) { + windmill_common::feature_usage::log_feature_usage("slack_agent", "invoked", source); + } + let db = db.clone(); + let question = question.to_string(); + // 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 { + if let Err(e) = answer(&db, &w_id, route, &question, msg).await { + tracing::error!("Slack agent answer failed: {e:#}"); + } + }); + Ok(true) +} + +/// `+name rest` or `~name rest` → (name, rest). +fn split_agent_name(text: &str) -> (Option<&str>, &str) { + let text = text.trim(); + let Some(after) = text.strip_prefix('+').or_else(|| text.strip_prefix('~')) else { + return (None, text); + }; + let (name, rest) = after.split_once(char::is_whitespace).unwrap_or((after, "")); + if name.is_empty() + || !name + .chars() + .all(|c| c.is_alphanumeric() || matches!(c, '_' | '-' | '/' | '.')) + { + return (None, text); + } + (Some(name), rest.trim()) +} + +/// Slack escapes these three in message text. +fn decode_entities(text: &str) -> String { + text.replace("<", "<") + .replace(">", ">") + .replace("&", "&") +} + +/// The workspace a message naming an agent goes to when neither its thread nor its channel says: +/// the one whose command script the team's messages already reach, else the team's only one. +async fn workspace_for_team(db: &DB, team_id: &str) -> Result> { + let rows = sqlx::query!( + "SELECT workspace_id, slack_command_script IS NOT NULL AS \"has_script!\" + FROM workspace_settings WHERE slack_team_id = $1 + ORDER BY slack_command_script IS NOT NULL DESC LIMIT 2", + team_id + ) + .fetch_all(db) + .await?; + Ok(match rows.as_slice() { + [only] => Some(only.workspace_id.clone()), + [first, ..] if first.has_script => Some(first.workspace_id.clone()), + _ => None, + }) +} + +/// The team's workspace when no workspace of the team has a command script to take the message. +async fn workspace_without_command_script(db: &DB, team_id: &str) -> Result> { + let has_script = sqlx::query_scalar!( + "SELECT EXISTS(SELECT 1 FROM workspace_settings + WHERE slack_team_id = $1 AND slack_command_script IS NOT NULL)", + team_id + ) + .fetch_one(db) + .await? + .unwrap_or(false); + if has_script { + return Ok(None); + } + workspace_for_team(db, team_id).await +} + +/// How to reach an agent, with the ones this member can use. +async fn help_text(user_db: &UserDB, authed: &ApiAuthed, w_id: &str) -> Result { + let mut tx = user_db.clone().begin(authed).await?; + let agents = sqlx::query_scalar!( + "SELECT path FROM resource WHERE workspace_id = $1 AND resource_type = 'ai_agent' + ORDER BY path LIMIT 11", + w_id + ) + .fetch_all(&mut *tx) + .await?; + tx.commit().await?; + let Some(first) = agents.first() else { + return Ok(format!( + "No agent answers here by default, and there is no agent in the Windmill workspace *{w_id}* you can use yet." + )); + }; + let handle = |p: &str| p.rsplit('/').next().unwrap_or(p).to_string(); + let mut list = agents + .iter() + .take(10) + .map(|p| format!("• `+{}` ({p})", handle(p))) + .collect::>() + .join("\n"); + if agents.len() > 10 { + list.push_str("\n• …"); + } + Ok(format!( + "No agent answers here by default. Start your message with an agent's name, for example `+{} what can you do?`\n\nAgents you can use:\n{list}\n\nA workspace admin can make an agent the default for a channel from the agent's page in Windmill.", + handle(first) + )) +} + +async fn answer( + db: &DB, + w_id: &str, + route: Route, + question: &str, + msg: SlackMessage, +) -> anyhow::Result<()> { + let token = bot_token(db, w_id).await?; + let channel = msg.channel_id.clone(); + + // A slash command leaves no message to thread under, so the question is posted first. + let thread_root = match msg.thread_ts.clone().or_else(|| msg.ts.clone()) { + Some(ts) => ts, + None => { + let posted = slack_call( + &token, + "chat.postMessage", + &[ + ("channel", channel.clone()), + ("text", format!("<@{}>: {}", msg.user_id, question)), + ], + ) + .await?; + posted["ts"].as_str().unwrap_or_default().to_string() + } + }; + let reply = Reply::start(&token, &channel, &thread_root).await?; + + let authed = match member_for_slack_user(db, &token, w_id, &msg.user_id).await { + Ok(Some(authed)) => authed, + Ok(None) => { + return reply.send( + format!( + "Your Slack email isn't a member of the Windmill workspace *{w_id}*. Ask an admin to invite you." + ), + None, + ) + .await; + } + Err(e) if e.to_string().contains("missing_scope") => { + return reply.send( + "Windmill can't read your Slack email. A workspace admin needs to reconnect Slack in the workspace settings to grant it.".to_string(), + None, + ) + .await; + } + Err(e) => return Err(e), + }; + let user_db = UserDB::new(db.clone()); + + let agent_path = match route { + Route::Help => { + return reply + .send(help_text(&user_db, &authed, w_id).await?, None) + .await; + } + Route::Agent(path) => path, + Route::Named(name) => { + let mut tx = user_db.clone().begin(&authed).await?; + let matches = sqlx::query_scalar!( + "SELECT path FROM resource + WHERE workspace_id = $1 AND resource_type = 'ai_agent' + AND (path = $2 OR right(path, length($2) + 1) = '/' || $2) + ORDER BY path LIMIT 6", + w_id, + name + ) + .fetch_all(&mut *tx) + .await?; + tx.commit().await?; + match matches.as_slice() { + [only] => only.clone(), + [] => { + return reply + .send( + format!("I can't find an agent named *{name}* that you can use."), + None, + ) + .await; + } + several => { + let list = several + .iter() + .map(|p| format!("`+{p}`")) + .collect::>() + .join(", "); + return reply + .send( + format!("Several agents are named *{name}*, pick one: {list}"), + None, + ) + .await; + } + } + } + }; + + if question.is_empty() { + return reply + .send(format!("What would you like to ask `{agent_path}`?"), None) + .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?; + + let run_query = RunJobQuery { + memory_id: Some(format!( + "slack:{}:{}:{}:{}", + msg.team_id, channel, thread_root, msg.user_id + )), + ..Default::default() + }; + let args: HashMap> = [( + "user_message".to_string(), + windmill_common::worker::to_raw_value(&question), + )] + .into(); + let job_id = + match run_agent_as(&authed, db, &user_db, w_id, &agent_path, &run_query, &args).await { + Ok(id) => id, + Err(e) => { + return reply + .send(format!("`{agent_path}` couldn't start: {e}"), None) + .await; + } + }; + + let run_link = format!( + "<{}/run/{job_id}?workspace={w_id}|View run> · `{agent_path}`", + BASE_URL.load() + ); + let thinking = reply.keep_thinking(); + let result = wait_for_result(db, w_id, job_id).await; + if let Some(thinking) = thinking { + thinking.abort(); + } + let text = match result? { + Some((result, true)) => answer_text(&result), + Some((result, false)) => format!("The agent failed: {}", error_text(&result)), + None => "Still working, follow it in the run.".to_string(), + }; + reply.send(text, Some(run_link)).await +} + +async fn member_for_slack_user( + db: &DB, + token: &str, + w_id: &str, + slack_user: &str, +) -> anyhow::Result> { + let info = slack_call(token, "users.info", &[("user", slack_user.to_string())]).await?; + let Some(email) = info["user"]["profile"]["email"].as_str() else { + return Ok(None); + }; + let member = sqlx::query!( + "SELECT username, email FROM usr + WHERE workspace_id = $1 AND lower(email) = lower($2) AND NOT disabled", + w_id, + email + ) + .fetch_optional(db) + .await?; + let Some(member) = member else { + return Ok(None); + }; + Ok(Some( + fetch_api_authed(member.username, member.email, w_id, db, None).await?, + )) +} + +async fn wait_for_result( + db: &DB, + w_id: &str, + job_id: Uuid, +) -> Result> { + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(MAX_WAIT_SECS); + while std::time::Instant::now() < deadline { + let row = sqlx::query!( + "SELECT result AS \"result: sqlx::types::Json\", + status = 'success' AS \"success!\" + FROM v2_job_completed WHERE id = $1 AND workspace_id = $2", + job_id, + w_id + ) + .fetch_optional(db) + .await?; + if let Some(row) = row { + return Ok(Some(( + row.result.map(|r| r.0).unwrap_or_default(), + row.success, + ))); + } + tokio::time::sleep(std::time::Duration::from_secs(2)).await; + } + Ok(None) +} + +/// The agent step's `output`: its text, or a structured output as a JSON block. +fn answer_text(result: &serde_json::Value) -> String { + match &result["output"] { + serde_json::Value::String(s) if !s.trim().is_empty() => s.clone(), + serde_json::Value::Null => "The agent returned no answer.".to_string(), + other => format!( + "```\n{}\n```", + serde_json::to_string_pretty(other).unwrap_or_default() + ), + } +} + +fn error_text(result: &serde_json::Value) -> String { + result["error"]["message"] + .as_str() + .map(str::to_string) + .unwrap_or_else(|| result.to_string()) +} + +/// The bot's reply in a thread. While it is being worked out the thread shows the bot as thinking, +/// or, where Slack refuses that status, a placeholder message the reply then replaces. +#[derive(Clone)] +struct Reply { + token: String, + channel: String, + thread_ts: String, + placeholder: Option, +} + +impl Reply { + async fn start(token: &str, channel: &str, thread_ts: &str) -> anyhow::Result { + let mut reply = Reply { + token: token.to_string(), + channel: channel.to_string(), + thread_ts: thread_ts.to_string(), + placeholder: None, + }; + if let Err(e) = reply.set_status("is thinking…").await { + tracing::debug!("Slack refused the thinking status, posting a placeholder: {e:#}"); + let posted = slack_call( + token, + "chat.postMessage", + &[ + ("channel", channel.to_string()), + ("thread_ts", thread_ts.to_string()), + ("text", "Thinking…".to_string()), + ], + ) + .await?; + reply.placeholder = posted["ts"].as_str().map(str::to_string); + } + Ok(reply) + } + + async fn set_status(&self, status: &str) -> anyhow::Result<()> { + slack_call( + &self.token, + "assistant.threads.setStatus", + &[ + ("channel_id", self.channel.clone()), + ("thread_ts", self.thread_ts.clone()), + ("status", status.to_string()), + ], + ) + .await + .map(|_| ()) + } + + /// Slack drops the status after two minutes without a reply, so a long run renews it. + fn keep_thinking(&self) -> Option> { + if self.placeholder.is_some() { + return None; + } + let reply = self.clone(); + Some(tokio::spawn(async move { + loop { + tokio::time::sleep(std::time::Duration::from_secs(90)).await; + let _ = reply.set_status("is thinking…").await; + } + })) + } + + async fn send(&self, text: String, footer: Option) -> anyhow::Result<()> { + let (text, blocks) = reply_blocks(&text, footer)?; + let mut form = vec![ + ("channel", self.channel.clone()), + ("text", text), + ("blocks", blocks), + ]; + match &self.placeholder { + Some(ts) => { + form.push(("ts", ts.clone())); + slack_call(&self.token, "chat.update", &form).await?; + } + None => { + form.push(("thread_ts", self.thread_ts.clone())); + slack_call(&self.token, "chat.postMessage", &form).await?; + let _ = self.set_status("").await; + } + } + Ok(()) + } +} + +/// The reply's notification text and its blocks: the answer as markdown, and an optional footer. +fn reply_blocks(text: &str, footer: Option) -> anyhow::Result<(String, String)> { + let text = if text.chars().count() > MAX_REPLY_CHARS { + let cut: String = text.chars().take(MAX_REPLY_CHARS).collect(); + format!("{cut}\n\n_The answer is cut short here; the run has all of it._") + } else { + text.to_string() + }; + let mut blocks = vec![serde_json::json!({ "type": "markdown", "text": text })]; + if let Some(footer) = footer { + blocks.push(serde_json::json!({ + "type": "context", + "elements": [{ "type": "mrkdwn", "text": footer }] + })); + } + let fallback: String = text.chars().take(300).collect(); + Ok((fallback, serde_json::to_string(&blocks)?)) +} + +#[cfg(test)] +mod tests { + use super::split_agent_name; + + #[test] + fn agent_name_prefix() { + assert_eq!( + split_agent_name("+sql_analyst how many?"), + (Some("sql_analyst"), "how many?") + ); + assert_eq!( + split_agent_name("~f/data/sql_analyst hi"), + (Some("f/data/sql_analyst"), "hi") + ); + assert_eq!(split_agent_name("+helper"), (Some("helper"), "")); + assert_eq!(split_agent_name("what is +x"), (None, "what is +x")); + assert_eq!(split_agent_name("+ spaced"), (None, "+ spaced")); + } +} diff --git a/frontend/src/lib/components/flows/content/AgentUsePanel.svelte b/frontend/src/lib/components/flows/content/AgentUsePanel.svelte new file mode 100644 index 0000000000..4dfad01c5a --- /dev/null +++ b/frontend/src/lib/components/flows/content/AgentUsePanel.svelte @@ -0,0 +1,249 @@ + + +
+ + + + + {#snippet content()} +
+ + {#if !slackTeamName} + + Connect Slack to the workspace to ask this agent from Slack, by its name or as a + channel's default. + + {#if isAdmin} +
+ +
+ {:else} + A workspace admin can connect it. + {/if} + {:else} +
+ Ask it anywhere + + In {slackTeamName}, mention the Windmill bot and start your message with the agent's + name, in a channel or in a direct message. Follow-ups in the same thread go to the + same agent. + +
+ {handle} your question + +
+
+ +
+ Default in these channels + + Mentions of the Windmill bot in these channels go to this agent without its name. + + {#if channels.length === 0} + No channel yet. + {:else} +
+ {#each channels as channel (channel.channel_id)} +
+ #{channel.channel_name} + {#if isAdmin} +
+ {/each} +
+ {/if} + {#if isAdmin} +
+