diff --git a/docs/AUDIT_REPAIRS.md b/docs/AUDIT_REPAIRS.md
new file mode 100644
index 0000000..8bd7e81
--- /dev/null
+++ b/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.
diff --git a/main.py b/main.py
index 22352bd..8437c58 100644
--- a/main.py
+++ b/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!")
diff --git a/src/classify/apply_org.py b/src/classify/apply_org.py
index fdfdccb..5e17ed8 100644
--- a/src/classify/apply_org.py
+++ b/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")
diff --git a/src/classify/pipeline.py b/src/classify/pipeline.py
index 1541703..da1f917 100644
--- a/src/classify/pipeline.py
+++ b/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 = {
diff --git a/src/compressor.py b/src/compressor.py
index 729d7ee..39d84cc 100644
--- a/src/compressor.py
+++ b/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,7 +121,6 @@ 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:
@@ -128,7 +128,7 @@ def compress_image(
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)
diff --git a/src/duplicate_handler.py b/src/duplicate_handler.py
index 7b3780f..e65d433 100644
--- a/src/duplicate_handler.py
+++ b/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
diff --git a/src/file_operations.py b/src/file_operations.py
new file mode 100644
index 0000000..1fa9364
--- /dev/null
+++ b/src/file_operations.py
@@ -0,0 +1,239 @@
+"""Verified local transfers with atomic, non-replacing publication.
+
+The final name is a hard link to a fully written and verified temporary file.
+Filesystems without hard-link support fail closed; no replacing fallback is used.
+Locks coordinate Clean_Backup operations, not unrelated editors or filesystem tools.
+"""
+
+from contextlib import contextmanager
+import hashlib
+import os
+from pathlib import Path
+import shutil
+import stat
+import tempfile
+import threading
+
+try:
+ import fcntl
+except ImportError:
+ # Windows still gets process-local locking and exclusive publication.
+ fcntl = None
+
+
+class FileConflictError(OSError):
+ """A source or recorded output no longer matches its expected revision."""
+
+
+_mutation_lock = threading.RLock()
+_held_locks = threading.local()
+
+
+@contextmanager
+def locked_paths(*paths):
+ """Serialize cooperating writers, including separate POSIX app processes."""
+ with _mutation_lock:
+ held = getattr(_held_locks, "paths", {})
+ _held_locks.paths = held
+ acquired = []
+ try:
+ for name in sorted(
+ {str(Path(p).absolute().parent.resolve() / Path(p).name) for p in paths}
+ ):
+ if name in held:
+ continue
+ lock_dir = (
+ Path(tempfile.gettempdir())
+ / f"clean-backup-locks-{getattr(os, 'getuid', lambda: 0)()}"
+ )
+ lock_dir.mkdir(mode=0o700, exist_ok=True)
+ lock_path = lock_dir / hashlib.sha256(name.encode()).hexdigest()
+ fd = os.open(
+ lock_path,
+ os.O_CREAT | os.O_RDWR | getattr(os, "O_NOFOLLOW", 0),
+ 0o600,
+ )
+ try:
+ if fcntl:
+ fcntl.flock(fd, fcntl.LOCK_EX)
+ except BaseException:
+ os.close(fd)
+ raise
+ held[name] = fd
+ acquired.append(name)
+ yield
+ finally:
+ for name in reversed(acquired):
+ os.close(held.pop(name))
+
+
+def fsync_directory(path):
+ """Persist a directory entry after link, rename or unlink on POSIX."""
+ if os.name != "posix":
+ return
+ fd = os.open(path, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
+ try:
+ os.fsync(fd)
+ finally:
+ os.close(fd)
+
+
+def _revision(info):
+ return (info.st_dev, info.st_ino, info.st_size, info.st_mtime_ns, info.st_ctime_ns)
+
+
+def file_identity(path):
+ """Hash a regular file and refuse symlinks or a revision changed during reading."""
+ path = Path(path)
+ fd = os.open(
+ path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0)
+ )
+ with os.fdopen(fd, "rb") as stream:
+ before = os.fstat(stream.fileno())
+ if not stat.S_ISREG(before.st_mode):
+ raise FileConflictError(f"Not a regular file: {path}")
+ digest = hashlib.file_digest(stream, "sha256").hexdigest()
+ after = os.fstat(stream.fileno())
+ if _revision(before) != _revision(after) or _revision(after) != _revision(
+ path.lstat()
+ ):
+ raise FileConflictError(f"File changed while reading: {path}")
+ return dict(
+ zip(("device", "inode", "size", "mtime_ns", "ctime_ns"), _revision(after)),
+ sha256=digest,
+ )
+
+
+def require_identity(path, expected, *, published=False):
+ if not expected:
+ raise FileConflictError(f"No recorded file identity: {path}")
+ actual = file_identity(path)
+ # Linking/unlinking the staging name changes ctime. Pending journal intents
+ # still identify the exact inode and bytes even across that publication step.
+ keys = (
+ ("device", "inode", "size", "mtime_ns", "sha256")
+ if published
+ else expected.keys()
+ )
+ if any(actual.get(key) != expected.get(key) for key in keys):
+ raise FileConflictError(
+ f"File changed since the operation was recorded: {path}"
+ )
+ return actual
+
+
+def validate_roots(source, destination):
+ source, destination = Path(source).resolve(), Path(destination).resolve()
+ if not source.is_dir():
+ raise ValueError(f"Source directory not found: {source}")
+ if (
+ source == destination
+ or source in destination.parents
+ or destination in source.parents
+ ):
+ raise ValueError("Source and destination directories must not overlap")
+ return source, destination
+
+
+def _temporary_path(target, *, suffix=".partial"):
+ fd, name = tempfile.mkstemp(
+ prefix=".clean-backup-", suffix=suffix, dir=target.parent
+ )
+ os.close(fd)
+ return Path(name)
+
+
+def _publish(temporary, target):
+ # os.link is atomic and refuses all existing targets, including dangling links.
+ os.link(temporary, target, follow_symlinks=False)
+ fsync_directory(target.parent)
+
+
+def verified_transfer(
+ source, target, operation="copy", *, journal=None, collision="rename"
+):
+ """Copy, read back and publish; only then remove an unchanged Move source.
+
+ A journal, when supplied, persists intent before publication and completion
+ afterward. A failure after publication retains the verified output and its
+ pending intent for recovery rather than attempting a destructive rollback.
+ """
+ if operation not in {"copy", "move"} or collision not in {"rename", "error"}:
+ raise ValueError("Invalid transfer operation or collision policy")
+ source, target = Path(source).absolute(), Path(target).absolute()
+ if source.resolve() == target.resolve():
+ raise FileConflictError("Source and destination must differ")
+ target.parent.mkdir(parents=True, exist_ok=True)
+ with locked_paths(source, target):
+ original = file_identity(source)
+ temporary = _temporary_path(target)
+ try:
+ shutil.copy2(source, temporary)
+ with temporary.open("rb") as stream:
+ os.fsync(stream.fileno())
+ copied = file_identity(temporary)
+ if (copied["size"], copied["sha256"]) != (
+ original["size"],
+ original["sha256"],
+ ):
+ raise FileConflictError(f"Copy verification failed: {source}")
+ require_identity(source, original)
+ candidate = target
+ counter = 0
+ while True:
+ entry = (
+ journal.prepare_action(
+ operation, source, candidate, original, copied
+ )
+ if journal
+ else None
+ )
+ try:
+ _publish(temporary, candidate)
+ break
+ except FileExistsError:
+ if journal:
+ journal.cancel_action(entry)
+ if collision == "error":
+ raise
+ counter += 1
+ tail = "" if counter == 1 else f"-{counter}"
+ candidate = target.with_name(
+ f"{target.stem}-{original['sha256'][:12]}{tail}{target.suffix}"
+ )
+ temporary.unlink()
+ # Read back once more under the final name before a Move can remove its source.
+ output_identity = require_identity(candidate, copied, published=True)
+ if operation == "move":
+ require_identity(source, original)
+ source.unlink()
+ fsync_directory(source.parent)
+ if journal:
+ journal.complete_action(entry, output_identity)
+ return candidate
+ finally:
+ temporary.unlink(missing_ok=True)
+
+
+@contextmanager
+def staged_output(source, target):
+ """Give an encoder private output space, then publish without replacement."""
+ source, target = Path(source).absolute(), Path(target).absolute()
+ if source.resolve() == target.resolve():
+ raise FileConflictError("Compression cannot overwrite its source")
+ target.parent.mkdir(parents=True, exist_ok=True)
+ with locked_paths(source, target):
+ if os.path.lexists(target):
+ raise FileExistsError(f"Output already exists: {target}")
+ original = file_identity(source)
+ temporary = _temporary_path(target, suffix=f".partial{target.suffix}")
+ try:
+ yield temporary
+ with temporary.open("rb") as stream:
+ os.fsync(stream.fileno())
+ if not temporary.stat().st_size:
+ raise FileConflictError("Encoder produced an empty output")
+ require_identity(source, original)
+ _publish(temporary, target)
+ finally:
+ temporary.unlink(missing_ok=True)
diff --git a/src/organiser.py b/src/organiser.py
index 64e9dfa..9deb3c0 100644
--- a/src/organiser.py
+++ b/src/organiser.py
@@ -1,92 +1,34 @@
-import shutil
-import re
from pathlib import Path
from collections import defaultdict
-from multiprocessing import Pool, cpu_count
+from concurrent.futures import ThreadPoolExecutor
+from multiprocessing import cpu_count
from tqdm import tqdm
from src import constants
from src.metadata import get_image_date, get_video_date, get_file_modification_date
from src.logger import logger
from src.phash import find_duplicates_from_paths, is_rust_available, THRESHOLD_SIMILAR
-from src.undo_manager import undo_manager
+from src.undo_manager import UndoManager
+from src.file_operations import file_identity, validate_roots, verified_transfer
def detect_name_based_duplicates(file_paths):
+ """Legacy API: return content-identical files, never infer identity from names.
+
+ Organization retains all inputs even when filename hints are requested. Use
+ duplicate review to decide whether a redundant copy should be omitted.
"""
- Detect duplicate files based on common naming patterns from various OS.
-
- Patterns detected:
- - Windows (Download): photo (1).jpg, photo (2).jpg
- - Generic: photo-1.jpg, photo_1.jpg
- - macOS (Duplicate): photo copy.jpg, photoCopy.jpg
- - Windows (Copy): Notes - Copy.txt
- - Linux (Nautilus): Data (copy).csv
- - macOS (Drag/Drop): Image 2.png, Image 3.png
-
- Args:
- file_paths: List of Path objects
-
- Returns:
- set: Paths that should be considered duplicates (keeps original without suffix)
- """
- duplicate_files = set()
- base_name_groups = defaultdict(list)
-
- # Regex patterns to detect duplicate suffixes (order matters - most specific first)
- patterns = [
- r'\s*\((\d+)\)$', # (1), (2) - Windows/Android/iOS downloads
- r'\s*\(copy\)$', # (copy) - Linux Nautilus
- r'\s*-\s*Copy(\d*)$', # - Copy, - Copy2 - Windows
- r'\s+copy(\d*)$', # copy, copy2 - macOS (space before)
- r'[-_]copy(\d*)$', # -copy, _copy - generic
- r'[-_]Copy(\d*)$', # -Copy, _Copy - generic
- r'[-_]COPY(\d*)$', # -COPY, _COPY - generic
- r'[-_](\d+)$', # -1, _1 - generic
- r'\s+(\d+)$', # (space)2, (space)3 - macOS drag/drop
- ]
-
- # Examples of matched patterns:
- # "Installer (1).exe" → base: "Installer.exe"
- # "Data (copy).csv" → base: "Data.csv"
- # "Notes - Copy.txt" → base: "Notes.txt"
- # "Image copy.png" → base: "Image.png"
- # "Photo_1.jpg" → base: "Photo.jpg"
- # "Image 2.png" → base: "Image.png"
-
- for file_path in file_paths:
- if not file_path.is_file():
+ seen = {}
+ duplicates = set()
+ for path in file_paths:
+ path = Path(path)
+ if not path.is_file() or path.is_symlink():
continue
-
- stem = file_path.stem # filename without extension
- suffix = file_path.suffix
-
- # Try to extract base name by removing duplicate suffixes
- base_name = stem
- is_duplicate = False
-
- for pattern in patterns:
- match = re.search(pattern, stem, re.IGNORECASE)
- if match:
- # Remove the duplicate suffix to get base name
- base_name = re.sub(pattern, '', stem, flags=re.IGNORECASE)
- is_duplicate = True
- break
-
- # Group files by base name + extension
- key = f"{base_name}{suffix}".lower()
- base_name_groups[key].append((file_path, is_duplicate, stem))
-
- # Analyze groups to find duplicates
- for base_key, files in base_name_groups.items():
- if len(files) > 1:
- # Sort: originals (no suffix) first, then by name
- files.sort(key=lambda x: (x[1], x[2])) # (is_duplicate, stem)
-
- # Keep the first file (original without suffix), mark rest as duplicates
- for file_path, is_dup, stem in files[1:]:
- duplicate_files.add(file_path)
- logger.info(f"Name-based duplicate detected: {file_path.name} (base: {base_key})")
-
- return duplicate_files
+ identity = file_identity(path)
+ key = (identity["size"], identity["sha256"])
+ if key in seen:
+ duplicates.add(path)
+ else:
+ seen[key] = path
+ return duplicates
def _process_single_file(args):
@@ -99,7 +41,8 @@ def _process_single_file(args):
Returns:
dict: Result dictionary with status, file info, and statistics
"""
- file_path, dest_path, operation, duplicate_files, name_duplicate_files = args
+ file_path, dest_path, operation, duplicate_files, name_duplicate_files = args[:5]
+ journal = args[5] if len(args) > 5 else None
result = {
'status': 'skipped',
@@ -112,13 +55,6 @@ def _process_single_file(args):
}
try:
- # Skip name-based duplicates
- if name_duplicate_files and file_path in name_duplicate_files:
- result['status'] = 'skipped'
- result['reason'] = 'name_duplicate'
- logger.info(f"Skipping name-based duplicate: {file_path.name}")
- return result
-
# Skip perceptual duplicates (use absolute path for comparison)
if duplicate_files and file_path.resolve() in duplicate_files:
result['status'] = 'skipped'
@@ -163,21 +99,9 @@ def _process_single_file(args):
target_file = target_folder / file_path.name
- # Check if file already exists
- if target_file.exists():
- result['status'] = 'skipped'
- result['reason'] = 'duplicate_filename'
- logger.info(f"Skipped duplicate: {file_path.name} already exists in {target_folder}")
- return result
-
- # Move or copy file
- if operation == 'copy':
- shutil.copy2(str(file_path), str(target_file))
- logger.info(f"Copied: {file_path.name} -> {target_folder}")
- else:
- shutil.move(str(file_path), str(target_file))
- logger.info(f"Moved: {file_path.name} -> {target_folder}")
-
+ target_file = verified_transfer(file_path, target_file, operation, journal=journal)
+ logger.info(f"Verified {operation}: {file_path.name} -> {target_file}")
+
result['status'] = 'success'
result['folder_key'] = folder_key
result['source'] = file_path
@@ -210,14 +134,13 @@ def organise_files(
operation: 'move' or 'copy'
check_duplicates: If True, detect and skip duplicate images before organizing (perceptual hashing)
duplicate_threshold: Hamming distance threshold for duplicate detection
- check_name_duplicates: If True, detect and skip files with duplicate naming patterns (e.g., photo(1).jpg)
+ check_name_duplicates: Legacy option; named variants are retained for content review
progress_callback: Optional callback(completed, total, result_dict) invoked per processed file
target_files: Optional list of specific absolute file paths to organize. If provided, only these files are processed.
"""
- src_path = Path(src_dir)
- dest_path = Path(dest_dir)
-
- undo_manager.start_session()
+ if operation not in {"copy", "move"}:
+ raise ValueError("operation must be copy or move")
+ src_path, dest_path = validate_roots(src_dir, dest_dir)
# Statistics tracking
stats = {
@@ -234,19 +157,18 @@ def organise_files(
}
# Gather files
- if target_files:
- all_files = [Path(f) for f in target_files]
+ if target_files is not None:
+ all_files = [Path(f).absolute() for f in target_files]
+ if any(p.is_symlink() or src_path not in p.resolve().parents for p in all_files):
+ raise ValueError("target_files must be regular files inside the source directory")
else:
all_files = list(src_path.rglob('*'))
- # Name-based duplicate detection (if enabled)
+ # Filename resemblance never authorizes omitting an input.
name_duplicate_files = set()
if check_name_duplicates:
- logger.info("Scanning for name-based duplicates (e.g., photo(1).jpg)...")
- name_duplicate_files = detect_name_based_duplicates(all_files)
- stats['name_duplicates'] = len(name_duplicate_files)
- logger.info(f"Found {len(name_duplicate_files)} name-based duplicates")
-
+ logger.info("Filename hints no longer skip inputs; retaining all named variants")
+
# Perceptual duplicate detection (if enabled)
duplicate_files = set() # Files to skip due to perceptual duplication
if check_duplicates:
@@ -299,13 +221,12 @@ def organise_files(
# Collect all files to process
file_list = []
for file_path in all_files:
- if file_path.is_file():
+ if file_path.is_file() and not file_path.is_symlink():
stats['total_scanned'] += 1
file_list.append(file_path)
if not file_list:
logger.info("No files found to process")
- undo_manager.end_session()
if progress_callback:
try:
progress_callback(0, 0, None)
@@ -313,33 +234,25 @@ def organise_files(
pass
return stats
- # Determine number of workers (leave 1 core free)
- num_workers = max(1, cpu_count() - 1)
- logger.info(f"Using {num_workers} worker processes for parallel file organization")
- print(f"\n🚀 Processing {len(file_list)} files with {num_workers} workers...")
-
- # Prepare arguments for worker function
- worker_args = [
- (file_path, dest_path, operation, duplicate_files if check_duplicates else None,
- name_duplicate_files if check_name_duplicates else None)
- for file_path in file_list
- ]
-
- # Process files in parallel with progress bar
+ num_workers = min(8, max(1, cpu_count() - 1), len(file_list))
results = []
total_files = len(file_list)
- with Pool(processes=num_workers) as pool:
- for completed, result in enumerate(tqdm(pool.imap(_process_single_file, worker_args),
- total=total_files,
- desc="Organizing files",
- unit="file"), 1):
- results.append(result)
- if progress_callback:
- try:
+ # Threads share one explicitly scoped journal; no forked singleton state.
+ with UndoManager() as journal:
+ worker_args = [
+ (p, dest_path, operation, duplicate_files if check_duplicates else None,
+ name_duplicate_files, journal)
+ for p in file_list
+ ]
+ with ThreadPoolExecutor(max_workers=num_workers) as pool:
+ for completed, result in enumerate(tqdm(
+ pool.map(_process_single_file, worker_args), total=total_files,
+ desc="Organizing files", unit="file",
+ ), 1):
+ results.append(result)
+ if progress_callback:
progress_callback(completed, total_files, result)
- except Exception:
- pass
-
+
# Aggregate results and update statistics
for result in results:
# Track file types
@@ -356,8 +269,6 @@ def organise_files(
if result['folder_key']:
stats['folders_created'][result['folder_key']] += 1
- # Log to undo manager
- undo_manager.log_action(operation, result['source'], result['destination'])
elif result['status'] == 'skipped':
if result['reason'] == 'duplicate_filename':
@@ -367,7 +278,6 @@ def organise_files(
elif result['status'] == 'error':
stats['errors'] += 1
- undo_manager.end_session()
return stats
diff --git a/src/undo_manager.py b/src/undo_manager.py
index d31273d..8d03106 100644
--- a/src/undo_manager.py
+++ b/src/undo_manager.py
@@ -1,168 +1,277 @@
+"""Durable local operation journals and conflict-aware, retryable undo."""
+
+from datetime import datetime
import json
import os
-import shutil
from pathlib import Path
-from datetime import datetime
+import tempfile
+import threading
+import uuid
+
+from src.file_operations import (
+ FileConflictError,
+ fcntl,
+ file_identity,
+ fsync_directory,
+ locked_paths,
+ require_identity,
+ verified_transfer,
+)
from src.logger import logger
JOURNAL_DIR = Path("logs/undo_journals")
+
+def _write_journal(path, actions):
+ """Never truncate the only durable copy of a journal."""
+ fd, name = tempfile.mkstemp(prefix=".journal-", suffix=".tmp", dir=path.parent)
+ temporary = Path(name)
+ try:
+ with os.fdopen(fd, "w") as stream:
+ json.dump(actions, stream, indent=2)
+ stream.flush()
+ os.fsync(stream.fileno())
+ os.replace(temporary, path)
+ fsync_directory(path.parent)
+ finally:
+ temporary.unlink(missing_ok=True)
+
+
+def _session_lock(path):
+ fd = os.open(
+ str(path) + ".lock",
+ os.O_CREAT | os.O_RDWR | getattr(os, "O_NOFOLLOW", 0),
+ 0o600,
+ )
+ try:
+ if fcntl:
+ fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+ return fd
+ except BaseException:
+ os.close(fd)
+ raise
+
+
class UndoManager:
- def __init__(self):
+ def __init__(self, journal_dir=None):
+ self._journal_dir = Path(journal_dir) if journal_dir is not None else None
self.session_id = None
self.journal_path = None
self.actions = []
- JOURNAL_DIR.mkdir(parents=True, exist_ok=True)
+ self._lock = threading.RLock()
+ self._session_fd = None
+
+ def __enter__(self):
+ self.start_session()
+ return self
+
+ def __exit__(self, *exc):
+ self.end_session()
+
+ @property
+ def journal_dir(self):
+ return (self._journal_dir or JOURNAL_DIR).absolute()
def start_session(self):
- """Start a new transaction session."""
- self.session_id = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
- self.journal_path = JOURNAL_DIR / f"journal_{self.session_id}.json"
- self.actions = []
- logger.info(f"Started undo session: {self.session_id}")
+ with self._lock:
+ if self.session_id is not None:
+ raise RuntimeError("An undo session is already active on this manager")
+ self.journal_dir.mkdir(parents=True, exist_ok=True)
+ self.session_id = (
+ datetime.now().strftime("%Y-%m-%d_%H-%M-%S-%f") + "_" + uuid.uuid4().hex
+ )
+ self.journal_path = self.journal_dir / f"journal_{self.session_id}.json"
+ self.actions = []
+ self._session_fd = _session_lock(self.journal_path)
+ try:
+ self._save_journal()
+ except BaseException:
+ self.end_session()
+ raise
+ return self.session_id
+
+ def _save_journal(self):
+ if self.journal_path is None:
+ raise RuntimeError("No active journal")
+ _write_journal(self.journal_path, self.actions)
+
+ def prepare_action(self, action, src, dst, source_identity, output_identity):
+ """Record intent before a verified temporary output is published."""
+ with self._lock:
+ if self.session_id is None:
+ raise RuntimeError("No active undo session")
+ entry = {
+ "id": uuid.uuid4().hex,
+ "action": action,
+ "src": str(Path(src).absolute()),
+ "dst": str(Path(dst).absolute()),
+ "source_identity": source_identity,
+ "output_identity": output_identity,
+ "state": "prepared",
+ "timestamp": datetime.now().isoformat(),
+ }
+ self.actions.append(entry)
+ self._save_journal() # An error propagates and prevents publication.
+ return entry
+
+ def cancel_action(self, entry):
+ with self._lock:
+ entry["state"] = "cancelled"
+ self._save_journal()
+
+ def complete_action(self, entry, output_identity):
+ with self._lock:
+ entry["output_identity"] = output_identity
+ entry["state"] = "completed"
+ self._save_journal()
def log_action(self, action_type, src, dst):
- """Log a file operation.
-
- Args:
- action_type (str): 'move' or 'copy'
- src (str): Source path
- dst (str): Destination path
+ """Compatibility for older callers; new transfers must record intent first.
+
+ Old path-only journals remain readable, but undo refuses them because they
+ cannot prove ownership of the current file. Missing outputs are recorded
+ as unresolved rather than later acquiring an unrelated file's identity.
"""
- if not self.session_id:
- return # No active session
-
- entry = {
- "action": action_type,
- "src": str(src),
- "dst": str(dst),
- "timestamp": datetime.now().isoformat()
- }
- self.actions.append(entry)
-
- # Write to disk immediately for crash safety (append mode would be better in production, but JSON list is simple)
- self._save_journal()
-
- def _cleanup_empty_dirs(self, directory):
- """Recursively delete empty directories up the tree."""
- path = Path(directory)
+ if self.session_id is None:
+ raise RuntimeError("No active undo session")
try:
- # Stop if we reach root or get a permission error
- while path.is_dir() and path != path.parent:
- if not any(path.iterdir()):
- path.rmdir()
- logger.info(f"Cleanup: Deleted empty directory {path}")
- path = path.parent
- else:
- break
- except Exception:
- # Stops bubbling up if duplicate_handler or other process is using the dir, or permission denied
- pass
-
- def _save_journal(self):
+ identity = file_identity(dst)
+ except OSError:
+ identity = None
try:
- with open(self.journal_path, 'w') as f:
- json.dump(self.actions, f, indent=4)
- except Exception as e:
- logger.error(f"Failed to save undo journal: {e}")
+ source_identity = file_identity(src)
+ except OSError:
+ source_identity = None
+ entry = self.prepare_action(action_type, src, dst, source_identity, identity)
+ self.complete_action(entry, identity)
def end_session(self):
- """Close the current session."""
- if self.actions:
- logger.info(f"Ended undo session {self.session_id} with {len(self.actions)} actions.")
- else:
- # Cleanup empty journal
- if self.journal_path and self.journal_path.exists():
- os.remove(self.journal_path)
-
- self.session_id = None
- self.actions = []
- self.journal_path = None
+ with self._lock:
+ try:
+ if self.journal_path and not self.actions:
+ self.journal_path.unlink(missing_ok=True)
+ finally:
+ if self._session_fd is not None:
+ os.close(self._session_fd)
+ self._session_fd = None
+ self.session_id = None
+ self.actions = []
+ self.journal_path = None
def list_sessions(self):
- """List available undo sessions."""
- files = sorted(JOURNAL_DIR.glob("journal_*.json"), reverse=True)
sessions = []
- for p in files:
+ for path in sorted(self.journal_dir.glob("journal_*.json"), reverse=True):
+ fd = None
try:
- # filename format: journal_YYYY-MM-DD_HH-MM-SS.json
- # session_id matches the timestamp part
- sid = p.stem.replace("journal_", "")
- with open(p, 'r') as f:
- data = json.load(f)
- count = len(data)
- sessions.append({"id": sid, "path": p, "count": count})
- except Exception:
+ fd = _session_lock(path)
+ actions = json.loads(path.read_text())
+ count = sum(
+ a.get("state") not in {"undone", "cancelled"} for a in actions
+ )
+ if count:
+ sessions.append(
+ {
+ "id": path.stem.removeprefix("journal_"),
+ "path": path,
+ "count": count,
+ }
+ )
+ except (OSError, ValueError, TypeError):
continue
+ finally:
+ if fd is not None:
+ os.close(fd)
return sessions
+ @staticmethod
+ def _check_copy_survivor(entry):
+ identity = file_identity(entry["src"])
+ expected = entry.get("source_identity") or entry.get("output_identity")
+ if not expected or any(identity[k] != expected[k] for k in ("sha256", "size")):
+ raise FileConflictError(
+ "The original copy source is missing or changed; retaining the output"
+ )
+
+ def _undo_entry(self, entry, save):
+ action, src, dst = entry["action"], Path(entry["src"]), Path(entry["dst"])
+ expected = entry.get("output_identity")
+ if action not in {"copy", "move"} or not expected:
+ raise FileConflictError(
+ "Journal has no supported, verified output identity; manual review required"
+ )
+ with locked_paths(src, dst):
+ if not os.path.lexists(dst):
+ if entry.get("undo_phase") == "removing":
+ if action == "copy":
+ self._check_copy_survivor(entry)
+ else:
+ require_identity(src, entry["restore_identity"])
+ return
+ if entry.get("state") == "prepared":
+ require_identity(src, entry["source_identity"])
+ return # The intent was written but no final output was published.
+ raise FileConflictError(f"Recorded output is missing: {dst}")
+ require_identity(dst, expected, published=entry.get("state") == "prepared")
+ if action == "copy":
+ self._check_copy_survivor(entry)
+ elif os.path.lexists(src):
+ if entry.get("restore_identity"):
+ require_identity(src, entry["restore_identity"])
+ elif entry.get("state") == "prepared":
+ require_identity(src, entry["source_identity"])
+ else:
+ raise FileConflictError(
+ f"Restore path is occupied; retaining both files: {src}"
+ )
+ else:
+ restored = verified_transfer(dst, src, "copy", collision="error")
+ entry["restore_identity"] = file_identity(restored)
+ save()
+ entry["undo_phase"] = "removing"
+ save() # A retry can distinguish an interrupted undo from a missing file.
+ if action == "copy":
+ self._check_copy_survivor(entry)
+ else:
+ require_identity(
+ src, entry.get("restore_identity") or entry["source_identity"]
+ )
+ require_identity(dst, expected, published=entry.get("state") == "prepared")
+ dst.unlink()
+ fsync_directory(dst.parent)
+
def undo_session(self, session_path):
- """Revert a specific session."""
- logger.info(f"Reverting session from {session_path}")
+ path = Path(session_path).absolute()
+ fd = None
try:
- with open(session_path, 'r') as f:
- actions = json.load(f)
- except Exception as e:
- logger.error(f"Could not read journal file: {e}")
- return False
+ fd = _session_lock(path)
+ actions = json.loads(path.read_text())
- success_count = 0
- fail_count = 0
+ def save():
+ _write_journal(path, actions)
- # Reverse order to undo correctly
- for entry in reversed(actions):
- action = entry.get("action")
- src = entry.get("src") # This was the ORIGINAL source
- dst = entry.get("dst") # This was the destination
+ for entry in reversed(actions):
+ if entry.get("state") in {"undone", "cancelled"}:
+ continue
+ try:
+ self._undo_entry(entry, save)
+ entry["state"] = "undone"
+ entry.pop("error", None)
+ save()
+ except Exception as exc:
+ logger.error("Undo conflict for %s: %s", entry.get("dst"), exc)
+ entry["error"] = str(exc)
+ save()
+ if any(a.get("state") not in {"undone", "cancelled"} for a in actions):
+ return False
+ os.replace(path, str(path) + ".reverted")
+ fsync_directory(path.parent)
+ return True
+ except (OSError, ValueError, TypeError) as exc:
+ logger.error("Could not revert journal %s: %s", path, exc)
+ return False
+ finally:
+ if fd is not None:
+ os.close(fd)
- try:
- if action == "move":
- # Undo move: Move dst back to src
- if os.path.exists(dst):
- # Ensure src directory exists
- os.makedirs(os.path.dirname(src), exist_ok=True)
- shutil.move(dst, src)
- logger.info(f"Undo Move: {dst} -> {src}")
- success_count += 1
-
- # Cleanup empty directories at destination
- self._cleanup_empty_dirs(os.path.dirname(dst))
- else:
- logger.warning(f"Undo failed: File not found at {dst}")
- fail_count += 1
-
- elif action == "copy":
- # Undo copy: Delete dst
- if os.path.exists(dst):
- os.remove(dst)
- logger.info(f"Undo Copy: Deleted {dst}")
- success_count += 1
-
- # Cleanup empty directories at destination
- self._cleanup_empty_dirs(os.path.dirname(dst))
- else:
- logger.warning(f"Undo failed: File not found at {dst}")
- fail_count += 1
- # Note: we don't strictly need to do anything to 'src' for a copy undo,
- # as it is essentially "delete the copy".
-
- except Exception as e:
- logger.error(f"Error undoing action {entry}: {e}")
- fail_count += 1
-
- print("\nUndo Complete.")
- print(f"Successfully reverted: {success_count} files")
- print(f"Failed to revert: {fail_count} files")
-
- # Rename journal to indicate it's undone (or delete it?)
- # Let's rename it to .undone
- try:
- new_path = str(session_path) + ".reverted"
- os.rename(session_path, new_path)
- logger.info("Marked journal as reverted.")
- except OSError:
- pass
-
- return True
undo_manager = UndoManager()
diff --git a/src/web_app.py b/src/web_app.py
index 0799587..4408729 100644
--- a/src/web_app.py
+++ b/src/web_app.py
@@ -21,7 +21,7 @@
from src.logger import logger
from src.organiser import organise_files
from src.phash import get_backend
-from src.undo_manager import undo_manager
+from src.undo_manager import UndoManager, undo_manager
from src.demo import demo_guard, is_demo_mode
# Classification module
@@ -802,7 +802,8 @@ def api_delete_selected():
deleted_paths: list[str] = []
errors: list[str] = []
- undo_manager.start_session()
+ session_manager = UndoManager()
+ session_manager.start_session()
try:
for item in selected_images:
@@ -827,7 +828,7 @@ def api_delete_selected():
target = _trash_target_for(file_path)
target.parent.mkdir(parents=True, exist_ok=True)
file_path.rename(target)
- undo_manager.log_action("move", str(file_path), str(target))
+ session_manager.log_action("move", str(file_path), str(target))
else:
file_path.unlink()
@@ -835,9 +836,9 @@ def api_delete_selected():
except Exception as exc:
errors.append(str(exc))
- undo_manager.end_session()
+ session_manager.end_session()
except Exception:
- undo_manager.end_session()
+ session_manager.end_session()
raise
return jsonify(
diff --git a/tests/test_compression_safety.py b/tests/test_compression_safety.py
new file mode 100644
index 0000000..6f005c0
--- /dev/null
+++ b/tests/test_compression_safety.py
@@ -0,0 +1,121 @@
+"""Compression publication must preserve inputs and existing outputs."""
+
+from pathlib import Path
+from subprocess import CompletedProcess
+
+from PIL import Image
+import pytest
+
+from src import compressor
+
+
+@pytest.fixture
+def photo(tmp_path):
+ source = tmp_path / "original.jpg"
+ Image.new("RGB", (100, 80), "blue").save(source, quality=95)
+ return source
+
+
+@pytest.mark.parametrize("existing", ["source", "other_file", "symlink"])
+def test_image_compression_never_replaces_existing_content(photo, existing):
+ output = photo.parent / "output.jpg"
+ before = photo.read_bytes()
+ if existing == "source":
+ output = photo
+ elif existing == "other_file":
+ output.write_bytes(b"previous output")
+ else:
+ output.symlink_to(photo)
+ expected = output.read_bytes()
+ assert not compressor.compress_image(photo, output, level=3)
+ assert output.read_bytes() == expected
+ assert photo.read_bytes() == before
+ assert not list(photo.parent.glob(".clean-backup-*"))
+
+
+def test_image_output_is_decodable_and_source_unchanged(photo):
+ output = photo.parent / "output.jpg"
+ before = photo.read_bytes()
+ assert compressor.compress_image(photo, output)
+ with Image.open(output) as image:
+ image.load()
+ assert image.size == (100, 80)
+ assert photo.read_bytes() == before
+ assert not list(photo.parent.glob(".clean-backup-*"))
+
+
+def test_encoder_failure_does_not_leave_partial_final_output(photo, monkeypatch):
+ output = photo.parent / "output.jpg"
+ before = photo.read_bytes()
+
+ def interrupted(image, path, **kwargs):
+ Path(path).write_bytes(b"partial")
+ raise OSError("Disk full")
+
+ monkeypatch.setattr(Image.Image, "save", interrupted)
+ assert not compressor.compress_image(photo, output)
+ assert photo.read_bytes() == before
+ assert not output.exists()
+ assert not list(photo.parent.glob(".clean-backup-*"))
+
+
+def test_concurrent_output_is_not_replaced_at_publication(photo, monkeypatch):
+ output = photo.parent / "output.jpg"
+ encode = compressor._encode_image
+
+ def race(source, temporary, level):
+ result = encode(source, temporary, level)
+ output.write_text("other process created this")
+ return result
+
+ monkeypatch.setattr(compressor, "_encode_image", race)
+ assert not compressor.compress_image(photo, output)
+ assert output.read_text() == "other process created this"
+
+
+@pytest.mark.parametrize("nested", ["same", "inside", "outside"])
+def test_directory_compression_refuses_overlapping_roots(photo, nested):
+ source = photo.parent
+ output = {
+ "same": source,
+ "inside": source / "compressed",
+ "outside": source.parent,
+ }[nested]
+ before = photo.read_bytes()
+ with pytest.raises(ValueError, match="overlap"):
+ compressor.compress_files(source, output)
+ assert photo.read_bytes() == before
+
+
+@pytest.mark.parametrize("success", [True, False])
+def test_ffmpeg_only_writes_private_staging_output(tmp_path, monkeypatch, success):
+ source, output = tmp_path / "input.mp4", tmp_path / "output.mp4"
+ source.write_bytes(b"fixture video")
+
+ def ffmpeg(command, **kwargs):
+ if "-version" in command:
+ return CompletedProcess(command, 0)
+ temporary = Path(command[-1])
+ assert temporary != output and temporary != source
+ assert not output.exists()
+ temporary.write_bytes(b"encoded video" if success else b"partial")
+ return CompletedProcess(command, 0 if success else 1, stderr="encoder failed")
+
+ monkeypatch.setattr(compressor.subprocess, "run", ffmpeg)
+ assert compressor.compress_video(source, output) == success
+ assert source.read_bytes() == b"fixture video"
+ assert output.exists() == success
+ assert not list(tmp_path.glob(".clean-backup-*"))
+
+
+def test_ffmpeg_never_runs_for_existing_destination(tmp_path, monkeypatch):
+ source, output = tmp_path / "input.mp4", tmp_path / "output.mp4"
+ source.write_bytes(b"video")
+ output.write_bytes(b"keep existing output")
+
+ def unexpected(*args, **kwargs):
+ pytest.fail("FFmpeg must not run for an occupied target")
+
+ monkeypatch.setattr(compressor.subprocess, "run", unexpected)
+ assert not compressor.compress_video(source, output)
+ assert output.read_bytes() == b"keep existing output"
diff --git a/tests/test_file_operations.py b/tests/test_file_operations.py
new file mode 100644
index 0000000..b55edc1
--- /dev/null
+++ b/tests/test_file_operations.py
@@ -0,0 +1,111 @@
+"""Failure and concurrency regressions for the shared transfer boundary."""
+
+from concurrent.futures import ProcessPoolExecutor
+from pathlib import Path
+import shutil
+
+import pytest
+
+from src import file_operations as operations
+
+
+def _transfer(args):
+ source, target, operation = args
+ return str(operations.verified_transfer(source, target, operation))
+
+
+@pytest.mark.parametrize("operation", ["copy", "move"])
+def test_processes_cannot_replace_colliding_targets(tmp_path, operation):
+ sources = [tmp_path / f"source-{n}" for n in range(6)]
+ for n, source in enumerate(sources):
+ source.write_bytes(bytes([n]) * 4096)
+ target = tmp_path / "output" / "same.jpg"
+ with ProcessPoolExecutor(max_workers=3) as pool:
+ results = list(pool.map(_transfer, [(s, target, operation) for s in sources]))
+ assert len(set(results)) == len(sources)
+ assert {Path(p).read_bytes() for p in results} == {
+ bytes([n]) * 4096 for n in range(6)
+ }
+ assert all(s.exists() == (operation == "copy") for s in sources)
+
+
+@pytest.mark.parametrize("failure", ["disk_full", "corrupt_copy", "changed_source"])
+def test_failed_transfer_keeps_source_and_never_publishes(
+ tmp_path, monkeypatch, failure
+):
+ source, target = tmp_path / "source", tmp_path / "output" / "file.jpg"
+ source.write_bytes(b"original content")
+ real_copy = shutil.copy2
+
+ def faulty_copy(src, dst):
+ real_copy(src, dst)
+ if failure == "disk_full":
+ raise OSError("No space left on device")
+ if failure == "corrupt_copy":
+ Path(dst).write_bytes(b"corrupt bytes")
+ if failure == "changed_source":
+ Path(src).write_bytes(b"new user edit")
+
+ monkeypatch.setattr(operations.shutil, "copy2", faulty_copy)
+ with pytest.raises(OSError):
+ operations.verified_transfer(source, target, "move")
+ assert source.exists()
+ assert not target.exists()
+ assert list(target.parent.iterdir()) == []
+
+
+def test_symlink_and_same_file_are_refused(tmp_path):
+ source = tmp_path / "original"
+ source.write_text("keep")
+ alias = tmp_path / "alias"
+ alias.symlink_to(source)
+ with pytest.raises(OSError):
+ operations.verified_transfer(alias, tmp_path / "copy")
+ with pytest.raises(operations.FileConflictError):
+ operations.verified_transfer(source, alias)
+ assert source.read_text() == "keep"
+
+
+def test_existing_and_dangling_targets_are_never_replaced(tmp_path):
+ source, target = tmp_path / "source", tmp_path / "target"
+ source.write_text("new")
+ target.symlink_to(tmp_path / "missing")
+ with pytest.raises(FileExistsError):
+ operations.verified_transfer(source, target, collision="error")
+ actual = operations.verified_transfer(source, target)
+ assert actual != target
+ assert target.is_symlink()
+ assert actual.read_text() == "new"
+
+
+def test_unsupported_publication_fails_without_losing_source(tmp_path, monkeypatch):
+ source, target = tmp_path / "source", tmp_path / "target"
+ source.write_text("keep")
+
+ def unsupported(*args, **kwargs):
+ raise OSError("Hard links unsupported")
+
+ monkeypatch.setattr(operations.os, "link", unsupported)
+ with pytest.raises(OSError):
+ operations.verified_transfer(source, target, "move")
+ assert source.read_text() == "keep"
+ assert not target.exists()
+ assert not list(tmp_path.glob(".clean-backup-*"))
+
+
+def test_move_source_is_retained_if_it_changes_during_publication(
+ tmp_path, monkeypatch
+):
+ source, target = tmp_path / "source", tmp_path / "target"
+ source.write_text("original")
+ publish = operations._publish
+
+ def changed(temporary, destination):
+ publish(temporary, destination)
+ source.write_text("new edit")
+
+ monkeypatch.setattr(operations, "_publish", changed)
+ with pytest.raises(operations.FileConflictError):
+ operations.verified_transfer(source, target, "move")
+ assert source.read_text() == "new edit"
+ assert target.read_text() == "original"
diff --git a/tests/test_operation_recovery.py b/tests/test_operation_recovery.py
new file mode 100644
index 0000000..ca70a17
--- /dev/null
+++ b/tests/test_operation_recovery.py
@@ -0,0 +1,315 @@
+"""Exercise recovery preconditions and interrupted journal transitions."""
+
+from concurrent.futures import ThreadPoolExecutor
+import json
+import multiprocessing
+import os
+from contextlib import nullcontext
+from types import SimpleNamespace
+
+import pytest
+
+from src import file_operations as operations
+from src import undo_manager as journals
+
+
+def _crash_transfer(root, operation, stage):
+ """Exit without finally blocks to exercise the actual persisted intent."""
+ manager = journals.UndoManager(root / "journals")
+ manager.start_session()
+ if stage == "before_publication":
+ prepare = manager.prepare_action
+
+ def crash(*args):
+ prepare(*args)
+ os._exit(73)
+
+ manager.prepare_action = crash
+ elif stage == "after_publication":
+ publish = operations._publish
+
+ def crash(*args):
+ publish(*args)
+ os._exit(73)
+
+ operations._publish = crash
+ else:
+ manager.complete_action = lambda *args: os._exit(73)
+ operations.verified_transfer(
+ root / "source", root / "output", operation, journal=manager
+ )
+
+
+@pytest.mark.parametrize("operation", ["copy", "move"])
+@pytest.mark.parametrize(
+ "stage", ["before_publication", "after_publication", "before_completion"]
+)
+def test_recovery_after_process_exit(state, operation, stage):
+ source = state / "source"
+ source.write_text("original")
+ process = multiprocessing.get_context("spawn").Process(
+ target=_crash_transfer, args=(state, operation, stage)
+ )
+ process.start()
+ process.join(timeout=15)
+ if process.is_alive():
+ process.kill()
+ process.join()
+ pytest.fail("Crash probe did not exit")
+ assert process.exitcode == 73
+ manager = journals.UndoManager()
+ sessions = manager.list_sessions()
+ assert len(sessions) == 1
+ assert manager.undo_session(sessions[0]["path"])
+ assert source.read_text() == "original"
+ assert not (state / "output").exists()
+
+
+@pytest.fixture
+def state(tmp_path, monkeypatch):
+ monkeypatch.chdir(tmp_path)
+ monkeypatch.setattr(journals, "JOURNAL_DIR", tmp_path / "journals")
+ return tmp_path
+
+
+def transfer(state, operation="copy"):
+ source, target = state / "source", state / "out" / "file"
+ source.write_text("original")
+ with journals.UndoManager() as manager:
+ path = manager.journal_path
+ operations.verified_transfer(source, target, operation, journal=manager)
+ return source, target, manager, path
+
+
+@pytest.mark.parametrize("operation", ["copy", "move"])
+def test_undo_roundtrip(state, operation):
+ source, target, manager, path = transfer(state, operation)
+ assert manager.undo_session(path)
+ assert source.read_text() == "original"
+ assert not target.exists()
+ assert not path.exists()
+
+
+@pytest.mark.parametrize("operation", ["copy", "move"])
+def test_undo_retains_edited_output_and_journal(state, operation):
+ source, target, manager, path = transfer(state, operation)
+ target.write_text("new user edit")
+ assert not manager.undo_session(path)
+ assert target.read_text() == "new user edit"
+ assert path.exists()
+ assert manager.list_sessions()[0]["count"] == 1
+
+
+def test_undo_move_conflict_can_be_retried(state):
+ source, target, manager, path = transfer(state, "move")
+ source.write_text("new file at old location")
+ assert not manager.undo_session(path)
+ assert source.read_text() == "new file at old location"
+ assert target.read_text() == "original"
+ source.rename(state / "new-file-kept")
+ assert manager.undo_session(path)
+ assert source.read_text() == "original"
+ assert (state / "new-file-kept").read_text() == "new file at old location"
+
+
+@pytest.mark.parametrize("change", ["missing", "edited"])
+def test_copy_undo_keeps_last_original_content(state, change):
+ source, target, manager, path = transfer(state)
+ if change == "missing":
+ source.unlink()
+ else:
+ source.write_text("different content")
+ assert not manager.undo_session(path)
+ assert target.read_text() == "original"
+ assert path.exists()
+
+
+def test_intent_failure_prevents_publication_and_source_removal(state, monkeypatch):
+ source, target = state / "source", state / "target"
+ source.write_text("keep")
+ with journals.UndoManager() as manager:
+
+ def full_disk():
+ raise OSError("Journal disk full")
+
+ monkeypatch.setattr(manager, "_save_journal", full_disk)
+ with pytest.raises(OSError):
+ operations.verified_transfer(source, target, "move", journal=manager)
+ assert source.read_text() == "keep"
+ assert not target.exists()
+
+
+@pytest.mark.parametrize("operation", ["copy", "move"])
+def test_pending_intent_recovers_after_completion_write_failure(
+ state, monkeypatch, operation
+):
+ source, target = state / "source", state / "target"
+ source.write_text("original")
+ with journals.UndoManager() as manager:
+ path = manager.journal_path
+
+ def failed_commit(*args):
+ raise OSError("Completion write failed")
+
+ monkeypatch.setattr(manager, "complete_action", failed_commit)
+ with pytest.raises(OSError):
+ operations.verified_transfer(source, target, operation, journal=manager)
+ assert json.loads(path.read_text())[0]["state"] == "prepared"
+ assert target.read_text() == "original"
+ assert manager.undo_session(path)
+ assert source.read_text() == "original"
+ assert not target.exists()
+
+
+def test_partial_undo_does_not_replay_completed_actions(state):
+ with journals.UndoManager() as manager:
+ path = manager.journal_path
+ for name in ("a", "b"):
+ (state / name).write_text(name)
+ operations.verified_transfer(
+ state / name, state / "out" / name, "move", journal=manager
+ )
+ (state / "a").write_text("new occupant")
+ assert not manager.undo_session(path)
+ assert (state / "b").read_text() == "b"
+ (state / "a").unlink()
+ assert manager.undo_session(path)
+ assert (state / "a").read_text() == "a"
+ assert (state / "b").read_text() == "b"
+
+
+def test_legacy_path_only_journal_is_preserved_for_manual_review(state):
+ source, target = state / "source", state / "target"
+ target.write_text("possibly edited")
+ path = state / "legacy.json"
+ path.write_text(
+ json.dumps([{"action": "move", "src": str(source), "dst": str(target)}])
+ )
+ assert not journals.UndoManager().undo_session(path)
+ assert target.read_text() == "possibly edited"
+ assert not source.exists()
+ assert path.exists()
+
+
+def test_active_sessions_are_not_available_for_undo(state):
+ with journals.UndoManager() as manager:
+ (state / "a").write_text("a")
+ operations.verified_transfer(state / "a", state / "b", journal=manager)
+ assert manager.list_sessions() == []
+ assert not journals.UndoManager().undo_session(manager.journal_path)
+ path = manager.journal_path
+ assert manager.undo_session(path)
+
+
+def test_concurrent_session_ids_and_records_are_independent(state):
+ def run(n):
+ source = state / f"source-{n}"
+ source.write_text(str(n))
+ with journals.UndoManager() as manager:
+ path = manager.journal_path
+ operations.verified_transfer(source, state / f"target-{n}", journal=manager)
+ return path
+
+ with ThreadPoolExecutor(max_workers=4) as pool:
+ paths = list(pool.map(run, range(8)))
+ assert len(set(paths)) == 8
+ for path in paths:
+ records = json.loads(path.read_text())
+ assert len(records) == 1 and records[0]["state"] == "completed"
+
+
+@pytest.mark.parametrize("operation", ["copy", "move"])
+def test_organizer_keeps_distinct_names_and_colliding_basenames(state, operation):
+ from PIL import Image
+ from src.organiser import organise_files
+
+ source, output = state / "source", state / "out"
+ paths = [
+ source / "a" / "photo.jpg",
+ source / "b" / "photo.jpg",
+ source / "a" / "photo (1).jpg",
+ ]
+ for path, color in zip(paths, ("red", "blue", "green")):
+ path.parent.mkdir(parents=True, exist_ok=True)
+ Image.new("RGB", (30, 30), color).save(path)
+ expected = {p.read_bytes() for p in paths}
+ stats = organise_files(source, output, operation, check_name_duplicates=True)
+ assert stats["processed"] == 3 and stats["errors"] == 0
+ assert {p.read_bytes() for p in output.rglob("*.jpg")} == expected
+ manager = journals.UndoManager()
+ assert manager.undo_session(manager.list_sessions()[0]["path"])
+ assert {p.read_bytes() for p in paths} == expected
+
+
+@pytest.mark.parametrize("nested", ["same", "inside", "outside"])
+def test_organizer_rejects_overlapping_roots_before_mutation(state, nested):
+ from src.organiser import organise_files
+
+ source = state / "photos"
+ source.mkdir()
+ target = {"same": source, "inside": source / "out", "outside": state}[nested]
+ with pytest.raises(ValueError, match="overlap"):
+ organise_files(source, target)
+ assert not list(state.rglob("journal_*.json"))
+
+
+def test_legacy_duplicate_report_releases_its_session(state, monkeypatch):
+ from src import duplicate_handler
+
+ managers = []
+
+ def manager_factory():
+ manager = journals.UndoManager()
+ managers.append(manager)
+ return manager
+
+ monkeypatch.setattr(duplicate_handler, "UndoManager", manager_factory)
+ monkeypatch.setattr(duplicate_handler, "scan_for_duplicates", lambda *args: [])
+ for _ in range(2):
+ duplicate_handler.handle_duplicates(str(state), action="report")
+ assert all(m.session_id is None and m._session_fd is None for m in managers)
+
+
+@pytest.mark.parametrize("callback_failure", [False, True])
+def test_legacy_classification_releases_journal_on_return_or_error(
+ state, monkeypatch, callback_failure
+):
+ from src.classify import apply_org
+
+ source = state / "photo.jpg"
+ source.write_bytes(b"original fixture")
+ manager = journals.UndoManager()
+ monkeypatch.setattr(apply_org, "UndoManager", lambda: manager)
+ monkeypatch.setattr(
+ apply_org, "get_run_config", lambda _: {"multi_category": "tags"}
+ )
+ monkeypatch.setattr(
+ apply_org,
+ "get_all_tags_for_run",
+ lambda _: [
+ {"path": str(source), "category_label": "example"},
+ ],
+ )
+ cursor = SimpleNamespace(execute=lambda *args: None, fetchone=lambda: None)
+ monkeypatch.setattr(
+ apply_org,
+ "_get_conn",
+ lambda: nullcontext(SimpleNamespace(cursor=lambda: cursor)),
+ )
+
+ def failed_callback(*args):
+ raise RuntimeError("Callback failed")
+
+ if callback_failure:
+ with pytest.raises(RuntimeError, match="Callback failed"):
+ apply_org.apply_classification(
+ "test", state / "out", "copy", failed_callback
+ )
+ else:
+ result = apply_org.apply_classification("test", state / "out", "copy")
+ assert result["session_id"]
+ assert manager.session_id is None and manager._session_fd is None
+ sessions = manager.list_sessions()
+ assert len(sessions) == 1
+ assert manager.undo_session(sessions[0]["path"])
+ assert source.read_bytes() == b"original fixture"
diff --git a/web/src/App.jsx b/web/src/App.jsx
index effbd4d..b65720f 100644
--- a/web/src/App.jsx
+++ b/web/src/App.jsx
@@ -118,7 +118,6 @@ export default function App() {
const [orgDestination, setOrgDestination] = useState("");
const [orgOperation, setOrgOperation] = useState("move");
const [orgCheckDuplicates, setOrgCheckDuplicates] = useState(false);
- const [orgCheckNameDuplicates, setOrgCheckNameDuplicates] = useState(false);
const [orgThreshold, setOrgThreshold] = useState(5);
const [orgJobId, setOrgJobId] = useState(null);
const [orgJob, setOrgJob] = useState(null);
@@ -555,7 +554,7 @@ export default function App() {
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
source_dir: sourceDir.trim(),
- threshold: Number(threshold) || 5
+ threshold: threshold === "" ? 5 : Number(threshold)
})
});
const data = await res.json();
@@ -682,8 +681,7 @@ export default function App() {
destination_dir: orgDestination.trim(),
operation: orgOperation,
check_duplicates: orgCheckDuplicates,
- duplicate_threshold: Number(orgThreshold) || 5,
- check_name_duplicates: orgCheckNameDuplicates
+ duplicate_threshold: orgThreshold === "" ? 5 : Number(orgThreshold)
})
});
const data = await res.json();
@@ -1273,10 +1271,7 @@ export default function App() {
setOrgCheckDuplicates(e.target.checked)} />
Check perceptual duplicates
-
+ Files with similar names are retained. Naming conflicts receive a unique suffix.
@@ -1291,7 +1286,6 @@ export default function App() {
Images{orgResult.stats.images}
Videos{orgResult.stats.videos}
Perceptual duplicates{orgResult.stats.perceptual_duplicates}
- Name duplicates{orgResult.stats.name_duplicates}
) : null}