Harden organizer transfers, undo, and compression publication

docs/AUDIT_REPAIRS.md

@@ -0,0 +1,57 @@
+# Audit repairs: local file safety
+
+This first repair phase addresses the local file-operation and recovery problems
+in the 13 November 2026 audit. Findings outside this phase remain open.
+
+| Finding | Progress |
+| --- | --- |
+| F01: filename-based omission | Organization retains named variants. The legacy detection helper compares full content digests instead of name patterns. |
+| F02: destination collisions | Organizer Copy and Move publish exclusively and choose a digest suffix for collisions. Classification and the CLI duplicate handler still need migration to this shared executor. |
+| F03: partial/unverified copies | Organizer transfers write privately, verify SHA-256 and size by reading back the output, check source stability, then publish. Move removes its source only after verification and publication. |
+| F05: destructive undo | New records contain output identity. Undo refuses edited/replaced outputs, occupied restore paths, and removal of a copy whose original content is no longer available. Old path-only journals require manual review. |
+| F06: journals and retry | UUID sessions, atomic journal replacement, intent before organizer publication, per-action recovery state, active-session exclusion and retryable conflicts are implemented. Other legacy mutation callers still record operations afterward and need migration. |
+| F07: compression overwrites | Direct compression refuses source aliases and existing outputs. Directory compression rejects overlapping roots. Encoders write privately; failed output is removed and publication never replaces another file. |
+| F34: threshold zero | The scan and organize forms preserve zero instead of substituting five. The obsolete filename-skip option is removed from the UI and CLI. |
+
+Transfers and undo coordinate cooperating app processes with POSIX advisory locks.
+Organizer workers share one journal scoped to the job. A failed journal write
+propagates instead of being silently treated as successful recovery bookkeeping.
+Completed undo actions are persisted individually; conflicts keep the remaining
+journal available through the existing API.
+
+Validation at this phase: 65 Python tests, including the 16 existing API/workflow
+tests, plus Ruff and a frontend production build. The added regressions cover separate-process name collisions,
+disk-full errors, corrupted copies, changed sources, occupied restore paths,
+partial undo retries, completion-write failures, abrupt process exits at three
+transfer boundaries, and compression failures/races. All fixtures are disposable.
+The 49 tests added by this phase also run from an export of the committed tree.
+Legacy callers now close their journals on return and exceptions, and API Trash
+requests own independent session managers. The read-only classification scan no
+longer opens an unused file-operation journal.
+
+## Current limits and next repairs
+
+- Publication currently requires hard-link support on the destination filesystem.
+  Unsupported filesystems fail without replacing a destination or removing a Move
+  source. FAT/exFAT support needs another verified, atomic non-replacing primitive.
+- Locks coordinate Clean_Backup operations. They cannot exclude unrelated editors
+  or tools; revision checks detect changes at verification boundaries.
+- Abrupt termination can leave hidden `.clean-backup-*.partial*` staging files.
+  Recovery preserves them; automatic discovery and cleanup are still needed.
+  Unpublished staging files are not successful final outputs.
+- The compression encoder's metadata, animation, HDR and stream-preservation
+  limitations (F08) are unchanged. The new publication boundary does not certify
+  the encoded media's semantic fidelity.
+- Perceptual matching and automatic perceptual omission remain heuristic. This
+  phase does not repair F11–F16 or make a safe-exact-mode claim.
+- Prioritize server-owned deletion plans (F04), API authentication/origin/root
+  boundaries (F09), cloud object identity (F27), and safe Docker build inputs
+  (F37) next. These high-priority findings remain open.
+- Migrate classification, CLI duplicate handling and Trash to the executor, fix
+  classification catalog/undo contracts, and add an explicit resume/reconciliation
+  interface. Recovery currently offers safe undo, not automatic forward resume.
+- API job outcome reporting (F17), other workflow/provider/watcher defects,
+  distribution, dependency triage and UI changes remain separate work.
+
+There is no throughput claim or cross-platform certification for this phase.
+Existing uncommitted CI, README, demo and test changes were left out of its commits.

main.py

@@ -201,16 +201,11 @@ def main():
                 check_dups = input("\nCheck for perceptual duplicates before organizing? (y/n): ").strip().lower()
                 check_duplicates = (check_dups == 'y')
 
-                check_name_dups = input("Check for name-based duplicates (e.g., photo(1).jpg, Image copy.png)? (y/n): ").strip().lower()
-                check_name_duplicates = (check_name_dups == 'y')
-
                 if check_duplicates:
                     print(f"  Using Rust perceptual hashing to detect duplicate images (Threshold: {current_threshold})")
 
-                if check_name_duplicates:
-                    print("  Detecting OS duplicate patterns: (1), (copy), - Copy, copy, drag/drop numbers")
-
-                stats = organise_files(source, destination, operation, check_duplicates=check_duplicates, duplicate_threshold=current_threshold, check_name_duplicates=check_name_duplicates)
+                print("  Files with similar names are retained; naming conflicts receive a unique suffix.")
+                stats = organise_files(source, destination, operation, check_duplicates=check_duplicates, duplicate_threshold=current_threshold)
                 print_summary(stats, operation)
 
                 print("Check logs for detailed information!")

src/classify/apply_org.py

@@ -11,13 +11,15 @@ def apply_classification(run_id, dest_dir, operation, progress_cb=None):
     """
     Applies the classification results from the DB to physical files.
     """
+    with UndoManager() as undo_mgr:
+        return _apply_classification(run_id, dest_dir, operation, progress_cb, undo_mgr)
+
+
+def _apply_classification(run_id, dest_dir, operation, progress_cb, undo_mgr):
     dest_path = Path(dest_dir)
     if not dest_path.exists():
         dest_path.mkdir(parents=True)
 
-    undo_mgr = UndoManager()
-    undo_mgr.start_session()
-
     config = get_run_config(run_id)
     if not config:
         raise ValueError(f"Run ID {run_id} not found in DB")

src/classify/pipeline.py

@@ -36,7 +36,6 @@
 from src.classify.tag_resolver import resolve_tags
 from src.constants import IMAGE_EXTENSIONS, VIDEO_EXTENSIONS
 from src.logger import logger
-from src.undo_manager import undo_manager
 
 
 def run_classify_pipeline(
@@ -118,9 +117,6 @@ class DummyModel:
     # Pre-load known face embeddings for Stage E matching
     known_embeddings = db.get_all_known_embeddings() if face_recognizer.available else []
 
-    # ── Start undo session ────────────────────────────────────────────
-    undo_manager.start_session()
-
     # ── Process each file ─────────────────────────────────────────────
     stats: dict[str, int] = defaultdict(int)
     category_counts: dict[str, int] = defaultdict(int)
@@ -275,7 +271,6 @@ class DummyModel:
                 total_review += 1
 
     # ── Finalise ──────────────────────────────────────────────────────
-    undo_manager.end_session()
     progress_cb(98, "Building summary")
 
     summary = {

src/compressor.py

@@ -12,6 +12,7 @@
 
 from src.constants import IMAGE_EXTENSIONS, VIDEO_EXTENSIONS
 from src.logger import logger
+from src.file_operations import staged_output, validate_roots
 
 pillow_heif.register_heif_opener()
 
@@ -75,7 +76,7 @@ def get_compression_settings(level: CompressionLevel) -> dict:
     return settings.get(level, settings[2])
 
 
-def compress_image(
+def _encode_image(
     image_path: Path,
     output_path: Path,
     level: CompressionLevel = 2
@@ -120,15 +121,14 @@ def compress_image(
                 del save_kwargs['quality']  # PNG doesn't use quality parameter
 
             img.save(output_path, **save_kwargs)
-            logger.info(f"Compressed image: {image_path.name} -> {output_path.name}")
             return True
 
     except Exception as e:
         logger.error(f"Failed to compress image {image_path}: {e}")
         return False
 
 
-def compress_video(
+def _encode_video(
     video_path: Path,
     output_path: Path,
     level: CompressionLevel = 2
@@ -168,7 +168,7 @@ def compress_video(
             '-c:a', 'aac',  # Audio codec
             '-b:a', '128k',  # Audio bitrate
             '-movflags', '+faststart',  # Web optimization
-            '-y',  # Overwrite output
+            '-y',  # Only the exclusively created private staging file.
             str(output_path)
         ]
 
@@ -181,7 +181,6 @@ def compress_video(
         )
 
         if result.returncode == 0:
-            logger.info(f"Compressed video: {video_path.name} -> {output_path.name}")
             return True
         else:
             logger.error(f"FFmpeg failed for {video_path.name}: {result.stderr}")
@@ -198,6 +197,29 @@ def compress_video(
         return False
 
 
+def _compress_to_new_output(source, target, level, encoder):
+    source, target = Path(source), Path(target)
+    try:
+        with staged_output(source, target) as temporary:
+            if not encoder(source, temporary, level):
+                raise OSError("Encoder did not complete successfully")
+        logger.info("Compressed and published: %s -> %s", source, target)
+        return True
+    except Exception as exc:
+        logger.error("Compression output was not published for %s: %s", source, exc)
+        return False
+
+
+def compress_image(image_path: Path, output_path: Path, level: CompressionLevel = 2) -> bool:
+    """Encode a new image without replacing a source or existing destination."""
+    return _compress_to_new_output(image_path, output_path, level, _encode_image)
+
+
+def compress_video(video_path: Path, output_path: Path, level: CompressionLevel = 2) -> bool:
+    """Encode to private staging space; publish only after FFmpeg succeeds."""
+    return _compress_to_new_output(video_path, output_path, level, _encode_video)
+
+
 def compress_files(
     source_dir: str,
     output_dir: str,
@@ -218,12 +240,7 @@ def compress_files(
     Returns:
         CompressionStats: Statistics about the compression operation
     """
-    source = Path(source_dir)
-    output = Path(output_dir)
-
-    if not source.exists():
-        logger.error(f"Source directory not found: {source}")
-        return CompressionStats()
+    source, output = validate_roots(source_dir, output_dir)
 
     # Create output directory
     output.mkdir(parents=True, exist_ok=True)

src/duplicate_handler.py

@@ -22,7 +22,7 @@
     THRESHOLD_SIMILAR
 )
 from src.logger import logger
-from src.undo_manager import undo_manager
+from src.undo_manager import UndoManager
 
 
 @dataclass
@@ -250,10 +250,13 @@ def handle_duplicates(
     Returns:
         DuplicateReport with statistics and details
     """
+    with UndoManager() as session:
+        return _handle_duplicates(source_dir, duplicates_dir, action, threshold, keep_best, session)
+
+
+def _handle_duplicates(source_dir, duplicates_dir, action, threshold, keep_best, undo_manager):
     source_path = Path(source_dir)
 
-    undo_manager.start_session()
-
     # Find duplicates
     groups = scan_for_duplicates(source_dir, threshold)
 
@@ -327,7 +330,6 @@ def handle_duplicates(
                     logger.error(f"Error deleting {dup}: {e}")
                     report.errors += 1
 
-    undo_manager.end_session()
     return report