mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-10-13 08:02:47 +00:00
feat: agents run on behalf of their deployer through a run-by-path endpoint
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5.5
parent
95c4b9ee77
commit
c6ece17452
+23
@@ -0,0 +1,23 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "SELECT resource_type = 'ai_agent' FROM resource WHERE workspace_id = $1 AND path = $2",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
"ordinal": 0,
|
||||
"name": "?column?",
|
||||
"type_info": "Bool"
|
||||
}
|
||||
],
|
||||
"parameters": {
|
||||
"Left": [
|
||||
"Text",
|
||||
"Text"
|
||||
]
|
||||
},
|
||||
"nullable": [
|
||||
null
|
||||
]
|
||||
},
|
||||
"hash": "1564e75bd7888dc54baec981bf9d855df12d32b0102c38b15e5f81162a00a66b"
|
||||
}
|
||||
+9
-3
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "SELECT resource.workspace_id, resource.path, resource.value, resource.description,\n resource.resource_type, resource.extra_perms, resource.created_by, resource.edited_at,\n resource.labels,\n folder_labels(resource.workspace_id, resource.path) as \"inherited_labels?\",\n (now() > account.expires_at) as is_expired, account.refresh_token != '' as is_refreshed,\n account.refresh_error,\n variable.path IS NOT NULL as is_linked,\n variable.is_oauth as \"is_oauth?\",\n variable.account,\n ws_specific.path IS NOT NULL as ws_specific,\n null::bool as draft_only,\n null::bool as is_draft\n FROM resource\n LEFT JOIN variable ON variable.path = resource.path AND variable.workspace_id = $2\n LEFT JOIN account ON variable.account = account.id AND account.workspace_id = $2\n LEFT JOIN ws_specific ON ws_specific.path = resource.path AND ws_specific.workspace_id = $2 AND ws_specific.item_kind = 'resource'\n WHERE resource.path = $1 AND resource.workspace_id = $2",
|
||||
"query": "SELECT resource.workspace_id, resource.path, resource.value, resource.description,\n resource.resource_type, resource.extra_perms, resource.created_by, resource.edited_at,\n resource.labels,\n folder_labels(resource.workspace_id, resource.path) as \"inherited_labels?\",\n (now() > account.expires_at) as is_expired, account.refresh_token != '' as is_refreshed,\n account.refresh_error,\n variable.path IS NOT NULL as is_linked,\n variable.is_oauth as \"is_oauth?\",\n variable.account,\n ws_specific.path IS NOT NULL as ws_specific,\n null::bool as draft_only,\n null::bool as is_draft,\n resource.on_behalf_of\n FROM resource\n LEFT JOIN variable ON variable.path = resource.path AND variable.workspace_id = $2\n LEFT JOIN account ON variable.account = account.id AND account.workspace_id = $2\n LEFT JOIN ws_specific ON ws_specific.path = resource.path AND ws_specific.workspace_id = $2 AND ws_specific.item_kind = 'resource'\n WHERE resource.path = $1 AND resource.workspace_id = $2",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
@@ -97,6 +97,11 @@
|
||||
"ordinal": 18,
|
||||
"name": "is_draft",
|
||||
"type_info": "Bool"
|
||||
},
|
||||
{
|
||||
"ordinal": 19,
|
||||
"name": "on_behalf_of",
|
||||
"type_info": "Varchar"
|
||||
}
|
||||
],
|
||||
"parameters": {
|
||||
@@ -124,8 +129,9 @@
|
||||
true,
|
||||
null,
|
||||
null,
|
||||
null
|
||||
null,
|
||||
true
|
||||
]
|
||||
},
|
||||
"hash": "9b88afcbecd66e5e29658463a2b8c2aa44d714bcc95e242ef329f6fd71c345c1"
|
||||
"hash": "1d5c68d62b35386ada241fe84a511e33926d151d1a916dfa6c0959217b0232b3"
|
||||
}
|
||||
+29
@@ -0,0 +1,29 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "SELECT value AS \"value: sqlx::types::Json<serde_json::Value>\", on_behalf_of\n FROM resource WHERE workspace_id = $1 AND path = $2 AND resource_type = 'ai_agent'",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
"ordinal": 0,
|
||||
"name": "value: sqlx::types::Json<serde_json::Value>",
|
||||
"type_info": "Jsonb"
|
||||
},
|
||||
{
|
||||
"ordinal": 1,
|
||||
"name": "on_behalf_of",
|
||||
"type_info": "Varchar"
|
||||
}
|
||||
],
|
||||
"parameters": {
|
||||
"Left": [
|
||||
"Text",
|
||||
"Text"
|
||||
]
|
||||
},
|
||||
"nullable": [
|
||||
true,
|
||||
true
|
||||
]
|
||||
},
|
||||
"hash": "38a233bd5816deb70773796a682a5bceb415da87b69296f9ac50679392274f2d"
|
||||
}
|
||||
+16
@@ -0,0 +1,16 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "UPDATE resource SET on_behalf_of = $1 WHERE workspace_id = $2 AND path = $3",
|
||||
"describe": {
|
||||
"columns": [],
|
||||
"parameters": {
|
||||
"Left": [
|
||||
"Varchar",
|
||||
"Text",
|
||||
"Text"
|
||||
]
|
||||
},
|
||||
"nullable": []
|
||||
},
|
||||
"hash": "c6c5e40893aa766cd55af5d65d5a19ac213f3a99656e6efea94d13c9af4d904a"
|
||||
}
|
||||
@@ -0,0 +1,3 @@
|
||||
DROP TRIGGER IF EXISTS reset_agent_on_behalf_of ON resource;
|
||||
DROP FUNCTION IF EXISTS reset_agent_on_behalf_of();
|
||||
ALTER TABLE resource DROP COLUMN IF EXISTS on_behalf_of;
|
||||
@@ -0,0 +1,58 @@
|
||||
-- The identity an AI agent runs as when someone runs it from its own page, as a flow's or a
|
||||
-- script's `on_behalf_of` is: a principal (`u/<name>`, `g/<group>`), resolved at deploy time.
|
||||
-- Only `ai_agent` resources read it; every other resource type leaves it NULL.
|
||||
ALTER TABLE resource ADD COLUMN IF NOT EXISTS on_behalf_of VARCHAR(255);
|
||||
|
||||
-- An agent's value is code it runs (inline tools, instructions), so a write of it that does not
|
||||
-- say who the agent now runs as must not keep running it as whoever wrote the previous one:
|
||||
-- anyone who can write the value would otherwise act with that account's permissions. The API's
|
||||
-- writers set the column again in the same transaction; any other path (a direct SQL update, a
|
||||
-- writer added later) leaves it cleared, which runs the agent as whoever runs it.
|
||||
CREATE OR REPLACE FUNCTION reset_agent_on_behalf_of() RETURNS trigger AS $$
|
||||
BEGIN
|
||||
IF NEW.resource_type = 'ai_agent'
|
||||
AND NEW.value IS DISTINCT FROM OLD.value
|
||||
AND NEW.on_behalf_of IS NOT DISTINCT FROM OLD.on_behalf_of THEN
|
||||
NEW.on_behalf_of := NULL;
|
||||
END IF;
|
||||
RETURN NEW;
|
||||
END;
|
||||
$$ LANGUAGE plpgsql;
|
||||
|
||||
CREATE TRIGGER reset_agent_on_behalf_of
|
||||
BEFORE UPDATE ON resource
|
||||
FOR EACH ROW EXECUTE FUNCTION reset_agent_on_behalf_of();
|
||||
|
||||
-- Mirrors `users::username_to_permissioned_as`, as the sibling backfills do. `pg_temp` lives for
|
||||
-- the whole session and migrations share one connection, so replace any earlier copy and drop
|
||||
-- this one at the end.
|
||||
CREATE OR REPLACE FUNCTION pg_temp.username_to_permissioned_as(name VARCHAR)
|
||||
RETURNS VARCHAR AS $$
|
||||
SELECT CASE
|
||||
WHEN $1 LIKE '%@%' AND $1 LIKE '%/%' THEN 'u/' || $1
|
||||
WHEN $1 LIKE '%@%' THEN $1
|
||||
WHEN $1 LIKE 'group-%' THEN 'g/' || substr($1, 7)
|
||||
ELSE 'u/' || $1
|
||||
END;
|
||||
$$ LANGUAGE SQL IMMUTABLE;
|
||||
|
||||
-- An agent saved before this column ran as whoever ran it; give it the account that wrote its
|
||||
-- current version, the one a deploy now records. A principal wider than
|
||||
-- `v2_job.permissioned_as` could not be enqueued, so it is left unset, as the sibling backfills
|
||||
-- leave it (20260801043001, 20260911085221).
|
||||
UPDATE resource r
|
||||
SET on_behalf_of = pg_temp.username_to_permissioned_as(v.created_by)
|
||||
FROM (
|
||||
SELECT DISTINCT ON (workspace_id, path) workspace_id, path, created_by
|
||||
FROM resource_version
|
||||
ORDER BY workspace_id, path, id DESC
|
||||
) v
|
||||
WHERE r.workspace_id = v.workspace_id
|
||||
AND r.path = v.path
|
||||
AND r.resource_type = 'ai_agent'
|
||||
AND r.on_behalf_of IS NULL
|
||||
AND v.created_by IS NOT NULL
|
||||
AND v.created_by <> ''
|
||||
AND length(pg_temp.username_to_permissioned_as(v.created_by)) <= 55;
|
||||
|
||||
DROP FUNCTION pg_temp.username_to_permissioned_as(VARCHAR);
|
||||
@@ -621,12 +621,17 @@ const FLOW_JOBS: [&'static str; 6] = [
|
||||
"jobs/run_and_stream/f",
|
||||
];
|
||||
|
||||
/// Runs of a saved AI agent: the server builds the run from the stored agent, so the route is
|
||||
/// scoped by the agent's path like a deployed runnable's, as `jobs:run:agents:<path>`.
|
||||
const AGENT_JOBS: [&'static str; 1] = ["jobs/run/agent"];
|
||||
|
||||
lazy_static::lazy_static! {
|
||||
static ref RUN_PATH_ACTIONS: Vec<&'static str> = {
|
||||
let mut v = vec!["jobs/resume/", "jobs/run/batch_rerun_jobs", "jobs/run/workflow_as_code", "jobs/run/dependencies","jobs/run/flow_dependencies", "apps_u/execute_component", "apps_u/upload_s3_file"];
|
||||
|
||||
v.extend(SCRIPT_JOBS);
|
||||
v.extend(FLOW_JOBS);
|
||||
v.extend(AGENT_JOBS);
|
||||
v
|
||||
};
|
||||
}
|
||||
@@ -646,13 +651,13 @@ fn map_http_method_to_action(method: &str, route_path: &str) -> ScopeAction {
|
||||
}
|
||||
}
|
||||
|
||||
/// Checks the route path to determine the runnable kind (either "flows" or "scripts").
|
||||
/// Checks the route path to determine the runnable kind ("agents", "flows" or "scripts").
|
||||
///
|
||||
/// The order of checks is important:
|
||||
/// - Flow-related paths are checked first to avoid false positives, as some flow paths
|
||||
/// (e.g., `/run_preview_flow`) share prefixes with script paths (e.g., `/run_preview`).
|
||||
///
|
||||
/// Returns `"flows"` or `"scripts"` based on the match, or `None` if no match is found.
|
||||
/// Returns the kind based on the match, or `None` if no match is found.
|
||||
fn determine_kind_from_route(route_path: &str) -> Option<String> {
|
||||
if route_path.starts_with("jobs") {
|
||||
// Preview/bundle runs execute arbitrary code with no deployed path, so
|
||||
@@ -666,6 +671,9 @@ fn determine_kind_from_route(route_path: &str) -> Option<String> {
|
||||
{
|
||||
return None;
|
||||
}
|
||||
if AGENT_JOBS.iter().any(|path| route_path.starts_with(path)) {
|
||||
return Some("agents".to_string());
|
||||
}
|
||||
if FLOW_JOBS.iter().any(|path| route_path.starts_with(path)) {
|
||||
return Some("flows".to_string());
|
||||
} else if SCRIPT_JOBS.iter().any(|path| route_path.starts_with(path)) {
|
||||
@@ -1725,6 +1733,10 @@ mod tests {
|
||||
scope_for_route("POST", "/api/w/ws/jobs/run/f/u/x/y").as_deref(),
|
||||
Some("jobs:run:flows")
|
||||
);
|
||||
assert_eq!(
|
||||
scope_for_route("POST", "/api/w/ws/jobs/run/agent/u/x/y").as_deref(),
|
||||
Some("jobs:run:agents")
|
||||
);
|
||||
|
||||
// Preview/bundle runs have no deployed path and their handlers require the
|
||||
// broad `jobs:run` scope, so the derived scope must not carry a kind.
|
||||
|
||||
@@ -15955,6 +15955,46 @@ paths:
|
||||
type: string
|
||||
format: uuid
|
||||
|
||||
/w/{workspace}/jobs/run/agent/{path}:
|
||||
post:
|
||||
summary: run a saved AI agent
|
||||
description: |
|
||||
Runs the `ai_agent` resource at `path` as stored, for any caller who can read it
|
||||
(operators included), as the agent's `on_behalf_of`. With `memory_id`, the run is a turn
|
||||
of the caller's conversation with the agent; without, a single run.
|
||||
operationId: runAgent
|
||||
tags:
|
||||
- job
|
||||
parameters:
|
||||
- $ref: "#/components/parameters/WorkspaceId"
|
||||
- $ref: "#/components/parameters/Path"
|
||||
- name: memory_id
|
||||
description: The conversation this turn belongs to. A uuid is used as is; any other string is hashed within the workspace and agent.
|
||||
in: query
|
||||
schema:
|
||||
type: string
|
||||
requestBody:
|
||||
description: the run's inputs
|
||||
required: true
|
||||
content:
|
||||
application/json:
|
||||
schema:
|
||||
type: object
|
||||
properties:
|
||||
user_message:
|
||||
type: string
|
||||
user_attachments:
|
||||
type: array
|
||||
items: {}
|
||||
responses:
|
||||
"201":
|
||||
description: job created
|
||||
content:
|
||||
text/plain:
|
||||
schema:
|
||||
type: string
|
||||
format: uuid
|
||||
|
||||
/w/{workspace}/jobs/run_wait_result/preview_flow:
|
||||
post:
|
||||
summary: run flow preview and wait for result
|
||||
@@ -31566,6 +31606,12 @@ components:
|
||||
type: string
|
||||
ws_specific:
|
||||
type: boolean
|
||||
on_behalf_of:
|
||||
type: string
|
||||
description: ai_agent only. The identity the agent runs as (`u/<username>`, `g/<group>`), kept only with preserve_on_behalf_of by a caller allowed to; otherwise the agent runs as the caller.
|
||||
preserve_on_behalf_of:
|
||||
type: boolean
|
||||
description: ai_agent only. Keep on_behalf_of rather than running as the caller. Honoured for admins and members of wm_deployers.
|
||||
required:
|
||||
- path
|
||||
- value
|
||||
@@ -31590,6 +31636,12 @@ components:
|
||||
type: string
|
||||
ws_specific:
|
||||
type: boolean
|
||||
on_behalf_of:
|
||||
type: string
|
||||
description: ai_agent only. The identity the agent runs as (`u/<username>`, `g/<group>`), kept only with preserve_on_behalf_of by a caller allowed to; a new value otherwise makes the agent run as the caller.
|
||||
preserve_on_behalf_of:
|
||||
type: boolean
|
||||
description: ai_agent only. Keep on_behalf_of rather than running as the caller. Honoured for admins and members of wm_deployers.
|
||||
|
||||
Resource:
|
||||
type: object
|
||||
@@ -31710,6 +31762,9 @@ components:
|
||||
(over a deployed row or a synthesized draft-only row).
|
||||
Frontend appends a `*` to the displayed name.
|
||||
type: boolean
|
||||
on_behalf_of:
|
||||
type: string
|
||||
description: ai_agent only. The identity the agent runs as when run from its own page.
|
||||
agent_memory:
|
||||
description: |
|
||||
For an `ai_agent` resource only, its `memory` setting: the one part
|
||||
|
||||
@@ -0,0 +1,188 @@
|
||||
//! Running a saved AI agent from its own page.
|
||||
//!
|
||||
//! The run is built here from the stored agent rather than sent by the client, which is what lets
|
||||
//! anyone who can read the agent run it, operators included, the way a deployed flow runs by path:
|
||||
//! a preview takes arbitrary code, this takes only the run's inputs. It runs as the agent's
|
||||
//! `on_behalf_of`, resolved when the agent was deployed.
|
||||
|
||||
use std::collections::HashMap;
|
||||
|
||||
use axum::{
|
||||
extract::{Path, Query},
|
||||
Extension, Json,
|
||||
};
|
||||
use hyper::StatusCode;
|
||||
use serde_json::value::RawValue;
|
||||
use windmill_api_auth::check_scopes;
|
||||
use windmill_common::{
|
||||
db::UserDB,
|
||||
error::{Error, Result},
|
||||
jobs::JobPayload,
|
||||
users::username_to_permissioned_as,
|
||||
utils::{not_found_if_none, StripPath},
|
||||
};
|
||||
use windmill_queue::{push, PushArgs, PushIsolationLevel};
|
||||
|
||||
use crate::ai_evals::run::config_to_draft;
|
||||
use crate::ai_evals::subject::AgentDraft;
|
||||
use crate::db::{ApiAuthed, DB};
|
||||
use crate::jobs::{handle_chat_conversation_messages, set_flow_memory_id, RunJobQuery};
|
||||
|
||||
/// The id the agent's step carries in the run, as the editor's chat names it.
|
||||
const AGENT_NODE_ID: &str = "__wm_agent_root";
|
||||
|
||||
/// A turn of the agent's chat when `memory_id` names a conversation, a single run otherwise.
|
||||
/// Returns the job id.
|
||||
pub(crate) async fn run_agent(
|
||||
authed: ApiAuthed,
|
||||
Extension(db): Extension<DB>,
|
||||
Extension(user_db): Extension<UserDB>,
|
||||
Path((w_id, path)): Path<(String, StripPath)>,
|
||||
Query(run_query): Query<RunJobQuery>,
|
||||
Json(args): Json<HashMap<String, Box<RawValue>>>,
|
||||
) -> Result<(StatusCode, String)> {
|
||||
let path = path.to_path();
|
||||
check_scopes(&authed, || format!("jobs:run:agents:{path}"))?;
|
||||
|
||||
// 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 row = sqlx::query!(
|
||||
"SELECT value AS \"value: sqlx::types::Json<serde_json::Value>\", on_behalf_of
|
||||
FROM resource WHERE workspace_id = $1 AND path = $2 AND resource_type = 'ai_agent'",
|
||||
w_id,
|
||||
path
|
||||
)
|
||||
.fetch_optional(&mut *tx)
|
||||
.await?;
|
||||
tx.commit().await?;
|
||||
let row = not_found_if_none(row, "Agent", path)?;
|
||||
let Some(value) = row.value else {
|
||||
return Err(Error::BadRequest(format!(
|
||||
"Agent {path} has no configuration"
|
||||
)));
|
||||
};
|
||||
let config = config_to_draft(value.0)?;
|
||||
|
||||
// A chat turn is filed where the agent's chat lists its conversations, which a flow cannot
|
||||
// name (flow paths carry no `.`), so the two never share a conversation list.
|
||||
let chat = run_query.memory_id.is_some();
|
||||
let run_path = if chat {
|
||||
format!("{path}.chat")
|
||||
} else {
|
||||
path.to_string()
|
||||
};
|
||||
let flow_value = agent_flow(&config, chat)?;
|
||||
|
||||
let on_behalf_of =
|
||||
windmill_common::on_behalf_of_from_permissioned_as(row.on_behalf_of.as_deref(), &w_id, &db)
|
||||
.await?;
|
||||
// As `run_flow` picks the identity of a deployed flow: the agent's own when it has one, the
|
||||
// caller's otherwise, which lends nobody's permissions.
|
||||
let (email, permissioned_as, push_authed, isolation) = if let Some(obo) = on_behalf_of.as_ref()
|
||||
{
|
||||
(
|
||||
&obo.email,
|
||||
obo.permissioned_as.clone(),
|
||||
None,
|
||||
PushIsolationLevel::IsolatedRoot(db.clone()),
|
||||
)
|
||||
} else {
|
||||
(
|
||||
&authed.email,
|
||||
username_to_permissioned_as(&authed.username),
|
||||
Some(authed.clone().into()),
|
||||
PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into()),
|
||||
)
|
||||
};
|
||||
|
||||
let (uuid, mut tx) = push(
|
||||
&db,
|
||||
isolation,
|
||||
&w_id,
|
||||
JobPayload::RawFlow {
|
||||
value: flow_value,
|
||||
path: Some(run_path.clone()),
|
||||
restarted_from: None,
|
||||
},
|
||||
PushArgs::from(&args),
|
||||
authed.display_username(),
|
||||
email,
|
||||
permissioned_as,
|
||||
authed.token_prefix.as_deref(),
|
||||
authed.username_override.as_deref(),
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
false,
|
||||
false,
|
||||
None,
|
||||
true,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
push_authed.as_ref(),
|
||||
false,
|
||||
None,
|
||||
authed.trigger_or_fallback(None),
|
||||
None,
|
||||
)
|
||||
.await?;
|
||||
|
||||
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 {
|
||||
// The caller's conversation, whoever the agent runs as, and a real one: this is the
|
||||
// agent as deployed, not a draft under test.
|
||||
handle_chat_conversation_messages(
|
||||
&mut tx,
|
||||
&authed,
|
||||
&w_id,
|
||||
&run_path,
|
||||
&run_query,
|
||||
args.get("user_message"),
|
||||
uuid,
|
||||
false,
|
||||
&args,
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
tx.commit().await?;
|
||||
|
||||
Ok((StatusCode::CREATED, uuid.to_string()))
|
||||
}
|
||||
|
||||
/// The agent as a one-module flow reading the message and files from the run's inputs, shaped
|
||||
/// through `FlowValue` rather than trusted as raw JSON.
|
||||
fn agent_flow(config: &AgentDraft, chat: bool) -> Result<windmill_common::flows::FlowValue> {
|
||||
let mut input_transforms = match &config.input_transforms {
|
||||
serde_json::Value::Object(map) => map.clone(),
|
||||
_ => serde_json::Map::new(),
|
||||
};
|
||||
// A step memory id would replace the conversation's, and the agent would forget the turns
|
||||
// before; history comes from the conversation alone.
|
||||
for key in ["memory_id", "previous_messages"] {
|
||||
input_transforms.remove(key);
|
||||
}
|
||||
for key in ["user_message", "user_attachments"] {
|
||||
input_transforms.insert(
|
||||
key.to_string(),
|
||||
serde_json::json!({ "type": "javascript", "expr": format!("flow_input.{}", key) }),
|
||||
);
|
||||
}
|
||||
Ok(serde_json::from_value(serde_json::json!({
|
||||
"chat_input_enabled": chat,
|
||||
"modules": [{
|
||||
"id": AGENT_NODE_ID,
|
||||
"value": {
|
||||
"type": "aiagent",
|
||||
"tools": config.tools,
|
||||
"input_transforms": serde_json::Value::Object(input_transforms),
|
||||
}
|
||||
}]
|
||||
}))?)
|
||||
}
|
||||
@@ -348,7 +348,7 @@ pub(crate) async fn require_agent(
|
||||
/// transforms, its tools the module's tools. The same conversion for a draft and for what is
|
||||
/// deployed, so the two hash comparably — which is what lets a draft run be recognised as the
|
||||
/// version it became.
|
||||
fn config_to_draft(value: serde_json::Value) -> Result<AgentDraft> {
|
||||
pub(crate) fn config_to_draft(value: serde_json::Value) -> Result<AgentDraft> {
|
||||
let mut config = match value {
|
||||
serde_json::Value::Object(map) => map,
|
||||
_ => return Err(Error::BadRequest("The agent is not an object".to_string())),
|
||||
|
||||
@@ -279,6 +279,7 @@ pub fn workspaced_service() -> Router {
|
||||
)
|
||||
.route("/add_batch_jobs/{n}", post(add_batch_jobs))
|
||||
.route("/run/preview_flow", post(run_preview_flow_job))
|
||||
.route("/run/agent/{*path}", post(crate::agent_runs::run_agent))
|
||||
.route(
|
||||
"/run_wait_result/preview_flow",
|
||||
post(run_wait_result_preview_flow),
|
||||
|
||||
@@ -121,6 +121,7 @@ pub mod storage_list_ee;
|
||||
mod storage_list_oss;
|
||||
mod workspace_dependencies;
|
||||
|
||||
mod agent_runs;
|
||||
mod ai_evals;
|
||||
mod approvals;
|
||||
#[cfg(all(feature = "enterprise", feature = "private"))]
|
||||
|
||||
@@ -204,6 +204,10 @@ pub struct ListableResource {
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
#[sqlx(default)]
|
||||
pub is_draft: Option<bool>,
|
||||
/// `ai_agent` only: the identity it runs as when run from its own page.
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
#[sqlx(default)]
|
||||
pub on_behalf_of: Option<String>,
|
||||
}
|
||||
|
||||
/// A row of the resource listing: the resource as a single read returns it, plus what only the
|
||||
@@ -234,6 +238,12 @@ pub struct CreateResource {
|
||||
pub labels: Option<Vec<String>>,
|
||||
#[serde(default)]
|
||||
pub ws_specific: Option<bool>,
|
||||
/// `ai_agent` only: the identity to keep, honoured with `preserve_on_behalf_of` for a caller
|
||||
/// allowed to keep one (see `set_agent_on_behalf_of`).
|
||||
#[serde(default)]
|
||||
pub on_behalf_of: Option<String>,
|
||||
#[serde(default)]
|
||||
pub preserve_on_behalf_of: Option<bool>,
|
||||
}
|
||||
#[derive(Deserialize)]
|
||||
struct EditResource {
|
||||
@@ -243,6 +253,8 @@ struct EditResource {
|
||||
resource_type: Option<String>,
|
||||
labels: Option<Vec<String>>,
|
||||
ws_specific: Option<bool>,
|
||||
on_behalf_of: Option<String>,
|
||||
preserve_on_behalf_of: Option<bool>,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
@@ -557,6 +569,8 @@ async fn list_resources(
|
||||
draft_only: Some(true),
|
||||
// Synthesized rows are the authed user's draft.
|
||||
is_draft: Some(true),
|
||||
// Resolved on the first deploy; a draft runs nowhere but the editor's preview.
|
||||
on_behalf_of: None,
|
||||
};
|
||||
rows.push(ListedResource { resource, agent_memory, draft_path });
|
||||
}
|
||||
@@ -590,7 +604,8 @@ async fn get_resource(
|
||||
variable.account,
|
||||
ws_specific.path IS NOT NULL as ws_specific,
|
||||
null::bool as draft_only,
|
||||
null::bool as is_draft
|
||||
null::bool as is_draft,
|
||||
resource.on_behalf_of
|
||||
FROM resource
|
||||
LEFT JOIN variable ON variable.path = resource.path AND variable.workspace_id = $2
|
||||
LEFT JOIN account ON variable.account = account.id AND account.workspace_id = $2
|
||||
@@ -1286,6 +1301,20 @@ async fn create_resource(
|
||||
}
|
||||
}
|
||||
|
||||
if resource.resource_type == "ai_agent" {
|
||||
set_agent_on_behalf_of(
|
||||
&mut tx,
|
||||
&authed,
|
||||
&db,
|
||||
&w_id,
|
||||
&resource.path,
|
||||
resource.on_behalf_of.as_deref(),
|
||||
resource.preserve_on_behalf_of.unwrap_or(false),
|
||||
true,
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
|
||||
// Mirror update_resource: Some(true) inserts, Some(false) clears (only
|
||||
// meaningful on the upsert path, since a pure create has no existing row),
|
||||
// None leaves the existing flag alone.
|
||||
@@ -2172,6 +2201,33 @@ async fn update_resource(
|
||||
|
||||
let npath = not_found_if_none(npath_o, "Resource", path)?;
|
||||
|
||||
// A description or label edit changes nothing the agent runs, so only a new value or an
|
||||
// identity asked for re-resolves who it runs as.
|
||||
if ns.value.is_some() || ns.on_behalf_of.is_some() {
|
||||
let is_agent = sqlx::query_scalar!(
|
||||
"SELECT resource_type = 'ai_agent' FROM resource WHERE workspace_id = $1 AND path = $2",
|
||||
&w_id,
|
||||
&npath
|
||||
)
|
||||
.fetch_optional(&mut *tx)
|
||||
.await?
|
||||
.flatten()
|
||||
.unwrap_or(false);
|
||||
if is_agent {
|
||||
set_agent_on_behalf_of(
|
||||
&mut tx,
|
||||
&authed,
|
||||
&db,
|
||||
&w_id,
|
||||
&npath,
|
||||
ns.on_behalf_of.as_deref(),
|
||||
ns.preserve_on_behalf_of.unwrap_or(false),
|
||||
false,
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
}
|
||||
|
||||
if let Some(nlabels) = &ns.labels {
|
||||
sqlx::query!(
|
||||
"UPDATE resource SET labels = $1 WHERE path = $2 AND workspace_id = $3",
|
||||
@@ -2333,6 +2389,74 @@ async fn update_resource_value(
|
||||
Ok(format!("value of resource {} updated", path))
|
||||
}
|
||||
|
||||
/// Records the identity an `ai_agent` runs as, resolved on every write of it as a flow's
|
||||
/// `on_behalf_of` is on every deploy: the writer's own, unless one allowed to keep another asks
|
||||
/// to, and a folder's default for a new agent. Called in the writer's transaction after the
|
||||
/// value lands, since a value change clears the column (`reset_agent_on_behalf_of`).
|
||||
async fn set_agent_on_behalf_of(
|
||||
tx: &mut Transaction<'_, Postgres>,
|
||||
authed: &ApiAuthed,
|
||||
db: &DB,
|
||||
w_id: &str,
|
||||
path: &str,
|
||||
requested: Option<&str>,
|
||||
preserve: bool,
|
||||
created: bool,
|
||||
) -> Result<()> {
|
||||
let (mut requested, mut preserve) = (requested.map(str::to_string), preserve);
|
||||
let explicit =
|
||||
requested.is_some() && preserve && windmill_common::can_preserve_on_behalf_of(authed);
|
||||
if created && !explicit && windmill_common::can_preserve_on_behalf_of(authed) {
|
||||
if let Some((_, folder_default)) =
|
||||
windmill_common::folders::resolve_folder_default_on_behalf_of(db, w_id, path).await?
|
||||
{
|
||||
requested = Some(folder_default);
|
||||
preserve = true;
|
||||
}
|
||||
}
|
||||
let own = windmill_common::users::username_to_permissioned_as(&authed.username);
|
||||
// Always an identity: nothing requested is the writer's own.
|
||||
let resolved = windmill_common::resolve_on_behalf_of(
|
||||
None,
|
||||
Some(requested.as_deref().unwrap_or(&own)),
|
||||
preserve,
|
||||
authed,
|
||||
w_id,
|
||||
db,
|
||||
)
|
||||
.await?;
|
||||
sqlx::query!(
|
||||
"UPDATE resource SET on_behalf_of = $1 WHERE workspace_id = $2 AND path = $3",
|
||||
resolved,
|
||||
w_id,
|
||||
path
|
||||
)
|
||||
.execute(&mut **tx)
|
||||
.await?;
|
||||
if let Some(kept) = windmill_common::check_on_behalf_of_preservation(
|
||||
resolved.as_deref(),
|
||||
preserve,
|
||||
authed,
|
||||
&own,
|
||||
) {
|
||||
audit_log(
|
||||
&mut **tx,
|
||||
authed,
|
||||
"resources.on_behalf_of",
|
||||
if created {
|
||||
ActionKind::Create
|
||||
} else {
|
||||
ActionKind::Update
|
||||
},
|
||||
w_id,
|
||||
Some(path),
|
||||
Some([("on_behalf_of", kept.as_str())].into()),
|
||||
)
|
||||
.await?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Write a resource's value and run everything that has to follow it: a version row, the audit
|
||||
/// entry, deployment metadata, the webhook, and dependent CI tests. Shared with version restore
|
||||
/// so a restored value is indistinguishable downstream from any other edit.
|
||||
@@ -2375,6 +2499,9 @@ async fn set_resource_value(
|
||||
let Some(resource_type) = updated else {
|
||||
return Err(Error::NotFound(format!("Resource {} not found", path)));
|
||||
};
|
||||
if resource_type == "ai_agent" {
|
||||
set_agent_on_behalf_of(&mut tx, authed, db, w_id, path, None, false, false).await?;
|
||||
}
|
||||
audit_log(
|
||||
&mut *tx,
|
||||
authed,
|
||||
|
||||
Reference in New Issue
Block a user