diff --git a/backend/packages/harness/deerflow/extensions/notify.py b/backend/packages/harness/deerflow/extensions/notify.py index b0d77d60c9..57785e9a52 100644 --- a/backend/packages/harness/deerflow/extensions/notify.py +++ b/backend/packages/harness/deerflow/extensions/notify.py @@ -117,15 +117,26 @@ async def _notify_each( """Invoke contributors in order, fail-open, within one shared budget.""" loop = asyncio.get_running_loop() deadline = None if timeout is None else loop.time() + timeout + # ``budget_spent`` latches once the shared budget runs out, so the skip + # decision never depends on a second clock read. The loop runs a scheduled + # timeout up to one ``clock_resolution`` *early* (``BaseEventLoop._run_once`` + # compares each handle's ``_when`` against ``now + clock_resolution``), so on + # a coarse clock — Windows' ~15.6 ms against budgets as small as 20 ms — + # ``loop.time()`` can still sit strictly before ``deadline`` at the moment + # the budget has in fact expired. That raced both the log classification and + # the skip of later contributors (#6453). + budget_spent = False for source, contributor in contributors: + timeout_scope: asyncio.Timeout | None = None try: call = invoke(contributor) if deadline is None: await call continue - remaining = deadline - loop.time() + remaining = 0.0 if budget_spent else deadline - loop.time() if remaining <= 0: + budget_spent = True close = getattr(call, "close", None) if callable(close): close() @@ -137,9 +148,15 @@ async def _notify_each( timeout, ) continue - await asyncio.wait_for(call, remaining) + # ``asyncio.timeout`` records whether *it* expired, so the handler + # below never re-reads the clock to tell a spent budget from a + # ``TimeoutError`` the contributor raised on its own. + timeout_scope = asyncio.timeout(remaining) + async with timeout_scope: + await call except TimeoutError: - if deadline is not None and loop.time() >= deadline: + if timeout_scope is not None and timeout_scope.expired(): + budget_spent = True # Budget exhaustion mid-hook is the same expected operational # condition as the skip above, so it stays a warning rather # than a hook failure with an asyncio-internal traceback. diff --git a/backend/tests/test_extension_task_lifecycle.py b/backend/tests/test_extension_task_lifecycle.py index 15b4d3b890..c6f2c86835 100644 --- a/backend/tests/test_extension_task_lifecycle.py +++ b/backend/tests/test_extension_task_lifecycle.py @@ -143,6 +143,66 @@ async def test_budget_exhaustion_mid_hook_logs_a_warning_not_a_traceback(caplog) assert records[0].exc_info is None +@pytest.mark.asyncio +async def test_budget_exhaustion_is_classified_without_a_clock_recheck(monkeypatch, caplog): + # ``BaseEventLoop._run_once`` runs a scheduled timeout up to one + # ``clock_resolution`` *early* (it compares each handle's ``_when`` against + # ``now + clock_resolution``), so on a coarse clock the budget can expire + # while ``loop.time()`` still sits strictly before ``deadline`` — the + # Windows behaviour in #6453. Freezing the clock the handler reads makes + # that race deterministic on every platform; the timeout itself still + # schedules against the real loop, exactly as it does in production. + class _FrozenClockLoop: + """Loop proxy whose ``time()`` never advances.""" + + def __init__(self, loop, frozen_at): + self._loop = loop + self._frozen_at = frozen_at + + def time(self): + return self._frozen_at + + def __getattr__(self, name): + return getattr(self._loop, name) + + class _AsyncioWithFrozenLoop: + """The ``asyncio`` module, but reporting a frozen running loop.""" + + def __init__(self, module, loop): + self._module = module + self._loop = loop + + def get_running_loop(self): + return self._loop + + def __getattr__(self, name): + return getattr(self._module, name) + + class _Hang: + async def on_task_stop(self, app_store, task_store, info, outcome): + await asyncio.sleep(10) + + real_loop = asyncio.get_running_loop() + monkeypatch.setattr( + "deerflow.extensions.notify.asyncio", + _AsyncioWithFrozenLoop(asyncio, _FrozenClockLoop(real_loop, real_loop.time())), + ) + + with caplog.at_level(logging.WARNING, logger="deerflow.extensions.notify"): + await notify_task_stop( + _extensions(_Hang()), + ExtensionData("task-1"), + _info(), + TaskOutcome.COMPLETED, + timeout=0.02, + ) + + records = [record for record in caplog.records if record.name == "deerflow.extensions.notify"] + assert [record.levelno for record in records] == [logging.WARNING] + assert "timed out" in records[0].getMessage() + assert records[0].exc_info is None + + @pytest.mark.asyncio async def test_contributor_timeout_error_before_budget_is_a_hook_failure(caplog): # A TimeoutError the contributor raises on its own is not budget