From bede0b7acf4175f992dbdcafe9357798cd96f6a6 Mon Sep 17 00:00:00 2001 From: Eugene Yurtsev Date: Fri, 10 Apr 2026 17:05:37 -0400 Subject: [PATCH] test(langgraph): use monotonic clock in flaky streaming test (#7477) Switches test_sync_streaming_with_functional_api to time.monotonic() for both emitted task timestamps and observed arrival times so the assertion is based on a monotonic clock instead of wall time. This makes the streaming timing check less flaky on systems where time.time() can jump or lack sufficient precision. Created with [Deep Agents CLI](https://docs.langchain.com/oss/python/deepagents/cli/overview) using gpt-5.4 (provider: openai). --- libs/langgraph/tests/test_pregel.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/libs/langgraph/tests/test_pregel.py b/libs/langgraph/tests/test_pregel.py index 95afe3e6e..ad18f7843 100644 --- a/libs/langgraph/tests/test_pregel.py +++ b/libs/langgraph/tests/test_pregel.py @@ -6271,7 +6271,7 @@ def test_sync_streaming_with_functional_api() -> None: @task() def slow() -> dict: time.sleep(time_delay) # Simulate a delay of 10 ms - return {"tic": time.time()} + return {"tic": time.monotonic()} @entrypoint() def graph(inputs: dict) -> list: @@ -6284,7 +6284,7 @@ def test_sync_streaming_with_functional_api() -> None: for chunk in graph.stream({}): if "slow" not in chunk: # We'll just look at the updates from `slow` continue - arrival_times.append(time.time()) + arrival_times.append(time.monotonic()) assert len(arrival_times) == 2 delta = arrival_times[1] - arrival_times[0]