fix: clean interrupted staging temporaries

src/archive_clients/jobs.py

@@ -32,6 +32,7 @@ from archive_clients.syncthing import RouteSetupError, SyncthingTransferObserver
 from archive_clients.transfer import (
     TransferError,
     TransferIntegrityError,
+    cleanup_partial_transfer,
     cleanup_transfer,
     load_published_transfer,
     materialize_transfer,
@@ -833,6 +834,11 @@ class ClientJobExecutor:
             qb_root=job_directory,
             store=self.store,
         )
+        cleanup_partial_transfer(
+            job_directory,
+            job_id=definition.job_id,
+            store=self.store,
+        )
         for name in ("ready.json", "manifest.json"):
             candidate = job_directory / name
             if candidate.is_file() and not candidate.is_symlink():

src/archive_clients/transfer.py

@@ -456,6 +456,36 @@ def cleanup_transfer(job_directory: Path) -> bool:
     return True
 
 
+def cleanup_partial_transfer(
+    job_directory: Path,
+    *,
+    job_id: str,
+    store: ClientStore,
+) -> None:
+    """Remove copy/reflink temporaries left before a transfer is published.
+
+    A source-stage failure before ``manifest.json``/``ready.json`` exists
+    cannot use :func:`cleanup_transfer`. The file-operation journal is the
+    authoritative list of owned destinations, so derive each temporary path
+    from it rather than recursively removing arbitrary content from a shared
+    Syncthing folder.
+    """
+
+    for row in store.file_operation_rows(job_id):
+        intent = json.loads(str(row["intent_json"]))
+        destination = job_directory / _relative_path(str(intent["destination"]))
+        temporary = _temporary_path(destination, str(row["operation_id"]))
+        try:
+            metadata = temporary.lstat()
+        except FileNotFoundError:
+            continue
+        if not stat.S_ISREG(metadata.st_mode):
+            raise TransferIntegrityError(
+                "job-owned temporary cleanup path is not a regular file"
+            )
+        temporary.unlink()
+
+
 def canonical_message_json(message: object) -> bytes:
     value = json_format.MessageToDict(
         message,