fix: add job_id, step_id and flow_path to native ts step errors (#12012)

* fix: add job_id, step_id and flow_path to every executor's step error

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

* fix: take the step error's flow_path from the pulled job, not a query

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

* fix: add step context to native ts errors in the executor instead of process_result

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

---------

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

backend/windmill-worker/src/bun_executor.rs

@@ -3170,7 +3170,8 @@ try {{
                     None,
                     has_stream,
                 )
-                .await?;
+                .await
+                .map_err(|e| crate::js_eval::with_step_context(e, &reserved_variables))?;
 
                 let preprocessed: HashMap<String, Box<RawValue>> =
                     serde_json::from_str(pre_result.get()).map_err(|e| {
@@ -3202,7 +3203,8 @@ try {{
                     stream_notifier,
                     has_stream,
                 )
-                .await?;
+                .await
+                .map_err(|e| crate::js_eval::with_step_context(e, &reserved_variables))?;
                 tracing::info!(
                     "Executed native code (with preprocessor) in {}ms",
                     started_at.elapsed().as_millis()
@@ -3230,7 +3232,8 @@ try {{
                 stream_notifier,
                 has_stream,
             )
-            .await?;
+            .await
+            .map_err(|e| crate::js_eval::with_step_context(e, &reserved_variables))?;
             tracing::info!(
                 "Executed native code in {}ms",
                 started_at.elapsed().as_millis()
@@ -4190,8 +4193,7 @@ pub async fn handle_wac_v2_output(
             let resume_id: u32 = windmill_common::wac::approval_resume_id(&key);
             // Read before the tx: it holds the queue row lock, and a pool checkout under it
             // waits on the connections every other suspending parent is holding.
-            let wkey =
-                windmill_common::variables::get_workspace_key(&job.workspace_id, db).await?;
+            let wkey = windmill_common::variables::get_workspace_key(&job.workspace_id, db).await?;
             let approval_token =
                 windmill_common::variables::approval_token_from_key(&wkey, job.id)?;
 

backend/windmill-worker/src/js_eval.rs

@@ -107,6 +107,31 @@ pub async fn eval_timeout(
     }
 }
 
+/// Adds `step_id`, `job_id` and `flow_path` to a native script's thrown error, as the
+/// subprocess wrappers do from the same env vars.
+pub fn with_step_context(
+    e: windmill_common::error::Error,
+    reserved_variables: &HashMap<String, String>,
+) -> windmill_common::error::Error {
+    use windmill_common::error::Error;
+    let Error::ExecutionRawError(raw) = &e else {
+        return e;
+    };
+    let Ok(serde_json::Value::Object(mut err)) = serde_json::from_str(raw.get()) else {
+        return e;
+    };
+    for (key, var) in [
+        ("step_id", "WM_FLOW_STEP_ID"),
+        ("job_id", "WM_JOB_ID"),
+        ("flow_path", "WM_FLOW_PATH"),
+    ] {
+        if let Some(v) = reserved_variables.get(var).filter(|v| !v.is_empty()) {
+            err.insert(key.to_string(), serde_json::Value::String(v.clone()));
+        }
+    }
+    Error::ExecutionRawError(windmill_common::worker::to_raw_value(&err))
+}
+
 #[cfg(feature = "deno_core")]
 pub fn transpile_ts(expr: String) -> anyhow::Result<String> {
     windmill_runtime_nativets::transpile_ts(expr)

backend/windmill-worker/src/worker.rs

@@ -6695,7 +6695,8 @@ pub async fn run_language_executor(
             occupancy_metrics,
             has_stream,
         ))
-        .await?;
+        .await
+        .map_err(|e| crate::js_eval::with_step_context(e, &reserved_variables))?;
         return Ok(result);
     }