fix: set WM_END_USER_EMAIL on direct script and flow runs (#11636)

* fix: set WM_END_USER_EMAIL on direct script and flow runs

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

* fix: pass the end user to workflow-as-code tasks

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

* docs: describe when WM_END_USER_EMAIL is set

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

* test: pin the end user a job token passes on

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

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

backend/tests/end_user_email.rs

@@ -1,10 +1,8 @@
 //! Tests for WM_END_USER_EMAIL environment variable.
 //!
 //! These tests verify that WM_END_USER_EMAIL is populated with the authenticated
-//! user's email when executing app components.
-//!
-//! TODO: Add tests for scripts and flows once public execution endpoints are identified.
-//! Currently only apps support non-workspace-member execution via OptAuthed + token lookup.
+//! user's email when executing app components and when running scripts and flows directly.
+//! Only apps support non-workspace-member execution via OptAuthed + token lookup.
 
 use serde_json::json;
 use sqlx::{Pool, Postgres};
@@ -27,40 +25,6 @@ fn authed(builder: reqwest::RequestBuilder, token: &str) -> reqwest::RequestBuil
     builder.header("Authorization", format!("Bearer {}", token))
 }
 
-// TODO: Script tests - need to identify public execution endpoints for non-workspace-members
-// async fn run_script(port: u16, token: &str) -> anyhow::Result<String> {
-//     let url = format!(
-//         "http://localhost:{}/api/w/test-workspace/jobs/run_wait_result/p/f/test/get_end_user_email",
-//         port
-//     );
-//     let resp = authed(client().post(&url), token)
-//         .json(&json!({}))
-//         .send()
-//         .await?;
-//     if !resp.status().is_success() {
-//         anyhow::bail!("script run failed: {} - {}", resp.status(), resp.text().await?);
-//     }
-//     Ok(resp.json::<serde_json::Value>().await?
-//         .as_str().unwrap_or("").to_string())
-// }
-
-// TODO: Flow tests - need to identify public execution endpoints for non-workspace-members
-// async fn run_flow(port: u16, token: &str) -> anyhow::Result<String> {
-//     let url = format!(
-//         "http://localhost:{}/api/w/test-workspace/jobs/run_wait_result/f/f/test/get_end_user_email_flow",
-//         port
-//     );
-//     let resp = authed(client().post(&url), token)
-//         .json(&json!({}))
-//         .send()
-//         .await?;
-//     if !resp.status().is_success() {
-//         anyhow::bail!("flow run failed: {} - {}", resp.status(), resp.text().await?);
-//     }
-//     Ok(resp.json::<serde_json::Value>().await?
-//         .as_str().unwrap_or("").to_string())
-// }
-
 /// Create an app with inline script via API
 async fn create_app_with_inline_script(port: u16, path: &str) -> anyhow::Result<()> {
     let url = format!("http://localhost:{}/api/w/test-workspace/apps/create", port);
@@ -312,41 +276,111 @@ async fn wait_for_job_result(port: u16, token: &str, job_id: &str) -> anyhow::Re
     anyhow::bail!("timeout waiting for job result")
 }
 
-// TODO: Script tests - need to identify public execution endpoints for non-workspace-members
-// #[cfg(feature = "deno_core")]
-// #[sqlx::test(fixtures("base", "end_user_email"))]
-// async fn test_script_wm_end_user_email(db: Pool<Postgres>) -> anyhow::Result<()> {
-//     initialize_tracing().await;
-//     set_jwt_secret().await;
-//     let server = ApiServer::start(db.clone()).await?;
-//     let port = server.addr.port();
-//
-//     in_test_worker(Connection::Sql(db.clone()), async move {
-//         let result = run_script(port, SAME_WS_TOKEN).await?;
-//         assert_eq!(result, SAME_WS_EMAIL, "same workspace user should get their email");
-//         Ok::<(), anyhow::Error>(())
-//     }, port).await?;
-//
-//     Ok(())
-// }
+async fn run_direct(port: u16, token: &str, route: &str) -> anyhow::Result<String> {
+    let url = format!(
+        "http://localhost:{}/api/w/test-workspace/jobs/run/{}",
+        port, route
+    );
+    let resp = authed(client().post(&url), token)
+        .json(&json!({}))
+        .send()
+        .await?;
+    if !resp.status().is_success() {
+        anyhow::bail!(
+            "run {route} failed: {} - {}",
+            resp.status(),
+            resp.text().await?
+        );
+    }
+    let job_id = resp.text().await?;
+    wait_for_job_result(port, token, &job_id).await
+}
 
-// TODO: Flow tests - need to identify public execution endpoints for non-workspace-members
-// #[cfg(feature = "deno_core")]
-// #[sqlx::test(fixtures("base", "end_user_email"))]
-// async fn test_flow_wm_end_user_email(db: Pool<Postgres>) -> anyhow::Result<()> {
-//     initialize_tracing().await;
-//     set_jwt_secret().await;
-//     let server = ApiServer::start(db.clone()).await?;
-//     let port = server.addr.port();
-//
-//     in_test_worker(Connection::Sql(db.clone()), async move {
-//         let result = run_flow(port, SAME_WS_TOKEN).await?;
-//         assert_eq!(result, SAME_WS_EMAIL, "same workspace user should get their email");
-//         Ok::<(), anyhow::Error>(())
-//     }, port).await?;
-//
-//     Ok(())
-// }
+#[cfg(feature = "deno_core")]
+#[sqlx::test(fixtures("base", "end_user_email"))]
+async fn test_direct_run_wm_end_user_email(db: Pool<Postgres>) -> anyhow::Result<()> {
+    initialize_tracing().await;
+    set_jwt_secret().await;
+    let server = ApiServer::start(db.clone()).await?;
+    let port = server.addr.port();
+
+    in_test_worker(
+        Connection::Sql(db.clone()),
+        async move {
+            let result = run_direct(port, SAME_WS_TOKEN, "p/f/test/get_end_user_email").await?;
+            assert_eq!(result, SAME_WS_EMAIL, "a script run should see its caller");
+
+            let result =
+                run_direct(port, SAME_WS_TOKEN, "f/f/test/get_end_user_email_flow").await?;
+            assert_eq!(
+                result, SAME_WS_EMAIL,
+                "every flow step should see the flow's caller"
+            );
+
+            Ok::<(), anyhow::Error>(())
+        },
+        port,
+    )
+    .await?;
+
+    Ok(())
+}
+
+/// A job's `WM_TOKEN` names the identity the job runs as, which differs from its end user when
+/// the job runs on behalf of another account. A run it starts must keep the job's end user.
+#[cfg(feature = "deno_core")]
+#[sqlx::test(fixtures("base", "end_user_email"))]
+async fn test_job_token_run_keeps_end_user_email(db: Pool<Postgres>) -> anyhow::Result<()> {
+    initialize_tracing().await;
+    set_jwt_secret().await;
+    let server = ApiServer::start(db.clone()).await?;
+    let port = server.addr.port();
+
+    let calling_job = uuid::Uuid::new_v4();
+    sqlx::query(
+        "INSERT INTO job_perms (job_id, email, username, is_admin, is_operator, folders, groups, workspace_id, end_user_email)
+         VALUES ($1, 'test2@windmill.dev', 'test-user-2', false, false, '{}', '{}', 'test-workspace', $2)",
+    )
+    .bind(calling_job)
+    .bind(SAME_WS_EMAIL)
+    .execute(&db)
+    .await?;
+    let wm_token = windmill_common::auth::create_jwt_token(
+        windmill_common::db::Authed {
+            email: "test2@windmill.dev".to_string(),
+            username: "test-user-2".to_string(),
+            is_admin: false,
+            is_operator: false,
+            groups: vec!["all".to_string()],
+            folders: vec![],
+            scopes: None,
+            token_prefix: None,
+        },
+        "test-workspace",
+        3600,
+        Some(calling_job),
+        None,
+        None,
+        None,
+    )
+    .await?;
+
+    in_test_worker(
+        Connection::Sql(db.clone()),
+        async move {
+            let result = run_direct(port, &wm_token, "p/f/test/get_end_user_email").await?;
+            assert_eq!(
+                result, SAME_WS_EMAIL,
+                "the calling job's end user, not its run-as identity"
+            );
+            Ok::<(), anyhow::Error>(())
+        },
+        port,
+    )
+    .await?;
+
+    Ok(())
+}
 
 #[cfg(feature = "deno_core")]
 #[sqlx::test(fixtures("base", "end_user_email"))]

backend/windmill-api-jobs/src/execution.rs

@@ -245,6 +245,27 @@ pub async fn unclaimable_run_lineage(
     Ok(referenced)
 }
 
+/// `WM_END_USER_EMAIL` of a run started through a `/jobs/run*` route. `WM_EMAIL` names the
+/// runnable's `on_behalf_of` identity when it has one, so this is the only place the caller's
+/// address reaches the job. A job's own `WM_TOKEN` passes on the end user of that job, and a run
+/// a trigger fired has no end user.
+pub async fn run_end_user_email(
+    db: &DB,
+    w_id: &str,
+    authed: &ApiAuthed,
+    trigger: Option<&TriggerMetadata>,
+) -> error::Result<Option<String>> {
+    if trigger.is_some() {
+        return Ok(None);
+    }
+    match authed.job_id {
+        Some(job_id) => Ok(windmill_common::auth::get_job_perms(db, &job_id, w_id)
+            .await?
+            .and_then(|p| p.end_user_email)),
+        None => Ok(Some(authed.email.clone())),
+    }
+}
+
 #[cfg(feature = "enterprise")]
 pub async fn check_license_key_valid() -> error::Result<()> {
     use windmill_common::ee_oss::LICENSE_KEY_VALID;
@@ -1013,6 +1034,7 @@ pub async fn run_flow<'c>(
         )
     };
 
+    let end_user_email = run_end_user_email(db, w_id, authed, trigger.as_ref()).await?;
     let scope_ceiling = windmill_api_auth::caller_scope_ceiling(db, authed).await?;
     let (uuid, mut tx) = push(
         &db,
@@ -1048,7 +1070,7 @@ pub async fn run_flow<'c>(
         None,
         push_authed.as_ref(),
         false,
-        None,
+        end_user_email,
         authed.trigger_or_fallback(trigger),
         run_query.suspended_mode,
         scope_ceiling.as_deref(),
@@ -1244,6 +1266,7 @@ pub async fn push_script_job_by_path_into_queue<'c>(
         )
     };
 
+    let end_user_email = run_end_user_email(&db, &w_id, &authed, trigger.as_ref()).await?;
     let scope_ceiling = windmill_api_auth::caller_scope_ceiling(&db, &authed).await?;
     let (uuid, tx) = push(
         &db,
@@ -1277,7 +1300,7 @@ pub async fn push_script_job_by_path_into_queue<'c>(
         },
         push_authed.as_ref(),
         false,
-        None,
+        end_user_email,
         authed.trigger_or_fallback(trigger),
         run_query.suspended_mode,
         scope_ceiling.as_deref(),

backend/windmill-api/src/jobs.rs

@@ -7826,6 +7826,7 @@ pub async fn run_workflow_as_code(
             )
         };
 
+    let end_user_email = run_end_user_email(&db, &w_id, &authed, None).await?;
     let scope_ceiling = windmill_api_auth::caller_scope_ceiling(&db, &authed).await?;
     let (uuid, mut tx) = push(
         &db,
@@ -7854,7 +7855,7 @@ pub async fn run_workflow_as_code(
         None,
         push_authed.as_ref(),
         false,
-        None,
+        end_user_email,
         None,
         None,
         scope_ceiling.as_deref(),
@@ -8141,6 +8142,7 @@ pub async fn run_wait_result_job_by_path_get(
             )
         };
 
+    let end_user_email = run_end_user_email(&db, &w_id, &authed, None).await?;
     let scope_ceiling = windmill_api_auth::caller_scope_ceiling(&db, &authed).await?;
     let (uuid, tx) = push(
         &db,
@@ -8169,7 +8171,7 @@ pub async fn run_wait_result_job_by_path_get(
         None,
         push_authed.as_ref(),
         false,
-        None,
+        end_user_email,
         authed.trigger_or_fallback(None),
         run_query.suspended_mode,
         scope_ceiling.as_deref(),
@@ -8291,6 +8293,7 @@ pub async fn run_wait_result_script_by_path_internal(
             )
         };
 
+    let end_user_email = run_end_user_email(&db, &w_id, &authed, None).await?;
     let scope_ceiling = windmill_api_auth::caller_scope_ceiling(&db, &authed).await?;
     let (uuid, tx) = push(
         &db,
@@ -8319,7 +8322,7 @@ pub async fn run_wait_result_script_by_path_internal(
         None,
         push_authed.as_ref(),
         false,
-        None,
+        end_user_email,
         authed.trigger_or_fallback(None),
         run_query.suspended_mode,
         scope_ceiling.as_deref(),
@@ -8409,6 +8412,7 @@ pub async fn run_wait_result_script_by_hash(
         )
     };
 
+    let end_user_email = run_end_user_email(&db, &w_id, &authed, None).await?;
     let scope_ceiling = windmill_api_auth::caller_scope_ceiling(&db, &authed).await?;
     let (uuid, tx) = push(
         &db,
@@ -8451,7 +8455,7 @@ pub async fn run_wait_result_script_by_hash(
         None,
         push_authed.as_ref(),
         false,
-        None,
+        end_user_email,
         authed.trigger_or_fallback(None),
         run_query.suspended_mode,
         scope_ceiling.as_deref(),
@@ -10556,6 +10560,7 @@ pub async fn run_job_by_hash_inner(
     };
     let job_payload = with_run_retry(&run_query, job_payload, &push_args, &tag)?;
 
+    let end_user_email = run_end_user_email(&db, &w_id, &authed, trigger.as_ref()).await?;
     let scope_ceiling = windmill_api_auth::caller_scope_ceiling(&db, &authed).await?;
     let (uuid, tx) = push(
         &db,
@@ -10584,7 +10589,7 @@ pub async fn run_job_by_hash_inner(
         None,
         push_authed.as_ref(),
         false,
-        None,
+        end_user_email,
         authed.trigger_or_fallback(trigger),
         run_query.suspended_mode,
         scope_ceiling.as_deref(),

backend/windmill-common/src/variables.rs

@@ -597,7 +597,7 @@ pub async fn get_reserved_variables(
     ContextualVariable {
         name: "WM_END_USER_EMAIL".to_string(),
         value: end_user_email.unwrap_or_else(|| "".to_string()),
-        description: "Email of the end user that executed the current script, propagated to flow steps. Only set when the run was triggered from an app (empty otherwise).".to_string(),
+        description: "Email of the end user that executed the current script, even when it runs on behalf of another identity, propagated to flow steps and child jobs. Set for app and API runs (for a webhook, the owner of the token it was called with), empty for runs fired by a schedule or by a trigger such as an HTTP route or Kafka.".to_string(),
         is_custom: false,
     },
     ContextualVariable {

backend/windmill-worker/src/bun_executor.rs

@@ -3273,9 +3273,9 @@ pub async fn handle_wac_v2_output(
                             .or(if own_runnable { None } else { job.priority }),
                         None,  // authed
                         false, // running
-                        None,  // end_user_email
-                        None,  // trigger
-                        None,  // suspended_mode
+                        job.permissioned_as_end_user_email.clone(),
+                        None, // trigger
+                        None, // suspended_mode
                         job.job_token_scopes.as_deref(),
                     )
                     .await?;