test(mcp): cover failed-owner retry across event loops (#7458)

backend/tests/blocking_io/test_dynamic_context_middleware.py

@@ -401,20 +401,23 @@ async def test_abefore_agent_records_checkpointed_memory_on_timeout() -> None:
 
     with (
         mock.patch.object(mw, "_inject", blocking_inject),
+        mock.patch.object(mw, "_read_failures_are_fatal", return_value=False),
         mock.patch(
             "deerflow.agents.middlewares.dynamic_context_middleware._INJECT_TIMEOUT_SECONDS",
             0.01,
         ),
     ):
-        result = await mw.abefore_agent(state, runtime)
+        try:
+            result = await mw.abefore_agent(state, runtime)
+            assert result is None
+            # The first model call assembles with the frozen checkpoint block: the
+            # recorded identity is the checkpointed content, not the late replacement.
+            _drive_model_call(mw, state["messages"], runtime)
+            recorded_call = journal.record_memory_context.call_args
+        finally:
+            release.set()
+            assert await asyncio.to_thread(finished.wait, 1)
 
-    assert result is None
-    # The first model call assembles with the frozen checkpoint block: the
-    # recorded identity is the checkpointed content, not the late replacement.
-    _drive_model_call(mw, state["messages"], runtime)
-    recorded_call = journal.record_memory_context.call_args
-    release.set()
-    assert await asyncio.to_thread(finished.wait, 1)
     assert started.is_set()
     assert recorded_call == mock.call(
         content_sha256=hashlib.sha256(memory_content.encode("utf-8")).hexdigest(),

backend/tests/test_mcp_oauth.py

@@ -5,12 +5,13 @@ from __future__ import annotations
 import asyncio
 import logging
 import threading
+from concurrent.futures import ThreadPoolExecutor
 from datetime import UTC, datetime, timedelta
 from typing import Any
 
 import pytest
 
-from deerflow.config.extensions_config import ExtensionsConfig
+from deerflow.config.extensions_config import ExtensionsConfig, McpOAuthConfig
 from deerflow.mcp.oauth import OAuthTokenManager, build_oauth_tool_interceptor, get_initial_oauth_headers
 
 
@@ -710,6 +711,69 @@ def test_get_authorization_header_cancelled_while_waiting_does_not_leak_lock(mon
     assert len(post_calls) == 1
 
 
+@pytest.mark.parametrize("failure", ["cancel", "error"])
+def test_failed_refresh_owner_allows_cross_loop_waiters_to_retry(monkeypatch, failure):
+    """Pending callers on other loops elect one replacement after owner failure."""
+    owner_entered = threading.Event()
+    release_owner = threading.Event()
+    retry_entered = threading.Event()
+    release_retry = threading.Event()
+    waiters_ready = [threading.Event(), threading.Event()]
+    attempts = 0
+    attempts_guard = threading.Lock()
+    post_calls: list[dict[str, Any]] = []
+
+    class HeldClient(_MockAsyncClient):
+        async def post(self, url, data):
+            nonlocal attempts
+            with attempts_guard:
+                attempts += 1
+                attempt = attempts
+            if attempt == 1:
+                owner_entered.set()
+                assert await asyncio.to_thread(release_owner.wait, 5), "test did not release owner"
+                if failure == "cancel":
+                    asyncio.current_task().cancel()
+                    await asyncio.sleep(0)
+                raise RuntimeError("first refresh failed")
+            retry_entered.set()
+            assert await asyncio.to_thread(release_retry.wait, 5), "test did not release retry"
+            return await super().post(url, data)
+
+    monkeypatch.setattr("httpx.AsyncClient", lambda **kwargs: HeldClient(payload={"access_token": "recovered-token", "expires_in": 3600}, post_calls=post_calls, **kwargs))
+    manager = OAuthTokenManager({"secure-http": McpOAuthConfig(enabled=True, token_url="https://auth.example.com/token", client_id="client-id", client_secret="client-secret")})
+
+    async def call(ready=None):
+        request = asyncio.create_task(manager.get_authorization_header("secure-http"))
+        if ready is not None:
+            # Run after the request has suspended while the original owner is held.
+            asyncio.get_running_loop().call_soon(ready.set)
+        return await asyncio.wait_for(request, 5)
+
+    with ThreadPoolExecutor(max_workers=3) as callers:
+        owner = callers.submit(asyncio.run, call())
+        try:
+            assert owner_entered.wait(5), "owner did not start refresh"
+            waiters = [callers.submit(asyncio.run, call(ready)) for ready in waiters_ready]
+            for ready in waiters_ready:
+                assert ready.wait(5), "waiter did not suspend its authorization request"
+            assert attempts == 1
+            release_owner.set()
+            with pytest.raises(asyncio.CancelledError if failure == "cancel" else RuntimeError):
+                owner.result(timeout=5)
+            assert retry_entered.wait(5), "waiters did not retry refresh"
+            assert attempts == 2
+            release_retry.set()
+            assert [waiter.result(timeout=5) for waiter in waiters] == ["Bearer recovered-token", "Bearer recovered-token"]
+        finally:
+            release_owner.set()
+            release_retry.set()
+
+    assert asyncio.run(manager.get_authorization_header("secure-http")) == "Bearer recovered-token"
+    assert attempts == 2
+    assert len(post_calls) == 1
+
+
 # --- Illegal header values ---------------------------------------------------
 #
 # What the token endpoint returns is not this process's to control. An