fix(datatables): give the role catalog its own row, out of reach of the config machinery

Putting it inside `custom_instance_pg_databases` was the wrong call, and it cost two ways.
The catalog serializes a generated Postgres password per role, and that row is the
operator-facing instance config, so the passwords reached `get_instance_config` and its YAML
editor — a live cluster credential in a response body, a UI field and any log of either.
Worse in the other direction: `to_settings_map` strips the catalog, so a full-row upsert of
that key writes the row back without it and the catalog is gone, while the cluster keeps every
login it described.

`custom_instance_replication_pwd` is the precedent and says exactly why — a generated secret,
written only by the server, never operator-authored, hidden so the config machinery cannot
read, rewrite or drop it. The catalog is the same thing, so it now has the same shape:
`datatable_roles`, in `HIDDEN_SETTINGS`, `PROTECTED_SETTINGS` and the agent-worker denylist.
No redaction to keep in step with three code paths, and no way for a neighbouring write to
take it out.

Two races on the same shared documents. `edit_datatable_config` read the stored data tables
outside its transaction and then wrote the whole `datatable` document, so a permissions save
committing in between was silently rolled back; it now reads under `FOR UPDATE`. And
`set_datatable_permissions` validated role ids against the catalog before opening its
transaction, so a deletion in between let it write a deleted role back — including as the
default, which every later job then fails on; it now holds the catalog lock and the settings
row across validation and write.

Completes the authorization contracts the previous commit claimed but did not finish:
`read_datatable_entry` (which it named and missed), `resolve_governing_datatable`, whose whole
job is to answer for a workspace the caller may not belong to, and
`converge_connect_grants_with`, which had not inherited its wrapper's.

Also the generic Python SDK reference: `_format_py_params` learned the bare `*` last time, but
`extract_py_functions` is a second formatter and still rendered `datatable(name, role)`, so
code written from that page passed a keyword-only argument positionally.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012ti5HyeTikPMYyW8YSdiHR

backend/.sqlx/query-06abbf945bee93349ff88f64906b96ea1e853ef202510281427cfa9beeff81b3.json

@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT 1 AS one FROM workspace_settings WHERE workspace_id = $1 FOR UPDATE",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "one",
+        "type_info": "Int4"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "06abbf945bee93349ff88f64906b96ea1e853ef202510281427cfa9beeff81b3"
+}

backend/.sqlx/query-6f9fb5d72f486358fa25d6887bd69b93910e028f140c07048f2c1c8d63ee6909.json

@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO global_settings (name, value) VALUES ($1, $2)\n         ON CONFLICT (name) DO UPDATE SET value = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Varchar",
+        "Jsonb"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "6f9fb5d72f486358fa25d6887bd69b93910e028f140c07048f2c1c8d63ee6909"
+}

backend/.sqlx/query-b42af37fb474bea4c5419b0a46d9eadfe384013ab970ccf9c5effd1c78321b7c.json

@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT ws.datatable->'datatables' FROM workspace_settings ws\n             WHERE ws.workspace_id = $1 FOR UPDATE",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "?column?",
+        "type_info": "Jsonb"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "b42af37fb474bea4c5419b0a46d9eadfe384013ab970ccf9c5effd1c78321b7c"
+}

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

@@ -354,15 +354,7 @@ async fn concurrent_role_catalog_writes_do_not_lose_an_entry(db: Pool<Postgres>)
                 id.to_string(),
                 InstanceDatatableRole { name: id.to_string(), enabled: true, pwd: None },
             );
-            let value = serde_json::to_value(&catalog)?;
-            sqlx::query(
-                "UPDATE global_settings
-                 SET value = jsonb_set(COALESCE(value, '{}'::jsonb), '{roles}', $1)
-                 WHERE name = 'custom_instance_pg_databases'",
-            )
-            .bind(value)
-            .execute(&mut *tx)
-            .await?;
+            windmill_common::datatable_roles::write_role_catalog(&mut tx, &catalog).await?;
             tx.commit().await?;
             Ok::<_, anyhow::Error>(())
         }

backend/windmill-api-integration-tests/tests/fixtures/datatable_roles.sql

@@ -2,14 +2,12 @@
 -- carrying a copy. `test-user-2` is a non-admin of the parent and an admin of the fork: the shape
 -- the pointer exists for.
 
-UPDATE global_settings SET value = jsonb_set(value, '{roles}',
-    '{"role1": {"name": "analytics", "enabled": true, "pwd": "pw"}}'::jsonb)
-    WHERE name = 'custom_instance_pg_databases';
-INSERT INTO global_settings (name, value)
-    SELECT 'custom_instance_pg_databases',
-           '{"user_pwd": "pw", "databases": {"dt_main": {}},
-             "roles": {"role1": {"name": "analytics", "enabled": true, "pwd": "pw"}}}'::jsonb
-    WHERE NOT EXISTS (SELECT 1 FROM global_settings WHERE name = 'custom_instance_pg_databases');
+INSERT INTO global_settings (name, value) VALUES
+    ('custom_instance_pg_databases', '{"user_pwd": "pw", "databases": {"dt_main": {}}}'::jsonb),
+    -- The role catalog has its own row: it holds generated credentials and must stay out of the
+    -- operator-facing config the neighbouring row belongs to.
+    ('datatable_roles', '{"role1": {"name": "analytics", "enabled": true, "pwd": "pw"}}'::jsonb)
+    ON CONFLICT (name) DO UPDATE SET value = EXCLUDED.value;
 
 UPDATE workspace_settings SET datatable = '{
     "datatables": {

backend/windmill-api-settings/src/lib.rs

@@ -2578,40 +2578,6 @@ fn datatable_role_infos(
         .collect()
 }
 
-/// Persist the catalog next to the instance Postgres password, in the same `global_settings` row
-/// the instance database registry lives in.
-///
-/// Errors when it matches nothing. The cluster is written first, so a silent no-op here would
-/// leave a live Postgres login with a password nobody recorded: invisible to the catalog,
-/// un-recreatable (the name is taken) and un-deletable (there is no entry to delete). The row is
-/// normally planted by the boot converge, but that swallows its own failures, so this is a check
-/// rather than an assumption.
-async fn write_role_catalog(
-    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
-    catalog: &windmill_common::datatable_roles::DatatableRoleCatalog,
-) -> error::Result<()> {
-    let value = serde_json::to_value(catalog).map_err(to_anyhow)?;
-    let written = sqlx::query!(
-        "UPDATE global_settings SET value = jsonb_set(COALESCE(value, '{}'::jsonb), '{roles}', $1)
-         WHERE name = 'custom_instance_pg_databases'",
-        value
-    )
-    .execute(&mut **tx)
-    .await?
-    .rows_affected();
-    if written == 0 {
-        return Err(error::Error::internal_err(
-            concat!(
-                "The instance Postgres settings row is missing, so the data table role catalog ",
-                "could not be recorded. Refresh the custom instance user password in instance ",
-                "settings to recreate it, then try again."
-            )
-            .to_string(),
-        ));
-    }
-    Ok(())
-}
-
 async fn list_datatable_roles(
     authed: ApiAuthed,
     Extension(db): Extension<DB>,
@@ -2656,7 +2622,7 @@ async fn create_datatable_role(
             pwd: Some(pwd),
         },
     );
-    write_role_catalog(&mut tx, &catalog).await?;
+    windmill_common::datatable_roles::write_role_catalog(&mut tx, &catalog).await?;
     tx.commit().await?;
     converge_connect_grants_everywhere(&db, &catalog).await;
     windmill_common::feature_usage::log_feature_usage("datatable", "role_created", "");
@@ -2713,7 +2679,7 @@ async fn update_datatable_role(
     }
 
     catalog.insert(id.clone(), updated.clone());
-    write_role_catalog(&mut tx, &catalog).await?;
+    windmill_common::datatable_roles::write_role_catalog(&mut tx, &catalog).await?;
     tx.commit().await?;
     converge_connect_grants_everywhere(&db, &catalog).await;
 
@@ -2756,7 +2722,7 @@ async fn delete_datatable_role(
 
     windmill_common::datatable_roles::drop_instance_role(&db, &mut tx, &role.name).await?;
     catalog.remove(&id);
-    write_role_catalog(&mut tx, &catalog).await?;
+    windmill_common::datatable_roles::write_role_catalog(&mut tx, &catalog).await?;
     tx.commit().await?;
     // After the drop commits: a tenant naming a role that still exists is harmless, one naming a
     // role that is gone is not, so this only ever runs once the cluster agrees it is gone.

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

@@ -274,8 +274,21 @@ async fn set_datatable_permissions(
         )));
     }
 
+    // One transaction for the whole save, holding both locks the decision depends on: the role
+    // catalog, so a role cannot be deleted between validating an id and writing it back, and the
+    // workspace settings row, so a concurrent settings save cannot carry a stale copy of this
+    // block forward over what is written here.
+    let mut tx = db.begin().await?;
+    windmill_common::datatable_roles::lock_role_catalog(&mut tx).await?;
+    sqlx::query!(
+        "SELECT 1 AS one FROM workspace_settings WHERE workspace_id = $1 FOR UPDATE",
+        &governing.workspace_id
+    )
+    .fetch_optional(&mut *tx)
+    .await?;
+
     let permissions = if req.permissioned {
-        let catalog = read_role_catalog(&db).await?;
+        let catalog = windmill_common::datatable_roles::read_role_catalog_tx(&mut tx).await?;
         let mut roles: BTreeMap<String, DataTableRoleTenants> = BTreeMap::new();
         for role in req.roles {
             if role.id != ADMIN_DATATABLE_ROLE && !catalog.contains_key(&role.id) {
@@ -306,30 +319,6 @@ async fn set_datatable_permissions(
         None
     };
 
-    // An instance database provisioned before data table roles existed has neither the grant
-    // options the admin connection needs to delegate privileges, nor a CONNECT grant for any role
-    // — so a role would be refused at login however its tenants read. Repair it here, at the one
-    // moment someone is deciding this data table's roles. Best-effort: neither is worth failing a
-    // tenant edit over, and both converge again on the next save.
-    if permissions.is_some() {
-        if let Some(database) = governing.datatable.database.as_ref() {
-            if database.resource_type == DataTableCatalogResourceType::Instance {
-                let dbname = &database.resource_path;
-                if let Err(e) =
-                    windmill_common::ensure_instance_db_grant_options_unchecked(&db, dbname).await
-                {
-                    tracing::warn!("Could not refresh grant options on '{dbname}': {e}");
-                }
-                if let Err(e) =
-                    windmill_common::datatable_roles::converge_connect_grants(&db, dbname).await
-                {
-                    tracing::warn!("Could not refresh CONNECT grants on '{dbname}': {e}");
-                }
-            }
-        }
-    }
-
-    let mut tx = db.begin().await?;
     let value = match &permissions {
         Some(p) => serde_json::to_value(p).map_err(|e| Error::internal_err(e.to_string()))?,
         None => serde_json::Value::Null,
@@ -362,6 +351,32 @@ async fn set_datatable_permissions(
     .await?;
     tx.commit().await?;
 
+    // An instance database provisioned before data table roles existed has neither the grant
+    // options the admin connection needs to delegate privileges, nor a CONNECT grant for any role
+    // — so a role would be refused at login however its tenants read. Repair it here, at the one
+    // moment someone is deciding this data table's roles. Best-effort: neither is worth failing a
+    // tenant edit over, and both converge again on the next save.
+    //
+    // Runs after the commit: it opens its own connections to other databases, which has no place
+    // inside a transaction holding two locks.
+    if permissions.is_some() {
+        if let Some(database) = governing.datatable.database.as_ref() {
+            if database.resource_type == DataTableCatalogResourceType::Instance {
+                let dbname = &database.resource_path;
+                if let Err(e) =
+                    windmill_common::ensure_instance_db_grant_options_unchecked(&db, dbname).await
+                {
+                    tracing::warn!("Could not refresh grant options on '{dbname}': {e}");
+                }
+                if let Err(e) =
+                    windmill_common::datatable_roles::converge_connect_grants(&db, dbname).await
+                {
+                    tracing::warn!("Could not refresh CONNECT grants on '{dbname}': {e}");
+                }
+            }
+        }
+    }
+
     // A live replication stream holds a connection it opened under the old decision. Bouncing the
     // rows makes every listener reconnect and re-authorize.
     restart_streams_reaching(&db, &governing).await?;

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

@@ -3544,12 +3544,17 @@ async fn edit_datatable_config(
 
     let mut tx = db.begin().await?;
 
+    // Read under the row lock this transaction will write with. `permissions`, `reference` and
+    // `forked_from` are carried across from what this read returns, so a permissions save
+    // committing between the read and the whole-document write below would be silently rolled back
+    // by it.
     let old_datatables: HashMap<String, DataTable> = serde_json::from_value(
         sqlx::query_scalar!(
-            "SELECT ws.datatable->'datatables' FROM workspace_settings ws WHERE ws.workspace_id = $1",
+            "SELECT ws.datatable->'datatables' FROM workspace_settings ws
+             WHERE ws.workspace_id = $1 FOR UPDATE",
             &w_id
         )
-        .fetch_one(&db)
+        .fetch_one(&mut *tx)
         .await?
         .unwrap_or(serde_json::Value::Null),
     )

backend/windmill-common/src/datatable_roles.rs

@@ -21,6 +21,7 @@ use serde::{Deserialize, Serialize};
 
 use crate::{
     error::{Error, Result},
+    global_settings::DATATABLE_ROLES_SETTING,
     DB,
 };
 
@@ -33,9 +34,8 @@ pub const ADMIN_DATATABLE_ROLE: &str = "admin";
 /// membership is what later lets it `ALTER ... OWNER TO` a role and drop it.
 pub const CUSTOM_INSTANCE_USER: &str = "custom_instance_user";
 
-/// One catalog entry. The password is per role and instance-wide; it lives here rather than in any
-/// workspace's settings, next to the `custom_instance_user` password in the same
-/// `custom_instance_pg_databases` row.
+/// One catalog entry. The password is per role and instance-wide, and lives in the instance's own
+/// [`DATATABLE_ROLES_SETTING`] row rather than in any workspace's settings.
 #[derive(Deserialize, Serialize, Clone)]
 #[cfg_attr(feature = "instance_config_schema", derive(schemars::JsonSchema))]
 pub struct InstanceDatatableRole {
@@ -134,11 +134,11 @@ pub async fn lock_role_catalog(tx: &mut sqlx::Transaction<'_, sqlx::Postgres>) -
 /// response, a log line or an audit record.
 pub async fn read_role_catalog(db: &DB) -> Result<DatatableRoleCatalog> {
     let value = sqlx::query_scalar!(
-        "SELECT value->'roles' FROM global_settings WHERE name = 'custom_instance_pg_databases'"
+        "SELECT value FROM global_settings WHERE name = $1",
+        DATATABLE_ROLES_SETTING
     )
     .fetch_optional(db)
-    .await?
-    .flatten();
+    .await?;
     Ok(parse_role_catalog(value))
 }
 
@@ -148,14 +148,36 @@ pub async fn read_role_catalog_tx(
     tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
 ) -> Result<DatatableRoleCatalog> {
     let value = sqlx::query_scalar!(
-        "SELECT value->'roles' FROM global_settings WHERE name = 'custom_instance_pg_databases'"
+        "SELECT value FROM global_settings WHERE name = $1",
+        DATATABLE_ROLES_SETTING
     )
     .fetch_optional(&mut **tx)
-    .await?
-    .flatten();
+    .await?;
     Ok(parse_role_catalog(value))
 }
 
+/// Persist the catalog, in the caller's transaction so it commits with the cluster DDL it
+/// describes. Upserts: the row does not exist until the first role is created.
+///
+/// Authorization: writes generated Postgres credentials. Callers MUST restrict this to superadmin
+/// paths and MUST hold [`lock_role_catalog`] on `tx`.
+pub async fn write_role_catalog(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    catalog: &DatatableRoleCatalog,
+) -> Result<()> {
+    let value = serde_json::to_value(catalog)
+        .map_err(|e| Error::internal_err(format!("serializing the role catalog: {e}")))?;
+    sqlx::query!(
+        "INSERT INTO global_settings (name, value) VALUES ($1, $2)
+         ON CONFLICT (name) DO UPDATE SET value = $2",
+        DATATABLE_ROLES_SETTING,
+        value
+    )
+    .execute(&mut **tx)
+    .await?;
+    Ok(())
+}
+
 /// A catalog that will not deserialize is an empty one, which fails closed: tenants are keyed by
 /// id independently of it, so every role then resolves to "no longer exists on this instance"
 /// rather than to admin.
@@ -212,6 +234,8 @@ pub async fn converge_connect_grants(db: &DB, dbname: &str) -> Result<()> {
     converge_connect_grants_with(db, dbname, &catalog).await
 }
 
+/// As [`converge_connect_grants`], with a catalog the caller already read. Same authorization
+/// contract: it rewrites a database's ACL with the server's own credentials and checks nothing.
 pub async fn converge_connect_grants_with(
     db: &DB,
     dbname: &str,

backend/windmill-common/src/global_settings.rs

@@ -243,8 +243,18 @@ pub const AGENT_WORKER_BLOCKED_SETTINGS: &[&str] = &[
     // resolve datatable connections through the dedicated datatable endpoints, never these.
     "custom_instance_pg_databases",
     "custom_instance_replication_pwd",
+    // The data table role catalog: one generated Postgres password per role.
+    DATATABLE_ROLES_SETTING,
 ];
 
+/// The instance's data table role catalog, `{ "<id>": { name, enabled, pwd } }`.
+///
+/// Its own row rather than a field of `custom_instance_pg_databases`, for the same reason
+/// `custom_instance_replication_pwd` is: it holds generated credentials and is written only by the
+/// server, so the config machinery must not be able to read it into an export, rewrite it, or drop
+/// it on a full-row upsert of a neighbour.
+pub const DATATABLE_ROLES_SETTING: &str = "datatable_roles";
+
 /// Whether an agent worker may read the given global setting over HTTP.
 /// Deny-by-exception: everything is readable except [`AGENT_WORKER_BLOCKED_SETTINGS`].
 pub fn is_setting_readable_by_agent_worker(name: &str) -> bool {
@@ -677,6 +687,7 @@ mod tests {
             OTEL_TRACING_PROXY_SETTING,
             "custom_instance_pg_databases",
             "custom_instance_replication_pwd",
+            DATATABLE_ROLES_SETTING,
         ] {
             assert!(
                 !is_setting_readable_by_agent_worker(key),

backend/windmill-common/src/instance_config.rs

@@ -438,13 +438,11 @@ impl GlobalSettings {
             serde_json::Value::Object(map) => map.into_iter().collect(),
             _ => unreachable!(),
         };
-        // Strip the runtime-only sub-fields of custom_instance_pg_databases: `databases` is setup
-        // status/logs managed by the setup endpoint, and `roles` is the data table role catalog,
-        // which carries one Postgres password per role. Neither is configuration.
+        // Strip runtime-only `databases` sub-field from custom_instance_pg_databases.
+        // It contains setup status/logs managed by the setup endpoint, not configuration.
         if let Some(pg) = map.get_mut("custom_instance_pg_databases") {
             if let Some(obj) = pg.as_object_mut() {
                 obj.remove("databases");
-                obj.remove("roles");
             }
         }
         map
@@ -795,10 +793,6 @@ pub struct CustomInstancePgDatabases {
     pub user_pwd: Option<StringOrSecretRef>,
     #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
     pub databases: BTreeMap<String, CustomInstanceDb>,
-    /// The instance's data table role catalog, keyed by generated id. Runtime state carrying one
-    /// password per role, so it is stripped from config sync exactly like `databases`.
-    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
-    pub roles: BTreeMap<String, crate::datatable_roles::InstanceDatatableRole>,
 }
 
 /// Status of a single custom instance database.
@@ -965,6 +959,7 @@ pub const PROTECTED_SETTINGS: &[&str] = &[
     "ducklake_settings",
     "custom_instance_pg_databases",
     "custom_instance_replication_pwd",
+    crate::global_settings::DATATABLE_ROLES_SETTING,
     "uid",
     "rsa_keys",
     "jwt_secret",
@@ -990,6 +985,9 @@ pub const HIDDEN_SETTINGS: &[&str] = &[
     // Server-only (written by setup/refresh via direct SQL), never operator-authored —
     // hidden so the config machinery can't read, rewrite, or drop it.
     "custom_instance_replication_pwd",
+    // The data table role catalog, one generated Postgres password per role. Same reasoning as
+    // the line above: server-written, never operator-authored, and it must not reach an export.
+    crate::global_settings::DATATABLE_ROLES_SETTING,
 ];
 
 /// Top-level settings whose entire value is sensitive and must be fully redacted in logs.
@@ -1201,17 +1199,14 @@ pub fn diff_global_settings(
         } else {
             desired_value.clone()
         };
-        // Preserve the runtime-only sub-fields inside `custom_instance_pg_databases` so that
-        // config sync never wipes setup status/logs managed by the setup endpoint, nor the data
-        // table role catalog — the latter mirrors real Postgres roles, so losing it would leave
-        // the cluster holding logins Windmill can no longer name.
+        // Preserve the runtime-only `databases` sub-field inside
+        // `custom_instance_pg_databases` so that config sync never wipes
+        // setup status/logs that are managed by the setup endpoint.
         if key == "custom_instance_pg_databases" {
             if let Some(existing) = current.get(key) {
-                for runtime_field in ["databases", "roles"] {
-                    if let Some(kept) = existing.get(runtime_field) {
-                        if let Some(obj) = value.as_object_mut() {
-                            obj.entry(runtime_field).or_insert_with(|| kept.clone());
-                        }
+                if let Some(databases) = existing.get("databases") {
+                    if let Some(obj) = value.as_object_mut() {
+                        obj.entry("databases").or_insert_with(|| databases.clone());
                     }
                 }
             }

backend/windmill-common/src/workspaces.rs

@@ -1381,6 +1381,11 @@ fn datatable_not_found_error(name: &str, datatables: Option<&serde_json::Value>)
 }
 
 /// Read one workspace's data table entry, without following a pointer.
+///
+/// Authorization: reads a workspace's stored configuration by id and checks nothing — not that the
+/// caller belongs to that workspace, nor that they may see the data table. Callers MUST have
+/// authorized access to `w_id` already, and MUST NOT return the entry to a caller from another
+/// workspace: it names the database and, on a governing entry, who may reach it as what.
 pub async fn read_datatable_entry(db: &DB, w_id: &str, name: &str) -> Result<DataTable> {
     let datatables = sqlx::query_scalar!(
         r#"
@@ -1408,6 +1413,13 @@ pub async fn read_datatable_entry(db: &DB, w_id: &str, name: &str) -> Result<Dat
 /// Every decision downstream — which database to connect to, whose `permissions` apply, whose
 /// members tenants are evaluated against, who may administer it — is taken on this, never on the
 /// entry the caller named.
+///
+/// Authorization: resolving deliberately crosses into the governing workspace, so it answers for a
+/// workspace the caller may not belong to and checks nothing itself. It is the input to the
+/// checks, not one of them: callers MUST pass what it returns to
+/// [`can_use_datatable_role_in_governing_workspace`] or [`ensure_datatable_admin_access`] before
+/// acting on it, and MUST NOT return its `permissions` or `workspace_id` to a caller from
+/// elsewhere without gating on the answer.
 pub struct GoverningDatatable {
     pub workspace_id: String,
     pub name: String,

cli/src/guidance/skills.gen.ts

@@ -4414,7 +4414,7 @@ def send_teams_message(conversation_id: str, text: str, success: bool = True, ca
 # 
 # Returns:
 #     DataTableClient instance
-def datatable(name: str = 'main', role: Optional[str] = None)
+def datatable(name: str = 'main', *, role: Optional[str] = None)
 
 # Get a DuckLake client for DuckDB queries.
 # 
@@ -4624,7 +4624,7 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]]
 # 
 #     @task(retry={"attempts": 3, "delay": 30, "multiplier": 2})
 #     async def call_api(payload: dict): ...
-def task(_func = None, path: Optional[str] = None, tag: Optional[str] = None, timeout: Optional[int] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
+def task(_func = None, *, path: Optional[str] = None, tag: Optional[str] = None, timeout: Optional[int] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
 
 # Create a task that dispatches to a separate Windmill script.
 # 
@@ -4637,7 +4637,7 @@ def task(_func = None, path: Optional[str] = None, tag: Optional[str] = None, ti
 #     @workflow
 #     async def main():
 #         data = await extract(url="https://...")
-def task_script(path: str, timeout: Optional[int] = None, tag: Optional[str] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
+def task_script(path: str, *, timeout: Optional[int] = None, tag: Optional[str] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
 
 # Create a task that dispatches to a separate Windmill flow.
 # 
@@ -4650,7 +4650,7 @@ def task_script(path: str, timeout: Optional[int] = None, tag: Optional[str] = N
 #     @workflow
 #     async def main():
 #         result = await pipeline(input=data)
-def task_flow(path: str, timeout: Optional[int] = None, tag: Optional[str] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
+def task_flow(path: str, *, timeout: Optional[int] = None, tag: Optional[str] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
 
 # Decorator marking an async function as a workflow-as-code entry point.
 # 
@@ -4711,7 +4711,7 @@ async def wait_for_approval(timeout: int = 1800, form: dict | None = None, self_
 #         ...
 # 
 #     results = await parallel(items, process, concurrency=5)
-async def parallel(items, fn, concurrency: Optional[int] = None)
+async def parallel(items, fn, *, concurrency: Optional[int] = None)
 
 # Commit Kafka offsets for a trigger with auto_commit disabled.
 # 

system_prompts/auto-generated/prompts.ts

@@ -2377,7 +2377,7 @@ def send_teams_message(conversation_id: str, text: str, success: bool = True, ca
 # 
 # Returns:
 #     DataTableClient instance
-def datatable(name: str = 'main', role: Optional[str] = None)
+def datatable(name: str = 'main', *, role: Optional[str] = None)
 
 # Get a DuckLake client for DuckDB queries.
 # 
@@ -2587,7 +2587,7 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]]
 # 
 #     @task(retry={"attempts": 3, "delay": 30, "multiplier": 2})
 #     async def call_api(payload: dict): ...
-def task(_func = None, path: Optional[str] = None, tag: Optional[str] = None, timeout: Optional[int] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
+def task(_func = None, *, path: Optional[str] = None, tag: Optional[str] = None, timeout: Optional[int] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
 
 # Create a task that dispatches to a separate Windmill script.
 # 
@@ -2600,7 +2600,7 @@ def task(_func = None, path: Optional[str] = None, tag: Optional[str] = None, ti
 #     @workflow
 #     async def main():
 #         data = await extract(url="https://...")
-def task_script(path: str, timeout: Optional[int] = None, tag: Optional[str] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
+def task_script(path: str, *, timeout: Optional[int] = None, tag: Optional[str] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
 
 # Create a task that dispatches to a separate Windmill flow.
 # 
@@ -2613,7 +2613,7 @@ def task_script(path: str, timeout: Optional[int] = None, tag: Optional[str] = N
 #     @workflow
 #     async def main():
 #         result = await pipeline(input=data)
-def task_flow(path: str, timeout: Optional[int] = None, tag: Optional[str] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
+def task_flow(path: str, *, timeout: Optional[int] = None, tag: Optional[str] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
 
 # Decorator marking an async function as a workflow-as-code entry point.
 # 
@@ -2674,7 +2674,7 @@ async def wait_for_approval(timeout: int = 1800, form: dict | None = None, self_
 #         ...
 # 
 #     results = await parallel(items, process, concurrency=5)
-async def parallel(items, fn, concurrency: Optional[int] = None)
+async def parallel(items, fn, *, concurrency: Optional[int] = None)
 
 # Commit Kafka offsets for a trigger with auto_commit disabled.
 # 

system_prompts/auto-generated/script.md

@@ -2532,7 +2532,7 @@ def send_teams_message(conversation_id: str, text: str, success: bool = True, ca
 # 
 # Returns:
 #     DataTableClient instance
-def datatable(name: str = 'main', role: Optional[str] = None)
+def datatable(name: str = 'main', *, role: Optional[str] = None)
 
 # Get a DuckLake client for DuckDB queries.
 # 
@@ -2742,7 +2742,7 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]]
 # 
 #     @task(retry={"attempts": 3, "delay": 30, "multiplier": 2})
 #     async def call_api(payload: dict): ...
-def task(_func = None, path: Optional[str] = None, tag: Optional[str] = None, timeout: Optional[int] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
+def task(_func = None, *, path: Optional[str] = None, tag: Optional[str] = None, timeout: Optional[int] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
 
 # Create a task that dispatches to a separate Windmill script.
 # 
@@ -2755,7 +2755,7 @@ def task(_func = None, path: Optional[str] = None, tag: Optional[str] = None, ti
 #     @workflow
 #     async def main():
 #         data = await extract(url="https://...")
-def task_script(path: str, timeout: Optional[int] = None, tag: Optional[str] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
+def task_script(path: str, *, timeout: Optional[int] = None, tag: Optional[str] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
 
 # Create a task that dispatches to a separate Windmill flow.
 # 
@@ -2768,7 +2768,7 @@ def task_script(path: str, timeout: Optional[int] = None, tag: Optional[str] = N
 #     @workflow
 #     async def main():
 #         result = await pipeline(input=data)
-def task_flow(path: str, timeout: Optional[int] = None, tag: Optional[str] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
+def task_flow(path: str, *, timeout: Optional[int] = None, tag: Optional[str] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
 
 # Decorator marking an async function as a workflow-as-code entry point.
 # 
@@ -2829,7 +2829,7 @@ async def wait_for_approval(timeout: int = 1800, form: dict | None = None, self_
 #         ...
 # 
 #     results = await parallel(items, process, concurrency=5)
-async def parallel(items, fn, concurrency: Optional[int] = None)
+async def parallel(items, fn, *, concurrency: Optional[int] = None)
 
 # Commit Kafka offsets for a trigger with auto_commit disabled.
 # 

system_prompts/auto-generated/sdks/python.md

@@ -475,7 +475,7 @@ def send_teams_message(conversation_id: str, text: str, success: bool = True, ca
 # 
 # Returns:
 #     DataTableClient instance
-def datatable(name: str = 'main', role: Optional[str] = None)
+def datatable(name: str = 'main', *, role: Optional[str] = None)
 
 # Get a DuckLake client for DuckDB queries.
 # 
@@ -685,7 +685,7 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]]
 # 
 #     @task(retry={"attempts": 3, "delay": 30, "multiplier": 2})
 #     async def call_api(payload: dict): ...
-def task(_func = None, path: Optional[str] = None, tag: Optional[str] = None, timeout: Optional[int] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
+def task(_func = None, *, path: Optional[str] = None, tag: Optional[str] = None, timeout: Optional[int] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
 
 # Create a task that dispatches to a separate Windmill script.
 # 
@@ -698,7 +698,7 @@ def task(_func = None, path: Optional[str] = None, tag: Optional[str] = None, ti
 #     @workflow
 #     async def main():
 #         data = await extract(url="https://...")
-def task_script(path: str, timeout: Optional[int] = None, tag: Optional[str] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
+def task_script(path: str, *, timeout: Optional[int] = None, tag: Optional[str] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
 
 # Create a task that dispatches to a separate Windmill flow.
 # 
@@ -711,7 +711,7 @@ def task_script(path: str, timeout: Optional[int] = None, tag: Optional[str] = N
 #     @workflow
 #     async def main():
 #         result = await pipeline(input=data)
-def task_flow(path: str, timeout: Optional[int] = None, tag: Optional[str] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
+def task_flow(path: str, *, timeout: Optional[int] = None, tag: Optional[str] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
 
 # Decorator marking an async function as a workflow-as-code entry point.
 # 
@@ -772,7 +772,7 @@ async def wait_for_approval(timeout: int = 1800, form: dict | None = None, self_
 #         ...
 # 
 #     results = await parallel(items, process, concurrency=5)
-async def parallel(items, fn, concurrency: Optional[int] = None)
+async def parallel(items, fn, *, concurrency: Optional[int] = None)
 
 # Commit Kafka offsets for a trigger with auto_commit disabled.
 # 

system_prompts/auto-generated/skills/write-script-python3/SKILL.md

@@ -660,7 +660,7 @@ def send_teams_message(conversation_id: str, text: str, success: bool = True, ca
 # 
 # Returns:
 #     DataTableClient instance
-def datatable(name: str = 'main', role: Optional[str] = None)
+def datatable(name: str = 'main', *, role: Optional[str] = None)
 
 # Get a DuckLake client for DuckDB queries.
 # 
@@ -870,7 +870,7 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]]
 # 
 #     @task(retry={"attempts": 3, "delay": 30, "multiplier": 2})
 #     async def call_api(payload: dict): ...
-def task(_func = None, path: Optional[str] = None, tag: Optional[str] = None, timeout: Optional[int] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
+def task(_func = None, *, path: Optional[str] = None, tag: Optional[str] = None, timeout: Optional[int] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
 
 # Create a task that dispatches to a separate Windmill script.
 # 
@@ -883,7 +883,7 @@ def task(_func = None, path: Optional[str] = None, tag: Optional[str] = None, ti
 #     @workflow
 #     async def main():
 #         data = await extract(url="https://...")
-def task_script(path: str, timeout: Optional[int] = None, tag: Optional[str] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
+def task_script(path: str, *, timeout: Optional[int] = None, tag: Optional[str] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
 
 # Create a task that dispatches to a separate Windmill flow.
 # 
@@ -896,7 +896,7 @@ def task_script(path: str, timeout: Optional[int] = None, tag: Optional[str] = N
 #     @workflow
 #     async def main():
 #         result = await pipeline(input=data)
-def task_flow(path: str, timeout: Optional[int] = None, tag: Optional[str] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
+def task_flow(path: str, *, timeout: Optional[int] = None, tag: Optional[str] = None, cache_ttl: Optional[int] = None, priority: Optional[int] = None, concurrency_limit: Optional[int] = None, concurrency_key: Optional[str] = None, concurrency_time_window_s: Optional[int] = None, retry: Optional[dict] = None)
 
 # Decorator marking an async function as a workflow-as-code entry point.
 # 
@@ -957,7 +957,7 @@ async def wait_for_approval(timeout: int = 1800, form: dict | None = None, self_
 #         ...
 # 
 #     results = await parallel(items, process, concurrency=5)
-async def parallel(items, fn, concurrency: Optional[int] = None)
+async def parallel(items, fn, *, concurrency: Optional[int] = None)
 
 # Commit Kafka offsets for a trigger with auto_commit disabled.
 # 

system_prompts/generate.py

@@ -223,7 +223,12 @@ def extract_py_functions(content: str) -> list[dict]:
         if args.vararg:
             params.append(f"*{args.vararg.arg}")
 
-        # Handle keyword-only args
+        # Handle keyword-only args. Same bare `*` as `_format_py_params`: without it the rendered
+        # signature reads as all-positional, and code written against this reference passes a
+        # keyword-only argument positionally and gets a TypeError.
+        if args.kwonlyargs and not args.vararg:
+            params.append('*')
+
         for i, arg in enumerate(args.kwonlyargs):
             param_str = arg.arg
             if arg.annotation: