diff --git a/backend/Cargo.lock b/backend/Cargo.lock index f044478b25..30138085c9 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -17500,6 +17500,7 @@ dependencies = [ "gcp_auth", "git-version", "hex", + "hmac", "hudsucker", "hyper-http-proxy", "hyper-tls", diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index d55596bc72..6576633319 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -11088,6 +11088,129 @@ paths: "200": description: Interactive slack approval message sent successfully + /w/{workspace}/jobs_u/flow/resume_suspended/{job_id}: + post: + summary: resume or cancel a suspended flow/WAC job + description: > + Resume or cancel a suspended flow/WAC job. Uses approval rules to + determine authorization. Either a valid approval_token or an + authenticated session is required. + operationId: resumeSuspended + tags: + - job + parameters: + - $ref: "#/components/parameters/WorkspaceId" + - name: job_id + in: path + required: true + schema: + type: string + format: uuid + requestBody: + required: true + content: + application/json: + schema: + type: object + properties: + payload: + description: payload to send to the resumed job + approval_token: + type: string + description: approval token for unauthenticated access + approved: + type: boolean + description: whether to approve (true) or cancel (false) the job + default: true + responses: + "201": + description: job resumed + content: + text/plain: + schema: + type: string + + /w/{workspace}/jobs_u/flow/approval_info/{job_id}: + get: + summary: get approval info for a suspended flow/WAC job + description: > + Get approval info for a suspended flow/WAC job. Returns form schema, + approval rules, and whether the current user can approve. Either a + valid token query parameter or an authenticated session is required. + operationId: getApprovalInfo + tags: + - job + parameters: + - $ref: "#/components/parameters/WorkspaceId" + - name: job_id + in: path + required: true + schema: + type: string + format: uuid + - name: token + in: query + required: false + schema: + type: string + description: approval token for unauthenticated access + responses: + "200": + description: approval info + content: + application/json: + schema: + type: object + required: + - flow_id + - can_approve + - user_auth_required + - approvers + properties: + flow_id: + type: string + format: uuid + form_schema: + description: form schema for the approval step + description: + description: description of the approval step + approval_conditions: + type: object + properties: + user_auth_required: + type: boolean + user_groups_required: + type: array + items: + type: string + self_approval_disabled: + type: boolean + required: + - user_auth_required + - user_groups_required + - self_approval_disabled + can_approve: + type: boolean + description: whether the current user/token holder can approve + user_auth_required: + type: boolean + description: whether user authentication is required to approve + hide_cancel: + type: boolean + description: whether to hide the cancel button in the UI + approvers: + type: array + items: + type: object + required: + - resume_id + - approver + properties: + resume_id: + type: integer + approver: + type: string + /w/{workspace}/jobs_u/resume/{id}/{resume_id}/{signature}: get: summary: resume a job for a suspended flow diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index a99b2737a8..bde3b81a14 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -103,7 +103,7 @@ use windmill_common::{ cache, db::UserDB, error::{self, to_anyhow, Error}, - flow_status::{Approval, FlowStatus, FlowStatusModule}, + flow_status::{Approval, ApprovalConditions, FlowStatus, FlowStatusModule}, flows::{add_virtual_items_if_necessary, resolve_maybe_value, FlowValue}, jobs::{script_path_to_payload, CompletedJob, JobKind, JobPayload, QueuedJob, RawCode}, oauth2::HmacSha256, @@ -401,6 +401,8 @@ pub fn workspace_unauthed_service() -> Router { post(cancel_persistent_script_api), ) .route("/queue/force_cancel/:id", post(force_cancel)) + .route("/flow/resume_suspended/:job_id", post(resume_suspended)) + .route("/flow/approval_info/:job_id", get(get_approval_info)) } pub fn global_root_service() -> Router { @@ -1058,6 +1060,7 @@ impl<'a> GetQuery<'a> { Self { with_code: false, ..self } } + #[allow(dead_code)] fn without_flow(self) -> Self { Self { with_flow: false, ..self } } @@ -2181,7 +2184,7 @@ pub async fn resume_suspended_flow_as_owner( ) -> error::Result { let mut tx = db.begin().await?; - let (flow, job_id) = get_suspended_flow_info(flow_id, &mut tx).await?; + let (flow, job_id, is_wac) = get_suspended_flow_info(flow_id, &mut tx).await?; let flow_path = flow.script_path.as_deref().unwrap_or_else(|| ""); require_owner_of_path(&authed, flow_path)?; @@ -2189,10 +2192,17 @@ pub async fn resume_suspended_flow_as_owner( // Check approval conditions (self-approval, required groups, etc.) if let Some(ref flow_status_value) = flow.flow_status { - if let Ok(flow_status) = serde_json::from_value::(flow_status_value.clone()) { - let trigger_email = flow.email.as_deref().unwrap_or(""); - conditionally_require_authed_user(Some(authed.clone()), flow_status, trigger_email)?; - } + let trigger_email = flow.email.as_deref().unwrap_or(""); + let ac = serde_json::from_value::(flow_status_value.clone()) + .ok() + .and_then(|fs| fs.approval_conditions) + .or_else(|| { + // WAC flows store approval_conditions directly in flow_status JSONB + flow_status_value + .get("approval_conditions") + .and_then(|v| serde_json::from_value::(v.clone()).ok()) + }); + conditionally_require_authed_user(Some(authed.clone()), ac, trigger_email)?; } let value = value.unwrap_or(serde_json::Value::Null); @@ -2208,12 +2218,426 @@ pub async fn resume_suspended_flow_as_owner( ) .await?; - resume_immediately_if_relevant(flow, job_id, &mut tx).await?; + if is_wac { + // WAC: directly decrement suspend counter + if flow.suspend > 0 { + sqlx::query!( + "UPDATE v2_job_queue SET suspend = GREATEST(suspend - 1, 0) WHERE id = $1", + flow.id, + ) + .execute(&mut *tx) + .await?; + } + } else { + resume_immediately_if_relevant(flow, job_id, &mut tx).await?; + } tx.commit().await?; Ok(StatusCode::CREATED) } +// --- New approval system endpoints --- + +use windmill_common::variables::generate_approval_token; + +/// Verify an approval token against the workspace key + job_id. +async fn validate_approval_token( + db: &DB, + token: &str, + job_id: Uuid, + workspace_id: &str, +) -> error::Result<()> { + let expected = generate_approval_token(workspace_id, job_id, db).await?; + if token != expected { + return Err(Error::NotAuthorized("Invalid approval token".to_string())); + } + Ok(()) +} + +#[derive(Deserialize)] +struct ResumeSuspendedBody { + payload: Option, + approval_token: Option, + approved: Option, +} + +async fn resume_suspended( + OptAuthed(opt_authed): OptAuthed, + Extension(db): Extension, + Path((w_id, job_id)): Path<(String, Uuid)>, + Json(body): Json, +) -> error::Result { + let approved = body.approved.unwrap_or(true); + let value = body.payload.unwrap_or(serde_json::Value::Null); + + // Determine if we have a valid authed user or token + let has_token = if let Some(ref token) = body.approval_token { + validate_approval_token(&db, token, job_id, &w_id) + .await + .is_ok() + } else { + false + }; + + if opt_authed.is_none() && !has_token { + return Err(Error::NotAuthorized( + "Must be logged in or provide a valid approval token".to_string(), + )); + } + + let mut tx = db.begin().await?; + + // Resolve the suspended flow (works for both WAC and classic flows) + let (flow, resume_job_id, is_wac) = get_suspended_flow_info(job_id, &mut tx).await?; + + // Verify the job belongs to this workspace + let job_workspace: Option = + sqlx::query_scalar("SELECT workspace_id FROM v2_job WHERE id = $1") + .bind(&flow.id) + .fetch_optional(&mut *tx) + .await?; + if job_workspace.as_deref() != Some(w_id.as_str()) { + return Err(Error::NotFound( + "Job not found in this workspace".to_string(), + )); + } + + // Check approval conditions + let approval_conditions = if is_wac { + flow.flow_status + .as_ref() + .and_then(|v| v.get("approval_conditions")) + .and_then(|v| serde_json::from_value::(v.clone()).ok()) + } else { + flow.flow_status + .as_ref() + .and_then(|v| serde_json::from_value::(v.clone()).ok()) + .and_then(|fs| fs.approval_conditions) + }; + + if let Some(ref ac) = approval_conditions { + if ac.user_auth_required && opt_authed.is_none() { + return Err(Error::NotAuthorized( + "This approval requires a logged-in user. Please sign in.".to_string(), + )); + } + } + + // If logged in, check authorization rules + if let Some(ref authed) = opt_authed { + let is_admin = authed.is_admin; + let is_owner = flow + .script_path + .as_deref() + .map(|p| require_owner_of_path(authed, p).is_ok()) + .unwrap_or(false); + + if !is_admin && !is_owner { + let trigger_email = flow.email.as_deref().unwrap_or(""); + conditionally_require_authed_user( + Some(authed.clone()), + approval_conditions.clone(), + trigger_email, + )?; + } + } else if !has_token { + return Err(Error::NotAuthorized( + "Must be logged in or provide a valid approval token".to_string(), + )); + } + + // Generate a unique resume_id + let resume_id: u32 = rand::random(); + + // Check for duplicate + let exists: bool = sqlx::query_scalar("SELECT EXISTS (SELECT 1 FROM resume_job WHERE id = $1)") + .bind(Uuid::from_u128(resume_job_id.as_u128() ^ resume_id as u128)) + .fetch_one(&mut *tx) + .await?; + + if exists { + return Err(Error::BadRequest("Resume request already sent".to_string())); + } + + let approver_value = opt_authed.as_ref().map(|a| a.username.clone()); + + insert_resume_job( + resume_id, + resume_job_id, + &flow, + value, + approver_value.clone(), + approved, + &mut tx, + ) + .await?; + + if !approved { + sqlx::query("UPDATE v2_job_queue SET suspend = 0 WHERE id = $1") + .bind(&flow.id) + .execute(&mut *tx) + .await?; + } else if is_wac { + if flow.suspend > 0 { + sqlx::query("UPDATE v2_job_queue SET suspend = GREATEST(suspend - 1, 0) WHERE id = $1") + .bind(&flow.id) + .execute(&mut *tx) + .await?; + } + } else { + resume_immediately_if_relevant(flow, resume_job_id, &mut tx).await?; + } + + let approver = approver_value.unwrap_or_else(|| "anonymous".to_string()); + let audit_author = if let Some(ref authed) = opt_authed { + AuditAuthor::from(authed) + } else { + AuditAuthor { + email: approver.clone(), + username: approver.clone(), + username_override: None, + token_prefix: None, + } + }; + + audit_log( + &mut *tx, + &audit_author, + "jobs.suspend_resume", + ActionKind::Update, + &w_id, + Some( + &serde_json::json!({ + "approved": approved, + "job_id": job_id, + "details": if approved { + format!("Approved by {}", &approver) + } else { + format!("Cancelled by {}", &approver) + } + }) + .to_string(), + ), + None, + ) + .await?; + + tx.commit().await?; + Ok(StatusCode::CREATED) +} + +#[derive(Deserialize)] +struct ApprovalInfoQuery { + token: Option, +} + +#[derive(Serialize)] +struct ApprovalInfo { + flow_id: Uuid, + #[serde(skip_serializing_if = "Option::is_none")] + form_schema: Option, + #[serde(skip_serializing_if = "Option::is_none")] + description: Option, + #[serde(skip_serializing_if = "Option::is_none")] + approval_conditions: Option, + can_approve: bool, + user_auth_required: bool, + #[serde(skip_serializing_if = "Option::is_none")] + hide_cancel: Option, + approvers: Vec, +} + +async fn get_approval_info( + OptAuthed(opt_authed): OptAuthed, + Extension(db): Extension, + Path((w_id, job_id)): Path<(String, Uuid)>, + Query(query): Query, +) -> error::Result> { + // Validate access: either logged in or valid token + let has_token = if let Some(ref token) = query.token { + validate_approval_token(&db, token, job_id, &w_id) + .await + .is_ok() + } else { + false + }; + + if opt_authed.is_none() && !has_token { + return Err(Error::NotAuthorized( + "Must be logged in or provide a valid approval token".to_string(), + )); + } + + // Fetch job info + #[derive(sqlx::FromRow)] + struct ApprovalJobRow { + id: Uuid, + script_path: Option, + email: String, + flow_status: Option, + workflow_as_code_status: Option, + } + let row = sqlx::query_as::<_, ApprovalJobRow>( + "SELECT j.id, j.runnable_path as script_path, j.permissioned_as_email as email, + s.flow_status, s.workflow_as_code_status + FROM v2_job j + LEFT JOIN v2_job_status s ON s.id = j.id + WHERE j.id = $1 AND j.workspace_id = $2", + ) + .bind(&job_id) + .bind(&w_id) + .fetch_optional(&db) + .await? + .ok_or_else(|| Error::NotFound(format!("Job {job_id} not found")))?; + + let is_wac = row.workflow_as_code_status.is_some(); + + // Extract approval info based on WAC vs classic flow + let (form_schema, description, approval_conditions, hide_cancel) = if is_wac { + let approval_meta = row + .workflow_as_code_status + .as_ref() + .and_then(|v| v.get("_approval")); + let form = approval_meta.and_then(|m| m.get("form").cloned()); + let ac = row + .flow_status + .as_ref() + .and_then(|v| v.get("approval_conditions")) + .and_then(|v| serde_json::from_value::(v.clone()).ok()); + (form, None, ac, None) + } else { + let fs = row + .flow_status + .as_ref() + .and_then(|v| serde_json::from_value::(v.clone()).ok()); + let ac = fs.as_ref().and_then(|s| s.approval_conditions.clone()); + + // For classic flows, form/description come from the flow definition and step result + let approval_step = fs.as_ref().map(|s| (s.step as usize).saturating_sub(1)); + + // Fetch flow definition to get suspend settings (form schema, hide_cancel). + // Try raw_flow on the job first, fall back to flow_version for deployed flows. + let raw_flow: Option = { + let from_job: Option = sqlx::query_scalar( + "SELECT raw_flow FROM v2_job WHERE id = $1 AND workspace_id = $2", + ) + .bind(&job_id) + .bind(&w_id) + .fetch_optional(&db) + .await? + .flatten(); + + if let Some(v) = from_job { + serde_json::from_value(v).ok() + } else { + // Deployed flow: fetch from flow_version using runnable_id + let from_version: Option = sqlx::query_scalar( + "SELECT fv.value FROM v2_job j JOIN flow_version fv ON fv.id = j.runnable_id \ + WHERE j.id = $1 AND j.workspace_id = $2", + ) + .bind(&job_id) + .bind(&w_id) + .fetch_optional(&db) + .await? + .flatten(); + from_version.and_then(|v| serde_json::from_value(v).ok()) + } + }; + + let suspend_module = raw_flow + .as_ref() + .and_then(|rf| approval_step.and_then(|s| rf.modules.get(s))); + let suspend_settings = suspend_module.and_then(|m| m.suspend.as_ref()); + + let form = suspend_settings + .and_then(|s| s.resume_form.as_ref()) + .map(|rf| serde_json::json!(rf)); + let hc = suspend_settings.map(|s| s.hide_cancel.unwrap_or(false)); + + // Fetch description and default_args from the step's completed job result + let step_job_id = fs + .as_ref() + .and_then(|s| approval_step.and_then(|step| s.modules.get(step))) + .and_then(|m| m.job()); + let (desc, _default_args) = if let Some(sjid) = step_job_id { + let result: Option = sqlx::query_scalar( + "SELECT result FROM v2_job_completed WHERE id = $1 AND workspace_id = $2", + ) + .bind(sjid) + .bind(&w_id) + .fetch_optional(&db) + .await? + .flatten(); + let desc = result.as_ref().and_then(|r| r.get("description").cloned()); + let da = result.as_ref().and_then(|r| r.get("default_args").cloned()); + (desc, da) + } else { + (None, None) + }; + + (form, desc, ac, hc) + }; + + let user_auth_required = approval_conditions + .as_ref() + .map(|ac| ac.user_auth_required) + .unwrap_or(false); + + // Determine if current user can approve + let can_approve = if let Some(ref authed) = opt_authed { + if authed.is_admin { + true + } else { + let is_owner = row + .script_path + .as_deref() + .map(|p| require_owner_of_path(authed, p).is_ok()) + .unwrap_or(false); + if is_owner { + true + } else { + let trigger_email = row.email.as_str(); + conditionally_require_authed_user( + Some(authed.clone()), + approval_conditions.clone(), + trigger_email, + ) + .is_ok() + } + } + } else { + // Not logged in — can approve only if no auth required + !user_auth_required + }; + + // Get existing approvers + let approvers: Vec = sqlx::query_as::<_, (i32, Option)>( + "SELECT resume_id, approver FROM resume_job WHERE flow = $1", + ) + .bind(&job_id) + .fetch_all(&db) + .await? + .into_iter() + .map(|(rid, approver)| Approval { + resume_id: rid as u16, + approver: approver.unwrap_or_else(|| "anonymous".to_string()), + }) + .collect(); + + Ok(Json(ApprovalInfo { + flow_id: row.id, + form_schema, + description, + approval_conditions, + can_approve, + user_auth_required, + hide_cancel, + approvers, + })) +} + +// --- End new approval system endpoints --- + pub async fn resume_suspended_job( authed: Option, opt_tokened: OptTokened, @@ -2255,26 +2679,8 @@ async fn resume_suspended_job_internal( // Get flow info - works for step-level, flow-level, and WAC approval let (flow_info, is_flow_level, is_wac) = get_flow_info_for_resume(job_id, &db).await?; - // For step-level resumes, verify user auth and flow status - // For flow-level resumes (pre-approvals), the flow might not be at a suspended step yet - // For WAC approvals, skip flow status checks (there is no flow) - if !is_flow_level && !is_wac { - let parent_flow = GetQuery::new() - .without_logs() - .without_code() - .without_flow() - .fetch(&db, &flow_info.id, &w_id) - .await?; - let flow_status = parent_flow - .flow_status() - .ok_or_else(|| anyhow::anyhow!("unable to find the flow status in the flow job"))?; - - let trigger_email = match &parent_flow { - Job::CompletedJob(job) => &job.email, - Job::QueuedJob(job) => &job.email, - }; - conditionally_require_authed_user(authed.clone(), flow_status, trigger_email)?; - } + // HMAC secret = full capability. Skip approval_conditions checks. + // Authorization rules are enforced by the new resume_suspended endpoint instead. let exists = sqlx::query_scalar!( r#" @@ -2540,7 +2946,7 @@ async fn get_flow_info_for_resume(job_id: Uuid, db: &DB) -> error::Result<(FlowI async fn get_suspended_flow_info<'c>( job_id: Uuid, tx: &mut Transaction<'c, Postgres>, -) -> error::Result<(FlowInfo, Uuid)> { +) -> error::Result<(FlowInfo, Uuid, bool)> { let flow = sqlx::query_as!( FlowInfo, r#" @@ -2553,7 +2959,9 @@ async fn get_suspended_flow_info<'c>( .fetch_optional(&mut **tx) .await? .ok_or_else(|| anyhow::anyhow!("parent flow job not found"))?; - let job_id = flow + + // Try to extract step job_id from FlowStatus modules (classic flow path) + let step_job_id = flow .flow_status .as_ref() .and_then(|v| serde_json::from_value::(v.clone()).ok()) @@ -2562,8 +2970,31 @@ async fn get_suspended_flow_info<'c>( _ => None, }); - if let Some(job_id) = job_id { - Ok((flow, job_id)) + if let Some(step_job_id) = step_job_id { + // Classic flow + Ok((flow, step_job_id, false)) + } else if flow.suspend > 0 { + // WAC approval: no FlowStatus modules, but the job is suspended + // The flow_status here comes from COALESCE(flow_status, workflow_as_code_status), + // so for WAC it may contain approval_conditions from flow_status column + // or the WAC checkpoint from workflow_as_code_status column. + // We need the approval_conditions which are in flow_status column. + // Re-fetch just flow_status (without COALESCE fallback) for the auth check. + let flow_status_only: Option = + sqlx::query_scalar("SELECT flow_status FROM v2_job_status WHERE id = $1") + .bind(&job_id) + .fetch_optional(&mut **tx) + .await? + .flatten(); + + let flow = FlowInfo { + id: flow.id, + flow_status: flow_status_only, + suspend: flow.suspend, + script_path: flow.script_path, + email: flow.email, + }; + Ok((flow, job_id, true)) } else { Err(anyhow::anyhow!("the flow is not in a suspended state anymore").into()) } @@ -2640,7 +3071,11 @@ pub async fn get_suspended_job_flow( Job::CompletedJob(job) => &job.email, Job::QueuedJob(job) => &job.email, }; - conditionally_require_authed_user(authed.clone(), flow_status.clone(), trigger_email)?; + conditionally_require_authed_user( + authed.clone(), + flow_status.approval_conditions.clone(), + trigger_email, + )?; let approvers_from_status = match flow_module_status { FlowStatusModule::Success { approvers, .. } => approvers.to_owned(), @@ -2681,16 +3116,25 @@ pub async fn get_suspended_job_flow( fn conditionally_require_authed_user( _authed: Option, - flow_status: FlowStatus, + approval_conditions_opt: Option, _trigger_email: &str, ) -> error::Result<()> { - let approval_conditions_opt = flow_status.approval_conditions; - if approval_conditions_opt.is_none() { return Ok(()); } let approval_conditions = approval_conditions_opt.unwrap(); + // Check self-approval independently of user_auth_required + if approval_conditions.self_approval_disabled { + if let Some(ref authed) = _authed { + if !authed.is_admin && authed.email.eq(_trigger_email) { + return Err(Error::PermissionDenied( + "Self-approval is disabled for this flow step".to_string(), + )); + } + } + } + if approval_conditions.user_auth_required { { #[cfg(not(feature = "enterprise"))] @@ -2708,13 +3152,6 @@ fn conditionally_require_authed_user( let authed = _authed.unwrap(); if !authed.is_admin { - if approval_conditions.self_approval_disabled && authed.email.eq(_trigger_email) - { - return Err(Error::PermissionDenied( - "Self-approval is disabled for this flow step".to_string(), - )); - } - if !approval_conditions.user_groups_required.is_empty() { #[cfg(feature = "enterprise")] { @@ -2860,11 +3297,18 @@ pub async fn get_resume_urls_internal( .map(|x| format!("?approver={}", encode(x))) .unwrap_or_else(String::new); + // Generate approval token for the new approval page URL. + // The token targets the parent flow/WAC job for proper resolution. + let approval_target_id = get_flow_id_for_job(&db, job_id) + .await + .unwrap_or(target_job_id); + let approval_token = generate_approval_token(&w_id, approval_target_id, &db).await?; + let base_url_str = BASE_URL.read().await.clone(); let base_url = base_url_str.as_str(); let res = ResumeUrls { approvalPage: format!( - "{base_url}/approve/{w_id}/{target_job_id}/{resume_id}/{signature}{approver_query}" + "{base_url}/approve/{w_id}/{approval_target_id}?token={approval_token}" ), cancel: build_resume_url( "cancel", diff --git a/backend/windmill-common/src/variables.rs b/backend/windmill-common/src/variables.rs index 8129f39b3f..e7595f9d1b 100644 --- a/backend/windmill-common/src/variables.rs +++ b/backend/windmill-common/src/variables.rs @@ -140,6 +140,24 @@ pub async fn get_workspace_key(w_id: &str, db: &DB) -> crate::error::Result crate::error::Result { + use hmac::{Hmac, Mac}; + use sha2::Sha256; + let key = get_workspace_key(w_id, db).await?; + let mut mac = Hmac::::new_from_slice(key.as_bytes()) + .map_err(|e| crate::Error::internal_err(format!("HMAC key error: {e}")))?; + mac.update(job_id.as_bytes()); + mac.update(b"approval_token"); + Ok(hex::encode(mac.finalize().into_bytes())) +} + pub async fn get_secret_value_as_admin( db: &DB, w_id: &str, diff --git a/backend/windmill-worker/Cargo.toml b/backend/windmill-worker/Cargo.toml index ed82298038..c1e2a4927a 100644 --- a/backend/windmill-worker/Cargo.toml +++ b/backend/windmill-worker/Cargo.toml @@ -110,6 +110,7 @@ gcp_auth = { workspace = true, optional = true } rust_decimal.workspace = true jsonwebtoken.workspace = true sha2.workspace = true +hmac.workspace = true pem = { workspace = true, optional = true } urlencoding.workspace = true nix.workspace = true diff --git a/backend/windmill-worker/src/bun_executor.rs b/backend/windmill-worker/src/bun_executor.rs index f0ed8793a7..7dbaa6723d 100644 --- a/backend/windmill-worker/src/bun_executor.rs +++ b/backend/windmill-worker/src/bun_executor.rs @@ -1504,7 +1504,7 @@ async function run() {{ return {{ type: "inline_checkpoint", key: dispatch.key, result: dispatch.result ?? null, started_at: dispatch.started_at, duration_ms: dispatch.duration_ms }}; }} if (dispatch.mode === "approval") {{ - return {{ type: "approval", key: dispatch.key, timeout: dispatch.timeout, form: dispatch.form }}; + return {{ type: "approval", key: dispatch.key, timeout: dispatch.timeout, form: dispatch.form, self_approval_disabled: dispatch.self_approval_disabled }}; }} if (dispatch.mode === "sleep") {{ return {{ type: "sleep", key: dispatch.key, seconds: dispatch.seconds }}; @@ -2634,7 +2634,7 @@ pub async fn handle_wac_v2_output( job.id, num_steps ))) } - WacOutput::Approval { key, timeout, form } => { + WacOutput::Approval { key, timeout, form, self_approval_disabled } => { let db = match conn { Connection::Sql(db) => db, _ => { @@ -2676,11 +2676,91 @@ pub async fn handle_wac_v2_output( .await .map_err(|e| error::Error::internal_err(format!("Failed to save checkpoint: {e}")))?; + // Store approval_conditions in flow_status for resume endpoint auth checks + let sad = self_approval_disabled.unwrap_or(false); + if sad { + #[cfg(not(feature = "enterprise"))] + return Err(error::Error::ExecutionErr( + "Disabling self-approval is an enterprise only feature".to_string(), + )); + + #[cfg(feature = "enterprise")] + { + use windmill_common::flow_status::ApprovalConditions; + let approval_conditions = ApprovalConditions { + user_auth_required: true, + user_groups_required: vec![], + self_approval_disabled: true, + }; + sqlx::query( + "UPDATE v2_job_status SET flow_status = JSONB_SET( + COALESCE(flow_status, '{}'::jsonb), + '{approval_conditions}', + $2::jsonb + ) WHERE id = $1", + ) + .bind(&job.id) + .bind(&serde_json::json!(approval_conditions)) + .execute(&mut *tx) + .await + .map_err(|e| { + error::Error::internal_err(format!( + "Failed to save approval conditions: {e}" + )) + })?; + } + } + + // Generate resume URLs for the inline approval buttons. + // Use a hash of the step key as resume_id so each waitForApproval() + // in the same workflow gets a unique resume_job record. + let resume_id: u32 = { + use std::hash::{Hash, Hasher}; + let mut hasher = std::collections::hash_map::DefaultHasher::new(); + key.hash(&mut hasher); + (hasher.finish() & 0xFFFF_FFFF) as u32 + }; + // Generate stateless approval token using shared utility + let approval_token = + windmill_common::variables::generate_approval_token(&job.workspace_id, job.id, db) + .await?; + + let (resume_url, cancel_url, approval_page_url) = { + use hmac::{Hmac, Mac}; + use sha2::Sha256; + use windmill_common::variables::get_workspace_key; + + let wkey = get_workspace_key(&job.workspace_id, db).await?; + let mut mac = Hmac::::new_from_slice(wkey.as_bytes()) + .map_err(|e| error::Error::internal_err(format!("HMAC key error: {e}")))?; + mac.update(job.id.as_bytes()); + mac.update(resume_id.to_be_bytes().as_ref()); + let signature = hex::encode(mac.finalize().into_bytes()); + + let base_url = windmill_common::BASE_URL.read().await.clone(); + let w_id = &job.workspace_id; + let job_id = &job.id; + + let resume = format!( + "{base_url}/api/w/{w_id}/jobs_u/resume/{job_id}/{resume_id}/{signature}" + ); + let cancel = format!( + "{base_url}/api/w/{w_id}/jobs_u/cancel/{job_id}/{resume_id}/{signature}" + ); + let approval_page = + format!("{base_url}/approve/{w_id}/{job_id}?token={approval_token}"); + (resume, cancel, approval_page) + }; + // Store approval form metadata for the approval page endpoint let approval_meta = serde_json::json!({ "key": key, "form": form, "timeout": timeout_secs as u32, + "self_approval_disabled": sad, + "resume": resume_url, + "cancel": cancel_url, + "approvalPage": approval_page_url, }); sqlx::query( "UPDATE v2_job_status SET workflow_as_code_status = jsonb_set( @@ -2705,6 +2785,11 @@ pub async fn handle_wac_v2_output( "started_at": &now_str, "name": key, "approval": true, + "self_approval_disabled": sad, + "form": form, + "resume": &resume_url, + "cancel": &cancel_url, + "approvalPage": &approval_page_url, }); let step_timeline_key = format!("_step/{}", key); sqlx::query( diff --git a/backend/windmill-worker/src/wac_executor.rs b/backend/windmill-worker/src/wac_executor.rs index 208012ccce..9b4ba3d92a 100644 --- a/backend/windmill-worker/src/wac_executor.rs +++ b/backend/windmill-worker/src/wac_executor.rs @@ -59,7 +59,13 @@ pub enum WacOutput { /// No child job is dispatched — the parent suspends directly and resumes /// when a user hits the resume/cancel endpoint. #[serde(rename = "approval")] - Approval { key: String, timeout: Option, form: Option }, + Approval { + key: String, + timeout: Option, + form: Option, + #[serde(default)] + self_approval_disabled: Option, + }, /// Server-side sleep — suspend the workflow for a duration without holding a worker. #[serde(rename = "sleep")] Sleep { key: String, seconds: u32 }, @@ -306,15 +312,13 @@ pub async fn prepare_checkpoint_for_resume( } /// Detect WAC v2 patterns in TypeScript/Bun code. -/// Checks for `import ... from "windmill-client"` containing workflow/task, +/// Checks for `import ... from "windmill-client"` containing workflow, /// skipping comment lines. Handles both single-line and multi-line imports. pub fn is_wac_v2_ts(code: &str) -> bool { let mut has_wac_import = false; let mut has_workflow = false; - let mut has_task = false; let mut in_import_block = false; let mut import_block_has_workflow = false; - let mut import_block_has_task = false; for line in code.lines() { let trimmed = line.trim(); if trimmed.starts_with("//") { @@ -328,34 +332,24 @@ pub fn is_wac_v2_ts(code: &str) -> bool { if trimmed.contains("workflow") { has_workflow = true; } - if trimmed.contains("task") { - has_task = true; - } in_import_block = false; } // Start of multi-line import: import { else if trimmed.starts_with("import") && trimmed.contains("{") && !trimmed.contains("}") { in_import_block = true; import_block_has_workflow = trimmed.contains("workflow"); - import_block_has_task = trimmed.contains("task"); } // Inside multi-line import block else if in_import_block { if trimmed.contains("workflow") { import_block_has_workflow = true; } - if trimmed.contains("task") { - import_block_has_task = true; - } // End of multi-line import: } from "windmill-client" if trimmed.contains("windmill-client") { has_wac_import = true; if import_block_has_workflow { has_workflow = true; } - if import_block_has_task { - has_task = true; - } in_import_block = false; } // End of import block but not windmill-client @@ -367,7 +361,7 @@ pub fn is_wac_v2_ts(code: &str) -> bool { has_workflow = true; } } - has_wac_import && has_workflow && has_task + has_wac_import && has_workflow } /// Detect WAC v2 patterns in Python code. diff --git a/cli/src/guidance/skills.ts b/cli/src/guidance/skills.ts index e1354302e6..662c55eb8e 100644 --- a/cli/src/guidance/skills.ts +++ b/cli/src/guidance/skills.ts @@ -740,7 +740,7 @@ workflow(fn: (...args: any[]) => Promise): void * await step("notify", () => sendEmail(urls.approvalPage)); * const { value, approver } = await waitForApproval({ timeout: 3600 }); */ -waitForApproval(options?: { timeout?: number; form?: object; }): PromiseLike<{ value: any; approver: string; approved: boolean }> +waitForApproval(options?: { timeout?: number; form?: object; selfApproval?: boolean; }): PromiseLike<{ value: any; approver: string; approved: boolean }> /** * Process items in parallel with optional concurrency control. @@ -1403,7 +1403,7 @@ workflow(fn: (...args: any[]) => Promise): void * await step("notify", () => sendEmail(urls.approvalPage)); * const { value, approver } = await waitForApproval({ timeout: 3600 }); */ -waitForApproval(options?: { timeout?: number; form?: object; }): PromiseLike<{ value: any; approver: string; approved: boolean }> +waitForApproval(options?: { timeout?: number; form?: object; selfApproval?: boolean; }): PromiseLike<{ value: any; approver: string; approved: boolean }> /** * Process items in parallel with optional concurrency control. @@ -2129,7 +2129,7 @@ workflow(fn: (...args: any[]) => Promise): void * await step("notify", () => sendEmail(urls.approvalPage)); * const { value, approver } = await waitForApproval({ timeout: 3600 }); */ -waitForApproval(options?: { timeout?: number; form?: object; }): PromiseLike<{ value: any; approver: string; approved: boolean }> +waitForApproval(options?: { timeout?: number; form?: object; selfApproval?: boolean; }): PromiseLike<{ value: any; approver: string; approved: boolean }> /** * Process items in parallel with optional concurrency control. @@ -3069,7 +3069,7 @@ workflow(fn: (...args: any[]) => Promise): void * await step("notify", () => sendEmail(urls.approvalPage)); * const { value, approver } = await waitForApproval({ timeout: 3600 }); */ -waitForApproval(options?: { timeout?: number; form?: object; }): PromiseLike<{ value: any; approver: string; approved: boolean }> +waitForApproval(options?: { timeout?: number; form?: object; selfApproval?: boolean; }): PromiseLike<{ value: any; approver: string; approved: boolean }> /** * Process items in parallel with optional concurrency control. @@ -4078,12 +4078,17 @@ async def sleep(seconds: int) # # Returns a dict with \`\`value\`\` (form data), \`\`approver\`\`, and \`\`approved\`\`. # +# Args: +# timeout: Approval timeout in seconds (default 1800). +# form: Optional form schema for the approval page. +# self_approval: Whether the user who triggered the flow can approve it (default True). +# # Example:: # # urls = await step("urls", lambda: get_resume_urls()) # await step("notify", lambda: send_email(urls["approvalPage"])) # result = await wait_for_approval(timeout=3600) -async def wait_for_approval(timeout: int = 1800, form: dict | None = None) -> dict +async def wait_for_approval(timeout: int = 1800, form: dict | None = None, self_approval: bool = True) -> dict # Process items in parallel with optional concurrency control. # diff --git a/frontend/src/lib/components/FlowStatusViewerInner.svelte b/frontend/src/lib/components/FlowStatusViewerInner.svelte index e9f2361629..2c580bb02b 100644 --- a/frontend/src/lib/components/FlowStatusViewerInner.svelte +++ b/frontend/src/lib/components/FlowStatusViewerInner.svelte @@ -2084,6 +2084,7 @@ stepResults={getStepResults(node.workflow_as_code_status)} result={node.result} success={node.type === 'Success'} + jobId={node.job_id} /> {/if} diff --git a/frontend/src/lib/components/FlowStatusWaitingForEvents.svelte b/frontend/src/lib/components/FlowStatusWaitingForEvents.svelte index 60f30b1adc..195a1d6fad 100644 --- a/frontend/src/lib/components/FlowStatusWaitingForEvents.svelte +++ b/frontend/src/lib/components/FlowStatusWaitingForEvents.svelte @@ -18,11 +18,9 @@ light?: boolean } - let { isOwner, workspaceId, job, light = false }: Props = $props() + let { isOwner: _isOwner, workspaceId, job, light = false }: Props = $props() let default_payload: object = $state({}) - let resumeUrl: string | undefined = $state(undefined) - let cancelUrl: string | undefined = $state(undefined) let description: any = $state(undefined) let hide_cancel = $state(false) @@ -49,8 +47,6 @@ defaultValues = JSON.parse(JSON.stringify(args)) default_payload = args - resumeUrl = job_result?.['resume'] - cancelUrl = job_result?.['cancel'] hide_cancel = job?.raw_flow?.modules?.[approvalStep]?.suspend?.hide_cancel ?? false schema = mergeSchema( job?.raw_flow?.modules?.[approvalStep]?.suspend?.resume_form?.schema ?? {}, @@ -61,61 +57,19 @@ let loading = $state(false) async function continu(approve: boolean) { loading = true - if ((resumeUrl && approve) || (cancelUrl && !approve)) { - let split = (approve ? resumeUrl : cancelUrl)!.split('/') - let signatureUrl = split.pop() ?? '' - const regex = /([^?]+)(?:\?[^=]+=(\w+))?/ - - const matches = signatureUrl.match(regex) - - const signature = matches?.[1] - if (!signature) { - sendUserToast(`Could not parse signature: ${signatureUrl}`, true) - return - } - const approver = matches?.[2] || undefined - - let resumeId = -1 - let parsedResumeId = split.pop() ?? '' - try { - resumeId = new Number(parsedResumeId).valueOf() - } catch (e) { - console.error(`Could not parse resume id: ${parsedResumeId}`) - } - let jobId = split.pop() ?? '' - if (approve) { - await JobService.resumeSuspendedJobPost({ - workspace: workspaceId ?? $workspaceStore ?? '', - id: jobId, - requestBody: default_payload as any, - resumeId, - signature, - approver - }) - } else { - await JobService.cancelSuspendedJobPost({ - workspace: workspaceId ?? $workspaceStore ?? '', - id: jobId, - resumeId, - signature, - approver, - requestBody: {} - }) - } - } else { - if (approve) { - await JobService.resumeSuspendedFlowAsOwner({ - workspace: workspaceId ?? $workspaceStore ?? '', - id: job?.id ?? '', - requestBody: default_payload as any - }) - } else { - await JobService.cancelQueuedJob({ - workspace: workspaceId ?? $workspaceStore ?? '', - id: job?.id ?? '', - requestBody: {} - }) - } + try { + await JobService.resumeSuspended({ + workspace: workspaceId ?? $workspaceStore ?? '', + jobId: job?.id ?? '', + requestBody: { + payload: approve ? (default_payload as any) : undefined, + approved: approve + } + }) + } catch (e: any) { + sendUserToast(e?.body ?? e?.message ?? 'Failed', true) + } finally { + loading = false } } let approvalStep = $derived((job?.flow_status?.step ?? 1) - 1) @@ -130,51 +84,41 @@
{/if}
- {#if isOwner || resumeUrl} -
- {#if !hide_cancel} -
-
- {/if} +
+ {#if !hide_cancel}
- +
- - {#if job?.raw_flow?.modules?.[approvalStep]?.suspend?.resume_form?.schema} -
- -
- - The payload is optional, it is passed to the following step through the `resume` - variable - - {/if} + {/if} +
+
- {:else} - You cannot resume the flow yourself without receiving the resume secret since you are not an - owner of {job.script_path} and the approval step did not contain the resume url at key `resume` - {/if} + + {#if job?.raw_flow?.modules?.[approvalStep]?.suspend?.resume_form?.schema} +
+ +
+ + The payload is optional, it is passed to the following step through the `resume` variable + + {/if} +
diff --git a/frontend/src/lib/components/WorkflowTimeline.svelte b/frontend/src/lib/components/WorkflowTimeline.svelte index 3dea7279d6..2ea5ebba50 100644 --- a/frontend/src/lib/components/WorkflowTimeline.svelte +++ b/frontend/src/lib/components/WorkflowTimeline.svelte @@ -1,6 +1,6 @@ {#if flow_status} @@ -167,22 +216,74 @@ sleep ({(v as any).sleep_duration_s}s) {:else if isApproval} -
-
- - - {v.name ?? stepKey(k)} - - {#if !isDone} - - - waiting + {@const selfApprovalDisabled = (v as any).self_approval_disabled === true} + {@const formSchema = (v as any).form?.schema ?? (v as any).form} + {@const hasForm = + formSchema && typeof formSchema === 'object' && Object.keys(formSchema).length > 0} + {@const canApprove = !isDone && jobId} +
+
+
+ + + {v.name ?? stepKey(k)} - {:else} - {msToSec(v.duration_ms ?? 0)}s + {#if !isDone} + + + waiting + + {#if canApprove} +
+ + +
+ {/if} + {:else} + {msToSec(v.duration_ms ?? 0)}s + {/if} +
+ {#if canApprove && selfApprovalDisabled && $userStore?.is_admin} +
+ Self-approval is disabled but allowed because you are an admin/owner +
+ {/if} + {#if canApprove && hasForm} +
+ {#if emptyString($enterpriseLicense)} + + {:else} + + {/if} +
{/if}
{:else} @@ -275,13 +376,13 @@ {@const result = stepResults[stepKey(k)]} {#if isDone && result !== undefined}
-
Result
+
Result
{:else} -
Step completed (no result)
+
Step completed (no result)
{/if} {:else if loadingJobs[k] && !childJobs[k]}
@@ -293,7 +394,7 @@ {#if job.logs || isRunning}
-
Logs
+
Logs
{#if isDone && job.result !== undefined}
-
Result
+
Result
diff --git a/frontend/src/lib/components/flows/content/SuspendDrawer.svelte b/frontend/src/lib/components/flows/content/SuspendDrawer.svelte index ee22349cac..3ac2afe001 100644 --- a/frontend/src/lib/components/flows/content/SuspendDrawer.svelte +++ b/frontend/src/lib/components/flows/content/SuspendDrawer.svelte @@ -39,27 +39,9 @@ render a cancel button, providing the operator with an option to cancel the step. e.g: - {#snippet content()} - - -
{/if} diff --git a/frontend/src/lib/components/scriptEditor/LogPanel.svelte b/frontend/src/lib/components/scriptEditor/LogPanel.svelte index 271e2a765c..82845c4a88 100644 --- a/frontend/src/lib/components/scriptEditor/LogPanel.svelte +++ b/frontend/src/lib/components/scriptEditor/LogPanel.svelte @@ -158,6 +158,7 @@ result={previewJob?.result} success={previewJob?.success !== false} autoExpandResult + jobId={previewJob?.id} />
{:else} diff --git a/frontend/src/lib/script_helpers.ts b/frontend/src/lib/script_helpers.ts index d05d78f6ef..6865637ba7 100644 --- a/frontend/src/lib/script_helpers.ts +++ b/frontend/src/lib/script_helpers.ts @@ -1601,7 +1601,7 @@ export function canHaveApproval(language: SupportedLanguage | undefined): boolea return false } - return ['python3', 'bun', 'deno'].includes(language) + return ['python3', 'bun'].includes(language) } export function canHaveFailure(language: SupportedLanguage | undefined): boolean { diff --git a/frontend/src/routes/(root)/(logged)/run/[...run]/+page.svelte b/frontend/src/routes/(root)/(logged)/run/[...run]/+page.svelte index 5bf8f21ac5..5f7e426fe9 100644 --- a/frontend/src/routes/(root)/(logged)/run/[...run]/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/run/[...run]/+page.svelte @@ -797,6 +797,7 @@ stepResults={getStepResults(job.workflow_as_code_status)} result={job.result} success={(job as any).success !== false} + jobId={job.id} />
diff --git a/frontend/src/routes/approve/[workspace]/[job]/+page.svelte b/frontend/src/routes/approve/[workspace]/[job]/+page.svelte new file mode 100644 index 0000000000..6c9638a948 --- /dev/null +++ b/frontend/src/routes/approve/[workspace]/[job]/+page.svelte @@ -0,0 +1,359 @@ + + + + + + {#if error} +
+ {#if error.includes('logged in') || error.includes('sign in') || error.includes('Not authorized')} +
+ +

Not Authorized

+
+

{error}

+ + {:else if error.includes('Permission denied') || error.includes('Self-approval')} +
+ +

Permission denied

+
+

{error}

+ {:else} +
+ +

Error

+
+

{error}

+ {/if} +
+ {:else if approvalInfo} +
+
+

Approvers

+
+ {#if approvalInfo.approvers?.length > 0} +
    + {#each approvalInfo.approvers as a} +
  • +

    + {a.approver} + Unique id of approval: {a.resume_id} +

    +
    • + {/each} +
+ {:else} +

+ No current approvers for this step (approval steps can require more than one approval) +

+ {/if} +
+
+
+ {#if job && job.raw_flow} + + {/if} +
+
+ + {#if !completed} +

+ {isWac ? 'Workflow' : 'Flow'} arguments +

+ + {/if} + +
+ +
+ {#if completed} + + The flow is not running anymore. You cannot cancel or resume it. + + {/if} + + {#if approvalInfo.description != undefined} + + {/if} + + {#if hasForm && !completed} + {#if emptyString($enterpriseLicense)} + + {:else} + + {/if} + {/if} + + {#if !completed && approvalInfo.can_approve} +
+ {#if approvalInfo.hide_cancel !== true} + + {:else} +
+ {/if} + +
+ {:else if !completed && !approvalInfo.can_approve} + {#if approvalInfo.user_auth_required && !$userStore} + + {:else} +
+

You are not authorized to approve this flow.

+ {#if approvalInfo.approval_conditions?.self_approval_disabled && $userStore && $userStore.email === (job as any)?.email} +

Self-approval is disabled for this step.

+ {/if} + {#if approvalInfo.approval_conditions?.user_groups_required?.length > 0} +

Only members of the following groups can approve: {approvalInfo.approval_conditions.user_groups_required.join(', ')}

+ {/if} +
+ {/if} + {:else if completed} + + {/if} + + {#if !completed && isSelfApprovalBypass} +
+ + As an administrator, by resuming or cancelling this stage of the flow, you bypass the + self-approval interdiction. + +
+ {/if} +
+ +
+ + Open run details (require auth) + +
+ + {#if job && job.raw_flow && !completed} +

Flow details

+
+ +
+ {/if} + {:else} +

Loading...

+ {/if} + diff --git a/python-client/wmill/wmill/client.py b/python-client/wmill/wmill/client.py index cdd3342771..d952584e5a 100644 --- a/python-client/wmill/wmill/client.py +++ b/python-client/wmill/wmill/client.py @@ -2463,7 +2463,7 @@ class WorkflowCtx: ) async def _wait_for_approval( - self, timeout: int = 1800, form: dict | None = None + self, timeout: int = 1800, form: dict | None = None, self_approval: bool = True ): key = self._alloc_key("approval") @@ -2479,6 +2479,7 @@ class WorkflowCtx: "key": key, "timeout": timeout, "form": form, + "self_approval_disabled": not self_approval, "steps": [], }) @@ -2762,6 +2763,7 @@ async def sleep(seconds: int): async def wait_for_approval( timeout: int = 1800, form: dict | None = None, + self_approval: bool = True, ) -> dict: """Suspend the workflow and wait for an external approval. @@ -2770,6 +2772,11 @@ async def wait_for_approval( Returns a dict with ``value`` (form data), ``approver``, and ``approved``. + Args: + timeout: Approval timeout in seconds (default 1800). + form: Optional form schema for the approval page. + self_approval: Whether the user who triggered the flow can approve it (default True). + Example:: urls = await step("urls", lambda: get_resume_urls()) @@ -2778,7 +2785,7 @@ async def wait_for_approval( """ ctx: WorkflowCtx | None = _workflow_ctx.get(None) if ctx is not None: - return await ctx._wait_for_approval(timeout=timeout, form=form) + return await ctx._wait_for_approval(timeout=timeout, form=form, self_approval=self_approval) raise RuntimeError("wait_for_approval can only be called inside a @workflow") diff --git a/system_prompts/auto-generated/prompts.ts b/system_prompts/auto-generated/prompts.ts index 0170d7edca..d30a5be3eb 100644 --- a/system_prompts/auto-generated/prompts.ts +++ b/system_prompts/auto-generated/prompts.ts @@ -632,7 +632,7 @@ workflow(fn: (...args: any[]) => Promise): void * await step("notify", () => sendEmail(urls.approvalPage)); * const { value, approver } = await waitForApproval({ timeout: 3600 }); */ -waitForApproval(options?: { timeout?: number; form?: object; }): PromiseLike<{ value: any; approver: string; approved: boolean }> +waitForApproval(options?: { timeout?: number; form?: object; selfApproval?: boolean; }): PromiseLike<{ value: any; approver: string; approved: boolean }> /** * Process items in parallel with optional concurrency control. @@ -1336,12 +1336,17 @@ async def sleep(seconds: int) # # Returns a dict with \`\`value\`\` (form data), \`\`approver\`\`, and \`\`approved\`\`. # +# Args: +# timeout: Approval timeout in seconds (default 1800). +# form: Optional form schema for the approval page. +# self_approval: Whether the user who triggered the flow can approve it (default True). +# # Example:: # # urls = await step("urls", lambda: get_resume_urls()) # await step("notify", lambda: send_email(urls["approvalPage"])) # result = await wait_for_approval(timeout=3600) -async def wait_for_approval(timeout: int = 1800, form: dict | None = None) -> dict +async def wait_for_approval(timeout: int = 1800, form: dict | None = None, self_approval: bool = True) -> dict # Process items in parallel with optional concurrency control. # diff --git a/system_prompts/auto-generated/script.md b/system_prompts/auto-generated/script.md index ec776de97e..674c9986b9 100644 --- a/system_prompts/auto-generated/script.md +++ b/system_prompts/auto-generated/script.md @@ -1605,7 +1605,7 @@ workflow(fn: (...args: any[]) => Promise): void * await step("notify", () => sendEmail(urls.approvalPage)); * const { value, approver } = await waitForApproval({ timeout: 3600 }); */ -waitForApproval(options?: { timeout?: number; form?: object; }): PromiseLike<{ value: any; approver: string; approved: boolean }> +waitForApproval(options?: { timeout?: number; form?: object; selfApproval?: boolean; }): PromiseLike<{ value: any; approver: string; approved: boolean }> /** * Process items in parallel with optional concurrency control. @@ -2309,12 +2309,17 @@ async def sleep(seconds: int) # # Returns a dict with ``value`` (form data), ``approver``, and ``approved``. # +# Args: +# timeout: Approval timeout in seconds (default 1800). +# form: Optional form schema for the approval page. +# self_approval: Whether the user who triggered the flow can approve it (default True). +# # Example:: # # urls = await step("urls", lambda: get_resume_urls()) # await step("notify", lambda: send_email(urls["approvalPage"])) # result = await wait_for_approval(timeout=3600) -async def wait_for_approval(timeout: int = 1800, form: dict | None = None) -> dict +async def wait_for_approval(timeout: int = 1800, form: dict | None = None, self_approval: bool = True) -> dict # Process items in parallel with optional concurrency control. # diff --git a/system_prompts/auto-generated/sdks/python.md b/system_prompts/auto-generated/sdks/python.md index 7163e76a4a..241d438f58 100644 --- a/system_prompts/auto-generated/sdks/python.md +++ b/system_prompts/auto-generated/sdks/python.md @@ -648,12 +648,17 @@ async def sleep(seconds: int) # # Returns a dict with ``value`` (form data), ``approver``, and ``approved``. # +# Args: +# timeout: Approval timeout in seconds (default 1800). +# form: Optional form schema for the approval page. +# self_approval: Whether the user who triggered the flow can approve it (default True). +# # Example:: # # urls = await step("urls", lambda: get_resume_urls()) # await step("notify", lambda: send_email(urls["approvalPage"])) # result = await wait_for_approval(timeout=3600) -async def wait_for_approval(timeout: int = 1800, form: dict | None = None) -> dict +async def wait_for_approval(timeout: int = 1800, form: dict | None = None, self_approval: bool = True) -> dict # Process items in parallel with optional concurrency control. # diff --git a/system_prompts/auto-generated/sdks/typescript.md b/system_prompts/auto-generated/sdks/typescript.md index f38ba274c1..8d96473313 100644 --- a/system_prompts/auto-generated/sdks/typescript.md +++ b/system_prompts/auto-generated/sdks/typescript.md @@ -481,7 +481,7 @@ workflow(fn: (...args: any[]) => Promise): void * await step("notify", () => sendEmail(urls.approvalPage)); * const { value, approver } = await waitForApproval({ timeout: 3600 }); */ -waitForApproval(options?: { timeout?: number; form?: object; }): PromiseLike<{ value: any; approver: string; approved: boolean }> +waitForApproval(options?: { timeout?: number; form?: object; selfApproval?: boolean; }): PromiseLike<{ value: any; approver: string; approved: boolean }> /** * Process items in parallel with optional concurrency control. diff --git a/system_prompts/auto-generated/skills/write-script-bun/SKILL.md b/system_prompts/auto-generated/skills/write-script-bun/SKILL.md index ba40a2d624..b4db20ae80 100644 --- a/system_prompts/auto-generated/skills/write-script-bun/SKILL.md +++ b/system_prompts/auto-generated/skills/write-script-bun/SKILL.md @@ -610,7 +610,7 @@ workflow(fn: (...args: any[]) => Promise): void * await step("notify", () => sendEmail(urls.approvalPage)); * const { value, approver } = await waitForApproval({ timeout: 3600 }); */ -waitForApproval(options?: { timeout?: number; form?: object; }): PromiseLike<{ value: any; approver: string; approved: boolean }> +waitForApproval(options?: { timeout?: number; form?: object; selfApproval?: boolean; }): PromiseLike<{ value: any; approver: string; approved: boolean }> /** * Process items in parallel with optional concurrency control. diff --git a/system_prompts/auto-generated/skills/write-script-bunnative/SKILL.md b/system_prompts/auto-generated/skills/write-script-bunnative/SKILL.md index cdd015863a..ecf7fe2103 100644 --- a/system_prompts/auto-generated/skills/write-script-bunnative/SKILL.md +++ b/system_prompts/auto-generated/skills/write-script-bunnative/SKILL.md @@ -608,7 +608,7 @@ workflow(fn: (...args: any[]) => Promise): void * await step("notify", () => sendEmail(urls.approvalPage)); * const { value, approver } = await waitForApproval({ timeout: 3600 }); */ -waitForApproval(options?: { timeout?: number; form?: object; }): PromiseLike<{ value: any; approver: string; approved: boolean }> +waitForApproval(options?: { timeout?: number; form?: object; selfApproval?: boolean; }): PromiseLike<{ value: any; approver: string; approved: boolean }> /** * Process items in parallel with optional concurrency control. diff --git a/system_prompts/auto-generated/skills/write-script-deno/SKILL.md b/system_prompts/auto-generated/skills/write-script-deno/SKILL.md index fddae85f6e..563d01ed48 100644 --- a/system_prompts/auto-generated/skills/write-script-deno/SKILL.md +++ b/system_prompts/auto-generated/skills/write-script-deno/SKILL.md @@ -614,7 +614,7 @@ workflow(fn: (...args: any[]) => Promise): void * await step("notify", () => sendEmail(urls.approvalPage)); * const { value, approver } = await waitForApproval({ timeout: 3600 }); */ -waitForApproval(options?: { timeout?: number; form?: object; }): PromiseLike<{ value: any; approver: string; approved: boolean }> +waitForApproval(options?: { timeout?: number; form?: object; selfApproval?: boolean; }): PromiseLike<{ value: any; approver: string; approved: boolean }> /** * Process items in parallel with optional concurrency control. diff --git a/system_prompts/auto-generated/skills/write-script-nativets/SKILL.md b/system_prompts/auto-generated/skills/write-script-nativets/SKILL.md index 4687be55e4..1d52290283 100644 --- a/system_prompts/auto-generated/skills/write-script-nativets/SKILL.md +++ b/system_prompts/auto-generated/skills/write-script-nativets/SKILL.md @@ -575,7 +575,7 @@ workflow(fn: (...args: any[]) => Promise): void * await step("notify", () => sendEmail(urls.approvalPage)); * const { value, approver } = await waitForApproval({ timeout: 3600 }); */ -waitForApproval(options?: { timeout?: number; form?: object; }): PromiseLike<{ value: any; approver: string; approved: boolean }> +waitForApproval(options?: { timeout?: number; form?: object; selfApproval?: boolean; }): PromiseLike<{ value: any; approver: string; approved: boolean }> /** * Process items in parallel with optional concurrency control. 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 c860ee696c..e6aa3b848c 100644 --- a/system_prompts/auto-generated/skills/write-script-python3/SKILL.md +++ b/system_prompts/auto-generated/skills/write-script-python3/SKILL.md @@ -783,12 +783,17 @@ async def sleep(seconds: int) # # Returns a dict with ``value`` (form data), ``approver``, and ``approved``. # +# Args: +# timeout: Approval timeout in seconds (default 1800). +# form: Optional form schema for the approval page. +# self_approval: Whether the user who triggered the flow can approve it (default True). +# # Example:: # # urls = await step("urls", lambda: get_resume_urls()) # await step("notify", lambda: send_email(urls["approvalPage"])) # result = await wait_for_approval(timeout=3600) -async def wait_for_approval(timeout: int = 1800, form: dict | None = None) -> dict +async def wait_for_approval(timeout: int = 1800, form: dict | None = None, self_approval: bool = True) -> dict # Process items in parallel with optional concurrency control. # diff --git a/typescript-client/client.ts b/typescript-client/client.ts index 49087a8b60..1aded4a720 100644 --- a/typescript-client/client.ts +++ b/typescript-client/client.ts @@ -1577,6 +1577,7 @@ export class WorkflowCtx { _waitForApproval(options?: { timeout?: number; form?: object; + selfApproval?: boolean; }): PromiseLike<{ value: any; approver: string; approved: boolean }> { const key = this._allocKey("approval"); @@ -1597,6 +1598,7 @@ export class WorkflowCtx { key, timeout: options?.timeout ?? 1800, form: options?.form, + self_approval_disabled: !(options?.selfApproval ?? true), steps: [], }); } @@ -1842,6 +1844,7 @@ export function workflow(fn: (...args: any[]) => Promise) { export function waitForApproval(options?: { timeout?: number; form?: object; + selfApproval?: boolean; }): PromiseLike<{ value: any; approver: string; approved: boolean }> { const ctx: WorkflowCtx | null = _workflowCtx ?? Reflect.get(globalThis, "__wmill_wf_ctx"); if (!ctx) {