diff --git a/docs/guides/request_loaders.mdx b/docs/guides/request_loaders.mdx
index a63393142d..a1e95130d3 100644
--- a/docs/guides/request_loaders.mdx
+++ b/docs/guides/request_loaders.mdx
@@ -72,6 +72,7 @@ class RequestManager {
     + add_request()
     + add_requests_batched()
     + reclaim_request()
+    + purge()
     + drop()
 }
 
diff --git a/docs/guides/storages.mdx b/docs/guides/storages.mdx
index a4348e28ab..8506396f1b 100644
--- a/docs/guides/storages.mdx
+++ b/docs/guides/storages.mdx
@@ -206,7 +206,7 @@ Crawlee provides the following helper function to simplify interactions with the
 
 ## Cleaning up the storages
 
-By default, Crawlee cleans up all unnamed storages (including the default one) at the start of each run, so every crawl begins with a clean state. This behavior is controlled by <ApiLink to="class/Configuration#purge_on_start">`Configuration.purge_on_start`</ApiLink> (default: True). In contrast, named storages are never purged automatically and persist across runs. The exact behavior may vary depending on the storage client implementation.
+By default, Crawlee cleans up all unnamed storages (including the default one) once per process, when each one is first opened, so every process begins with a clean state. This behavior is controlled by <ApiLink to="class/Configuration#purge_on_start">`Configuration.purge_on_start`</ApiLink> (default: True). In contrast, named storages are never purged automatically and persist across runs. The exact behavior may vary depending on the storage client implementation.
 
 ### When purging happens
 
@@ -214,7 +214,7 @@ The cleanup occurs as soon as a storage is accessed:
 
 - When opening a storage explicitly (e.g., <ApiLink to="class/RequestQueue#open">`RequestQueue.open`</ApiLink>, <ApiLink to="class/Dataset#open">`Dataset.open`</ApiLink>, <ApiLink to="class/KeyValueStore#open">`KeyValueStore.open`</ApiLink>).
 - When using helper functions that implicitly open storages (e.g., <ApiLink to="class/PushDataFunction">`push_data`</ApiLink>).
-- Automatically when <ApiLink to="class/BasicCrawler#run">`BasicCrawler.run`</ApiLink> is invoked.
+- Automatically when <ApiLink to="class/BasicCrawler#run">`BasicCrawler.run`</ApiLink> opens the crawler's storages.
 
 ### Disabling automatic purging
 
diff --git a/docs/upgrading/upgrading_to_v2.md b/docs/upgrading/upgrading_to_v2.md
index c3b64a2f3b..927116a7e5 100644
--- a/docs/upgrading/upgrading_to_v2.md
+++ b/docs/upgrading/upgrading_to_v2.md
@@ -179,3 +179,39 @@ queue = await RequestQueue.open()
 crawler_1 = BasicCrawler(request_manager=queue)
 crawler_2 = BasicCrawler(request_manager=queue)
 ```
+
+## Repeated `run()` calls no longer purge the request queue
+
+In v1, calling `crawler.run()` again on the same instance purged the request queue, unless the queue was named or the previous run had failed. In v2, nothing is purged between runs, the same as Crawlee JS. A repeated `run()` continues with the same request manager and skips the requests an earlier run handled, failed ones included.
+
+The `purge_request_queue` argument of `run()` is removed. If you passed `purge_request_queue=False`, drop the argument. To crawl the same requests again, purge the request manager yourself:
+
+```python
+# v1
+await crawler.run(urls)
+await crawler.run(urls)
+
+# v2
+await crawler.run(urls)
+request_manager = await crawler.get_request_manager()
+await request_manager.purge()
+await crawler.run(urls)
+```
+
+`RequestManagerTandem.purge()` now purges the manager it wraps, where v1 did nothing. The loader isn't purged, so requests it has already passed to the manager don't come back.
+
+Instead of purging, you can create a new crawler for each crawl. Every crawler after the first opens a request queue of its own, as described in [Only the first crawler uses the default request queue](#only-the-first-crawler-uses-the-default-request-queue). Crawlers created with the same `id` share that queue, so leave `id` unset or give each crawler a different one:
+
+```python
+for urls in batches:
+    crawler = BasicCrawler()
+    await crawler.run(urls)
+```
+
+When a crawl processes no requests because its request manager holds only handled ones, the crawler now logs a warning that explains how to crawl them again.
+
+`Configuration.purge_on_start` isn't affected. Unnamed storages are still purged once per process, when they're first opened.
+
+## `RequestQueue.purge` raises when the storage client can't purge
+
+In v1, `RequestQueue.purge()` worked around a storage client that didn't support purging. It dropped an unnamed queue and opened the default queue in its place, and it skipped a named queue with a warning. In v2, it raises the client's `NotImplementedError`, the same as Crawlee JS. The Apify platform's storage client doesn't support purging, so on the platform, call `drop()` or open a new queue instead.
diff --git a/src/crawlee/crawlers/_basic/_basic_crawler.py b/src/crawlee/crawlers/_basic/_basic_crawler.py
index fd4bcee6b3..b54e6aa988 100644
--- a/src/crawlee/crawlers/_basic/_basic_crawler.py
+++ b/src/crawlee/crawlers/_basic/_basic_crawler.py
@@ -530,7 +530,7 @@ async def persist_state_factory() -> KeyValueStore:
         self._keep_alive = keep_alive
         self._running = False
         self._has_finished_before = False
-        self._last_run_failed = False
+        self._handled_request_this_run = False
         self._failed = False
         self._unexpected_stop = False
         self._logger_once = LoggerOnce(self._logger)
@@ -717,20 +717,15 @@ def on_skipped_request(self, callback: SkippedRequestCallback) -> SkippedRequest
         self._on_skipped_request = callback
         return callback
 
-    async def run(
-        self,
-        requests: Sequence[str | Request] | None = None,
-        *,
-        purge_request_queue: bool = True,
-    ) -> FinalStatistics:
+    async def run(self, requests: Sequence[str | Request] | None = None) -> FinalStatistics:
         """Run the crawler until all requests are processed.
 
+        Calling `run()` again on the same instance keeps crawling the same request manager. Requests that a previous
+        run handled, failed ones included, aren't processed again. To crawl them again, purge the request manager, or
+        create a new crawler with a freshly opened request queue.
+
         Args:
             requests: The requests to be enqueued before the crawler starts.
-            purge_request_queue: If this is `True` and the crawler is not being run for the first time, the request
-                queue will be purged. A run that ended with an exception does not count as a previous run, so a
-                retry keeps the requests that were still pending even when this is `True`. Named request queues
-                are considered persistent and are never purged implicitly.
 
         Raises:
             PersistentRateLimitError: If a domain throttled by a `ThrottlingRequestManager` has rate-limited every
@@ -742,6 +737,7 @@ async def run(
             )
 
         self._running = True
+        self._handled_request_this_run = False
 
         try:
             if self._respect_robots_txt_file and not isinstance(self._request_manager, ThrottlingRequestManager):
@@ -758,22 +754,6 @@ async def run(
                 if self._use_session_pool:
                     await self._session_pool.reset_store()
 
-                # A failed run does not count as a previous run, so its pending requests survive into the retry.
-                if purge_request_queue and not self._last_run_failed:
-                    request_manager = await self.get_request_manager()
-                    # A `ThrottlingRequestManager` delegates `purge` to the manager it wraps, so inspect the wrapped
-                    # manager when deciding whether the purge would hit a named queue.
-                    inner_manager = (
-                        request_manager.inner
-                        if isinstance(request_manager, ThrottlingRequestManager)
-                        else request_manager
-                    )
-                    # Named storages are persistent and shared across runs, so they are never purged implicitly
-                    # (the same named-storage exemption as in `StorageClient._purge_if_needed`).
-                    is_named_queue = isinstance(inner_manager, RequestQueue) and inner_manager.name is not None
-                    if not is_named_queue:
-                        await request_manager.purge()
-
             if requests is not None:
                 await self.add_requests(requests)
 
@@ -827,7 +807,6 @@ def sigint_handler() -> None:
                 )
 
             self._has_finished_before = True
-            self._last_run_failed = False
 
             await self._save_crawler_state()
 
@@ -836,10 +815,31 @@ def sigint_handler() -> None:
                 self._logger.info(f'Final request statistics:\n{final_statistics.to_table()}')
             else:
                 self._logger.info('Final request statistics:', extra=final_statistics.to_dict())
-        except BaseException:
-            self._last_run_failed = True
-            raise
-        else:
+
+            # A crawl that did nothing while the manager holds only handled requests is a mistake, whoever handled them.
+            # Starting against handled requests is not, since that is what resuming a crawl looks like. The statistics
+            # can't tell, as they don't count skipped requests and may be restored from a previous process.
+            if not self._handled_request_this_run:
+                request_manager = await self.get_request_manager()
+                # The diagnostic must never fail the run.
+                try:
+                    is_finished = await request_manager.is_finished()
+                    handled_count = await request_manager.get_handled_count()
+                except Exception:
+                    is_finished = False
+                    handled_count = 0
+
+                # Pending requests mean the crawl ended early, such as after an abort, not that everything was handled.
+                if is_finished and handled_count > 0:
+                    noun = 'request' if handled_count == 1 else 'requests'
+                    message = (
+                        f'This crawl processed no requests - the request manager holds {handled_count} {noun}, all of '
+                        'them already handled, and a failed request counts as handled too. Nothing empties a queue '
+                        'between runs, so to crawl them again, purge it (`await queue.purge()`) or use a fresh one '
+                        "(e.g. `RequestQueue.open(alias='second-run')`) with a freshly created crawler instance."
+                    )
+                    self._logger_once.log(message, key=message, level=logging.WARNING)
+
             return final_statistics
         finally:
             # A failed run must leave the instance usable, so that the caller can retry after handling the error.
@@ -1953,3 +1953,4 @@ async def _mark_request_as_handled(self, request: Request) -> None:
             logger=self._logger,
             max_retries=3,
         )
+        self._handled_request_this_run = True
diff --git a/src/crawlee/request_loaders/_request_manager_tandem.py b/src/crawlee/request_loaders/_request_manager_tandem.py
index a605c25d25..325863bc18 100644
--- a/src/crawlee/request_loaders/_request_manager_tandem.py
+++ b/src/crawlee/request_loaders/_request_manager_tandem.py
@@ -109,3 +109,12 @@ async def mark_request_as_handled(self, request: Request) -> None:
     @override
     async def drop(self) -> None:
         await self._read_write_manager.drop()
+
+    @override
+    async def purge(self) -> None:
+        """Purge the wrapped request manager.
+
+        The loader can't be purged. Requests it already passed to the manager don't come back, and the ones it still
+        holds are fetched as usual.
+        """
+        await self._read_write_manager.purge()
diff --git a/src/crawlee/storages/_request_queue.py b/src/crawlee/storages/_request_queue.py
index b614928a2b..b5a9bef1b2 100644
--- a/src/crawlee/storages/_request_queue.py
+++ b/src/crawlee/storages/_request_queue.py
@@ -151,28 +151,7 @@ async def drop(self) -> None:
 
     @override
     async def purge(self) -> None:
-        try:
-            await self._client.purge()
-        except NotImplementedError:
-            client_name = type(self._client).__name__
-            if self._name is not None:
-                logger.warning(
-                    f'Storage client "{client_name}" does not support purging the request queue. '
-                    f'Skipping purge for named queue "{self._name}" to avoid destroying persistent data; '
-                    'the queue contents are left intact.'
-                )
-                return
-            logger.warning(
-                f'Storage client "{client_name}" does not support purging the request queue. '
-                'Falling back to dropping and recreating the unnamed queue; the request queue ID may change.'
-            )
-            await self.drop()
-            # Override `purge_on_start` so the storage client does not try to purge the freshly recreated
-            # (and necessarily empty) queue and re-raise the same `NotImplementedError`.
-            recreate_config = service_locator.get_configuration().model_copy(update={'purge_on_start': False})
-            new_rq = await RequestQueue.open(configuration=recreate_config)
-            self._client = new_rq._client  # noqa: SLF001
-            self._id = new_rq._id  # noqa: SLF001
+        await self._client.purge()
 
     @override
     async def add_request(
diff --git a/tests/unit/crawlers/_basic/test_basic_crawler.py b/tests/unit/crawlers/_basic/test_basic_crawler.py
index 422b0aac37..5b96c9bca6 100644
--- a/tests/unit/crawlers/_basic/test_basic_crawler.py
+++ b/tests/unit/crawlers/_basic/test_basic_crawler.py
@@ -30,6 +30,7 @@
 from crawlee.configuration import Configuration
 from crawlee.crawlers import BasicCrawler
 from crawlee.errors import (
+    ContextPipelineInterruptedError,
     PersistentRateLimitError,
     RequestCollisionError,
     RequestThrottledError,
@@ -147,35 +148,8 @@ async def handler(context: BasicCrawlingContext) -> None:
     assert handled_urls == ['https://a.placeholder.com', 'https://b.placeholder.com']
 
 
-async def test_purge_resumes_once_a_run_succeeds_again(monkeypatch: pytest.MonkeyPatch) -> None:
-    """The purge exemption lasts only until a run succeeds, so the run after the retry starts from a clean queue."""
-    queue = await RequestQueue.open()
-    crawler = BasicCrawler(request_manager=queue)
-    handled_urls = []
-
-    @crawler.router.default_handler
-    async def handler(context: BasicCrawlingContext) -> None:
-        handled_urls.append(context.request.url)
-
-    await crawler.run(['https://a.placeholder.com'])
-
-    with monkeypatch.context() as monkey:
-        monkey.setattr(queue, 'is_empty', AsyncMock(side_effect=RuntimeError('Queue status unavailable')))
-        with pytest.raises(RuntimeError, match='Queue status unavailable'):
-            await crawler.run(['https://b.placeholder.com'])
-
-    await crawler.run()
-    await crawler.run(['https://a.placeholder.com'])
-
-    assert handled_urls == [
-        'https://a.placeholder.com',
-        'https://b.placeholder.com',
-        'https://a.placeholder.com',
-    ]
-
-
-async def test_failure_after_the_crawl_marks_the_run_as_failed(monkeypatch: pytest.MonkeyPatch) -> None:
-    """A run whose post-crawl state save raises counts as failed too, so the retry keeps the queue intact."""
+async def test_crawler_is_usable_after_a_failure_after_the_crawl(monkeypatch: pytest.MonkeyPatch) -> None:
+    """A failure after the crawl leaves the instance usable, and the next run continues the same queue."""
     queue = await RequestQueue.open()
     crawler = BasicCrawler(request_manager=queue)
     handled_urls = []
@@ -236,7 +210,8 @@ async def handler(context: BasicCrawlingContext) -> None:
     assert calls == ['https://a.placeholder.com', 'https://b.placeholder.com', 'https://c.placeholder.com']
 
 
-async def test_allows_multiple_run_calls() -> None:
+async def test_allows_multiple_run_calls(caplog: pytest.LogCaptureFixture) -> None:
+    """A repeated run continues the same queue, so it crawls only the new requests and doesn't warn."""
     crawler = BasicCrawler()
     calls = list[str]()
 
@@ -245,16 +220,23 @@ async def handler(context: BasicCrawlingContext) -> None:
         calls.append(context.request.url)
 
     await crawler.run(['https://a.placeholder.com', 'https://b.placeholder.com', 'https://c.placeholder.com'])
-    await crawler.run(['https://a.placeholder.com', 'https://b.placeholder.com', 'https://c.placeholder.com'])
+    with caplog.at_level(logging.WARNING):
+        await crawler.run(
+            [
+                'https://a.placeholder.com',
+                'https://b.placeholder.com',
+                'https://c.placeholder.com',
+                'https://d.placeholder.com',
+            ]
+        )
 
     assert calls == [
         'https://a.placeholder.com',
         'https://b.placeholder.com',
         'https://c.placeholder.com',
-        'https://a.placeholder.com',
-        'https://b.placeholder.com',
-        'https://c.placeholder.com',
+        'https://d.placeholder.com',
     ]
+    assert processed_no_requests_warnings(caplog) == []
 
 
 async def test_retries_failed_requests() -> None:
@@ -1487,7 +1469,8 @@ def test_configure_logging_sets_up_unconfigured_root_logger() -> None:
         root_logger.setLevel(saved_level)
 
 
-async def test_consecutive_runs_purge_request_queue() -> None:
+async def test_consecutive_runs_recrawl_after_purge() -> None:
+    """Purging the request manager between runs lets the same requests be crawled again."""
     crawler = BasicCrawler()
     visit = Mock()
 
@@ -1495,9 +1478,10 @@ async def test_consecutive_runs_purge_request_queue() -> None:
     async def handler(context: BasicCrawlingContext) -> None:
         visit(context.request.url)
 
-    await crawler.run(['https://a.placeholder.com', 'https://b.placeholder.com', 'https://c.placeholder.com'])
-    await crawler.run(['https://a.placeholder.com', 'https://b.placeholder.com', 'https://c.placeholder.com'])
-    await crawler.run(['https://a.placeholder.com', 'https://b.placeholder.com', 'https://c.placeholder.com'])
+    for _ in range(3):
+        await crawler.run(['https://a.placeholder.com', 'https://b.placeholder.com', 'https://c.placeholder.com'])
+        request_manager = await crawler.get_request_manager()
+        await request_manager.purge()
 
     counter = Counter(args[0][0] for args in visit.call_args_list)
     assert counter == {
@@ -1546,6 +1530,169 @@ async def handler(context: BasicCrawlingContext) -> None:
     assert visited == ['https://a.placeholder.com', 'https://b.placeholder.com']
 
 
+async def test_repeated_run_over_throttling_manager_does_not_recrawl() -> None:
+    """A second `run()` over a `ThrottlingRequestManager` doesn't crawl a request its domain queue already handled."""
+    throttler = ThrottlingRequestManager(
+        inner=await RequestQueue.open(), domains=['a.placeholder.com'], request_manager_opener=RequestQueue.open
+    )
+    crawler = BasicCrawler(request_manager=throttler)
+    visited = list[str]()
+
+    @crawler.router.default_handler
+    async def handler(context: BasicCrawlingContext) -> None:
+        visited.append(context.request.url)
+
+    await crawler.run(['https://a.placeholder.com'])
+    await crawler.run(['https://a.placeholder.com'])
+
+    assert visited == ['https://a.placeholder.com']
+
+
+def processed_no_requests_warnings(caplog: pytest.LogCaptureFixture) -> list[str]:
+    """Return the logged warnings about a crawl that processed no requests."""
+    return [record.message for record in caplog.records if 'processed no requests' in record.message]
+
+
+async def test_repeated_run_over_handled_requests_warns_once(caplog: pytest.LogCaptureFixture) -> None:
+    """A repeated run over requests that are all handled warns once, explaining how to crawl them again."""
+    crawler = BasicCrawler()
+    handler = AsyncMock()
+    crawler.router.default_handler(handler)
+
+    with caplog.at_level(logging.WARNING):
+        for _ in range(3):
+            await crawler.run(['https://a.placeholder.com'])
+
+    handler.assert_awaited_once()
+    warnings = processed_no_requests_warnings(caplog)
+    assert len(warnings) == 1
+    assert 'holds 1 request,' in warnings[0]
+
+
+async def test_crawler_over_exhausted_queue_warns(caplog: pytest.LogCaptureFixture) -> None:
+    """A second crawler over a queue that another crawler worked through warns that it processed nothing."""
+    queue = await RequestQueue.open()
+    first_crawler = BasicCrawler(request_manager=queue)
+    first_crawler.router.default_handler(AsyncMock())
+    await first_crawler.run(['https://a.placeholder.com', 'https://b.placeholder.com'])
+
+    second_crawler = BasicCrawler(request_manager=queue)
+    second_crawler.router.default_handler(AsyncMock())
+    with caplog.at_level(logging.WARNING):
+        await second_crawler.run()
+
+    warnings = processed_no_requests_warnings(caplog)
+    assert len(warnings) == 1
+    assert 'holds 2 requests,' in warnings[0]
+
+
+async def test_run_with_nothing_to_crawl_does_not_warn(caplog: pytest.LogCaptureFixture) -> None:
+    """A run over an empty queue processes nothing without warning."""
+    crawler = BasicCrawler()
+
+    with caplog.at_level(logging.WARNING):
+        await crawler.run()
+
+    assert processed_no_requests_warnings(caplog) == []
+
+
+async def test_failed_request_counts_as_handled_for_a_repeated_run(caplog: pytest.LogCaptureFixture) -> None:
+    """A request that failed isn't crawled again by a repeated run, which warns instead."""
+    crawler = BasicCrawler(max_request_retries=0)
+    handler = AsyncMock(side_effect=RuntimeError('Handler failed'))
+    crawler.router.default_handler(handler)
+
+    with caplog.at_level(logging.WARNING):
+        first_stats = await crawler.run(['https://a.placeholder.com'])
+        assert first_stats.requests_failed == 1
+        assert processed_no_requests_warnings(caplog) == []
+
+        await crawler.run(['https://a.placeholder.com'])
+
+    handler.assert_awaited_once()
+    assert len(processed_no_requests_warnings(caplog)) == 1
+
+
+async def test_handled_count_error_does_not_fail_the_run(
+    monkeypatch: pytest.MonkeyPatch,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A request manager that fails to report its handled count doesn't fail the run, which then doesn't warn."""
+    queue = await RequestQueue.open()
+    crawler = BasicCrawler(request_manager=queue)
+    crawler.router.default_handler(AsyncMock())
+    await crawler.run(['https://a.placeholder.com'])
+
+    get_handled_count = AsyncMock(side_effect=RuntimeError('Handled count unavailable'))
+    monkeypatch.setattr(queue, 'get_handled_count', get_handled_count)
+    with caplog.at_level(logging.WARNING):
+        stats = await crawler.run(['https://a.placeholder.com'])
+
+    assert stats.requests_total == 0
+    get_handled_count.assert_awaited_once()
+    assert processed_no_requests_warnings(caplog) == []
+
+
+@pytest.mark.parametrize(
+    ('method_name', 'skip'),
+    [
+        pytest.param('_is_allowed_based_on_robots_txt_file', AsyncMock(return_value=False), id='robots-disallowed'),
+        pytest.param(
+            '_run_request_handler', AsyncMock(side_effect=ContextPipelineInterruptedError), id='pipeline-interrupted'
+        ),
+    ],
+)
+async def test_run_that_only_skips_requests_does_not_warn(
+    caplog: pytest.LogCaptureFixture,
+    monkeypatch: pytest.MonkeyPatch,
+    method_name: str,
+    skip: AsyncMock,
+) -> None:
+    """A run that only skips requests marks them handled without warning that it processed nothing."""
+    queue = await RequestQueue.open()
+    await queue.add_request('https://a.placeholder.com')
+    crawler = BasicCrawler(request_manager=queue)
+    crawler.router.default_handler(AsyncMock())
+    monkeypatch.setattr(crawler, method_name, skip)
+
+    with caplog.at_level(logging.WARNING):
+        stats = await crawler.run()
+
+    assert stats.requests_total == 0
+    assert await queue.get_handled_count() == 1
+    assert processed_no_requests_warnings(caplog) == []
+
+
+async def test_aborted_crawl_with_pending_requests_does_not_warn(caplog: pytest.LogCaptureFixture) -> None:
+    """A crawl aborted before it processed anything doesn't warn while the aborted request is still pending."""
+    crawler = BasicCrawler(
+        abort_on_error=True,
+        concurrency_settings=ConcurrencySettings(desired_concurrency=1, max_concurrency=1),
+    )
+    handler = AsyncMock(side_effect=[None, RuntimeError('Handler failed')])
+    crawler.router.default_handler(handler)
+    await crawler.run(['https://a.placeholder.com'])
+
+    with caplog.at_level(logging.WARNING):
+        stats = await crawler.run(['https://b.placeholder.com'])
+
+    assert stats.requests_total == 0
+    assert processed_no_requests_warnings(caplog) == []
+
+
+async def test_handled_count_not_read_when_requests_were_processed(monkeypatch: pytest.MonkeyPatch) -> None:
+    """A run that processed requests doesn't ask the request manager for its handled count."""
+    queue = await RequestQueue.open()
+    get_handled_count = AsyncMock(wraps=queue.get_handled_count)
+    monkeypatch.setattr(queue, 'get_handled_count', get_handled_count)
+    crawler = BasicCrawler(request_manager=queue)
+    crawler.router.default_handler(AsyncMock())
+
+    await crawler.run(['https://a.placeholder.com'])
+
+    get_handled_count.assert_not_awaited()
+
+
 @pytest.mark.skipif(os.name == 'nt' and 'CI' in os.environ, reason='Skipped in Windows CI')
 @pytest.mark.parametrize(
     ('statistics_log_format'),
@@ -2333,29 +2480,6 @@ async def handler(context: BasicCrawlingContext) -> None:
     await crawler.run(['https://start.placeholder.com'])
 
 
-async def test_crawler_purge_request_queue_uses_same_storage_client() -> None:
-    """Make sure that purge on start does not replace the storage client in the underlying storage manager"""
-
-    # Set some different storage_client globally and different for Crawlee.
-    service_locator.set_storage_client(FileSystemStorageClient())
-    unrelated_rq = await RequestQueue.open()
-    unrelated_request = Request.from_url('https://x.placeholder.com')
-    await unrelated_rq.add_request(unrelated_request)
-
-    crawler = BasicCrawler(storage_client=MemoryStorageClient())
-
-    @crawler.router.default_handler
-    async def handler(context: BasicCrawlingContext) -> None:
-        context.log.info(context.request.url)
-
-    for _ in (1, 2):
-        await crawler.run(requests=[Request.from_url('https://a.placeholder.com')], purge_request_queue=True)
-        assert crawler.statistics.state.requests_finished == 1
-
-    # Crawler should not fall back to the default storage after the purge
-    assert await unrelated_rq.fetch_next_request() == unrelated_request
-
-
 async def _run_crawler(crawler_id: str | None, requests: list[str], storage_dir: str) -> StatisticsState:
     """Run crawler and return its statistics state.
 
diff --git a/tests/unit/storages/test_request_manager_tandem.py b/tests/unit/storages/test_request_manager_tandem.py
index 69bd944348..26166c8c0b 100644
--- a/tests/unit/storages/test_request_manager_tandem.py
+++ b/tests/unit/storages/test_request_manager_tandem.py
@@ -6,7 +6,7 @@
 import pytest
 
 from crawlee import Request
-from crawlee.request_loaders import RequestLoader, RequestManagerTandem
+from crawlee.request_loaders import RequestLoader, RequestManager, RequestManagerTandem
 from crawlee.storages import RequestQueue
 
 
@@ -80,3 +80,14 @@ async def test_basic_functionality(test_input: TestInput) -> None:
         await tandem.mark_request_as_handled(request)
 
     assert processed == test_input.expected_result
+
+
+async def test_purge_forwards_to_the_manager() -> None:
+    """Test that purging the tandem purges its manager and leaves the loader alone."""
+    loader = create_autospec(RequestLoader, instance=True, spec_set=True)
+    manager = create_autospec(RequestManager, instance=True, spec_set=True)
+
+    await RequestManagerTandem(loader, manager).purge()
+
+    manager.purge.assert_awaited_once_with()
+    assert loader.mock_calls == []
diff --git a/tests/unit/storages/test_request_queue.py b/tests/unit/storages/test_request_queue.py
index 112adfaa96..3eaa1f52b0 100644
--- a/tests/unit/storages/test_request_queue.py
+++ b/tests/unit/storages/test_request_queue.py
@@ -3,6 +3,7 @@
 import asyncio
 from datetime import timedelta
 from typing import TYPE_CHECKING
+from unittest.mock import AsyncMock
 
 import pytest
 
@@ -798,90 +799,30 @@ async def test_purge(
     await rq.drop()
 
 
-async def test_purge_falls_back_to_drop_for_unnamed_queue_when_not_implemented(
-    storage_client: StorageClient,
-    caplog: pytest.LogCaptureFixture,
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    """Test that `purge` falls back to drop+recreate for unnamed queues when the client raises `NotImplementedError`.
-
-    Some storage clients (e.g. the Apify platform client) do not support purging. For the default unnamed queue
-    used by `BasicCrawler`, `purge` should drop and recreate the queue so that callers keep working on repeated
-    runs. Named queues are handled separately to avoid destroying persistent data.
-    """
-    rq = await RequestQueue.open(storage_client=storage_client)
-    assert rq.name is None
-
-    await rq.add_requests(['https://example.com/1', 'https://example.com/2'])
-    metadata = await rq.get_metadata()
-    assert metadata.pending_request_count == 2
-
-    async def _raise_not_implemented(self: object) -> None:
-        raise NotImplementedError('Purge is not supported.')
-
-    monkeypatch.setattr(type(rq._client), 'purge', _raise_not_implemented)
-
-    with caplog.at_level('WARNING'):
-        await rq.purge()
-
-    assert any(
-        'does not support purging' in rec.message and 'dropping and recreating' in rec.message for rec in caplog.records
-    )
-
-    # The queue should be empty, usable, and backed by a fresh client (id may differ for backends that mint new ids).
-    metadata = await rq.get_metadata()
-    assert metadata.pending_request_count == 0
-    assert metadata.total_request_count == 0
-    assert metadata.handled_request_count == 0
-    assert rq.id is not None
-
-    await rq.add_request('https://example.com/after-purge')
-    request = await rq.fetch_next_request()
-    assert request is not None
-    assert request.url == 'https://example.com/after-purge'
-
-    await rq.drop()
-
-
-async def test_purge_skips_named_queue_when_not_implemented(
+@pytest.mark.parametrize(
+    'name',
+    [
+        pytest.param(None, id='unnamed'),
+        pytest.param('purge-unsupported-named-test', id='named'),
+    ],
+)
+async def test_purge_unsupported_by_client_raises(
     storage_client: StorageClient,
-    caplog: pytest.LogCaptureFixture,
     monkeypatch: pytest.MonkeyPatch,
+    name: str | None,
 ) -> None:
-    """Test that `purge` is a logged no-op for named queues when the client raises `NotImplementedError`.
-
-    Named queues are considered persistent (e.g. shared across runs on the Apify platform), so falling back
-    to drop+recreate would silently destroy user data. Instead `purge` logs a warning and leaves the queue
-    intact.
-    """
-    rq = await RequestQueue.open(
-        name='purge-fallback-named-test',
-        storage_client=storage_client,
-    )
+    """Test that `purge` raises when the storage client doesn't support it, leaving the queue intact."""
+    rq = await RequestQueue.open(name=name, storage_client=storage_client)
     original_id = rq.id
-
     await rq.add_requests(['https://example.com/1', 'https://example.com/2'])
-    metadata = await rq.get_metadata()
-    assert metadata.pending_request_count == 2
+    monkeypatch.setattr(rq._client, 'purge', AsyncMock(side_effect=NotImplementedError('Purge is not supported.')))
 
-    async def _raise_not_implemented(self: object) -> None:
-        raise NotImplementedError('Purge is not supported.')
-
-    monkeypatch.setattr(type(rq._client), 'purge', _raise_not_implemented)
-
-    with caplog.at_level('WARNING'):
+    with pytest.raises(NotImplementedError, match='Purge is not supported'):
         await rq.purge()
 
-    assert any(
-        'does not support purging' in rec.message and 'Skipping purge for named queue' in rec.message
-        for rec in caplog.records
-    )
-
-    # Queue identity and contents must be preserved.
     assert rq.id == original_id
     metadata = await rq.get_metadata()
     assert metadata.pending_request_count == 2
-    assert metadata.total_request_count == 2
 
     await rq.drop()