diff --git a/backend/.sqlx/query-06abbf945bee93349ff88f64906b96ea1e853ef202510281427cfa9beeff81b3.json b/backend/.sqlx/query-06abbf945bee93349ff88f64906b96ea1e853ef202510281427cfa9beeff81b3.json new file mode 100644 index 0000000000..7adefa7eb8 --- /dev/null +++ b/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" +} diff --git a/backend/.sqlx/query-6f9fb5d72f486358fa25d6887bd69b93910e028f140c07048f2c1c8d63ee6909.json b/backend/.sqlx/query-6f9fb5d72f486358fa25d6887bd69b93910e028f140c07048f2c1c8d63ee6909.json new file mode 100644 index 0000000000..d87d3aeba5 --- /dev/null +++ b/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" +} diff --git a/backend/.sqlx/query-b42af37fb474bea4c5419b0a46d9eadfe384013ab970ccf9c5effd1c78321b7c.json b/backend/.sqlx/query-b42af37fb474bea4c5419b0a46d9eadfe384013ab970ccf9c5effd1c78321b7c.json new file mode 100644 index 0000000000..65c7f47b30 --- /dev/null +++ b/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" +} diff --git a/backend/windmill-api-integration-tests/tests/datatable_roles.rs b/backend/windmill-api-integration-tests/tests/datatable_roles.rs index d3e9255934..2f6d7ca37b 100644 --- a/backend/windmill-api-integration-tests/tests/datatable_roles.rs +++ b/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) 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>(()) } diff --git a/backend/windmill-api-integration-tests/tests/fixtures/datatable_roles.sql b/backend/windmill-api-integration-tests/tests/fixtures/datatable_roles.sql index 0a3fb9372f..42f88e3f13 100644 --- a/backend/windmill-api-integration-tests/tests/fixtures/datatable_roles.sql +++ b/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": { diff --git a/backend/windmill-api-settings/src/lib.rs b/backend/windmill-api-settings/src/lib.rs index 6d0168776d..d5d584bd9b 100644 --- a/backend/windmill-api-settings/src/lib.rs +++ b/backend/windmill-api-settings/src/lib.rs @@ -2633,40 +2633,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, @@ -2711,7 +2677,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", ""); @@ -2768,7 +2734,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; @@ -2811,7 +2777,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. diff --git a/backend/windmill-api-workspaces/src/datatable_permissions.rs b/backend/windmill-api-workspaces/src/datatable_permissions.rs index 41df4d32bc..41ffbb6d35 100644 --- a/backend/windmill-api-workspaces/src/datatable_permissions.rs +++ b/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 = 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?; diff --git a/backend/windmill-api-workspaces/src/workspaces.rs b/backend/windmill-api-workspaces/src/workspaces.rs index e9f0580162..bc05965b2d 100644 --- a/backend/windmill-api-workspaces/src/workspaces.rs +++ b/backend/windmill-api-workspaces/src/workspaces.rs @@ -3599,12 +3599,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 = 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), ) diff --git a/backend/windmill-common/src/datatable_roles.rs b/backend/windmill-common/src/datatable_roles.rs index 19d396e763..522f37410e 100644 --- a/backend/windmill-common/src/datatable_roles.rs +++ b/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 { 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 { 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, diff --git a/backend/windmill-common/src/global_settings.rs b/backend/windmill-common/src/global_settings.rs index ef63fc2347..f9ee875848 100644 --- a/backend/windmill-common/src/global_settings.rs +++ b/backend/windmill-common/src/global_settings.rs @@ -355,8 +355,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, `{ "": { 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 { @@ -988,6 +998,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), diff --git a/backend/windmill-common/src/instance_config.rs b/backend/windmill-common/src/instance_config.rs index 1f897f115f..6ef3a3a09f 100644 --- a/backend/windmill-common/src/instance_config.rs +++ b/backend/windmill-common/src/instance_config.rs @@ -440,13 +440,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 @@ -797,10 +795,6 @@ pub struct CustomInstancePgDatabases { pub user_pwd: Option, #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] pub databases: BTreeMap, - /// 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, } /// Status of a single custom instance database. @@ -975,6 +969,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", @@ -1000,6 +995,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. @@ -1211,17 +1209,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()); } } } diff --git a/backend/windmill-common/src/workspaces.rs b/backend/windmill-common/src/workspaces.rs index 2d5c6a4901..63ba4aaf50 100644 --- a/backend/windmill-common/src/workspaces.rs +++ b/backend/windmill-common/src/workspaces.rs @@ -1426,6 +1426,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 { let datatables = sqlx::query_scalar!( r#" @@ -1453,6 +1458,13 @@ pub async fn read_datatable_entry(db: &DB, w_id: &str, name: &str) -> Result 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. # @@ -4657,7 +4657,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. # @@ -4670,7 +4670,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. # @@ -4735,7 +4735,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. # diff --git a/system_prompts/auto-generated/prompts.ts b/system_prompts/auto-generated/prompts.ts index 4feacfd7db..08ad8ef6c1 100644 --- a/system_prompts/auto-generated/prompts.ts +++ b/system_prompts/auto-generated/prompts.ts @@ -2396,7 +2396,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. # @@ -2614,7 +2614,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. # @@ -2627,7 +2627,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. # @@ -2640,7 +2640,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. # @@ -2705,7 +2705,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. # diff --git a/system_prompts/auto-generated/script.md b/system_prompts/auto-generated/script.md index eba7fd06f4..afc064df40 100644 --- a/system_prompts/auto-generated/script.md +++ b/system_prompts/auto-generated/script.md @@ -2536,7 +2536,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. # @@ -2754,7 +2754,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. # @@ -2767,7 +2767,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. # @@ -2780,7 +2780,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. # @@ -2845,7 +2845,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. # diff --git a/system_prompts/auto-generated/sdks/python.md b/system_prompts/auto-generated/sdks/python.md index 08fe8dd973..97db78b3d4 100644 --- a/system_prompts/auto-generated/sdks/python.md +++ b/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. # @@ -693,7 +693,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. # @@ -706,7 +706,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. # @@ -719,7 +719,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. # @@ -784,7 +784,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. # diff --git a/system_prompts/auto-generated/skills/write-script-python3/SKILL.md b/system_prompts/auto-generated/skills/write-script-python3/SKILL.md index c0a0c2b023..42916f7af5 100644 --- a/system_prompts/auto-generated/skills/write-script-python3/SKILL.md +++ b/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. # @@ -878,7 +878,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. # @@ -891,7 +891,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. # @@ -904,7 +904,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. # @@ -969,7 +969,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. # diff --git a/system_prompts/generate.py b/system_prompts/generate.py index 45b0296747..2860b9d15c 100644 --- a/system_prompts/generate.py +++ b/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: