fix: key a cached inline task on a fingerprint of its code and its arguments

python-client/wmill/tests/test_workflow.py

@@ -1847,3 +1847,19 @@ 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"]

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,22 @@ class WorkflowCtx:
         })
 
 
+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
+    import marshal
+
+    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()
+
+
 def task(
     _func=None,
     *,
@@ -3328,12 +3344,10 @@ 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. A task is keyed on its step key (its name and
-    call order) and the workflow's input, not on the arguments it is called
-    with, so cache one only when whether it runs, and what it receives, follow
-    from the workflow's input alone. A ``task_script`` target is keyed on the
-    arguments it is called with. 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 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.
 
     Usage::
 
@@ -3364,6 +3378,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)
 
@@ -3389,7 +3404,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 (