From 7c77edaeff9b984f97fd6a5b112d6e26b512c6c7 Mon Sep 17 00:00:00 2001
From: Ankita Luthra <lankita@google.com>
Date: Fri, 31 Jul 2026 07:24:00 +0000
Subject: [PATCH 01/10] makes prefetcher will none cache default

---
 docs/source/prefetcher.rst     | 20 ++++++++++++--------
 gcsfs/core.py                  | 34 +++++++++++++++++++---------------
 gcsfs/tests/test_core.py       | 16 +++++++++++++---
 gcsfs/tests/test_zonal_file.py |  6 +++---
 gcsfs/zb_hns_utils.py          |  2 +-
 5 files changed, 48 insertions(+), 30 deletions(-)

diff --git a/docs/source/prefetcher.rst b/docs/source/prefetcher.rst
index a641d8102..cbf5505c9 100644
--- a/docs/source/prefetcher.rst
+++ b/docs/source/prefetcher.rst
@@ -2,8 +2,7 @@
 GCSFS Adaptive Concurrent Prefetching: Architecture & Usage Guide
 =================================================================
 
-Prefetcher is not enabled by default. To enable, you need to pass the environment variable
-`USE_EXPERIMENTAL_ADAPTIVE_PREFETCHING='true'` and `DEFAULT_GCSFS_CONCURRENCY`=4. As currently written, this implementation is
+Prefetcher is enabled by default with `DEFAULT_GCSFS_CONCURRENCY=4` and `cache_type="none"`. To disable, you can pass the environment variable `USE_EXPERIMENTAL_ADAPTIVE_PREFETCHING='false'` or pass `use_experimental_adaptive_prefetching=False` when opening a file. As currently written, this implementation is
 separate from the fsspec-style caching layer, but the intent is to eventually make this available to all
 asynchronous filesystems using the standard `cache_type=` argument. How it interacts with the
 existing cache types ("readahead", "first", etc.) remains to be decided, and in the meantime, use at your own risk.
@@ -71,17 +70,22 @@ Interaction with GCSFile
 
 The prefetcher is integrated into the ``GCSFile`` and replaces the standard sequential fetching mechanism when enabled.
 
-Enabling the Feature
---------------------
+Feature Configuration & Disabling
+---------------------------------
 
-To use this architecture, set the following environment variables:
+Adaptive prefetching is enabled by default with ``DEFAULT_GCSFS_CONCURRENCY=4`` and ``cache_type="none"``.
+
+To disable prefetching and revert to the legacy readahead cache, set the environment variable:
 
 .. code-block:: bash
 
-    export DEFAULT_GCSFS_CONCURRENCY=4
-    export USE_EXPERIMENTAL_ADAPTIVE_PREFETCHING='true'
+    export USE_EXPERIMENTAL_ADAPTIVE_PREFETCHING='false'
+
+or pass ``use_experimental_adaptive_prefetching=False`` directly when opening a file:
+
+.. code-block:: python
 
-We recommend setting ``cache_type="none"`` for optimal results. The engine avoids prefetching for random workloads, and other cache types create unnecessary memory copies that degrade performance.
+    gcs.open("bucket/file.txt", "rb", use_experimental_adaptive_prefetching=False)
 
 Under the Hood Lifecycle
 ------------------------
diff --git a/gcsfs/core.py b/gcsfs/core.py
index 1bae97e60..e2686a130 100644
--- a/gcsfs/core.py
+++ b/gcsfs/core.py
@@ -2313,7 +2313,7 @@ def __init__(
         mode="rb",
         block_size=DEFAULT_BLOCK_SIZE,
         autocommit=True,
-        cache_type="readahead",
+        cache_type=None,
         cache_options=None,
         acl=None,
         consistency="md5",
@@ -2376,6 +2376,24 @@ def __init__(
             raise OSError("Attempt to open a bucket")
         self.generation = _coalesce_generation(generation, path_generation)
         self.concurrency = kwargs.get("concurrency", DEFAULT_CONCURRENCY)
+        # Ideally, all of these fields should be part of `cache_options`. Because current
+        # `fsspec` caches do not accept arbitrary `*args` and `**kwargs`, passing them
+        # there currently causes instantiation errors. We are holding off on introducing
+        # them as explicit keyword arguments to ensure existing user workloads are not
+        # disrupted. This will be refactored once the upstream `fsspec` changes are merged.
+        if "use_experimental_adaptive_prefetching" in kwargs:
+            use_prefetch_reader = bool(kwargs["use_experimental_adaptive_prefetching"])
+        else:
+            use_prefetch_reader = os.environ.get(
+                "USE_EXPERIMENTAL_ADAPTIVE_PREFETCHING", "true"
+            ).lower() in (
+                "true",
+                "1",
+            )
+
+        if cache_type is None:
+            cache_type = "none" if use_prefetch_reader else "readahead"
+
         super().__init__(
             gcsfs,
             path,
@@ -2394,20 +2412,6 @@ def __init__(
         self.consistency = consistency
         self.checker = get_consistency_checker(consistency)
 
-        # Ideally, all of these fields should be part of `cache_options`. Because current
-        # `fsspec` caches do not accept arbitrary `*args` and `**kwargs`, passing them
-        # there currently causes instantiation errors. We are holding off on introducing
-        # them as explicit keyword arguments to ensure existing user workloads are not
-        # disrupted. This will be refactored once the upstream `fsspec` changes are merged.
-        use_prefetch_reader = kwargs.get(
-            "use_experimental_adaptive_prefetching", False
-        ) or os.environ.get(
-            "USE_EXPERIMENTAL_ADAPTIVE_PREFETCHING", "false"
-        ).lower() in (
-            "true",
-            "1",
-        )
-
         if "r" in mode and use_prefetch_reader:
             max_prefetch_size = kwargs.get("max_prefetch_size", MAX_PREFETCH_SIZE)
             from .prefetcher import BackgroundPrefetcher
diff --git a/gcsfs/tests/test_core.py b/gcsfs/tests/test_core.py
index ae42c8d66..1a6caa093 100644
--- a/gcsfs/tests/test_core.py
+++ b/gcsfs/tests/test_core.py
@@ -1598,7 +1598,7 @@ def test_errors(gcs):
 
 def test_read_small(gcs):
     fn = TEST_BUCKET + "/2014-01-01.csv"
-    with gcs.open(fn, "rb", block_size=10) as f:
+    with gcs.open(fn, "rb", block_size=10, cache_type="readahead") as f:
         out = []
         while True:
             data = f.read(3)
@@ -1725,7 +1725,7 @@ def test_readline_from_cache(gcs):
     with gcs.open(a, "wb") as f:
         f.write(data)
 
-    with gcs.open(a, "rb") as f:
+    with gcs.open(a, "rb", cache_type="readahead") as f:
         result = f.readline()
         assert result == b"a,b\n"
         assert f.loc == 4
@@ -2814,12 +2814,22 @@ async def mock_fail_seq(path, start, end, **kwargs):
 
 
 def test_gcsfile_prefetch_disabled_fallback(gcs):
-    """Verify that omitting the flag entirely skips the prefetcher initialization."""
+    """Verify that disabling prefetcher defaults to readahead cache unless cache_type="none" is explicitly specified."""
     fn = f"{TEST_BUCKET}/no_prefetch.txt"
     gcs.pipe(fn, b"HelloWorld")
 
+    # When prefetcher disabled and no cache_type specified, defaults to readahead
     with gcs.open(fn, "rb", use_experimental_adaptive_prefetching=False) as f:
         assert getattr(f, "_prefetch_engine", None) is None
+        assert f.cache_type == "readahead"
+        assert f.read() == b"HelloWorld"
+
+    # When prefetcher disabled and cache_type="none" explicitly specified, remains "none"
+    with gcs.open(
+        fn, "rb", cache_type="none", use_experimental_adaptive_prefetching=False
+    ) as f:
+        assert getattr(f, "_prefetch_engine", None) is None
+        assert f.cache_type == "none"
         assert f.read() == b"HelloWorld"
 
 
diff --git a/gcsfs/tests/test_zonal_file.py b/gcsfs/tests/test_zonal_file.py
index c9d7962fc..31a830fde 100644
--- a/gcsfs/tests/test_zonal_file.py
+++ b/gcsfs/tests/test_zonal_file.py
@@ -635,7 +635,7 @@ def fake_sync(loop, func, *args, **kwargs):
         assert result == [b"split_data"]
         mock_gcsfs._fetch_range_split.assert_awaited_once_with(
             zf.path,
-            concurrency=1,
+            concurrency=4,
             start=10,
             chunk_lengths=[5],
             size=zf.size,
@@ -711,7 +711,7 @@ def test_zonal_file_pool_size_initialization(mock_sync, mock_gcsfs):
         mode="rb",
         use_experimental_adaptive_prefetching=True,
     )
-    assert zf2.pool_size == 1
+    assert zf2.pool_size == 4
     assert zf2._prefetch_engine is not None
     zf2.close()
 
@@ -721,7 +721,7 @@ def test_zonal_file_pool_size_initialization(mock_sync, mock_gcsfs):
         mode="rb",
         use_experimental_adaptive_prefetching=False,
     )
-    assert zf3.pool_size == 1
+    assert zf3.pool_size == 4
     assert zf3._prefetch_engine is None
     zf3.close()
 
diff --git a/gcsfs/zb_hns_utils.py b/gcsfs/zb_hns_utils.py
index 3ced93a51..88c95a767 100644
--- a/gcsfs/zb_hns_utils.py
+++ b/gcsfs/zb_hns_utils.py
@@ -19,7 +19,7 @@
 )
 
 MRD_MAX_RANGES = 1000  # MRD supports up to 1000 ranges per request
-DEFAULT_CONCURRENCY = int(os.environ.get("DEFAULT_GCSFS_CONCURRENCY", "1"))
+DEFAULT_CONCURRENCY = int(os.environ.get("DEFAULT_GCSFS_CONCURRENCY", "4"))
 MAX_PREFETCH_SIZE = 256 * 1024 * 1024
 logger = logging.getLogger("gcsfs")
 

From d83f24d7931cb19464ee343ad8fe8ceccaeb95c5 Mon Sep 17 00:00:00 2001
From: Ankita Luthra <lankita@google.com>
Date: Fri, 31 Jul 2026 07:57:54 +0000
Subject: [PATCH 02/10] updates zonal file to use none cache as default | fixes
 flag check to include string value

---
 gcsfs/core.py                      |  5 ++++-
 gcsfs/tests/test_extended_gcsfs.py |  4 ++--
 gcsfs/zb_hns_utils.py              |  5 ++++-
 gcsfs/zonal_file.py                | 15 ++++++++++++++-
 4 files changed, 24 insertions(+), 5 deletions(-)

diff --git a/gcsfs/core.py b/gcsfs/core.py
index e2686a130..7b7cfb4bc 100644
--- a/gcsfs/core.py
+++ b/gcsfs/core.py
@@ -2382,7 +2382,10 @@ def __init__(
         # them as explicit keyword arguments to ensure existing user workloads are not
         # disrupted. This will be refactored once the upstream `fsspec` changes are merged.
         if "use_experimental_adaptive_prefetching" in kwargs:
-            use_prefetch_reader = bool(kwargs["use_experimental_adaptive_prefetching"])
+            val = kwargs["use_experimental_adaptive_prefetching"]
+            use_prefetch_reader = (
+                val.lower() in ("true", "1") if isinstance(val, str) else bool(val)
+            )
         else:
             use_prefetch_reader = os.environ.get(
                 "USE_EXPERIMENTAL_ADAPTIVE_PREFETCHING", "true"
diff --git a/gcsfs/tests/test_extended_gcsfs.py b/gcsfs/tests/test_extended_gcsfs.py
index 52b7d3760..deaf2e3ab 100644
--- a/gcsfs/tests/test_extended_gcsfs.py
+++ b/gcsfs/tests/test_extended_gcsfs.py
@@ -985,7 +985,7 @@ def test_get_list_from_zonal_bucket(extended_gcsfs):
             with open(l2, "rb") as f:
                 assert f.read() == files[file2]
 
-    assert mock_create_mrd.call_count == 2
+    assert mock_create_mrd.call_count == 4
 
 
 def test_get_directory_from_zonal_bucket(extended_gcsfs):
@@ -1033,7 +1033,7 @@ def test_get_directory_from_zonal_bucket(extended_gcsfs):
             with open(os.path.join(local_dir, "accounts.2.json"), "rb") as f:
                 assert f.read() == files[file2]
 
-    assert mock_create_mrd.call_count == 2
+    assert mock_create_mrd.call_count == 4
 
 
 @pytest.mark.asyncio
diff --git a/gcsfs/zb_hns_utils.py b/gcsfs/zb_hns_utils.py
index 88c95a767..a884e2b6a 100644
--- a/gcsfs/zb_hns_utils.py
+++ b/gcsfs/zb_hns_utils.py
@@ -19,7 +19,10 @@
 )
 
 MRD_MAX_RANGES = 1000  # MRD supports up to 1000 ranges per request
-DEFAULT_CONCURRENCY = int(os.environ.get("DEFAULT_GCSFS_CONCURRENCY", "4"))
+try:
+    DEFAULT_CONCURRENCY = int(os.environ.get("DEFAULT_GCSFS_CONCURRENCY", "4"))
+except ValueError:
+    DEFAULT_CONCURRENCY = 4
 MAX_PREFETCH_SIZE = 256 * 1024 * 1024
 logger = logging.getLogger("gcsfs")
 
diff --git a/gcsfs/zonal_file.py b/gcsfs/zonal_file.py
index 908f82b35..9e3d58054 100644
--- a/gcsfs/zonal_file.py
+++ b/gcsfs/zonal_file.py
@@ -1,4 +1,5 @@
 import logging
+import os
 
 from fsspec import asyn
 from google.cloud.storage.asyncio.async_appendable_object_writer import (
@@ -28,7 +29,7 @@ def __init__(
         mode="rb",
         block_size=DEFAULT_BLOCK_SIZE,
         autocommit=True,
-        cache_type="readahead_chunked",
+        cache_type=None,
         cache_options=None,
         acl=None,
         consistency="md5",
@@ -94,6 +95,18 @@ def __init__(
                 "Only read, write and append operations are currently supported for Zonal buckets."
             )
 
+        if cache_type is None:
+            val = kwargs.get("use_experimental_adaptive_prefetching")
+            if val is not None:
+                use_prefetch = (
+                    val.lower() in ("true", "1") if isinstance(val, str) else bool(val)
+                )
+            else:
+                use_prefetch = os.environ.get(
+                    "USE_EXPERIMENTAL_ADAPTIVE_PREFETCHING", "true"
+                ).lower() in ("true", "1")
+            cache_type = "none" if use_prefetch else "readahead_chunked"
+
         super().__init__(
             gcsfs,
             path,

From fe7c2b525e73aa2d135fec6d0a0c0b18310e8d43 Mon Sep 17 00:00:00 2001
From: Ankita Luthra <lankita@google.com>
Date: Fri, 31 Jul 2026 10:11:45 +0000
Subject: [PATCH 03/10] fixes tests

---
 gcsfs/prefetcher.py                | 14 +++++++++-
 gcsfs/tests/test_extended_gcsfs.py | 25 ++++++++---------
 gcsfs/tests/test_zonal_file.py     | 45 ++++++++++++++++++++++++++++++
 3 files changed, 70 insertions(+), 14 deletions(-)

diff --git a/gcsfs/prefetcher.py b/gcsfs/prefetcher.py
index 9f3b52b9e..f0e551bae 100644
--- a/gcsfs/prefetcher.py
+++ b/gcsfs/prefetcher.py
@@ -872,4 +872,16 @@ async def aclose(self):
 
     def close(self):
         """Safely shuts down the prefetcher from a synchronous context."""
-        fsspec.asyn.sync(self.loop, self._async_close)
+        try:
+            loop = asyncio.get_running_loop()
+        except RuntimeError:
+            loop = None
+
+        if loop and loop.is_running():
+            loop.create_task(self._async_close())
+        else:
+            try:
+                fsspec.asyn.sync(self.loop, self._async_close)
+            except NotImplementedError:
+                if self.loop and self.loop.is_running():
+                    self.loop.create_task(self._async_close())
diff --git a/gcsfs/tests/test_extended_gcsfs.py b/gcsfs/tests/test_extended_gcsfs.py
index deaf2e3ab..49a0aa788 100644
--- a/gcsfs/tests/test_extended_gcsfs.py
+++ b/gcsfs/tests/test_extended_gcsfs.py
@@ -161,7 +161,9 @@ def test_read_small_zb(extended_gcsfs, gcs_bucket_mocks):
     with gcs_bucket_mocks(
         csv_data, bucket_type_val=BucketType.ZONAL_HIERARCHICAL
     ) as mocks:
-        with extended_gcsfs.open(csv_file_path, "rb", block_size=10) as f:
+        with extended_gcsfs.open(
+            csv_file_path, "rb", block_size=10, cache_type="readahead_chunked"
+        ) as f:
             out = []
             i = 1
             while True:
@@ -194,7 +196,7 @@ def test_readline_zb(extended_gcsfs, gcs_bucket_mocks):
 def test_readline_from_cache_zb(extended_gcsfs, gcs_bucket_mocks):
     data = text_files["zonal/test/a"]
     with gcs_bucket_mocks(data, bucket_type_val=BucketType.ZONAL_HIERARCHICAL):
-        with extended_gcsfs.open(a, "rb") as f:
+        with extended_gcsfs.open(a, "rb", cache_type="readahead_chunked") as f:
             result = f.readline()
             assert result == b"a,b\n"
             assert f.loc == 4
@@ -376,10 +378,15 @@ def test_multithreaded_read_overlapping_ranges_zb(
             assert mocks["pool"].close.call_count == len(read_tasks)
 
 
-def test_default_cache_is_readahead_chunked(extended_gcsfs, gcs_bucket_mocks):
+def test_default_cache_is_none_with_prefetcher(extended_gcsfs, gcs_bucket_mocks):
     data = text_files["zonal/test/b"]
     with gcs_bucket_mocks(data, bucket_type_val=BucketType.ZONAL_HIERARCHICAL):
         with extended_gcsfs.open(b, "rb") as f:
+            assert isinstance(f.cache, caching.BaseCache)
+            assert f._prefetch_engine is not None
+        with extended_gcsfs.open(
+            b, "rb", use_experimental_adaptive_prefetching=False
+        ) as f:
             assert isinstance(f.cache, caching.ReadAheadChunked)
 
 
@@ -932,7 +939,7 @@ def test_get_file_from_zonal_bucket(extended_gcsfs, gcs_bucket_mocks):
                 assert f.read() == json_data
         if mocks:
             mocks["downloader"].download_ranges.assert_awaited()
-            mocks["downloader"].close.assert_awaited_once()
+            mocks["downloader"].close.assert_awaited()
 
 
 async def create_mrd_side_effect(client, bucket, object_name, generation):
@@ -1139,15 +1146,7 @@ def test_read_block_zb(extended_gcsfs, gcs_bucket_mocks, subtests):
                             # for delimiters. We just assert that it requested ranges.
                             assert len(actual_ranges) >= 1
                         else:
-                            req_end = offset + length
-                            if req_end >= file_size:
-                                expected_chunks = 1
-                            else:
-                                expected_chunks = 2
-
-                            assert (
-                                len(actual_ranges) == expected_chunks
-                            ), f"Expected {expected_chunks} chunks (Request + Readahead), got {len(actual_ranges)}"
+                            assert len(actual_ranges) >= 1
                             actual_offsets = sorted(
                                 range_[0] for range_ in actual_ranges
                             )
diff --git a/gcsfs/tests/test_zonal_file.py b/gcsfs/tests/test_zonal_file.py
index 31a830fde..6499cbb9f 100644
--- a/gcsfs/tests/test_zonal_file.py
+++ b/gcsfs/tests/test_zonal_file.py
@@ -726,6 +726,51 @@ def test_zonal_file_pool_size_initialization(mock_sync, mock_gcsfs):
     zf3.close()
 
 
+@mock.patch("gcsfs.zonal_file.asyn.sync")
+def test_zonal_file_cache_type_default_resolution(mock_sync, mock_gcsfs):
+    """Tests dynamic cache_type resolution for ZonalFile."""
+    # 1. Default prefetcher enabled -> cache_type="none"
+    zf_default = ZonalFile(
+        gcsfs=mock_gcsfs, path="gs://test-bucket/test-key", mode="rb"
+    )
+    assert zf_default.cache_type == "none"
+    assert zf_default._prefetch_engine is not None
+    zf_default.close()
+
+    # 2. Prefetcher disabled (opt-out) -> cache_type="readahead_chunked"
+    zf_no_prefetch = ZonalFile(
+        gcsfs=mock_gcsfs,
+        path="gs://test-bucket/test-key",
+        mode="rb",
+        use_experimental_adaptive_prefetching=False,
+    )
+    assert zf_no_prefetch.cache_type == "readahead_chunked"
+    assert zf_no_prefetch._prefetch_engine is None
+    zf_no_prefetch.close()
+
+    # 3. Explicit cache_type="none" with prefetcher disabled -> cache_type="none" and no prefetcher
+    zf_explicit_none = ZonalFile(
+        gcsfs=mock_gcsfs,
+        path="gs://test-bucket/test-key",
+        mode="rb",
+        cache_type="none",
+        use_experimental_adaptive_prefetching=False,
+    )
+    assert zf_explicit_none.cache_type == "none"
+    assert zf_explicit_none._prefetch_engine is None
+    zf_explicit_none.close()
+
+    # 4. Explicit cache_type="bytes" -> cache_type="bytes"
+    zf_bytes = ZonalFile(
+        gcsfs=mock_gcsfs,
+        path="gs://test-bucket/test-key",
+        mode="rb",
+        cache_type="bytes",
+    )
+    assert zf_bytes.cache_type == "bytes"
+    zf_bytes.close()
+
+
 @mock.patch("gcsfs.zonal_file.asyn.sync")
 def test_zonal_file_fetch_range_mutually_exclusive(mock_sync, mock_gcsfs):
     """Tests that providing both end and chunk_lengths raises a ValueError."""

From d2f33f23090876f2c62e687fb60ceb2f67de3a6d Mon Sep 17 00:00:00 2001
From: Ankita Luthra <lankita@google.com>
Date: Fri, 31 Jul 2026 12:03:57 +0000
Subject: [PATCH 04/10] fixes lint

---
 gcsfs/tests/test_extended_gcsfs.py | 4 ----
 1 file changed, 4 deletions(-)

diff --git a/gcsfs/tests/test_extended_gcsfs.py b/gcsfs/tests/test_extended_gcsfs.py
index 49a0aa788..336b25117 100644
--- a/gcsfs/tests/test_extended_gcsfs.py
+++ b/gcsfs/tests/test_extended_gcsfs.py
@@ -1109,10 +1109,6 @@ async def mock_is_zonal(bucket):
 
 
 def test_read_block_zb(extended_gcsfs, gcs_bucket_mocks, subtests):
-    file_size = len(
-        json_data
-    )  # We need the file size to predict if readahead will trigger
-
     for param in read_block_params:
         with subtests.test(id=param.id):
             offset, length, delimiter, expected_data = param.values

From 0b78caa161dc12ac3f9dfab251668bd566cfcdd8 Mon Sep 17 00:00:00 2001
From: Ankita Luthra <lankita@google.com>
Date: Wed, 5 Aug 2026 13:34:04 +0000
Subject: [PATCH 05/10] Makes prefetcher default true when cache is not set by
 the user

---
 gcsfs/core.py                      | 78 ++++++++++++++++--------------
 gcsfs/tests/test_core.py           | 30 +++++++++---
 gcsfs/tests/test_extended_gcsfs.py |  9 ++++
 gcsfs/tests/test_zonal_file.py     | 21 ++++++--
 gcsfs/zonal_file.py                | 13 -----
 5 files changed, 92 insertions(+), 59 deletions(-)

diff --git a/gcsfs/core.py b/gcsfs/core.py
index 7b7cfb4bc..c43200822 100644
--- a/gcsfs/core.py
+++ b/gcsfs/core.py
@@ -2302,7 +2302,32 @@ def sign(self, path, expiration=100, **kwargs):
         )
 
 
-GoogleCredentials.load_tokens()
+def _get_prefetcher_and_cache_value(cache_type, kwargs):
+    """
+    Resolves effective cache_type and whether prefetch reader should be enabled.
+
+    Rules:
+    - Prefetcher is only used when cache_type is not explicitly set by the user (cache_type is None).
+    - If user sets any cache_type (cache_type is not None), prefetcher is always disabled.
+    - By default when cache_type is not set, cache_type is "none".
+    """
+    if cache_type is not None:
+        use_prefetch_reader = False
+    else:
+        cache_type = "none"
+        if "use_experimental_adaptive_prefetching" in kwargs:
+            val = kwargs["use_experimental_adaptive_prefetching"]
+            use_prefetch_reader = (
+                val.lower() in ("true", "1") if isinstance(val, str) else bool(val)
+            )
+        else:
+            use_prefetch_reader = os.environ.get(
+                "USE_EXPERIMENTAL_ADAPTIVE_PREFETCHING", "true"
+            ).lower() in (
+                "true",
+                "1",
+            )
+    return cache_type, use_prefetch_reader
 
 
 class GCSFile(fsspec.spec.AbstractBufferedFile):
@@ -2376,26 +2401,9 @@ def __init__(
             raise OSError("Attempt to open a bucket")
         self.generation = _coalesce_generation(generation, path_generation)
         self.concurrency = kwargs.get("concurrency", DEFAULT_CONCURRENCY)
-        # Ideally, all of these fields should be part of `cache_options`. Because current
-        # `fsspec` caches do not accept arbitrary `*args` and `**kwargs`, passing them
-        # there currently causes instantiation errors. We are holding off on introducing
-        # them as explicit keyword arguments to ensure existing user workloads are not
-        # disrupted. This will be refactored once the upstream `fsspec` changes are merged.
-        if "use_experimental_adaptive_prefetching" in kwargs:
-            val = kwargs["use_experimental_adaptive_prefetching"]
-            use_prefetch_reader = (
-                val.lower() in ("true", "1") if isinstance(val, str) else bool(val)
-            )
-        else:
-            use_prefetch_reader = os.environ.get(
-                "USE_EXPERIMENTAL_ADAPTIVE_PREFETCHING", "true"
-            ).lower() in (
-                "true",
-                "1",
-            )
-
-        if cache_type is None:
-            cache_type = "none" if use_prefetch_reader else "readahead"
+        cache_type, use_prefetch_reader = _get_prefetcher_and_cache_value(
+            cache_type, kwargs
+        )
 
         super().__init__(
             gcsfs,
@@ -2415,20 +2423,6 @@ def __init__(
         self.consistency = consistency
         self.checker = get_consistency_checker(consistency)
 
-        if "r" in mode and use_prefetch_reader:
-            max_prefetch_size = kwargs.get("max_prefetch_size", MAX_PREFETCH_SIZE)
-            from .prefetcher import BackgroundPrefetcher
-
-            self._prefetch_engine = BackgroundPrefetcher(
-                self._async_fetch_range,
-                self.size,
-                max_prefetch_size=max_prefetch_size,
-                concurrency=self.concurrency,
-                loop=self.gcsfs.loop,
-            )
-        else:
-            self._prefetch_engine = None
-
         # _supports_append is an internal argument not meant to be used directly.
         # If True, allows opening file in append mode. This is generally not supported
         # by GCS, but may be supported by subclasses (e.g. ZonalFile). This flag should
@@ -2460,6 +2454,20 @@ def __init__(
                 self.blocksize = GCS_MIN_BLOCK_SIZE
             self.location = None
 
+        if "r" in mode and use_prefetch_reader:
+            max_prefetch_size = kwargs.get("max_prefetch_size", MAX_PREFETCH_SIZE)
+            from .prefetcher import BackgroundPrefetcher
+
+            self._prefetch_engine = BackgroundPrefetcher(
+                self._async_fetch_range,
+                self.size,
+                max_prefetch_size=max_prefetch_size,
+                concurrency=self.concurrency,
+                loop=self.gcsfs.loop,
+            )
+        else:
+            self._prefetch_engine = None
+
     @property
     def details(self):
         if self._details is None:
diff --git a/gcsfs/tests/test_core.py b/gcsfs/tests/test_core.py
index 1a6caa093..b61b5f130 100644
--- a/gcsfs/tests/test_core.py
+++ b/gcsfs/tests/test_core.py
@@ -2813,21 +2813,39 @@ async def mock_fail_seq(path, start, end, **kwargs):
             )
 
 
-def test_gcsfile_prefetch_disabled_fallback(gcs):
-    """Verify that disabling prefetcher defaults to readahead cache unless cache_type="none" is explicitly specified."""
-    fn = f"{TEST_BUCKET}/no_prefetch.txt"
+def test_gcsfile_prefetch_and_cache_type_rules(gcs):
+    """Verify that prefetcher is only used when cache_type is not set by user, and default cache_type is 'none'."""
+    fn = f"{TEST_BUCKET}/cache_rules.txt"
     gcs.pipe(fn, b"HelloWorld")
 
-    # When prefetcher disabled and no cache_type specified, defaults to readahead
+    # 1. Default: cache_type is not set -> prefetcher active, cache_type is "none"
+    with gcs.open(fn, "rb") as f:
+        assert getattr(f, "_prefetch_engine", None) is not None
+        assert f.cache_type == "none"
+        assert f.read() == b"HelloWorld"
+
+    # 2. Prefetcher disabled, no cache_type set -> no prefetcher, cache_type is "none"
     with gcs.open(fn, "rb", use_experimental_adaptive_prefetching=False) as f:
+        assert getattr(f, "_prefetch_engine", None) is None
+        assert f.cache_type == "none"
+        assert f.read() == b"HelloWorld"
+
+    # 3. User sets cache_type="readahead" -> prefetcher NOT used, cache_type is "readahead"
+    with gcs.open(fn, "rb", cache_type="readahead") as f:
         assert getattr(f, "_prefetch_engine", None) is None
         assert f.cache_type == "readahead"
         assert f.read() == b"HelloWorld"
 
-    # When prefetcher disabled and cache_type="none" explicitly specified, remains "none"
+    # 4. User sets cache_type="readahead" even with prefetcher=True -> prefetcher NOT used
     with gcs.open(
-        fn, "rb", cache_type="none", use_experimental_adaptive_prefetching=False
+        fn, "rb", cache_type="readahead", use_experimental_adaptive_prefetching=True
     ) as f:
+        assert getattr(f, "_prefetch_engine", None) is None
+        assert f.cache_type == "readahead"
+        assert f.read() == b"HelloWorld"
+
+    # 5. User explicitly sets cache_type="none" -> prefetcher NOT used
+    with gcs.open(fn, "rb", cache_type="none") as f:
         assert getattr(f, "_prefetch_engine", None) is None
         assert f.cache_type == "none"
         assert f.read() == b"HelloWorld"
diff --git a/gcsfs/tests/test_extended_gcsfs.py b/gcsfs/tests/test_extended_gcsfs.py
index 336b25117..66c0c0dfc 100644
--- a/gcsfs/tests/test_extended_gcsfs.py
+++ b/gcsfs/tests/test_extended_gcsfs.py
@@ -381,13 +381,22 @@ def test_multithreaded_read_overlapping_ranges_zb(
 def test_default_cache_is_none_with_prefetcher(extended_gcsfs, gcs_bucket_mocks):
     data = text_files["zonal/test/b"]
     with gcs_bucket_mocks(data, bucket_type_val=BucketType.ZONAL_HIERARCHICAL):
+        # 1. Default: cache_type not set -> prefetcher enabled, cache is BaseCache ("none")
         with extended_gcsfs.open(b, "rb") as f:
             assert isinstance(f.cache, caching.BaseCache)
             assert f._prefetch_engine is not None
+
+        # 2. Prefetcher disabled, cache_type not set -> no prefetcher, cache is BaseCache ("none")
         with extended_gcsfs.open(
             b, "rb", use_experimental_adaptive_prefetching=False
         ) as f:
+            assert isinstance(f.cache, caching.BaseCache)
+            assert f._prefetch_engine is None
+
+        # 3. Explicit cache_type="readahead_chunked" -> no prefetcher, cache is ReadAheadChunked
+        with extended_gcsfs.open(b, "rb", cache_type="readahead_chunked") as f:
             assert isinstance(f.cache, caching.ReadAheadChunked)
+            assert f._prefetch_engine is None
 
 
 def test_multithreaded_read_chunk_boundary_zb(
diff --git a/gcsfs/tests/test_zonal_file.py b/gcsfs/tests/test_zonal_file.py
index 6499cbb9f..dae11f3a4 100644
--- a/gcsfs/tests/test_zonal_file.py
+++ b/gcsfs/tests/test_zonal_file.py
@@ -737,30 +737,40 @@ def test_zonal_file_cache_type_default_resolution(mock_sync, mock_gcsfs):
     assert zf_default._prefetch_engine is not None
     zf_default.close()
 
-    # 2. Prefetcher disabled (opt-out) -> cache_type="readahead_chunked"
+    # 2. Prefetcher disabled (opt-out), no cache_type set -> cache_type="none", no prefetcher
     zf_no_prefetch = ZonalFile(
         gcsfs=mock_gcsfs,
         path="gs://test-bucket/test-key",
         mode="rb",
         use_experimental_adaptive_prefetching=False,
     )
-    assert zf_no_prefetch.cache_type == "readahead_chunked"
+    assert zf_no_prefetch.cache_type == "none"
     assert zf_no_prefetch._prefetch_engine is None
     zf_no_prefetch.close()
 
-    # 3. Explicit cache_type="none" with prefetcher disabled -> cache_type="none" and no prefetcher
+    # 3. Explicit cache_type="readahead_chunked" -> cache_type="readahead_chunked", no prefetcher
+    zf_readahead = ZonalFile(
+        gcsfs=mock_gcsfs,
+        path="gs://test-bucket/test-key",
+        mode="rb",
+        cache_type="readahead_chunked",
+    )
+    assert zf_readahead.cache_type == "readahead_chunked"
+    assert zf_readahead._prefetch_engine is None
+    zf_readahead.close()
+
+    # 4. Explicit cache_type="none" -> cache_type="none", no prefetcher
     zf_explicit_none = ZonalFile(
         gcsfs=mock_gcsfs,
         path="gs://test-bucket/test-key",
         mode="rb",
         cache_type="none",
-        use_experimental_adaptive_prefetching=False,
     )
     assert zf_explicit_none.cache_type == "none"
     assert zf_explicit_none._prefetch_engine is None
     zf_explicit_none.close()
 
-    # 4. Explicit cache_type="bytes" -> cache_type="bytes"
+    # 5. Explicit cache_type="bytes" -> cache_type="bytes", no prefetcher
     zf_bytes = ZonalFile(
         gcsfs=mock_gcsfs,
         path="gs://test-bucket/test-key",
@@ -768,6 +778,7 @@ def test_zonal_file_cache_type_default_resolution(mock_sync, mock_gcsfs):
         cache_type="bytes",
     )
     assert zf_bytes.cache_type == "bytes"
+    assert zf_bytes._prefetch_engine is None
     zf_bytes.close()
 
 
diff --git a/gcsfs/zonal_file.py b/gcsfs/zonal_file.py
index 9e3d58054..a2d08d527 100644
--- a/gcsfs/zonal_file.py
+++ b/gcsfs/zonal_file.py
@@ -1,5 +1,4 @@
 import logging
-import os
 
 from fsspec import asyn
 from google.cloud.storage.asyncio.async_appendable_object_writer import (
@@ -95,18 +94,6 @@ def __init__(
                 "Only read, write and append operations are currently supported for Zonal buckets."
             )
 
-        if cache_type is None:
-            val = kwargs.get("use_experimental_adaptive_prefetching")
-            if val is not None:
-                use_prefetch = (
-                    val.lower() in ("true", "1") if isinstance(val, str) else bool(val)
-                )
-            else:
-                use_prefetch = os.environ.get(
-                    "USE_EXPERIMENTAL_ADAPTIVE_PREFETCHING", "true"
-                ).lower() in ("true", "1")
-            cache_type = "none" if use_prefetch else "readahead_chunked"
-
         super().__init__(
             gcsfs,
             path,

From 4adf3b7b8ed910138d3ee5fe146be7c7283ef466 Mon Sep 17 00:00:00 2001
From: Ankita Luthra <lankita@google.com>
Date: Wed, 5 Aug 2026 13:55:44 +0000
Subject: [PATCH 06/10] removed unrelated close related changes to keep PR
 clean

---
 gcsfs/prefetcher.py | 14 +-------------
 1 file changed, 1 insertion(+), 13 deletions(-)

diff --git a/gcsfs/prefetcher.py b/gcsfs/prefetcher.py
index f0e551bae..9f3b52b9e 100644
--- a/gcsfs/prefetcher.py
+++ b/gcsfs/prefetcher.py
@@ -872,16 +872,4 @@ async def aclose(self):
 
     def close(self):
         """Safely shuts down the prefetcher from a synchronous context."""
-        try:
-            loop = asyncio.get_running_loop()
-        except RuntimeError:
-            loop = None
-
-        if loop and loop.is_running():
-            loop.create_task(self._async_close())
-        else:
-            try:
-                fsspec.asyn.sync(self.loop, self._async_close)
-            except NotImplementedError:
-                if self.loop and self.loop.is_running():
-                    self.loop.create_task(self._async_close())
+        fsspec.asyn.sync(self.loop, self._async_close)

From a63f73394fa608cce7dac85fb40023194f6032ad Mon Sep 17 00:00:00 2001
From: Ankita Luthra <lankita@google.com>
Date: Wed, 5 Aug 2026 14:09:32 +0000
Subject: [PATCH 07/10] reverts unrelated changes

---
 gcsfs/core.py | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/gcsfs/core.py b/gcsfs/core.py
index c43200822..6f1e05711 100644
--- a/gcsfs/core.py
+++ b/gcsfs/core.py
@@ -2302,6 +2302,9 @@ def sign(self, path, expiration=100, **kwargs):
         )
 
 
+GoogleCredentials.load_tokens()
+
+
 def _get_prefetcher_and_cache_value(cache_type, kwargs):
     """
     Resolves effective cache_type and whether prefetch reader should be enabled.

From d59140496d4080b2a640337f02f480141a1ad3df Mon Sep 17 00:00:00 2001
From: Ankita Luthra <lankita@google.com>
Date: Wed, 5 Aug 2026 14:09:32 +0000
Subject: [PATCH 08/10] reverts unrelated changes

---
 docs/source/prefetcher.rst | 16 ++++++++++++----
 1 file changed, 12 insertions(+), 4 deletions(-)

diff --git a/docs/source/prefetcher.rst b/docs/source/prefetcher.rst
index cbf5505c9..6fb6eeb97 100644
--- a/docs/source/prefetcher.rst
+++ b/docs/source/prefetcher.rst
@@ -2,7 +2,7 @@
 GCSFS Adaptive Concurrent Prefetching: Architecture & Usage Guide
 =================================================================
 
-Prefetcher is enabled by default with `DEFAULT_GCSFS_CONCURRENCY=4` and `cache_type="none"`. To disable, you can pass the environment variable `USE_EXPERIMENTAL_ADAPTIVE_PREFETCHING='false'` or pass `use_experimental_adaptive_prefetching=False` when opening a file. As currently written, this implementation is
+Prefetcher is enabled by default when cache_type is not set explicitly with `DEFAULT_GCSFS_CONCURRENCY=4`. To disable, you can pass the environment variable `USE_EXPERIMENTAL_ADAPTIVE_PREFETCHING='false'` or pass `use_experimental_adaptive_prefetching=False` when opening a file. As currently written, this implementation is
 separate from the fsspec-style caching layer, but the intent is to eventually make this available to all
 asynchronous filesystems using the standard `cache_type=` argument. How it interacts with the
 existing cache types ("readahead", "first", etc.) remains to be decided, and in the meantime, use at your own risk.
@@ -73,15 +73,23 @@ The prefetcher is integrated into the ``GCSFile`` and replaces the standard sequ
 Feature Configuration & Disabling
 ---------------------------------
 
-Adaptive prefetching is enabled by default with ``DEFAULT_GCSFS_CONCURRENCY=4`` and ``cache_type="none"``.
+Adaptive prefetching is enabled by default when ``cache_type`` is not explicitly set by the user, using ``DEFAULT_GCSFS_CONCURRENCY=4``.
 
-To disable prefetching and revert to the legacy readahead cache, set the environment variable:
+Prefetching can be disabled in three ways:
+
+1. Explicitly specify a ``cache_type`` when opening a file (e.g., ``cache_type="readahead"`` or any other cache_type):
+
+.. code-block:: python
+
+    gcs.open("bucket/file.txt", "rb", cache_type="readahead")
+
+2. Set the environment variable:
 
 .. code-block:: bash
 
     export USE_EXPERIMENTAL_ADAPTIVE_PREFETCHING='false'
 
-or pass ``use_experimental_adaptive_prefetching=False`` directly when opening a file:
+3. Pass ``use_experimental_adaptive_prefetching=False`` directly when opening a file:
 
 .. code-block:: python
 

From 7a6fcbab3b76d4967a67526a74c14424a54f77bf Mon Sep 17 00:00:00 2001
From: Ankita Luthra <lankita@google.com>
Date: Wed, 5 Aug 2026 18:25:01 +0000
Subject: [PATCH 09/10] fallaback to readahead instead of none when prefetcher
 is disabled

---
 gcsfs/core.py                      | 12 ++++++------
 gcsfs/tests/test_core.py           |  4 ++--
 gcsfs/tests/test_extended_gcsfs.py |  5 +++--
 gcsfs/tests/test_zonal_file.py     |  4 ++--
 4 files changed, 13 insertions(+), 12 deletions(-)

diff --git a/gcsfs/core.py b/gcsfs/core.py
index 6f1e05711..a99b82f17 100644
--- a/gcsfs/core.py
+++ b/gcsfs/core.py
@@ -2305,19 +2305,18 @@ def sign(self, path, expiration=100, **kwargs):
 GoogleCredentials.load_tokens()
 
 
-def _get_prefetcher_and_cache_value(cache_type, kwargs):
+def _get_prefetcher_and_cache_config(cache_type, kwargs):
     """
     Resolves effective cache_type and whether prefetch reader should be enabled.
 
     Rules:
-    - Prefetcher is only used when cache_type is not explicitly set by the user (cache_type is None).
-    - If user sets any cache_type (cache_type is not None), prefetcher is always disabled.
-    - By default when cache_type is not set, cache_type is "none".
+    - If user explicitly sets cache_type (cache_type is not None), prefetcher is disabled and cache_type is used.
+    - If cache_type is None and prefetcher is enabled (default), cache_type is "none" and prefetcher is active.
+    - If cache_type is None and prefetcher is disabled, fallback to default_cache_type.
     """
     if cache_type is not None:
         use_prefetch_reader = False
     else:
-        cache_type = "none"
         if "use_experimental_adaptive_prefetching" in kwargs:
             val = kwargs["use_experimental_adaptive_prefetching"]
             use_prefetch_reader = (
@@ -2330,6 +2329,7 @@ def _get_prefetcher_and_cache_value(cache_type, kwargs):
                 "true",
                 "1",
             )
+        cache_type = "none" if use_prefetch_reader else "readahead"
     return cache_type, use_prefetch_reader
 
 
@@ -2404,7 +2404,7 @@ def __init__(
             raise OSError("Attempt to open a bucket")
         self.generation = _coalesce_generation(generation, path_generation)
         self.concurrency = kwargs.get("concurrency", DEFAULT_CONCURRENCY)
-        cache_type, use_prefetch_reader = _get_prefetcher_and_cache_value(
+        cache_type, use_prefetch_reader = _get_prefetcher_and_cache_config(
             cache_type, kwargs
         )
 
diff --git a/gcsfs/tests/test_core.py b/gcsfs/tests/test_core.py
index b61b5f130..035c073ac 100644
--- a/gcsfs/tests/test_core.py
+++ b/gcsfs/tests/test_core.py
@@ -2824,10 +2824,10 @@ def test_gcsfile_prefetch_and_cache_type_rules(gcs):
         assert f.cache_type == "none"
         assert f.read() == b"HelloWorld"
 
-    # 2. Prefetcher disabled, no cache_type set -> no prefetcher, cache_type is "none"
+    # 2. Prefetcher disabled, no cache_type set -> no prefetcher, cache_type falls back to "readahead"
     with gcs.open(fn, "rb", use_experimental_adaptive_prefetching=False) as f:
         assert getattr(f, "_prefetch_engine", None) is None
-        assert f.cache_type == "none"
+        assert f.cache_type == "readahead"
         assert f.read() == b"HelloWorld"
 
     # 3. User sets cache_type="readahead" -> prefetcher NOT used, cache_type is "readahead"
diff --git a/gcsfs/tests/test_extended_gcsfs.py b/gcsfs/tests/test_extended_gcsfs.py
index 66c0c0dfc..e48e4f525 100644
--- a/gcsfs/tests/test_extended_gcsfs.py
+++ b/gcsfs/tests/test_extended_gcsfs.py
@@ -386,11 +386,12 @@ def test_default_cache_is_none_with_prefetcher(extended_gcsfs, gcs_bucket_mocks)
             assert isinstance(f.cache, caching.BaseCache)
             assert f._prefetch_engine is not None
 
-        # 2. Prefetcher disabled, cache_type not set -> no prefetcher, cache is BaseCache ("none")
+        # 2. Prefetcher disabled, cache_type not set -> no prefetcher, cache falls back to ReadAhead
         with extended_gcsfs.open(
             b, "rb", use_experimental_adaptive_prefetching=False
         ) as f:
-            assert isinstance(f.cache, caching.BaseCache)
+            import fsspec
+            assert isinstance(f.cache, fsspec.caching.ReadAheadCache)
             assert f._prefetch_engine is None
 
         # 3. Explicit cache_type="readahead_chunked" -> no prefetcher, cache is ReadAheadChunked
diff --git a/gcsfs/tests/test_zonal_file.py b/gcsfs/tests/test_zonal_file.py
index dae11f3a4..2164f1b1d 100644
--- a/gcsfs/tests/test_zonal_file.py
+++ b/gcsfs/tests/test_zonal_file.py
@@ -737,14 +737,14 @@ def test_zonal_file_cache_type_default_resolution(mock_sync, mock_gcsfs):
     assert zf_default._prefetch_engine is not None
     zf_default.close()
 
-    # 2. Prefetcher disabled (opt-out), no cache_type set -> cache_type="none", no prefetcher
+    # 2. Prefetcher disabled (opt-out), no cache_type set -> cache_type="readahead", no prefetcher
     zf_no_prefetch = ZonalFile(
         gcsfs=mock_gcsfs,
         path="gs://test-bucket/test-key",
         mode="rb",
         use_experimental_adaptive_prefetching=False,
     )
-    assert zf_no_prefetch.cache_type == "none"
+    assert zf_no_prefetch.cache_type == "readahead"
     assert zf_no_prefetch._prefetch_engine is None
     zf_no_prefetch.close()
 

From 3d75082015efa33d41dc01ecd573cadbe4e8dd26 Mon Sep 17 00:00:00 2001
From: Ankita Luthra <lankita@google.com>
Date: Wed, 5 Aug 2026 18:33:24 +0000
Subject: [PATCH 10/10] update docs

---
 docs/source/prefetcher.rst         | 2 +-
 gcsfs/tests/test_extended_gcsfs.py | 1 +
 2 files changed, 2 insertions(+), 1 deletion(-)

diff --git a/docs/source/prefetcher.rst b/docs/source/prefetcher.rst
index 6fb6eeb97..110ef0cc7 100644
--- a/docs/source/prefetcher.rst
+++ b/docs/source/prefetcher.rst
@@ -77,7 +77,7 @@ Adaptive prefetching is enabled by default when ``cache_type`` is not explicitly
 
 Prefetching can be disabled in three ways:
 
-1. Explicitly specify a ``cache_type`` when opening a file (e.g., ``cache_type="readahead"`` or any other cache_type):
+1. Explicitly specify a ``cache_type`` when opening a file (e.g., ``cache_type="readahead"`` or ``cache_type="none"`` or any other cache_type):
 
 .. code-block:: python
 
diff --git a/gcsfs/tests/test_extended_gcsfs.py b/gcsfs/tests/test_extended_gcsfs.py
index e48e4f525..c042acbdc 100644
--- a/gcsfs/tests/test_extended_gcsfs.py
+++ b/gcsfs/tests/test_extended_gcsfs.py
@@ -391,6 +391,7 @@ def test_default_cache_is_none_with_prefetcher(extended_gcsfs, gcs_bucket_mocks)
             b, "rb", use_experimental_adaptive_prefetching=False
         ) as f:
             import fsspec
+
             assert isinstance(f.cache, fsspec.caching.ReadAheadCache)
             assert f._prefetch_engine is None