From 6768357dcae215ecb9781441f3d8371eb7c6e49b Mon Sep 17 00:00:00 2001
From: axisrow <axisrow@users.noreply.github.com>
Date: Mon, 31 Aug 2026 10:30:53 +0800
Subject: [PATCH 01/10] fix: keep collecting after empty top retry timeout

---
 src/wordstat/collector.py     | 23 +++++++++--
 tests/test_collector_batch.py | 76 +++++++++++++++++++++++++++++++++++
 2 files changed, 95 insertions(+), 4 deletions(-)

diff --git a/src/wordstat/collector.py b/src/wordstat/collector.py
index 3f7d755..ea9bcb1 100644
--- a/src/wordstat/collector.py
+++ b/src/wordstat/collector.py
@@ -591,10 +591,25 @@ async def _collect_one(
                     if _should_retry_empty_export(view, dataset) or _is_untrustworthy_empty_export(view, dataset):
                         if self.empty_export_retry_seconds > 0:
                             await asyncio.sleep(self.empty_export_retry_seconds)
-                        source, escape_warning = await self._download_current_view(page, session, downloads_path)
-                        if escape_warning is not None:
-                            escaped_download_warnings.append(f"[{view.value}] {escape_warning}")
-                        dataset = parse_wordstat_csv(source, view)
+                        try:
+                            retry_source, retry_escape_warning = await self._download_current_view(
+                                page, session, downloads_path
+                            )
+                        except DownloadTimeoutError:
+                            # A repeated export can overwrite the original CSV
+                            # instead of creating a new path. In that case the
+                            # snapshot-based download wait times out even
+                            # though the first, legitimate empty export is
+                            # already available. Top exports are allowed to be
+                            # empty, so keep that result and continue with the
+                            # remaining views rather than failing the phrase.
+                            if not _should_retry_empty_export(view, dataset):
+                                raise
+                        else:
+                            source = retry_source
+                            if retry_escape_warning is not None:
+                                escaped_download_warnings.append(f"[{view.value}] {retry_escape_warning}")
+                            dataset = parse_wordstat_csv(source, view)
                     if _is_untrustworthy_empty_export(view, dataset):
                         raise InterfaceChangedError(
                             f"Wordstat returned an empty {view.value} CSV after a retry, but the page had "
diff --git a/tests/test_collector_batch.py b/tests/test_collector_batch.py
index 2ef1c1b..f26d4d6 100644
--- a/tests/test_collector_batch.py
+++ b/tests/test_collector_batch.py
@@ -16,6 +16,7 @@
 from wordstat.errors import (
     AuthenticationRequiredError,
     DownloadTimeoutError,
+    InterfaceChangedError,
     InvalidRequestError,
     PhraseEntryError,
     ResumeMismatchError,
@@ -521,6 +522,81 @@ async def fake_download(self, page, session, dl_path):
     assert all(export.row_count == 1 for export in result.manifest.exports[2:])
 
 
+def test_collect_one_keeps_empty_top_export_when_retry_times_out(monkeypatch, tmp_path):
+    """A failed top-view retry must fall back to the first empty CSV.
+
+    Chrome can overwrite the first download with the same filename, leaving
+    the snapshot-based downloader with no new path to observe. That timeout
+    must not prevent the other views from being collected.
+    """
+
+    _patch_common(monkeypatch)
+    downloads_path = tmp_path / "downloads"
+    downloads_path.mkdir()
+
+    async def fake_select_view(self, page, selector, view):
+        pass
+
+    download_count = 0
+
+    async def fake_download(self, page, session, dl_path):
+        nonlocal download_count
+        download_count += 1
+        if download_count == 1:
+            source = dl_path / "export.csv"
+            _write_empty_view_csv(source)
+            return source, None
+        if download_count == 2:
+            raise DownloadTimeoutError("simulated same-name retry timeout")
+        source = dl_path / f"export-{download_count}.csv"
+        _write_view_csv(source, "тест")
+        return source, None
+
+    monkeypatch.setattr(WordstatCollector, "_select_view", fake_select_view)
+    monkeypatch.setattr(WordstatCollector, "_download_current_view", fake_download)
+
+    collector = WordstatCollector("cdp", tmp_path, settling_seconds=0, empty_export_retry_seconds=0)
+    result = asyncio.run(
+        collector._collect_one(
+            _FakePage(), _FakeSession(), downloads_path, "тест", "Россия", set_region=False
+        )
+    )
+
+    assert download_count == 5
+    assert result.view_errors == {}
+    assert [export.row_count for export in result.manifest.exports] == [0, 1, 1, 1]
+    assert result.manifest.empty_views == [WordstatView.TOP_POPULAR]
+
+
+def test_collect_one_does_not_swallow_non_timeout_top_retry_error(monkeypatch, tmp_path):
+    """Only the known same-name timeout is recoverable for a top export."""
+
+    _patch_common(monkeypatch)
+    downloads_path = tmp_path / "downloads"
+    downloads_path.mkdir()
+
+    async def fake_select_view(self, page, selector, view):
+        pass
+
+    async def fake_download(self, page, session, dl_path):
+        source = dl_path / "export.csv"
+        if source.exists():
+            raise InterfaceChangedError("simulated changed export control")
+        _write_empty_view_csv(source)
+        return source, None
+
+    monkeypatch.setattr(WordstatCollector, "_select_view", fake_select_view)
+    monkeypatch.setattr(WordstatCollector, "_download_current_view", fake_download)
+
+    collector = WordstatCollector("cdp", tmp_path, settling_seconds=0, empty_export_retry_seconds=0)
+    with pytest.raises(InterfaceChangedError, match="changed export control"):
+        asyncio.run(
+            collector._collect_one(
+                _FakePage(), _FakeSession(), downloads_path, "тест", "Россия", set_region=False
+            )
+        )
+
+
 def test_collect_one_keeps_dynamics_empty_export_fail_closed(monkeypatch, tmp_path):
     """The top-view retry must not change DYNAMICS' existing safety path."""
 

From d47eba3c398df37096b6ad8aa7ca341754f3d587 Mon Sep 17 00:00:00 2001
From: axisrow <axisrow@users.noreply.github.com>
Date: Mon, 31 Aug 2026 10:34:36 +0800
Subject: [PATCH 02/10] test: preserve dynamics retry fail-closed guard

---
 tests/test_collector_batch.py | 42 +++++++++++++++++++++++++++++++++++
 1 file changed, 42 insertions(+)

diff --git a/tests/test_collector_batch.py b/tests/test_collector_batch.py
index f26d4d6..2801cd8 100644
--- a/tests/test_collector_batch.py
+++ b/tests/test_collector_batch.py
@@ -597,6 +597,48 @@ async def fake_download(self, page, session, dl_path):
         )
 
 
+def test_collect_one_does_not_swallow_dynamics_retry_timeout(monkeypatch, tmp_path):
+    """An empty dynamics retry timeout must preserve fail-closed behavior."""
+
+    _patch_common(monkeypatch)
+    downloads_path = tmp_path / "downloads"
+    downloads_path.mkdir()
+
+    async def fake_select_view(self, page, selector, view):
+        pass
+
+    download_count = 0
+
+    async def fake_download(self, page, session, dl_path):
+        nonlocal download_count
+        download_count += 1
+        source = dl_path / f"export-{download_count}.csv"
+        if download_count == 3:
+            _write_empty_view_csv(source)
+            return source, None
+        if download_count == 4:
+            raise DownloadTimeoutError("simulated dynamics retry timeout")
+        _write_view_csv(source, "тест")
+        return source, None
+
+    monkeypatch.setattr(WordstatCollector, "_select_view", fake_select_view)
+    monkeypatch.setattr(WordstatCollector, "_download_current_view", fake_download)
+
+    collector = WordstatCollector("cdp", tmp_path, settling_seconds=0, empty_export_retry_seconds=0)
+    result = asyncio.run(
+        collector._collect_one(
+            _FakePage(), _FakeSession(), downloads_path, "тест", "Россия", set_region=False
+        )
+    )
+
+    assert result.manifest.missing_views == [WordstatView.DYNAMICS, WordstatView.REGIONS]
+    assert result.manifest.exports[0].view is WordstatView.TOP_POPULAR
+    assert result.manifest.exports[1].view is WordstatView.TOP_RELATED
+    assert result.view_errors[WordstatView.DYNAMICS] == (
+        "DownloadTimeoutError: simulated dynamics retry timeout"
+    )
+
+
 def test_collect_one_keeps_dynamics_empty_export_fail_closed(monkeypatch, tmp_path):
     """The top-view retry must not change DYNAMICS' existing safety path."""
 

From 7ec6037dc0e575fda1b39cc20c39f1bdbaf24fc2 Mon Sep 17 00:00:00 2001
From: axisrow <axisrow@users.noreply.github.com>
Date: Mon, 31 Aug 2026 10:44:51 +0800
Subject: [PATCH 03/10] fix: distinguish missing download path timeout

---
 src/wordstat/collector.py     | 5 +++--
 src/wordstat/errors.py        | 8 ++++++++
 tests/test_collector_batch.py | 7 ++++---
 3 files changed, 15 insertions(+), 5 deletions(-)

diff --git a/src/wordstat/collector.py b/src/wordstat/collector.py
index ea9bcb1..5774dac 100644
--- a/src/wordstat/collector.py
+++ b/src/wordstat/collector.py
@@ -17,6 +17,7 @@
 from wordstat.errors import (
     AuthenticationRequiredError,
     DownloadEscapedError,
+    DownloadNoNewPathError,
     DownloadTimeoutError,
     InterfaceChangedError,
     InvalidRequestError,
@@ -595,7 +596,7 @@ async def _collect_one(
                             retry_source, retry_escape_warning = await self._download_current_view(
                                 page, session, downloads_path
                             )
-                        except DownloadTimeoutError:
+                        except DownloadNoNewPathError:
                             # A repeated export can overwrite the original CSV
                             # instead of creating a new path. In that case the
                             # snapshot-based download wait times out even
@@ -1294,7 +1295,7 @@ async def _download_current_view(
                     "automatically. Move it manually if it belongs to this run."
                 )
             await asyncio.sleep(0.25)
-        raise DownloadTimeoutError("Wordstat did not produce a CSV before the download timeout")
+        raise DownloadNoNewPathError("Wordstat did not produce a new CSV before the download timeout")
 
     async def _click(self, page, selector: str) -> None:
         result = await page.evaluate(
diff --git a/src/wordstat/errors.py b/src/wordstat/errors.py
index df500ff..c6b38d8 100644
--- a/src/wordstat/errors.py
+++ b/src/wordstat/errors.py
@@ -29,6 +29,14 @@ class DownloadTimeoutError(WordstatError):
     """The UI accepted an export request but did not produce a CSV file."""
 
 
+class DownloadNoNewPathError(DownloadTimeoutError):
+    """The export completed without exposing a new file path.
+
+    This is distinct from other download failures (notably multiple new CSVs)
+    because Chrome can handle a repeated export by reusing the existing path.
+    """
+
+
 class DownloadEscapedError(WordstatError):
     """Chrome reported a download outside the run's own downloads directory.
 
diff --git a/tests/test_collector_batch.py b/tests/test_collector_batch.py
index 2801cd8..6a3f4a7 100644
--- a/tests/test_collector_batch.py
+++ b/tests/test_collector_batch.py
@@ -15,6 +15,7 @@
 from wordstat.collector import WordstatCollector
 from wordstat.errors import (
     AuthenticationRequiredError,
+    DownloadNoNewPathError,
     DownloadTimeoutError,
     InterfaceChangedError,
     InvalidRequestError,
@@ -547,7 +548,7 @@ async def fake_download(self, page, session, dl_path):
             _write_empty_view_csv(source)
             return source, None
         if download_count == 2:
-            raise DownloadTimeoutError("simulated same-name retry timeout")
+            raise DownloadNoNewPathError("simulated same-name retry timeout")
         source = dl_path / f"export-{download_count}.csv"
         _write_view_csv(source, "тест")
         return source, None
@@ -617,7 +618,7 @@ async def fake_download(self, page, session, dl_path):
             _write_empty_view_csv(source)
             return source, None
         if download_count == 4:
-            raise DownloadTimeoutError("simulated dynamics retry timeout")
+            raise DownloadNoNewPathError("simulated dynamics retry timeout")
         _write_view_csv(source, "тест")
         return source, None
 
@@ -635,7 +636,7 @@ async def fake_download(self, page, session, dl_path):
     assert result.manifest.exports[0].view is WordstatView.TOP_POPULAR
     assert result.manifest.exports[1].view is WordstatView.TOP_RELATED
     assert result.view_errors[WordstatView.DYNAMICS] == (
-        "DownloadTimeoutError: simulated dynamics retry timeout"
+        "DownloadNoNewPathError: simulated dynamics retry timeout"
     )
 
 

From 3dd6f8dde01ec10bb5e2c7494dfa0f349a623012 Mon Sep 17 00:00:00 2001
From: axisrow <axisrow@users.noreply.github.com>
Date: Mon, 31 Aug 2026 10:48:01 +0800
Subject: [PATCH 04/10] test: preserve ambiguous download failures

---
 tests/test_collector_batch.py | 29 +++++++++++++++++++++++++++++
 1 file changed, 29 insertions(+)

diff --git a/tests/test_collector_batch.py b/tests/test_collector_batch.py
index 6a3f4a7..93d6c5a 100644
--- a/tests/test_collector_batch.py
+++ b/tests/test_collector_batch.py
@@ -598,6 +598,35 @@ async def fake_download(self, page, session, dl_path):
         )
 
 
+def test_collect_one_does_not_swallow_ambiguous_top_retry_timeout(monkeypatch, tmp_path):
+    """A multi-download timeout is not the recoverable no-new-path signal."""
+
+    _patch_common(monkeypatch)
+    downloads_path = tmp_path / "downloads"
+    downloads_path.mkdir()
+
+    async def fake_select_view(self, page, selector, view):
+        pass
+
+    async def fake_download(self, page, session, dl_path):
+        source = dl_path / "export.csv"
+        if source.exists():
+            raise DownloadTimeoutError("Wordstat produced more than one new CSV for a single export")
+        _write_empty_view_csv(source)
+        return source, None
+
+    monkeypatch.setattr(WordstatCollector, "_select_view", fake_select_view)
+    monkeypatch.setattr(WordstatCollector, "_download_current_view", fake_download)
+
+    collector = WordstatCollector("cdp", tmp_path, settling_seconds=0, empty_export_retry_seconds=0)
+    with pytest.raises(DownloadTimeoutError, match="more than one new CSV"):
+        asyncio.run(
+            collector._collect_one(
+                _FakePage(), _FakeSession(), downloads_path, "тест", "Россия", set_region=False
+            )
+        )
+
+
 def test_collect_one_does_not_swallow_dynamics_retry_timeout(monkeypatch, tmp_path):
     """An empty dynamics retry timeout must preserve fail-closed behavior."""
 

From 4257ea58cddf93d24321b586d222809ba81bd785 Mon Sep 17 00:00:00 2001
From: axisrow <axisrow@users.noreply.github.com>
Date: Mon, 31 Aug 2026 10:51:46 +0800
Subject: [PATCH 05/10] fix: preserve original export on retry fallback

---
 src/wordstat/collector.py     | 17 ++++++++++++++
 tests/test_collector_batch.py | 42 +++++++++++++++++++++++++++++++++++
 2 files changed, 59 insertions(+)

diff --git a/src/wordstat/collector.py b/src/wordstat/collector.py
index 5774dac..7c8fac1 100644
--- a/src/wordstat/collector.py
+++ b/src/wordstat/collector.py
@@ -592,6 +592,18 @@ async def _collect_one(
                     if _should_retry_empty_export(view, dataset) or _is_untrustworthy_empty_export(view, dataset):
                         if self.empty_export_retry_seconds > 0:
                             await asyncio.sleep(self.empty_export_retry_seconds)
+                        # The retry may reuse the same path and overwrite the
+                        # first file. Keep the original beside the temporary
+                        # downloads directory so a no-new-path fallback cannot
+                        # pair the old parsed dataset with new raw contents.
+                        with tempfile.NamedTemporaryFile(
+                            prefix=f".{view.value}-retry-",
+                            suffix=".csv",
+                            dir=self.output_root,
+                            delete=False,
+                        ) as backup_file:
+                            retry_backup = Path(backup_file.name)
+                        shutil.copy2(source, retry_backup)
                         try:
                             retry_source, retry_escape_warning = await self._download_current_view(
                                 page, session, downloads_path
@@ -606,11 +618,16 @@ async def _collect_one(
                             # remaining views rather than failing the phrase.
                             if not _should_retry_empty_export(view, dataset):
                                 raise
+                            source = retry_backup
+                            retry_backup = None
                         else:
                             source = retry_source
                             if retry_escape_warning is not None:
                                 escaped_download_warnings.append(f"[{view.value}] {retry_escape_warning}")
                             dataset = parse_wordstat_csv(source, view)
+                        finally:
+                            if retry_backup is not None:
+                                retry_backup.unlink(missing_ok=True)
                     if _is_untrustworthy_empty_export(view, dataset):
                         raise InterfaceChangedError(
                             f"Wordstat returned an empty {view.value} CSV after a retry, but the page had "
diff --git a/tests/test_collector_batch.py b/tests/test_collector_batch.py
index 93d6c5a..1cd606a 100644
--- a/tests/test_collector_batch.py
+++ b/tests/test_collector_batch.py
@@ -569,6 +569,48 @@ async def fake_download(self, page, session, dl_path):
     assert result.manifest.empty_views == [WordstatView.TOP_POPULAR]
 
 
+def test_collect_one_preserves_first_csv_when_retry_overwrites_path(monkeypatch, tmp_path):
+    """Fallback parquet and raw CSV must describe the same first export."""
+
+    _patch_common(monkeypatch)
+    downloads_path = tmp_path / "downloads"
+    downloads_path.mkdir()
+
+    async def fake_select_view(self, page, selector, view):
+        pass
+
+    download_count = 0
+
+    async def fake_download(self, page, session, dl_path):
+        nonlocal download_count
+        download_count += 1
+        source = dl_path / "export.csv"
+        if download_count == 1:
+            _write_empty_view_csv(source)
+            return source, None
+        if download_count == 2:
+            _write_view_csv(source, "тест")
+            raise DownloadNoNewPathError("simulated overwritten-path timeout")
+        source = dl_path / f"export-{download_count}.csv"
+        _write_view_csv(source, "тест")
+        return source, None
+
+    monkeypatch.setattr(WordstatCollector, "_select_view", fake_select_view)
+    monkeypatch.setattr(WordstatCollector, "_download_current_view", fake_download)
+
+    collector = WordstatCollector(
+        "cdp", tmp_path, keep_raw=True, settling_seconds=0, empty_export_retry_seconds=0
+    )
+    result = asyncio.run(
+        collector._collect_one(
+            _FakePage(), _FakeSession(), downloads_path, "тест", "Россия", set_region=False
+        )
+    )
+
+    assert result.manifest.exports[0].row_count == 0
+    assert (result.run_directory / "top_popular.csv").read_text(encoding="cp1251") == "Запрос;Показов\n"
+
+
 def test_collect_one_does_not_swallow_non_timeout_top_retry_error(monkeypatch, tmp_path):
     """Only the known same-name timeout is recoverable for a top export."""
 

From acc6aa6e2b770756a4cdfa0dc50bee06ef806fed Mon Sep 17 00:00:00 2001
From: axisrow <axisrow@users.noreply.github.com>
Date: Mon, 31 Aug 2026 10:54:23 +0800
Subject: [PATCH 06/10] test: ensure retry backups are cleaned up

---
 tests/test_collector_batch.py | 1 +
 1 file changed, 1 insertion(+)

diff --git a/tests/test_collector_batch.py b/tests/test_collector_batch.py
index 1cd606a..5399d63 100644
--- a/tests/test_collector_batch.py
+++ b/tests/test_collector_batch.py
@@ -475,6 +475,7 @@ async def fake_download(self, page, session, dl_path):
     assert download_count == 6  # top views each needed one content-based retry
     assert {export.view for export in result.manifest.exports} == set(WordstatView)
     assert all(export.row_count == 1 for export in result.manifest.exports)
+    assert not list(tmp_path.glob("*retry*"))
 
 
 def test_collect_one_accepts_persistent_empty_top_exports_after_retry(monkeypatch, tmp_path):

From b0526a5fcb44c7a1489492b5ed753f01c466e337 Mon Sep 17 00:00:00 2001
From: axisrow <axisrow@users.noreply.github.com>
Date: Mon, 31 Aug 2026 10:57:01 +0800
Subject: [PATCH 07/10] fix: clean retry backup on copy failure

---
 src/wordstat/collector.py     | 59 ++++++++++++++++++-----------------
 tests/test_collector_batch.py | 33 ++++++++++++++++++++
 2 files changed, 64 insertions(+), 28 deletions(-)

diff --git a/src/wordstat/collector.py b/src/wordstat/collector.py
index 7c8fac1..2234bf0 100644
--- a/src/wordstat/collector.py
+++ b/src/wordstat/collector.py
@@ -596,35 +596,38 @@ async def _collect_one(
                         # first file. Keep the original beside the temporary
                         # downloads directory so a no-new-path fallback cannot
                         # pair the old parsed dataset with new raw contents.
-                        with tempfile.NamedTemporaryFile(
-                            prefix=f".{view.value}-retry-",
-                            suffix=".csv",
-                            dir=self.output_root,
-                            delete=False,
-                        ) as backup_file:
-                            retry_backup = Path(backup_file.name)
-                        shutil.copy2(source, retry_backup)
+                        retry_backup: Path | None = None
                         try:
-                            retry_source, retry_escape_warning = await self._download_current_view(
-                                page, session, downloads_path
-                            )
-                        except DownloadNoNewPathError:
-                            # A repeated export can overwrite the original CSV
-                            # instead of creating a new path. In that case the
-                            # snapshot-based download wait times out even
-                            # though the first, legitimate empty export is
-                            # already available. Top exports are allowed to be
-                            # empty, so keep that result and continue with the
-                            # remaining views rather than failing the phrase.
-                            if not _should_retry_empty_export(view, dataset):
-                                raise
-                            source = retry_backup
-                            retry_backup = None
-                        else:
-                            source = retry_source
-                            if retry_escape_warning is not None:
-                                escaped_download_warnings.append(f"[{view.value}] {retry_escape_warning}")
-                            dataset = parse_wordstat_csv(source, view)
+                            with tempfile.NamedTemporaryFile(
+                                prefix=f".{view.value}-retry-",
+                                suffix=".csv",
+                                dir=self.output_root,
+                                delete=False,
+                            ) as backup_file:
+                                retry_backup = Path(backup_file.name)
+                            shutil.copy2(source, retry_backup)
+                            try:
+                                retry_source, retry_escape_warning = await self._download_current_view(
+                                    page, session, downloads_path
+                                )
+                            except DownloadNoNewPathError:
+                                # A repeated export can overwrite the original
+                                # CSV instead of creating a new path. In that
+                                # case the snapshot-based download wait times
+                                # out even though the first, legitimate empty
+                                # export is already available. Top exports are
+                                # allowed to be empty, so keep that result and
+                                # continue with the remaining views rather
+                                # than failing the phrase.
+                                if not _should_retry_empty_export(view, dataset):
+                                    raise
+                                source = retry_backup
+                                retry_backup = None
+                            else:
+                                source = retry_source
+                                if retry_escape_warning is not None:
+                                    escaped_download_warnings.append(f"[{view.value}] {retry_escape_warning}")
+                                dataset = parse_wordstat_csv(source, view)
                         finally:
                             if retry_backup is not None:
                                 retry_backup.unlink(missing_ok=True)
diff --git a/tests/test_collector_batch.py b/tests/test_collector_batch.py
index 5399d63..c4735e7 100644
--- a/tests/test_collector_batch.py
+++ b/tests/test_collector_batch.py
@@ -612,6 +612,39 @@ async def fake_download(self, page, session, dl_path):
     assert (result.run_directory / "top_popular.csv").read_text(encoding="cp1251") == "Запрос;Показов\n"
 
 
+def test_collect_one_cleans_retry_backup_when_copy_fails(monkeypatch, tmp_path):
+    """A failed backup copy must not leave a temporary file behind."""
+
+    _patch_common(monkeypatch)
+    downloads_path = tmp_path / "downloads"
+    downloads_path.mkdir()
+
+    async def fake_select_view(self, page, selector, view):
+        pass
+
+    async def fake_download(self, page, session, dl_path):
+        source = dl_path / "export.csv"
+        _write_empty_view_csv(source)
+        return source, None
+
+    def failing_copy(source, destination):
+        raise OSError("simulated backup filesystem failure")
+
+    monkeypatch.setattr(WordstatCollector, "_select_view", fake_select_view)
+    monkeypatch.setattr(WordstatCollector, "_download_current_view", fake_download)
+    monkeypatch.setattr(collector_module.shutil, "copy2", failing_copy)
+
+    collector = WordstatCollector("cdp", tmp_path, settling_seconds=0, empty_export_retry_seconds=0)
+    with pytest.raises(OSError, match="backup filesystem failure"):
+        asyncio.run(
+            collector._collect_one(
+                _FakePage(), _FakeSession(), downloads_path, "тест", "Россия", set_region=False
+            )
+        )
+
+    assert not list(tmp_path.glob("*retry*"))
+
+
 def test_collect_one_does_not_swallow_non_timeout_top_retry_error(monkeypatch, tmp_path):
     """Only the known same-name timeout is recoverable for a top export."""
 

From eee127d3fc07c7e375d11c4fcba15934c1d4b87d Mon Sep 17 00:00:00 2001
From: axisrow <axisrow@users.noreply.github.com>
Date: Mon, 31 Aug 2026 11:07:16 +0800
Subject: [PATCH 08/10] refactor: isolate empty export retry

---
 src/wordstat/collector.py | 112 ++++++++++++++++++++++++--------------
 1 file changed, 71 insertions(+), 41 deletions(-)

diff --git a/src/wordstat/collector.py b/src/wordstat/collector.py
index 2234bf0..ceb312e 100644
--- a/src/wordstat/collector.py
+++ b/src/wordstat/collector.py
@@ -7,8 +7,10 @@
 import tempfile
 import time
 from collections.abc import Callable
+from dataclasses import dataclass
 from datetime import UTC, date, datetime, timedelta
 from pathlib import Path
+from typing import Literal
 
 from browser_use.browser import BrowserSession
 
@@ -167,6 +169,15 @@ def _without_traceback(error: Exception) -> Exception:
     return error.with_traceback(None)
 
 
+@dataclass(frozen=True)
+class _RetryExportResult:
+    """A parsed export whose path and contents are guaranteed to match."""
+
+    source: Path
+    dataset: CsvDataset
+    kind: Literal["success", "no-new-path"]
+
+
 class WordstatCollector:
     """Export four Wordstat reports through an already authenticated CDP session."""
 
@@ -590,47 +601,16 @@ async def _collect_one(
                     # separate from the fail-closed predicate below.
                     dataset = parse_wordstat_csv(source, view)
                     if _should_retry_empty_export(view, dataset) or _is_untrustworthy_empty_export(view, dataset):
-                        if self.empty_export_retry_seconds > 0:
-                            await asyncio.sleep(self.empty_export_retry_seconds)
-                        # The retry may reuse the same path and overwrite the
-                        # first file. Keep the original beside the temporary
-                        # downloads directory so a no-new-path fallback cannot
-                        # pair the old parsed dataset with new raw contents.
-                        retry_backup: Path | None = None
-                        try:
-                            with tempfile.NamedTemporaryFile(
-                                prefix=f".{view.value}-retry-",
-                                suffix=".csv",
-                                dir=self.output_root,
-                                delete=False,
-                            ) as backup_file:
-                                retry_backup = Path(backup_file.name)
-                            shutil.copy2(source, retry_backup)
-                            try:
-                                retry_source, retry_escape_warning = await self._download_current_view(
-                                    page, session, downloads_path
-                                )
-                            except DownloadNoNewPathError:
-                                # A repeated export can overwrite the original
-                                # CSV instead of creating a new path. In that
-                                # case the snapshot-based download wait times
-                                # out even though the first, legitimate empty
-                                # export is already available. Top exports are
-                                # allowed to be empty, so keep that result and
-                                # continue with the remaining views rather
-                                # than failing the phrase.
-                                if not _should_retry_empty_export(view, dataset):
-                                    raise
-                                source = retry_backup
-                                retry_backup = None
-                            else:
-                                source = retry_source
-                                if retry_escape_warning is not None:
-                                    escaped_download_warnings.append(f"[{view.value}] {retry_escape_warning}")
-                                dataset = parse_wordstat_csv(source, view)
-                        finally:
-                            if retry_backup is not None:
-                                retry_backup.unlink(missing_ok=True)
+                        retry = await self._retry_empty_export(
+                            page,
+                            session,
+                            downloads_path,
+                            source,
+                            dataset,
+                            view,
+                            escaped_download_warnings,
+                        )
+                        source, dataset = retry.source, retry.dataset
                     if _is_untrustworthy_empty_export(view, dataset):
                         raise InterfaceChangedError(
                             f"Wordstat returned an empty {view.value} CSV after a retry, but the page had "
@@ -1217,6 +1197,56 @@ async def _table_snapshot(self, page) -> str | None:
             f"() => document.querySelector({json.dumps(TABLE_ROW_SELECTOR)})?.textContent ?? null"
         )
 
+    async def _retry_empty_export(
+        self,
+        page,
+        session: BrowserSession,
+        downloads_path: Path,
+        source: Path,
+        dataset: CsvDataset,
+        view: WordstatView,
+        escaped_download_warnings: list[str],
+    ) -> _RetryExportResult:
+        """Retry an empty export and return a matching path/dataset pair.
+
+        A no-new-path timeout can mean Chrome reused and overwrote ``source``.
+        The original is backed up before the retry; on that specific signal,
+        restore it to the original path before returning. All other failures
+        propagate unchanged. The single outer ``finally`` owns the temporary
+        file through creation, copy, retry, restore, and cleanup.
+        """
+        if self.empty_export_retry_seconds > 0:
+            await asyncio.sleep(self.empty_export_retry_seconds)
+        retry_backup: Path | None = None
+        try:
+            with tempfile.NamedTemporaryFile(
+                prefix=f".{view.value}-retry-",
+                suffix=".csv",
+                dir=self.output_root,
+                delete=False,
+            ) as backup_file:
+                retry_backup = Path(backup_file.name)
+            shutil.copy2(source, retry_backup)
+            try:
+                retry_source, retry_escape_warning = await self._download_current_view(
+                    page, session, downloads_path
+                )
+            except DownloadNoNewPathError:
+                if not _should_retry_empty_export(view, dataset):
+                    raise
+                shutil.copy2(retry_backup, source)
+                return _RetryExportResult(source=source, dataset=dataset, kind="no-new-path")
+            if retry_escape_warning is not None:
+                escaped_download_warnings.append(f"[{view.value}] {retry_escape_warning}")
+            return _RetryExportResult(
+                source=retry_source,
+                dataset=parse_wordstat_csv(retry_source, view),
+                kind="success",
+            )
+        finally:
+            if retry_backup is not None:
+                retry_backup.unlink(missing_ok=True)
+
     async def _download_current_view(
         self, page, session: BrowserSession, downloads_path: Path
     ) -> tuple[Path, str | None]:

From b39d6dfb53a9f0341d647ebd20b9d0c51617aad9 Mon Sep 17 00:00:00 2001
From: axisrow <axisrow@users.noreply.github.com>
Date: Mon, 31 Aug 2026 11:09:23 +0800
Subject: [PATCH 09/10] refactor: remove unused retry result kind

---
 src/wordstat/collector.py | 5 +----
 1 file changed, 1 insertion(+), 4 deletions(-)

diff --git a/src/wordstat/collector.py b/src/wordstat/collector.py
index ceb312e..cb23f9c 100644
--- a/src/wordstat/collector.py
+++ b/src/wordstat/collector.py
@@ -10,7 +10,6 @@
 from dataclasses import dataclass
 from datetime import UTC, date, datetime, timedelta
 from pathlib import Path
-from typing import Literal
 
 from browser_use.browser import BrowserSession
 
@@ -175,7 +174,6 @@ class _RetryExportResult:
 
     source: Path
     dataset: CsvDataset
-    kind: Literal["success", "no-new-path"]
 
 
 class WordstatCollector:
@@ -1235,13 +1233,12 @@ async def _retry_empty_export(
                 if not _should_retry_empty_export(view, dataset):
                     raise
                 shutil.copy2(retry_backup, source)
-                return _RetryExportResult(source=source, dataset=dataset, kind="no-new-path")
+                return _RetryExportResult(source=source, dataset=dataset)
             if retry_escape_warning is not None:
                 escaped_download_warnings.append(f"[{view.value}] {retry_escape_warning}")
             return _RetryExportResult(
                 source=retry_source,
                 dataset=parse_wordstat_csv(retry_source, view),
-                kind="success",
             )
         finally:
             if retry_backup is not None:

From 46f917058004e1e8739e0220c173d3e8339882dd Mon Sep 17 00:00:00 2001
From: axisrow <axisrow@users.noreply.github.com>
Date: Mon, 31 Aug 2026 11:12:09 +0800
Subject: [PATCH 10/10] test: verify raw export matches retry parquet

---
 tests/test_collector_batch.py | 9 ++++++++-
 1 file changed, 8 insertions(+), 1 deletion(-)

diff --git a/tests/test_collector_batch.py b/tests/test_collector_batch.py
index c4735e7..7bcf4c0 100644
--- a/tests/test_collector_batch.py
+++ b/tests/test_collector_batch.py
@@ -9,6 +9,7 @@
 import asyncio
 from datetime import UTC, datetime
 
+import pyarrow.parquet as parquet
 import pytest
 
 import wordstat.collector as collector_module
@@ -465,7 +466,9 @@ async def fake_download(self, page, session, dl_path):
     monkeypatch.setattr(WordstatCollector, "_select_view", fake_select_view)
     monkeypatch.setattr(WordstatCollector, "_download_current_view", fake_download)
 
-    collector = WordstatCollector("cdp", tmp_path, settling_seconds=0, empty_export_retry_seconds=0)
+    collector = WordstatCollector(
+        "cdp", tmp_path, keep_raw=True, settling_seconds=0, empty_export_retry_seconds=0
+    )
     result = asyncio.run(
         collector._collect_one(
             _FakePage(), _FakeSession(), downloads_path, "тест", "Россия", set_region=False
@@ -476,6 +479,10 @@ async def fake_download(self, page, session, dl_path):
     assert {export.view for export in result.manifest.exports} == set(WordstatView)
     assert all(export.row_count == 1 for export in result.manifest.exports)
     assert not list(tmp_path.glob("*retry*"))
+    top_export = next(export for export in result.manifest.exports if export.view is WordstatView.TOP_POPULAR)
+    assert top_export.raw_file == "top_popular.csv"
+    assert "январь 2024;100" in (result.run_directory / top_export.raw_file).read_text(encoding="cp1251")
+    assert parquet.read_table(result.run_directory / top_export.file).num_rows == top_export.row_count
 
 
 def test_collect_one_accepts_persistent_empty_top_exports_after_retry(monkeypatch, tmp_path):