@@ -252,7 +252,7 @@ async def _worker_task_orchestrator(self, run: _AutoscaledPoolRun) -> None:
         finally:
             if finished:
                 logger.debug('`is_finished_function` reports that we are finished')
-            elif run.result.done() and run.result.exception() is not None:
+            elif run.result.done() and not run.result.cancelled() and run.result.exception() is not None:
                 logger.debug('Unhandled exception in `run_task_function`')
 
             if run.worker_tasks:
@@ -269,7 +269,7 @@ async def _worker_task_orchestrator(self, run: _AutoscaledPoolRun) -> None:
                     run.result.set_result(object())
             elif orchestrator_error is not None:
                 # A worker failure or an abort already decided the run, so this error has no way out.
-                logger.error('Exception in worker task orchestrator', exc_info=orchestrator_error)
+                logger.error('Unpropagated exception in worker task orchestrator', exc_info=orchestrator_error)
 
     def _reap_worker_task(self, task: asyncio.Task, run: _AutoscaledPoolRun) -> None:
         """Handle cleanup and tracking of a completed worker task.
 
@@ -507,6 +507,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._failed = False
         self._unexpected_stop = False
         self._logger_once = LoggerOnce(self._logger)
@@ -696,8 +697,8 @@ async def run(
             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. Named request queues are considered persistent
-                and are never purged implicitly.
+                retry keeps the requests that were still pending even when this is `True`. Named request queues
+                are considered persistent and are never purged implicitly.
         """
         if self._running:
             raise RuntimeError(
@@ -706,86 +707,97 @@ async def run(
 
         self._running = True
 
-        if self._respect_robots_txt_file and not isinstance(self._request_manager, ThrottlingRequestManager):
-            self._logger.warning(
-                'The `respect_robots_txt_file` option is enabled, but the crawler is not using '
-                '`ThrottlingRequestManager`. Crawl-delay directives from robots.txt will not be enforced. To enable '
-                'crawl-delay support, configure the crawler to use `ThrottlingRequestManager` as the request manager.'
-            )
-
-        if self._has_finished_before:
-            await self._statistics.reset()
-
-            if self._use_session_pool:
-                await self._session_pool.reset_store()
-
-            if purge_request_queue:
-                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
+        try:
+            if self._respect_robots_txt_file and not isinstance(self._request_manager, ThrottlingRequestManager):
+                self._logger.warning(
+                    'The `respect_robots_txt_file` option is enabled, but the crawler is not using '
+                    '`ThrottlingRequestManager`. Crawl-delay directives from robots.txt will not be enforced. To '
+                    'enable crawl-delay support, configure the crawler to use `ThrottlingRequestManager` as the '
+                    '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)
-
-        interrupted = False
+            if self._has_finished_before:
+                await self._statistics.reset()
+
+                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()
 
-        def sigint_handler() -> None:
-            nonlocal interrupted
+            if requests is not None:
+                await self.add_requests(requests)
 
-            if not interrupted:
-                interrupted = True
-                self._logger.info('Pausing... Press CTRL+C again to force exit.')
+            interrupted = False
 
-            run_task.cancel()
+            def sigint_handler() -> None:
+                nonlocal interrupted
 
-        run_task = asyncio.create_task(self._run_crawler(), name='run_crawler_task')
+                if not interrupted:
+                    interrupted = True
+                    self._logger.info('Pausing... Press CTRL+C again to force exit.')
 
-        if threading.current_thread() is threading.main_thread():  # `add_signal_handler` works only in the main thread
-            with suppress(NotImplementedError):  # event loop signal handlers are not supported on Windows
-                asyncio.get_running_loop().add_signal_handler(signal.SIGINT, sigint_handler)
+                run_task.cancel()
 
-        try:
-            await run_task
-        except CancelledError:
-            pass
-        finally:
-            # A failed run must leave the instance usable, so that the caller can retry after handling the error.
-            self._running = False
+            run_task = asyncio.create_task(self._run_crawler(), name='run_crawler_task')
 
+            # `add_signal_handler` works only in the main thread
             if threading.current_thread() is threading.main_thread():
-                with suppress(NotImplementedError):
-                    asyncio.get_running_loop().remove_signal_handler(signal.SIGINT)
+                with suppress(NotImplementedError):  # event loop signal handlers are not supported on Windows
+                    asyncio.get_running_loop().add_signal_handler(signal.SIGINT, sigint_handler)
 
-        if self._statistics.error_tracker.total > 0:
-            self._logger.info(
-                'Error analysis:'
-                f' total_errors={self._statistics.error_tracker.total}'
-                f' unique_errors={self._statistics.error_tracker.unique_error_count}'
-            )
+            try:
+                await run_task
+            except CancelledError:
+                pass
+            finally:
+                if threading.current_thread() is threading.main_thread():
+                    with suppress(NotImplementedError):
+                        asyncio.get_running_loop().remove_signal_handler(signal.SIGINT)
+
+            if self._statistics.error_tracker.total > 0:
+                self._logger.info(
+                    'Error analysis:'
+                    f' total_errors={self._statistics.error_tracker.total}'
+                    f' unique_errors={self._statistics.error_tracker.unique_error_count}'
+                )
 
-        if interrupted:
-            self._logger.info(
-                f'The crawl was interrupted. To resume, do: CRAWLEE_PURGE_ON_START=0 python {sys.argv[0]}'
-            )
+            if interrupted:
+                self._logger.info(
+                    f'The crawl was interrupted. To resume, do: CRAWLEE_PURGE_ON_START=0 python {sys.argv[0]}'
+                )
 
-        self._has_finished_before = True
+            self._has_finished_before = True
+            self._last_run_failed = False
 
-        await self._save_crawler_state()
+            await self._save_crawler_state()
 
-        final_statistics = self._statistics.calculate()
-        if self._statistics_log_format == 'table':
-            self._logger.info(f'Final request statistics:\n{final_statistics.to_table()}')
+            final_statistics = self._statistics.calculate()
+            if self._statistics_log_format == 'table':
+                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:
-            self._logger.info('Final request statistics:', extra=final_statistics.to_dict())
-        return final_statistics
+            return final_statistics
+        finally:
+            # A failed run must leave the instance usable, so that the caller can retry after handling the error.
+            self._running = False
 
     async def _run_crawler(self) -> None:
         local_event_manager = self._service_locator.get_event_manager()
 
@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import asyncio
+import logging
 from contextlib import suppress
 from datetime import timedelta
 from itertools import chain, repeat
@@ -164,6 +165,31 @@ async def is_finished() -> bool:
         await asyncio.gather(pool_run_task, return_exceptions=True)
 
 
+async def test_orchestrator_error_is_logged_when_the_run_is_cancelled(
+    system_status: SystemStatus | Mock, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A scheduling error raised while the run is being cancelled is logged, as the cancelled result cannot carry it."""
+
+    async def is_finished() -> bool:
+        pool_run_task.cancel()
+        raise RuntimeError('Queue status unavailable')
+
+    pool = AutoscaledPool(
+        system_status=system_status,
+        run_task_function=lambda: future(None),
+        is_task_ready_function=lambda: future(False),
+        is_finished_function=is_finished,
+    )
+
+    with caplog.at_level(logging.ERROR, logger='crawlee._autoscaling.autoscaled_pool'):
+        pool_run_task = asyncio.create_task(pool.run())
+        with pytest.raises(asyncio.CancelledError):
+            await pool_run_task
+
+    assert 'Unpropagated exception in worker task orchestrator' in caplog.text
+    assert 'Queue status unavailable' in caplog.text
+
+
 async def test_propagates_exceptions_after_finished(system_status: SystemStatus | Mock) -> None:
     started_count = 0
 
 
@@ -97,6 +97,97 @@ async def handler(context: BasicCrawlingContext) -> None:
     assert await queue.is_finished()
 
 
+async def test_crawler_is_usable_after_a_failed_run_setup(monkeypatch: pytest.MonkeyPatch) -> None:
+    """A failure before the crawl starts leaves the instance usable, so the caller can retry after handling it."""
+    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)
+
+    with monkeypatch.context() as monkey:
+        monkey.setattr(queue, 'add_requests', AsyncMock(side_effect=RuntimeError('Queue unavailable')))
+        with pytest.raises(RuntimeError, match='Queue unavailable'):
+            await crawler.run(['https://a.placeholder.com'])
+
+    await crawler.run(['https://a.placeholder.com'])
+    assert handled_urls == ['https://a.placeholder.com']
+
+
+async def test_failed_run_keeps_pending_requests_for_the_retry(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Requests left pending by a failed run survive into the retry even when an earlier run completed."""
+    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'])
+    assert handled_urls == ['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()
+    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."""
+    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'])
+    assert handled_urls == ['https://a.placeholder.com']
+
+    with monkeypatch.context() as monkey:
+        monkey.setattr(crawler, '_save_crawler_state', AsyncMock(side_effect=RuntimeError('Key-value store down')))
+        with pytest.raises(RuntimeError, match='Key-value store down'):
+            await crawler.run()
+
+    await queue.add_request('https://b.placeholder.com')
+    await crawler.run()
+    assert handled_urls == ['https://a.placeholder.com', 'https://b.placeholder.com']
+
+
 async def test_processes_requests_from_request_source_tandem() -> None:
     request_queue = await RequestQueue.open()
     await request_queue.add_requests(