fix: harden retention cleanup safety

backend/tests/test_cleanup_service.py

@@ -6,6 +6,7 @@ import sys
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
 
+import pytest
 from fastapi.testclient import TestClient
 
 
@@ -173,6 +174,48 @@ def test_restored_product_is_not_deleted_when_original_purge_time_arrives(tmp_pa
     assert service.product_store.get_product(product.product_id).status == "active"
 
 
+def test_atomic_purge_claim_blocks_restore_before_product_files_are_deleted(
+    tmp_path, monkeypatch
+):
+    service = make_cleanup_service(tmp_path)
+    product = create_pending_product(service, purge_after=NOW)
+    restore_conflicts: list[str] = []
+
+    def attempt_restore(product_id: str) -> None:
+        with pytest.raises(ValueError, match="purging"):
+            service.product_store.restore_product(product_id, now=NOW)
+        restore_conflicts.append(product_id)
+
+    monkeypatch.setattr(
+        service, "_before_product_files_delete", attempt_restore, raising=False
+    )
+
+    report = service.apply(now=NOW)
+
+    assert restore_conflicts == [product.product_id]
+    assert report.deleted_product_ids == [product.product_id]
+    assert service.product_store.get_product(product.product_id).status == "purged"
+
+
+def test_failed_product_file_deletion_releases_claim_back_to_pending(tmp_path, monkeypatch):
+    service = make_cleanup_service(tmp_path)
+    product = create_pending_product(service, purge_after=NOW)
+
+    monkeypatch.setattr(
+        "service.cleanup_service.shutil.rmtree",
+        lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError("disk failure")),
+    )
+
+    with pytest.raises(OSError, match="disk failure"):
+        service.apply(now=NOW)
+
+    assert service.product_store.get_product(product.product_id).status == "pending_cleanup"
+    assert service.product_store._fetchone(
+        "SELECT status FROM cleanup_records WHERE product_id = ?", (product.product_id,)
+    )["status"] == "pending_cleanup"
+    assert service.product_store.due_product_cleanups(NOW)[0].product_id == product.product_id
+
+
 def test_apply_rechecks_archive_reference_immediately_before_job_deletion(tmp_path):
     service = make_cleanup_service(tmp_path)
     due_job = create_success_job(service, age_days=31)
@@ -241,6 +284,31 @@ def test_scheduled_cleanup_is_preview_only_until_apply_flag_is_enabled(tmp_path,
     assert not service.storage.job_root(second_job).exists()
 
 
+def test_scheduler_stop_retains_live_worker_and_restart_does_not_duplicate_cycle(monkeypatch):
+    class LiveThread:
+        def __init__(self):
+            self.join_calls: list[float | None] = []
+
+        def is_alive(self) -> bool:
+            return True
+
+        def join(self, timeout=None) -> None:
+            self.join_calls.append(timeout)
+
+    worker = LiveThread()
+    cycles: list[str] = []
+    monkeypatch.setattr(service_app, "_cleanup_scheduler_thread", worker)
+    monkeypatch.setattr(service_app, "_run_scheduled_cleanup_once", lambda: cycles.append("run"))
+
+    service_app._stop_cleanup_scheduler()
+    service_app._start_cleanup_scheduler()
+
+    assert service_app._cleanup_scheduler_stop.is_set()
+    assert service_app._cleanup_scheduler_thread is worker
+    assert worker.join_calls
+    assert cycles == []
+
+
 def test_storage_summary_separates_archive_protection_from_asset_references(tmp_path, monkeypatch):
     service = make_cleanup_service(tmp_path)
     archived_job = create_success_job(service, age_days=3650, archived=True)
@@ -313,3 +381,44 @@ def test_storage_metrics_protects_archives_but_not_old_asset_references(
     assert report["archive_protected_job_ids"] == [archived_job]
     assert report["asset_referenced_job_ids"] == [asset_job]
     assert [item["job_id"] for item in report["stale_jobs"]] == [asset_job]
+
+
+def test_storage_metrics_apply_never_deletes_unproven_or_failed_workspaces(
+    tmp_path, monkeypatch
+):
+    workspace = tmp_path / "workspace"
+    metadata = tmp_path / "metadata"
+    storage = Storage(workspace)
+    metadata_store = MetadataStore(metadata / "app.db")
+    orphan_job = "c" * 32
+    failed_job = "d" * 32
+    for job_id in (orphan_job, failed_job):
+        root = storage.job_root(job_id)
+        root.mkdir(parents=True)
+        (root / "payload.bin").write_bytes(b"payload")
+        old_timestamp = (datetime.now(timezone.utc) - timedelta(days=31)).timestamp()
+        os.utime(root, (old_timestamp, old_timestamp))
+    metadata_store.upsert_job(
+        JobStatus(
+            job_id=failed_job,
+            status="failed",
+            stage="failed",
+            progress_percent=100,
+            message="failed",
+            artifacts={"db": str(storage.job_root(failed_job) / "output" / "db.sqlite")},
+            error="generation failed",
+            created_at=datetime.now(timezone.utc) - timedelta(days=31),
+            updated_at=datetime.now(timezone.utc) - timedelta(days=31),
+        )
+    )
+    monkeypatch.setattr(storage_metrics, "WORKSPACE_DIR", workspace)
+    monkeypatch.setattr(storage_metrics, "ASSETS_DIR", tmp_path / "assets")
+    monkeypatch.setattr(storage_metrics, "METADATA_DIR", metadata)
+    monkeypatch.setattr(storage_metrics, "PRODUCTS_DIR", tmp_path / "products")
+    monkeypatch.setattr(sys, "argv", ["storage_metrics", "--apply"])
+    monkeypatch.setenv("CLEANUP_APPLY_ENABLED", "true")
+
+    storage_metrics.main()
+
+    assert storage.job_root(orphan_job).exists()
+    assert storage.job_root(failed_job).exists()