feat: env var in flow (#6852)

backend/.sqlx/query-0c0f3909b80c35210fc64c685905308621f9135c2c45a2fa0531ea750387da1f.json

@@ -0,0 +1,25 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n            SELECT \n                CASE \n                    WHEN flow_version.id IS NOT NULL THEN\n                        (flow_version.value -> 'flow_env' -> $3) #> $4\n                    ELSE\n                        (root_job.raw_flow -> 'flow_env' -> $3) #> $4\n                END AS \"flow_env: sqlx::types::Json<Box<RawValue>>\"\n            FROM \n                v2_job current_job\n            JOIN \n                v2_job root_job ON root_job.id = COALESCE(current_job.root_job, current_job.flow_innermost_root_job, current_job.parent_job, current_job.id)\n                AND root_job.workspace_id = current_job.workspace_id\n            LEFT JOIN\n                flow_version ON flow_version.id = root_job.runnable_id\n                AND flow_version.path = root_job.runnable_path\n                AND flow_version.workspace_id = root_job.workspace_id\n           WHERE \n                current_job.id = $1 AND \n                current_job.workspace_id = $2",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "flow_env: sqlx::types::Json<Box<RawValue>>",
+        "type_info": "Jsonb"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "Text",
+        "Text",
+        "TextArray"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "0c0f3909b80c35210fc64c685905308621f9135c2c45a2fa0531ea750387da1f"
+}

backend/.sqlx/query-5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55.json

@@ -15,7 +15,7 @@
       ]
     },
     "nullable": [
-      null
+      true
     ]
   },
   "hash": "5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55"

backend/windmill-api/src/flows.rs

@@ -1634,6 +1634,7 @@ mod tests {
             early_return: None,
             concurrency_key: None,
             chat_input_enabled: None,
+            flow_env: None,
             debounce_key: None,
             debounce_delay_s: None,
         };

backend/windmill-api/src/jobs.rs

@@ -281,6 +281,10 @@ pub fn workspaced_service() -> Router {
             "/result_by_id/:job_id/:node_id",
             get(get_result_by_id).layer(cors.clone()),
         )
+        .route(
+            "/flow_env_by_flow_job_id/:flow_job_id/:var_name",
+            get(get_flow_env_by_flow_job_id).layer(cors.clone()),
+        )
         .route("/run/dependencies", post(run_dependencies_job))
         .route("/run/flow_dependencies", post(run_flow_dependencies_job))
         .route(
@@ -376,6 +380,59 @@ async fn get_root_job(
     Ok(Json(res))
 }
 
+async fn get_flow_env_by_flow_job_id(
+    authed: ApiAuthed,
+    tokened: Tokened,
+    Extension(db): Extension<DB>,
+    Path((w_id, flow_job_id, var_name)): Path<(String, Uuid, String)>,
+    Query(JsonPath { json_path, .. }): Query<JsonPath>,
+) -> windmill_common::error::JsonResult<Box<JsonRawValue>> {
+    let flow_env = sqlx::query_scalar!(
+        r#"
+            SELECT 
+                CASE 
+                    WHEN flow_version.id IS NOT NULL THEN
+                        (flow_version.value -> 'flow_env' -> $3) #> $4
+                    ELSE
+                        (root_job.raw_flow -> 'flow_env' -> $3) #> $4
+                END AS "flow_env: sqlx::types::Json<Box<RawValue>>"
+            FROM 
+                v2_job current_job
+            JOIN 
+                v2_job root_job ON root_job.id = COALESCE(current_job.root_job, current_job.flow_innermost_root_job, current_job.parent_job, current_job.id)
+                AND root_job.workspace_id = current_job.workspace_id
+            LEFT JOIN
+                flow_version ON flow_version.id = root_job.runnable_id
+                AND flow_version.path = root_job.runnable_path
+                AND flow_version.workspace_id = root_job.workspace_id
+           WHERE 
+                current_job.id = $1 AND 
+                current_job.workspace_id = $2"#,
+        flow_job_id,
+        w_id,
+        var_name,
+        json_path
+            .as_ref()
+            .map(|x| x.split(".").collect::<Vec<_>>())
+            .unwrap_or_default() as Vec<&str>,
+    )
+    .fetch_optional(&db)
+    .await?
+    .map(|r| r.map(|x| x.0))
+    .flatten()
+    .unwrap_or_else(|| to_raw_value(&serde_json::Value::Null));
+
+    log_job_view(
+        &db,
+        Some(&authed),
+        Some(&tokened.token),
+        &w_id,
+        &flow_job_id,
+    )
+    .await?;
+    Ok(Json(flow_env))
+}
+
 async fn compute_root_job_for_flow(db: &DB, w_id: &str, job_id: Uuid) -> error::Result<String> {
     let root_job = sqlx::query_scalar!(
         r#"SELECT COALESCE(root_job, flow_innermost_root_job, parent_job, id) as "root_job!" FROM v2_job WHERE id = $1 AND workspace_id = $2"#,

backend/windmill-common/src/client.rs

@@ -49,14 +49,7 @@ impl AuthedClient {
             "{}/api/w/{}/oidc/token/{}",
             self.base_internal_url, self.workspace, audience
         );
-        let response = self.get(&url, vec![]).await?;
-        match response.status().as_u16() {
-            200u16 => Ok(response
-                .json::<String>()
-                .await
-                .context("decoding oidc token as json string")?),
-            _ => Err(anyhow::anyhow!(response.text().await.unwrap_or_default())),
-        }
+        make_basic_get_request(self, &url, None, Some("decoding oidc token as json string")).await
     }
 
     pub async fn get_resource_value<T: DeserializeOwned>(&self, path: &str) -> anyhow::Result<T> {
@@ -64,14 +57,7 @@ impl AuthedClient {
             "{}/api/w/{}/resources/get_value/{}",
             self.base_internal_url, self.workspace, path
         );
-        let response = self.get(&url, vec![]).await?;
-        match response.status().as_u16() {
-            200u16 => Ok(response
-                .json::<T>()
-                .await
-                .context("decoding resource value as json")?),
-            _ => Err(anyhow::anyhow!(response.text().await.unwrap_or_default())),
-        }
+        make_basic_get_request(self, &url, None, Some("decoding resource value as json")).await
     }
 
     pub async fn get_variable_value(&self, path: &str) -> anyhow::Result<String> {
@@ -79,14 +65,7 @@ impl AuthedClient {
             "{}/api/w/{}/variables/get_value/{}",
             self.base_internal_url, self.workspace, path
         );
-        let response = self.get(&url, vec![]).await?;
-        match response.status().as_u16() {
-            200u16 => Ok(response
-                .json::<String>()
-                .await
-                .context("decoding variable value as json")?),
-            _ => Err(anyhow::anyhow!(response.text().await.unwrap_or_default())),
-        }
+        make_basic_get_request(self, &url, None, Some("decoding variable value as json")).await
     }
 
     pub async fn get_resource_value_interpolated<T: DeserializeOwned>(
@@ -121,19 +100,34 @@ impl AuthedClient {
             "{}/api/w/{}/jobs_u/completed/get_result/{}",
             self.base_internal_url, self.workspace, path
         );
-        let query = if let Some(json_path) = json_path {
-            vec![("json_path", json_path)]
-        } else {
-            vec![]
-        };
-        let response = self.get(&url, query).await?;
-        match response.status().as_u16() {
-            200u16 => Ok(response
-                .json::<T>()
-                .await
-                .context("decoding completed job result as json")?),
-            _ => Err(anyhow::anyhow!(response.text().await.unwrap_or_default())),
-        }
+        let query = query_from_json_path(json_path);
+        make_basic_get_request(
+            self,
+            &url,
+            Some(query),
+            Some("decoding completed job result as json"),
+        )
+        .await
+    }
+
+    pub async fn get_flow_env_by_flow_job_id<T: DeserializeOwned>(
+        &self,
+        root_job_id: &str,
+        var_name: &str,
+        json_path: Option<String>,
+    ) -> anyhow::Result<T> {
+        let url = format!(
+            "{}/api/w/{}/jobs/flow_env_by_flow_job_id/{}/{}",
+            self.base_internal_url, self.workspace, root_job_id, var_name
+        );
+        let query = query_from_json_path(json_path);
+        make_basic_get_request(
+            self,
+            &url,
+            Some(query),
+            Some("decoding flow env variable as json"),
+        )
+        .await
     }
 
     pub async fn get_result_by_id<T: DeserializeOwned>(
@@ -146,19 +140,14 @@ impl AuthedClient {
             "{}/api/w/{}/jobs/result_by_id/{}/{}",
             self.base_internal_url, self.workspace, flow_job_id, node_id
         );
-        let query = if let Some(json_path) = json_path {
-            vec![("json_path", json_path)]
-        } else {
-            vec![]
-        };
-        let response = self.get(&url, query).await?;
-        match response.status().as_u16() {
-            200u16 => Ok(response
-                .json::<T>()
-                .await
-                .context("decoding result by id as json")?),
-            _ => Err(anyhow::anyhow!(response.text().await.unwrap_or_default())),
-        }
+        let query = query_from_json_path(json_path);
+        make_basic_get_request(
+            self,
+            &url,
+            Some(query),
+            Some("decoding result by id as json"),
+        )
+        .await
     }
 
     pub async fn upload_s3_file<S>(
@@ -245,3 +234,33 @@ impl AuthedClient {
         }
     }
 }
+
+#[inline]
+fn query_from_json_path(json_path: Option<String>) -> Vec<(&'static str, String)> {
+    json_path
+        .map(|json_path| vec![("json_path", json_path)])
+        .unwrap_or_else(|| Vec::new())
+}
+
+#[inline]
+async fn make_basic_get_request<T: DeserializeOwned>(
+    client: &AuthedClient,
+    url: &str,
+    query: Option<Vec<(&'static str, String)>>,
+    context: Option<&'static str>,
+) -> anyhow::Result<T> {
+    let response = client
+        .get(&url, query.unwrap_or_else(|| Vec::new()))
+        .await?;
+
+    match response.status().as_u16() {
+        200u16 => {
+            let json_body = response
+                .json::<T>()
+                .await
+                .context(context.unwrap_or("error decoding body as json"))?;
+            Ok(json_body)
+        }
+        _ => Err(anyhow::anyhow!(response.text().await.unwrap_or_default())),
+    }
+}

backend/windmill-common/src/flows.rs

@@ -192,6 +192,8 @@ pub struct FlowValue {
     pub priority: Option<i16>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub chat_input_enabled: Option<bool>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub flow_env: Option<HashMap<String, Box<RawValue>>>
 }
 
 impl FlowValue {

backend/windmill-common/src/variables.rs

@@ -401,8 +401,7 @@ pub async fn get_reserved_variables(
     value,
     description: "Custom workspace environment variable".to_string(),
     is_custom: true,
-})
-).collect()
+})).collect()
 }
 
 async fn get_cached_workspace_envs(conn: &Connection, w_id: &str) -> Vec<(String, String)> {
@@ -439,7 +438,11 @@ async fn get_cached_workspace_envs(conn: &Connection, w_id: &str) -> Vec<(String
     custom_envs
 }
 
-pub async fn get_variable_or_self(path: String, db: &DB, w_id: &str) -> crate::error::Result<String> {
+pub async fn get_variable_or_self(
+    path: String,
+    db: &DB,
+    w_id: &str,
+) -> crate::error::Result<String> {
     if !path.starts_with("$var:") {
         return Ok(path);
     }

backend/windmill-queue/src/jobs.rs

@@ -4329,6 +4329,7 @@ pub async fn push<'c, 'd>(
                 skip_expr: None,
                 preprocessor_module: None,
                 chat_input_enabled: None,
+                flow_env: None,
             };
             // this is a new flow being pushed, flow_status is set to flow_value:
             let flow_status: FlowStatus = FlowStatus::new(&flow_value);

backend/windmill-worker/src/ai/tools.rs

@@ -338,6 +338,7 @@ async fn execute_windmill_tool(
             transform,
             last_result.clone(),
             flow_inputs.clone(),
+            None,
             Some(ctx.client),
             ctx.id_context.as_ref(),
         )

backend/windmill-worker/src/common.rs

@@ -1068,7 +1068,6 @@ pub fn build_http_client(timeout_duration: std::time::Duration) -> error::Result
 }
 
 pub fn get_root_job_id(job: &MiniPulledJob) -> uuid::Uuid {
-    // fallback to flow_innermost_root_job and parent_job as root_job is not set if equal to innermost root job or parent job
     job.root_job
         .or(job.flow_innermost_root_job)
         .or(job.parent_job)

backend/windmill-worker/src/js_eval.rs

@@ -165,10 +165,98 @@ impl NetPermissions for PermissionsContainer {
 #[cfg(feature = "deno_core")]
 pub struct OptAuthedClient(Option<AuthedClient>);
 
+const FLOW_INPUT_PREFIX: &'static str = "flow_input";
+const ENV_KEY_PREFIX: &'static str = "flow_env";
+const DOT_PATTERN: &'static str = ".";
+const START_BRACKET_PATTERN: &'static str = "[\"";
+const END_BRACKET_PATTERN: &'static str = "\"]";
+
+fn try_exact_property_access(
+    expr: &str,
+    flow_input: Option<&mappable_rc::Marc<HashMap<String, Box<RawValue>>>>,
+    flow_env: Option<&HashMap<String, Box<RawValue>>>,
+) -> Option<Box<RawValue>> {
+    let obj = if expr.starts_with(FLOW_INPUT_PREFIX) {
+        Some((
+            FLOW_INPUT_PREFIX,
+            flow_input.as_ref().map(|obj| obj.as_ref()),
+        ))
+    } else if expr.starts_with(ENV_KEY_PREFIX) {
+        Some((ENV_KEY_PREFIX, flow_env))
+    } else {
+        None
+    };
+
+    if let Some((prefix, obj)) = obj {
+        let access_pattern_pos = prefix.len();
+        let suffix = &expr[access_pattern_pos..];
+        let maybe_key_name = if suffix.starts_with(DOT_PATTERN) {
+            let key_name_pos = DOT_PATTERN.len();
+            Some(&expr[key_name_pos..])
+        } else if suffix.starts_with(START_BRACKET_PATTERN) {
+            let key_name_pos = START_BRACKET_PATTERN.len();
+            let suffix = &suffix[key_name_pos..];
+
+            let flow_arg_name = suffix
+                .ends_with(END_BRACKET_PATTERN)
+                .then(|| {
+                    let start_key_name_pos = access_pattern_pos + key_name_pos;
+                    let end_key_name_pos = expr.len() - END_BRACKET_PATTERN.len();
+                    &expr[start_key_name_pos..end_key_name_pos]
+                })
+                .filter(|s| s.len() > 0);
+            flow_arg_name
+        } else {
+            None
+        };
+
+        if let Some(key_name) = maybe_key_name {
+            if let Some(key_value) = obj.and_then(|obj| obj.get(key_name)) {
+                return Some(key_value.clone());
+            }
+        }
+    }
+    None
+}
+
+async fn handle_full_regex(
+    captures: regex::Captures<'_>,
+    authed_client: &AuthedClient,
+    by_id: &IdContext,
+) -> anyhow::Result<Box<RawValue>> {
+    let obj_name = captures.get(1).unwrap().as_str();
+    let obj_key = captures.get(2).unwrap().as_str();
+    let idx_o = captures.get(3).map(|y| y.as_str());
+    let rest = captures.get(4).map(|y| y.as_str());
+    let query = if let Some(idx) = idx_o {
+        match rest {
+            Some(rest) => Some(format!("{}{}", idx, rest)),
+            None => Some(idx.to_string()),
+        }
+    } else {
+        rest.map(|x| x.trim_start_matches('.').to_string())
+    };
+
+    let result = if obj_name == "results" {
+        authed_client
+            .get_result_by_id(&by_id.flow_job.to_string(), obj_key, query)
+            .await
+    } else if obj_name == "flow_env" {
+        authed_client
+            .get_flow_env_by_flow_job_id(&by_id.flow_job.to_string(), obj_key, query)
+            .await
+    } else {
+        unreachable!();
+    };
+
+    return result;
+}
+
 pub async fn eval_timeout(
     expr: String,
     transform_context: HashMap<String, Arc<Box<RawValue>>>,
     flow_input: Option<mappable_rc::Marc<HashMap<String, Box<RawValue>>>>,
+    flow_env: Option<&HashMap<String, Box<RawValue>>>,
     authed_client: Option<&AuthedClient>,
     by_id: Option<&IdContext>,
     #[allow(unused_variables)] ctx: Option<Vec<(String, String)>>,
@@ -180,21 +268,13 @@ pub async fn eval_timeout(
         expr,
         transform_context
     );
-    for (k, v) in transform_context.iter() {
-        if k == &expr {
-            return Ok(v.as_ref().clone());
-        }
+
+    if let Some(value) = transform_context.get(&expr) {
+        return Ok(value.as_ref().to_owned());
     }
 
-    if expr.starts_with("flow_input.") || expr.starts_with("flow_input[") {
-        if let Some(ref flow_input) = flow_input {
-            for (k, v) in flow_input.iter() {
-                if &format!("flow_input.{k}") == &expr || &format!("flow_input[\"{k}\"]") == &expr {
-                    // tracing::error!("FLOW_INPUT");
-                    return Ok(v.clone());
-                }
-            }
-        }
+    if let Some(value) = try_exact_property_access(&expr, flow_input.as_ref(), flow_env) {
+        return Ok(value);
     }
 
     let p_ids = by_id.map(|x| {
@@ -219,24 +299,8 @@ pub async fn eval_timeout(
     }
 
     if let (Some(by_id), Some(authed_client)) = (by_id, authed_client) {
-        if let Some((id, idx_o, rest)) = RE_FULL.captures(&expr).map(|x| {
-            (
-                x.get(1).unwrap().as_str(),
-                x.get(2).map(|y| y.as_str()),
-                x.get(3).map(|y| y.as_str()),
-            )
-        }) {
-            let query = if let Some(idx) = idx_o {
-                match rest {
-                    Some(rest) => Some(format!("{}{}", idx, rest)),
-                    None => Some(idx.to_string()),
-                }
-            } else {
-                rest.map(|x| x.trim_start_matches('.').to_string())
-            };
-            return authed_client
-                .get_result_by_id(&by_id.flow_job.to_string(), id, query)
-                .await;
+        if let Some(captures) = RE_FULL.captures(&expr) {
+            return handle_full_regex(captures, authed_client, by_id).await;
         }
     }
 
@@ -271,8 +335,8 @@ pub async fn eval_timeout(
                 if by_id.is_some() && authed_client.is_some() {
                     ops.push(op_get_result());
                     ops.push(op_get_id());
+                    ops.push(op_get_flow_env());
                 }
-
                 let ext = Extension { name: "js_eval", ops: ops.into(), ..Default::default() };
                 let exts = vec![ext];
                 // Use our snapshot to provision our new runtime
@@ -307,13 +371,15 @@ pub async fn eval_timeout(
                     let mut client = authed_client.clone();
                     if let Some(client) = client.as_mut() {
                         client.force_client = Some(
-                            configure_client(reqwest::ClientBuilder::new()
-                                .user_agent("windmill/beta")
-                                .danger_accept_invalid_certs(
-                                    std::env::var("ACCEPT_INVALID_CERTS").is_ok(),
-                                ))
-                                .build()
-                                .unwrap(),
+                            configure_client(
+                                reqwest::ClientBuilder::new()
+                                    .user_agent("windmill/beta")
+                                    .danger_accept_invalid_certs(
+                                        std::env::var("ACCEPT_INVALID_CERTS").is_ok(),
+                                    ),
+                            )
+                            .build()
+                            .unwrap(),
                         );
                     }
                     op_state.put(OptAuthedClient(client));
@@ -323,7 +389,7 @@ pub async fn eval_timeout(
                             .into_iter()
                             .filter(|(a, _)| context_keys.contains(a))
                             .collect(),
-                    })
+                    });
                 }
 
                 sender
@@ -379,10 +445,12 @@ fn replace_with_await(expr: String, fn_name: &str) -> String {
 }
 lazy_static! {
     static ref RE: Regex =
-        Regex::new(r#"(?m)(?P<r>results(?:\?)?(?:(?:\.[a-zA-Z_0-9]+)|(?:\[\".*?\"\])))"#).unwrap();
-    static ref RE_FULL: Regex =
-        Regex::new(r"(?m)^results(?:\?)?\.([a-zA-Z_0-9]+)(?:\[(\d+)\])?((?:\.[a-zA-Z_0-9]+)+)?$")
+        Regex::new(r#"(?m)(?P<r>(?:results|flow_env)(?:\?)?(?:(?:\.[a-zA-Z_0-9]+)|(?:\[\".*?\"\])))"#)
             .unwrap();
+    static ref RE_FULL: Regex = Regex::new(
+        r"(?m)^(results|flow_env)(?:\?)?\.([a-zA-Z_0-9]+)(?:\[(\d+)\])?((?:\.[a-zA-Z_0-9]+)+)?$"
+    )
+    .unwrap();
     static ref RE_PROXY: Regex =
         Regex::new(r"^(https?)://(([^:@\s]+):([^:@\s]+)@)?([^:@\s]+)(:(\d+))?$").unwrap();
 }
@@ -453,6 +521,17 @@ const results = new Proxy({{}}, {{
     }}
 }});
 
+async function flow_env_by_var_name(var_name) {{
+    let root_job_id = "{}";
+    return JSON.parse(await Deno.core.ops.op_get_flow_env(root_job_id, var_name, null));
+}}
+
+const flow_env = new Proxy({{}}, {{
+    get: function(target, name, receiver) {{
+        return flow_env_by_var_name(name);
+    }}
+}});
+
 "#,
                 by_id
                     .steps_results
@@ -469,6 +548,7 @@ const results = new Proxy({{}}, {{
                     .join(","),
                 by_id.previous_id,
                 by_id.flow_job,
+                by_id.flow_job
             )
         } else {
             String::new()
@@ -641,6 +721,33 @@ async fn op_resource(
     }
 }
 
+#[cfg(feature = "deno_core")]
+#[op2(async)]
+#[string]
+async fn op_get_flow_env(
+    op_state: Rc<RefCell<OpState>>,
+    #[string] root_job_id: String,
+    #[string] var_name: String,
+    #[string] json_path: Option<String>,
+) -> Result<Option<String>, deno_error::JsErrorBox> {
+    let client = op_state.borrow().borrow::<OptAuthedClient>().0.clone();
+    if let Some(client) = client {
+        client
+            .get_flow_env_by_flow_job_id::<Option<Box<RawValue>>>(
+                &root_job_id,
+                &var_name,
+                json_path,
+            )
+            .await
+            .map(|value| value.map(|val| val.get().to_string()))
+            .map_err(|e| deno_error::JsErrorBox::generic(e.to_string()))
+    } else {
+        Err(deno_error::JsErrorBox::generic(
+            "No client found in op state",
+        ))
+    }
+}
+
 #[cfg(feature = "deno_core")]
 pub struct TransformContext {
     pub envs: HashMap<String, Arc<Box<RawValue>>>,
@@ -1318,7 +1425,7 @@ multiline template`";
             op_state.put(TransformContext { flow_input: None, envs: env.clone() })
         }
 
-        let res = eval_timeout(code.to_string(), env, None, None, None, None).await?;
+        let res = eval_timeout(code.to_string(), env, None, None, None, None, None).await?;
         assert_eq!(res.get(), "2");
         Ok(())
     }

backend/windmill-worker/src/worker.rs

@@ -877,7 +877,7 @@ pub fn start_interactive_worker_shell(
                             token,
                             precomputed_agent_info: precomputed_bundle,
                         } = extract_job_and_perms(job, &conn).await;
-
+                        
                         let authed_client = AuthedClient::new(
                             base_internal_url.to_owned(),
                             job.workspace_id.clone(),
@@ -886,7 +886,7 @@ pub fn start_interactive_worker_shell(
                         );
 
                         let arc_job = Arc::new(job);
-
+                        
                         let _ = handle_queued_job(
                             arc_job.clone(),
                             raw_code,

backend/windmill-worker/src/worker_flow.rs

@@ -245,6 +245,7 @@ async fn evaluate_stop_after_all_iters_if(
     let stop_early_after_all_iters = compute_bool_from_expr(
         &stop_after_all_iters_if.expr,
         Marc::new(args),
+        None,
         iters_result.clone(),
         None,
         None,
@@ -427,17 +428,20 @@ pub async fn update_flow_status_after_job_completion_internal(
             )
             .fetch_one(db)
             .await;
+            let args =
+                args.map(|flow_args| flow_args.map(|flow_args| flow_args.0).unwrap_or_default());
+
             args
         }));
 
-        let from_result_to_args =
-            |args: &Result<Option<Json<HashMap<String, Box<RawValue>>>>, sqlx::Error>| {
-                let args = args.as_ref().map_err(|e| {
-                    Error::internal_err(format!("retrieval of args from state: {e:#}"))
-                })?;
 
-                Ok::<_, Error>(args.clone().unwrap_or_default().0)
-            };
+        let from_result_to_args = |args: &Result<HashMap<String, Box<RawValue>>, sqlx::Error>| {
+            let args = args
+                .as_ref()
+                .map_err(|e| Error::internal_err(format!("retrieval of args from state: {e:#}")))?;
+
+            Ok::<_, Error>(args.clone())
+        };
 
         let (mut stop_early, mut stop_early_err_msg, mut skip_if_stop_early, continue_on_error) =
             if stop_early_override.is_some()
@@ -469,9 +473,11 @@ pub async fn update_flow_status_after_job_completion_internal(
                                 _ => None,
                             };
                         let args = from_result_to_args(args.as_ref().await.get_ref())?;
+                       
                         compute_bool_from_expr(
                             &expr,
                             Marc::new(args),
+                            None,
                             result.clone(),
                             all_iters,
                             None,
@@ -736,6 +742,7 @@ pub async fn update_flow_status_after_job_completion_internal(
                             &mut stop_early_err_msg,
                             &mut nresult,
                             args,
+
                         )
                         .await?;
                     }
@@ -924,6 +931,7 @@ pub async fn update_flow_status_after_job_completion_internal(
                         .and_then(|x| x.stop_after_all_iters_if.as_ref())
                     {
                         let args = from_result_to_args(args.as_ref().await.get_ref())?;
+
                         evaluate_stop_after_all_iters_if(
                             db,
                             stop_after_all_iters_if,
@@ -1016,6 +1024,7 @@ pub async fn update_flow_status_after_job_completion_internal(
                             &old_status.retry,
                             result.clone(),
                             Marc::new(args),
+                            None,
                             Some(client),
                         )
                         .await?
@@ -1334,6 +1343,7 @@ pub async fn update_flow_status_after_job_completion_internal(
                 &old_status.retry,
                 result.clone(),
                 Marc::new(args),
+                None,
                 Some(client),
             )
             .await?
@@ -1824,6 +1834,7 @@ async fn evaluate_retry(
     status: &RetryStatus,
     result: Arc<Box<RawValue>>,
     flow_args: Marc<HashMap<String, Box<RawValue>>>,
+    flow_env: Option<&HashMap<String, Box<RawValue>>>,
     client: Option<&AuthedClient>,
 ) -> anyhow::Result<Option<(u32, Duration)>> {
     if status.fail_count > MAX_RETRY_ATTEMPTS {
@@ -1834,6 +1845,7 @@ async fn evaluate_retry(
         let should_retry = compute_bool_from_expr(
             &retry_if.expr,
             flow_args,
+            flow_env,
             result,
             None,
             None,
@@ -1857,6 +1869,7 @@ async fn evaluate_retry(
 async fn compute_bool_from_expr(
     expr: &str,
     flow_args: Marc<HashMap<String, Box<RawValue>>>,
+    flow_env: Option<&HashMap<String, Box<RawValue>>>,
     result: Arc<Box<RawValue>>,
     all_iters: Option<Arc<Box<RawValue>>>,
     by_id: Option<&IdContext>,
@@ -1881,6 +1894,7 @@ async fn compute_bool_from_expr(
         format!("Boolean({expr})"),
         context,
         Some(flow_args),
+        flow_env,
         client,
         by_id,
         ctx,
@@ -1906,6 +1920,7 @@ pub async fn evaluate_input_transform<T>(
     transform: &InputTransform,
     last_result: Arc<Box<RawValue>>,
     flow_args: Option<Marc<HashMap<String, Box<RawValue>>>>,
+    flow_env: Option<&HashMap<String, Box<RawValue>>>,
     authed_client: Option<&AuthedClient>,
     by_id: Option<&IdContext>,
 ) -> error::Result<T>
@@ -1927,6 +1942,7 @@ where
                 expr.to_string(),
                 context,
                 flow_args,
+                flow_env,
                 authed_client,
                 by_id,
                 None,
@@ -1955,6 +1971,7 @@ where
 #[instrument(level = "trace", skip_all)]
 async fn transform_input(
     flow_args: Marc<HashMap<String, Box<RawValue>>>,
+    flow_env: Option<&HashMap<String, Box<RawValue>>>,
     last_result: Arc<Box<RawValue>>,
     input_transforms: &HashMap<String, InputTransform>,
     resumes: Arc<Box<RawValue>>,
@@ -1996,6 +2013,7 @@ async fn transform_input(
                     expr.to_string(),
                     env.clone(),
                     Some(flow_args.clone()),
+                    flow_env,
                     Some(client),
                     Some(by_id),
                     None,
@@ -2069,6 +2087,7 @@ pub async fn handle_flow(
             );
         }
     }
+
     let mut rec = PushNextFlowJobRec { flow_job: flow_job, status: status };
     loop {
         let PushNextFlowJobRec { flow_job, status } = rec;
@@ -2204,13 +2223,14 @@ async fn push_next_flow_job(
     // tracing::error!("status_module: {status_module:#?}");
 
     let fj: mappable_rc::Marc<MiniPulledJob> = flow_job.clone().into();
-    let arc_flow_job_args: Marc<HashMap<String, Box<RawValue>>> = Marc::map(fj, |x| {
-        if let Some(args) = &x.args {
-            &args.0
-        } else {
-            &EHM
-        }
-    });
+    let arc_flow_job_args: Marc<HashMap<String, Box<RawValue>>> =
+        Marc::map(fj, |x: &MiniPulledJob| {
+            if let Some(args) = &x.args {
+                &args.0
+            } else {
+                &EHM
+            }
+        });
 
     // if this is an empty module without preprocessor of if the module has already been completed, successfully, update the parent flow
     if (flow.modules.is_empty() && !step.is_preprocessor_step())
@@ -2300,6 +2320,7 @@ async fn push_next_flow_job(
             let skip = compute_bool_from_expr(
                 &skip_expr,
                 arc_flow_job_args.clone(),
+                flow.flow_env.as_ref(),
                 Arc::new(to_raw_value(&json!("{}"))),
                 None,
                 None,
@@ -2421,6 +2442,7 @@ async fn push_next_flow_job(
                                      expr.to_string(),
                                      context,
                                      Some(arc_flow_job_args.clone()),
+                                     flow.flow_env.as_ref(),
                                      None,
                                      None,
                                      None
@@ -2681,6 +2703,7 @@ async fn push_next_flow_job(
                     &input_transform,
                     arc_last_job_result.clone(),
                     Some(arc_flow_job_args.clone()),
+                    flow.flow_env.as_ref(),
                     Some(client),
                     None,
                 )
@@ -2718,6 +2741,7 @@ async fn push_next_flow_job(
             &status.retry,
             arc_last_job_result.clone(),
             arc_flow_job_args.clone(),
+            flow.flow_env.as_ref(),
             Some(client),
         )
         .await?
@@ -2807,6 +2831,7 @@ async fn push_next_flow_job(
         compute_bool_from_expr(
             &skip_if.expr,
             arc_flow_job_args.clone(),
+            flow.flow_env.as_ref(),
             arc_last_job_result.clone(),
             None,
             Some(&idcontext),
@@ -2898,6 +2923,7 @@ async fn push_next_flow_job(
                 };
                 transform_input(
                     arc_flow_job_args.clone(),
+                    flow.flow_env.as_ref(),
                     arc_last_job_result.clone(),
                     input_transforms,
                     resumes.clone(),
@@ -2924,6 +2950,7 @@ async fn push_next_flow_job(
     let next_flow_transform = compute_next_flow_transform(
         arc_flow_job_args.clone(),
         arc_last_job_result.clone(),
+        flow.flow_env.as_ref(),
         &flow_job,
         &flow,
         transform_context,
@@ -3072,6 +3099,7 @@ async fn push_next_flow_job(
                         .await?;
                     let ti = transform_input(
                         Marc::new(args),
+                        flow.flow_env.as_ref(),
                         arc_last_job_result.clone(),
                         input_transforms,
                         resumes.clone(),
@@ -3122,6 +3150,7 @@ async fn push_next_flow_job(
                             .await?;
                         let ti = transform_input(
                             Marc::new(hm),
+                            flow.flow_env.as_ref(),
                             arc_last_job_result.clone(),
                             input_transforms,
                             resumes.clone(),
@@ -3225,6 +3254,7 @@ async fn push_next_flow_job(
                 timeout_transform,
                 arc_last_job_result.clone(),
                 Some(arc_flow_job_args.clone()),
+                flow.flow_env.as_ref(),
                 Some(client),
                 Some(&ctx),
             )
@@ -3304,6 +3334,7 @@ async fn push_next_flow_job(
                     parallelism_transform,
                     arc_last_job_result.clone(),
                     Some(arc_flow_job_args.clone()),
+                    flow.flow_env.as_ref(),
                     Some(client),
                     Some(&ctx),
                 )
@@ -3761,6 +3792,7 @@ pub fn get_path(flow_job: &MiniPulledJob, status: &FlowStatus, module: &FlowModu
 async fn compute_next_flow_transform(
     arc_flow_job_args: Marc<HashMap<String, Box<RawValue>>>,
     arc_last_job_result: Arc<Box<RawValue>>,
+    flow_env: Option<&HashMap<String, Box<RawValue>>>,
     flow_job: &MiniPulledJob,
     flow: &FlowValue,
     by_id: Option<IdContext>,
@@ -3979,6 +4011,7 @@ async fn compute_next_flow_transform(
                 resume,
                 approvers,
                 arc_flow_job_args,
+                flow_env,
                 client,
                 &parallel,
             )
@@ -4074,6 +4107,7 @@ async fn compute_next_flow_transform(
                         let pred = compute_bool_from_expr(
                             &b.expr,
                             arc_flow_job_args.clone(),
+                            flow.flow_env.as_ref(),
                             arc_last_job_result.clone(),
                             None,
                             Some(&idcontext),
@@ -4338,6 +4372,7 @@ async fn next_forloop_status(
     resume: Arc<Box<RawValue>>,
     approvers: Arc<Box<RawValue>>,
     arc_flow_job_args: Marc<HashMap<String, Box<RawValue>>>,
+    flow_env: Option<&HashMap<String, Box<RawValue>>>,
     client: &AuthedClient,
     parallel: &bool,
 ) -> Result<ForLoopStatus, Error> {
@@ -4370,6 +4405,7 @@ async fn next_forloop_status(
                         expr.to_string(),
                         context,
                         Some(arc_flow_job_args),
+                        flow_env,
                         Some(client),
                         Some(&by_id),
                         None,
@@ -4433,6 +4469,7 @@ async fn next_forloop_status(
                             expr.to_string(),
                             context,
                             Some(arc_flow_job_args),
+                            flow_env,
                             Some(client),
                             Some(&by_id),
                             None,

frontend/src/lib/components/InputTransformForm.svelte

@@ -3,7 +3,8 @@
 		'flow_input',
 		'results',
 		'resource',
-		'variable'
+		'variable',
+		'flow_env'
 	])
 </script>
 

frontend/src/lib/components/flows/content/FlowConstants.svelte

@@ -1,157 +0,0 @@
-<script lang="ts">
-	import { dfs } from '$lib/components/flows/dfs'
-	import FlowCard from '../common/FlowCard.svelte'
-	import { Alert, Badge } from '$lib/components/common'
-	import type { FlowModule, FlowModuleValue, InputTransform, PathScript, RawScript } from '$lib/gen'
-	import { getContext, setContext } from 'svelte'
-	import type { PropPickerWrapperContext } from '../propPicker/PropPickerWrapper.svelte'
-	import { writable } from 'svelte/store'
-	import Toggle from '../../Toggle.svelte'
-	import InputTransformSchemaForm from '$lib/components/InputTransformSchemaForm.svelte'
-	import type { FlowEditorContext } from '../types'
-
-	interface Props {
-		noEditor: boolean
-	}
-
-	let { noEditor }: Props = $props()
-
-	let hideOptional = $state(false)
-	const { flowStateStore, flowStore } = getContext<FlowEditorContext>('FlowEditorContext')
-
-	let scriptModules = $derived(
-		dfs(flowStore.val.value.modules, (x) => x)
-			.map((x) => [x.value, x] as [FlowModuleValue, FlowModule])
-			.filter((x) => x[0].type == 'script' || x[0].type == 'rawscript' || x[0].type == 'flow') as [
-			PathScript | RawScript,
-			FlowModule
-		][]
-	)
-
-	let resources = $derived(
-		Object.fromEntries(
-			scriptModules
-				.map(([v, m]) => [
-					m.id,
-					Object.entries(v.input_transforms)
-						.map((x) => {
-							let schema = flowStateStore.val[m.id]?.schema
-							let val: { argName: string; type: string } | undefined = undefined
-
-							const [k, inputTransform] = x
-							const v = schema?.properties[k]
-
-							if (
-								v?.format?.includes('resource') &&
-								inputTransform.type === 'static' &&
-								(inputTransform.value === '' ||
-									inputTransform.value === undefined ||
-									inputTransform.value === null)
-							) {
-								val = {
-									argName: k,
-									type: v.format.split('-')[1]
-								}
-							}
-							return val
-						})
-						.filter(Boolean)
-				])
-				.filter((x) => x[1].length > 0)
-		) as {
-			[k: string]: {
-				argName: string
-				type: string
-			}[]
-		}
-	)
-
-	let steps = $derived(
-		scriptModules
-			.map(
-				([v, m]) =>
-					[
-						v.input_transforms,
-						Object.entries(v.input_transforms)
-							.filter((x) => {
-								const shouldDisplay = hideOptional
-									? flowStateStore.val[m.id]?.schema?.required?.includes(x[0])
-									: true
-								return x[1].type == 'static' && shouldDisplay
-							})
-							.map((x) => x[0]),
-						m
-					] as [Record<string, InputTransform>, string[], FlowModule]
-			)
-			.filter(([i, f, m]) => f.length > 0)
-	)
-
-	setContext<PropPickerWrapperContext>('PropPickerWrapper', {
-		inputMatches: writable(undefined),
-		focusProp: () => {},
-		propPickerConfig: writable(undefined),
-		clearFocus: () => {}
-	})
-</script>
-
-<div class="min-h-full">
-	<FlowCard {noEditor} title="All Static Inputs">
-		{#snippet header()}
-			<Toggle bind:checked={hideOptional} options={{ left: 'Hide optional inputs' }} />
-		{/snippet}
-		<div class="min-h-full flex-1">
-			<Alert type="info" title="Static Inputs" class="m-4"
-				>This page centralizes the static inputs of every steps. It is aking to a file containing
-				all constants. Modifying a value here modifies it in the step input directly. It is
-				especially useful when forking a flow to get an overview of all the variables to parametrize
-				that are not exposed directly as flow inputs.</Alert
-			>
-			{#if Object.keys(resources).length > 0}
-				<Alert type="warning" title="Missing resources" class="m-4">
-					The following resources are missing and the flow will not be fully runnable until they are
-					set. Add your own resources:
-					{#each Object.entries(resources) as [id, r]}
-						{#each r as resource}
-							<div class="mt-2">
-								<Badge color="red">{id}</Badge> is missing a resource of type{' '}
-								<Badge color="red">{resource?.type}</Badge> for the input{' '}
-								<Badge color="red">{resource?.argName}</Badge>
-							</div>
-						{/each}
-					{/each}
-				</Alert>
-			{/if}
-			{#if steps.length == 0}
-				<div class="mt-2"></div>
-				{#if flowStore.val.value.modules.length == 0}
-					<Alert type="warning" title="No steps" class="m-4">
-						This flow has no steps. Add a step to see its static inputs.
-					</Alert>
-				{:else}
-					<Alert type="warning" title="No static inputs" class="m-4">
-						This flow has no steps with static inputs. Add a step with static inputs to see them
-						here.
-					</Alert>
-				{/if}
-			{/if}
-			{#each steps as [_args, filter, m], index (m.id + index)}
-				{#if filter.length > 0}
-					<div class="relative h-full border-t p-4">
-						<p class="text-sm font-bold sticky w-full top-0 z-10 inline-flex items-center">
-							<span class="mr-4">{m.summary || m.value['path'] || 'Inline script'}</span>
-							<Badge large color="indigo">{m.id}</Badge>
-						</p>
-
-						<InputTransformSchemaForm
-							noDynamicToggle
-							{filter}
-							class="mt-2"
-							schema={flowStateStore.val[m.id]?.schema ?? {}}
-							bind:args={steps[index][0]}
-						/>
-					</div>
-				{/if}
-			{/each}
-		</div>
-	</FlowCard>
-</div>

frontend/src/lib/components/flows/content/FlowEditorPanel.svelte

@@ -5,7 +5,7 @@
 	import FlowSettings from './FlowSettings.svelte'
 	import FlowInput from './FlowInput.svelte'
 	import FlowFailureModule from './FlowFailureModule.svelte'
-	import FlowConstants from './FlowConstants.svelte'
+	import FlowEnvironmentVariables from './FlowEnvironmentVariables.svelte'
 	import type { FlowModule, Flow, Job } from '$lib/gen'
 	import FlowPreprocessorModule from './FlowPreprocessorModule.svelte'
 	import type { TriggerContext } from '$lib/components/triggers'
@@ -102,7 +102,7 @@
 {:else if $selectedId === 'Result'}
 	<FlowResult {noEditor} {job} {isOwner} {suspendStatus} {onOpenDetails} />
 {:else if $selectedId === 'constants'}
-	<FlowConstants {noEditor} />
+	<FlowEnvironmentVariables {noEditor} />
 {:else if $selectedId === 'failure'}
 	<FlowFailureModule {noEditor} savedModule={savedFlow?.value.failure_module} />
 {:else if $selectedId === 'preprocessor'}

frontend/src/lib/components/flows/content/FlowEnvironmentVariables.svelte

@@ -0,0 +1,286 @@
+<script lang="ts">
+	import { Alert } from '$lib/components/common'
+	import { getContext, setContext } from 'svelte'
+	import type { PropPickerWrapperContext } from '../propPicker/PropPickerWrapper.svelte'
+	import { writable } from 'svelte/store'
+	import type { FlowEditorContext } from '../types'
+	import { Button } from '$lib/components/common'
+	import { Plus, Trash2 } from 'lucide-svelte'
+	import FlowCard from '../common/FlowCard.svelte'
+	import JsonEditor from '$lib/components/JsonEditor.svelte'
+	import Label from '$lib/components/Label.svelte'
+	import Select from '$lib/components/select/Select.svelte'
+
+	interface Props {
+		noEditor: boolean
+	}
+
+	type EnvVarType = 'string' | 'json'
+
+	interface EnvVarEntry {
+		id: string
+		key: string
+		value: any
+		type: EnvVarType
+		displayValue: string
+		error?: string
+	}
+
+	let { noEditor }: Props = $props()
+
+	const { flowStore } = getContext<FlowEditorContext>('FlowEditorContext')
+
+	if (!flowStore.val.value.flow_env) {
+		flowStore.val.value.flow_env = {}
+	}
+	let flowEnvVarsMap = $derived(new Map(Object.entries(flowStore.val.value.flow_env || {})))
+
+	function determineValueType(value: any): EnvVarType {
+		if (typeof value === 'string') {
+			try {
+				JSON.parse(value)
+				return value.trim().startsWith('{') ||
+					value.trim().startsWith('[') ||
+					value.trim().startsWith('"')
+					? 'json'
+					: 'string'
+			} catch {
+				return 'string'
+			}
+		}
+		return 'json'
+	}
+
+	let flowEnvTypes = $state<Record<string, EnvVarType>>({})
+
+	const typeOptions = [
+		{ label: 'String', value: 'string' as EnvVarType },
+		{ label: 'JSON', value: 'json' as EnvVarType }
+	]
+
+	$effect(() => {
+		for (const [key, value] of flowEnvVarsMap.entries()) {
+			if (!flowEnvTypes[key]) {
+				flowEnvTypes[key] = determineValueType(value)
+			}
+		}
+	})
+
+	$effect(() => {
+		for (const [key, type] of Object.entries(flowEnvTypes)) {
+			if (flowStore.val.value.flow_env && key in flowStore.val.value.flow_env) {
+				const currentType = determineValueType(flowStore.val.value.flow_env[key])
+				if (currentType !== type) {
+					updateEnvType(key, type)
+				}
+			}
+		}
+	})
+
+	let flowEnvEntries = $derived(
+		Array.from(flowEnvVarsMap.entries()).map(([key, value]): EnvVarEntry => {
+			const stringValue = typeof value === 'string' ? value : JSON.stringify(value, null, 2)
+			const type = flowEnvTypes[key] || determineValueType(value)
+			return {
+				id: key,
+				key,
+				value,
+				type,
+				displayValue: stringValue,
+				error: undefined
+			}
+		})
+	)
+
+	function addEnvVar() {
+		const existingKeys = Array.from(flowEnvVarsMap.keys())
+		let counter = 1
+		let newKey = `VAR_${counter}`
+		while (existingKeys.includes(newKey)) {
+			counter++
+			newKey = `VAR_${counter}`
+		}
+
+		if (!flowStore.val.value.flow_env) {
+			flowStore.val.value.flow_env = {}
+		}
+		flowStore.val.value.flow_env[newKey] = ''
+		flowEnvTypes[newKey] = 'string'
+		flowStore.val = flowStore.val
+	}
+
+	function removeEnvVar(key: string) {
+		if (flowStore.val.value.flow_env && key in flowStore.val.value.flow_env) {
+			delete flowStore.val.value.flow_env[key]
+			delete flowEnvTypes[key]
+			flowStore.val = flowStore.val
+		}
+	}
+
+	function updateEnvValue(key: string, value: string, type: EnvVarType) {
+		if (flowStore.val.value.flow_env) {
+			if (type === 'json') {
+				try {
+					const parsed = JSON.parse(value)
+					flowStore.val.value.flow_env[key] = parsed
+				} catch (e) {
+					flowStore.val.value.flow_env[key] = value
+				}
+			} else {
+				flowStore.val.value.flow_env[key] = value
+			}
+			flowStore.val = flowStore.val
+		}
+	}
+
+	function updateEnvKey(oldKey: string, newKey: string) {
+		if (flowStore.val.value.flow_env && oldKey !== newKey && newKey.trim() !== '') {
+			const value = flowStore.val.value.flow_env[oldKey]
+			const type = flowEnvTypes[oldKey] || 'string'
+
+			const newEnvVars: Record<string, any> = {}
+			for (const [k, v] of flowEnvVarsMap.entries()) {
+				if (k === oldKey) {
+					newEnvVars[newKey] = value
+				} else {
+					newEnvVars[k] = v
+				}
+			}
+
+			flowStore.val.value.flow_env = newEnvVars
+			delete flowEnvTypes[oldKey]
+			flowEnvTypes[newKey] = type
+			flowStore.val = flowStore.val
+		}
+	}
+
+	function updateEnvType(key: string, newType: EnvVarType) {
+		if (flowStore.val.value.flow_env && key in flowStore.val.value.flow_env) {
+			const currentValue = flowStore.val.value.flow_env[key]
+			const stringValue =
+				typeof currentValue === 'string' ? currentValue : JSON.stringify(currentValue, null, 2)
+
+			flowEnvTypes[key] = newType
+
+			if (newType === 'json') {
+				try {
+					const parsed = JSON.parse(stringValue)
+					flowStore.val.value.flow_env[key] = parsed
+				} catch {
+					flowStore.val.value.flow_env[key] = stringValue
+				}
+			} else {
+				flowStore.val.value.flow_env[key] = stringValue
+			}
+			flowStore.val = flowStore.val
+		}
+	}
+
+	setContext<PropPickerWrapperContext>('PropPickerWrapper', {
+		inputMatches: writable(undefined),
+		focusProp: () => {},
+		propPickerConfig: writable(undefined),
+		clearFocus: () => {}
+	})
+</script>
+
+<div class="min-h-full">
+	<FlowCard {noEditor} title="Flow Env Variables">
+		<div class="min-h-full flex-1">
+			<Alert type="info" title="Flow Env Variables" class="m-4">
+				Flow envs can be referenced in any flow step input using the syntax{' '}
+				<code>flow_env.VARIABLE_NAME</code> or <code>flow_env["VARIABLE_NAME"]</code>. These
+				variables are available in the property picker and can be used in JavaScript expressions and
+				input bindings. You can choose between String or JSON types for each variable - JSON types
+				allow complex data structures.
+			</Alert>
+
+			{#if flowEnvEntries.length === 0}
+				<Alert type="warning" title="No flow env variables" class="m-4">
+					This flow has no flow env variables defined. Click "Add Variable" to create your first
+					flow env variable.
+				</Alert>
+			{:else}
+				<div class="space-y-4 p-4">
+					{#each flowEnvEntries as entry (entry.id)}
+						<div class="flex flex-col gap-4 p-4 border rounded-lg bg-surface-secondary">
+							<div class="flex items-end gap-3">
+								<div class="flex-1 min-w-0 max-w-xs">
+									<Label label="Variable Name">
+										<input
+											type="text"
+											value={entry.key}
+											onblur={(e) => {
+												const newKey = e.currentTarget.value.trim()
+												if (newKey !== entry.key && newKey !== '') {
+													updateEnvKey(entry.key, newKey)
+												}
+											}}
+											disabled={noEditor}
+											class="input w-full"
+											placeholder="VARIABLE_NAME"
+										/>
+									</Label>
+								</div>
+
+								<Label label="Type">
+									<Select
+										bind:value={flowEnvTypes[entry.key]}
+										items={typeOptions}
+										disabled={noEditor}
+										size="sm"
+										class="text-sm"
+									/>
+								</Label>
+
+								{#if !noEditor}
+									<Button
+										size="sm"
+										color="red"
+										startIcon={{ icon: Trash2 }}
+										onClick={() => removeEnvVar(entry.key)}
+									>
+										Remove
+									</Button>
+								{/if}
+							</div>
+
+							<div class="flex flex-col gap-1">
+								<!-- svelte-ignore a11y_label_has_associated_control -->
+								<label class="text-sm font-medium">Value</label>
+								{#if entry.type === 'json'}
+									<div class="w-full">
+										<JsonEditor
+											bind:code={entry.displayValue}
+											disabled={noEditor}
+											class="min-h-[60px] max-h-[200px]"
+											on:change={() => {
+												updateEnvValue(entry.key, entry.displayValue, 'json')
+											}}
+										/>
+									</div>
+								{:else}
+									<input
+										type="text"
+										value={entry.displayValue}
+										oninput={(e) => updateEnvValue(entry.key, e.currentTarget.value, 'string')}
+										disabled={noEditor}
+										class="input w-full"
+										placeholder="Variable value"
+									/>
+								{/if}
+							</div>
+						</div>
+					{/each}
+				</div>
+			{/if}
+			<div class=" p-4">
+				{#if !noEditor}
+					<Button size="sm" startIcon={{ icon: Plus }} onClick={addEnvVar} color="light">
+						Add Variable
+					</Button>
+				{/if}
+			</div>
+		</div>
+	</FlowCard>
+</div>

frontend/src/lib/components/flows/map/FlowStickyNode.svelte

@@ -66,7 +66,7 @@
 				onClick={() => ($selectedId = 'constants')}
 			/>
 			{#snippet text()}
-				Static inputs
+				Environment Variables
 			{/snippet}
 		</Popover>
 	{/if}

frontend/src/lib/components/flows/previousResults.ts

@@ -9,6 +9,7 @@ export type PickableProperties = {
 	priorIds: Record<string, any>
 	previousId: string | undefined
 	hasResume: boolean
+	flow_env?: Record<string, any>
 }
 
 type StepPropPicker = {
@@ -156,7 +157,8 @@ export function getFailureStepPropPicker(flowState: FlowState, flow: OpenFlow, a
 			flow_input: schemaToObject(flow.schema as any, args),
 			priorIds: priorIds,
 			previousId: undefined,
-			hasResume: false
+			hasResume: false,
+			flow_env: flow.value.flow_env
 		},
 		extraLib: `
 /**
@@ -178,6 +180,17 @@ declare const results = ${JSON.stringify(priorIds)}
 * flow input as an object
 */
 declare const flow_input = ${JSON.stringify(flowInput)};
+
+${
+	flow.value.flow_env
+		? `
+/**
+* flow environment variables
+*/
+declare const flow_env = ${JSON.stringify(flow.value.flow_env)};
+`
+		: ''
+}
 		`
 	}
 }
@@ -218,7 +231,8 @@ export function getStepPropPicker(
 		flow_input: flowInput,
 		priorIds: priorIds,
 		previousId: previousIds[0],
-		hasResume: previousModule?.suspend != undefined
+		hasResume: previousModule?.suspend != undefined,
+		flow_env: flow.value.flow_env
 	}
 
 	if (pickableProperties.hasResume) {
@@ -230,7 +244,8 @@ export function getStepPropPicker(
 			flowInput,
 			priorIds,
 			previousModule?.suspend != undefined,
-			previousModule?.id
+			previousModule?.id,
+			flow.value.flow_env
 		),
 		pickableProperties
 	}
@@ -240,7 +255,8 @@ export function buildExtraLib(
 	flowInput: Record<string, any>,
 	results: Record<string, any>,
 	resume: boolean,
-	previousId: string | undefined
+	previousId: string | undefined,
+	flowEnv?: Record<string, any>
 ): string {
 	return `
 /**
@@ -275,6 +291,17 @@ declare const results = ${JSON.stringify(results)};
  */
 declare const previous_result: ${previousId ? JSON.stringify(results[previousId]) : 'any'};
 
+${
+	flowEnv
+		? `
+/**
+ * flow environment variables
+ */
+declare const flow_env = ${JSON.stringify(flowEnv)};
+`
+		: ''
+}
+
 ${
 	resume
 		? `

frontend/src/lib/components/flows/propPicker/PropPickerWrapper.svelte

@@ -31,6 +31,7 @@
 	import type { PickableProperties } from '../previousResults'
 	import AnimatedButton from '$lib/components/common/button/AnimatedButton.svelte'
 	import type { PropPickerContext } from '$lib/components/prop_picker'
+	import type { FlowEditorContext } from '../types'
 
 	interface Props {
 		pickableProperties: PickableProperties | undefined
@@ -69,6 +70,10 @@
 
 	const { flowPropPickerConfig } = getContext<PropPickerContext>('PropPickerContext')
 	flowPropPickerConfig.set(undefined)
+
+	const { flowStore } = getContext<FlowEditorContext>('FlowEditorContext')
+
+	let flow_env = $derived(pickableProperties?.flow_env || flowStore.val.value.flow_env)
 	setContext<PropPickerWrapperContext>('PropPickerWrapper', {
 		propPickerConfig,
 		inputMatches,
@@ -156,6 +161,7 @@
 						<PropPicker
 							{displayContext}
 							{error}
+							{flow_env}
 							previousId={pickableProperties?.previousId}
 							{pickableProperties}
 							allowCopy={!notSelectable && !$propPickerConfig}

frontend/src/lib/components/propertyPicker/PropPicker.svelte

@@ -18,11 +18,13 @@
 	export let error: boolean = false
 	export let allowCopy = false
 	export let previousId: string | undefined = undefined
+	export let flow_env: Record<string, string> | undefined = undefined
 
 	let variables: Record<string, string> = {}
 	let resources: Record<string, any> = {}
 	let displayVariable = false
 	let displayResources = false
+	let displayFlowEnv = false
 
 	let allResultsCollapsed = true
 	let collapsableInitialState:
@@ -30,6 +32,7 @@
 				allResultsCollapsed: boolean
 				displayVariable: boolean
 				displayResources: boolean
+				displayFlowEnv: boolean
 		  }
 		| undefined
 
@@ -46,6 +49,7 @@
 
 	let flowInputsFiltered: any = pickableProperties.flow_input
 	let resultByIdFiltered: any = pickableProperties.priorIds
+	let flowEnvFiltered: any = pickableProperties.flow_env
 
 	let timeout: number | undefined
 	function onSearch(search: string) {
@@ -63,6 +67,9 @@
 				search === EMPTY_STRING
 					? pickableProperties.priorIds
 					: keepByKey(pickableProperties.priorIds, search)
+
+			flowEnvFiltered =
+				search === EMPTY_STRING ? pickableProperties.flow_env : keepByKey(pickableProperties.flow_env, search)
 		}, 50)
 	}
 
@@ -98,6 +105,7 @@
 			if (search === EMPTY_STRING) {
 				flowInputsFiltered = pickableProperties.flow_input
 				resultByIdFiltered = pickableProperties.priorIds
+				flowEnvFiltered = pickableProperties.flow_env
 			}
 			filteringFlowInputsOrResult = ''
 			return
@@ -109,6 +117,9 @@
 		if (!$inputMatches?.some((match) => match.word === 'results')) {
 			resultByIdFiltered = {}
 		}
+		if (!$inputMatches?.some((match) => match.word === 'flow_env')) {
+			flowEnvFiltered = {}
+		}
 		if ($inputMatches?.length == 1) {
 			filteringFlowInputsOrResult = $inputMatches[0].value
 			if ($inputMatches[0].word === 'flow_input') {
@@ -125,6 +136,13 @@
 				if (Object.keys(filtered).length > 0) {
 					resultByIdFiltered = filtered
 				}
+			} else if ($inputMatches[0].word === 'flow_env') {
+				flowEnvFiltered = pickableProperties.flow_env
+				let [, ...nestedKeys] = $inputMatches[0].value.split('.')
+				let filtered = filterNestedObject(flowEnvFiltered, nestedKeys)
+				if (Object.keys(filtered).length > 0) {
+					flowEnvFiltered = filtered
+				}
 			}
 		} else {
 			filteringFlowInputsOrResult = ''
@@ -143,7 +161,12 @@
 		}
 
 		if (!collapsableInitialState) {
-			collapsableInitialState = { allResultsCollapsed, displayVariable, displayResources }
+			collapsableInitialState = {
+				allResultsCollapsed,
+				displayVariable,
+				displayResources,
+				displayFlowEnv
+			}
 		}
 
 		if ($inputMatches[0].word === 'variable') {
@@ -156,6 +179,10 @@
 			displayResources = true
 			return
 		}
+		if ($inputMatches[0].word === 'flow_env') {
+			displayFlowEnv = true
+			return
+		}
 		if ($inputMatches[0].word === 'results') {
 			allResultsCollapsed = false
 			return
@@ -166,7 +193,8 @@
 		if (!collapsableInitialState) {
 			return
 		}
-		;({ allResultsCollapsed, displayVariable, displayResources } = collapsableInitialState)
+		;({ allResultsCollapsed, displayVariable, displayResources, displayFlowEnv } =
+			collapsableInitialState)
 		collapsableInitialState = undefined
 	}
 
@@ -183,6 +211,7 @@
 		if (prev && !filterActive) {
 			flowInputsFiltered = pickableProperties.flow_input
 			resultByIdFiltered = pickableProperties.priorIds
+			flowEnvFiltered = pickableProperties.flow_env
 		}
 	}
 
@@ -192,7 +221,7 @@
 		await updateCollapsable()
 	}
 
-	$: (search, $inputMatches, $propPickerConfig, pickableProperties, updateState())
+	$: search, $inputMatches, $propPickerConfig, pickableProperties, updateState()
 
 	onDestroy(() => {
 		clearTimeout(timeout)
@@ -400,6 +429,45 @@
 					{/if}
 				</div>
 			{/if}
+			{#if flow_env && Object.keys(flow_env).length > 0 && (!filterActive || $inputMatches?.some((match) => match.word === 'flow_env'))}
+				<div class="overflow-y-auto pb-2">
+					<span class="font-normal text-xs text-secondary">Flow Env Variables:</span>
+
+					{#if displayFlowEnv}
+						<Button
+							color="light"
+							size="xs2"
+							variant="border"
+							on:click={() => {
+								displayFlowEnv = false
+							}}
+							wrapperClasses="inline-flex whitespace-nowrap w-fit"
+							btnClasses="font-mono h-4 text-2xs font-thin px-1 rounded-[0.275rem]">-</Button
+						>
+						<ObjectViewer
+							{allowCopy}
+							pureViewer={!$propPickerConfig}
+							rawKey={false}
+							json={flowEnvFiltered}
+							prefix="flow_env"
+							on:select
+						/>
+					{:else}
+						<Button
+							color="light"
+							size="xs2"
+							variant="border"
+							on:click={() => {
+								displayFlowEnv = true
+							}}
+							wrapperClasses="inline-flex whitespace-nowrap w-fit"
+							btnClasses="font-normal text-2xs rounded-[0.275rem] h-4 px-1"
+						>
+							{'{...}'}
+						</Button>
+					{/if}
+				</div>
+			{/if}
 		{/if}
 		<!-- </div> -->
 	</Scrollable>

frontend/src/lib/components/propertyPicker/PropPickerResult.svelte

@@ -5,6 +5,7 @@
 	export let result: any
 	export let extraResults: any = undefined
 	export let flow_input: any = undefined
+	export let flow_env: any = undefined
 </script>
 
 <div class="w-full px-2">
@@ -18,4 +19,10 @@
 			<ObjectViewer {allowCopy} json={flow_input} prefix="flow_input" on:select />
 		</div>
 	{/if}
+	{#if flow_env}
+		<span class="font-normal text-sm text-secondary">Flow Environment Variables</span>
+		<div class="overflow-y-auto w-full">
+			<ObjectViewer {allowCopy} json={flow_env} prefix="flow_env" on:select />
+		</div>
+	{/if}
 </div>

openflow.openapi.yaml

@@ -62,6 +62,10 @@ components:
           type: string
         cache_ttl:
           type: number
+        flow_env:
+          type: object
+          additionalProperties:
+            type: string
         priority:
           type: number
         early_return: