diff --git a/libs/langgraph/langgraph/pregel/_runner.py b/libs/langgraph/langgraph/pregel/_runner.py index 4c5037091..f5e8832cb 100644 --- a/libs/langgraph/langgraph/pregel/_runner.py +++ b/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, diff --git a/libs/langgraph/tests/test_pregel.py b/libs/langgraph/tests/test_pregel.py index d574e8511..4ce5ca24d 100644 --- a/libs/langgraph/tests/test_pregel.py +++ b/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) == [ diff --git a/libs/langgraph/tests/test_pregel_async.py b/libs/langgraph/tests/test_pregel_async.py index e57c09b3d..d3a655bcf 100644 --- a/libs/langgraph/tests/test_pregel_async.py +++ b/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