fix: detect missing files when resource path coincides with datapath

Single-resource ArchiveDownloader returns dataset.datapath as its path,
which always exists (holds .state.json), so the "COMPLETE but files
missing" and "adopt pre-existing" checks were fooled into skipping
downloads.

- Add Resource.files_present() with FolderResource override that checks
  for actual content beyond metadata files when path == datapath
- Fix _move_path to merge directory contents instead of nesting when
  destination already exists
- Add 5 regression tests covering re-download and transient scenarios

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: bpiwowar

src/datamaestro/definitions.py

@@ -125,8 +125,24 @@ def _delete_path(path: Path) -> None:
 
 
 def _move_path(src: Path, dst: Path) -> None:
-    """Move a file or directory from src to dst."""
-    if src.exists():
+    """Move a file or directory from src to dst.
+
+    When dst is an existing directory, merges src contents into dst
+    (shutil.move would nest src inside dst, which is not what we want).
+    """
+    if not src.exists():
+        return
+
+    if dst.is_dir() and src.is_dir():
+        # Merge contents of src into dst
+        for child in src.iterdir():
+            child_dst = dst / child.name
+            if child_dst.exists():
+                _delete_path(child_dst)
+            shutil.move(str(child), str(child_dst))
+        # Remove now-empty src
+        shutil.rmtree(src, ignore_errors=True)
+    else:
         dst.parent.mkdir(parents=True, exist_ok=True)
         shutil.move(str(src), str(dst))
 
@@ -402,7 +418,7 @@ def _download_locked(self, force, ResourceState):
 
             if current_state == ResourceState.COMPLETE and not force:
                 # Verify files are actually present on disk
-                if resource.has_files() and not resource.path.exists():
+                if resource.has_files() and not resource.files_present():
                     logging.warning(
                         "Resource %s marked COMPLETE but files "
                         "missing at %s — re-downloading",
@@ -420,7 +436,7 @@ def _download_locked(self, force, ResourceState):
                 current_state == ResourceState.NONE
                 and not force
                 and resource.has_files()
-                and resource.path.exists()
+                and resource.files_present()
             ):
                 logging.info(
                     "Resource %s already exists at %s — marking COMPLETE",

src/datamaestro/download/__init__.py

@@ -346,6 +346,16 @@ def has_files(self) -> bool:
         """
         return True
 
+    def files_present(self) -> bool:
+        """Whether this resource's output files actually exist on disk.
+
+        Used by the download orchestrator to detect COMPLETE resources
+        whose files have been removed. Default checks ``path.exists()``.
+        Subclasses may override for more thorough checks (e.g., when
+        ``path`` coincides with ``dataset.datapath``).
+        """
+        return self.path.exists()
+
     # Backward compat alias
     def hasfiles(self) -> bool:
         """Deprecated: use has_files() instead."""
@@ -554,6 +564,8 @@ class FolderResource(Resource):
     the given destination (which is ``self.transient_path``).
     """
 
+    _METADATA_FILES = {".state.json", ".state.json.tmp", ".state.lock"}
+
     @property
     def path(self) -> Path:
         """Final path to the produced directory.
@@ -562,6 +574,27 @@ def path(self) -> Path:
         """
         return self.dataset.datapath / self.name
 
+    def files_present(self) -> bool:
+        """Check that the directory exists and has content.
+
+        When ``path`` coincides with ``dataset.datapath`` (e.g.,
+        single-resource ArchiveDownloader), the directory always
+        exists because it holds metadata files (.state.json).
+        In that case, verify it contains at least one non-metadata entry.
+        """
+        p = self.path
+        if not p.exists():
+            return False
+
+        if p != self.dataset.datapath:
+            return True
+
+        # path == datapath: check for actual content beyond metadata
+        for child in p.iterdir():
+            if child.name not in self._METADATA_FILES:
+                return True
+        return False
+
     @property
     def transient_path(self) -> Path:
         """Temporary path for writing during download.

src/datamaestro/test/test_resource.py

@@ -1655,3 +1655,167 @@ class Squad(Base):
         ann = dataset_dec(base=Squad, url="http://test.com")
         dw = DatasetWrapper(ann, Squad)
         assert dw.id == "stanford.qa.squad"
+
+
+# ==== Redownload with folder-path == datapath ====
+
+
+class FolderAtDatapath(FolderResource):
+    """A FolderResource whose path collapses to dataset.datapath.
+
+    Simulates ArchiveDownloader with a single resource,
+    where ``path`` returns ``dataset.datapath`` instead of a subdirectory.
+    """
+
+    def __init__(self, **kw):
+        super().__init__(**kw)
+        self._download_called = False
+
+    @property
+    def path(self) -> Path:
+        return self.dataset.datapath
+
+    def _download(self, destination: Path):
+        destination.mkdir(parents=True, exist_ok=True)
+        (destination / "data.tsv").write_text("col1\tcol2\n")
+        self._download_called = True
+
+
+class TestRedownloadFolderAtDatapath:
+    """Regression: single-resource FolderResource whose path == datapath.
+
+    When resource.path is the same as dataset.datapath, the directory
+    always exists (it holds .state.json), so the "COMPLETE but files
+    missing" check must look deeper than just path.exists().
+    """
+
+    def test_redownload_when_content_missing(self, dataset):
+        """COMPLETE folder resource at datapath re-downloads when empty."""
+        r = FolderAtDatapath()
+        r.bind("DATA", dataset)
+
+        dataset.ordered_resources = [r]
+        _compute_dependents(dataset.resources)
+
+        # Simulate: state is COMPLETE but actual content is gone
+        # (datapath exists because .state.json is stored there)
+        dataset.datapath.mkdir(parents=True, exist_ok=True)
+        r.state = ResourceState.COMPLETE
+        assert r.path.exists()  # datapath exists (has .state.json)
+        assert not (r.path / "data.tsv").exists()  # but no data files
+
+        dataset.download()
+
+        assert r._download_called is True
+        assert r.state == ResourceState.COMPLETE
+        assert (r.path / "data.tsv").exists()
+
+    def test_no_redownload_when_content_present(self, dataset):
+        """COMPLETE folder resource at datapath is NOT re-downloaded."""
+        r = FolderAtDatapath()
+        r.bind("DATA", dataset)
+
+        dataset.ordered_resources = [r]
+        _compute_dependents(dataset.resources)
+
+        # Simulate: state is COMPLETE and actual content is present
+        dataset.datapath.mkdir(parents=True, exist_ok=True)
+        (dataset.datapath / "data.tsv").write_text("existing\n")
+        r.state = ResourceState.COMPLETE
+
+        dataset.download()
+
+        assert r._download_called is False
+        assert r.state == ResourceState.COMPLETE
+
+    def _make_transient_dag(self, dataset):
+        """Helper: transient source -> dependent file resource."""
+        source = DummyFolderResource(transient=True)
+        source.bind("SOURCE", dataset)
+
+        dependent = DummyFileResource("result.txt")
+        dependent._dependencies = [source]
+        dependent.bind("RESULT", dataset)
+
+        dataset.ordered_resources = [source, dependent]
+        _compute_dependents(dataset.resources)
+        return source, dependent
+
+    def test_transient_first_download(self, dataset):
+        """First prep: transient source is downloaded then cleaned up."""
+        source, dependent = self._make_transient_dag(dataset)
+
+        dataset.download()
+
+        # Both should have been downloaded
+        assert source._download_called is True
+        assert dependent._download_called is True
+        assert dependent.path.exists()
+        assert dependent.state == ResourceState.COMPLETE
+
+        # Transient source was cleaned up after dependent completed
+        assert source.state == ResourceState.NONE
+        assert not source.path.exists()
+
+    def test_transient_second_download_all_ok(self, dataset):
+        """Second prep: transient source is skipped (dependent is COMPLETE)."""
+        source, dependent = self._make_transient_dag(dataset)
+
+        # First download
+        dataset.download()
+        assert source._download_called is True
+        assert dependent._download_called is True
+
+        # Reset call flags
+        source._download_called = False
+        dependent._download_called = False
+
+        # Second download — everything is already done
+        dataset.download()
+
+        # Transient source skipped (all dependents COMPLETE)
+        assert source._download_called is False
+        # Dependent still COMPLETE, not re-downloaded
+        assert dependent._download_called is False
+        assert dependent.state == ResourceState.COMPLETE
+
+    def test_transient_second_download_after_first_failure(self, dataset):
+        """Second prep after first failed: transient re-downloads."""
+        source = DummyFolderResource(transient=True)
+        source.bind("SOURCE", dataset)
+
+        # Use a resource that will fail on first attempt
+        dependent = FailingResource("result.txt")
+        dependent._dependencies = [source]
+        dependent.bind("RESULT", dataset)
+
+        dataset.ordered_resources = [source, dependent]
+        _compute_dependents(dataset.resources)
+
+        # First download — source succeeds, dependent fails
+        result = dataset.download()
+        assert result is False
+        assert source._download_called is True
+        # Source is COMPLETE but NOT cleaned up (dependent not COMPLETE)
+        assert source.state == ResourceState.COMPLETE
+        assert dependent.state == ResourceState.NONE
+
+        # Now replace the failing resource with a working one for retry
+        source._download_called = False
+        good_dependent = DummyFileResource("result.txt")
+        good_dependent._dependencies = [source]
+        # Re-bind: remove old, add new
+        del dataset.resources["RESULT"]
+        good_dependent.bind("RESULT", dataset)
+        dataset.ordered_resources = [source, good_dependent]
+        _compute_dependents(dataset.resources)
+
+        dataset.download()
+
+        # Source is already COMPLETE, not re-downloaded
+        assert source._download_called is False
+        # Dependent should now succeed
+        assert good_dependent._download_called is True
+        assert good_dependent.state == ResourceState.COMPLETE
+        # Transient source cleaned up now that dependent is COMPLETE
+        assert source.state == ResourceState.NONE