nebius/SWE-rebench-V2-PRs huggingface dataset:
Repository: dask/dask
Pull request #11944 (2025-05-12 10:02:47)

Raise on local executor if tasks are missing dependency
If graphs are provided with keys that are missing dependencies, the local executor is silently getting stuck. The distributed scheduler cancels the computation and logs and raises an appropriate error message. The local one should do the same. This is otherwise infuriating to debug.

xref https://github.com/dask/dask/issues/11927

github-actions[bot]: ## Unit Test Results
_See [test report](https://dask.github.io/dask/test_report.html) for an extended history of previous test failures. This is useful for diagnosing flaky tests._

      9 files  ±0        9 suites  ±0   3h 19m 9s ⏱️ -1s
 18 029 tests +1   16 815 ✅ +1   1 214 💤 ±0  0 ❌ ±0 
161 335 runs  +9  149 217 ✅ +8  12 118 💤 +1  0 ❌ ±0 

Results for commit c68ce459. ± Comparison against base commit 7df5674f.

[test-results]:data:application/gzip;base64,H4sIAAnQIWgC/03PTQrDIBCG4asE1104ZjTay5Q4UZDmpxhdld69GlKb5fsMfDBv5sPsdnbvzK1jew6pxZTjmMK2lgQwWKkcUz2D5qL1Y89EFZUG+cdneFUUgM38GOZivIGLcYunxLwe0wr6Xp7dptEIGBq2adA/u0wffV2mbVlCKsFIaXIoDXivjBTAHaEd/GTLf5ZQKzth70CxzxcxXbjYGAEAAA==


Patch:
diff --git a/dask/local.py b/dask/local.py
index 66b72f5ea..0c276aecb 100644
--- a/dask/local.py
+++ b/dask/local.py
@@ -187,6 +187,12 @@ def start_state_from_dask(dsk, cache=None, sortkey=None, keys=None):
         dependencies[key]
         task = dsk.get(key, None)
         if task is None:
+            if dependents[key] and not cache.get(key, None):
+                raise ValueError(
+                    "Missing dependency {} for dependents {}".format(
+                        key, dependents[key]
+                    )
+                )
             continue
         elif isinstance(task, DataNode):
             cache[key] = task()


Test patch:
diff --git a/dask/tests/test_local.py b/dask/tests/test_local.py
index 684e43949..016443820 100644
--- a/dask/tests/test_local.py
+++ b/dask/tests/test_local.py
@@ -3,7 +3,7 @@ from __future__ import annotations
 import pytest
 
 import dask
-from dask._task_spec import Task
+from dask._task_spec import Task, TaskRef
 from dask.local import finish_task, get_sync, sortkey, start_state_from_dask
 from dask.order import order
 from dask.utils_test import GetFunctionTestMixin, add, inc
@@ -213,3 +213,10 @@ def test_ensure_calculate_only_whats_needed():
     ]
     dsk = {t.key: t for t in tasks}
     assert get_sync(dsk, "y") == 2
+
+
+def test_detect_malformed_graph():
+    dsk = {"y": Task("y", lambda x: x, TaskRef("x"))}
+    with pytest.raises(ValueError, match="Missing dependency"):
+        get_sync(dsk, "y")
+    assert get_sync(dsk, "y", cache={"x": 1}) == 1
Repository: dask/dask
Pull request #11953 (2025-05-16 09:42:33)

Revert "Dont handle tuple in task_spec.parse_input"
Reverts dask/dask#11948

This broke a bunch of things in distributed


Patch:
diff --git a/dask/_task_spec.py b/dask/_task_spec.py
index f751439d3..9fd4f75a7 100644
--- a/dask/_task_spec.py
+++ b/dask/_task_spec.py
@@ -143,13 +143,17 @@ def parse_input(obj: Any) -> object:
         if any(isinstance(v, GraphNode) for v in parsed_dict.values()):
             return Dict(parsed_dict)
 
-    if isinstance(obj, (list, set)):
+    if isinstance(obj, (list, set, tuple)):
         parsed_collection = tuple(parse_input(o) for o in obj)
         if any(isinstance(o, GraphNode) for o in parsed_collection):
             if isinstance(obj, list):
                 return List(*parsed_collection)
             if isinstance(obj, set):
                 return Set(*parsed_collection)
+            if isinstance(obj, tuple):
+                if is_namedtuple_instance(obj):
+                    return _wrap_namedtuple_task(None, obj, parse_input)
+                return Tuple(*parsed_collection)
 
     return obj
 


Test patch:
diff --git a/dask/tests/test_task_spec.py b/dask/tests/test_task_spec.py
index 36074e98e..4986c3efb 100644
--- a/dask/tests/test_task_spec.py
+++ b/dask/tests/test_task_spec.py
@@ -499,6 +499,32 @@ def test_parse_graph_namedtuple_legacy(typ, args, kwargs):
     assert new_dsk["foo"]() == typ(*args, **kwargs)
 
 
+@pytest.mark.parametrize(
+    "typ, args, kwargs",
+    [
+        (PlainNamedTuple, ["some-data"], {}),
+        (NewArgsNamedTuple, ["some", "data", "more"], {}),
+        (NewArgsExNamedTuple, ["some", "data", "more"], {"another": "data"}),
+    ],
+)
+def test_parse_namedtuple(typ, args, kwargs):
+    def func(x):
+        return x
+
+    obj = typ(*args, **kwargs)
+    t = Task("foo", func, parse_input(obj))
+
+    assert t() == obj
+
+    # The other test tuple do weird things to their input
+    if typ is PlainNamedTuple:
+        args = tuple([TaskRef("b")] + list(args)[1:])
+        obj = typ(*args, **kwargs)
+        t = Task("foo", func, parse_input(obj))
+        assert t.dependencies == {"b"}
+        assert t({"b": "foo"}) == typ(*tuple(["foo"] + list(args)[1:]), **kwargs)
+
+
 def test_pickle_literals():
     np = pytest.importorskip("numpy")
     obj = DataNode("foo", np.transpose)