fix: render a task's code the same way in every process

backend/tests/bun_jobs.rs

@@ -1209,6 +1209,100 @@ export function main() {
     Ok(())
 }
 
+/// A deployed flow runs an inline step as the `flow_node` its deploy rewrote it into,
+/// a `FlowScript` job rather than the preview job the editor runs. A workflow-as-code
+/// step's `task()` children must dispatch from that kind too, as re-runs of the same
+/// node, or the step passes its editor test and fails once deployed.
+///
+/// The step is cached: a child that shared the parent's result-cache key would hand
+/// its own result (`10`) back to the parent on resume, in place of the workflow's.
+#[sqlx::test(fixtures("base", "wac_flow_script"))]
+async fn test_bun_wac_task_dispatch_from_flow_script(db: Pool<Postgres>) -> anyhow::Result<()> {
+    use windmill_common::flows::FlowNodeId;
+    initialize_tracing().await;
+    let server = ApiServer::start(db.clone()).await?;
+    let port = server.addr.port();
+
+    let node = FlowNodeId(3000000000000011);
+    let job = RunJob::from(JobPayload::FlowScript {
+        id: node,
+        path: "f/system/wac_flow_script/a".to_string(),
+        language: ScriptLang::Bun,
+        cache_ttl: Some(60),
+        cache_ignore_s3_path: None,
+        dedicated_worker: None,
+        concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default(),
+    })
+    .arg("n", serde_json::json!(5))
+    .run_until_complete(&db, false, port)
+    .await;
+
+    assert_eq!(
+        job.json_result().unwrap(),
+        serde_json::json!({"doubled": 10})
+    );
+
+    let children: Vec<(String, Option<i64>, Option<i32>)> = sqlx::query_as(
+        "SELECT kind::text, runnable_id, cache_ttl FROM v2_job WHERE parent_job = $1",
+    )
+    .bind(job.id)
+    .fetch_all(&db)
+    .await?;
+    assert_eq!(
+        children,
+        vec![("flowscript".to_string(), Some(node.0), None)],
+        "the task child re-runs the parent's flow node, outside the result cache"
+    );
+    Ok(())
+}
+
+/// `task(fn, { cache_ttl })` on an inline task of a deployed flow's step: the child runs
+/// the parent's code with the parent's arguments, so its result-cache key carries its
+/// step key, or the parent and every sibling would read its result back as their own.
+#[sqlx::test(fixtures("base", "wac_flow_script"))]
+async fn test_bun_wac_inline_task_cache_is_per_task(db: Pool<Postgres>) -> anyhow::Result<()> {
+    use windmill_common::flows::FlowNodeId;
+    initialize_tracing().await;
+    let server = ApiServer::start(db.clone()).await?;
+    let port = server.addr.port();
+
+    let payload = || JobPayload::FlowScript {
+        id: FlowNodeId(3000000000000012),
+        path: "f/system/wac_flow_script/a".to_string(),
+        language: ScriptLang::Bun,
+        cache_ttl: None,
+        cache_ignore_s3_path: None,
+        dedicated_worker: None,
+        concurrency_settings: windmill_common::runnable_settings::ConcurrencySettings::default(),
+    };
+
+    let mut children_from_cache = Vec::new();
+    for _ in 0..2 {
+        let job = RunJob::from(payload())
+            .arg("n", serde_json::json!(5))
+            .run_until_complete(&db, false, port)
+            .await;
+        assert_eq!(
+            job.json_result().unwrap(),
+            serde_json::json!({"doubled": 10, "tripled": 15})
+        );
+        let from_cache: i64 = sqlx::query_scalar(
+            "SELECT count(*) FROM job_logs l JOIN v2_job j ON j.id = l.job_id \
+             WHERE j.parent_job = $1 AND l.logs LIKE '%found in cache%'",
+        )
+        .bind(job.id)
+        .fetch_one(&db)
+        .await?;
+        children_from_cache.push(from_cache);
+    }
+    assert_eq!(
+        children_from_cache,
+        vec![0, 2],
+        "the second run serves each task from its own cache entry"
+    );
+    Ok(())
+}
+
 // ============================================================================
 // Environment Variable Tests
 // ============================================================================

backend/tests/fixtures/wac_flow_script.sql

@@ -0,0 +1,53 @@
+-- A deployed flow whose inline bun step is workflow-as-code calling task(), in the
+-- shape the deploy leaves behind: the RawScript module rewritten into a flow_node that
+-- the step then runs as a FlowScript job. No lock, so the worker resolves
+-- windmill-client at run time like the other bun fixtures.
+INSERT INTO public.flow(workspace_id, summary, description, path, versions, schema, value, edited_by) VALUES (
+'test-workspace', '', '',
+'f/system/wac_flow_script',
+'{}',
+'{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{"n":{"type":"integer","description":""}},"required":[],"type":"object"}',
+'{"modules":[{"id":"a","value":{"type":"flowscript","id":3000000000000011,"language":"bun","input_transforms":{"n":{"expr":"flow_input.n","type":"javascript"}}}}]}',
+'system'
+);
+
+INSERT INTO public.flow_node(id, workspace_id, path, hash_v2, lock, code) VALUES (
+3000000000000011,
+'test-workspace',
+'f/system/wac_flow_script',
+'0000000000000000000000000000000000000000000000000000000000000011',
+NULL,
+E'import { workflow, task } from "windmill-client";
+
+const double = task(async (n: number) => {
+  return n * 2;
+});
+
+export const main = workflow(async (n: number) => {
+  const d = await double(n);
+  return { doubled: d };
+});'
+);
+
+-- The same flow's step with two tasks that cache their own result.
+INSERT INTO public.flow_node(id, workspace_id, path, hash_v2, lock, code) VALUES (
+3000000000000012,
+'test-workspace',
+'f/system/wac_flow_script',
+'0000000000000000000000000000000000000000000000000000000000000012',
+NULL,
+E'import { workflow, task } from "windmill-client";
+
+const double = task(async (n: number) => {
+  return n * 2;
+}, { cache_ttl: 60 });
+const triple = task(async (n: number) => {
+  return n * 3;
+}, { cache_ttl: 60 });
+
+export const main = workflow(async (n: number) => {
+  const d = await double(n);
+  const t = await triple(n);
+  return { doubled: d, tripled: t };
+});'
+);

backend/tests/wac_task_cache_key.rs

@@ -0,0 +1,150 @@
+//! A workflow-as-code task child runs its parent's code with its parent's
+//! arguments, so its result-cache key comes from what the dispatch seeded in its
+//! checkpoint: the SDK's fingerprint of the task and the arguments the task was
+//! called with, or, from an SDK that sends no fingerprint, its step key and the
+//! parent's arguments.
+
+use serde_json::{json, value::RawValue, Value};
+use sqlx::{Pool, Postgres};
+use std::collections::HashMap;
+use uuid::Uuid;
+use windmill_common::client::AuthedClient;
+use windmill_common::jobs::JobKind;
+use windmill_common::scripts::{ScriptHash, ScriptLang};
+use windmill_queue::MiniPulledJob;
+use windmill_worker::common::cached_result_path;
+
+const W_ID: &str = "test-workspace";
+
+async fn insert_job(db: &Pool<Postgres>, id: Uuid, parent: Option<Uuid>) -> anyhow::Result<()> {
+    sqlx::query(
+        "INSERT INTO v2_job (id, workspace_id, created_by, created_at, permissioned_as, \
+            permissioned_as_email, kind, script_lang, runnable_path, tag, visible_to_owner, parent_job) \
+         VALUES ($1, $2, 'test-user', now(), 'u/test-user', 'test@windmill.dev', \
+            'flowscript', 'bun', 'f/system/wac/a', 'bun', true, $3)",
+    )
+    .bind(id)
+    .bind(W_ID)
+    .bind(parent)
+    .execute(db)
+    .await?;
+    sqlx::query(
+        "INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, running, tag) \
+         VALUES ($1, $2, now(), true, 'bun')",
+    )
+    .bind(id)
+    .bind(W_ID)
+    .execute(db)
+    .await?;
+    Ok(())
+}
+
+/// The cache path of a task child carrying `parent_args`, whose checkpoint was
+/// seeded with step key `key`, fingerprint `fn_id` and call arguments `{n}`.
+async fn child_cache_path(
+    db: &Pool<Postgres>,
+    parent_args: Value,
+    key: &str,
+    fn_id: Option<&str>,
+    n: i64,
+) -> anyhow::Result<String> {
+    let parent = Uuid::new_v4();
+    let child = Uuid::new_v4();
+    insert_job(db, parent, None).await?;
+    insert_job(db, child, Some(parent)).await?;
+    let mut checkpoint = json!({
+        "completed_steps": {},
+        "_executing_key": key,
+        "_executing_args": { "n": n },
+    });
+    if let Some(fn_id) = fn_id {
+        checkpoint["_executing_fn"] = json!(fn_id);
+    }
+    sqlx::query("INSERT INTO v2_job_status (id, workflow_as_code_status) VALUES ($1, $2)")
+        .bind(child)
+        .bind(json!({ "_checkpoint": checkpoint }))
+        .execute(db)
+        .await?;
+
+    let args: HashMap<String, Box<RawValue>> = serde_json::from_str(&parent_args.to_string())?;
+    let mut job = MiniPulledJob::new_inline(
+        W_ID.to_string(),
+        Some(args),
+        "test-user".to_string(),
+        "u/test-user".to_string(),
+        "test@windmill.dev".to_string(),
+        Some("f/system/wac/a".to_string()),
+        JobKind::FlowScript,
+        Some(ScriptHash(42)),
+        "bun".to_string(),
+        Some(ScriptLang::Bun),
+    );
+    job.id = child;
+    job.parent_job = Some(parent);
+    job.cache_ttl = Some(60);
+    let client = AuthedClient::new(
+        "http://localhost".to_string(),
+        W_ID.to_string(),
+        "tok".to_string(),
+        None,
+    );
+    Ok(cached_result_path(db, &client, &job, None).await?)
+}
+
+#[sqlx::test(fixtures("base"))]
+async fn a_task_child_is_cached_under_its_fingerprint_or_else_its_step_key(
+    db: Pool<Postgres>,
+) -> anyhow::Result<()> {
+    let parent_args = || json!({ "x": 1 });
+
+    let fingerprinted = child_cache_path(&db, parent_args(), "step", Some("f1"), 1).await?;
+    assert_ne!(
+        fingerprinted,
+        child_cache_path(&db, parent_args(), "step", Some("f2"), 1).await?,
+        "two tasks called at one position: the fingerprint tells them apart"
+    );
+    assert_eq!(
+        fingerprinted,
+        child_cache_path(&db, parent_args(), "step", Some("f1"), 1).await?,
+        "one task at one step with one set of arguments is one entry"
+    );
+    assert_ne!(
+        fingerprinted,
+        child_cache_path(&db, parent_args(), "step_2", Some("f1"), 1).await?,
+        "the step key stays in the key: a fingerprint cannot separate two bound \
+         functions of one name, or two lambdas sharing a source line"
+    );
+    assert_eq!(
+        fingerprinted,
+        child_cache_path(&db, json!({ "x": 2 }), "step", Some("f1"), 1).await?,
+        "with a fingerprint, the parent's arguments are not in the key"
+    );
+    assert_ne!(
+        fingerprinted,
+        child_cache_path(&db, parent_args(), "step", Some("f1"), 2).await?,
+        "and the task's own arguments are"
+    );
+
+    let unfingerprinted = child_cache_path(&db, parent_args(), "step", None, 1).await?;
+    assert_ne!(
+        unfingerprinted,
+        child_cache_path(&db, parent_args(), "step_2", None, 1).await?,
+        "without a fingerprint, the step key is the identity"
+    );
+    assert_ne!(
+        unfingerprinted,
+        child_cache_path(&db, json!({ "x": 2 }), "step", None, 1).await?,
+        "and the parent's arguments are in the key"
+    );
+    assert_eq!(
+        unfingerprinted,
+        child_cache_path(&db, parent_args(), "step", None, 2).await?,
+        "and the task's own arguments are not"
+    );
+    assert_ne!(
+        child_cache_path(&db, parent_args(), "f1", None, 1).await?,
+        fingerprinted,
+        "a step key never reads as a fingerprint"
+    );
+    Ok(())
+}

backend/windmill-common/src/wac.rs

@@ -40,6 +40,16 @@ pub struct WacCheckpoint {
     #[serde(skip_serializing_if = "Option::is_none")]
     #[serde(default)]
     pub _executing_key: Option<String>,
+    /// With `_executing_key`: the SDK's fingerprint of the task's code and the
+    /// arguments it was called with. A cached result is keyed on all three, or,
+    /// for a child whose SDK sent no fingerprint, on the step key and the
+    /// parent's arguments (`cached_result_path`).
+    #[serde(skip_serializing_if = "Option::is_none")]
+    #[serde(default)]
+    pub _executing_fn: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    #[serde(default)]
+    pub _executing_args: Option<serde_json::Map<String, Value>>,
     /// `resume_job.id` values already consumed by earlier approval steps (the
     /// row primary key, not the distinct integer `resume_id` column). Rows are
     /// never deleted, so a workflow with several sequential wait_for_approval()

backend/windmill-worker/src/bun_executor.rs

@@ -2874,15 +2874,17 @@ pub async fn handle_wac_v2_output(
                 .collect();
 
             // Resolve job_payload once (same for all children since they re-run
-            // the parent script)
+            // the parent script). The step's cache setting is for the workflow's
+            // result; a task is cached only through its own `cache_ttl` option,
+            // under a key of its own (see `cached_result_path`).
             let job_payload_template = match job.kind {
                 JobKind::Script => {
                     if let Some(hash) = job.runnable_id {
                         Ok(JobPayload::ScriptHash {
                             hash,
                             path: job.runnable_path.clone().unwrap_or_default(),
-                            cache_ttl: job.cache_ttl,
-                            cache_ignore_s3_path: job.cache_ignore_s3_path,
+                            cache_ttl: None,
+                            cache_ignore_s3_path: None,
                             dedicated_worker: None,
                             language: job.script_lang.unwrap_or(ScriptLang::Bun),
                             priority: job.priority,
@@ -2897,6 +2899,27 @@ pub async fn handle_wac_v2_output(
                         ))
                     }
                 }
+                // A deployed flow runs an inline step as the `flow_node` its deploy
+                // rewrote it into; the child re-runs that node the way a `Script`
+                // child re-runs its hash, so `runnable_id` (the checkpoint's source
+                // hash) stays the same across parent and children.
+                JobKind::FlowScript => {
+                    if let Some(id) = job.runnable_id {
+                        Ok(JobPayload::FlowScript {
+                            id: windmill_common::flows::FlowNodeId(id.0),
+                            path: job.runnable_path.clone().unwrap_or_default(),
+                            language: job.script_lang.unwrap_or(ScriptLang::Bun),
+                            cache_ttl: None,
+                            cache_ignore_s3_path: None,
+                            dedicated_worker: None,
+                            concurrency_settings: ConcurrencySettings::default(),
+                        })
+                    } else {
+                        Err(error::Error::internal_err(
+                            "WAC v2 FlowScript job missing runnable_id".to_string(),
+                        ))
+                    }
+                }
                 JobKind::Preview => {
                     let row: Option<(Option<String>, Option<String>)> = sqlx::query_as(
                         "SELECT raw_code, raw_lock FROM v2_job WHERE id = $1 AND workspace_id = $2",
@@ -2912,8 +2935,8 @@ pub async fn handle_wac_v2_output(
                         hash: None,
                         language: job.script_lang.unwrap_or(ScriptLang::Bun),
                         lock: lock,
-                        cache_ttl: job.cache_ttl,
-                        cache_ignore_s3_path: job.cache_ignore_s3_path,
+                        cache_ttl: None,
+                        cache_ignore_s3_path: None,
                         dedicated_worker: None,
                         concurrency_settings: ConcurrencySettingsWithCustom::default(),
                         debouncing_settings: DebouncingSettings::default(),
@@ -3012,6 +3035,12 @@ pub async fn handle_wac_v2_output(
             let mut pushed_ids: Vec<Uuid> = Vec::with_capacity(num_steps);
             let push_result: error::Result<()> = async {
                 for (step, (_, child_uuid)) in steps.iter().zip(job_ids.iter()) {
+                    // A task with a runnable of its own (a deployed script or flow) queues
+                    // at that runnable's priority; any other task is the parent's code and
+                    // queues at the parent's.
+                    let own_runnable = matches!(step.dispatch_type.as_str(), "script" | "flow")
+                        && !step.script.starts_with("./");
+
                     // Resolve job payload based on dispatch_type
                     let (job_payload, child_args, is_external, on_behalf_of) =
                         match step.dispatch_type.as_str() {
@@ -3025,8 +3054,8 @@ pub async fn handle_wac_v2_output(
                                     hash: None,
                                     language: module.language,
                                     lock: module.lock,
-                                    cache_ttl: job.cache_ttl,
-                                    cache_ignore_s3_path: job.cache_ignore_s3_path,
+                                    cache_ttl: None,
+                                    cache_ignore_s3_path: None,
                                     dedicated_worker: None,
                                     concurrency_settings: ConcurrencySettingsWithCustom::default(),
                                     debouncing_settings: DebouncingSettings::default(),
@@ -3110,7 +3139,8 @@ pub async fn handle_wac_v2_output(
                     let mut job_payload = job_payload;
                     if let Some(cache_ttl) = step.cache_ttl {
                         match &mut job_payload {
-                            JobPayload::ScriptHash { cache_ttl: ref mut ct, .. } => {
+                            JobPayload::ScriptHash { cache_ttl: ref mut ct, .. }
+                            | JobPayload::FlowScript { cache_ttl: ref mut ct, .. } => {
                                 *ct = Some(cache_ttl)
                             }
                             JobPayload::Code(ref mut code) => code.cache_ttl = Some(cache_ttl),
@@ -3122,7 +3152,8 @@ pub async fn handle_wac_v2_output(
                         || step.concurrency_time_window_s.is_some()
                     {
                         match &mut job_payload {
-                            JobPayload::ScriptHash { concurrency_settings: ref mut cs, .. } => {
+                            JobPayload::ScriptHash { concurrency_settings: ref mut cs, .. }
+                            | JobPayload::FlowScript { concurrency_settings: ref mut cs, .. } => {
                                 if let Some(limit) = step.concurrent_limit {
                                     cs.concurrent_limit = Some(limit);
                                 }
@@ -3188,13 +3219,14 @@ pub async fn handle_wac_v2_output(
                         job.visible_to_owner,
                         step.tag.clone().or_else(|| Some(job.tag.clone())),
                         step.timeout.or(job.timeout),
-                        None,          // flow_step_id
-                        step.priority, // priority_override
-                        None,          // authed
-                        false,         // running
-                        None,          // end_user_email
-                        None,          // trigger
-                        None,          // suspended_mode
+                        None, // flow_step_id
+                        step.priority
+                            .or(if own_runnable { None } else { job.priority }),
+                        None,  // authed
+                        false, // running
+                        None,  // end_user_email
+                        None,  // trigger
+                        None,  // suspended_mode
                     )
                     .await?;
 
@@ -3202,10 +3234,14 @@ pub async fn handle_wac_v2_output(
                     // _executing_key to know which step to run). External
                     // scripts/flows don't need a WAC checkpoint.
                     if !is_external {
-                        let child_checkpoint_json = serde_json::json!({
+                        let mut child_checkpoint_json = serde_json::json!({
                             "completed_steps": &checkpoint.completed_steps,
                             "_executing_key": &step.key,
+                            "_executing_args": &step.args,
                         });
+                        if let Some(fn_id) = &step.fn_id {
+                            child_checkpoint_json["_executing_fn"] = serde_json::json!(fn_id);
+                        }
                         sqlx::query(
                             "INSERT INTO v2_job_status (id, workflow_as_code_status)
                              VALUES ($1, jsonb_build_object('_checkpoint', $2::jsonb))

backend/windmill-worker/src/common.rs

@@ -1559,7 +1559,7 @@ pub async fn cached_result_path(
     client: &AuthedClient,
     job: &MiniPulledJob,
     raw_data: Option<&RawData>,
-) -> String {
+) -> windmill_common::error::Result<String> {
     let mut hasher = sha2::Sha256::new();
     hasher.update(&[job.kind as u8]);
     if let Some(ScriptHash(hash)) = job.runnable_id {
@@ -1574,17 +1574,76 @@ pub async fn cached_result_path(
             _ => {}
         }
     }
+    // A task child runs its parent's code with its parent's arguments. With the SDK's
+    // fingerprint it is keyed on that, the step it runs as and its own call arguments.
+    // Without one, on its step key and the parent's arguments, which is what tells apart
+    // tasks a branch puts at one step.
+    let args = match wac_task_identity(db, job).await? {
+        Some(WacTaskIdentity::Fingerprint { key, fn_id, args }) => {
+            // The step key stays in: a fingerprint cannot separate two tasks whose
+            // difference it never sees, such as two bound functions of one name or
+            // two lambdas sharing a source line, and the step key can.
+            hasher.update(b"wac_fn:");
+            hasher.update(fn_id.as_bytes());
+            hasher.update(b"@");
+            hasher.update(key.as_bytes());
+            Some(Json(args))
+        }
+        Some(WacTaskIdentity::StepKey(key)) => {
+            hasher.update(b"wac_step:");
+            hasher.update(key.as_bytes());
+            job.args.clone()
+        }
+        None => job.args.clone(),
+    };
     hash_args(
         db,
         client,
         &job.workspace_id,
-        &job.args,
+        &args,
         &mut hasher,
         &job.id,
         job.cache_ignore_s3_path.unwrap_or(false),
     )
     .await;
-    format!("g/results/{:064x}", hasher.finalize())
+    Ok(format!("g/results/{:064x}", hasher.finalize()))
+}
+
+/// What a workflow-as-code parent seeded in a task child's checkpoint at push time
+/// to key the child's cached result on.
+enum WacTaskIdentity {
+    Fingerprint { key: String, fn_id: String, args: HashMap<String, Box<RawValue>> },
+    StepKey(String),
+}
+
+/// `None` for any job that is not a workflow-as-code task child.
+async fn wac_task_identity(
+    db: &DB,
+    job: &MiniPulledJob,
+) -> windmill_common::error::Result<Option<WacTaskIdentity>> {
+    if job.parent_job.is_none() || job.flow_step_id.is_some() {
+        return Ok(None);
+    }
+    let identity: Option<(
+        Option<String>,
+        Option<String>,
+        Option<Json<HashMap<String, Box<RawValue>>>>,
+    )> = sqlx::query_as(
+        "SELECT workflow_as_code_status->'_checkpoint'->>'_executing_fn', \
+                workflow_as_code_status->'_checkpoint'->>'_executing_key', \
+                workflow_as_code_status->'_checkpoint'->'_executing_args' \
+         FROM v2_job_status WHERE id = $1",
+    )
+    .bind(job.id)
+    .fetch_optional(db)
+    .await?;
+    Ok(match identity {
+        Some((Some(fn_id), Some(key), Some(Json(args)))) => {
+            Some(WacTaskIdentity::Fingerprint { key, fn_id, args })
+        }
+        Some((_, Some(key), _)) => Some(WacTaskIdentity::StepKey(key)),
+        _ => None,
+    })
 }
 
 #[cfg(feature = "parquet")]

backend/windmill-worker/src/wac_executor.rs

@@ -85,6 +85,10 @@ pub struct WacStepDispatch {
     pub concurrency_key: Option<String>,
     #[serde(default)]
     pub concurrency_time_window_s: Option<i32>,
+    /// The SDK's fingerprint of the task's code, the identity its cached result
+    /// is keyed on; absent from an SDK that predates it.
+    #[serde(default)]
+    pub fn_id: Option<String>,
 }
 
 fn default_dispatch_type() -> String {

backend/windmill-worker/src/worker.rs

@@ -4645,7 +4645,7 @@ pub async fn handle_queued_job(
     let cached_res_path = if job.cache_ttl.is_some() {
         match conn {
             Connection::Sql(db) => {
-                Some(cached_result_path(db, &client, &job, preview_data.as_ref()).await)
+                Some(cached_result_path(db, &client, &job, preview_data.as_ref()).await?)
             }
             Connection::Http(_) => None,
         }

backend/windmill-worker/src/worker_flow.rs

@@ -2005,7 +2005,8 @@ pub async fn update_flow_status_after_job_completion_internal(
 
             if flow_job.cache_ttl.is_some() && success {
                 let flow = RawData::Flow(flow_data.clone());
-                let cached_res_path = cached_result_path(db, client, &flow_job, Some(&flow)).await;
+                let cached_res_path =
+                    cached_result_path(db, client, &flow_job, Some(&flow)).await?;
 
                 save_in_cache(
                     db,

cli/src/guidance/skills.gen.ts

@@ -4608,6 +4608,13 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]]
 # it grows with both the width of the fan-out and \`\`attempts\`\`. Retries with
 # no \`\`delay\`\` all go out in a single round.
 # 
+# \`\`cache_ttl\`\` serves a previous result of the task for that many seconds
+# instead of running it again. The result is keyed on the task, the step it
+# runs as and the arguments it is called with, so anything a cached task reads
+# from its closure, the receiver of a bound method included, must be passed in
+# as an argument. It has no effect on a \`\`task_flow\`\` target, which keeps its
+# flow's own cache policy.
+# 
 # Usage::
 # 
 #     @task
@@ -6741,6 +6748,12 @@ export interface TaskRetry {
 export interface TaskOptions {
   timeout?: number;
   tag?: string;
+  /** Seconds during which a previous result of this task is served instead of
+  *  running it again. The result is keyed on the task, the step it runs as and
+  *  the arguments it is called with, so anything a cached task reads from its
+  *  closure, the receiver of a bound method included, must be passed in as an
+  *  argument. It has no effect on a \`taskFlow\` target, which keeps its flow's
+  *  own cache policy. */
   cache_ttl?: number;
   priority?: number;
   concurrency_limit?: number;
@@ -6932,6 +6945,13 @@ def get_resume_urls(approver: str = None, flow_level: bool = None) -> dict
 # it grows with both the width of the fan-out and \`\`attempts\`\`. Retries with
 # no \`\`delay\`\` all go out in a single round.
 #
+# \`\`cache_ttl\`\` serves a previous result of the task for that many seconds
+# instead of running it again. The result is keyed on the task, the step it
+# runs as and the arguments it is called with, so anything a cached task reads
+# from its closure, the receiver of a bound method included, must be passed in
+# as an argument. It has no effect on a \`\`task_flow\`\` target, which keeps its
+# flow's own cache policy.
+#
 # Usage::
 #
 #     @task

python-client/wmill/tests/test_workflow.py

@@ -1847,3 +1847,84 @@ class TestApprovalKeys:
             await wait_for_approval()
 
         assert _run_workflow(wf, {"completed_steps": {"approval": {}}}, {})["key"] == "approval_2"
+
+
+class TestTaskFingerprint:
+    """The worker keys a task child's cached result on the ``fn_id`` its dispatch
+    carries, so it has to follow the task, not the position it is called at."""
+
+    def test_one_task_keeps_its_fingerprint_and_another_gets_its_own(self):
+        first = _run_workflow(double_parallel_wf, {}, {})
+        assert [s["key"] for s in first["steps"]] == ["double", "double_2"]
+        assert first["steps"][0]["fn_id"] == first["steps"][1]["fn_id"]
+
+        second = _run_workflow(
+            double_parallel_wf, {"completed_steps": {"double": 2, "double_2": 4}}, {}
+        )
+        assert [s["key"] for s in second["steps"]] == ["add_one", "add_one_2"]
+        assert second["steps"][0]["fn_id"] != first["steps"][0]["fn_id"]
+
+    def test_a_builtin_task_still_decorates_and_dispatches(self):
+        """The fingerprint is taken for every task, cached or not, so a callable
+        with neither source nor code object must not break the decorator."""
+        builtin_task = task(pow)
+
+        @workflow
+        async def wf():
+            return await builtin_task(2, 3)
+
+        result = _run_workflow(wf, {}, {})
+        assert result["type"] == "dispatch"
+        assert result["steps"][0]["key"] == "pow"
+
+    def test_two_lambdas_on_one_line_are_told_apart(self):
+        """``inspect.getsource`` gives each the whole line, so the code's shape is
+        what separates them, and it must not depend on where the line sits."""
+        first, second = task(lambda x: x + 1), task(lambda x: x + 2)
+
+        @workflow
+        async def wf():
+            return await asyncio.gather(first(x=1), second(x=1))
+
+        result = _run_workflow(wf, {}, {})
+        assert result["steps"][0]["fn_id"] != result["steps"][1]["fn_id"]
+
+    def test_two_lambdas_differing_inside_a_genexp_are_told_apart(self):
+        """The difference lives in a nested code object, and they share a source
+        line, so the shape has to be read recursively."""
+        first, second = task(lambda xs: sum(x + 1 for x in xs)), task(lambda xs: sum(x + 2 for x in xs))
+
+        @workflow
+        async def wf():
+            return await asyncio.gather(first(xs=[1]), second(xs=[1]))
+
+        result = _run_workflow(wf, {}, {})
+        assert result["steps"][0]["fn_id"] != result["steps"][1]["fn_id"]
+
+    def test_fingerprint_does_not_move_with_the_interpreter_hash_seed(self):
+        """A set constant renders in hash order, which the interpreter randomizes
+        per process: a fingerprint that moved with it would never hit its cache."""
+        import os
+        import pathlib
+        import subprocess
+        import sys
+
+        script = (
+            "from wmill.client import _fn_fingerprint\n"
+            "def t(x):\n"
+            "    return x in frozenset({'a', 'b', 'c', 'd', 'e'})\n"
+            "print(_fn_fingerprint(t))\n"
+        )
+        root = str(pathlib.Path(__file__).resolve().parents[1])
+        seen = set()
+        for seed in ("1", "2"):
+            env = {**os.environ, "PYTHONHASHSEED": seed, "PYTHONPATH": root}
+            out = subprocess.run(
+                [sys.executable, "-c", script],
+                capture_output=True,
+                text=True,
+                env=env,
+                check=True,
+            )
+            seen.add(out.stdout.strip())
+        assert len(seen) == 1

python-client/wmill/wmill/client.py

@@ -3026,7 +3026,7 @@ class WorkflowCtx:
             print(f"\n--- WAC: {key} ---")
             info = {"name": name or key, "script": script or key, "args": kwargs, "key": key, "dispatch_type": dispatch_type}
             if _task_options:
-                for opt_key in ("timeout", "tag", "cache_ttl", "priority", "concurrent_limit", "concurrency_key", "concurrency_time_window_s"):
+                for opt_key in ("timeout", "tag", "cache_ttl", "priority", "concurrent_limit", "concurrency_key", "concurrency_time_window_s", "fn_id"):
                     if opt_key in _task_options and _task_options[opt_key] is not None:
                         info[opt_key] = _task_options[opt_key]
             self._pending.append(info)
@@ -3285,6 +3285,60 @@ class WorkflowCtx:
         })
 
 
+def _code_shape(code) -> bytes:
+    """What a code object does, rendered the same way in every process.
+
+    A set renders in hash order, which the interpreter randomizes per process, so
+    a fingerprint built on it would change between jobs and never find its cached
+    result; sorting the members fixes that. A nested code object is where a
+    generator expression's body lives, and two of those can be the only
+    difference between two functions, so it is rendered rather than skipped.
+    """
+    import types
+
+    def const(c) -> bytes:
+        if isinstance(c, types.CodeType):
+            return b"code:" + _code_shape(c)
+        if isinstance(c, (frozenset, set)):
+            return b"set:" + b",".join(sorted(const(x) for x in c))
+        if isinstance(c, tuple):
+            return b"tuple:" + b",".join(const(x) for x in c)
+        return f"{type(c).__name__}:{c!r}".encode()
+
+    return b"|".join(
+        [
+            code.co_code,
+            repr(code.co_names).encode(),
+            repr(code.co_varnames).encode(),
+            b",".join(const(c) for c in code.co_consts),
+        ]
+    )
+
+
+def _fn_fingerprint(func) -> str:
+    """A stable identity for a task's code, what its cached result is keyed on: a
+    name is shared by any two tasks called the same, and a step key by any two
+    tasks called at the same position, so neither can tell them apart."""
+    import hashlib
+    import inspect
+
+    # Runs for every task at decoration, cached or not, so it must never raise: a
+    # builtin has neither source nor code object. Source alone cannot separate two
+    # lambdas written on one line, so the code's shape goes in as well, without the
+    # line numbers that would move whenever the file is edited above it.
+    parts = []
+    try:
+        parts.append(inspect.getsource(func).encode())
+    except Exception:
+        pass
+    code = getattr(func, "__code__", None)
+    if code is not None:
+        parts.append(_code_shape(code))
+    if not parts:
+        parts.append(repr(func).encode())
+    return hashlib.sha256(b"\x1f".join(parts)).hexdigest()
+
+
 def task(
     _func=None,
     *,
@@ -3327,6 +3381,13 @@ def task(
     it grows with both the width of the fan-out and ``attempts``. Retries with
     no ``delay`` all go out in a single round.
 
+    ``cache_ttl`` serves a previous result of the task for that many seconds
+    instead of running it again. The result is keyed on the task, the step it
+    runs as and the arguments it is called with, so anything a cached task reads
+    from its closure, the receiver of a bound method included, must be passed in
+    as an argument. It has no effect on a ``task_flow`` target, which keeps its
+    flow's own cache policy.
+
     Usage::
 
         @task
@@ -3356,6 +3417,7 @@ def task(
     def decorator(func) -> Callable[..., Any]:
         task_path = path
         task_name = func.__name__
+        _fn_opts = {**(_task_opts or {}), "fn_id": _fn_fingerprint(func)}
 
         _params_list = list(_sig(func).parameters)
 
@@ -3381,7 +3443,7 @@ def task(
             if ctx is not None:
                 script = task_path if task_path else task_name
                 merged = _merge_args(args, kwargs)
-                return ctx._next_step(task_name, script, func, _task_options=_task_opts, **merged)
+                return ctx._next_step(task_name, script, func, _task_options=_fn_opts, **merged)
 
             # WAC v1: running inside a Windmill job but not in a @workflow
             if (

system_prompts/auto-generated/prompts.ts

@@ -2587,6 +2587,13 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]]
 # it grows with both the width of the fan-out and \`\`attempts\`\`. Retries with
 # no \`\`delay\`\` all go out in a single round.
 # 
+# \`\`cache_ttl\`\` serves a previous result of the task for that many seconds
+# instead of running it again. The result is keyed on the task, the step it
+# runs as and the arguments it is called with, so anything a cached task reads
+# from its closure, the receiver of a bound method included, must be passed in
+# as an argument. It has no effect on a \`\`task_flow\`\` target, which keeps its
+# flow's own cache policy.
+# 
 # Usage::
 # 
 #     @task
@@ -2737,6 +2744,12 @@ export interface TaskRetry {
 export interface TaskOptions {
   timeout?: number;
   tag?: string;
+  /** Seconds during which a previous result of this task is served instead of
+  *  running it again. The result is keyed on the task, the step it runs as and
+  *  the arguments it is called with, so anything a cached task reads from its
+  *  closure, the receiver of a bound method included, must be passed in as an
+  *  argument. It has no effect on a \`taskFlow\` target, which keeps its flow's
+  *  own cache policy. */
   cache_ttl?: number;
   priority?: number;
   concurrency_limit?: number;
@@ -2928,6 +2941,13 @@ def get_resume_urls(approver: str = None, flow_level: bool = None) -> dict
 # it grows with both the width of the fan-out and \`\`attempts\`\`. Retries with
 # no \`\`delay\`\` all go out in a single round.
 #
+# \`\`cache_ttl\`\` serves a previous result of the task for that many seconds
+# instead of running it again. The result is keyed on the task, the step it
+# runs as and the arguments it is called with, so anything a cached task reads
+# from its closure, the receiver of a bound method included, must be passed in
+# as an argument. It has no effect on a \`\`task_flow\`\` target, which keeps its
+# flow's own cache policy.
+#
 # Usage::
 #
 #     @task

system_prompts/auto-generated/script.md

@@ -2728,6 +2728,13 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]]
 # it grows with both the width of the fan-out and ``attempts``. Retries with
 # no ``delay`` all go out in a single round.
 # 
+# ``cache_ttl`` serves a previous result of the task for that many seconds
+# instead of running it again. The result is keyed on the task, the step it
+# runs as and the arguments it is called with, so anything a cached task reads
+# from its closure, the receiver of a bound method included, must be passed in
+# as an argument. It has no effect on a ``task_flow`` target, which keeps its
+# flow's own cache policy.
+# 
 # Usage::
 # 
 #     @task

system_prompts/auto-generated/sdks/python.md

@@ -672,6 +672,13 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]]
 # it grows with both the width of the fan-out and ``attempts``. Retries with
 # no ``delay`` all go out in a single round.
 # 
+# ``cache_ttl`` serves a previous result of the task for that many seconds
+# instead of running it again. The result is keyed on the task, the step it
+# runs as and the arguments it is called with, so anything a cached task reads
+# from its closure, the receiver of a bound method included, must be passed in
+# as an argument. It has no effect on a ``task_flow`` target, which keeps its
+# flow's own cache policy.
+# 
 # Usage::
 # 
 #     @task

system_prompts/auto-generated/sdks/wac-python.md

@@ -58,6 +58,13 @@ def get_resume_urls(approver: str = None, flow_level: bool = None) -> dict
 # it grows with both the width of the fan-out and ``attempts``. Retries with
 # no ``delay`` all go out in a single round.
 #
+# ``cache_ttl`` serves a previous result of the task for that many seconds
+# instead of running it again. The result is keyed on the task, the step it
+# runs as and the arguments it is called with, so anything a cached task reads
+# from its closure, the receiver of a bound method included, must be passed in
+# as an argument. It has no effect on a ``task_flow`` target, which keeps its
+# flow's own cache policy.
+#
 # Usage::
 #
 #     @task

system_prompts/auto-generated/sdks/wac-typescript.md

@@ -34,6 +34,12 @@ export interface TaskRetry {
 export interface TaskOptions {
   timeout?: number;
   tag?: string;
+  /** Seconds during which a previous result of this task is served instead of
+  *  running it again. The result is keyed on the task, the step it runs as and
+  *  the arguments it is called with, so anything a cached task reads from its
+  *  closure, the receiver of a bound method included, must be passed in as an
+  *  argument. It has no effect on a `taskFlow` target, which keeps its flow's
+  *  own cache policy. */
   cache_ttl?: number;
   priority?: number;
   concurrency_limit?: number;

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

@@ -857,6 +857,13 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]]
 # it grows with both the width of the fan-out and ``attempts``. Retries with
 # no ``delay`` all go out in a single round.
 # 
+# ``cache_ttl`` serves a previous result of the task for that many seconds
+# instead of running it again. The result is keyed on the task, the step it
+# runs as and the arguments it is called with, so anything a cached task reads
+# from its closure, the receiver of a bound method included, must be passed in
+# as an argument. It has no effect on a ``task_flow`` target, which keeps its
+# flow's own cache policy.
+# 
 # Usage::
 # 
 #     @task

system_prompts/auto-generated/skills/write-workflow-as-code/SKILL.md

@@ -277,6 +277,12 @@ export interface TaskRetry {
 export interface TaskOptions {
   timeout?: number;
   tag?: string;
+  /** Seconds during which a previous result of this task is served instead of
+  *  running it again. The result is keyed on the task, the step it runs as and
+  *  the arguments it is called with, so anything a cached task reads from its
+  *  closure, the receiver of a bound method included, must be passed in as an
+  *  argument. It has no effect on a `taskFlow` target, which keeps its flow's
+  *  own cache policy. */
   cache_ttl?: number;
   priority?: number;
   concurrency_limit?: number;
@@ -468,6 +474,13 @@ def get_resume_urls(approver: str = None, flow_level: bool = None) -> dict
 # it grows with both the width of the fan-out and ``attempts``. Retries with
 # no ``delay`` all go out in a single round.
 #
+# ``cache_ttl`` serves a previous result of the task for that many seconds
+# instead of running it again. The result is keyed on the task, the step it
+# runs as and the arguments it is called with, so anything a cached task reads
+# from its closure, the receiver of a bound method included, must be passed in
+# as an argument. It has no effect on a ``task_flow`` target, which keeps its
+# flow's own cache policy.
+#
 # Usage::
 #
 #     @task

typescript-client/client.ts

@@ -1712,6 +1712,12 @@ export interface TaskRetry {
 export interface TaskOptions {
   timeout?: number;
   tag?: string;
+  /** Seconds during which a previous result of this task is served instead of
+   *  running it again. The result is keyed on the task, the step it runs as and
+   *  the arguments it is called with, so anything a cached task reads from its
+   *  closure, the receiver of a bound method included, must be passed in as an
+   *  argument. It has no effect on a `taskFlow` target, which keeps its flow's
+   *  own cache policy. */
   cache_ttl?: number;
   priority?: number;
   concurrency_limit?: number;
@@ -1849,6 +1855,7 @@ export class WorkflowCtx {
     args: Record<string, any> = {},
     dispatch_type: string = "inline",
     options?: TaskOptions,
+    fnId?: string,
   ): PromiseLike<any> {
     this._rethrowSwallowed();
     const stepName = name || script || "step";
@@ -1902,6 +1909,7 @@ export class WorkflowCtx {
       }
 
       const stepInfo: any = { name: name || key, script: script || key, args, key, dispatch_type };
+      if (fnId) stepInfo.fn_id = fnId;
       if (options) {
         if (options.timeout !== undefined) stepInfo.timeout = options.timeout;
         if (options.tag !== undefined) stepInfo.tag = options.tag;
@@ -2278,6 +2286,20 @@ export async function step<T>(
   return jsonRoundTrip(await fn());
 }
 
+// A stable identity for a task's code, what its cached result is keyed on: a
+// name is shared by any two tasks called the same, and a step key by any two
+// tasks called at the same position, so neither can tell them apart. The name
+// stays in: every bound function stringifies to the same native stub, and only
+// its name (`bound orders`) tells two of them apart.
+function fnFingerprint(fn: Function): string {
+  const src = JSON.stringify([fn.name, fn.toString()]);
+  let h = 0xcbf29ce484222325n;
+  for (let i = 0; i < src.length; i++) {
+    h = ((h ^ BigInt(src.charCodeAt(i))) * 0x100000001b3n) & 0xffffffffffffffffn;
+  }
+  return h.toString(16);
+}
+
 /**
  * Wrap an async function as a workflow task.
  *
@@ -2315,6 +2337,7 @@ export function task<T extends (...args: any[]) => Promise<any>>(
   assertUsableRetry(taskOptions?.retry);
 
   const taskName = fn.name || taskPath || "";
+  const fnId = fnFingerprint(fn);
 
   // NOT async — in workflow context we return the thenable directly so that
   // unawaited task calls leave the step in ctx.pending (for _flushPending).
@@ -2335,7 +2358,7 @@ export function task<T extends (...args: any[]) => Promise<any>>(
           kwargs[`arg${i}`] = args[i];
         }
       }
-      const stepResult = ctx._nextStep(taskName, script, kwargs, "inline", taskOptions);
+      const stepResult = ctx._nextStep(taskName, script, kwargs, "inline", taskOptions, fnId);
       // If this step should execute directly (child job mode), run the inner function
       // and throw StepSuspend with mode "step_complete" to signal that we're done
       if ((stepResult as any)?._execute_directly) {

typescript-client/tests/workflow_task_identity.test.ts

@@ -0,0 +1,111 @@
+/**
+ * The fingerprint a dispatched task carries, against the real client.
+ *
+ * Run with: bun test typescript-client/tests/workflow_task_identity.test.ts
+ *
+ * The worker keys a task child's cached result on this `fn_id`, so two tasks
+ * that share a step key (anonymous tasks in exclusive branches) or whose source
+ * reads the same (bound functions) must not share one. Imports client.ts itself,
+ * with the two generated modules stubbed so the import works without ./build.sh.
+ */
+import { expect, test, describe, mock, beforeAll, afterAll } from "bun:test";
+
+mock.module("../services.gen", () => ({
+  ResourceService: {},
+  VariableService: {},
+  JobService: {},
+  HelpersService: {},
+  AppService: {},
+  MetricsService: {},
+  OidcService: {},
+  UserService: {},
+  KafkaTriggerService: {},
+}));
+mock.module("../core/OpenAPI", () => ({
+  OpenAPI: { BASE: "http://localhost:8000/api", TOKEN: "tok" },
+}));
+
+const { WorkflowCtx, task, setWorkflowCtx, StepSuspend } = await import("../client.ts");
+import { isSuspendSignal } from "../wacError";
+
+/** The steps a first round of `body` dispatches. */
+async function dispatched(body: () => Promise<any>): Promise<any[]> {
+  const ctx = new WorkflowCtx({ completed_steps: {} } as any);
+  setWorkflowCtx(ctx);
+  try {
+    await body();
+  } catch (e: any) {
+    if (isSuspendSignal(e, StepSuspend)) return e.dispatchInfo.steps;
+    throw e;
+  } finally {
+    setWorkflowCtx(null);
+  }
+  throw new Error("the body completed without dispatching");
+}
+
+// These assert the suspend a worker acts on, the legacy inline path; the v2 fast
+// path is on by default, so pin it off rather than depend on `WM_JOB_ID` being
+// absent.
+const priorFastPath = process.env.WM_WAC_INLINE_FAST_PATH;
+beforeAll(() => {
+  process.env.WM_WAC_INLINE_FAST_PATH = "0";
+});
+afterAll(() => {
+  if (priorFastPath === undefined) delete process.env.WM_WAC_INLINE_FAST_PATH;
+  else process.env.WM_WAC_INLINE_FAST_PATH = priorFastPath;
+});
+
+describe("task fingerprint", () => {
+  test("two anonymous tasks dispatched at the same position carry different fingerprints", async () => {
+    const tasks = {
+      a: task(async () => "A", { cache_ttl: 60 }),
+      b: task(async () => "B", { cache_ttl: 60 }),
+    };
+    const [a] = await dispatched(() => tasks.a() as Promise<any>);
+    const [b] = await dispatched(() => tasks.b() as Promise<any>);
+    expect(a.key).toBe(b.key);
+    expect(typeof a.fn_id).toBe("string");
+    expect(a.fn_id).not.toBe(b.fn_id);
+  });
+
+  test("two bound methods of one name share a fingerprint, and their step keys separate them", async () => {
+    const a = {
+      async read() {
+        return "A";
+      },
+    };
+    const b = {
+      async read() {
+        return "B";
+      },
+    };
+    const steps = await dispatched(async () => {
+      await Promise.all([task(a.read.bind(a))(), task(b.read.bind(b))()]);
+    });
+    expect(steps[0].fn_id).toBe(steps[1].fn_id);
+    expect(steps.map((s) => s.key)).toEqual(["bound read", "bound read_2"]);
+  });
+
+  test("one task carries one fingerprint at every position", async () => {
+    const double = task(async (n: number) => n * 2, { cache_ttl: 60 });
+    const steps = await dispatched(async () => {
+      await Promise.all([double(1), double(2)]);
+    });
+    expect(steps.map((s) => s.key)).toEqual(["step", "step_2"]);
+    expect(steps[0].fn_id).toBe(steps[1].fn_id);
+  });
+
+  test("two bound functions carry different fingerprints", async () => {
+    const api = {
+      async orders() {
+        return "orders";
+      },
+      async users() {
+        return "users";
+      },
+    };
+    const [orders] = await dispatched(() => task(api.orders.bind(api))() as Promise<any>);
+    const [users] = await dispatched(() => task(api.users.bind(api))() as Promise<any>);
+    expect(orders.fn_id).not.toBe(users.fn_id);
+  });
+});