Preserve shared ArchiveResult outputs

archivebox/core/models.py

@@ -4115,10 +4115,13 @@ class ArchiveResult(ModelWithDeleteAfter, ModelWithOutputDir, ModelWithNotes):
     def schedule_delete_cleanup(self, *, using: str | None = None) -> None:
         """Remove shared plugin output and refresh persisted Snapshot metadata after commit."""
         snapshot_id = self.snapshot_id
+        plugin = self.plugin
         paths = self.validate_output_paths_for_delete(self.output_paths_for_delete())
 
         def cleanup() -> None:
-            type(self).delete_output_paths(paths)
+            results = type(self).objects.using(using) if using else type(self).objects
+            if not results.filter(snapshot_id=snapshot_id, plugin=plugin).exists():
+                type(self).delete_output_paths(paths)
             type(self).refresh_snapshot_output_sizes({snapshot_id})
             snapshot = Snapshot.objects.filter(pk=snapshot_id).first()
             if snapshot:

archivebox/tests/test_ui_admin_archiveresult.py

@@ -141,3 +141,29 @@ class TestArchiveResultAdminListView:
         assert not output_dir.exists()
         snapshot.refresh_from_db()
         assert snapshot.output_size == 0
+
+    def test_deleting_duplicate_preserves_shared_plugin_output(self, snapshot):
+        from archivebox.core.models import ArchiveResult
+
+        output_dir = Path(snapshot.output_dir) / "responses"
+        output_dir.mkdir(parents=True, exist_ok=True)
+        output_file = output_dir / "index.jsonl"
+        output_file.write_text("captured response\n")
+        primary = ArchiveResult.objects.create(
+            snapshot=snapshot,
+            plugin="responses",
+            hook_name="on_Snapshot__24_responses.daemon.bg",
+            status=ArchiveResult.StatusChoices.SUCCEEDED,
+            output_size=18,
+        )
+        duplicate = ArchiveResult.objects.create(
+            snapshot=snapshot,
+            plugin="responses",
+            hook_name="on_Snapshot__24_responses.daemon.bg.replayed",
+            status=ArchiveResult.StatusChoices.NORESULTS,
+        )
+
+        duplicate.delete()
+
+        assert ArchiveResult.objects.filter(pk=primary.pk).exists()
+        assert output_file.read_text() == "captured response\n"