Record partial deletion when archive is trashed but original removal fails

CHANGELOG.md

@@ -7,6 +7,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- Compress-before-trash where the archive reaches trash but the original
+  directory cannot be removed is now a distinct `partial` outcome on
+  `DeleteResult` (not success, not plain failure). The trashed archive is
+  recorded in metadata with `status = "partial"` (schema version 6) so
+  `history` and `doctor` can see it; CLI output warns plainly instead of
+  reporting a generic failure.
+
 ## [1.1.0] - 2026-07-09
 
 ### Added

src/devklean/__init__.py

@@ -1,6 +1,6 @@
 """devklean — scan and remove node_modules / venvs to reclaim disk space."""
 
 from devklean._version import __version__
-from devklean.models import CleanableItem, DeleteFailure, DeleteResult
+from devklean.models import CleanableItem, DeleteFailure, DeleteResult, PartialDeletion
 
-__all__ = ["CleanableItem", "DeleteFailure", "DeleteResult", "__version__"]
+__all__ = ["CleanableItem", "DeleteFailure", "DeleteResult", "PartialDeletion", "__version__"]

src/devklean/deletion/metadata.py

@@ -16,7 +16,13 @@
 TRASH_STRATEGY = "trash"
 # Bumped for the archive dict gaining compressed/original_size/compressed_size;
 # the new fields are optional on read so schema_version <= 4 records still parse.
-SCHEMA_VERSION = 5
+# Version 6 adds the top-level `status` field ("deleted" vs "partial") so a
+# compress-before-trash run whose archive reached trash but whose original
+# could not be removed stays visible to history/doctor instead of vanishing.
+SCHEMA_VERSION = 6
+
+STATUS_DELETED = "deleted"
+STATUS_PARTIAL = "partial"
 
 
 @dataclass(frozen=True)
@@ -67,6 +73,7 @@ class DeletionMetadataRecord:
     strategy: str
     item: DeletionMetadataItem
     archive: DeletionArchive | None = None
+    status: str = STATUS_DELETED
 
     def to_dict(self) -> dict[str, object]:
         payload = {
@@ -78,6 +85,7 @@ def to_dict(self) -> dict[str, object]:
                 "strategy": self.strategy,
             },
             "item": self.item.to_dict(),
+            "status": self.status,
         }
         if self.archive is not None:
             payload["archive"] = self.archive.to_dict()
@@ -127,6 +135,11 @@ def _parse_record(data: dict[str, object]) -> DeletionMetadataRecord:
     # version is accepted as-is; there are no migrations yet.
     schema_version = data.get("schema_version", 1)
 
+    # `status` postdates schema_version 5; absence means a legacy "deleted"
+    # record, so absence is not an error. Unknown values are rejected as
+    # corrupt so doctor can flag them rather than silently misreporting.
+    status = data.get("status", STATUS_DELETED)
+
     if not (
         isinstance(deletion_id, str)
         and (run_id is None or isinstance(run_id, str))
@@ -171,6 +184,9 @@ def _parse_record(data: dict[str, object]) -> DeletionMetadataRecord:
     if strategy != TRASH_STRATEGY:
         raise ValueError(f"unrecognized strategy {strategy!r}")
 
+    if not isinstance(status, str) or status not in (STATUS_DELETED, STATUS_PARTIAL):
+        raise ValueError(f"unrecognized status {status!r}")
+
     return DeletionMetadataRecord(
         schema_version=schema_version,
         deletion_id=deletion_id,
@@ -183,6 +199,7 @@ def _parse_record(data: dict[str, object]) -> DeletionMetadataRecord:
             size=size,
         ),
         archive=archive,
+        status=status,
     )
 
 
@@ -240,7 +257,8 @@ def record_successes(
         archives: Mapping[str, DeletionArchive] | None = None,
     ) -> None:
         deleted_paths = set(result.deleted)
-        if not deleted_paths:
+        partial_paths = {p.path for p in result.partial}
+        if not deleted_paths and not partial_paths:
             return
 
         self._storage_dir.mkdir(parents=True, exist_ok=True)
@@ -250,7 +268,11 @@ def record_successes(
         archives = archives or {}
 
         for item in items:
-            if item.path not in deleted_paths:
+            if item.path in deleted_paths:
+                status = STATUS_DELETED
+            elif item.path in partial_paths:
+                status = STATUS_PARTIAL
+            else:
                 continue
 
             archive = archives.get(item.path)
@@ -267,6 +289,7 @@ def record_successes(
                     size=item.size,
                 ),
                 archive=archive,
+                status=status,
             )
             stamp = record.timestamp.replace(":", "").replace("+00:00", "Z")
             filename = f"{stamp}_{record.deletion_id}.json"

src/devklean/deletion/trash.py

@@ -15,14 +15,27 @@
 from devklean.deletion.metadata import TRASH_STRATEGY, DeletionArchive, MetadataManager
 from devklean.deletion.safety import SafetyValidator
 from devklean.logging_setup import get_logger
-from devklean.models import CleanableItem, DeleteFailure, DeleteResult
+from devklean.models import CleanableItem, DeleteFailure, DeleteResult, PartialDeletion
 
 # The single deletion backend is the native OS trash (Recycle Bin on Windows,
 # ~/.Trash on macOS, the freedesktop trash on Linux) via send2trash. The name
 # recorded in metadata/history is the shared constant defined in metadata.py.
 STRATEGY_NAME = TRASH_STRATEGY
 
 
+class OriginalCleanupError(OSError):
+    """Archive reached trash but the original directory could not be removed.
+
+    Carries the trashed archive so callers can record it instead of losing
+    it in a plain failure. No data is lost: the recoverable copy is in trash,
+    the source is still on disk.
+    """
+
+    def __init__(self, message: str, archive: DeletionArchive) -> None:
+        super().__init__(message)
+        self.archive = archive
+
+
 def delete_items(
     items: Sequence[CleanableItem],
     total_size: int,
@@ -73,6 +86,7 @@ def delete_items(
 
     deleted: list[str] = []
     failures: list[DeleteFailure] = []
+    partials: list[PartialDeletion] = []
     archives: dict[str, DeletionArchive] = {}
     for item in safe:
         try:
@@ -85,6 +99,21 @@ def delete_items(
             if archive is not None:
                 archives[item.path] = archive
             deleted.append(item.path)
+        except OriginalCleanupError as exc:
+            # Archive is already safe in trash; the original is still on disk.
+            # A distinct outcome, not success nor plain failure, so history
+            # and doctor can see the trashed archive.
+            archives[item.path] = exc.archive
+            partials.append(
+                PartialDeletion(
+                    path=item.path,
+                    error=str(exc),
+                    archive_path=exc.archive.path,
+                    archive_format=exc.archive.format,
+                    original_size=exc.archive.original_size,
+                    compressed_size=exc.archive.compressed_size,
+                )
+            )
         except (OSError, CompressionVerificationError) as exc:
             # TrashPermissionError subclasses OSError; ENOENT/EACCES and
             # platform-specific failures surface here too, alongside
@@ -97,17 +126,27 @@ def delete_items(
         deleted=tuple(deleted),
         failed=tuple(failures) + blocked_failures,
         total_size=safe_total,
+        partial=tuple(partials),
     )
 
     for path in result.deleted:
         logger.info("deleted strategy=%s path=%s", STRATEGY_NAME, path)
+    for partial in result.partial:
+        logger.warning(
+            "partial delete strategy=%s path=%s archive=%s error=%s",
+            STRATEGY_NAME,
+            partial.path,
+            partial.archive_path,
+            partial.error,
+        )
     for failure in result.failed:
         logger.warning("delete failed path=%s error=%s", failure.path, failure.error)
     logger.info(
-        "deletion summary strategy=%s deleted=%d failed=%d size=%d compressed=%d",
+        "deletion summary strategy=%s deleted=%d failed=%d partial=%d size=%d compressed=%d",
         STRATEGY_NAME,
         result.deleted_count,
         result.failed_count,
+        result.partial_count,
         result.total_size,
         len(archives),
     )
@@ -167,9 +206,17 @@ def _send_to_trash(
     try:
         shutil.rmtree(source)
     except OSError as exc:
-        raise OSError(
+        archive = DeletionArchive(
+            path=str(result.archive_path),
+            format=result.format,
+            compressed=True,
+            original_size=result.original_size,
+            compressed_size=compressed_size,
+        )
+        raise OriginalCleanupError(
             f"compressed archive was trashed, but the original directory {source} "
-            f"could not be removed ({exc}); remove it manually to reclaim the disk space"
+            f"could not be removed ({exc}); remove it manually to reclaim the disk space",
+            archive,
         ) from exc
 
     return DeletionArchive(

src/devklean/models.py

@@ -19,11 +19,29 @@ class DeleteFailure:
     error: str
 
 
+@dataclass(frozen=True)
+class PartialDeletion:
+    """Archive trashed but original directory could not be removed.
+
+    Distinct from success (original gone) and failure (nothing trashed):
+    the compressed archive genuinely exists in trash, the source is still
+    on disk and must be removed manually.
+    """
+
+    path: str
+    error: str
+    archive_path: str
+    archive_format: str = "gzip"
+    original_size: int | None = None
+    compressed_size: int | None = None
+
+
 @dataclass(frozen=True)
 class DeleteResult:
     deleted: tuple[str, ...]
     failed: tuple[DeleteFailure, ...]
     total_size: int
+    partial: tuple[PartialDeletion, ...] = ()
 
     @property
     def deleted_count(self) -> int:
@@ -32,3 +50,7 @@ def deleted_count(self) -> int:
     @property
     def failed_count(self) -> int:
         return len(self.failed)
+
+    @property
+    def partial_count(self) -> int:
+        return len(self.partial)

src/devklean/output/text.py

@@ -6,7 +6,7 @@
 from devklean.deletion.integrity import IntegrityReport
 from devklean.formatting import format_size, format_timestamp, truncate
 from devklean.models import CleanableItem, DeleteResult
-from devklean.output.console import SYM_ERROR, SYM_SUCCESS, Console
+from devklean.output.console import SYM_ERROR, SYM_SUCCESS, SYM_WARNING, Console
 from devklean.output.sorting import items_by_size_desc
 from devklean.signatures import ArtifactSignature
 from devklean.signatures.analysis import AnalysisReport
@@ -86,6 +86,8 @@ def deletion_result(self, result: DeleteResult) -> None:
         self._println()
         for path in result.deleted:
             self._println(f"  {c.paint(SYM_SUCCESS, 'success')} {c.paint(path, 'detail')}")
+        for partial in result.partial:
+            self._println(f"  {c.paint(SYM_WARNING, 'warning')} {partial.path} — {partial.error}")
         for failure in result.failed:
             self._println(f"  {c.paint(SYM_ERROR, 'error')} {failure.path} — {failure.error}")
 
@@ -95,6 +97,10 @@ def deletion_result(self, result: DeleteResult) -> None:
         self._console.success(
             c.paint(f"Cleaned {deleted} {word}, freed ~{format_size(result.total_size)}.", "bold")
         )
+        if result.partial_count:
+            self._console.warning(
+                f"{result.partial_count} partial: archive in trash, original still on disk."
+            )
         if result.failed_count:
             self._console.error(f"{result.failed_count} failed.")
         self._println()

tests/test_deletion.py

@@ -184,12 +184,14 @@ def _boom(path) -> None:
         [item], item.size, metadata_manager=manager, compress=True, compress_min_size=0
     )
 
-    # Not a silent success: the item is a reported failure, not a deletion.
+    # Distinct outcome: not success, not plain failure — partial.
     assert result.deleted == ()
-    assert len(result.failed) == 1
-    assert result.failed[0].path == str(source)
+    assert result.failed == ()
+    assert len(result.partial) == 1
+    assert result.partial[0].path == str(source)
+    assert result.partial_count == 1
 
-    error = result.failed[0].error
+    error = result.partial[0].error
     # Distinct from an ordinary failure: says the archive made it to trash...
     assert "compressed archive was trashed" in error
     # ...names the actual reason (proves {exc} was interpolated, not hardcoded)...
@@ -207,8 +209,15 @@ def _boom(path) -> None:
     assert source.exists()
     assert (source / "a.txt").exists()
 
-    # A failed item is never recorded as a successful deletion.
-    assert manager.load_records().records == ()
+    # A partial item is recorded (status=partial) so history/doctor can see
+    # the trashed archive, unlike a plain failure.
+    records = manager.load_records()
+    assert len(records.records) == 1
+    stored = records.records[0]
+    assert stored.record.item.original_path == str(source)
+    assert stored.record.status == "partial"
+    assert stored.record.archive is not None
+    assert stored.record.archive.format == "gzip"
 
 
 def test_delete_items_does_not_call_send2trash_on_dry_run(tmp_path: Path, fake_trash) -> None:
@@ -356,12 +365,13 @@ def _selective(path) -> None:
     payload = json.loads(records[0].read_text(encoding="utf-8"))
 
     assert result.deleted == ("/tmp/a",)
-    assert payload["schema_version"] == 5
+    assert payload["schema_version"] == 6
     assert payload["deletion"]["strategy"] == "trash"
     assert isinstance(payload["deletion"]["run_id"], str) and payload["deletion"]["run_id"]
     assert payload["item"]["original_path"] == "/tmp/a"
     assert payload["item"]["display_name"] == "A"
     assert payload["item"]["size"] == 10
+    assert payload["status"] == "deleted"
 
 
 def test_metadata_manager_records_archive_details(tmp_path: Path) -> None:
@@ -380,7 +390,7 @@ def test_metadata_manager_records_archive_details(tmp_path: Path) -> None:
     records = sorted(storage_dir.glob("*.json"))
     payload = json.loads(records[0].read_text(encoding="utf-8"))
 
-    assert payload["schema_version"] == 5
+    assert payload["schema_version"] == 6
     assert payload["archive"] == {"path": "/tmp/a.zip", "format": "zip", "compressed": True}