chore: shallow copy futures (#6755)

libs/langgraph/langgraph/pregel/_runner.py

@@ -565,7 +565,7 @@ def _call(
         if fut := next(
             (
                 f
-                for f, t in futures().items()  # type: ignore[union-attr]
+                for f, t in list(futures().items())  # type: ignore[union-attr]
                 if t is not None and t == next_task.id
             ),
             None,
@@ -708,7 +708,7 @@ async def _acall_impl(
             if fut := next(
                 (
                     f
-                    for f, t in futures().items()  # type: ignore[union-attr]
+                    for f, t in list(futures().items())  # type: ignore[union-attr]
                     if t is not None and t == next_task.id
                 ),
                 None,

libs/langgraph/tests/test_pregel.py

@@ -1264,18 +1264,20 @@ def test_imp_task(
     }
 
     thread1 = {"configurable": {"thread_id": "1"}}
-    assert [*graph.stream([0, 1], thread1, durability=durability)] == [
+    result = [*graph.stream([0, 1], thread1, durability=durability)]
+    # mapper tasks run concurrently so output order is non-deterministic
+    assert sorted(result[:-1], key=lambda d: str(d)) == [
         {"mapper": "00"},
         {"mapper": "11"},
-        {
-            "__interrupt__": (
-                Interrupt(
-                    value="question",
-                    id=AnyStr(),
-                ),
-            )
-        },
     ]
+    assert result[-1] == {
+        "__interrupt__": (
+            Interrupt(
+                value="question",
+                id=AnyStr(),
+            ),
+        )
+    }
     assert mapper_calls == 2
 
     assert graph.invoke(Command(resume="answer"), thread1, durability=durability) == [

libs/langgraph/tests/test_pregel_async.py

@@ -2262,18 +2262,20 @@ async def test_imp_task(
 
     tracer = FakeTracer()
     thread1 = {"configurable": {"thread_id": "1"}, "callbacks": [tracer]}
-    assert [c async for c in graph.astream([0, 1], thread1, durability=durability)] == [
+    result = [c async for c in graph.astream([0, 1], thread1, durability=durability)]
+    # mapper tasks run concurrently so output order is non-deterministic
+    assert sorted(result[:-1], key=lambda d: str(d)) == [
         {"mapper": "00"},
         {"mapper": "11"},
-        {
-            "__interrupt__": (
-                Interrupt(
-                    value="question",
-                    id=AnyStr(),
-                ),
-            )
-        },
     ]
+    assert result[-1] == {
+        "__interrupt__": (
+            Interrupt(
+                value="question",
+                id=AnyStr(),
+            ),
+        )
+    }
     assert mapper_calls == 2
     assert len(tracer.runs) == 1
     assert len(tracer.runs[0].child_runs) == 1