fix(extensions): classify a spent notification budget without a clock recheck (#6469)

The extension notify loop told a spent shared budget apart from a
TimeoutError a contributor raised on its own by re-reading loop.time()
after the await and comparing it with the deadline. The event 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 -- that read can
still land strictly before the deadline. Budget exhaustion was then logged as
a phantom hook failure (ERROR plus an asyncio-internal traceback), and the
same lag let a later contributor run on a budget that had already run out.

Classify from asyncio.timeout's own expiry (expired()) instead of a second
clock read, and latch budget_spent once the budget is gone so the skip of
later contributors is deterministic too.

Fixes #6453

Co-authored-by: sxh313 <sxh313@users.noreply.github.com>

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.

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