feat: let operators compose flows when the workspace grants the right

Adds operator_settings.builder_flows: a workspace setting that lets every
operator compose flows out of runnables that already exist. It does not make
them authors. The boundary the operator role draws is authoring code and running
arbitrary code, and this does not move it: check_flow_is_composition_only walks
the value and refuses anything carrying code, including the shapes an obvious
walk misses (code hoisted into a flow_node, an AI agent step's tools, and a
linked ai_agent resource whose tool list is resolved at run time).

What the walk cannot settle it returns for the caller to authorize under RLS:
the worker tags the steps pin, every runnable they reference, and the (path,
hash) of every version-pinned step. Composing a path is enough to run it and to
run it as whoever it runs as, since the worker resolves a step's path with the
root DB handle and adopts that runnable's on_behalf_of. A pinned hash needs its
own check because dispatch ignores the path beside it.

The gate runs on every write and on both request-supplied-value paths, flow
preview and flow dependencies, or either becomes the way to run what the write
path refuses.

Operators of a builder workspace consume a full author seat; the EE companion
carries the counting.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Dsf6VC4MVLisiEoeQkgbr4

AGENTS.md

@@ -33,6 +33,8 @@ Open-source platform for internal tools, workflows, API integrations, background
 - **Operator write rights**: `docs/operator-write-rights.md` — which `operator_settings` flags are
   enforced rather than cosmetic, and why a right that is granted-unless-withdrawn needs `Option`
   fields and a jsonb merge rather than a serde default
+- **Operator builder rights**: `docs/operator-builder-rights.md` — the workspace setting that lets
+  operators compose flows, what the composition check must cover, and why it costs a full seat
 - **Auth surface**: `docs/auth-surface.md` — credential precedence, session/cache invalidation
   scope, which token labels email their owner at expiry, how OAuth login matches `login_type`, and
   that every superadmin route refuses `$WM_TOKEN`. Read before designing anything that creates

backend/.sqlx/query-094a07eaa5714ec739786717f110ea539e84c33de23476e042a4a1a7cb529dac.json

@@ -0,0 +1,34 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT COALESCE((operator_settings->>'builder_flows')::boolean, false) AS \"flows!\",\n                COALESCE((operator_settings->>'manage_schedules')::boolean, true) AS \"schedules!\",\n                COALESCE((operator_settings->>'manage_triggers')::boolean, true) AS \"triggers!\"\n         FROM workspace_settings WHERE workspace_id = $1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "flows!",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 1,
+        "name": "schedules!",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 2,
+        "name": "triggers!",
+        "type_info": "Bool"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      null,
+      null,
+      null
+    ]
+  },
+  "hash": "094a07eaa5714ec739786717f110ea539e84c33de23476e042a4a1a7cb529dac"
+}

backend/.sqlx/query-298e0e30afdc973c517a9b4ea99f5dd1616a62c6ecbaaea312c686038d21e6fd.json

@@ -0,0 +1,24 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT EXISTS(SELECT 1 FROM script WHERE workspace_id = $1 AND path = $2 AND hash = $3)",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "exists",
+        "type_info": "Bool"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Int8"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "298e0e30afdc973c517a9b4ea99f5dd1616a62c6ecbaaea312c686038d21e6fd"
+}

backend/.sqlx/query-3dccc8e745f4a0973541088f66172e74af6828c1ab52cf7a6fd10b305deade85.json

@@ -0,0 +1,23 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT EXISTS(SELECT 1 FROM flow WHERE workspace_id = $1 AND path = $2)",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "exists",
+        "type_info": "Bool"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "3dccc8e745f4a0973541088f66172e74af6828c1ab52cf7a6fd10b305deade85"
+}

backend/.sqlx/query-4bc47050c74a02ab3169c3165898b2af07e995de71564d867b171f97f719fbde.json

@@ -0,0 +1,23 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT EXISTS(SELECT 1 FROM script WHERE workspace_id = $1 AND path = $2)",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "exists",
+        "type_info": "Bool"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "4bc47050c74a02ab3169c3165898b2af07e995de71564d867b171f97f719fbde"
+}

backend/.sqlx/query-4c5ca111c2ebee9733025cbe74d94a903d509d23b202982e39cbed4cb60108ef.json

@@ -0,0 +1,38 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "WITH all_audit AS (SELECT username, operation, timestamp FROM audit_partitioned UNION ALL SELECT username, operation, timestamp FROM audit),\n        active_users as (SELECT distinct username as email FROM all_audit WHERE timestamp > NOW() - INTERVAL '1 month' AND (operation = 'users.login' OR operation = 'oauth.login' OR operation = 'users.token.refresh')),\n        active_authors as (SELECT distinct t.email FROM usr t LEFT JOIN workspace_settings ws ON ws.workspace_id = t.workspace_id WHERE NOT (t.operator AND NOT COALESCE((ws.operator_settings->>'builder_flows')::boolean, false)) AND t.email IN (SELECT email FROM active_users)),\n        active_authors_agg as (SELECT array_agg(email) as authors FROM active_authors),\n        active_ops_agg as (SELECT array_agg(email) as operators from active_users WHERE email NOT IN (SELECT email FROM active_authors))\n        SELECT active_authors_agg.authors, active_ops_agg.operators, array_length(active_authors_agg.authors, 1) as author_count, array_length(active_ops_agg.operators, 1) as operator_count FROM active_authors_agg, active_ops_agg",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "authors",
+        "type_info": "VarcharArray"
+      },
+      {
+        "ordinal": 1,
+        "name": "operators",
+        "type_info": "VarcharArray"
+      },
+      {
+        "ordinal": 2,
+        "name": "author_count",
+        "type_info": "Int4"
+      },
+      {
+        "ordinal": 3,
+        "name": "operator_count",
+        "type_info": "Int4"
+      }
+    ],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": [
+      null,
+      null,
+      null,
+      null
+    ]
+  },
+  "hash": "4c5ca111c2ebee9733025cbe74d94a903d509d23b202982e39cbed4cb60108ef"
+}

backend/ee-repo-ref.txt

@@ -1 +1 @@
-bc3ef08c8e4233508c023e6ee847a3cd0b8be43b
+3c7964b87197daf3cb36d39e42390078c77388b4

backend/windmill-api-flows/src/flows.rs

@@ -16,10 +16,12 @@ use axum::{
 };
 use windmill_api_auth::{
     auth::{list_tokens_internal, TruncatedTokenWithEmail},
-    build_scope_path_predicate, check_scopes, maybe_refresh_folders, require_owner_of_path,
-    ApiAuthed,
+    build_scope_path_predicate, check_scopes, get_scope_tags, maybe_refresh_folders,
+    require_owner_of_path, ApiAuthed,
+};
+use windmill_common::workspaces::{
+    check_deploy_rules, check_operator_can_build_flows, RuleCheckResult,
 };
-use windmill_common::workspaces::{check_deploy_rules, RuleCheckResult};
 use windmill_common::{
     user_drafts::{overlay_or_draft_only, DraftUserRef, UserDraftItemKind, WithDraftOverlay},
     utils::HTTP_CLIENT,
@@ -35,7 +37,7 @@ use sqlx::{FromRow, Postgres, Transaction};
 use windmill_audit::audit_oss::{audit_log, AuditAuthorable};
 use windmill_audit::ActionKind;
 use windmill_common::assets::{clear_static_asset_usage, AssetUsageKind};
-use windmill_common::flows::FlowModule;
+use windmill_common::flows::{FlowModule, FlowValue};
 use windmill_common::min_version::{
     MIN_VERSION_SUPPORTS_DEBOUNCING, MIN_VERSION_SUPPORTS_DEBOUNCING_V2,
     MIN_VERSION_SUPPORTS_NODE_DEBOUNCING,
@@ -567,7 +569,13 @@ async fn list_paths_linking_agent(
     Ok(Json(flows))
 }
 
-async fn validate_flow(new_flow: &NewFlow) -> error::Result<()> {
+async fn validate_flow(
+    new_flow: &NewFlow,
+    authed: &ApiAuthed,
+    db: &DB,
+    user_db: &UserDB,
+    w_id: &str,
+) -> error::Result<()> {
     #[cfg(not(feature = "enterprise"))]
     if new_flow.ws_error_handler_muted.is_some_and(|val| val) {
         return Err(Error::BadRequest(
@@ -578,9 +586,113 @@ async fn validate_flow(new_flow: &NewFlow) -> error::Result<()> {
 
     guard_flow_from_debounce_data(new_flow).await?;
 
+    if authed.is_operator {
+        validate_operator_composed_flow(
+            &new_flow.parse_flow_value()?,
+            &new_flow.tag,
+            authed,
+            db,
+            user_db,
+            w_id,
+        )
+        .await?;
+    }
+
     return Ok(());
 }
 
+/// Runs on every write and every preview of a flow authored by an operator with builder rights.
+/// The walk in `check_flow_is_composition_only` only sees the value; what it collects is
+/// authorized here against the caller's own permissions.
+pub async fn validate_operator_composed_flow(
+    value: &FlowValue,
+    flow_tag: &Option<String>,
+    authed: &ApiAuthed,
+    db: &DB,
+    user_db: &UserDB,
+    w_id: &str,
+) -> error::Result<()> {
+    let mut refs = windmill_common::flows::check_flow_is_composition_only(value)?;
+
+    // A tag is how a step picks the worker group it runs on: unauthorized, a builder could route
+    // a job onto a privileged one.
+    refs.tags.extend(flow_tag.clone());
+    if refs.tags.iter().any(|t| !t.is_empty()) {
+        // Job-aware: a WM_TOKEN running as a superadmin must not unlock restricted tags.
+        let is_super_admin = windmill_api_auth::is_super_admin_authed(db, authed).await?;
+        for tag in refs.tags.iter().filter(|t| !t.is_empty()) {
+            windmill_common::jobs::check_tag_available_for_workspace_internal(
+                db,
+                w_id,
+                tag,
+                is_super_admin,
+                get_scope_tags(authed),
+            )
+            .await?;
+        }
+    }
+
+    if refs.runnables.is_empty() && refs.pinned_scripts.is_empty() {
+        return Ok(());
+    }
+    // A flow can step through the same script thirty times; this runs on every write, preview and
+    // dependency job.
+    refs.runnables.sort();
+    refs.runnables.dedup();
+    refs.pinned_scripts
+        .sort_by_key(|(path, hash)| (path.clone(), hash.0));
+    refs.pinned_scripts
+        .dedup_by_key(|(path, hash)| (path.clone(), hash.0));
+    // Composing a runnable is enough to run it: the worker resolves a step's path with the root DB
+    // handle and adopts that runnable's `on_behalf_of`, so an unreadable path would let a builder
+    // execute code it cannot see, as whoever that code runs as. RLS on this transaction is the
+    // check. A pinned `hash` needs its own comparison on top: the dispatch ignores the path beside
+    // it, so a readable path paired with another script's hash still runs that other script.
+    let mut tx = user_db.clone().begin(authed).await?;
+    for (is_flow, path) in &refs.runnables {
+        let readable = if *is_flow {
+            sqlx::query_scalar!(
+                "SELECT EXISTS(SELECT 1 FROM flow WHERE workspace_id = $1 AND path = $2)",
+                w_id,
+                path,
+            )
+        } else {
+            sqlx::query_scalar!(
+                "SELECT EXISTS(SELECT 1 FROM script WHERE workspace_id = $1 AND path = $2)",
+                w_id,
+                path,
+            )
+        }
+        .fetch_one(&mut *tx)
+        .await?
+        .unwrap_or(false);
+        if !readable {
+            return Err(Error::NotAuthorized(format!(
+                "{} {path} does not exist or is not readable by you",
+                if *is_flow { "Flow" } else { "Script" }
+            )));
+        }
+    }
+    for (path, hash) in &refs.pinned_scripts {
+        let exists = sqlx::query_scalar!(
+            "SELECT EXISTS(SELECT 1 FROM script WHERE workspace_id = $1 AND path = $2 AND hash = $3)",
+            w_id,
+            path,
+            hash.0,
+        )
+        .fetch_one(&mut *tx)
+        .await?
+        .unwrap_or(false);
+        if !exists {
+            return Err(Error::NotAuthorized(format!(
+                "Version {hash} is not a readable version of {path}"
+            )));
+        }
+    }
+    tx.commit().await?;
+    Ok(())
+}
+
 async fn create_flow(
     authed: ApiAuthed,
     Extension(db): Extension<DB>,
@@ -589,11 +701,13 @@ async fn create_flow(
     Path(w_id): Path<String>,
     Json(mut nf): Json<NewFlow>,
 ) -> Result<(StatusCode, String)> {
-    if authed.is_operator {
-        return Err(Error::NotAuthorized(
-            "Operators cannot create flows for security reasons".to_string(),
-        ));
-    }
+    check_operator_can_build_flows(
+        &db,
+        &w_id,
+        authed.is_operator,
+        "create flows",
+    )
+    .await?;
     check_scopes(&authed, || format!("flows:write:{}", nf.path))?;
 
     // A `<= 0` flow timeout is "unset", not a 0-second limit that kills every run instantly.
@@ -613,7 +727,7 @@ async fn create_flow(
         return Err(Error::PermissionDenied(msg));
     }
 
-    validate_flow(&nf).await?;
+    validate_flow(&nf, &authed, &db, &user_db, &w_id).await?;
     if *CLOUD_HOSTED {
         let nb_flows =
             sqlx::query_scalar!("SELECT COUNT(*) FROM flow WHERE workspace_id = $1", &w_id)
@@ -1165,11 +1279,13 @@ async fn update_flow(
     Path((w_id, flow_path)): Path<(String, StripPath)>,
     Json(ef): Json<EditFlow>,
 ) -> Result<String> {
-    if authed.is_operator {
-        return Err(Error::NotAuthorized(
-            "Operators cannot update flows for security reasons".to_string(),
-        ));
-    }
+    check_operator_can_build_flows(
+        &db,
+        &w_id,
+        authed.is_operator,
+        "update flows",
+    )
+    .await?;
     let flow_path = flow_path.to_path();
     // The URL identifies the flow being updated; the body path is only needed to rename.
     let mut nf = ef.into_new_flow(flow_path);
@@ -1196,7 +1312,7 @@ async fn update_flow(
         return Err(Error::PermissionDenied(msg));
     }
 
-    validate_flow(&nf).await?;
+    validate_flow(&nf, &authed, &db, &user_db, &w_id).await?;
 
     let authed = maybe_refresh_folders(&flow_path, &w_id, authed, &db).await;
     let mut tx = user_db.clone().begin(&authed).await?;
@@ -1819,11 +1935,13 @@ async fn archive_flow_by_path(
     Path((w_id, path)): Path<(String, StripPath)>,
     Json(archived): Json<Archived>,
 ) -> Result<String> {
-    if authed.is_operator {
-        return Err(Error::NotAuthorized(
-            "Operators cannot archive flows for security reasons".to_string(),
-        ));
-    }
+    check_operator_can_build_flows(
+        &db,
+        &w_id,
+        authed.is_operator,
+        "archive flows",
+    )
+    .await?;
     let path = path.to_path();
     check_scopes(&authed, || format!("flows:write:{}", path))?;
     if let RuleCheckResult::Blocked(msg) = check_deploy_rules(
@@ -1964,11 +2082,13 @@ async fn delete_flow_by_path(
     Path((w_id, path)): Path<(String, StripPath)>,
     Query(query): Query<DeleteFlowQuery>,
 ) -> Result<String> {
-    if authed.is_operator {
-        return Err(Error::NotAuthorized(
-            "Operators cannot delete flows for security reasons".to_string(),
-        ));
-    }
+    check_operator_can_build_flows(
+        &db,
+        &w_id,
+        authed.is_operator,
+        "delete flows",
+    )
+    .await?;
     let path = path.to_path();
     check_scopes(&authed, || format!("flows:write:{}", path))?;
     if let RuleCheckResult::Blocked(msg) = check_deploy_rules(

backend/windmill-api-integration-tests/tests/operator_builder_flows.rs

@@ -0,0 +1,213 @@
+use serde_json::json;
+use sqlx::{Pool, Postgres};
+use windmill_common::workspaces::invalidate_operator_rights_cache;
+use windmill_test_utils::*;
+
+const WS: &str = "test-workspace";
+
+fn operator_client() -> reqwest::Client {
+    let mut headers = reqwest::header::HeaderMap::new();
+    headers.insert(
+        reqwest::header::AUTHORIZATION,
+        reqwest::header::HeaderValue::from_str("Bearer OPERATOR_TOKEN_1").unwrap(),
+    );
+    reqwest::ClientBuilder::new()
+        .default_headers(headers)
+        .build()
+        .unwrap()
+}
+
+async fn set_builder(db: &Pool<Postgres>, flows: bool) -> anyhow::Result<()> {
+    sqlx::query(
+        "UPDATE workspace_settings SET operator_settings = $1::text::jsonb WHERE workspace_id = $2",
+    )
+    .bind(format!(r#"{{"builder_flows": {flows}}}"#))
+    .bind(WS)
+    .execute(db)
+    .await?;
+    // The right is read through a process-global 60s cache keyed by workspace id.
+    invalidate_operator_rights_cache(WS);
+    Ok(())
+}
+
+async fn add_script(db: &Pool<Postgres>, hash: i64, path: &str, owner: &str) -> anyhow::Result<()> {
+    sqlx::query(
+        "INSERT INTO script (workspace_id, hash, path, content, language, kind, created_by, schema,
+             summary, description, lock, extra_perms)
+         VALUES ($1, $2, $3, 'x', 'bun', 'script', $4, '{}', '', '', '', '{}')",
+    )
+    .bind(WS)
+    .bind(hash)
+    .bind(path)
+    .bind(owner)
+    .execute(db)
+    .await?;
+    Ok(())
+}
+
+fn composition_flow_at(path: &str, step_path: &str) -> serde_json::Value {
+    json!({
+        "path": path,
+        "summary": "",
+        "description": "",
+        "schema": {},
+        "value": {"modules": [{
+            "id": "a",
+            "value": {"type": "script", "path": step_path, "input_transforms": {}}
+        }]}
+    })
+}
+
+fn inline_code_flow(path: &str) -> serde_json::Value {
+    json!({
+        "path": path,
+        "summary": "",
+        "description": "",
+        "schema": {},
+        "value": {"modules": [{
+            "id": "a",
+            "value": {
+                "type": "rawscript",
+                "content": "export async function main() { return 1 }",
+                "language": "bun",
+                "input_transforms": {}
+            }
+        }]}
+    })
+}
+
+/// The whole boundary in one pass: the builder right lets an operator compose runnables that are
+/// already deployed and nothing more, and the endpoints that author code stay shut whether or not
+/// it is granted.
+#[sqlx::test(migrations = "../migrations", fixtures("base", "permissions_test"))]
+async fn test_operator_builder_flows_boundary(db: Pool<Postgres>) -> anyhow::Result<()> {
+    initialize_tracing().await;
+    let server = ApiServer::start(db.clone()).await?;
+    let port = server.addr.port();
+    let api = format!("http://localhost:{port}/api/w/{WS}");
+    let c = operator_client();
+
+    // A composition-only flow references a runnable that exists and the builder can read, so the
+    // fixture needs one.
+    add_script(&db, 4241, "u/operator/some_script", "operator").await?;
+
+    set_builder(&db, false).await?;
+    let resp = c
+        .post(format!("{api}/flows/create"))
+        .json(&composition_flow_at(
+            "u/operator/f1",
+            "u/operator/some_script",
+        ))
+        .send()
+        .await?;
+    assert!(
+        !resp.status().is_success(),
+        "an operator without the builder right must not create a flow"
+    );
+
+    set_builder(&db, true).await?;
+
+    let resp = c
+        .post(format!("{api}/flows/create"))
+        .json(&composition_flow_at(
+            "u/operator/f1",
+            "u/operator/some_script",
+        ))
+        .send()
+        .await?;
+    assert!(
+        resp.status().is_success(),
+        "a builder must be able to create a composition-only flow: {}",
+        resp.text().await?
+    );
+
+    let resp = c
+        .post(format!("{api}/flows/create"))
+        .json(&inline_code_flow("u/operator/f2"))
+        .send()
+        .await?;
+    assert!(
+        !resp.status().is_success(),
+        "a builder must not deploy a flow carrying inline code"
+    );
+
+    // Same for the preview path, which runs a request-supplied flow value rather than a stored one.
+    let resp = c
+        .post(format!("{api}/jobs/run/preview_flow"))
+        .json(&json!({"value": inline_code_flow("u/operator/f2")["value"], "args": {}}))
+        .send()
+        .await?;
+    assert!(
+        !resp.status().is_success(),
+        "a builder must not preview a flow carrying inline code"
+    );
+
+    // Authoring code directly stays shut with the right granted.
+    let resp = c
+        .post(format!("{api}/scripts/create"))
+        .json(&json!({
+            "path": "u/operator/s1",
+            "summary": "",
+            "description": "",
+            "content": "export async function main() { return 1 }",
+            "language": "bun",
+            "is_template": false
+        }))
+        .send()
+        .await?;
+    assert!(
+        !resp.status().is_success(),
+        "a builder must not create a script"
+    );
+
+    // Composing a runnable is enough to run it: the worker resolves a step's path with the root DB
+    // handle and adopts that runnable's `on_behalf_of`. `permissions_test` gives the operator
+    // fixture no rights on `u/alice/**`.
+    add_script(&db, 4243, "u/alice/private", "alice").await?;
+    let resp = c
+        .post(format!("{api}/flows/create"))
+        .json(&composition_flow_at("u/operator/f4", "u/alice/private"))
+        .send()
+        .await?;
+    assert!(
+        !resp.status().is_success(),
+        "a builder must not compose a runnable it cannot read"
+    );
+
+    // A version-pinned step dispatches on its hash alone, so the pair must be real and readable:
+    // otherwise a builder pins the hash of a script it cannot reach and runs that instead.
+    add_script(&db, 4242, "u/operator/pinned", "operator").await?;
+    let pinned = |hash: &str| {
+        json!({
+            "path": "u/operator/f3", "summary": "", "description": "", "schema": {},
+            "value": {"modules": [{
+                "id": "a",
+                "value": {
+                    "type": "script", "path": "u/operator/pinned", "hash": hash,
+                    "input_transforms": {}
+                }
+            }]}
+        })
+    };
+    let resp = c
+        .post(format!("{api}/flows/create"))
+        .json(&pinned("0000000000000000"))
+        .send()
+        .await?;
+    assert!(
+        !resp.status().is_success(),
+        "a builder must not pin a hash that is not a version of the step's path"
+    );
+    let resp = c
+        .post(format!("{api}/flows/create"))
+        .json(&pinned("0000000000001092"))
+        .send()
+        .await?;
+    assert!(
+        resp.status().is_success(),
+        "a builder must be able to pin the real version of a readable script: {}",
+        resp.text().await?
+    );
+
+    Ok(())
+}

backend/windmill-api-workspaces/src/workspaces.rs

@@ -9772,8 +9772,12 @@ async fn invite_user(
     nu.email = nu.email.to_lowercase();
 
     #[cfg(feature = "enterprise")]
-    if let Some(msg) =
-        windmill_common::ee_oss::check_seat_cap_for_new_user(&db, &nu.email, nu.operator).await?
+    if let Some(msg) = windmill_common::ee_oss::check_seat_cap_for_new_user(
+        &db,
+        &nu.email,
+        windmill_common::workspaces::consumes_operator_seat(&db, &w_id, nu.operator).await?,
+    )
+    .await?
     {
         return Err(Error::BadRequest(msg));
     }
@@ -9924,8 +9928,12 @@ async fn add_user(
     };
 
     #[cfg(feature = "enterprise")]
-    if let Some(msg) =
-        windmill_common::ee_oss::check_seat_cap_for_new_user(&db, &nu.email, nu.operator).await?
+    if let Some(msg) = windmill_common::ee_oss::check_seat_cap_for_new_user(
+        &db,
+        &nu.email,
+        windmill_common::workspaces::consumes_operator_seat(&db, &w_id, nu.operator).await?,
+    )
+    .await?
     {
         return Err(Error::BadRequest(msg));
     }
@@ -10488,6 +10496,11 @@ struct ChangeOperatorSettings {
     folders: bool,
     #[serde(default)]
     workers: bool,
+    /// Lets every operator of this workspace compose flows out of already-deployed runnables.
+    /// Unlike the visibility flags above this is a write right, and it makes each operator
+    /// consume a full author seat instead of half of one.
+    #[serde(default)]
+    builder_flows: bool,
     /// Writes operators may perform unless withdrawn, so `None` (key absent) must mean "leave as
     /// stored" rather than a value: the row is merged, not overwritten, and this endpoint takes
     /// whole-object payloads from git-sync files that predate the key. Defaulting either way here
@@ -10506,6 +10519,17 @@ async fn update_operator_settings(
 ) -> Result<String> {
     require_admin(authed.is_admin, &authed.username)?;
 
+    // Every operator of the workspace turns into a full seat, which an offline license may not
+    // cover. It is a no-op delta when the right is already on.
+    #[cfg(feature = "enterprise")]
+    if settings.builder_flows {
+        if let Some(msg) =
+            windmill_common::ee_oss::check_seat_cap_for_operator_builder(&db, &w_id).await?
+        {
+            return Err(Error::BadRequest(msg));
+        }
+    }
+
     let mut tx = db.begin().await?;
 
     let settings_json = serde_json::json!(settings);

backend/windmill-api/openapi.yaml

@@ -36584,6 +36584,9 @@ components:
         workers:
           type: boolean
           description: Whether operators can view workers page
+        builder_flows:
+          type: boolean
+          description: Whether operators can compose flows out of existing runnables (consumes a full seat)
         manage_schedules:
           type: boolean
           description: Whether operators can create, edit and delete schedules. Granted unless withdrawn; omitting the field leaves the stored value unchanged.

backend/windmill-api/src/drafts.rs

@@ -21,6 +21,7 @@ use windmill_common::{
     users::resolve_username_to_email,
     utils::strip_json_nul,
     variables::{build_crypt, encrypt},
+    workspaces::operator_can_build_flows,
 };
 
 pub fn workspaced_service() -> Router {
@@ -100,10 +101,10 @@ async fn list_drafts(
     Path(w_id): Path<String>,
     Query(query): Query<ListDraftsQuery>,
 ) -> Result<Json<Vec<DraftListItem>>> {
-    // Operators have no drafts of their own (they can't write any, see
-    // `require_can_write_path`), so this list is always empty for them. They
-    // can still READ some collaborators' drafts via `/drafts/get`.
-    if authed.is_operator {
+    // Without builder rights an operator has no drafts of their own (they can't write any, see
+    // `require_can_write_path`), so this list is always empty for them. They can still READ some
+    // collaborators' drafts via `/drafts/get`.
+    if authed.is_operator && !operator_can_build_flows(&db, &w_id).await? {
         return Ok(Json(vec![]));
     }
     let all_users = query.all_users.unwrap_or(false);
@@ -686,14 +687,19 @@ async fn require_can_write_path(
     if authed.is_admin {
         return Ok(());
     }
-    // Operators are read-only and never WRITE drafts. Read access is
-    // deliberately asymmetric: `require_can_read_path` has no operator block,
-    // so an operator can still READ a draft they can read via `/drafts/get`,
-    // mirroring their read access to deployed content. Intended.
+    // Operators are read-only and never WRITE drafts, except a flow draft where the workspace
+    // granted the builder right: the kind has to be checked, or the right would open drafts of
+    // kinds it says nothing about. Read access is deliberately asymmetric:
+    // `require_can_read_path` has no operator block, so an operator can still READ a draft they
+    // can read via `/drafts/get`, mirroring their read access to deployed content. Intended.
     if authed.is_operator {
-        return Err(Error::NotAuthorized(
-            "operators cannot save drafts".to_string(),
-        ));
+        let granted =
+            matches!(kind, UserDraftItemKind::Flow) && operator_can_build_flows(db, w_id).await?;
+        if !granted {
+            return Err(Error::NotAuthorized(
+                "operators cannot save drafts".to_string(),
+            ));
+        }
     }
     // Cheap claim-based namespace checks first: they evaluate the same JWT
     // claims RLS reads, so the outcome matches the policies while sparing the

backend/windmill-api/src/jobs.rs

@@ -26,6 +26,7 @@ use std::time::Instant;
 use tokio::io::AsyncReadExt;
 use tower::ServiceBuilder;
 use url::Url;
+use windmill_api_flows::flows::validate_operator_composed_flow;
 use windmill_common::assets::AssetUsageAccessType;
 use windmill_common::auth::TOKEN_PREFIX_LEN;
 #[cfg(feature = "run_inline")]
@@ -53,7 +54,9 @@ use windmill_common::worker::{Connection, CLOUD_HOSTED, WINDMILL_DIR};
 use windmill_common::workspace_dependencies::{
     RawWorkspaceDependencies, MIN_VERSION_WORKSPACE_DEPENDENCIES,
 };
-use windmill_common::workspaces::{check_user_against_rule, ProtectionRuleKind, RuleCheckResult};
+use windmill_common::workspaces::{
+    check_operator_can_build_flows, check_user_against_rule, ProtectionRuleKind, RuleCheckResult,
+};
 use windmill_common::DYNAMIC_INPUT_CACHE;
 #[cfg(all(feature = "enterprise", feature = "instance_smtp"))]
 use windmill_common::{email_oss::send_email_html, server::load_smtp_config};
@@ -9108,14 +9111,17 @@ pub struct RunFlowDependenciesResponse {
 async fn push_flow_dependencies_job(
     authed: &ApiAuthed,
     db: &DB,
+    user_db: &UserDB,
     w_id: &str,
     req: RunFlowDependenciesRequest,
 ) -> error::Result<Uuid> {
     check_scopes(authed, || format!("jobs:run"))?;
+    check_operator_can_build_flows(db, w_id, authed.is_operator, "run dependencies jobs").await?;
+    // The dependency job locks whatever inline code this request carries, on a worker. A
+    // composition-only flow has none, so validating here costs a builder nothing and keeps the
+    // lock step from becoming the way to run code the write path refuses.
     if authed.is_operator {
-        return Err(error::Error::NotAuthorized(
-            "Operators cannot run dependencies jobs for security reasons".to_string(),
-        ));
+        validate_operator_composed_flow(&req.flow_value, &None, authed, db, user_db, w_id).await?;
     }
 
     if req.raw_deps.is_some() {
@@ -9181,20 +9187,22 @@ async fn push_flow_dependencies_job(
 async fn run_flow_dependencies_job(
     authed: ApiAuthed,
     Extension(db): Extension<DB>,
+    Extension(user_db): Extension<UserDB>,
     Path(w_id): Path<String>,
     Json(req): Json<RunFlowDependenciesRequest>,
 ) -> error::Result<Response> {
-    let uuid = push_flow_dependencies_job(&authed, &db, &w_id, req).await?;
+    let uuid = push_flow_dependencies_job(&authed, &db, &user_db, &w_id, req).await?;
     run_wait_result(&db, uuid, &w_id, None, false, &authed.username).await
 }
 
 async fn run_flow_dependencies_job_async(
     authed: ApiAuthed,
     Extension(db): Extension<DB>,
+    Extension(user_db): Extension<UserDB>,
     Path(w_id): Path<String>,
     Json(req): Json<RunFlowDependenciesRequest>,
 ) -> error::Result<(StatusCode, String)> {
-    let uuid = push_flow_dependencies_job(&authed, &db, &w_id, req).await?;
+    let uuid = push_flow_dependencies_job(&authed, &db, &user_db, &w_id, req).await?;
     Ok((StatusCode::CREATED, uuid.to_string()))
 }
 
@@ -9495,15 +9503,24 @@ async fn run_preview_flow_job(
     Query(run_query): Query<RunJobQuery>,
     Json(raw_flow): Json<PreviewFlow>,
 ) -> error::Result<(StatusCode, String)> {
-    if authed.is_operator {
-        return Err(error::Error::NotAuthorized(
-            "Operators cannot run preview jobs for security reasons".to_string(),
-        ));
-    }
+    check_operator_can_build_flows(&db, &w_id, authed.is_operator, "run preview jobs").await?;
     // Flow preview runs an arbitrary, request-supplied flow definition; require the broad
     // jobs:run scope so a narrowly-scoped token cannot escape its scope. See run_preview_script.
     check_scopes(&authed, || format!("jobs:run"))?;
     require_path_read_access_for_preview(&authed, &raw_flow.path)?;
+    // A builder must be able to test what it composes, but the submitted value is not the stored
+    // one: without this the preview is a way to run inline code the write path refuses.
+    if authed.is_operator {
+        validate_operator_composed_flow(
+            &raw_flow.value,
+            &raw_flow.tag,
+            &authed,
+            &db,
+            &user_db,
+            &w_id,
+        )
+        .await?;
+    }
     let scheduled_for = run_query.get_scheduled_for(&db).await?;
     let tag = run_query.tag.clone().or(raw_flow.tag.clone());
     check_tag_available_for_workspace(&db, &w_id, &tag, &authed).await?;

backend/windmill-common/src/ee_oss.rs

@@ -71,6 +71,14 @@ pub async fn check_seat_cap_for_reactivation(
     Ok(None)
 }
 
+#[cfg(all(feature = "enterprise", not(feature = "private")))]
+pub async fn check_seat_cap_for_operator_builder(
+    _db: &DB,
+    _w_id: &str,
+) -> anyhow::Result<Option<String>> {
+    Ok(None)
+}
+
 #[cfg(all(feature = "enterprise", not(feature = "private")))]
 pub async fn compute_instance_hash(_db: &DB) -> anyhow::Result<Option<String>> {
     // Implementation is not open source

backend/windmill-common/src/flows.rs

@@ -17,6 +17,7 @@ use crate::{
     cache::{self, FlowExtras},
     db::DB,
     error::{to_anyhow, Error},
+    scripts::ScriptHash,
     utils::{http_get_from_hub, StripPath},
     worker::{to_raw_value, Connection},
     DEFAULT_HUB_BASE_URL, HUB_BASE_URL, PRIVATE_HUB_MIN_VERSION,
@@ -246,6 +247,159 @@ pub async fn resolve_modules(
     Ok(())
 }
 
+/// Checks a flow value contains nothing but composition of runnables that already exist, which is
+/// all an operator with builder rights may author. Walks the modules, the preprocessor and failure
+/// modules, every branch, and the `tools` of an AI agent step.
+///
+/// Returns what the caller still has to authorize against its own permissions, which this
+/// value-only walk cannot: every runnable the steps reference, the worker tags they pin, and the
+/// `(path, hash)` pairs of version-pinned steps. See [`ComposedFlowRefs`] for why each one is not
+/// already settled by the walk.
+pub fn check_flow_is_composition_only(value: &FlowValue) -> Result<ComposedFlowRefs, Error> {
+    let mut refs = ComposedFlowRefs::default();
+    for module in value
+        .modules
+        .iter()
+        .chain(value.preprocessor_module.as_deref())
+        .chain(value.failure_module.as_deref())
+    {
+        check_module_is_composition_only(module, &mut refs)?;
+    }
+    Ok(refs)
+}
+
+/// What [`check_flow_is_composition_only`] collects for the caller to authorize.
+#[derive(Default)]
+pub struct ComposedFlowRefs {
+    pub tags: Vec<String>,
+    /// Every workspace runnable a step references, as `(is_flow, path)`. The worker resolves
+    /// these with the root DB handle and adopts the referenced runnable's `on_behalf_of`, so
+    /// composing a path is enough to run it, and to run it as whoever it runs as.
+    pub runnables: Vec<(bool, String)>,
+    /// Version-pinned script steps. A step carrying a `hash` is dispatched by that hash alone,
+    /// with the path beside it ignored, so the pair has to be checked on top of the path.
+    pub pinned_scripts: Vec<(String, ScriptHash)>,
+}
+
+fn check_module_is_composition_only(
+    module: &FlowModule,
+    refs: &mut ComposedFlowRefs,
+) -> Result<(), Error> {
+    let value = module
+        .get_value()
+        .map_err(|e| Error::BadRequest(format!("Step {} could not be read: {e}", module.id)))?;
+    check_module_value_is_composition_only(&value, &module.id, refs)
+}
+
+fn check_module_value_is_composition_only(
+    value: &FlowModuleValue,
+    id: &str,
+    refs: &mut ComposedFlowRefs,
+) -> Result<(), Error> {
+    let refuse = |what: &str| {
+        Err(Error::NotAuthorized(format!(
+            "Step {id}: {what}. Operators with builder rights compose runnables that are already \
+             deployed; they cannot author code."
+        )))
+    };
+    // A node id points at code stored in a `flow_node` row. Only the dependency job produces them,
+    // by hoisting a step's code out of the flow value, so an authored value carrying one names
+    // code that belongs to some other flow. `modules` is what the walk below covers, and an
+    // editor payload comes from the un-hoisted `flow_version.value`, so refusing them costs
+    // nothing legitimate.
+    let refuse_node = |node: &Option<FlowNodeId>| match node {
+        Some(_) => refuse("references code stored outside the flow"),
+        None => Ok(()),
+    };
+    let mut push_tag = |tag: &Option<String>| {
+        if let Some(tag) = tag.as_deref().filter(|t| !t.is_empty()) {
+            refs.tags.push(tag.to_string());
+        }
+    };
+
+    match value {
+        FlowModuleValue::RawScript { .. } => return refuse("has inline code"),
+        FlowModuleValue::FlowScript { .. } => {
+            return refuse("references code stored outside the flow")
+        }
+        FlowModuleValue::Identity => {}
+        FlowModuleValue::Script { path, hash, tag_override, .. } => {
+            check_composable_path(path, id)?;
+            push_tag(tag_override);
+            refs.runnables.push((false, path.clone()));
+            if let Some(hash) = hash {
+                refs.pinned_scripts.push((path.clone(), *hash));
+            }
+        }
+        FlowModuleValue::Flow { path, .. } => {
+            check_composable_path(path, id)?;
+            refs.runnables.push((true, path.clone()));
+        }
+        FlowModuleValue::ForloopFlow { modules, modules_node, .. }
+        | FlowModuleValue::WhileloopFlow { modules, modules_node, .. } => {
+            refuse_node(modules_node)?;
+            for module in modules {
+                check_module_is_composition_only(module, refs)?;
+            }
+        }
+        FlowModuleValue::BranchOne { branches, default, default_node } => {
+            refuse_node(default_node)?;
+            for module in default {
+                check_module_is_composition_only(module, refs)?;
+            }
+            check_branches_are_composition_only(branches, id, refs)?;
+        }
+        FlowModuleValue::BranchAll { branches, .. } => {
+            check_branches_are_composition_only(branches, id, refs)?
+        }
+        FlowModuleValue::AIAgent { tools, tag, agent, .. } => {
+            // A linked agent resolves its tools from an `ai_agent` resource at run time, and
+            // operators may write resources, so those tools are outside this check: the list can
+            // be swapped for a raw script after the flow is deployed.
+            if agent.is_some() {
+                return refuse("links an AI agent resource, whose tools live outside the flow");
+            }
+            push_tag(tag);
+            for tool in tools {
+                if let ToolValue::FlowModule(value) = &tool.value {
+                    check_module_value_is_composition_only(value, &tool.id, refs)?;
+                }
+            }
+        }
+    }
+    Ok(())
+}
+
+fn check_branches_are_composition_only(
+    branches: &[Branch],
+    id: &str,
+    refs: &mut ComposedFlowRefs,
+) -> Result<(), Error> {
+    for branch in branches {
+        if branch.modules_node.is_some() {
+            return Err(Error::NotAuthorized(format!(
+                "Step {id}: a branch references code stored outside the flow. Operators with \
+                 builder rights compose runnables that are already deployed; they cannot author \
+                 code."
+            )));
+        }
+        for module in &branch.modules {
+            check_module_is_composition_only(module, refs)?;
+        }
+    }
+    Ok(())
+}
+
+fn check_composable_path(path: &str, id: &str) -> Result<(), Error> {
+    if path.starts_with("hub/") {
+        return Err(Error::NotAuthorized(format!(
+            "Step {id}: hub runnables are not available to operators with builder rights. Deploy \
+             it to the workspace first."
+        )));
+    }
+    Ok(())
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -281,4 +435,95 @@ mod tests {
         let err = extract_hub_flow_id_from_path("hub/flows/0").unwrap_err();
         assert!(matches!(err, Error::BadRequest(_)));
     }
+
+    fn flow(value: serde_json::Value) -> FlowValue {
+        serde_json::from_value(value).unwrap()
+    }
+
+    #[test]
+    fn composition_check_accepts_a_composed_flow_and_collects_its_tags() {
+        let refs = check_flow_is_composition_only(&flow(serde_json::json!({"modules": [{
+            "id": "a",
+            "value": {"type": "forloopflow", "iterator": {"type": "static", "value": []},
+                "parallel": false, "modules": [
+                    {"id": "b", "value": {"type": "script", "path": "f/x/s", "tag_override": "gpu"}},
+                    {"id": "c", "value": {"type": "flow", "path": "f/x/f"}},
+                    {"id": "d", "value": {"type": "aiagent", "input_transforms": {}, "tag": "ai",
+                        "tools": [{"id": "t", "value": {"tool_type": "flowmodule",
+                            "type": "script", "path": "f/x/tool"}}]}}
+                ]}
+        }]})))
+        .unwrap();
+        assert_eq!(refs.tags, vec!["gpu".to_string(), "ai".to_string()]);
+    }
+
+    /// The walk covers `modules`, so a node reference is a way past it: it names code hoisted
+    /// into a `flow_node` row, possibly another flow's.
+    #[test]
+    fn composition_check_rejects_node_references() {
+        for value in [
+            serde_json::json!({"modules": [{"id": "a", "value": {"type": "forloopflow",
+                "iterator": {"type": "static", "value": []}, "parallel": false,
+                "modules": [], "modules_node": 7}}]}),
+            serde_json::json!({"modules": [{"id": "a", "value": {"type": "branchone",
+                "branches": [], "default": [], "default_node": 7}}]}),
+            serde_json::json!({"modules": [{"id": "a", "value": {"type": "branchall",
+                "branches": [{"expr": "true", "modules": [], "modules_node": 7}]}}]}),
+            serde_json::json!({"modules": [{"id": "a", "value": {"type": "flowscript",
+                "id": 7, "language": "bun"}}]}),
+        ] {
+            assert!(check_flow_is_composition_only(&flow(value)).is_err());
+        }
+    }
+
+    /// An agent tool wraps a whole `FlowModuleValue`, and a linked agent resolves its tools from a
+    /// resource an operator may rewrite after the flow is deployed.
+    #[test]
+    fn composition_check_rejects_code_reachable_through_an_ai_agent() {
+        for value in [
+            serde_json::json!({"modules": [{"id": "a", "value": {"type": "aiagent",
+                "input_transforms": {}, "tools": [{"id": "t", "value": {"tool_type": "flowmodule",
+                    "type": "rawscript", "content": "x", "language": "bun"}}]}}]}),
+            serde_json::json!({"modules": [{"id": "a", "value": {"type": "aiagent",
+                "input_transforms": {}, "tools": [], "agent": "$res:f/x/agent"}}]}),
+        ] {
+            assert!(check_flow_is_composition_only(&flow(value)).is_err());
+        }
+    }
+
+    #[test]
+    fn composition_check_rejects_code_in_every_module_slot() {
+        let inline = serde_json::json!({"type": "rawscript", "content": "x", "language": "bun"});
+        for value in [
+            serde_json::json!({"modules": [{"id": "a", "value": inline}]}),
+            serde_json::json!({"modules": [], "failure_module": {"id": "f", "value": inline}}),
+            serde_json::json!({"modules": [], "preprocessor_module": {"id": "p", "value": inline}}),
+        ] {
+            assert!(check_flow_is_composition_only(&flow(value)).is_err());
+        }
+    }
+
+    /// A step carrying a `hash` dispatches on that hash alone: the caller must verify the pair
+    /// exists and is readable, so the walk has to surface it rather than pass it through.
+    #[test]
+    fn composition_check_reports_version_pinned_steps() {
+        let refs = check_flow_is_composition_only(&flow(serde_json::json!({"modules": [
+            {"id": "a", "value": {"type": "script", "path": "f/x/s", "hash": "000000000000007b"}},
+            {"id": "b", "value": {"type": "script", "path": "f/x/t"}}
+        ]})))
+        .unwrap();
+        assert_eq!(
+            refs.pinned_scripts,
+            vec![("f/x/s".to_string(), ScriptHash(123))]
+        );
+    }
+
+    #[test]
+    fn composition_check_rejects_hub_runnables() {
+        for kind in ["script", "flow"] {
+            let value = serde_json::json!({"modules": [{"id": "a",
+                "value": {"type": kind, "path": "hub/1234/thing"}}]});
+            assert!(check_flow_is_composition_only(&flow(value)).is_err());
+        }
+    }
 }

backend/windmill-common/src/workspaces.rs

@@ -1189,7 +1189,17 @@ pub fn invalidate_protection_rules_cache(workspace_id: &str) {
 // Operator rights cache
 
 lazy_static::lazy_static! {
-    static ref OPERATOR_RIGHTS_CACHE: Cache<String, (OperatorManageRights, i64)> = Cache::new(1000);
+    static ref OPERATOR_RIGHTS_CACHE: Cache<String, (OperatorRights, i64)> = Cache::new(1000);
+}
+
+/// Every operator right of a workspace, read and cached together because they share one
+/// `operator_settings` column and one invalidation. The two groups have opposite polarity:
+/// `builder_flows` is granted on request and costs a seat, the `manage` rights are held by
+/// default and cost nothing.
+#[derive(Copy, Clone, Debug, Default, PartialEq, Eq)]
+pub struct OperatorRights {
+    pub builder_flows: bool,
+    pub manage: OperatorManageRights,
 }
 
 /// Writes an operator may perform unless the workspace withdraws them. Unlike the visibility
@@ -1236,7 +1246,7 @@ impl OperatorManageRights {
 ///
 /// Call it before opening an RLS transaction: it takes a connection from the root pool, and a
 /// second pooled connection held alongside a transaction self-deadlocks on a one-connection pool.
-pub async fn operator_manage_rights(db: &DB, workspace_id: &str) -> Result<OperatorManageRights> {
+pub async fn operator_rights(db: &DB, workspace_id: &str) -> Result<OperatorRights> {
     let now = chrono::Utc::now().timestamp();
 
     if let Some((rights, expiry)) = OPERATOR_RIGHTS_CACHE.get(workspace_id) {
@@ -1245,10 +1255,12 @@ pub async fn operator_manage_rights(db: &DB, workspace_id: &str) -> Result<Opera
         }
     }
 
-    // Coalesced to true, matching `OperatorManageRights::default`: an absent key means the
-    // workspace never configured the right, not that it withdrew it.
+    // The builder key defaults to false, a right the workspace has to grant; the manage keys to
+    // true, rights it has to withdraw. An absent key means "never configured" for both, so the
+    // defaults have to differ here rather than at the call sites.
     let row = sqlx::query!(
-        "SELECT COALESCE((operator_settings->>'manage_schedules')::boolean, true) AS \"schedules!\",
+        "SELECT COALESCE((operator_settings->>'builder_flows')::boolean, false) AS \"flows!\",
+                COALESCE((operator_settings->>'manage_schedules')::boolean, true) AS \"schedules!\",
                 COALESCE((operator_settings->>'manage_triggers')::boolean, true) AS \"triggers!\"
          FROM workspace_settings WHERE workspace_id = $1",
         workspace_id
@@ -1262,7 +1274,10 @@ pub async fn operator_manage_rights(db: &DB, workspace_id: &str) -> Result<Opera
     })?;
 
     let rights = row
-        .map(|r| OperatorManageRights { schedules: r.schedules, triggers: r.triggers })
+        .map(|r| OperatorRights {
+            builder_flows: r.flows,
+            manage: OperatorManageRights { schedules: r.schedules, triggers: r.triggers },
+        })
         .unwrap_or_default();
 
     OPERATOR_RIGHTS_CACHE.insert(workspace_id.to_string(), (rights, now + 60));
@@ -1270,6 +1285,42 @@ pub async fn operator_manage_rights(db: &DB, workspace_id: &str) -> Result<Opera
     Ok(rights)
 }
 
+pub async fn operator_manage_rights(db: &DB, workspace_id: &str) -> Result<OperatorManageRights> {
+    Ok(operator_rights(db, workspace_id).await?.manage)
+}
+
+/// Whether operators of this workspace may compose flows out of already-deployed runnables. Per
+/// workspace, not per user: every operator gets it, and consumes a full seat for it.
+pub async fn operator_can_build_flows(db: &DB, workspace_id: &str) -> Result<bool> {
+    Ok(operator_rights(db, workspace_id).await?.builder_flows)
+}
+
+/// Gate for a write only a workspace that granted builder rights lets operators perform. `action`
+/// completes "Operators cannot {action} for security reasons".
+pub async fn check_operator_can_build_flows(
+    db: &DB,
+    workspace_id: &str,
+    is_operator: bool,
+    action: &str,
+) -> Result<()> {
+    if is_operator && !operator_can_build_flows(db, workspace_id).await? {
+        return Err(Error::NotAuthorized(format!(
+            "Operators cannot {action} for security reasons"
+        )));
+    }
+    Ok(())
+}
+
+/// Whether a membership consumes an operator (half) seat rather than an author seat. A builder
+/// right makes an operator an author of deployable artifacts, so it weighs a full seat.
+pub async fn consumes_operator_seat(
+    db: &DB,
+    workspace_id: &str,
+    is_operator: bool,
+) -> Result<bool> {
+    Ok(is_operator && !operator_can_build_flows(db, workspace_id).await?)
+}
+
 /// Invalidate the operator rights cache for a workspace
 pub fn invalidate_operator_rights_cache(workspace_id: &str) {
     OPERATOR_RIGHTS_CACHE.remove(workspace_id);

docs/operator-builder-rights.md

@@ -0,0 +1,60 @@
+# Operator builder rights
+
+`operator_settings.builder_flows` lets every operator of a workspace compose flows out of runnables
+that already exist. It does not make them authors: the boundary the operator role draws is
+**authoring code and running arbitrary code**, and this does not move it.
+
+It is a write right, unlike the visibility flags beside it, and unlike the withdrawable rights in
+`docs/operator-write-rights.md` it is granted on request and costs a seat. Read it with
+`windmill_common::workspaces::operator_can_build_flows` (60s cache, shared with the withdrawable
+rights) and gate a write with `check_operator_can_build_flows`.
+
+## What the check has to cover
+
+`check_flow_is_composition_only` (`windmill-common/src/flows.rs`) walks a `FlowValue` and refuses
+anything that carries code. Three of its rules exist because the obvious walk misses them:
+
+- **`FlowScript` and any populated `modules_node` / `default_node`.** These name code hoisted into
+  a `flow_node` row. Only the dependency job produces them, so an authored value carrying one
+  names code stored under some other flow. The walk covers `modules`, so a node reference is a way
+  past it.
+- **An AI agent step's `tools`.** `ToolValue::FlowModule` wraps a whole `FlowModuleValue`, so a
+  tool can be a raw script.
+- **An AI agent step's `agent` link.** A linked agent resolves its tools from an `ai_agent`
+  resource at run time, and operators may write resources, so the tool list is outside this check
+  and can be swapped for a raw script after the flow is approved.
+
+It also returns what a value-only walk cannot authorize, for the caller to check against its own
+permissions:
+
+- **the worker tags the steps pin**, or a builder routes a job onto a privileged worker group;
+- **every runnable a step references**. `script_to_payload` resolves a step's path with the root DB
+  handle (`db_authed = None`) and returns the referenced runnable's `on_behalf_of`, which
+  `worker_flow` then applies to the step job. So composing a path is enough to run it, and to run
+  it as whoever it runs as: `validate_operator_composed_flow` re-checks each path under the
+  caller's RLS. This is the general case; the one below is on top of it, not instead of it.
+- **the `(path, hash)` of every version-pinned step**. A step carrying a `hash` is dispatched by
+  that hash alone, with the path beside it never consulted, so a readable path paired with another
+  script's hash still runs that other script.
+
+Call it on every write **and** every preview: `run_preview_flow_job` and
+`push_flow_dependencies_job` both take a request-supplied flow value, so leaving either out makes
+it the way to run what the write path refuses.
+
+## Billing
+
+An operator of a builder workspace consumes a full author seat: composing deployable artifacts
+makes them an author, and there is no half-author. `consumes_operator_seat` is the seat-role
+helper; the EE counting queries share `OPERATOR_SEAT_SQL` so the displayed, enforced and reported
+numbers agree.
+
+Granting the right runs `check_seat_cap_for_operator_builder`, which prices the change by counting
+seats twice rather than by counting the workspace's operators: an operator who already authors
+elsewhere must not be charged again, so re-saving settings that already have the right on is a
+zero delta and never blocks.
+
+## Accepted risks
+
+- All-or-nothing per workspace: there is no per-user builder role.
+- `operator_settings` is git-synced, so a pull can flip every operator's class in a workspace and
+  the billed seat count with it.

frontend/src/lib/components/FlowBuilder.svelte

@@ -19,6 +19,7 @@
 	} from '$lib/components/flows/linkedAgentToolsStore.svelte'
 	import {
 		enterpriseLicense,
+		operatorBuilderFlows,
 		userStore,
 		userWorkspaces,
 		workspaceStore,
@@ -1394,7 +1395,7 @@
 <AIChangesWarningModal bind:open={aiChangesWarningOpen} onConfirm={aiChangesConfirmCallback} />
 
 {#key renderCount}
-	{#if !$userStore?.operator}
+	{#if !$userStore?.operator || $operatorBuilderFlows}
 		{#if $pathStore}
 			<FlowHistory bind:this={flowHistory} path={$pathStore} {onHistoryRestore} />
 		{/if}
@@ -1548,7 +1549,9 @@
 					{forceTestTab}
 					{highlightArg}
 					aiChatOpen={aiChatManager.open}
-					showFlowAiButton={!disableAi && customUi?.topBar?.aiBuilder != false}
+					showFlowAiButton={!disableAi &&
+						customUi?.topBar?.aiBuilder != false &&
+						!$operatorBuilderFlows}
 					toggleAiChat={() => aiChatManager.toggleOpen()}
 					{sessionOpen}
 					onOpenPreview={flowPreviewButtons?.openPreview}
@@ -1577,7 +1580,9 @@
 			{/if}
 		</div>
 	{:else}
-		Flow Builder not available to operators
+		<div class="h-full w-full center-center text-sm text-secondary">
+			Flow builder not available to operators
+		</div>
 	{/if}
 {/key}
 

frontend/src/lib/components/common/table/FlowRow.svelte

@@ -8,7 +8,7 @@
 	import DraftBadge from '$lib/components/DraftBadge.svelte'
 	import type ShareModal from '$lib/components/ShareModal.svelte'
 	import { FlowService, type Flow } from '$lib/gen'
-	import { userStore, userWorkspaces, workspaceStore } from '$lib/stores'
+	import { operatorBuilderFlows, userStore, userWorkspaces, workspaceStore } from '$lib/stores'
 	import { UserDraftDbSyncer } from '$lib/userDraftDbSyncer.svelte'
 	import { createEventDispatcher } from 'svelte'
 	import Badge from '../badge/Badge.svelte'
@@ -237,7 +237,7 @@
 							// list endpoint only surfaces own/legacy draft-only rows), so
 							// discarding it never requires write permission on the path.
 							disabled: !showEditButton,
-							hide: $userStore?.operator
+							hide: $userStore?.operator && !$operatorBuilderFlows
 						}
 					]
 				}
@@ -252,7 +252,7 @@
 						icon: GitFork,
 						href: `${base}/flows/add?template=${path}`,
 						disabled: !showEditButton,
-						hide: $userStore?.operator
+						hide: $userStore?.operator && !$operatorBuilderFlows
 					},
 					{
 						displayName: editInForkLabel($workspaceStore, $userWorkspaces),
@@ -278,7 +278,7 @@
 							moveDrawer.openDrawer(path, flow.summary, 'flow')
 						},
 						disabled: !owner || archived || !canEdit,
-						hide: $userStore?.operator
+						hide: $userStore?.operator && !$operatorBuilderFlows
 					},
 					{
 						displayName: 'Copy path',
@@ -306,7 +306,7 @@
 						action: () => {
 							flowHistory?.open()
 						},
-						hide: $userStore?.operator
+						hide: $userStore?.operator && !$operatorBuilderFlows
 					},
 					{
 						displayName: 'Schedule',
@@ -315,7 +315,7 @@
 							scheduleEditor?.openNew(true, path)
 						},
 						disabled: archived,
-						hide: $userStore?.operator
+						hide: $userStore?.operator && !$operatorBuilderFlows
 					},
 					{
 						displayName: 'Permissions',
@@ -323,7 +323,7 @@
 						action: () => {
 							shareModal.openDrawer && shareModal.openDrawer(path, 'flow')
 						},
-						hide: $userStore?.operator
+						hide: $userStore?.operator && !$operatorBuilderFlows
 					},
 					{
 						displayName: archived ? 'Unarchive' : 'Archive',
@@ -333,7 +333,7 @@
 						},
 						type: 'delete',
 						disabled: !owner || !canEdit,
-						hide: $userStore?.operator
+						hide: $userStore?.operator && !$operatorBuilderFlows
 					},
 					{
 						displayName: 'Delete',
@@ -350,7 +350,7 @@
 						},
 						type: 'delete',
 						disabled: !owner || !canEdit,
-						hide: $userStore?.operator
+						hide: $userStore?.operator && !$operatorBuilderFlows
 					}
 				]
 			}}

frontend/src/lib/components/copilot/chat/AIChatManager.svelte.ts

@@ -154,6 +154,7 @@ import type { ArtifactVersionTarget } from '$lib/components/sessions/previewRout
 import { appendAttachedFilesRoster } from './files/fileTools'
 import { ENTER_PLAN_MODE_TOOL, EXIT_PLAN_MODE_TOOL } from './planMode'
 import { PlanModeController, type PlanModeHost } from './planModeController.svelte'
+import { operatorBuilderFlows } from '$lib/stores'
 
 // Compaction of the stored history: once the projected request size
 // (contextTokens — the provider's report when current, a fresh chars/4
@@ -283,6 +284,11 @@ export function supportsPlanMode(mode: AIMode): boolean {
 	return PLAN_MODES.has(mode)
 }
 
+// Rune mirror of the store: a class `$derived` cannot track a store read, and this one gates
+// which chat modes exist.
+const isOperatorBuilderFlows = $state({ val: false })
+operatorBuilderFlows.subscribe((v) => (isOperatorBuilderFlows.val = v))
+
 export function isAIModeVisible(mode: AIMode): boolean {
 	return mode !== AIMode.GLOBAL || isGlobalAiEnabled()
 }
@@ -1363,12 +1369,19 @@ export class AIChatManager implements ChatViewHost {
 			.map((s) => ({ ...s, kind: 'skill' as const }))
 	])
 
+	// The flow and script builders both write code, which the backend refuses from an operator
+	// with the builder right: leaving them reachable would only produce work that cannot be
+	// deployed.
 	allowedModes: Record<AIMode, boolean> = $derived({
 		script:
 			this.flowAiChatHelpers === undefined &&
 			this.scriptEditorOptions !== undefined &&
-			!this.disabledModes.script,
-		flow: this.flowAiChatHelpers !== undefined && !this.disabledModes.flow,
+			!this.disabledModes.script &&
+			!isOperatorBuilderFlows.val,
+		flow:
+			this.flowAiChatHelpers !== undefined &&
+			!this.disabledModes.flow &&
+			!isOperatorBuilderFlows.val,
 		app: this.appAiChatHelpers !== undefined && !this.disabledModes.app,
 		navigator: !this.disabledModes.navigator,
 		ask: !this.disabledModes.ask,

frontend/src/lib/components/flows/common/FlowCardHeader.svelte

@@ -19,7 +19,7 @@
 	import FlowPanelChrome from './FlowPanelChrome.svelte'
 	import type { FlowBuilderWhitelabelCustomUi } from '$lib/components/custom_ui'
 	import DropdownV2 from '$lib/components/DropdownV2.svelte'
-	import { hubBaseUrlStore, workspaceStore } from '$lib/stores'
+	import { hubBaseUrlStore, operatorBuilderFlows, workspaceStore } from '$lib/stores'
 	import { DEFAULT_HUB_BASE_URL, PRIVATE_HUB_MIN_VERSION } from '$lib/hub'
 	import { getLatestHashForScript } from '$lib/scripts'
 	import { sendUserToast, type Item } from '$lib/utils'
@@ -95,7 +95,7 @@
 	const scriptItems: Item[] = $derived.by(() => {
 		if (flowModuleValue?.type !== 'script') return []
 		const items: Item[] = []
-		if (!isHub && customUi?.scriptEdit != false) {
+		if (!isHub && customUi?.scriptEdit != false && !$operatorBuilderFlows) {
 			items.push({
 				displayName: "Edit the script's code",
 				icon: Pen,
@@ -139,7 +139,7 @@
 				})
 			}
 		}
-		if (customUi?.scriptFork != false) {
+		if (customUi?.scriptFork != false && !$operatorBuilderFlows) {
 			items.push({
 				displayName: 'Fork into an inline script',
 				icon: GitFork,

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

@@ -11,6 +11,7 @@
 		defaultScripts,
 		enterpriseLicense,
 		hubBaseUrlStore,
+		operatorBuilderFlows,
 		userStore,
 		workspaceStore
 	} from '$lib/stores'
@@ -153,7 +154,10 @@
 		preFilter: 'all' | 'workspace' | 'hub',
 		selectedKind: 'script' | 'flow' | 'approval' | 'trigger' | 'preprocessor' | 'failure'
 	) {
-		if (['script', 'trigger', 'failure', 'approval', 'preprocessor'].includes(selectedKind)) {
+		if (
+			!$operatorBuilderFlows &&
+			['script', 'trigger', 'failure', 'approval', 'preprocessor'].includes(selectedKind)
+		) {
 			if (!selected && preFilter == 'all') {
 				inlineScripts = langs.filter((lang) => {
 					return (
@@ -246,6 +250,7 @@
 	let showAiRows = $derived(
 		!disableAi &&
 			!$copilotInfo.workspaceDisabled &&
+			!$operatorBuilderFlows &&
 			funcDesc?.length > 0 &&
 			kind != 'failure' &&
 			kind != 'preprocessor' &&
@@ -274,9 +279,14 @@
 			preFilter === 'all' &&
 			!selected &&
 			customUi?.aiSandbox != false &&
+			!$operatorBuilderFlows &&
 			matchesAiSandbox
 	)
 
+	// Hub runnables carry code the workspace never reviewed, and the backend refuses them in a
+	// flow a builder authors, so the hub browser and its integration filters are not offered.
+	let showHub = $derived(!$operatorBuilderFlows)
+
 	// Every result row lives in one keyboard index space, and hovering a row moves that index, so
 	// mouse and keyboard can never highlight two different rows. Offsets follow the render order.
 	let inlineOffset = $derived(topLevelNodes.length)
@@ -329,7 +339,7 @@
 					{/if}
 				{/if}
 
-				{#if preFilter === 'hub' || preFilter === 'all'}
+				{#if showHub && (preFilter === 'hub' || preFilter === 'all')}
 					{#if preFilter == 'all'}
 						<div class="pb-0 text-2xs font-normal text-secondary ml-2 pt-1">Integrations</div>
 					{/if}
@@ -534,7 +544,7 @@
 				}}
 			/>
 		{/if}
-		{#if selectedKind != 'preprocessor' && selectedKind != 'flow'}
+		{#if showHub && selectedKind != 'preprocessor' && selectedKind != 'flow'}
 			{#if (!selected || selected?.kind === 'integrations') && (preFilter === 'hub' || preFilter === 'all')}
 				{#if !selected && preFilter !== 'hub'}
 					<div class=" pb-0 text-2xs font-normal text-secondary ml-2">Hub</div>

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

@@ -39,7 +39,7 @@
 	import type { ButtonProp } from '$lib/components/diffEditorTypes'
 	import { loadSchemaFromModule } from '../flowInfers'
 	import { type Job } from '$lib/gen'
-	import { workspaceStore } from '$lib/stores'
+	import { operatorBuilderFlows, workspaceStore } from '$lib/stores'
 	import { checkIfParentLoop } from '../utils.svelte'
 	import { useWorkspaceScriptSettings } from '../useWorkspaceScriptSettings.svelte'
 	import ScriptSettingsBadges from '$lib/components/ScriptSettingsBadges.svelte'
@@ -230,6 +230,7 @@
 			!flowModule.value.path?.startsWith('hub/') &&
 			flowModule.value.hash == undefined &&
 			customUi?.scriptEdit != false &&
+			!$operatorBuilderFlows &&
 			$workspaceScriptSettingsDrawer != undefined
 	)
 	let workspaceScriptNoEditReason = $derived(

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

@@ -12,7 +12,7 @@
 	import RefreshButton from '$lib/components/common/button/RefreshButton.svelte'
 	import Button from '$lib/components/common/button/Button.svelte'
 	import { ResourceService } from '$lib/gen'
-	import { workspaceStore } from '$lib/stores'
+	import { operatorBuilderFlows, workspaceStore } from '$lib/stores'
 	import type { FlowEditorContext } from '../types'
 	import { logReusableAgentUsage } from '../agentTelemetry'
 	import { BotIcon, Loader2, Plus } from 'lucide-svelte'
@@ -50,7 +50,11 @@
 		| 'failure'
 		| 'aisandbox'
 		| 'aiagent' = $state(untrack(() => kind))
-	let preFilter: 'all' | 'workspace' | 'hub' = $state('all')
+	// Builders compose what the workspace already deployed: hub scripts bring in code nobody here
+	// reviewed, and the backend refuses them, so never open on the hub for them.
+	let preFilter: 'all' | 'workspace' | 'hub' = $state(
+		untrack(() => $operatorBuilderFlows) ? 'workspace' : 'all'
+	)
 	let loading = $state(false)
 	let small = $derived(smallProp ?? (kind === 'preprocessor' || kind === 'failure'))
 
@@ -183,13 +187,13 @@
 		<StepGenQuick
 			bind:this={stepGen}
 			on:escape={() => dispatch('close')}
-			{disableAi}
+			disableAi={disableAi || $operatorBuilderFlows}
 			on:insert
 			bind:funcDesc
 			{preFilter}
 			{loading}
 		/>
-		{#if selectedKind != 'preprocessor' && selectedKind != 'flow'}
+		{#if selectedKind != 'preprocessor' && selectedKind != 'flow' && !$operatorBuilderFlows}
 			<ToggleHubWorkspaceQuick bind:selected={preFilter} />
 		{/if}
 		<RefreshButton
@@ -314,7 +318,7 @@
 							}}
 						/>
 					{/if}
-					{#if customUi?.aiSandbox != false}
+					{#if customUi?.aiSandbox != false && !$operatorBuilderFlows}
 						<TopLevelNode
 							label="AI Sandbox"
 							selected={selectedKind === 'aisandbox'}

frontend/src/lib/components/home/CreateActionsMenu.svelte

@@ -25,6 +25,7 @@
 	import { importScriptStore } from '$lib/components/scripts/scriptStore.svelte'
 	import { importStore } from '$lib/components/apps/store'
 	import { conditionalMelt, getLocalSetting, storeLocalSetting } from '$lib/utils'
+	import { operatorBuilderFlows } from '$lib/stores'
 	import { createDropdownMenu, melt } from '@melt-ui/svelte'
 	import YAML from 'yaml'
 	import type { Snippet } from 'svelte'
@@ -247,6 +248,14 @@
 		}
 	}
 
+	// A builder composes runnables that already exist, so only flows are offered: everything else
+	// here writes code, which the backend refuses from an operator.
+	// Derived, not computed once: switching workspace only sets `workspaceStore`, it does not
+	// remount this component, so a snapshot would keep the previous workspace's kinds.
+	const options: Option[] = $derived(
+		$operatorBuilderFlows ? allOptions.filter((o) => o.key === 'flow') : allOptions
+	)
+
 	// the doc panel only shows while an option is hovered or focused, so the menu opens compact
 	let activeKey: string | undefined = $state(undefined)
 	// every option's import action, surfaced together under the bottom "Import" submenu.
@@ -256,7 +265,7 @@
 		...(onImportHubProject
 			? [{ label: 'Import a hub project', onSelect: onImportHubProject }]
 			: []),
-		...allOptions.flatMap((o) => o.extras ?? [])
+		...options.flatMap((o) => o.extras ?? [])
 	])
 
 	// melt dropdown menu: arrow-key nav, typeahead, focus management and outside/escape
@@ -372,7 +381,7 @@
 			activeKey = undefined
 		}
 	})
-	let active = $derived(allOptions.find((o) => o.key === activeKey))
+	let active = $derived(options.find((o) => o.key === activeKey))
 	let activeAc = $derived(active ? accentClasses[active.accent] : undefined)
 
 	// shared YAML/JSON import drawer, reused by every "Import …" extra
@@ -525,7 +534,7 @@
 						</span>
 					{/if}
 				{/snippet}
-				{#each allOptions as option (option.key)}
+				{#each options as option (option.key)}
 					{@const ac = accentClasses[option.accent]}
 					{@const rowClass =
 						'w-full flex flex-row items-center gap-2.5 rounded-md px-2 py-1.5 text-left cursor-pointer transition-colors focus:outline-none data-[highlighted]:bg-surface-hover hover:bg-surface-hover'}

frontend/src/lib/components/home/ItemsList.svelte

@@ -16,7 +16,12 @@
 	} from '$lib/gen'
 	import { resource } from 'runed'
 	import { getDraftItems } from '$lib/workspaceDrafts.svelte'
-	import { disableHubStore, userStore, workspaceStore } from '$lib/stores'
+	import {
+		disableHubStore,
+		operatorBuilderFlows,
+		userStore,
+		workspaceStore
+	} from '$lib/stores'
 	import type uFuzzy from '@leeoniya/ufuzzy'
 	import {
 		ArrowDownUp,
@@ -340,7 +345,9 @@
 			canWrite:
 				canWrite(it.path, (it.extra_perms ?? {}) as any, $userStore) &&
 				(it.type === 'script' || it.workspace_id == $workspaceStore) &&
-				!$userStore?.operator
+				// The builder right covers flows only; a script or an app is still off limits, so
+				// the row must not offer edit or delete for those.
+				(!$userStore?.operator || (it.type === 'flow' && $operatorBuilderFlows))
 		}
 		// combinedItems reads a script's time from `created_at`; the endpoint's
 		// unified `edited_at` holds exactly that for scripts.
@@ -1061,7 +1068,7 @@
 	 * whose direct-deploy protection cleared `showEditButtons` — must not be shown them.
 	 * Reading archived items is not a write, so it is not gated on this.
 	 */
-	let canCreateHere = $derived(!$userStore?.operator && showEditButtons)
+	let canCreateHere = $derived((!$userStore?.operator || $operatorBuilderFlows) && showEditButtons)
 
 	// The workspace itself holds nothing — no filter is narrowing the list away. It stays
 	// false until the first load resolves: a skeleton already means "loading", and the

frontend/src/lib/components/settings/WorkspaceOperatorSettings.svelte

@@ -6,6 +6,7 @@
 	import Section from '$lib/components/Section.svelte'
 	import Head from '$lib/components/table/Head.svelte'
 	import Cell from '$lib/components/table/Cell.svelte'
+	import ConfirmationModal from '$lib/components/common/confirmationModal/ConfirmationModal.svelte'
 	import Toggle from '$lib/components/Toggle.svelte'
 	import { WorkspaceService } from '$lib/gen'
 	import { workspaceStore } from '$lib/stores'
@@ -27,25 +28,40 @@
 	})
 
 	// Kept out of `operatorWorkspaceSettings` so the visibility table's "Enable all" never flips a
-	// write right, and so these rows stay out of that table. Withdrawable rather than granted:
-	// operators hold them until an admin turns them off.
+	// write right, and so these rows stay out of that table.
+	let builderFlows = $state(false)
+	// Withdrawable rather than granted: operators hold these until an admin turns them off.
 	let manageSchedules = $state(true)
 	let manageTriggers = $state(true)
 
 	let originalSettings = $state({
 		...untrack(() => operatorWorkspaceSettings),
+		builder_flows: false,
 		manage_schedules: true,
 		manage_triggers: true
 	})
 	let isChanged = $state(false)
 	let currentWorkspace: string | null = $state(null)
+	let confirmBuilderOpen = $state(false)
 
 	const settingsPayload = $derived({
 		...operatorWorkspaceSettings,
+		builder_flows: builderFlows,
 		manage_schedules: manageSchedules,
 		manage_triggers: manageTriggers
 	})
 
+	// The seat cost lands when the right is first granted, so confirm only on that transition.
+	const grantsBuilderRight = $derived(builderFlows && !originalSettings.builder_flows)
+
+	function onSaveClicked() {
+		if (grantsBuilderRight) {
+			confirmBuilderOpen = true
+		} else {
+			saveSettings()
+		}
+	}
+
 	async function saveSettings() {
 		try {
 			await WorkspaceService.updateOperatorSettings({
@@ -83,15 +99,18 @@
 				})
 				if (settings.operator_settings !== null) {
 					const {
+						builder_flows: remoteFlows,
 						manage_schedules: remoteSchedules,
 						manage_triggers: remoteTriggers,
 						...remoteVisibility
 					} = settings.operator_settings ?? {}
 					operatorWorkspaceSettings = { ...operatorWorkspaceSettings, ...remoteVisibility }
+					builderFlows = remoteFlows ?? false
 					manageSchedules = remoteSchedules ?? true
 					manageTriggers = remoteTriggers ?? true
 					originalSettings = {
 						...operatorWorkspaceSettings,
+						builder_flows: builderFlows,
 						manage_schedules: manageSchedules,
 						manage_triggers: manageTriggers
 					}
@@ -120,7 +139,7 @@
 >
 	{#snippet action()}
 		<Button
-			on:click={saveSettings}
+			on:click={onSaveClicked}
 			startIcon={{ icon: SaveIcon }}
 			disabled={!isChanged}
 			variant="accent"
@@ -129,6 +148,20 @@
 		</Button>
 	{/snippet}
 
+	<div class="flex flex-col gap-y-1 mb-4">
+		<span class="text-xs font-semibold text-emphasis">Builder rights</span>
+		<span class="text-xs font-normal text-secondary">
+			Let operators compose flows out of scripts and flows that are already deployed. They still
+			cannot write code. Granting this makes each operator consume a full seat instead of half a
+			seat.
+		</span>
+		<Toggle
+			bind:checked={builderFlows}
+			options={{ right: 'Operators can build flows' }}
+			size="xs"
+		/>
+	</div>
+
 	<div class="flex flex-col gap-y-1 mb-4">
 		<span class="text-xs font-semibold text-emphasis">Schedules and triggers</span>
 		<span class="text-xs font-normal text-secondary">
@@ -199,3 +232,26 @@
 		</tbody>
 	</DataTable>
 </Section>
+
+<ConfirmationModal
+	open={confirmBuilderOpen}
+	title="Give operators builder rights"
+	confirmationText="Enable builder rights"
+	onCanceled={() => (confirmBuilderOpen = false)}
+	onConfirmed={async () => {
+		confirmBuilderOpen = false
+		await saveSettings()
+	}}
+>
+	<div class="flex flex-col gap-2 text-sm">
+		<span>This applies to every operator of this workspace, not to a chosen few.</span>
+		<span>
+			Each of them then consumes a full seat instead of half a seat, which changes what this
+			instance is billed.
+		</span>
+		<span>
+			They can create, edit and delete flows wherever their folder permissions already let them
+			write. Review those permissions before enabling.
+		</span>
+	</div>
+</ConfirmationModal>

frontend/src/lib/stores.ts

@@ -191,6 +191,19 @@ export const userWorkspaces: Readable<Array<UserWorkspace>> = derived(
 	}
 )
 
+/**
+ * True when the current user is an operator of a workspace that granted operators the right to
+ * compose flows out of runnables that are already deployed. They still author no code, and
+ * everywhere else `operator` keeps meaning read-only, so a gate on the operator role has to
+ * consult this before refusing.
+ */
+export const operatorBuilderFlows: Readable<boolean> = derived(
+	[userStore, userWorkspaces, workspaceStore],
+	([user, workspaces, workspace]) =>
+		(user?.operator ?? false) &&
+		workspaces.find((w) => w.id === workspace)?.operator_settings?.builder_flows === true
+)
+
 export const codeCompletionLoading = writable<boolean>(false)
 export const metadataCompletionEnabled = writable<boolean>(true)
 export const stepInputCompletionEnabled = writable<boolean>(true)

frontend/src/routes/(root)/(logged)/flows/get/[...path]/+page.svelte

@@ -28,7 +28,13 @@
 	import MoveDrawer from '$lib/components/MoveDrawer.svelte'
 	import RunForm from '$lib/components/RunForm.svelte'
 	import ShareModal from '$lib/components/ShareModal.svelte'
-	import { enterpriseLicense, userStore, userWorkspaces, workspaceStore } from '$lib/stores'
+	import {
+		enterpriseLicense,
+		operatorBuilderFlows,
+		userStore,
+		userWorkspaces,
+		workspaceStore
+	} from '$lib/stores'
 	import { sendUserToast } from '$lib/toast'
 	import DeployWorkspaceDrawer from '$lib/components/DeployWorkspaceDrawer.svelte'
 	import SavedInputsV2 from '$lib/components/SavedInputsV2.svelte'
@@ -292,6 +298,10 @@
 		}
 	}
 
+	// Operators with the builder right author flows out of deployed runnables; every other
+	// operator is read-only here.
+	let canAuthorFlow = $derived(!$userStore?.operator || $operatorBuilderFlows)
+
 	let moveDrawer: MoveDrawer | undefined = $state()
 	let deploymentDrawer: DeployWorkspaceDrawer | undefined = $state()
 	let runForm: RunForm | undefined = $state()
@@ -299,7 +309,7 @@
 	function getMainButtons(flow: Flow | undefined, args: object | undefined) {
 		const buttons: any = []
 
-		if (flow && !$userStore?.operator) {
+		if (flow && canAuthorFlow) {
 			buttons.push({
 				label: 'Fork',
 				description: `Start a new flow from a copy of this one`,
@@ -360,11 +370,11 @@
 			}
 		})
 
-		if (!flow || $userStore?.operator || !can_write) {
+		if (!flow || !canAuthorFlow || !can_write) {
 			return buttons
 		}
 
-		if (!$userStore?.operator) {
+		if (canAuthorFlow) {
 			buttons.push({
 				label: 'Build app',
 				narrow: 'menu',
@@ -413,7 +423,7 @@
 		flow: Flow | undefined,
 		deployUiSettings: WorkspaceDeployUISettings | undefined
 	) {
-		if (!flow || $userStore?.operator) return []
+		if (!flow || !canAuthorFlow) return []
 
 		const menuItems: any = []
 
@@ -440,7 +450,7 @@
 			}
 		})
 
-		if (isDeployable('flow', flow?.path ?? '', deployUiSettings)) {
+		if (isDeployable('flow', flow?.path ?? '', deployUiSettings) && !$userStore?.operator) {
 			menuItems.push({
 				label: 'Deploy to staging/prod',
 				onclick: () => deploymentDrawer?.openDrawer(flow?.path ?? '', 'flow'),