fix: keep the step key in a cached task's identity

backend/tests/wac_task_cache_key.rs

@@ -104,9 +104,15 @@ async fn a_task_child_is_cached_under_its_fingerprint_or_else_its_step_key(
         "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?,
-        "one task called at two positions keeps one identity"
+        "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,

backend/windmill-common/src/wac.rs

@@ -41,9 +41,9 @@ pub struct WacCheckpoint {
     #[serde(default)]
     pub _executing_key: Option<String>,
     /// With `_executing_key`: the SDK's fingerprint of the task's code and the
-    /// arguments the task was called with, what its cached result is keyed on.
-    /// A child whose SDK sent no fingerprint is keyed on its step key and its
-    /// parent's arguments instead (`cached_result_path`).
+    /// 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>,

backend/windmill-worker/src/common.rs

@@ -1575,13 +1575,18 @@ 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 and its own call arguments. Without one it is keyed
-    // on its step key and the parent's arguments: a step key names a position, not a
-    // task, and only the parent's arguments tell apart tasks a branch puts at one.
+    // 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 { fn_id, args }) => {
+        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)) => {
@@ -1607,7 +1612,7 @@ pub async fn cached_result_path(
 /// 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 { fn_id: String, args: HashMap<String, Box<RawValue>> },
+    Fingerprint { key: String, fn_id: String, args: HashMap<String, Box<RawValue>> },
     StepKey(String),
 }
 
@@ -1633,8 +1638,8 @@ async fn wac_task_identity(
     .fetch_optional(db)
     .await?;
     Ok(match identity {
-        Some((Some(fn_id), _, Some(Json(args)))) => {
-            Some(WacTaskIdentity::Fingerprint { fn_id, args })
+        Some((Some(fn_id), Some(key), Some(Json(args)))) => {
+            Some(WacTaskIdentity::Fingerprint { key, fn_id, args })
         }
         Some((_, Some(key), _)) => Some(WacTaskIdentity::StepKey(key)),
         _ => None,

cli/src/guidance/skills.gen.ts

@@ -4609,10 +4609,11 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]]
 # 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 and the
-# arguments it is called with, so anything a cached task reads from its
-# closure must be passed in as an argument. It has no effect on a
-# \`\`task_flow\`\` target, which keeps its flow's own cache policy.
+# 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::
 # 
@@ -6748,10 +6749,11 @@ 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 and the arguments it is
-  *  called with, so anything a cached task reads from its closure must be
-  *  passed in as an argument. It has no effect on a \`taskFlow\` target, which
-  *  keeps its flow's own cache policy. */
+  *  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;
@@ -6944,10 +6946,11 @@ def get_resume_urls(approver: str = None, flow_level: bool = None) -> dict
 # 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 and the
-# arguments it is called with, so anything a cached task reads from its
-# closure must be passed in as an argument. It has no effect on a
-# \`\`task_flow\`\` target, which keeps its flow's own cache policy.
+# 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::
 #

python-client/wmill/tests/test_workflow.py

@@ -1863,3 +1863,28 @@ class TestTaskFingerprint:
         )
         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"]

python-client/wmill/wmill/client.py

@@ -3291,14 +3291,23 @@ def _fn_fingerprint(func) -> str:
     tasks called at the same position, so neither can tell them apart."""
     import hashlib
     import inspect
-    import marshal
 
+    # 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:
-        src = inspect.getsource(func).encode()
-    except (OSError, TypeError):
-        # No source on disk: the whole code object, constants and names included.
-        src = marshal.dumps(func.__code__)
-    return hashlib.sha256(src).hexdigest()
+        parts.append(inspect.getsource(func).encode())
+    except Exception:
+        pass
+    code = getattr(func, "__code__", None)
+    if code is not None:
+        consts = tuple(c for c in code.co_consts if not isinstance(c, type(code)))
+        parts.append(repr((code.co_code, code.co_names, code.co_varnames, consts)).encode())
+    if not parts:
+        parts.append(repr(func).encode())
+    return hashlib.sha256(b"\x1f".join(parts)).hexdigest()
 
 
 def task(
@@ -3344,10 +3353,11 @@ def task(
     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 and the
-    arguments it is called with, so anything a cached task reads from its
-    closure must be passed in as an argument. It has no effect on a
-    ``task_flow`` target, which keeps its flow's own cache policy.
+    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::
 

system_prompts/auto-generated/prompts.ts

@@ -2588,10 +2588,11 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]]
 # 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 and the
-# arguments it is called with, so anything a cached task reads from its
-# closure must be passed in as an argument. It has no effect on a
-# \`\`task_flow\`\` target, which keeps its flow's own cache policy.
+# 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::
 # 
@@ -2744,10 +2745,11 @@ 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 and the arguments it is
-  *  called with, so anything a cached task reads from its closure must be
-  *  passed in as an argument. It has no effect on a \`taskFlow\` target, which
-  *  keeps its flow's own cache policy. */
+  *  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;
@@ -2940,10 +2942,11 @@ def get_resume_urls(approver: str = None, flow_level: bool = None) -> dict
 # 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 and the
-# arguments it is called with, so anything a cached task reads from its
-# closure must be passed in as an argument. It has no effect on a
-# \`\`task_flow\`\` target, which keeps its flow's own cache policy.
+# 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::
 #

system_prompts/auto-generated/script.md

@@ -2729,10 +2729,11 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]]
 # 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 and the
-# arguments it is called with, so anything a cached task reads from its
-# closure must be passed in as an argument. It has no effect on a
-# ``task_flow`` target, which keeps its flow's own cache policy.
+# 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::
 # 

system_prompts/auto-generated/sdks/python.md

@@ -673,10 +673,11 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]]
 # 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 and the
-# arguments it is called with, so anything a cached task reads from its
-# closure must be passed in as an argument. It has no effect on a
-# ``task_flow`` target, which keeps its flow's own cache policy.
+# 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::
 # 

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

@@ -59,10 +59,11 @@ def get_resume_urls(approver: str = None, flow_level: bool = None) -> dict
 # 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 and the
-# arguments it is called with, so anything a cached task reads from its
-# closure must be passed in as an argument. It has no effect on a
-# ``task_flow`` target, which keeps its flow's own cache policy.
+# 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::
 #

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

@@ -35,10 +35,11 @@ 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 and the arguments it is
-  *  called with, so anything a cached task reads from its closure must be
-  *  passed in as an argument. It has no effect on a `taskFlow` target, which
-  *  keeps its flow's own cache policy. */
+  *  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

@@ -858,10 +858,11 @@ def parse_sql_client_name(name: str) -> tuple[str, Optional[str]]
 # 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 and the
-# arguments it is called with, so anything a cached task reads from its
-# closure must be passed in as an argument. It has no effect on a
-# ``task_flow`` target, which keeps its flow's own cache policy.
+# 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::
 # 

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

@@ -278,10 +278,11 @@ 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 and the arguments it is
-  *  called with, so anything a cached task reads from its closure must be
-  *  passed in as an argument. It has no effect on a `taskFlow` target, which
-  *  keeps its flow's own cache policy. */
+  *  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;
@@ -474,10 +475,11 @@ def get_resume_urls(approver: str = None, flow_level: bool = None) -> dict
 # 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 and the
-# arguments it is called with, so anything a cached task reads from its
-# closure must be passed in as an argument. It has no effect on a
-# ``task_flow`` target, which keeps its flow's own cache policy.
+# 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::
 #

typescript-client/client.ts

@@ -1713,10 +1713,11 @@ 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 and the arguments it is
-   *  called with, so anything a cached task reads from its closure must be
-   *  passed in as an argument. It has no effect on a `taskFlow` target, which
-   *  keeps its flow's own cache policy. */
+   *  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;

typescript-client/tests/workflow_task_identity.test.ts

@@ -68,7 +68,25 @@ describe("task fingerprint", () => {
     expect(a.fn_id).not.toBe(b.fn_id);
   });
 
-  test("one task keeps its fingerprint across positions", async () => {
+  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)]);