Python: fix(redis): close an owned RedisContextProvider index on exit

python/packages/redis/agent_framework_redis/_context_provider.py

@@ -115,6 +115,9 @@ def __init__(
         self._token_escaper: TokenEscaper = TokenEscaper()
         self._index_initialized: bool = False
         self._schema_dict: dict[str, Any] | None = None
+        # Ownership rule matches RedisHistoryProvider: only the index we built
+        # ourselves is ours to close; an injected index stays the caller's.
+        self._owns_index: bool = redis_index is None
         index = redis_index or AsyncSearchIndex.from_dict(  # pyright: ignore[reportUnknownMemberType]
             self.schema_dict, redis_url=self.redis_url, validate_on_load=True
         )
@@ -440,8 +443,18 @@ async def __aenter__(self) -> Self:
         """Async context manager entry."""
         return self
 
+    async def aclose(self) -> None:
+        """Close the underlying Redis connection when this provider built the index.
+
+        An index injected via ``redis_index`` stays under the caller's control,
+        matching the ownership rule of ``RedisHistoryProvider``.
+        """
+        if self._owns_index and self.redis_index is not None:
+            await self.redis_index.disconnect()
+
     async def __aexit__(self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: Any) -> None:
-        """Async context manager exit."""
+        """Async context manager exit; closes an owned index's connections."""
+        await self.aclose()
 
 
 __all__ = ["RedisContextProvider"]

python/packages/redis/tests/test_providers.py

@@ -490,6 +490,32 @@ async def test_borrowed_client_is_not_closed(self):
         await borrowed.aclose()
 
 
+class TestRedisContextProviderClose:
+    """The context provider closes only the index it built itself, matching
+    the history provider's ownership rule."""
+
+    async def test_owned_index_is_disconnected(self, patch_index_from_dict: MagicMock, mock_index: AsyncMock):
+        provider = RedisContextProvider(source_id="ctx", user_id="u1")
+
+        await provider.aclose()
+
+        mock_index.disconnect.assert_awaited_once()
+
+    async def test_borrowed_index_is_not_disconnected(self):
+        borrowed = AsyncMock()
+        provider = RedisContextProvider(source_id="ctx", redis_index=borrowed)
+
+        await provider.aclose()
+
+        borrowed.disconnect.assert_not_awaited()
+
+    async def test_aexit_closes_owned_index(self, patch_index_from_dict: MagicMock, mock_index: AsyncMock):
+        async with RedisContextProvider(source_id="ctx", user_id="u1"):
+            pass
+
+        mock_index.disconnect.assert_awaited_once()
+
+
 class TestRedisHistoryProviderRedisKey:
     def test_scoped_key_is_deterministic_and_encodes_opaque_components(self, mock_redis_client: MagicMock):
         with patch("agent_framework_redis._history_provider.redis.from_url") as mock_from_url: