feat: agents run on behalf of their deployer (#11430)

* Revert "docs: agent editor comments describe runs as the caller"

This reverts commit d94919c9fe.

* Revert "refactor: agents run as the caller, with a note on what a shared one needs"

This reverts commit 0748a2fb51.

* fix: a run as an identity continues only the caller's own conversation

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix: an agent run as its identity takes only attachments its caller may read

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix: the resource editor keeps an agent's identity on an admin's save

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* chore: regenerate workspace tool schemas for preserve_on_behalf_of

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix: the rust client sets preserve_on_behalf_of on resource writes

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* feat: agents run on behalf of their identity only when they opt in

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix: an agent holds an identity only while it runs on behalf of it

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* chore: reduce the agent identity migration to clearing untrusted keys

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix: refuse decoded variable references in inputs to an agent run as its identity

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix: user renames and offboarding re-point an agent's identity

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix: refuse model-written tool arguments that the tool's job would interpolate

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* chore: agent on-behalf-of migration after main's

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix: the agent identity migration also clears trashed agents

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
hugocasa
2026-10-15 19:31:21 +02:00
committed by GitHub
co-authored by Claude Opus 5.5
parent 4361a41b3b
commit 0b980546b1
48 changed files with 1265 additions and 172 deletions
@@ -0,0 +1,29 @@
{
"db_name": "PostgreSQL",
"query": "SELECT resource_type, value AS \"value: sqlx::types::Json<Box<RawValue>>\"\n FROM resource WHERE workspace_id = $1 AND path = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "resource_type",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "value: sqlx::types::Json<Box<RawValue>>",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
false,
true
]
},
"hash": "15f53f40a49b538ca0d4456f69b8d801e8c121f1e0cd904a77ce55bcf9bc5053"
}
@@ -1,15 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO resource (workspace_id, path, value, description, resource_type, extra_perms, edited_at, created_by)\n SELECT $2, path, value, description, resource_type, extra_perms, edited_at, created_by\n FROM resource\n WHERE workspace_id = $1",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Varchar"
]
},
"nullable": []
},
"hash": "748b63f877498d6d279fdf23894611e11c7a05d69b4b7907f8909aec82fb1eb2"
}
@@ -0,0 +1,16 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO resource (workspace_id, path, value, description, resource_type, extra_perms, edited_at, created_by)\n SELECT $2, path,\n CASE WHEN $3::text IS NOT NULL AND resource_type = 'ai_agent'\n AND jsonb_typeof(value) = 'object' AND value ? 'on_behalf_of'\n THEN jsonb_set(value, '{on_behalf_of}', to_jsonb($3::text))\n ELSE value END,\n description, resource_type, extra_perms, edited_at, created_by\n FROM resource\n WHERE workspace_id = $1",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Varchar",
"Text"
]
},
"nullable": []
},
"hash": "bf763c3ae4951537033d18ce48d3eb83d5c19e32aef199bf161e8ba8d1adf176"
}
@@ -0,0 +1 @@
-- Nothing to undo: older versions ignore both keys, and clearing them would drop toggles set since.
@@ -0,0 +1,19 @@
-- An AI agent's `run_on_behalf_of` makes runs from its page take the identity in its
-- `on_behalf_of`, which the server resolves on every write. Before this, both keys were ordinary
-- JSON any writer of the agent could set, so a value written then could name someone else's
-- identity unchecked: clear them, and every agent starts with the toggle off.
UPDATE resource
SET value = value - 'on_behalf_of' - 'run_on_behalf_of'
WHERE resource_type = 'ai_agent'
AND jsonb_typeof(value) = 'object'
AND (value ? 'on_behalf_of' OR value ? 'run_on_behalf_of');
-- A deleted agent comes back from the trash as stored, so its copy there is cleared too.
UPDATE trashbin
SET item_data = jsonb_set(item_data, '{row,value}',
(item_data->'row'->'value') - 'on_behalf_of' - 'run_on_behalf_of')
WHERE item_kind = 'resource'
AND item_data->'row'->>'resource_type' = 'ai_agent'
AND jsonb_typeof(item_data->'row'->'value') = 'object'
AND (item_data->'row'->'value' ? 'on_behalf_of'
OR item_data->'row'->'value' ? 'run_on_behalf_of');
@@ -16,9 +16,17 @@ const CONV: Uuid = Uuid::from_u128(0x5eed);
/// Takes the conversation for a turn, as a run does, without keeping what it writes.
async fn take_conversation(db: &Pool<Postgres>) -> windmill_common::error::Result<Uuid> {
let mut tx = db.begin().await?;
let result =
get_or_create_conversation_with_id(&mut tx, WS, "f/flow", "test-user", "t", CONV, false)
.await;
let result = get_or_create_conversation_with_id(
&mut tx,
WS,
"f/flow",
"test-user",
false,
"t",
CONV,
false,
)
.await;
tx.rollback().await?;
result.map(|c| c.id)
}
+1
View File
@@ -325,6 +325,7 @@ async fn test_new_turns_wait_for_conversation_cleanup_and_recreate(
WS,
"f/flow",
"test-user",
false,
"hi again",
conv_id,
false,
@@ -0,0 +1,216 @@
use serde_json::json;
use sqlx::{Pool, Postgres};
use windmill_test_utils::*;
const AGENT: &str = "u/test-user-3/agent";
async fn post(
base_url: &str,
token: &str,
route: &str,
body: serde_json::Value,
) -> anyhow::Result<(u16, String)> {
let resp = reqwest::Client::new()
.post(format!("{base_url}/w/test-workspace/{route}"))
.header("Authorization", format!("Bearer {token}"))
.json(&body)
.send()
.await?;
Ok((resp.status().as_u16(), resp.text().await?))
}
async fn agent_identity(db: &Pool<Postgres>) -> anyhow::Result<Option<String>> {
Ok(sqlx::query_scalar::<_, Option<String>>(
"SELECT value->>'on_behalf_of' FROM resource
WHERE workspace_id = 'test-workspace' AND path = $1",
)
.bind(AGENT)
.fetch_one(db)
.await?)
}
async fn job_identity(db: &Pool<Postgres>, id: &str) -> anyhow::Result<(String, String)> {
Ok(
sqlx::query_as("SELECT permissioned_as, created_by FROM v2_job WHERE id = $1::uuid")
.bind(id)
.fetch_one(db)
.await?,
)
}
/// The identity an agent runs as is in its value, which anyone who can write the agent writes: it
/// is kept only for an admin asking to keep it, and every other write makes it the writer's own,
/// so a writer cannot change what runs while it keeps running as someone else. It exists only
/// while the agent opts into running as it.
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
async fn test_agent_write_resets_identity(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let base_url = format!("http://localhost:{}/api", server.addr.port());
let value = json!({ "system_prompt": "hi", "on_behalf_of": "u/test-user-2",
"run_on_behalf_of": true });
let (status, body) = post(
&base_url,
"SECRET_TOKEN",
"resources/create",
json!({ "path": AGENT, "resource_type": "ai_agent", "value": value,
"preserve_on_behalf_of": true }),
)
.await?;
assert!((200..300).contains(&status), "create: {status} {body}");
assert_eq!(agent_identity(&db).await?.as_deref(), Some("u/test-user-2"));
// The agent's owner, not an admin, asking to keep it.
let (status, body) = post(
&base_url,
"SECRET_TOKEN_3",
&format!("resources/update/{AGENT}"),
json!({ "value": value, "preserve_on_behalf_of": true }),
)
.await?;
assert!((200..300).contains(&status), "update: {status} {body}");
assert_eq!(agent_identity(&db).await?.as_deref(), Some("u/test-user-3"));
let mut off = value.clone();
off["run_on_behalf_of"] = json!(false);
let (status, body) = post(
&base_url,
"SECRET_TOKEN",
&format!("resources/update/{AGENT}"),
json!({ "value": off, "preserve_on_behalf_of": true }),
)
.await?;
assert!((200..300).contains(&status), "toggle off: {status} {body}");
assert_eq!(agent_identity(&db).await?, None);
Ok(())
}
/// Reading an agent is what allows running it. The run is the caller's unless the agent opts into
/// running as its identity: then `test-user-2`, a plain member, runs as `test-user-3`.
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
async fn test_agent_run_identity_and_needs_read(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let base_url = format!("http://localhost:{}/api", server.addr.port());
sqlx::query(
"INSERT INTO resource (workspace_id, path, value, resource_type, extra_perms, created_by)
VALUES ('test-workspace', $1, $2, 'ai_agent', '{}', 'test-user-3')",
)
.bind(AGENT)
.bind(json!({ "system_prompt": "hi", "on_behalf_of": "u/test-user-3" }))
.execute(&db)
.await?;
let run = format!("jobs/run/agent/{AGENT}");
let (status, body) = post(&base_url, "SECRET_TOKEN_2", &run, json!({})).await?;
assert_eq!(status, 404, "an unreadable agent: {body}");
sqlx::query(
"UPDATE resource SET extra_perms = '{\"u/test-user-2\": false}'
WHERE workspace_id = 'test-workspace' AND path = $1",
)
.bind(AGENT)
.execute(&db)
.await?;
let (status, body) = post(
&base_url,
"SECRET_TOKEN_2",
&run,
json!({ "user_message": "hi" }),
)
.await?;
assert_eq!(status, 201, "a readable agent: {body}");
assert_eq!(
job_identity(&db, &body).await?,
("u/test-user-2".into(), "test-user-2".into())
);
sqlx::query(
"UPDATE resource SET value = value || '{\"run_on_behalf_of\": true}'
WHERE workspace_id = 'test-workspace' AND path = $1",
)
.bind(AGENT)
.execute(&db)
.await?;
let (status, body) = post(
&base_url,
"SECRET_TOKEN_2",
&run,
json!({ "user_message": "hi" }),
)
.await?;
assert_eq!(status, 201, "an agent run on behalf of: {body}");
assert_eq!(
job_identity(&db, &body).await?,
("u/test-user-3".into(), "test-user-2".into())
);
// The worker would resolve it with the identity's token, however the JSON spells the `$`.
for raw in [
r#"{"user_message":"$var:u/test-user-3/secret"}"#,
r#"{"user_message":"\u0024var:u/test-user-3/secret"}"#,
] {
let resp = reqwest::Client::new()
.post(format!("{base_url}/w/test-workspace/{run}"))
.header("Authorization", "Bearer SECRET_TOKEN_2")
.header("Content-Type", "application/json")
.body(raw)
.send()
.await?;
assert_eq!(resp.status().as_u16(), 400, "a reference in {raw}");
}
// The run takes its conversation outside the caller's row policy, so a reader must not
// continue another user's conversation; an admin still may, as the policy allows.
let chat = format!("{run}?memory_id=7d3f6a52-1c4e-4b8a-9f2d-0e5c6b7a8d91");
let msg = json!({ "user_message": "hi" });
let (status, body) = post(&base_url, "SECRET_TOKEN_3", &chat, msg.clone()).await?;
assert_eq!(status, 201, "the owner starts it: {body}");
let (status, body) = post(&base_url, "SECRET_TOKEN_2", &chat, msg.clone()).await?;
assert_eq!(status, 400, "another reader: {body}");
assert!(body.contains("another user"), "{body}");
let (status, body) = post(&base_url, "SECRET_TOKEN", &chat, msg).await?;
assert_eq!(
status, 409,
"an admin passes ownership, held only by the running turn: {body}"
);
Ok(())
}
/// The identity is a username inside the value, so renaming the user has to move it there too.
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
async fn test_agent_identity_follows_user_rename(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let base_url = format!("http://localhost:{}/api", server.addr.port());
sqlx::query(
"INSERT INTO resource (workspace_id, path, value, resource_type, created_by)
VALUES ('test-workspace', $1, $2, 'ai_agent', 'test-user-3')",
)
.bind(AGENT)
.bind(json!({ "run_on_behalf_of": true, "on_behalf_of": "u/test-user-2" }))
.execute(&db)
.await?;
let email: String =
sqlx::query_scalar("SELECT email FROM usr WHERE username = 'test-user-2' LIMIT 1")
.fetch_one(&db)
.await?;
let resp = reqwest::Client::new()
.post(format!("{base_url}/users/rename/{email}"))
.header("Authorization", "Bearer SECRET_TOKEN")
.json(&json!({ "new_username": "renamed" }))
.send()
.await?;
let status = resp.status().as_u16();
assert_eq!(status, 200, "rename: {}", resp.text().await?);
assert_eq!(agent_identity(&db).await?.as_deref(), Some("u/renamed"));
Ok(())
}
@@ -1,60 +0,0 @@
use serde_json::json;
use sqlx::{Pool, Postgres};
use windmill_test_utils::*;
const AGENT: &str = "u/test-user-3/agent";
async fn run(base_url: &str, token: &str) -> anyhow::Result<(u16, String)> {
let resp = reqwest::Client::new()
.post(format!(
"{base_url}/w/test-workspace/jobs/run/agent/{AGENT}"
))
.header("Authorization", format!("Bearer {token}"))
.json(&json!({ "user_message": "hi" }))
.send()
.await?;
Ok((resp.status().as_u16(), resp.text().await?))
}
/// Reading an agent is what allows running it, and the run is the caller's own: `test-user-2`, a
/// plain member, runs `test-user-3`'s agent once it may read it, lending nobody's permissions.
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
async fn test_agent_run_needs_read_and_runs_as_caller(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await?;
let base_url = format!("http://localhost:{}/api", server.addr.port());
sqlx::query(
"INSERT INTO resource (workspace_id, path, value, resource_type, extra_perms, created_by)
VALUES ('test-workspace', $1, $2, 'ai_agent', '{}', 'test-user-3')",
)
.bind(AGENT)
.bind(json!({ "system_prompt": "hi" }))
.execute(&db)
.await?;
let (status, body) = run(&base_url, "SECRET_TOKEN_2").await?;
assert_eq!(status, 404, "an unreadable agent: {body}");
sqlx::query(
"UPDATE resource SET extra_perms = '{\"u/test-user-2\": false}'
WHERE workspace_id = 'test-workspace' AND path = $1",
)
.bind(AGENT)
.execute(&db)
.await?;
let (status, body) = run(&base_url, "SECRET_TOKEN_2").await?;
assert_eq!(status, 201, "a readable agent: {body}");
let (permissioned_as, created_by) = sqlx::query_as::<_, (String, String)>(
"SELECT permissioned_as, created_by FROM v2_job WHERE id = $1::uuid",
)
.bind(&body)
.fetch_one(&db)
.await?;
assert_eq!(permissioned_as, "u/test-user-2");
assert_eq!(created_by, "test-user-2");
Ok(())
}
@@ -62,14 +62,31 @@ async fn fork_with_public_app(
/// An app policy's `on_behalf_of` is the identity anonymous and publisher executions queue jobs
/// under, and the fork's endpoint outlives any revocation in the parent — so a creator who may
/// not preserve someone else's identity must not receive one by forking. `test-user-2` is a
/// plain member of the parent.
/// not preserve someone else's identity must not receive one by forking. The same goes for the
/// one an agent's value carries. `test-user-2` is a plain member of the parent.
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
async fn test_fork_repoints_app_identity_for_unprivileged_creator(
db: Pool<Postgres>,
) -> anyhow::Result<()> {
sqlx::query(
"INSERT INTO resource (workspace_id, path, value, resource_type, extra_perms, created_by)
VALUES ('test-workspace', 'u/test-user/agent', $1, 'ai_agent', '{}', 'test-user')",
)
.bind(json!({ "system_prompt": "hi", "on_behalf_of": "u/test-user" }))
.execute(&db)
.await?;
let (policy, custom_path) = fork_with_public_app(&db, "SECRET_TOKEN_2").await?;
assert_eq!(
sqlx::query_scalar::<_, String>(
"SELECT value->>'on_behalf_of' FROM resource
WHERE workspace_id = 'wm-fork-app' AND path = 'u/test-user/agent'"
)
.fetch_one(&db)
.await?,
"u/test-user-2"
);
assert_eq!(policy["on_behalf_of"], json!("u/test-user-2"));
assert_eq!(policy["on_behalf_of_email"], json!("test2@windmill.dev"));
// `execution_mode` rides along untouched — see `clone_apps`.
@@ -312,8 +329,9 @@ async fn test_fork_keeps_only_resolvable_on_behalf_of(db: Pool<Postgres>) -> any
Ok(())
}
/// Apps, schedules, triggers and their drafts cannot drop an identity the way scripts and flows
/// do, so one naming nobody in the fork goes to its creator while one that still resolves stays.
/// Apps, agents, schedules, triggers and their drafts cannot drop an identity the way scripts and
/// flows do, so one naming nobody in the fork goes to its creator while one that still resolves
/// stays.
/// Forked as an admin, whose app policies the clone otherwise keeps.
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
async fn test_fork_repoints_unresolvable_identities(db: Pool<Postgres>) -> anyhow::Result<()> {
@@ -367,6 +385,13 @@ async fn test_fork_repoints_unresolvable_identities(db: Pool<Postgres>) -> anyho
))
.execute(&db)
.await?;
sqlx::query(
"INSERT INTO resource (workspace_id, path, value, resource_type, extra_perms, created_by)
VALUES ('test-workspace', 'u/test-user/stranger', $1, 'ai_agent', '{}', 'test-user')",
)
.bind(json!({ "system_prompt": "hi", "on_behalf_of": "u/test-user-2" }))
.execute(&db)
.await?;
sqlx::query(
"INSERT INTO schedule (workspace_id, path, edited_by, schedule, script_path, email, permissioned_as, enabled)
VALUES ('test-workspace', 'u/test-user/stranger', 'test-user', '0 0 * * * *', 'u/test-user/s', 'test2@windmill.dev', 'u/test-user-2', false)",
@@ -424,6 +449,10 @@ async fn test_fork_repoints_unresolvable_identities(db: Pool<Postgres>) -> anyho
text("SELECT value->>'permissioned_as' FROM draft WHERE workspace_id = 'wm-fork-repoint' AND typ = 'trigger_websocket'").await?,
"u/test-user"
);
assert_eq!(
text("SELECT value->>'on_behalf_of' FROM resource WHERE workspace_id = 'wm-fork-repoint' AND path = 'u/test-user/stranger'").await?,
"u/test-user"
);
assert_eq!(
text("SELECT permissioned_as || ' ' || email FROM schedule WHERE workspace_id = 'wm-fork-repoint'").await?,
"u/test-user test@windmill.dev"
@@ -90,6 +90,25 @@ async fn test_offboard_to_user(db: Pool<Postgres>) -> anyhow::Result<()> {
.execute(&db)
.await?;
// An agent records who it runs as inside its value, on a path the user does not own.
sqlx::query(
"INSERT INTO resource (workspace_id, path, value, resource_type, created_by)
VALUES ('test-workspace', 'f/shared/agent', $1, 'ai_agent', 'test-user')",
)
.bind(json!({ "run_on_behalf_of": true, "on_behalf_of": "u/test-user-2" }))
.execute(&db)
.await?;
let preview: serde_json::Value =
authed(client().get(ws_url(port, "offboard_preview/test-user-2")))
.send()
.await?
.json()
.await?;
assert_eq!(
preview["executing_on_behalf"]["resources"],
json!(["f/shared/agent"])
);
let resp = authed(client().post(ws_url(port, "offboard/test-user-2")))
.json(&json!({
"reassign_to": "u/test-user",
@@ -100,6 +119,14 @@ async fn test_offboard_to_user(db: Pool<Postgres>) -> anyhow::Result<()> {
.await?;
assert_eq!(resp.status(), 200);
let agent_obo: Option<String> = sqlx::query_scalar(
"SELECT value->>'on_behalf_of' FROM resource
WHERE path = 'f/shared/agent' AND workspace_id = 'test-workspace'",
)
.fetch_one(&db)
.await?;
assert_eq!(agent_obo.as_deref(), Some("u/test-user"));
let body: serde_json::Value = resp.json().await?;
assert!(body["conflicts"].as_array().map_or(true, |a| a.is_empty()));
assert!(body["summary"].is_object());
@@ -117,7 +144,10 @@ async fn test_offboard_to_user(db: Pool<Postgres>) -> anyhow::Result<()> {
.fetch_one(&db)
.await?;
assert_eq!(
(obo.on_behalf_of.as_deref(), obo.on_behalf_of_email.as_deref()),
(
obo.on_behalf_of.as_deref(),
obo.on_behalf_of_email.as_deref()
),
(Some("u/test-user"), Some("test@windmill.dev")),
"the reassignment moves the whole identity, not just the half this release reads"
);
@@ -946,6 +946,7 @@ pub async fn handle_chat_conversation_messages(
w_id,
flow_path,
&authed.username,
authed.is_admin,
&user_message,
memory_id,
is_test,
+9
View File
@@ -2120,6 +2120,15 @@ async fn change_user_email(
.execute(&mut *tx)
.await?;
sqlx::query(
"UPDATE resource SET value = jsonb_set(value, '{on_behalf_of}', to_jsonb($1::text))
WHERE resource_type = 'ai_agent' AND value->>'on_behalf_of' = $2",
)
.bind(&new_principal)
.bind(&old_principal)
.execute(&mut *tx)
.await?;
sqlx::query!(
"UPDATE script SET on_behalf_of = $1 WHERE on_behalf_of = $2",
&new_principal,
@@ -7130,7 +7130,7 @@ async fn clone_workspace_data(
clone_resource_types(tx, source_workspace_id, target_workspace_id).await?;
// Clone resources
clone_resources(tx, source_workspace_id, target_workspace_id).await?;
clone_resources(tx, source_workspace_id, target_workspace_id, authed).await?;
// Clone variables (including external secret backend replication)
clone_variables(tx, db, source_workspace_id, target_workspace_id).await?;
@@ -7667,18 +7667,30 @@ async fn clone_resource_types(
Ok(())
}
/// An agent's value carries the identity it runs as, re-pointed at the fork's creator when they
/// may not preserve someone else's, as `clone_apps` does an app's policy. The rest is swept by
/// `repoint_unresolvable_cloned_identities` once the fork's membership is final.
async fn clone_resources(
tx: &mut Transaction<'_, Postgres>,
source_workspace_id: &str,
target_workspace_id: &str,
authed: &ApiAuthed,
) -> Result<()> {
let repoint = (!windmill_common::can_preserve_on_behalf_of(authed))
.then(|| username_to_permissioned_as(&authed.username));
sqlx::query!(
"INSERT INTO resource (workspace_id, path, value, description, resource_type, extra_perms, edited_at, created_by)
SELECT $2, path, value, description, resource_type, extra_perms, edited_at, created_by
SELECT $2, path,
CASE WHEN $3::text IS NOT NULL AND resource_type = 'ai_agent'
AND jsonb_typeof(value) = 'object' AND value ? 'on_behalf_of'
THEN jsonb_set(value, '{on_behalf_of}', to_jsonb($3::text))
ELSE value END,
description, resource_type, extra_perms, edited_at, created_by
FROM resource
WHERE workspace_id = $1",
source_workspace_id,
target_workspace_id,
repoint,
)
.execute(&mut **tx)
.await?;
@@ -8165,6 +8177,18 @@ async fn repoint_unresolvable_cloned_identities(
.await?;
}
sqlx::query(&format!(
"UPDATE resource SET value = jsonb_set(value, '{{on_behalf_of}}', to_jsonb($2::text))
WHERE workspace_id = $1 AND resource_type = 'ai_agent'
AND jsonb_typeof(value) = 'object' AND value ? 'on_behalf_of'
AND NOT ({})",
principal_resolves_sql("(value->>'on_behalf_of')")
))
.bind(target_workspace_id)
.bind(&principal)
.execute(&mut **tx)
.await?;
// `email` is still written for workers that predate `permissioned_as`.
sqlx::query(&format!(
"UPDATE schedule SET permissioned_as = $2, email = $3
@@ -13941,17 +13965,18 @@ async fn compare_two_flows(
});
}
/// The policy minus the identity pair. Deploying cannot converge a difference there — the
/// target recomputes the identity from the deployer's own choice, which offers its current
/// value, the deployer, or a typed-in one, never the source's — so listing an app for it
/// alone leaves an entry no deploy can clear. `script` and `flow` compare no identity either.
fn policy_without_identity(policy: &serde_json::Value) -> serde_json::Value {
let mut policy = policy.clone();
if let Some(obj) = policy.as_object_mut() {
/// An app's policy or an agent's value minus the identity it carries. Deploying cannot converge
/// a difference there — the target recomputes the identity from the deployer's own choice, which
/// offers its current value, the deployer, or a typed-in one, never the source's — so listing an
/// item for it alone leaves an entry no deploy can clear. `script` and `flow` compare no identity
/// either.
fn without_identity(value: &serde_json::Value) -> serde_json::Value {
let mut value = value.clone();
if let Some(obj) = value.as_object_mut() {
obj.remove("on_behalf_of");
obj.remove("on_behalf_of_email");
}
policy
value
}
async fn compare_two_apps(
@@ -13994,7 +14019,7 @@ async fn compare_two_apps(
// Check metadata and content differences
if let (Some(source), Some(target)) = (&source_app, &target_app) {
if source.summary != target.summary
|| policy_without_identity(&source.policy) != policy_without_identity(&target.policy)
|| without_identity(&source.policy) != without_identity(&target.policy)
|| source.value != target.value
|| source.raw_app != target.raw_app
{
@@ -14072,7 +14097,14 @@ async fn compare_two_resources(
// Check metadata differences
if let (Some(source), Some(target)) = (&source_resource, &target_resource) {
if source.value != target.value
let agent = source.resource_type == "ai_agent" && target.resource_type == "ai_agent";
let values_differ = if agent {
source.value.as_ref().map(without_identity)
!= target.value.as_ref().map(without_identity)
} else {
source.value != target.value
};
if values_differ
|| source.description != target.description
|| source.resource_type != target.resource_type
{
+9 -2
View File
@@ -16783,8 +16783,9 @@ paths:
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 caller, who needs access to the agent's AI resource and to
whatever its tools use. With `memory_id`, the run is a turn
(operators included). The run is the caller's, or the agent's `on_behalf_of` when its
`run_on_behalf_of` is set; the caller's inputs may then not reference variables or
resources. With `memory_id`, the run is a turn
of the caller's conversation with the agent; without, a single run.
operationId: runAgent
tags:
@@ -32686,6 +32687,9 @@ components:
type: string
ws_specific:
type: boolean
preserve_on_behalf_of:
type: boolean
description: ai_agent only. Keep the identity `value.on_behalf_of` names (`u/<username>`, `g/<group>`) rather than the caller's own. Honoured for admins and members of wm_deployers; otherwise every write of an agent sets it to the caller.
required:
- path
- value
@@ -32710,6 +32714,9 @@ components:
type: string
ws_specific:
type: boolean
preserve_on_behalf_of:
type: boolean
description: ai_agent only. Keep the identity `value.on_behalf_of` names (`u/<username>`, `g/<group>`) rather than the caller's own. Honoured for admins and members of wm_deployers; otherwise every write of an agent sets it to the caller.
Resource:
type: object
+105 -11
View File
@@ -2,8 +2,9 @@
//!
//! 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 caller, who
//! needs access to the agent's AI resource and to whatever its tools use.
//! a preview takes arbitrary code, this takes only the run's inputs. It runs as the caller, or as
//! the agent's `on_behalf_of` when its `run_on_behalf_of` is set; the server resolves
//! `on_behalf_of` on every write of the agent.
use std::collections::HashMap;
@@ -20,13 +21,16 @@ use windmill_common::{
jobs::JobPayload,
users::username_to_permissioned_as,
utils::{not_found_if_none, StripPath},
variables::arg_needs_interpolation,
};
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};
use crate::jobs::{
handle_chat_conversation_messages, run_end_user_email, set_flow_memory_id, RunJobQuery,
};
/// The id the agent's step carries in the run, as the editor's chat names it. A token scoped to
/// `jobs:run:agents:<path>` reads its runs back by it (`require_job_within_read_scope`).
@@ -59,11 +63,20 @@ pub(crate) async fn run_agent(
.fetch_optional(&mut *tx)
.await?;
tx.commit().await?;
let Some(sqlx::types::Json(value)) = not_found_if_none(value, "Agent", path)? else {
let Some(sqlx::types::Json(mut value)) = not_found_if_none(value, "Agent", path)? else {
return Err(Error::BadRequest(format!(
"Agent {path} has no configuration"
)));
};
// Who the agent runs as, not one of its inputs: its identity when it opts in, as a deployed
// flow does.
let obj = value.as_object_mut();
let (agent_obo, run_obo) = obj
.map(|o| (o.remove("on_behalf_of"), o.remove("run_on_behalf_of")))
.unwrap_or_default();
let agent_obo = agent_obo
.filter(|_| run_obo.and_then(|v| v.as_bool()).unwrap_or(false))
.and_then(|v| v.as_str().map(str::to_string));
let config = config_to_draft(value)?;
// A chat turn is filed where the agent's chat lists its conversations, which a flow cannot
@@ -83,11 +96,47 @@ pub(crate) async fn run_agent(
&["memory_id", "previous_messages"],
)?;
let push_authed = authed.clone().into();
let on_behalf_of =
windmill_common::on_behalf_of_from_permissioned_as(agent_obo.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()),
)
};
if let Some(obo) = on_behalf_of.as_ref() {
// The worker would interpolate such an input with the job's token, which is the
// identity's. Checked on the decoded value: the raw JSON can spell `$` as `\u0024`.
if args.values().any(|v| {
serde_json::from_str::<serde_json::Value>(v.get())
.is_ok_and(|v| arg_needs_interpolation(&v.to_string()))
}) {
return Err(Error::BadRequest(format!(
"Agent {path} runs on behalf of {}, so its inputs cannot reference variables or resources",
obo.permissioned_as
)));
}
#[cfg(all(feature = "parquet", feature = "private"))]
check_attachments_readable(&db, &authed, &w_id, &args).await?;
}
let scope_ceiling = windmill_api_auth::caller_scope_ceiling(&db, &authed).await?;
let end_user_email = run_end_user_email(&db, &w_id, &authed, None).await?;
let (uuid, mut tx) = push(
&db,
PushIsolationLevel::Isolated(user_db.clone(), authed.clone().into()),
isolation,
&w_id,
JobPayload::RawFlow {
value: flow_value,
@@ -96,8 +145,8 @@ pub(crate) async fn run_agent(
},
PushArgs::from(&args),
authed.display_username(),
&authed.email,
username_to_permissioned_as(&authed.username),
email,
permissioned_as,
authed.token_prefix.as_deref(),
authed.username_override.as_deref(),
None,
@@ -114,9 +163,9 @@ pub(crate) async fn run_agent(
None,
None,
None,
Some(&push_authed),
push_authed.as_ref(),
false,
None,
end_user_email,
authed.trigger_or_fallback(None),
None,
scope_ceiling.as_deref(),
@@ -127,7 +176,8 @@ pub(crate) async fn run_agent(
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.
// 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,
@@ -146,6 +196,50 @@ pub(crate) async fn run_agent(
Ok((StatusCode::CREATED, uuid.to_string()))
}
/// The step fetches the message's files with its job's token, so a run as the agent's identity
/// would read any file that identity can: a caller attaches only what they could download
/// themselves, checked as `sign_s3_objects` checks a key before signing it.
#[cfg(all(feature = "parquet", feature = "private"))]
async fn check_attachments_readable(
db: &DB,
authed: &ApiAuthed,
w_id: &str,
args: &HashMap<String, Box<RawValue>>,
) -> Result<()> {
use crate::job_helpers_oss::{
get_large_file_storage, get_workspace_s3_resource_and_check_paths,
};
use windmill_common::db::DbWithOptAuthed;
use windmill_types::s3::{S3Object, S3Permission};
let Some(raw) = args.get("user_attachments") else {
return Ok(());
};
let attachments: Option<Vec<S3Object>> = serde_json::from_str(raw.get())
.map_err(|e| Error::BadRequest(format!("invalid user_attachments: {e}")))?;
for attachment in attachments.iter().flatten().filter(|a| !a.s3.is_empty()) {
// The path check is skipped, not failed, when the storage does not resolve.
if get_large_file_storage(db, w_id, attachment.storage.clone())
.await?
.is_none()
{
return Err(windmill_object_store::workspace_storage_not_found(
attachment.storage.as_deref(),
));
}
get_workspace_s3_resource_and_check_paths(
&DbWithOptAuthed::from_authed(authed, db.clone(), None),
Some(authed),
w_id,
attachment.storage.clone(),
&[(&attachment.s3, S3Permission::READ)],
None,
)
.await?;
}
Ok(())
}
/// The agent as a one-step flow, validated by deserializing through `FlowValue` rather than
/// trusted as raw JSON. Its own transforms are the step's, minus `without`, and the run's inputs
/// supply the message and the attachments over the top. Always inlined, never a link to the
+22
View File
@@ -266,6 +266,17 @@ async fn get_offboard_preview(
departing.as_deref(), &user_prefix, w_id
).fetch_all(db).await?;
// An agent's identity sits in its value, under any path, the user's own included: those
// move with the path but keep running as the departing user until re-pointed.
let obo_agents: Vec<String> = sqlx::query_scalar(
"SELECT path FROM resource WHERE resource_type = 'ai_agent'
AND value->>'on_behalf_of' = $1 AND workspace_id = $2",
)
.bind(departing.as_deref())
.bind(w_id)
.fetch_all(db)
.await?;
let obo_schedules = sqlx::query_scalar!(
"SELECT path FROM schedule WHERE permissioned_as = $1 AND NOT path LIKE $2 AND workspace_id = $3",
&user_owner, &user_prefix, w_id
@@ -354,6 +365,7 @@ async fn get_offboard_preview(
scripts: obo_scripts,
flows: obo_flows,
apps: obo_apps,
resources: obo_agents,
schedules: obo_schedules,
triggers: obo_triggers,
..Default::default()
@@ -1007,6 +1019,16 @@ async fn offboard_user_from_workspace<'c>(
.await?
.unwrap_or(0);
sqlx::query(
"UPDATE resource SET value = jsonb_set(value, '{on_behalf_of}', to_jsonb($1::text))
WHERE resource_type = 'ai_agent' AND value->>'on_behalf_of' = $2 AND workspace_id = $3",
)
.bind(new_permissioned_as)
.bind(departing.as_deref())
.bind(w_id)
.execute(&mut **tx)
.await?;
// ---- eval datasets ----
// The foreign keys cascade the rename onto cases and experiments; the paths held inside JSONB
// (an experiment's subject, a dataset's scorers) are rewritten separately since the cascade
+26 -2
View File
@@ -297,8 +297,21 @@ async fn rename_user(
// keeps a like-named member of some other workspace out of it.
if let Some(old_username) = old_instance_username.filter(|u| *u != ru.new_username) {
let old_principal = windmill_common::users::username_to_permissioned_as(&old_username);
let new_principal =
windmill_common::users::username_to_permissioned_as(&ru.new_username);
let new_principal = windmill_common::users::username_to_permissioned_as(&ru.new_username);
// An agent records no address, so a workspace where the name is a member's own is
// left to the per-workspace sweep: there the principal is that member.
sqlx::query(
"UPDATE resource SET value = jsonb_set(value, '{on_behalf_of}', to_jsonb($1::text))
WHERE resource_type = 'ai_agent' AND value->>'on_behalf_of' = $2
AND NOT EXISTS (SELECT 1 FROM usr
WHERE usr.workspace_id = resource.workspace_id
AND usr.username = $3)",
)
.bind(&new_principal)
.bind(&old_principal)
.bind(&old_username)
.execute(&mut *tx)
.await?;
sqlx::query!(
"UPDATE app SET policy = jsonb_set(policy, ARRAY['on_behalf_of'], to_jsonb($1::text))
WHERE policy->>'on_behalf_of' = $2 AND policy->>'on_behalf_of_email' = $3",
@@ -817,6 +830,17 @@ async fn update_username_in_workpsace<'c>(
.execute(&mut **tx)
.await?;
sqlx::query(
"UPDATE resource SET value = jsonb_set(value, '{on_behalf_of}', to_jsonb('u/' || $1))
WHERE resource_type = 'ai_agent' AND value->>'on_behalf_of' = ('u/' || $2)
AND workspace_id = $3",
)
.bind(new_username)
.bind(old_username)
.bind(w_id)
.execute(&mut **tx)
.await?;
// An app draft carries a copy of the deployed policy, so the rename must reach it
// there too — same reason as the script/flow draft sweep above.
sqlx::query!(
@@ -894,7 +894,14 @@ pub(crate) async fn tarball_workspace(
.fetch_all(&mut *tx)
.await?;
for resource in resources {
for mut resource in resources {
// An agent's identity belongs to this workspace, and whoever writes it elsewhere gets
// their own, as `to_string_without_metadata` leaves out a flow's.
if resource.resource_type == "ai_agent" {
if let Some(obj) = resource.value.as_mut().and_then(|v| v.as_object_mut()) {
obj.remove("on_behalf_of");
}
}
let resource_str =
&to_string_without_metadata(&resource, ExtraPermsBehavior::Drop, None).unwrap();
archive
@@ -64,12 +64,13 @@ pub async fn get_or_create_conversation_with_id(
w_id: &str,
flow_path: &str,
username: &str,
is_admin: bool,
title: &str,
conversation_id: Uuid,
is_test: bool,
) -> Result<FlowConversation> {
if let Some(existing) = lock_conversation(tx, w_id, conversation_id).await? {
let existing = same_kind(existing, is_test)?;
let existing = same_kind(owned(existing, username, is_admin)?, is_test)?;
refuse_running_turn(tx, conversation_id).await?;
return Ok(existing);
}
@@ -106,7 +107,7 @@ pub async fn get_or_create_conversation_with_id(
"conversation {conversation_id} belongs to another workspace"
))
})?;
let existing = same_kind(existing, is_test)?;
let existing = same_kind(owned(existing, username, is_admin)?, is_test)?;
refuse_running_turn(tx, conversation_id).await?;
Ok(existing)
}
@@ -176,6 +177,19 @@ pub async fn running_turns<'e, E: sqlx::PgExecutor<'e>>(
.collect())
}
/// The `see_own` policy on `flow_conversation`, for a turn whose transaction bypasses it: a run
/// as a flow's or agent's `on_behalf_of` takes the conversation in a root transaction, and must
/// not continue, or read the memory of, another user's conversation.
fn owned(existing: FlowConversation, username: &str, is_admin: bool) -> Result<FlowConversation> {
if is_admin || existing.created_by == username {
return Ok(existing);
}
Err(crate::error::Error::BadRequest(format!(
"conversation {} belongs to another user; start a new conversation",
existing.id
)))
}
/// `memory_id` is the caller's to choose, so a preview run could name a deployed
/// conversation and the reverse. A conversation's kind is fixed at creation and nothing
/// would show the mixing afterwards, so the turn is refused before it starts.
+9
View File
@@ -23,6 +23,15 @@ use std::sync::atomic::{AtomicBool, Ordering};
lazy_static::lazy_static! {
pub static ref SECRET_SALT: Option<String> = std::env::var("SECRET_SALT").ok();
static ref RESERVED_WM_VAR_NAME: regex::Regex = regex::Regex::new(r"^WM_[A-Z_]+$").unwrap();
static ref RE_RES_VAR: regex::Regex =
regex::Regex::new(r#"\$(?:var|jsonvar|res|encrypted)\:"#).unwrap();
}
/// Whether a job argument, as JSON text, holds a `$var:`/`$jsonvar:`/`$res:`/`$encrypted:`
/// reference, which makes the worker interpolate the whole argument with the job's token,
/// `$`-prefixed reserved variables included.
pub fn arg_needs_interpolation(json: &str) -> bool {
RE_RES_VAR.is_match(json)
}
/// Escape a string so it can be safely embedded inside a single-quoted JS
+181 -4
View File
@@ -236,6 +236,10 @@ pub struct CreateResource {
pub labels: Option<Vec<String>>,
#[serde(default)]
pub ws_specific: Option<bool>,
/// `ai_agent` only: keep the identity `value.on_behalf_of` names, for a caller allowed to
/// (see `resolve_agent_on_behalf_of`).
#[serde(default)]
pub preserve_on_behalf_of: Option<bool>,
}
#[derive(Deserialize)]
struct EditResource {
@@ -245,6 +249,7 @@ struct EditResource {
resource_type: Option<String>,
labels: Option<Vec<String>>,
ws_specific: Option<bool>,
preserve_on_behalf_of: Option<bool>,
}
#[derive(Deserialize)]
@@ -1254,7 +1259,20 @@ async fn create_resource(
check_path_conflict(&mut tx, &w_id, &resource.path).await?;
}
let res_value = resource.value.unwrap_or_default();
let mut res_value = resource.value.unwrap_or_default();
if resource.resource_type == "ai_agent" {
res_value = resolve_agent_raw_value(
&mut tx,
&authed,
&db,
&w_id,
&resource.path,
&res_value,
resource.preserve_on_behalf_of.unwrap_or(false),
true,
)
.await?;
}
let raw_json = sqlx::types::Json(res_value.as_ref());
if resource.path.starts_with("f/app_themes/") {
@@ -2095,9 +2113,6 @@ async fn update_resource(
if let Some(npath) = &ns.path {
sqlb.set_str("path", npath);
}
if let Some(nvalue) = &ns.value {
sqlb.set_str("value", nvalue.to_string());
}
if let Some(nrt) = &ns.resource_type {
sqlb.set_str("resource_type", nrt);
}
@@ -2205,6 +2220,49 @@ async fn update_resource(
}
}
lock_resource_row(&mut tx, &w_id, path).await?;
let current = sqlx::query!(
"SELECT resource_type, value AS \"value: sqlx::types::Json<Box<RawValue>>\"
FROM resource WHERE workspace_id = $1 AND path = $2",
&w_id,
path
)
.fetch_optional(&mut *tx)
.await?;
let resource_type = ns
.resource_type
.clone()
.or_else(|| current.as_ref().map(|c| c.resource_type.clone()));
// Retyping a resource into an agent makes its value an agent's, identity included, so it is
// resolved as a written one is.
let retyped = ns.resource_type.as_deref() == Some("ai_agent")
&& current
.as_ref()
.is_some_and(|c| c.resource_type != "ai_agent");
let nvalue = match &ns.value {
Some(v) => Some(v.clone()),
None if retyped => current.and_then(|c| c.value).map(|v| v.0),
None => None,
};
if let Some(nvalue) = nvalue {
let nvalue = if resource_type.as_deref() == Some("ai_agent") {
resolve_agent_raw_value(
&mut tx,
&authed,
&db,
&w_id,
ns.path.as_deref().unwrap_or(path),
&nvalue,
ns.preserve_on_behalf_of.unwrap_or(false),
false,
)
.await?
} else {
nvalue
};
sqlb.set_str("value", nvalue.to_string());
}
let sql = sqlb.sql().map_err(|e| Error::internal_err(e.to_string()))?;
let npath_o: Option<String> = sqlx::query_scalar(&sql)
.fetch_optional(&mut *tx)
@@ -2374,6 +2432,110 @@ async fn update_resource_value(
Ok(format!("value of resource {} updated", path))
}
/// Rewrites the identity an `ai_agent` runs as (`value.on_behalf_of`), resolved on every write of
/// it as a flow's `on_behalf_of` is on every deploy. It exists only while the agent's
/// `run_on_behalf_of` is set: then the writer's own, unless one allowed to keep the one the value
/// names asks to, and a folder's default for a new agent. The value is written by anyone who can
/// write the agent, so the key it arrives with is only a request.
async fn resolve_agent_on_behalf_of(
tx: &mut Transaction<'_, Postgres>,
authed: &ApiAuthed,
db: &DB,
w_id: &str,
path: &str,
value: &mut serde_json::Value,
preserve: bool,
created: bool,
) -> Result<()> {
let Some(obj) = value.as_object_mut() else {
return Ok(());
};
let mut requested = obj
.remove("on_behalf_of")
.and_then(|v| v.as_str().map(str::to_string));
if obj.get("run_on_behalf_of").and_then(|v| v.as_bool()) != Some(true) {
return Ok(());
}
let mut preserve = 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);
// 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?;
if let Some(resolved) = resolved.as_ref() {
obj.insert("on_behalf_of".to_string(), resolved.clone().into());
}
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(())
}
/// Held until the write commits: whether a value is resolved as an agent's depends on the type
/// read after this, and a concurrent retype into an agent must not land between the two.
async fn lock_resource_row(
tx: &mut Transaction<'_, Postgres>,
w_id: &str,
path: &str,
) -> Result<()> {
sqlx::query("SELECT 1 FROM resource WHERE workspace_id = $1 AND path = $2 FOR UPDATE")
.bind(w_id)
.bind(path)
.execute(&mut **tx)
.await?;
Ok(())
}
/// `resolve_agent_on_behalf_of` on a value as the request carried it.
async fn resolve_agent_raw_value(
tx: &mut Transaction<'_, Postgres>,
authed: &ApiAuthed,
db: &DB,
w_id: &str,
path: &str,
value: &RawValue,
preserve: bool,
created: bool,
) -> Result<Box<RawValue>> {
let mut parsed: serde_json::Value = serde_json::from_str(value.get())?;
resolve_agent_on_behalf_of(tx, authed, db, w_id, path, &mut parsed, preserve, created).await?;
Ok(serde_json::value::to_raw_value(&parsed)?)
}
/// 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.
@@ -2402,6 +2564,21 @@ async fn set_resource_value(
let mut tx = user_db.clone().begin(authed).await?;
let mut value = value;
if let Some(v) = value.as_mut() {
lock_resource_row(&mut tx, w_id, path).await?;
let resource_type = sqlx::query_scalar!(
"SELECT resource_type FROM resource WHERE workspace_id = $1 AND path = $2",
w_id,
path
)
.fetch_optional(&mut *tx)
.await?;
if resource_type.as_deref() == Some("ai_agent") {
resolve_agent_on_behalf_of(&mut tx, authed, db, w_id, path, v, false, false).await?;
}
}
// `RETURNING resource_type` rather than a second lookup: the advisory below has to know the
// type to leave `state` and `cache` alone, and this statement already runs.
let updated = sqlx::query_scalar!(
+62 -5
View File
@@ -19,6 +19,7 @@ use std::{collections::HashMap, sync::Arc, time::Duration};
use uuid::Uuid;
use windmill_ai::{ai_types::OpenAIToolCall, query_builder::StreamEventSink, types::*};
use windmill_common::jobs::JobPayload;
use windmill_common::variables::arg_needs_interpolation;
#[cfg(feature = "mcp")]
use windmill_mcp::McpClient;
@@ -153,6 +154,8 @@ pub async fn execute_tool_calls(
final_events_str,
)
.await?;
} else if tool.module.is_some() && interpolating_args(tool_call) {
refuse_tool_call(&mut ctx, tool_call, &mut messages, final_events_str).await?;
} else if tool.module.is_some() {
let mut batch = vec![(tool_call, tool)];
while let Some(next_call) = calls.peek() {
@@ -168,6 +171,9 @@ pub async fn execute_tool_calls(
}) else {
break;
};
if interpolating_args(next_call) {
break;
}
batch.push((calls.next().unwrap(), next_tool));
}
execute_windmill_tools(&mut ctx, &batch, actions, &mut messages, final_events_str)
@@ -189,6 +195,44 @@ pub async fn execute_tool_calls(
Ok((messages, final_content, used_structured_output_tool))
}
/// Whether the model wrote an argument the tool's job would interpolate with the agent's token,
/// which may be an identity it runs on behalf of. Only the step's own input transforms may name
/// variables and resources; the model's arguments are text from whoever talks to it.
fn interpolating_args(tool_call: &OpenAIToolCall) -> bool {
serde_json::from_str::<serde_json::Value>(&tool_call.function.arguments)
.is_ok_and(|v| arg_needs_interpolation(&v.to_string()))
}
async fn refuse_tool_call(
ctx: &mut ToolExecutionContext<'_>,
tool_call: &OpenAIToolCall,
messages: &mut Vec<OpenAIMessage>,
final_events_str: &mut String,
) -> Result<(), Error> {
let error_msg = "Tool call refused: its arguments cannot reference variables or resources \
($var:, $jsonvar:, $res:, $encrypted:)"
.to_string();
messages.push(OpenAIMessage {
role: "tool".to_string(),
content: Some(OpenAIContent::Text(error_msg.clone())),
tool_call_id: Some(tool_call.id.clone()),
..Default::default()
});
if let Some(stream_event_processor) = ctx.stream_event_processor {
let event = StreamingEvent::ToolResult {
call_id: tool_call.id.clone(),
function_name: tool_call.function.name.clone(),
result: error_msg.clone(),
success: false,
};
stream_event_processor.send(event, final_events_str).await?;
}
let (content, extras) = windmill_tool_row(tool_call, false, &error_msg);
let agent_job_id = ctx.job.id;
add_tool_message_to_chat(ctx, Some(agent_job_id), &content, false, Some(extras)).await;
Ok(())
}
/// Execute an MCP tool call
async fn execute_mcp_tool_call(
ctx: &mut ToolExecutionContext<'_>,
@@ -618,10 +662,8 @@ async fn enqueue_windmill_tool(
// narrows that further.
windmill_common::scopes::intersect_job_token_scopes(
ctx.job.job_token_scopes.as_deref(),
windmill_common::scopes::step_job_token_scopes(
tool_module.job_token_scopes.as_deref(),
)
.as_deref(),
windmill_common::scopes::step_job_token_scopes(tool_module.job_token_scopes.as_deref())
.as_deref(),
)
.as_deref(),
)
@@ -1051,7 +1093,7 @@ async fn add_tool_message_to_chat(
#[cfg(test)]
mod tests {
use super::{extract_ai_agent_output, windmill_tool_row};
use super::{extract_ai_agent_output, interpolating_args, windmill_tool_row};
use serde_json::value::RawValue;
use windmill_ai::ai_types::{OpenAIFunction, OpenAIToolCall};
@@ -1088,6 +1130,21 @@ mod tests {
);
}
#[test]
fn a_model_written_reference_is_caught_however_it_is_spelled() {
let call = |arguments: &str| OpenAIToolCall {
id: "call_1".to_string(),
function: OpenAIFunction { name: "t".to_string(), arguments: arguments.to_string() },
r#type: "function".to_string(),
extra_content: None,
};
assert!(interpolating_args(&call(r#"{"x":"$var:u/a/s"}"#)));
assert!(interpolating_args(&call(
r#"{"x":{"y":["\u0024res:u/a/r"]}}"#
)));
assert!(!interpolating_args(&call(r#"{"x":"$5 and var: 3"}"#)));
}
#[test]
fn extracts_only_the_output_of_an_agent_result() {
let envelope = RawValue::from_string(
+5 -8
View File
@@ -13,6 +13,7 @@ use sqlx::types::Json;
use sqlx::{Pool, Postgres};
use tokio::process::Command;
use tokio::{fs::File, io::AsyncReadExt};
use windmill_common::variables::arg_needs_interpolation;
use windmill_common::flows::Step;
use windmill_common::global_settings::NSJAIL_TMP_BACKING_DISK;
@@ -173,15 +174,11 @@ pub async fn write_file_binary(dir: &str, path: &str, content: &[u8]) -> error::
Ok(file)
}
lazy_static::lazy_static! {
static ref RE_RES_VAR: Regex = Regex::new(r#"\$(?:var|jsonvar|res|encrypted)\:"#).unwrap();
}
/// Returns true if any value in `vs` contains a `$var:`/`$jsonvar:`/`$res:`/`$encrypted:`
/// reference that would need interpolation by `transform_json`. Cheap pre-check that
/// callers can use to skip the DB roundtrip + clone path when nothing requires resolution.
pub(crate) fn map_needs_resolution(vs: &HashMap<String, Box<RawValue>>) -> bool {
vs.values().any(|v| (*RE_RES_VAR).is_match(v.get()))
vs.values().any(|v| arg_needs_interpolation(v.get()))
}
pub async fn transform_json<'a>(
@@ -194,7 +191,7 @@ pub async fn transform_json<'a>(
let mut has_match = false;
for (_, v) in vs {
let inner_vs = v.get();
if (*RE_RES_VAR).is_match(inner_vs) {
if arg_needs_interpolation(inner_vs) {
has_match = true;
break;
}
@@ -205,7 +202,7 @@ pub async fn transform_json<'a>(
let mut r = HashMap::new();
for (k, v) in vs {
let inner_vs = v.get();
if (*RE_RES_VAR).is_match(inner_vs) {
if arg_needs_interpolation(inner_vs) {
let value = serde_json::from_str(inner_vs).map_err(|e| {
error::Error::internal_err(format!("Error while parsing inner arg: {e:#}"))
})?;
@@ -232,7 +229,7 @@ pub async fn transform_json_as_values<'a>(
let mut r: HashMap<String, serde_json::Value> = HashMap::new();
for (k, v) in vs {
let inner_vs = v.get();
if (*RE_RES_VAR).is_match(inner_vs) {
if arg_needs_interpolation(inner_vs) {
let value = serde_json::from_str(inner_vs).map_err(|e| {
error::Error::internal_err(format!("Error while parsing inner arg: {e:#}"))
})?;
+67 -2
View File
@@ -20,6 +20,11 @@ import { Resource } from "../../../gen/types.gen.ts";
import { readInlinePathSync, readTextFile } from "../../utils/utils.ts";
import { isWorkspaceSpecificFile } from "../../core/specific_items.ts";
import { getCurrentGitBranch } from "../../utils/git.ts";
import {
buildPermissionedAsContext,
type PermissionedAsContext,
} from "../../core/permissioned_as.ts";
import { readEffectiveSyncBehavior } from "../../core/conf.ts";
export interface ResourceFile {
value: any;
@@ -83,6 +88,7 @@ export async function pushResource(
// only round-trip the canonical layout); the standalone `resource push`
// command pushes a single explicit file, where any pointer is fine.
enforceCanonicalFileset?: boolean,
permissionedAsContext?: PermissionedAsContext,
): Promise<void> {
remotePath = removeType(remotePath, "resource");
try {
@@ -93,6 +99,11 @@ export async function pushResource(
} catch {
// flow doesn't exist
}
const claimed = agentOnBehalfOf(
resource,
localResource,
permissionedAsContext
);
// Helper function to resolve inline content
const resolveInlineContent = async () => {
@@ -133,11 +144,23 @@ export async function pushResource(
// Only resolve inline content if we're actually updating
await resolveInlineContent();
if (claimed) {
log.info(`Preserving ${claimed} as permissioned_as for agent ${remotePath}`);
}
await wmill.updateResource({
workspace: workspace,
path: remotePath.replaceAll(SEP, "/"),
requestBody: { ...localResource, ...(wsSpecific !== undefined ? { ws_specific: wsSpecific } : {}) },
requestBody: {
...localResource,
...(wsSpecific !== undefined ? { ws_specific: wsSpecific } : {}),
...(claimed
? {
value: { ...localResource.value, on_behalf_of: claimed },
preserve_on_behalf_of: true,
}
: {}),
},
});
} else {
// New resource - resolve inline content
@@ -163,6 +186,42 @@ export async function pushResource(
}
}
/**
* An agent with `run_on_behalf_of` runs as the identity in its value's
* `on_behalf_of`, which the backend sets to whoever writes the agent and clears
* while the toggle is off. Like an app's, it is never taken from the
* tracked file, which the pull leaves it out of: the deployed one is claimed
* back only by a pusher the backend lets keep it, an admin or a `wm_deployers`
* member. Strips the key from `localResource` (and from the deployed value, so
* an unchanged agent still compares equal) and returns the identity to claim.
*/
export function agentOnBehalfOf(
deployed: ResourceFile | Resource | undefined,
localResource: ResourceFile,
permissionedAsContext: PermissionedAsContext | undefined
): string | undefined {
// What the push writes is the tracked file's type: an agent retyped into
// another resource keeps its value exactly as the file states it.
if (localResource.resource_type !== "ai_agent") {
return undefined;
}
const local = localResource.value;
if (local && typeof local === "object") {
delete local.on_behalf_of;
}
const value = deployed?.resource_type === "ai_agent" ? deployed.value : undefined;
const onBehalfOf =
value && typeof value === "object" ? value.on_behalf_of : undefined;
if (deployed && onBehalfOf !== undefined) {
const { on_behalf_of: _, ...rest } = value;
deployed.value = rest;
}
if (!onBehalfOf || !permissionedAsContext?.userIsAdminOrDeployer) {
return undefined;
}
return onBehalfOf;
}
type PushOptions = GlobalOptions;
async function push(opts: PushOptions, filePath: string, remotePath: string) {
const workspace = await resolveWorkspace(opts);
@@ -184,7 +243,13 @@ async function push(opts: PushOptions, filePath: string, remotePath: string) {
remotePath,
undefined,
parseFromFile(filePath),
filePath // Pass the local file path for branch-specific inline content resolution
filePath, // Pass the local file path for branch-specific inline content resolution
undefined,
undefined,
await buildPermissionedAsContext(
workspace.workspaceId,
await readEffectiveSyncBehavior(opts, workspace)
)
);
log.info(colors.bold.underline.green(`Resource ${remotePath} pushed`));
}
+1
View File
@@ -6065,6 +6065,7 @@ export async function push(
resourceFilePath,
isFileResWsSpecific ? true : undefined,
true,
permissionedAsContext,
);
}
// Already-synced parents got the full content this run.
+12
View File
@@ -250,6 +250,18 @@ export async function preCheckPermissionedAs(
if (match) {
currentOwner = match[1];
}
} else if (typeStr === "resource") {
// An agent with `run_on_behalf_of` runs as someone, and the file never says who.
if (
/["']?resource_type["']?\s*:\s*["']?ai_agent\b/.test(beforeContent) &&
/["']?run_on_behalf_of["']?\s*:\s*true\b/.test(beforeContent)
) {
wouldChangeItems.push({
path: change.path,
currentOwner: "(agent owner)",
});
}
continue;
} else if (typeStr.endsWith("_trigger")) {
wouldChangeItems.push({
path: change.path,
+1 -1
View File
@@ -254,7 +254,7 @@ export async function pushObj(
} else if (typeEnding === "resource") {
if (!alreadySynced.includes(p)) {
alreadySynced.push(p);
await pushResource(workspace, p, befObj, newObj, originalLocalPath || p, wsSpecific, true);
await pushResource(workspace, p, befObj, newObj, originalLocalPath || p, wsSpecific, true, permissionedAsContext);
}
} else if (typeEnding === "resource-type") {
await pushResourceType(workspace, p, befObj, newObj);
+25
View File
@@ -105,3 +105,28 @@ test("deployItem: never sends the source workspace's on_behalf_of", async () =>
expect("on_behalf_of" in JSON.parse(JSON.stringify(identity))).toBe(false);
}
});
// An agent carries its principal inside its resource value, and is handed the chosen one as a
// principal. The source's must never be sent: without a choice the key is dropped, leaving the
// backend to make the agent the deployer's.
test("deployItem: an agent takes the chosen principal, never the source's", async () => {
const captured: any[] = [];
const provider = {
existsResource: async () => true,
getResource: async () => ({
path: "f/x/agent",
resource_type: "ai_agent",
description: "",
value: { system_prompt: "hi", on_behalf_of: "u/alice" },
}),
updateResource: async (p: any) => void captured.push(p.requestBody),
} as any;
await deployItem(provider, "resource" as any, "f/x/agent", "src", "dst", "u/bob");
await deployItem(provider, "resource" as any, "f/x/agent", "src", "dst", undefined);
expect(captured[0].value).toEqual({ system_prompt: "hi", on_behalf_of: "u/bob" });
expect(captured[0].preserve_on_behalf_of).toBe(true);
expect(captured[1].value).toEqual({ system_prompt: "hi" });
expect(captured[1].preserve_on_behalf_of).toBe(false);
});
@@ -145,3 +145,16 @@ test("an app is listed once however many of its files changed", async () => {
expect(message).toContain("2 item(s)");
});
// An agent has an identity only while `run_on_behalf_of` is on, so only then
// can a push take it over.
test("an agent warns only while it runs on behalf of someone", async () => {
const agent = (run: boolean) => ({
name: "edited" as const,
path: "f/test/bot.resource.yaml",
before: `resource_type: ai_agent\nvalue:\n system_prompt: hi\n run_on_behalf_of: ${run}\n`,
});
expect(await precheck([agent(true)])).toContain("f/test/bot.resource.yaml");
expect(await precheck([agent(false)])).toBeUndefined();
});
@@ -0,0 +1,57 @@
/**
* An agent's run-as identity lives in its value, which the pull leaves out. A
* push must neither see that gap as a change nor let the file pick who the
* agent runs as: only the deployed identity is claimed back, and only by a
* pusher the backend lets keep it.
*/
import { expect, test } from "bun:test";
import { isSuperset } from "../src/types.ts";
import { agentOnBehalfOf } from "../src/commands/resource/resource.ts";
const ctx = (userIsAdminOrDeployer: boolean) => ({
userCache: new Map(),
userIsAdminOrDeployer,
userEmail: "me@x.dev",
});
function agents(localObo?: string) {
const deployed = {
resource_type: "ai_agent",
value: { system_prompt: "hi", on_behalf_of: "u/bob" },
} as any;
const local = {
resource_type: "ai_agent",
value: { system_prompt: "hi", ...(localObo ? { on_behalf_of: localObo } : {}) },
} as any;
return { deployed, local };
}
test("an unchanged agent still compares equal to the deployed one", () => {
const { deployed, local } = agents();
agentOnBehalfOf(deployed, local, ctx(true));
expect(isSuperset(local, deployed)).toBe(true);
});
test("an admin or deployer claims the deployed identity, never the file's", () => {
const { deployed, local } = agents("u/admin");
expect(agentOnBehalfOf(deployed, local, ctx(true))).toBe("u/bob");
expect(local.value.on_behalf_of).toBeUndefined();
});
test("anyone else claims nothing", () => {
const { deployed, local } = agents("u/bob");
expect(agentOnBehalfOf(deployed, local, ctx(false))).toBeUndefined();
expect(agentOnBehalfOf(agents().deployed, agents().local, undefined)).toBeUndefined();
expect(local.value.on_behalf_of).toBeUndefined();
});
test("other resources are left alone, an agent retyped into one included", () => {
for (const deployedType of ["json", "ai_agent"]) {
const deployed = { resource_type: deployedType, value: { on_behalf_of: "u/bob" } } as any;
const local = { resource_type: "json", value: { on_behalf_of: "u/x" } } as any;
expect(agentOnBehalfOf(deployed, local, ctx(true))).toBeUndefined();
expect(local.value.on_behalf_of).toBe("u/x");
expect(deployed.value.on_behalf_of).toBe("u/bob");
}
});
+27 -3
View File
@@ -637,15 +637,26 @@ export async function deployItem(
workspace: workspaceFrom,
path,
});
let value = resource.value ?? "";
// An agent's value carries the principal it runs as, as a trigger's row does: the source's
// names nobody in the target, so it is replaced by the chosen one or dropped, leaving the
// backend to make it the deployer's.
const agent = resource.resource_type === "ai_agent" && typeof value === "object";
if (agent) {
const { on_behalf_of: _, ...rest } = value;
value = onBehalfOf ? { ...rest, on_behalf_of: onBehalfOf } : rest;
}
const agentPreserve = agent ? { preserve_on_behalf_of: preserveOnBehalfOf } : {};
if (alreadyExists) {
await provider.updateResource({
workspace: workspaceTo,
path,
requestBody: {
path,
value: resource.value ?? "",
value,
description: resource.description ?? "",
labels: resource.labels,
...agentPreserve,
},
});
} else {
@@ -653,10 +664,11 @@ export async function deployItem(
workspace: workspaceTo,
requestBody: {
path,
value: resource.value ?? "",
value,
resource_type: resource.resource_type,
description: resource.description ?? "",
labels: resource.labels,
...agentPreserve,
},
});
}
@@ -917,6 +929,11 @@ export async function getItemValue(
return variable.value;
} else if (kind === "resource") {
const resource = await provider.getResource({ workspace, path });
// Each workspace resolves its own agent identity, which a deploy never copies.
if (resource.resource_type === "ai_agent" && resource.value) {
const { on_behalf_of: _, ...value } = resource.value;
return value;
}
return resource.value;
} else if (kind === "resource_type") {
const rt = await provider.getResourceType({ workspace, path });
@@ -1002,7 +1019,8 @@ function stripTriggerOrScheduleRuntimeFields(row: unknown): unknown {
/**
* Fetch the on_behalf_of value for a deployable item.
* Returns an email for flows/scripts/apps, or undefined if not applicable.
* Returns an email for flows/scripts/apps, a principal (u/username, g/group) for
* triggers/schedules and agents, or undefined if not applicable.
*/
export async function getOnBehalfOf(
provider: DeployProvider,
@@ -1020,6 +1038,12 @@ export async function getOnBehalfOf(
} else if (kind === "app" || kind === "raw_app") {
const app = await provider.getAppByPath({ workspace, path });
return app.policy?.on_behalf_of_email;
} else if (kind === "resource") {
// Only an agent runs as anyone, and like a trigger it records the principal.
const resource = await provider.getResource({ workspace, path });
return resource.resource_type === "ai_agent"
? resource.value?.on_behalf_of
: undefined;
} else if (kind === "schedule") {
const schedule = await provider.getSchedule({ workspace, path });
return schedule.permissioned_as;
@@ -612,12 +612,13 @@
}
async function fetchOnBehalfOfInfo(diffs: WorkspaceItemDiff[]) {
// Runnables carry an email; triggers/schedules carry `permissioned_as`.
// `getOnBehalfOf` handles the dispatch — we just need to skip kinds that
// have no on_behalf_of concept (resource, variable, folder, resource_type).
// Runnables carry an email; triggers/schedules and agents carry `permissioned_as`.
// `getOnBehalfOf` handles the dispatch (a resource that is not an agent has none) — we
// just need to skip kinds that have no on_behalf_of concept (variable, folder, resource_type).
const itemsWithOnBehalfOf = diffs.filter(
(d) =>
['flow', 'script', 'app', 'raw_app'].includes(d.kind) || isTriggerOrScheduleKind(d.kind)
['flow', 'script', 'app', 'raw_app', 'resource'].includes(d.kind) ||
isTriggerOrScheduleKind(d.kind)
)
const pair = [
{ workspace: currentWorkspaceId, side: 'fork' as const },
@@ -679,17 +680,22 @@
})
)
/** Triggers, schedules and agents record who they run as by principal, the rest by address. */
function usesPrincipal(kind: Kind): boolean {
return kind === 'trigger' || isTriggerOrScheduleKind(kind) || kind === 'resource'
}
/**
* Get the on_behalf_of value for deployment based on user's choice.
* Returns an email for flows/scripts/apps, or permissioned_as (u/username, g/group) for triggers/schedules.
* Returns an email for flows/scripts/apps, or permissioned_as (u/username, g/group) for
* triggers/schedules and agents.
*/
function getOnBehalfOfForDeploy(itemKey: string, kind: Kind): string | undefined {
const choice = onBehalfOfChoice[itemKey]
if (choice === 'target') return getTargetOnBehalfOf(itemKey)
if (choice === 'custom') {
const details = customOnBehalfOf[itemKey]
const wantsPermissionedAs = kind === 'trigger' || isTriggerOrScheduleKind(kind)
return wantsPermissionedAs ? details?.permissionedAs : details?.email
return usesPrincipal(kind) ? details?.permissionedAs : details?.email
}
// 'me' or undefined = don't pass, backend will use deploying user's identity
return undefined
@@ -701,7 +707,7 @@
* returning undefined clears the source item's value rather than keeping it.
*/
function getOnBehalfOfPermissionedAsForDeploy(itemKey: string, kind: Kind): string | undefined {
if (kind === 'trigger' || isTriggerOrScheduleKind(kind)) return undefined
if (usesPrincipal(kind)) return undefined
if (onBehalfOfChoice[itemKey] !== 'custom') return undefined
return customOnBehalfOf[itemKey]?.permissionedAs
}
@@ -110,16 +110,22 @@
return needsOnBehalfOfSelection(kind, sourceOnBehalfOfInfo[statusPath])
}
/** Triggers and agents record who they run as by principal, the rest by address. */
function usesPrincipal(kind: Kind): boolean {
return kind === 'trigger' || kind === 'resource'
}
/**
* Get the on_behalf_of value for deployment based on user's choice.
* Returns an email for flows/scripts/apps, or permissioned_as (u/username, g/group) for triggers/schedules.
* Returns an email for flows/scripts/apps, or permissioned_as (u/username, g/group) for
* triggers/schedules and agents.
*/
function getOnBehalfOfForDeploy(statusPath: string, kind: Kind): string | undefined {
const choice = onBehalfOfChoice[statusPath]
if (choice === 'target') return targetOnBehalfOfInfo[statusPath]
if (choice === 'custom') {
const details = customOnBehalfOf[statusPath]
return kind === 'trigger' ? details?.permissionedAs : details?.email
return usesPrincipal(kind) ? details?.permissionedAs : details?.email
}
// 'me' or undefined = don't pass, backend will use deploying user's identity
return undefined
@@ -134,7 +140,7 @@
statusPath: string,
kind: Kind
): string | undefined {
if (kind === 'trigger' || onBehalfOfChoice[statusPath] !== 'custom') return undefined
if (usesPrincipal(kind) || onBehalfOfChoice[statusPath] !== 'custom') return undefined
return customOnBehalfOf[statusPath]?.permissionedAs
}
@@ -205,7 +211,7 @@
// Fetch on_behalf_of_email for flows, scripts, apps, and triggers from both workspaces
for (const dep of sortedSet.filter((d) =>
['flow', 'script', 'app', 'trigger'].includes(d.kind)
['flow', 'script', 'app', 'trigger', 'resource'].includes(d.kind)
)) {
const key = computeStatusPath(dep.kind, dep.path)
let source: string | undefined
@@ -19,7 +19,9 @@
kind === 'app' ||
kind === 'raw_app' ||
kind === 'trigger' ||
isTriggerOrScheduleKind(kind)
isTriggerOrScheduleKind(kind) ||
// Only an agent's value carries an identity, so no other resource has a source value.
kind === 'resource'
if (!supported) return false
return !!sourceValue
}
@@ -652,6 +652,11 @@
const s = states[ws].draft!
const ini = initialStates[ws]
if (existedInitially[ws]) {
const fetched = fetchedResources[ws]
// An agent keeps the identity its value names, as its own editor keeps it: without
// the flag an admin's save would take it over. The backend resets it to the writer
// for anyone not allowed to keep it.
const keepAgentObo = fetched?.resource_type === 'ai_agent' && !!s.args?.on_behalf_of
await ResourceService.updateResource({
workspace: ws,
path: ini.path,
@@ -660,10 +665,10 @@
value: s.args,
description: s.description,
labels: s.labels,
ws_specific: s.wsSpecific
ws_specific: s.wsSpecific,
preserve_on_behalf_of: keepAgentObo ? true : undefined
}
})
const fetched = fetchedResources[ws]
if (fetched?.resource_type === 'json_schema') {
clearJsonSchemaResourceCache(ini.path, ws)
}
@@ -501,7 +501,8 @@ export const resourceRequestSchema = z.object({
"description": z.string().describe("The description of the resource").optional(),
"resource_type": z.string().describe("The resource_type associated with the resource"),
"labels": z.array(z.string()).optional(),
"ws_specific": z.boolean().optional()
"ws_specific": z.boolean().optional(),
"preserve_on_behalf_of": z.boolean().describe("ai_agent only. Keep the identity `value.on_behalf_of` names (`u/<username>`, `g/<group>`) rather than the caller's own. Honoured for admins and members of wm_deployers; otherwise every write of an agent sets it to the caller.").optional()
})
export const triggerRequestSchemas = {
@@ -8,7 +8,7 @@
onBehalfOf: string | undefined
/** Address of the account `onBehalfOf` names, derived from it on the read paths. */
onBehalfOfEmail: string | undefined
kind: 'script' | 'flow'
kind: 'script' | 'flow' | 'agent'
}
let { onBehalfOf, onBehalfOfEmail, kind }: Props = $props()
@@ -25,7 +25,12 @@
{#if onBehalfOf}
<Tooltip>
{#snippet text()}
Every run of this {kind} is permissioned as {detailed}, whoever starts it.
{#if kind === 'agent'}
Runs of this agent from its page and the API are permissioned as {detailed}, whoever
starts them.
{:else}
Every run of this {kind} is permissioned as {detailed}, whoever starts it.
{/if}
{/snippet}
<Badge color="violet" icon={{ icon: UserCog, position: 'left' }}>
<span class="truncate max-w-40">On behalf of {onBehalfOf}</span>
@@ -144,18 +144,29 @@ export function agentDraftDeployRefusal(
*/
type AgentWriteResult = { ok: true } | { ok: false; error: string; notAnAgent?: true }
/** Who a deploy makes the agent run as. The server resolves it as it does a flow's: without
* `preserve`, or from someone not allowed to keep another identity, it is the deployer. */
export interface AgentRunAs {
permissionedAs: string | undefined
preserve: boolean
}
async function writeAgentResource(
workspace: string,
fromPath: string,
state: AgentResourceState,
noDeployed: boolean
noDeployed: boolean,
runAs?: AgentRunAs
): Promise<AgentWriteResult> {
const body = {
path: state.path,
value: state.args,
value: runAs?.permissionedAs
? { ...state.args, on_behalf_of: runAs.permissionedAs }
: state.args,
description: state.description,
labels: state.labels,
ws_specific: state.wsSpecific
ws_specific: state.wsSpecific,
preserve_on_behalf_of: runAs?.preserve || undefined
}
try {
if (noDeployed) {
@@ -221,9 +232,12 @@ export interface AgentDraftHandle {
/** Why this path cannot be edited here, if it cannot. Render it instead of the form. */
readonly refusal: string | undefined
readonly sync: TriggerDraftSync
/** Who the deployed agent runs as when run from its page; undefined when nothing is deployed
* or it runs as whoever runs it. */
readonly onBehalfOf: string | undefined
/** Write the current state to the resource and drop the draft. Resolves to the path written,
* which differs from the one loaded when the draft renames the agent, or undefined on failure. */
deploy: () => Promise<string | undefined>
deploy: (runAs?: AgentRunAs) => Promise<string | undefined>
}
/**
@@ -237,6 +251,7 @@ export function useAgentDraft(opts: AgentDraftOptions): AgentDraftHandle {
let loading = $state(true)
let noDeployed = $state(false)
let canWriteResource = $state(true)
let onBehalfOf = $state<string | undefined>(undefined)
/** Guards the load against a path that changed under a slow response. */
let loadedFor = $state<string | undefined>(undefined)
/** Why the loaded path cannot be edited here, if it cannot. Gates the sync for as long as that
@@ -324,10 +339,15 @@ export function useAgentDraft(opts: AgentDraftOptions): AgentDraftHandle {
return
}
noDeployed = Boolean((r as any).no_deployed)
// Who it runs as is the deploy's to decide, not a setting the form or a draft holds.
const { on_behalf_of, ...args } = (r.value ?? {}) as AIAgentConfig & {
on_behalf_of?: string
}
onBehalfOf = noDeployed ? undefined : on_behalf_of
const deployedState: AgentResourceState = {
path: r.path,
description: r.description ?? '',
args: (r.value ?? {}) as AIAgentConfig,
args,
// Only where nothing else answers for the type: with a deployed row both the
// create below and the review page's `deployDraft` read it from there, and
// carrying it would make every draft the generic resource editor writes — which
@@ -414,7 +434,7 @@ export function useAgentDraft(opts: AgentDraftOptions): AgentDraftHandle {
})
})
async function deploy(): Promise<string | undefined> {
async function deploy(runAs?: AgentRunAs): Promise<string | undefined> {
const ws = opts.workspace()
const fromPath = opts.path()
const s = state
@@ -430,7 +450,7 @@ export function useAgentDraft(opts: AgentDraftOptions): AgentDraftHandle {
// made during the request as saved, and the banner would clear on a value the server never
// received; against the snapshot it stays a draft, which is what it is.
const submitted = structuredClone($state.snapshot(s)) as AgentResourceState
const written = await writeAgentResource(ws, fromPath, submitted, noDeployed)
const written = await writeAgentResource(ws, fromPath, submitted, noDeployed, runAs)
if (!written.ok) {
// A path that is no longer an agent tears this editor down; anything else is a plain error
// the user can retry from the form as it stands.
@@ -449,6 +469,10 @@ export function useAgentDraft(opts: AgentDraftOptions): AgentDraftHandle {
const firstSave = noDeployed
deployed = submitted
noDeployed = false
// What the server resolved, which a request to keep an identity does not decide alone.
void ResourceService.getResource({ workspace: ws, path: submitted.path })
.then((r) => (onBehalfOf = (r.value as { on_behalf_of?: string } | undefined)?.on_behalf_of))
.catch(() => {})
const renamed = submitted.path !== fromPath
// Only when the form still holds exactly what was sent. `discard` resets the handle's cell to
// what it is given, and the apply-effect copies that back over the form: against an edit made
@@ -490,6 +514,9 @@ export function useAgentDraft(opts: AgentDraftOptions): AgentDraftHandle {
get refusal() {
return refusal
},
get onBehalfOf() {
return onBehalfOf
},
sync,
deploy
}
@@ -118,6 +118,8 @@ export interface AIAgentConfig {
temperature?: number
max_iterations?: number
tools?: AgentTool[]
/** Runs from the agent's page take its `on_behalf_of` rather than the caller's identity. */
run_on_behalf_of?: boolean
}
/** Extract the static brain values from a step's input_transforms into a flat agent config. */
@@ -50,7 +50,7 @@
keepsManagedMemory
} from '../agentFormFields'
import { toolDisplayName, type AgentTool } from '../agentToolUtils'
import { useAgentDraft } from '../agentDraft.svelte'
import { useAgentDraft, type AgentRunAs } from '../agentDraft.svelte'
import { sendUserToast } from '$lib/toast'
interface Props {
@@ -359,8 +359,8 @@
inputs?: Record<string, any>
): Promise<string | undefined> {
if (!agentModule) return undefined
// A view runs the deployed agent, which anyone who can read it may do; the editor previews
// the draft. Both run as the caller.
// A view runs the deployed agent, which anyone who can read it may do, as the server picks
// the identity; the editor previews the draft as the one editing it.
if (view && workspace) {
return await JobService.runAgent({
workspace,
@@ -456,12 +456,21 @@
pathErrorValue = error ?? ''
}
/** Who the next deploy makes the agent run as, as the "Permissioned as" line picked it. Unset
* while the line was never shown, which keeps the current identity as the line would have
* preselected it: an admin deploying without looking must not take the agent over. */
let runAs = $state<AgentRunAs | undefined>(undefined)
export function setRunAs(next: AgentRunAs) {
runAs = next
}
export function deploy(): Promise<boolean> {
if (pathErrorValue) {
sendUserToast(`Cannot deploy the agent: ${pathErrorValue}`, true)
return Promise.resolve(false)
}
return draft.deploy().then(async (written) => {
const keep = draft.onBehalfOf ? { permissionedAs: draft.onBehalfOf, preserve: true } : undefined
return draft.deploy(runAs ?? keep).then(async (written) => {
// The path the write landed on, which a rename moves off the one this editor opened.
if (written) await onSaved?.(written)
return written !== undefined
@@ -582,6 +582,7 @@
path={target?.path ?? ''}
workspace={ws}
bind:error={() => host?.pathError(), (error) => host?.setPathError(error)}
onRunAsChange={(runAs) => host?.setRunAs(runAs)}
/>
{/if}
{/snippet}
@@ -7,8 +7,9 @@
import ResourceDescriptionField from '$lib/components/ResourceDescriptionField.svelte'
import ResourcePathHint from '$lib/components/ResourcePathHint.svelte'
import Toggle from '$lib/components/Toggle.svelte'
import PermissionedAsLine from '$lib/components/triggers/PermissionedAsLine.svelte'
import { WorkspaceService } from '$lib/gen'
import type { AgentDraftHandle } from '../agentDraft.svelte'
import type { AgentDraftHandle, AgentRunAs } from '../agentDraft.svelte'
/** An agent's own settings, laid out as the top of the resource editor: what it is saved as
* rather than what it does. */
@@ -19,9 +20,11 @@
workspace: string | undefined
/** Why the path cannot be deployed to, if it cannot. */
error?: string
/** Who the next deploy makes the agent run as, as picked here. */
onRunAsChange?: (runAs: AgentRunAs) => void
}
let { draft, path, workspace, error = $bindable() }: Props = $props()
let { draft, path, workspace, error = $bindable(), onRunAsChange = undefined }: Props = $props()
let readOnly = $derived(!draft.canWrite)
// Only a workspace that deploys somewhere has anything to keep an agent out of.
@@ -64,6 +67,33 @@
{:else}
<LabelsInput bind:labels={draft.state.labels} {workspace} class="-mt-4" />
{/if}
<!-- The identity exists only while the toggle is on, resolved on deploy as a flow's
on-behalf-of identity is. -->
<div class="flex flex-col gap-2">
<Label
label="Run on behalf of"
tooltip="Runs from the agent's page and the API take the permissions of whoever deploys it, or of the user an admin picks. Otherwise they take the caller's."
>
<Toggle
bind:checked={
() => draft.state?.args.run_on_behalf_of ?? false,
(v) => {
if (draft.state) draft.state.args.run_on_behalf_of = v || undefined
}
}
disabled={readOnly}
/>
</Label>
{#if draft.state.args.run_on_behalf_of}
<PermissionedAsLine
permissionedAs={draft.onBehalfOf}
path={draft.state.path}
disabled={readOnly}
onPermissionedAsChange={(permissionedAs, preserve) =>
onRunAsChange?.({ permissionedAs, preserve })}
/>
{/if}
</div>
{#if deployTo.current}
<Label label="Workspace specific" tooltip="Keeps this agent out of deploys to prod/staging.">
<Toggle bind:checked={draft.state.wsSpecific} disabled={readOnly} />
@@ -122,6 +122,25 @@ describe('deployDraft preserves on_behalf_of', () => {
})
)
})
it('agent: claims the deployed identity, which its draft never carries', async () => {
vi.mocked(ResourceService.getResource).mockResolvedValueOnce({
path: 'f/support/triage_agent',
resource_type: 'ai_agent',
value: { system_prompt: 'deployed', on_behalf_of: 'u/alice' },
draft: { path: 'f/support/triage_agent', args: { system_prompt: 'drafted' } }
} as any)
expect(await deployDraft('resource', 'f/support/triage_agent', 'ws')).toEqual({ success: true })
expect(ResourceService.updateResource).toHaveBeenCalledWith(
expect.objectContaining({
requestBody: expect.objectContaining({
value: { system_prompt: 'drafted', on_behalf_of: 'u/alice' },
preserve_on_behalf_of: true
})
})
)
})
})
// The backend skips the dependency job when `lock` is set, so a draft that inherited the
+14 -2
View File
@@ -234,8 +234,14 @@ function canonicalizeDraftDiffValue(kind: DraftKind, raw: any, isDraft: boolean)
if (kind === 'resource') {
// draft: { args, description, resource_type, labels, wsSpecific }
// deployed row: { value, description, resource_type, labels, ws_specific }
// An agent's identity is in its deployed value and never in a draft, which is not an edit.
let value = (isDraft ? raw.args : raw.value) ?? {}
if (raw.resource_type === 'ai_agent') {
const { on_behalf_of: _, ...rest } = value
value = rest
}
return {
value: (isDraft ? raw.args : raw.value) ?? {},
value,
description: raw.description ?? '',
resource_type: raw.resource_type ?? undefined,
labels: raw.labels ?? undefined,
@@ -658,12 +664,18 @@ export async function deployDraft(
}
})
} else {
// As for an app: an agent draft carries no identity, so the deployed one is kept (the
// backend resets it to the deploying user without the flag, gated by
// can_preserve_on_behalf_of).
const agentObo =
deployed.resource_type === 'ai_agent' ? deployed.value?.on_behalf_of : undefined
await ResourceService.updateResource({
workspace,
path,
requestBody: {
path: d.path ?? path,
value: d.args ?? {},
value: agentObo ? { ...(d.args ?? {}), on_behalf_of: agentObo } : (d.args ?? {}),
preserve_on_behalf_of: agentObo ? true : undefined,
description: d.description ?? '',
labels: d.labels,
ws_specific: d.wsSpecific
+42 -3
View File
@@ -196,6 +196,11 @@ export type DeployConflict = { hit: boolean }
* A group kept as the target identity travels as its synthetic `group-*@windmill.dev`
* address, which an admin-created account holding it would win on the backend: known and
* accepted, see `users::permissioned_as_from_email`.
*
* An agent carries its principal inside its resource value, and is handed the chosen one as a
* principal, as a trigger is. Every read here drops the source's, so a diff never shows it and
* the published `deployItem` copies none; the write then stamps the chosen one, or none, which
* leaves the backend to make it the deployer's.
*/
function makeProvider(
onBehalfOfPrincipal?: string,
@@ -224,6 +229,17 @@ function makeProvider(
}
}
: app
const agents = new Set<string>()
const withAgentIdentity = <T extends Record<string, any>>(requestBody: T, path: string): T =>
agents.has(path) && requestBody.value && typeof requestBody.value === 'object'
? {
...requestBody,
value: onBehalfOf
? { ...requestBody.value, on_behalf_of: onBehalfOf }
: requestBody.value,
preserve_on_behalf_of: onBehalfOf !== undefined
}
: requestBody
const refuseUpdate = (): never => {
if (conflict) conflict.hit = true
throw new Error('item already exists in the target workspace')
@@ -290,11 +306,23 @@ function makeProvider(
updateVariable: (p) =>
VariableService.updateVariable({ ...p, requestBody: withLabels(p.requestBody) }),
deleteVariable: (p) => VariableService.deleteVariable(p),
getResource: (p) => ResourceService.getResource(p).then(readLabels),
getResource: async (p) => {
const resource = readLabels(await ResourceService.getResource(p))
if (resource.resource_type !== 'ai_agent' || !resource.value) return resource
agents.add(p.path)
const { on_behalf_of: _, ...value } = resource.value as Record<string, any>
return { ...resource, value }
},
createResource: (p) =>
ResourceService.createResource({ ...p, requestBody: withLabels(p.requestBody) }),
ResourceService.createResource({
...p,
requestBody: withAgentIdentity(withLabels(p.requestBody), p.requestBody.path)
}),
updateResource: (p) =>
ResourceService.updateResource({ ...p, requestBody: withLabels(p.requestBody) }),
ResourceService.updateResource({
...p,
requestBody: withAgentIdentity(withLabels(p.requestBody), p.path)
}),
deleteResource: (p) => ResourceService.deleteResource(p),
getResourceType: (p) => ResourceService.getResourceType(p),
createResourceType: (p) => ResourceService.createResourceType(p),
@@ -591,6 +619,17 @@ export async function getOnBehalfOf(
return undefined
}
}
if (kind === 'resource') {
// Only an agent runs as anyone, and like a trigger it records the principal.
try {
const resource = await ResourceService.getResource({ workspace, path })
return resource.resource_type === 'ai_agent'
? (resource.value as { on_behalf_of?: string } | undefined)?.on_behalf_of
: undefined
} catch {
return undefined
}
}
return sharedGetOnBehalfOf(makeProvider(), kind as DeployKind, path, workspace)
}
@@ -6,6 +6,7 @@
import { copilotInfo } from '$lib/aiStore'
import AgentEvalsModal from '$lib/components/flows/content/AgentEvalsModal.svelte'
import DetailPageHeader from '$lib/components/details/DetailPageHeader.svelte'
import OnBehalfOfBadge from '$lib/components/details/OnBehalfOfBadge.svelte'
import ToggleButtonGroup from '$lib/components/common/toggleButton-v2/ToggleButtonGroup.svelte'
import ToggleButton from '$lib/components/common/toggleButton-v2/ToggleButton.svelte'
import ConfirmationModal from '$lib/components/common/confirmationModal/ConfirmationModal.svelte'
@@ -156,6 +157,11 @@
: [])
]}
>
<OnBehalfOfBadge
onBehalfOf={config?.run_on_behalf_of ? agent?.onBehalfOf : undefined}
onBehalfOfEmail={undefined}
kind="agent"
/>
{#snippet leading_actions()}
<!-- Only an agent that keeps the conversation can chat: without managed memory every
message would be answered alone, so it is run from its inputs with nothing to switch. -->
+2
View File
@@ -412,6 +412,7 @@ impl Windmill {
resource_type: resource_type.to_owned(),
labels: None,
ws_specific: None,
preserve_on_behalf_of: None,
},
// Some(true),
None,
@@ -429,6 +430,7 @@ impl Windmill {
resource_type: Some(resource_type.to_owned()),
labels: None,
ws_specific: None,
preserve_on_behalf_of: None,
},
)
.await?;