diff --git a/backend/app/database/images.py b/backend/app/database/images.py
index 0d2f1ad7f..f84323dd5 100644
--- a/backend/app/database/images.py
+++ b/backend/app/database/images.py
@@ -128,6 +128,14 @@ def db_create_images_table() -> None:
     if "score" not in {row[1] for row in cursor.fetchall()}:
         cursor.execute("ALTER TABLE image_classes ADD COLUMN score REAL")
 
+    # isMetadataSynced: whether the file on disk carries the metadata we hold.
+    # Guarded for the same reason as score -- shipped databases predate it.
+    cursor.execute("PRAGMA table_info(images)")
+    if "isMetadataSynced" not in {row[1] for row in cursor.fetchall()}:
+        cursor.execute(
+            "ALTER TABLE images ADD COLUMN isMetadataSynced BOOLEAN DEFAULT 0"
+        )
+
     conn.commit()
     conn.close()
 
@@ -159,6 +167,9 @@ def db_bulk_insert_images(image_records: List[ImageRecord]) -> bool:
                 END,
                 latitude=COALESCE(excluded.latitude, images.latitude),
                 longitude=COALESCE(excluded.longitude, images.longitude),
+                -- Only changed files reach this branch, and a file that changed
+                -- underneath us may no longer carry the metadata we wrote.
+                isMetadataSynced=0,
                 -- Not COALESCE: every record here comes from a full re-read of
                 -- the file, so NULL means "no capture date exists" and has to
                 -- overwrite a bad one a previous extractor guessed.
@@ -586,7 +597,10 @@ def db_toggle_image_favourite_status(image_id: str) -> bool:
         cursor.execute(
             """
             UPDATE images
-            SET isFavourite = CASE WHEN isFavourite = 1 THEN 0 ELSE 1 END
+            SET isFavourite = CASE WHEN isFavourite = 1 THEN 0 ELSE 1 END,
+                -- The rating in the file no longer matches, so the photo goes
+                -- back in the queue for the next metadata pass.
+                isMetadataSynced = 0
             WHERE id = ?
             """,
             (image_id,),
diff --git a/backend/app/database/metadata_sync.py b/backend/app/database/metadata_sync.py
new file mode 100644
index 000000000..6feb17b67
--- /dev/null
+++ b/backend/app/database/metadata_sync.py
@@ -0,0 +1,246 @@
+"""
+Which photos still need their metadata written into the file, and what to write.
+
+The database stays the fast index; the file is what survives PictoPy. These
+queries pull together the pieces the packet is built from, none of which live
+in one table.
+"""
+
+import json
+import sqlite3
+from typing import Any, Dict, List, Mapping, Optional, Tuple, TypedDict
+
+from app.config.settings import DATABASE_PATH
+from app.logging.setup_logging import get_logger
+
+logger = get_logger(__name__)
+
+ImageId = str
+
+
+class SyncCandidate(TypedDict):
+    """Everything the packet for one photo is built from."""
+
+    id: ImageId
+    path: str
+    metadata: Mapping[str, Any]
+    is_favourite: bool
+    keywords: List[str]
+    faces: List[Dict[str, Any]]
+
+
+def _connect() -> sqlite3.Connection:
+    conn = sqlite3.connect(DATABASE_PATH)
+    conn.execute("PRAGMA foreign_keys = ON")
+    return conn
+
+
+def _parse_json(raw: Optional[str], fallback: Any) -> Any:
+    if not raw:
+        return fallback
+    try:
+        return json.loads(raw)
+    except (json.JSONDecodeError, TypeError):
+        return fallback
+
+
+def db_get_images_pending_metadata_sync(limit: int = 200) -> List[SyncCandidate]:
+    """
+    Photos whose file does not yet carry what the database knows.
+
+    Only tagged photos are returned: writing before the AI pass has run would
+    put an empty keyword list into the file and need a second write anyway.
+    """
+    conn = _connect()
+    cursor = conn.cursor()
+
+    try:
+        cursor.execute(
+            """
+            SELECT id, path, metadata, isFavourite
+            FROM images
+            WHERE isMetadataSynced = 0 AND isTagged = 1
+            ORDER BY id
+            LIMIT ?
+            """,
+            (limit,),
+        )
+        rows = cursor.fetchall()
+        if not rows:
+            return []
+
+        candidates: Dict[ImageId, SyncCandidate] = {}
+        for image_id, path, metadata, is_favourite in rows:
+            candidates[image_id] = SyncCandidate(
+                id=image_id,
+                path=path,
+                metadata=_parse_json(metadata, {}),
+                is_favourite=bool(is_favourite),
+                keywords=[],
+                faces=[],
+            )
+
+        placeholders = ",".join("?" for _ in candidates)
+        identifiers = list(candidates)
+
+        # Tags come through the display view so the file gets the same labels
+        # the gallery shows, rather than every low-scoring semantic match.
+        cursor.execute(
+            f"""
+            SELECT d.image_id, m.name
+            FROM image_classes_display d
+            JOIN mappings m ON m.class_id = d.class_id
+            WHERE d.image_id IN ({placeholders})
+            ORDER BY m.name
+            """,
+            identifiers,
+        )
+        for image_id, name in cursor.fetchall():
+            if name:
+                candidates[image_id]["keywords"].append(name)
+
+        # Only named clusters: an unnamed one carries no information a person
+        # reading the file in another application could use.
+        cursor.execute(
+            f"""
+            SELECT f.image_id, f.bbox, c.cluster_name
+            FROM faces f
+            JOIN face_clusters c ON c.cluster_id = f.cluster_id
+            WHERE f.image_id IN ({placeholders})
+              AND c.cluster_name IS NOT NULL
+              AND TRIM(c.cluster_name) != ''
+            ORDER BY c.cluster_name
+            """,
+            identifiers,
+        )
+        for image_id, bbox, cluster_name in cursor.fetchall():
+            box = _parse_json(bbox, None)
+            if isinstance(box, dict):
+                candidates[image_id]["faces"].append(
+                    {"name": cluster_name, "bbox": box}
+                )
+
+        return list(candidates.values())
+    except sqlite3.Error as e:
+        logger.error(f"Error collecting images pending metadata sync: {e}")
+        return []
+    finally:
+        conn.close()
+
+
+def db_count_images_pending_metadata_sync() -> int:
+    """How many photos are waiting for their file to be brought up to date."""
+    conn = _connect()
+    cursor = conn.cursor()
+
+    try:
+        cursor.execute(
+            "SELECT COUNT(*) FROM images WHERE isMetadataSynced = 0 AND isTagged = 1"
+        )
+        return cursor.fetchone()[0]
+    except sqlite3.Error as e:
+        logger.error(f"Error counting images pending metadata sync: {e}")
+        return 0
+    finally:
+        conn.close()
+
+
+def db_mark_metadata_synced(written: List[Tuple[ImageId, int, int]]) -> int:
+    """
+    Record that a photo's file now carries its metadata, and how it now looks.
+
+    The size and mtime go back into the metadata blob on purpose. Our own write
+    changes both, and without this the next folder scan would see the file as
+    user-modified, re-read it, and mark it for another write -- a loop the file
+    itself keeps feeding.
+    """
+    if not written:
+        return 0
+
+    conn = _connect()
+    cursor = conn.cursor()
+
+    try:
+        updated = 0
+        for image_id, file_size, file_mtime in written:
+            cursor.execute("SELECT metadata FROM images WHERE id = ?", (image_id,))
+            row = cursor.fetchone()
+            if row is None:
+                continue
+
+            metadata = _parse_json(row[0], {})
+            if not isinstance(metadata, dict):
+                metadata = {}
+            metadata["file_size"] = file_size
+            metadata["file_mtime"] = file_mtime
+
+            cursor.execute(
+                """
+                UPDATE images
+                SET isMetadataSynced = 1, metadata = ?
+                WHERE id = ?
+                """,
+                (json.dumps(metadata), image_id),
+            )
+            updated += cursor.rowcount
+
+        conn.commit()
+        return updated
+    except sqlite3.Error as e:
+        logger.error(f"Error marking metadata synced: {e}")
+        conn.rollback()
+        return 0
+    finally:
+        conn.close()
+
+
+def db_mark_metadata_dirty(image_ids: List[ImageId]) -> int:
+    """Flag photos whose file no longer matches what the database holds."""
+    if not image_ids:
+        return 0
+
+    conn = _connect()
+    cursor = conn.cursor()
+
+    try:
+        placeholders = ",".join("?" for _ in image_ids)
+        cursor.execute(
+            f"UPDATE images SET isMetadataSynced = 0 WHERE id IN ({placeholders})",
+            image_ids,
+        )
+        conn.commit()
+        return cursor.rowcount
+    except sqlite3.Error as e:
+        logger.error(f"Error marking metadata dirty: {e}")
+        conn.rollback()
+        return 0
+    finally:
+        conn.close()
+
+
+def db_mark_metadata_dirty_for_cluster(cluster_id: str) -> int:
+    """
+    Flag every photo showing a given person.
+
+    Renaming a cluster changes one row but invalidates the region name written
+    into every file that person appears in.
+    """
+    conn = _connect()
+    cursor = conn.cursor()
+
+    try:
+        cursor.execute(
+            """
+            UPDATE images SET isMetadataSynced = 0
+            WHERE id IN (SELECT image_id FROM faces WHERE cluster_id = ?)
+            """,
+            (cluster_id,),
+        )
+        conn.commit()
+        return cursor.rowcount
+    except sqlite3.Error as e:
+        logger.error(f"Error marking cluster {cluster_id} dirty: {e}")
+        conn.rollback()
+        return 0
+    finally:
+        conn.close()
diff --git a/backend/app/database/self_writes.py b/backend/app/database/self_writes.py
new file mode 100644
index 000000000..f1bfd8101
--- /dev/null
+++ b/backend/app/database/self_writes.py
@@ -0,0 +1,154 @@
+"""
+A short-lived record of files PictoPy modified itself.
+
+The sync microservice watches every registered folder and treats any file change
+as a user edit, so a write of our own comes straight back as a full folder
+resync. Recording what we wrote lets the watcher tell the two apart.
+
+The path format here is a contract with `sync-microservice/app/database/
+self_writes.py`, which reads this table: both sides key on
+`os.path.normcase(os.path.abspath(path))`.
+"""
+
+import os
+import sqlite3
+import time
+from typing import List, Set, Tuple
+
+from app.config.settings import DATABASE_PATH
+from app.logging.setup_logging import get_logger
+
+logger = get_logger(__name__)
+
+# An entry the watcher never claims is one it missed -- it was stopped, or the
+# change was coalesced away. Expiring them keeps the table bounded and limits
+# how long a stale row can mask a real edit to the same path.
+SELF_WRITE_TTL_SECONDS = 3600
+
+# (path, file_size, file_mtime) as observed on disk right now.
+ObservedFile = Tuple[str, int, int]
+
+
+def self_write_key(path: str) -> str:
+    """Normalise a path to the form both services store and look up by."""
+    return os.path.normcase(os.path.abspath(path))
+
+
+def _connect() -> sqlite3.Connection:
+    conn = sqlite3.connect(DATABASE_PATH)
+    # This table stands alone, keyed by path, so nothing here depends on it
+    # today. Set anyway to match every other _connect() in this package.
+    conn.execute("PRAGMA foreign_keys = ON")
+    return conn
+
+
+def db_create_self_writes_table() -> None:
+    conn = _connect()
+    cursor = conn.cursor()
+    try:
+        cursor.execute(
+            """
+            CREATE TABLE IF NOT EXISTS self_writes (
+                path TEXT PRIMARY KEY,
+                file_size INTEGER NOT NULL,
+                file_mtime INTEGER NOT NULL,
+                -- Epoch seconds rather than CURRENT_TIMESTAMP: this column only
+                -- exists to be subtracted from, and text timestamps make that
+                -- arithmetic a timezone question.
+                written_at INTEGER NOT NULL
+            )
+        """
+        )
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def db_record_self_write(path: str, file_size: int, file_mtime: int) -> bool:
+    """
+    Note that PictoPy is about to leave a file in this exact state.
+
+    Callers record before the bytes land, because the watcher can fire the
+    instant they do.
+    """
+    conn = _connect()
+    cursor = conn.cursor()
+    now = int(time.time())
+
+    try:
+        cursor.execute(
+            """
+            INSERT INTO self_writes (path, file_size, file_mtime, written_at)
+            VALUES (?, ?, ?, ?)
+            ON CONFLICT(path) DO UPDATE SET
+                file_size=excluded.file_size,
+                file_mtime=excluded.file_mtime,
+                written_at=excluded.written_at
+            """,
+            (self_write_key(path), file_size, file_mtime, now),
+        )
+        # Pruning here avoids needing a scheduler for a table this small.
+        cursor.execute(
+            "DELETE FROM self_writes WHERE written_at < ?",
+            (now - SELF_WRITE_TTL_SECONDS,),
+        )
+        conn.commit()
+        return True
+    except sqlite3.Error as e:
+        logger.error(f"Error recording self write for {path}: {e}")
+        conn.rollback()
+        return False
+    finally:
+        conn.close()
+
+
+def db_take_matching_self_writes(observed: List[ObservedFile]) -> Set[str]:
+    """
+    Return the observed paths that match a recorded write, and forget them.
+
+    Claiming an entry as it matches means a second event for the same write is
+    treated as a real change. That is the safe direction to be wrong in: it
+    costs one redundant rescan, where holding the entry could swallow a genuine
+    edit that happened to land in the same second at the same size.
+    """
+    if not observed:
+        return set()
+
+    conn = _connect()
+    cursor = conn.cursor()
+    matched: Set[str] = set()
+
+    try:
+        by_key = {self_write_key(path): path for path, _, _ in observed}
+        placeholders = ",".join("?" for _ in observed)
+        cursor.execute(
+            f"""
+            SELECT path, file_size, file_mtime
+            FROM self_writes
+            WHERE path IN ({placeholders})
+            """,
+            list(by_key),
+        )
+        recorded = {row[0]: (row[1], row[2]) for row in cursor.fetchall()}
+
+        claimed = []
+        for path, size, mtime in observed:
+            key = self_write_key(path)
+            if recorded.get(key) == (size, mtime):
+                matched.add(path)
+                claimed.append(key)
+
+        if claimed:
+            cursor.executemany(
+                "DELETE FROM self_writes WHERE path = ?",
+                [(key,) for key in claimed],
+            )
+            conn.commit()
+
+        return matched
+    except sqlite3.Error as e:
+        # Failing open means a redundant rescan, never a missed change.
+        logger.error(f"Error matching self writes: {e}")
+        return set()
+    finally:
+        conn.close()
diff --git a/backend/app/routes/face_clusters.py b/backend/app/routes/face_clusters.py
index 4749b2ebf..5e3e6c3a6 100644
--- a/backend/app/routes/face_clusters.py
+++ b/backend/app/routes/face_clusters.py
@@ -13,6 +13,7 @@
     db_get_images_by_cluster_id,
     db_get_images_by_face_clusters,
 )
+from app.database.metadata_sync import db_mark_metadata_dirty_for_cluster
 from starlette.datastructures import State
 
 from app.routes.dependencies import get_state
@@ -126,6 +127,10 @@ def rename_cluster(
         # on stale inputs.
         _rescore_memories_for_cluster(app_state, cluster_id)
 
+        # Same reach for the files themselves: the region written into every
+        # photo this person appears in still carries the old name.
+        db_mark_metadata_dirty_for_cluster(cluster_id)
+
         return RenameClusterResponse(
             success=True,
             message=f"Successfully renamed cluster to '{request.cluster_name}'",
diff --git a/backend/app/routes/folders.py b/backend/app/routes/folders.py
index c27e789a2..b99c5ef3b 100644
--- a/backend/app/routes/folders.py
+++ b/backend/app/routes/folders.py
@@ -46,6 +46,7 @@
     folder_util_get_filesystem_direct_child_folders,
 )
 from concurrent.futures import ProcessPoolExecutor
+from app.utils.metadata_sync import metadata_util_sync_pending
 from app.utils.images import (
     image_util_process_folder_images,
     image_util_process_untagged_images,
@@ -150,6 +151,8 @@ def post_AI_tagging_enabled_sequence():
         # Curate before the video pass: semantic labels are written by now,
         # and the video pass can run for minutes.
         _curate_memories("ai_tagging")
+        # Last of the photo passes: it writes out what all of them produced.
+        metadata_util_sync_pending()
         # Videos last: photos are the primary surface, so they finish first.
         video_util_process_untagged_videos()
         video_util_process_unembedded_frames()
@@ -191,6 +194,7 @@ def post_sync_folder_sequence(
         image_util_process_unembedded_images()
         semantic_util_score_images()
         _curate_memories("sync_folder")
+        metadata_util_sync_pending()
         video_util_process_untagged_videos()
         video_util_process_unembedded_frames()
         semantic_util_score_videos()
diff --git a/backend/app/routes/metadata_sync.py b/backend/app/routes/metadata_sync.py
new file mode 100644
index 000000000..af56b1762
--- /dev/null
+++ b/backend/app/routes/metadata_sync.py
@@ -0,0 +1,86 @@
+from fastapi import APIRouter, HTTPException, Query, status
+
+from app.database.metadata_sync import db_count_images_pending_metadata_sync
+from app.logging.setup_logging import get_logger
+from app.schemas.metadata_sync import (
+    ErrorResponse,
+    GetMetadataSyncStatusResponse,
+    MetadataSyncResult,
+    MetadataSyncStatus,
+    RunMetadataSyncResponse,
+)
+from app.utils.metadata_sync import (
+    metadata_util_sync_pending,
+    metadata_util_write_to_files_enabled,
+)
+
+logger = get_logger(__name__)
+
+router = APIRouter()
+
+
+@router.get(
+    "/status",
+    response_model=GetMetadataSyncStatusResponse,
+    responses={500: {"model": ErrorResponse}},
+)
+def get_metadata_sync_status():
+    """How many photos are waiting to have their metadata written to disk."""
+    try:
+        return GetMetadataSyncStatusResponse(
+            success=True,
+            message="Successfully retrieved metadata sync status",
+            data=MetadataSyncStatus(
+                enabled=metadata_util_write_to_files_enabled(),
+                pending=db_count_images_pending_metadata_sync(),
+            ),
+        )
+    except Exception as e:
+        logger.error(f"Error retrieving metadata sync status: {e}")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail=ErrorResponse(
+                success=False,
+                error="Internal Server Error",
+                message=f"Unable to retrieve metadata sync status: {e}",
+            ).model_dump(),
+        )
+
+
+@router.post(
+    "/run",
+    response_model=RunMetadataSyncResponse,
+    responses={500: {"model": ErrorResponse}},
+)
+def run_metadata_sync(limit: int = Query(default=200, ge=1, le=5000)):
+    """
+    Write pending metadata into the photo files.
+
+    Does nothing unless the user has turned file writing on; the response still
+    reports what was considered so the caller can tell the difference.
+    """
+    try:
+        summary = metadata_util_sync_pending(limit)
+        return RunMetadataSyncResponse(
+            success=True,
+            message=(
+                "Metadata sync complete"
+                if summary.get("disabled") is None
+                else "Writing metadata to files is turned off"
+            ),
+            data=MetadataSyncResult(
+                considered=summary["considered"],
+                written=summary["written"],
+                skipped=summary["skipped"],
+            ),
+        )
+    except Exception as e:
+        logger.error(f"Error running metadata sync: {e}")
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail=ErrorResponse(
+                success=False,
+                error="Internal Server Error",
+                message=f"Unable to run metadata sync: {e}",
+            ).model_dump(),
+        )
diff --git a/backend/app/schemas/metadata_sync.py b/backend/app/schemas/metadata_sync.py
new file mode 100644
index 000000000..00a957f7c
--- /dev/null
+++ b/backend/app/schemas/metadata_sync.py
@@ -0,0 +1,36 @@
+from pydantic import BaseModel
+
+
+class MetadataSyncStatus(BaseModel):
+    """How far behind the files are, and whether writing to them is allowed."""
+
+    enabled: bool
+    pending: int
+
+
+class GetMetadataSyncStatusResponse(BaseModel):
+    success: bool
+    message: str
+    data: MetadataSyncStatus
+
+
+class MetadataSyncResult(BaseModel):
+    """What one pass did. `skipped` counts photos left for a later attempt."""
+
+    considered: int
+    written: int
+    skipped: int
+
+
+class RunMetadataSyncResponse(BaseModel):
+    success: bool
+    message: str
+    data: MetadataSyncResult
+
+
+class ErrorResponse(BaseModel):
+    """Error response model"""
+
+    success: bool
+    error: str
+    message: str
diff --git a/backend/app/schemas/user_preferences.py b/backend/app/schemas/user_preferences.py
index 9f00c94ac..1b398f0bf 100644
--- a/backend/app/schemas/user_preferences.py
+++ b/backend/app/schemas/user_preferences.py
@@ -65,6 +65,14 @@ def _check_bounds(self) -> "MemoriesPreferences":
         return self
 
 
+class MetadataPreferences(BaseModel):
+    """Writing PictoPy's metadata into the photo files themselves."""
+
+    # Off unless asked for. Every other feature reads the user's files; this one
+    # rewrites them, and that is not a default worth assuming.
+    write_to_files: bool = False
+
+
 class UserPreferencesData(BaseModel):
     """User preferences data structure"""
 
@@ -78,6 +86,7 @@ class UserPreferencesData(BaseModel):
         le=VIDEO_FRAME_INTERVAL_MAX,
     )
     memories: MemoriesPreferences = Field(default_factory=MemoriesPreferences)
+    metadata: MetadataPreferences = Field(default_factory=MetadataPreferences)
 
 
 class GetUserPreferencesResponse(BaseModel):
@@ -118,6 +127,12 @@ class MemoriesPreferencesUpdate(BaseModel):
     weights: Optional[MemoryScoringWeightsUpdate] = None
 
 
+class MetadataPreferencesUpdate(BaseModel):
+    """Partial update for metadata preferences."""
+
+    write_to_files: Optional[bool] = None
+
+
 class UpdateUserPreferencesRequest(BaseModel):
     """Request model for updating user preferences"""
 
@@ -127,6 +142,7 @@ class UpdateUserPreferencesRequest(BaseModel):
         default=None, ge=VIDEO_FRAME_INTERVAL_MIN, le=VIDEO_FRAME_INTERVAL_MAX
     )
     memories: Optional[MemoriesPreferencesUpdate] = None
+    metadata: Optional[MetadataPreferencesUpdate] = None
 
 
 class UpdateUserPreferencesResponse(BaseModel):
diff --git a/backend/app/utils/images.py b/backend/app/utils/images.py
index 0eafb5fdd..fd6493b90 100644
--- a/backend/app/utils/images.py
+++ b/backend/app/utils/images.py
@@ -590,6 +590,22 @@ def _extract_gps_coordinates(exif_data: Any) -> Tuple[float | None, float | None
 # Pointer to the EXIF sub-IFD, where the capture timestamps actually live.
 EXIF_IFD_POINTER = 0x8769
 
+# How the camera was held. Face boxes are recorded in the undecoded pixel space,
+# so anything placing them against the displayed image needs this to undo it.
+ORIENTATION_TAG = 0x0112
+
+
+def _extract_orientation(exif_data: Any) -> int:
+    """Read the EXIF orientation flag, defaulting to upright."""
+    if exif_data is None:
+        return 1
+    try:
+        value = int(exif_data.get(ORIENTATION_TAG) or 1)
+    except (AttributeError, TypeError, ValueError):
+        return 1
+    return value if 1 <= value <= 8 else 1
+
+
 # Preference order. DateTimeOriginal is when the shutter fired; DateTime is
 # the file's own timestamp and can be a later edit, so it comes last.
 CAPTURE_DATE_TAGS = ("DateTimeOriginal", "DateTimeDigitized", "DateTime")
@@ -683,6 +699,7 @@ def image_util_extract_metadata(image_path: str) -> dict:
 
                 dt_original = _extract_capture_datetime(exif_data)
                 latitude, longitude = _extract_gps_coordinates(exif_data)
+                orientation = _extract_orientation(exif_data)
 
                 # A Google Takeout export drops EXIF from part of its own
                 # library and keeps the real values in a sibling JSON file.
@@ -726,6 +743,7 @@ def image_util_extract_metadata(image_path: str) -> dict:
                 # coarser mtime would otherwise never compare equal to its own
                 # float from a previous scan.
                 "file_mtime": int(stats.st_mtime),
+                "orientation": orientation,
                 "item_type": mime_type,
             }
 
diff --git a/backend/app/utils/metadata_sync.py b/backend/app/utils/metadata_sync.py
new file mode 100644
index 000000000..fd247c2f8
--- /dev/null
+++ b/backend/app/utils/metadata_sync.py
@@ -0,0 +1,223 @@
+"""
+Writing what PictoPy knows about a photo into the photo itself.
+
+Runs as a batch rather than on every change: each write touches one of the
+user's original files, and doing that once per edit would multiply the risk for
+no benefit. The database stays authoritative for speed; the file is what
+survives PictoPy being uninstalled.
+"""
+
+import datetime
+import os
+from typing import Any, Dict, List, Mapping, Optional
+
+from app.database.metadata import db_get_metadata
+from app.database.metadata_sync import (
+    SyncCandidate,
+    db_count_images_pending_metadata_sync,
+    db_get_images_pending_metadata_sync,
+    db_mark_metadata_synced,
+)
+from app.logging.setup_logging import get_logger
+from app.schemas.user_preferences import MetadataPreferences
+from app.utils.self_write import self_write_util_replace
+from app.utils.xmp_packet import (
+    FaceRegion,
+    PhotoMetadata,
+    UnreadablePacketError,
+    xmp_packet_applied_dimensions,
+    xmp_packet_build,
+    xmp_packet_orient_region,
+)
+from app.utils.xmp_segments import (
+    UnsupportedImageError,
+    xmp_segments_read,
+    xmp_segments_write,
+)
+
+logger = get_logger(__name__)
+
+# A favourite is a flag, but xmp:Rating is a scale, so it maps to the top of it.
+FAVOURITE_RATING = 5
+
+# Keeps our keywords identifiable in a tree a user may already be organising by
+# hand, which a flat name would not be.
+PERSON_KEYWORD_ROOT = "People"
+
+
+def _as_int(value: Any, fallback: int = 0) -> int:
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return fallback
+
+
+def metadata_util_build_packet_metadata(
+    candidate: SyncCandidate, written_at: Optional[str] = None
+) -> PhotoMetadata:
+    """
+    Turn one photo's database rows into the properties its packet should carry.
+
+    Only keys PictoPy is prepared to own are set. A key left out is a property
+    left alone in the file, which is how another application's work survives.
+    """
+    blob: Mapping[str, Any] = candidate.get("metadata") or {}
+    raw_width = _as_int(blob.get("width"))
+    raw_height = _as_int(blob.get("height"))
+    orientation = _as_int(blob.get("orientation"), 1)
+
+    people = [face["name"] for face in candidate["faces"]]
+    keywords = list(candidate["keywords"])
+
+    metadata: PhotoMetadata = {
+        "keywords": keywords,
+        "hierarchical_keywords": [
+            f"{PERSON_KEYWORD_ROOT}|{name}" for name in dict.fromkeys(people)
+        ],
+        "written_at": written_at
+        or datetime.datetime.now().isoformat(timespec="seconds"),
+    }
+
+    # Deliberately not written when the photo is not a favourite. PictoPy only
+    # has a yes/no, so claiming the rating either way would let un-favouriting
+    # erase a star rating the user set somewhere else.
+    if candidate["is_favourite"]:
+        metadata["rating"] = FAVOURITE_RATING
+
+    regions: List[FaceRegion] = []
+    for face in candidate["faces"]:
+        placed = xmp_packet_orient_region(
+            face["bbox"], raw_width, raw_height, orientation
+        )
+        if placed is None:
+            continue
+        center_x, center_y, width, height = placed
+        regions.append(
+            FaceRegion(
+                name=face["name"],
+                center_x=center_x,
+                center_y=center_y,
+                width=width,
+                height=height,
+            )
+        )
+
+    # Regions are meaningless without the frame they are measured against, so
+    # both are written together or neither is.
+    if regions and raw_width and raw_height:
+        applied_width, applied_height = xmp_packet_applied_dimensions(
+            raw_width, raw_height, orientation
+        )
+        metadata["regions"] = regions
+        metadata["applied_width"] = applied_width
+        metadata["applied_height"] = applied_height
+
+    return metadata
+
+
+def metadata_util_write_one(candidate: SyncCandidate) -> Optional[Dict[str, int]]:
+    """
+    Write one photo's packet into its file.
+
+    Returns the file's new size and mtime, or None if it was left untouched.
+    Every failure here is a skip rather than a raise: one unreadable photo must
+    not stop the pass, and the file is always left exactly as it was.
+    """
+    path = candidate["path"]
+
+    try:
+        with open(path, "rb") as handle:
+            original = handle.read()
+    except OSError as e:
+        logger.warning(f"Could not read {path} for metadata sync: {e}")
+        return None
+
+    try:
+        existing = xmp_segments_read(original)
+    except UnsupportedImageError as e:
+        logger.warning(f"Cannot write metadata into {path}: {e}")
+        return None
+
+    try:
+        packet = xmp_packet_build(
+            metadata_util_build_packet_metadata(candidate), existing=existing
+        )
+    except UnreadablePacketError as e:
+        # The photo carries a packet we cannot parse, so we cannot know what
+        # replacing it would destroy. Leaving it alone is the point.
+        logger.warning(f"Leaving {path} alone: {e}")
+        return None
+
+    try:
+        updated = xmp_segments_write(original, packet)
+    except (UnsupportedImageError, ValueError) as e:
+        logger.warning(f"Cannot embed metadata in {path}: {e}")
+        return None
+
+    if updated == original:
+        # Nothing changed, so there is no reason to rewrite the user's file.
+        stats = os.stat(path)
+        return {"file_size": stats.st_size, "file_mtime": int(stats.st_mtime)}
+
+    if not self_write_util_replace(path, updated):
+        return None
+
+    stats = os.stat(path)
+    return {"file_size": stats.st_size, "file_mtime": int(stats.st_mtime)}
+
+
+def metadata_util_write_to_files_enabled() -> bool:
+    """
+    Whether the user has asked PictoPy to write into their originals.
+
+    Read-only, like the memories curator: db_update_metadata rewrites the whole
+    blob, so writing from here would clobber a concurrent settings save.
+    """
+    metadata = db_get_metadata() or {}
+    stored = (metadata.get("user_preferences") or {}).get("metadata") or {}
+    try:
+        return MetadataPreferences.model_validate(stored).write_to_files
+    except ValueError as e:
+        logger.warning(f"Invalid metadata preferences, not writing to files: {e}")
+        return False
+
+
+def metadata_util_sync_pending(limit: int = 200) -> Dict[str, int]:
+    """
+    Write metadata into every photo whose file is behind the database.
+
+    Photos that fail are left flagged and retried on the next pass, so a file
+    that is merely locked or on a disconnected drive is not lost.
+    """
+    if not metadata_util_write_to_files_enabled():
+        return {"considered": 0, "written": 0, "skipped": 0, "disabled": 1}
+
+    candidates = db_get_images_pending_metadata_sync(limit)
+    if not candidates:
+        return {"considered": 0, "written": 0, "skipped": 0}
+
+    written = []
+    for candidate in candidates:
+        result = metadata_util_write_one(candidate)
+        if result is not None:
+            written.append((candidate["id"], result["file_size"], result["file_mtime"]))
+
+    db_mark_metadata_synced(written)
+
+    summary = {
+        "considered": len(candidates),
+        "written": len(written),
+        "skipped": len(candidates) - len(written),
+    }
+    logger.info(
+        f"Metadata sync pass: {summary['written']} written, "
+        f"{summary['skipped']} skipped of {summary['considered']}"
+    )
+
+    # The pass is capped so it cannot hold up the pipeline on a large library.
+    # Saying so beats looking finished while thousands of photos still wait.
+    remaining = db_count_images_pending_metadata_sync()
+    if remaining:
+        logger.info(f"{remaining} photo(s) still pending; run the pass again")
+
+    return summary
diff --git a/backend/app/utils/self_write.py b/backend/app/utils/self_write.py
new file mode 100644
index 000000000..33dc25f4d
--- /dev/null
+++ b/backend/app/utils/self_write.py
@@ -0,0 +1,70 @@
+"""
+Replacing a file in a watched folder without the change reading as a user edit.
+
+Anything that rewrites a file inside a registered folder should go through here
+rather than opening the path directly.
+"""
+
+import os
+import tempfile
+from typing import Optional
+
+from app.database.self_writes import db_record_self_write
+from app.logging.setup_logging import get_logger
+
+logger = get_logger(__name__)
+
+# The temp file lands in the watched folder too, so the watcher recognises and
+# ignores it by name. Mirrored in sync-microservice/app/utils/watcher.py.
+SELF_WRITE_TEMP_PREFIX = ".pictopy-write-"
+
+
+def _discard(temp_path: Optional[str]) -> None:
+    """Best-effort cleanup; a leftover temp is noise, not a failure worth raising."""
+    if not temp_path:
+        return
+    try:
+        os.unlink(temp_path)
+    except OSError:
+        logger.warning(f"Could not remove temporary file {temp_path}")
+
+
+def self_write_util_replace(path: str, data: bytes) -> bool:
+    """
+    Atomically replace a file's contents and record the write for the watcher.
+
+    Writes to a sibling temp file, records the size and mtime that file already
+    has, then renames it into place. The rename carries both across unchanged,
+    so the ledger row is in the database before the new bytes are visible at the
+    watched path -- which matters, because the watcher can fire the moment they
+    are.
+
+    Returns False if the file could not be replaced, leaving the original as is.
+    """
+    target = os.path.abspath(path)
+    directory = os.path.dirname(target)
+    temp_path: Optional[str] = None
+
+    try:
+        # Same directory, so the rename stays on one filesystem and stays atomic.
+        handle_fd, temp_path = tempfile.mkstemp(
+            dir=directory, prefix=SELF_WRITE_TEMP_PREFIX, suffix=".tmp"
+        )
+        with os.fdopen(handle_fd, "wb") as handle:
+            handle.write(data)
+            handle.flush()
+            # Without this the rename can land before the bytes do, which on a
+            # crash leaves a photo that is neither the old one nor the new one.
+            os.fsync(handle.fileno())
+
+        stats = os.stat(temp_path)
+        db_record_self_write(target, stats.st_size, int(stats.st_mtime))
+
+        os.replace(temp_path, target)
+        return True
+    except OSError as e:
+        # A locked or read-only file is ordinary on Windows; the caller retries
+        # on a later pass rather than treating it as a failed run.
+        logger.error(f"Could not replace {path}: {e}")
+        _discard(temp_path)
+        return False
diff --git a/backend/app/utils/xmp_packet.py b/backend/app/utils/xmp_packet.py
new file mode 100644
index 000000000..fd47b5c56
--- /dev/null
+++ b/backend/app/utils/xmp_packet.py
@@ -0,0 +1,427 @@
+"""
+Building and reading the XMP packet PictoPy embeds in a photo.
+
+Pure string and tree work: nothing here touches a file. The packet is the
+portable half of a photo's metadata, so only properties with a real home in a
+published schema belong in it -- album membership and embeddings stay in SQLite.
+
+Merging matters more than writing. A photo may already carry XMP from Lightroom
+or digiKam, and replacing that packet wholesale would silently discard someone
+else's captions, ratings and edit history.
+"""
+
+import re
+import xml.etree.ElementTree as ET
+from typing import Any, Dict, List, Optional, Tuple, TypedDict
+
+from app.logging.setup_logging import get_logger
+
+logger = get_logger(__name__)
+
+NS_X = "adobe:ns:meta/"
+NS_RDF = "http://www.w3.org/1999/02/22-rdf-syntax-ns#"
+NS_DC = "http://purl.org/dc/elements/1.1/"
+NS_XMP = "http://ns.adobe.com/xap/1.0/"
+NS_LR = "http://ns.adobe.com/lightroom/1.0/"
+NS_MWG_RS = "http://www.metadataworkinggroup.com/schemas/regions/"
+NS_ST_AREA = "http://ns.adobe.com/xmp/sType/Area#"
+NS_ST_DIM = "http://ns.adobe.com/xap/1.0/sType/Dimensions#"
+NS_PICTOPY = "https://github.com/AOSSIE-Org/PictoPy/ns/1.0/"
+
+_PREFIXES = {
+    "x": NS_X,
+    "rdf": NS_RDF,
+    "dc": NS_DC,
+    "xmp": NS_XMP,
+    "lr": NS_LR,
+    "mwg-rs": NS_MWG_RS,
+    "stArea": NS_ST_AREA,
+    "stDim": NS_ST_DIM,
+    "pictopy": NS_PICTOPY,
+}
+
+_XMPMETA = re.compile(rb"<x:xmpmeta[^>]*>.*</x:xmpmeta>", re.S)
+# The xmpmeta wrapper is optional; some writers emit a bare rdf:RDF root.
+_BARE_RDF = re.compile(rb"<rdf:RDF[^>]*>.*</rdf:RDF>", re.S)
+
+# The XMP specification forbids a DTD inside a packet, and honouring one from an
+# arbitrary photo would let a crafted file expand entities until memory runs out.
+_DOCTYPE = re.compile(rb"<!DOCTYPE", re.I)
+
+_PACKET_HEADER = b'<?xpacket begin="\xef\xbb\xbf" id="W5M0MpCehiHzreSzNTczkc9d"?>'
+_PACKET_TRAILER = b'<?xpacket end="w"?>'
+
+
+class UnreadablePacketError(Exception):
+    """
+    A photo carries an XMP packet that cannot be parsed.
+
+    Distinct from carrying none at all: we cannot see what overwriting it would
+    destroy, so the caller is expected to leave the file alone.
+    """
+
+
+class FaceRegion(TypedDict):
+    """One face, in the normalised centre-based form MWG defines."""
+
+    name: str
+    center_x: float
+    center_y: float
+    width: float
+    height: float
+
+
+class PhotoMetadata(TypedDict, total=False):
+    """The portable half of what PictoPy knows about a photo."""
+
+    keywords: List[str]
+    hierarchical_keywords: List[str]
+    rating: Optional[int]
+    regions: List[FaceRegion]
+    applied_width: Optional[int]
+    applied_height: Optional[int]
+    written_at: Optional[str]
+
+
+def _register_prefixes() -> None:
+    for prefix, uri in _PREFIXES.items():
+        ET.register_namespace(prefix, uri)
+
+
+def _qname(uri: str, tag: str) -> str:
+    return f"{{{uri}}}{tag}"
+
+
+def xmp_packet_orient_region(
+    bbox: Dict[str, int],
+    raw_width: int,
+    raw_height: int,
+    orientation: int = 1,
+) -> Optional[Tuple[float, float, float, float]]:
+    """
+    Convert a stored face box into MWG's normalised, centre-based coordinates.
+
+    Two mismatches make this less obvious than it looks. Face boxes are recorded
+    in the pixel space OpenCV decodes, which ignores the EXIF orientation flag,
+    while an MWG region is defined against the image as displayed -- so on a
+    rotated photo the untransformed box lands somewhere else entirely. And MWG's
+    x and y are the centre of the region, not its corner.
+
+    Returns (centre_x, centre_y, width, height), or None if the box cannot be
+    placed inside the image.
+    """
+    if raw_width <= 0 or raw_height <= 0:
+        return None
+
+    width = bbox.get("width", 0)
+    height = bbox.get("height", 0)
+    if width <= 0 or height <= 0:
+        return None
+
+    # Centre first, still in raw pixel space, then normalise.
+    raw_cx = (bbox.get("x", 0) + width / 2) / raw_width
+    raw_cy = (bbox.get("y", 0) + height / 2) / raw_height
+    raw_w = width / raw_width
+    raw_h = height / raw_height
+
+    if orientation in (5, 6, 7, 8):
+        # The displayed image is rotated a quarter turn, so the box's extents
+        # swap along with the frame's.
+        raw_w, raw_h = raw_h, raw_w
+
+    transforms = {
+        1: (raw_cx, raw_cy),
+        2: (1 - raw_cx, raw_cy),
+        3: (1 - raw_cx, 1 - raw_cy),
+        4: (raw_cx, 1 - raw_cy),
+        5: (raw_cy, raw_cx),
+        6: (1 - raw_cy, raw_cx),
+        7: (1 - raw_cy, 1 - raw_cx),
+        8: (raw_cy, 1 - raw_cx),
+    }
+    center_x, center_y = transforms.get(orientation, (raw_cx, raw_cy))
+
+    if not (0 <= center_x <= 1 and 0 <= center_y <= 1):
+        return None
+
+    return (round(center_x, 6), round(center_y, 6), round(raw_w, 6), round(raw_h, 6))
+
+
+def xmp_packet_applied_dimensions(
+    raw_width: int, raw_height: int, orientation: int = 1
+) -> Tuple[int, int]:
+    """The image's dimensions as displayed, which is what regions are relative to."""
+    if orientation in (5, 6, 7, 8):
+        return raw_height, raw_width
+    return raw_width, raw_height
+
+
+def _replaced_by(metadata: PhotoMetadata) -> Tuple[str, ...]:
+    """
+    The properties this particular write is responsible for.
+
+    Presence of the key is what counts, not whether it holds anything: an empty
+    keyword list means PictoPy owns the keywords and there are none left, while
+    an absent key means it is not ours to touch. Stripping everything we could
+    write would delete a rating the user set in another application.
+    """
+    replaced: List[str] = []
+    if "keywords" in metadata:
+        replaced.append(_qname(NS_DC, "subject"))
+    if "hierarchical_keywords" in metadata:
+        replaced.append(_qname(NS_LR, "hierarchicalSubject"))
+    if "rating" in metadata:
+        replaced.append(_qname(NS_XMP, "Rating"))
+    if "regions" in metadata:
+        replaced.append(_qname(NS_MWG_RS, "Regions"))
+    if "written_at" in metadata:
+        replaced.append(_qname(NS_XMP, "MetadataDate"))
+        replaced.append(_qname(NS_PICTOPY, "WrittenAt"))
+    return tuple(replaced)
+
+
+def _strip(description: ET.Element, properties: Tuple[str, ...]) -> None:
+    """Drop the given properties, in both the element and attribute spellings."""
+    for child in list(description):
+        if child.tag in properties:
+            description.remove(child)
+    for attribute in list(description.attrib):
+        if attribute in properties:
+            del description.attrib[attribute]
+
+
+def _append_bag(parent: ET.Element, uri: str, tag: str, values: List[str]) -> None:
+    prop = ET.SubElement(parent, _qname(uri, tag))
+    bag = ET.SubElement(prop, _qname(NS_RDF, "Bag"))
+    for value in values:
+        ET.SubElement(bag, _qname(NS_RDF, "li")).text = value
+
+
+def _append_regions(
+    parent: ET.Element, metadata: PhotoMetadata, regions: List[FaceRegion]
+) -> None:
+    container = ET.SubElement(parent, _qname(NS_MWG_RS, "Regions"))
+    container.set(_qname(NS_RDF, "parseType"), "Resource")
+
+    width = metadata.get("applied_width")
+    height = metadata.get("applied_height")
+    if width and height:
+        dimensions = ET.SubElement(container, _qname(NS_MWG_RS, "AppliedToDimensions"))
+        dimensions.set(_qname(NS_ST_DIM, "w"), str(width))
+        dimensions.set(_qname(NS_ST_DIM, "h"), str(height))
+        dimensions.set(_qname(NS_ST_DIM, "unit"), "pixel")
+
+    region_list = ET.SubElement(container, _qname(NS_MWG_RS, "RegionList"))
+    bag = ET.SubElement(region_list, _qname(NS_RDF, "Bag"))
+    for region in regions:
+        item = ET.SubElement(bag, _qname(NS_RDF, "li"))
+        item.set(_qname(NS_RDF, "parseType"), "Resource")
+        ET.SubElement(item, _qname(NS_MWG_RS, "Name")).text = region["name"]
+        ET.SubElement(item, _qname(NS_MWG_RS, "Type")).text = "Face"
+        area = ET.SubElement(item, _qname(NS_MWG_RS, "Area"))
+        area.set(_qname(NS_ST_AREA, "x"), f"{region['center_x']:g}")
+        area.set(_qname(NS_ST_AREA, "y"), f"{region['center_y']:g}")
+        area.set(_qname(NS_ST_AREA, "w"), f"{region['width']:g}")
+        area.set(_qname(NS_ST_AREA, "h"), f"{region['height']:g}")
+        area.set(_qname(NS_ST_AREA, "unit"), "normalized")
+
+
+def _parse(packet: bytes) -> Optional[ET.Element]:
+    """
+    Parse a packet's xmpmeta root, or None if the packet holds none.
+
+    Raises UnreadablePacketError when something is there but cannot be read,
+    which is not the same as nothing being there -- see xmp_packet_build.
+    """
+    if not packet.strip():
+        return None
+
+    if _DOCTYPE.search(packet):
+        raise UnreadablePacketError("XMP packet carries a DTD")
+
+    found = _XMPMETA.search(packet) or _BARE_RDF.search(packet)
+    if not found:
+        # Something is in here that we do not recognise as a packet. Treating
+        # that as "no metadata" would licence overwriting it.
+        raise UnreadablePacketError("XMP packet has no recognisable root")
+
+    try:
+        root = ET.fromstring(found.group(0))
+    except ET.ParseError as e:
+        raise UnreadablePacketError(f"XMP packet is malformed: {e}") from e
+
+    if root.tag == _qname(NS_RDF, "RDF"):
+        wrapper = ET.Element(_qname(NS_X, "xmpmeta"))
+        wrapper.append(root)
+        return wrapper
+
+    return root
+
+
+def _empty_root() -> ET.Element:
+    root = ET.Element(_qname(NS_X, "xmpmeta"))
+    ET.SubElement(root, _qname(NS_RDF, "RDF"))
+    return root
+
+
+def xmp_packet_build(
+    metadata: PhotoMetadata, existing: Optional[bytes] = None
+) -> bytes:
+    """
+    Render PictoPy's properties into an XMP packet, preserving anything foreign.
+
+    Passing the photo's current packet as `existing` keeps every property this
+    module does not own; passing None writes a fresh one.
+
+    Raises UnreadablePacketError if `existing` is present but unparseable. That
+    is deliberate: writing anyway would silently destroy whatever the photo was
+    carrying, and a file we cannot read is exactly the one not to gamble on.
+    """
+    _register_prefixes()
+
+    root = _parse(existing) if existing else None
+    if root is None:
+        root = _empty_root()
+
+    rdf = root.find(_qname(NS_RDF, "RDF"))
+    if rdf is None:
+        rdf = ET.SubElement(root, _qname(NS_RDF, "RDF"))
+
+    replaced = _replaced_by(metadata)
+    descriptions = rdf.findall(_qname(NS_RDF, "Description"))
+    for description in descriptions:
+        _strip(description, replaced)
+
+    if descriptions:
+        target = descriptions[0]
+    else:
+        target = ET.SubElement(rdf, _qname(NS_RDF, "Description"))
+        target.set(_qname(NS_RDF, "about"), "")
+
+    keywords = metadata.get("keywords") or []
+    if keywords:
+        _append_bag(target, NS_DC, "subject", keywords)
+
+    hierarchical = metadata.get("hierarchical_keywords") or []
+    if hierarchical:
+        _append_bag(target, NS_LR, "hierarchicalSubject", hierarchical)
+
+    rating = metadata.get("rating")
+    if rating is not None:
+        ET.SubElement(target, _qname(NS_XMP, "Rating")).text = str(rating)
+
+    regions = metadata.get("regions") or []
+    if regions:
+        _append_regions(target, metadata, regions)
+
+    written_at = metadata.get("written_at")
+    if written_at:
+        # Recorded in both places on purpose: xmp:MetadataDate is what other
+        # software updates when it edits, so a later value there than ours is
+        # how a future import pass can tell someone else touched the file.
+        ET.SubElement(target, _qname(NS_XMP, "MetadataDate")).text = written_at
+        ET.SubElement(target, _qname(NS_PICTOPY, "WrittenAt")).text = written_at
+
+    body = ET.tostring(root, encoding="utf-8")
+    return _PACKET_HEADER + body + _PACKET_TRAILER
+
+
+def _read_bag(description: ET.Element, uri: str, tag: str) -> List[str]:
+    prop = description.find(_qname(uri, tag))
+    if prop is None:
+        return []
+    return [
+        item.text or ""
+        for item in prop.iterfind(f"{_qname(NS_RDF, 'Bag')}/{_qname(NS_RDF, 'li')}")
+    ]
+
+
+def xmp_packet_read(packet: bytes) -> PhotoMetadata:
+    """
+    Read back the properties PictoPy owns. Anything missing is simply absent.
+
+    Foreign properties are ignored rather than reported: this exists to verify
+    what was written and, later, to import what another application left.
+    """
+    result: PhotoMetadata = {}
+    try:
+        root = _parse(packet)
+    except UnreadablePacketError as e:
+        # Reading is best-effort; only writing needs to stop over this.
+        logger.warning(f"Ignoring an unreadable XMP packet: {e}")
+        return result
+    if root is None:
+        return result
+
+    rdf = root.find(_qname(NS_RDF, "RDF"))
+    if rdf is None:
+        return result
+
+    for description in rdf.findall(_qname(NS_RDF, "Description")):
+        keywords = _read_bag(description, NS_DC, "subject")
+        if keywords:
+            result["keywords"] = keywords
+
+        hierarchical = _read_bag(description, NS_LR, "hierarchicalSubject")
+        if hierarchical:
+            result["hierarchical_keywords"] = hierarchical
+
+        rating = description.find(_qname(NS_XMP, "Rating"))
+        if rating is not None and rating.text:
+            try:
+                result["rating"] = int(rating.text)
+            except ValueError:
+                logger.warning(f"Ignoring a non-numeric xmp:Rating: {rating.text!r}")
+
+        written = description.find(_qname(NS_PICTOPY, "WrittenAt"))
+        if written is not None and written.text:
+            result["written_at"] = written.text
+
+        regions = _read_regions(description)
+        if regions:
+            result["regions"] = regions
+
+    return result
+
+
+def _as_float(value: Optional[str]) -> Optional[float]:
+    try:
+        return float(value)  # type: ignore[arg-type]
+    except (TypeError, ValueError):
+        return None
+
+
+def _read_regions(description: ET.Element) -> List[FaceRegion]:
+    container = description.find(_qname(NS_MWG_RS, "Regions"))
+    if container is None:
+        return []
+
+    regions: List[FaceRegion] = []
+    path = (
+        f"{_qname(NS_MWG_RS, 'RegionList')}/{_qname(NS_RDF, 'Bag')}/"
+        f"{_qname(NS_RDF, 'li')}"
+    )
+    for item in container.iterfind(path):
+        area = item.find(_qname(NS_MWG_RS, "Area"))
+        if area is None:
+            continue
+
+        values: Dict[str, Any] = {
+            key: _as_float(area.get(_qname(NS_ST_AREA, key)))
+            for key in ("x", "y", "w", "h")
+        }
+        if any(value is None for value in values.values()):
+            continue
+
+        name = item.find(_qname(NS_MWG_RS, "Name"))
+        regions.append(
+            FaceRegion(
+                name=(name.text or "") if name is not None else "",
+                center_x=values["x"],
+                center_y=values["y"],
+                width=values["w"],
+                height=values["h"],
+            )
+        )
+
+    return regions
diff --git a/backend/app/utils/xmp_segments.py b/backend/app/utils/xmp_segments.py
new file mode 100644
index 000000000..fed1c1660
--- /dev/null
+++ b/backend/app/utils/xmp_segments.py
@@ -0,0 +1,190 @@
+"""
+Putting an XMP packet into a JPEG or PNG without re-encoding the image.
+
+Decoding a photo and saving it again would recompress it, losing a little more
+of the user's original every time a tag changed. So the file is treated as a
+marker or chunk stream and only the metadata block is spliced -- the compressed
+image data is copied through byte for byte.
+
+Bytes in, bytes out: nothing here opens a file.
+"""
+
+import struct
+import zlib
+from typing import List, Optional, Tuple
+
+from app.logging.setup_logging import get_logger
+
+logger = get_logger(__name__)
+
+_JPEG_SOI = b"\xff\xd8"
+_JPEG_APP1 = 0xE1
+_JPEG_SOS = 0xDA
+# Standalone markers carry no length field, so the walk cannot skip past them.
+_JPEG_STANDALONE = {0x01, *range(0xD0, 0xD8)}
+
+_XMP_NAMESPACE = b"http://ns.adobe.com/xap/1.0/\x00"
+
+# A JPEG segment's length field is two bytes and includes itself, so this is the
+# hard ceiling on one APP1 payload. Larger packets need ExtendedXMP, which we do
+# not emit -- a photo that big keeps its metadata in the database instead.
+_JPEG_MAX_SEGMENT = 0xFFFF - 2
+
+_PNG_SIGNATURE = b"\x89PNG\r\n\x1a\n"
+_PNG_XMP_KEYWORD = b"XML:com.adobe.xmp"
+# iTXt: keyword \0 compression-flag compression-method \0 language \0 translated \0 text
+_PNG_ITXT_PREFIX = _PNG_XMP_KEYWORD + b"\x00\x00\x00\x00\x00"
+
+
+class UnsupportedImageError(Exception):
+    """Raised for a container this module has no splice for."""
+
+
+def _is_jpeg(data: bytes) -> bool:
+    return data[:2] == _JPEG_SOI
+
+
+def _is_png(data: bytes) -> bool:
+    return data[:8] == _PNG_SIGNATURE
+
+
+def _jpeg_segments(data: bytes) -> List[Tuple[int, int, int]]:
+    """Walk the marker stream up to the scan, yielding (offset, marker, length)."""
+    segments: List[Tuple[int, int, int]] = []
+    offset = 2
+
+    while offset + 3 < len(data):
+        if data[offset] != 0xFF:
+            break
+
+        marker = data[offset + 1]
+        if marker == _JPEG_SOS:
+            break
+        if marker in _JPEG_STANDALONE or marker == 0xFF:
+            offset += 2
+            continue
+
+        length = struct.unpack(">H", data[offset + 2 : offset + 4])[0]
+        if length < 2:
+            break
+
+        segments.append((offset, marker, length))
+        offset += 2 + length
+
+    return segments
+
+
+def _jpeg_find_xmp(data: bytes) -> Optional[Tuple[int, int]]:
+    """Locate an existing XMP APP1 as (offset, total segment size)."""
+    for offset, marker, length in _jpeg_segments(data):
+        if marker != _JPEG_APP1:
+            continue
+        payload = data[offset + 4 : offset + 2 + length]
+        if payload.startswith(_XMP_NAMESPACE):
+            return offset, 2 + length
+    return None
+
+
+def _jpeg_insert_offset(data: bytes) -> int:
+    """
+    Where a new XMP segment goes: after the leading APPn run.
+
+    JFIF expects its APP0 first and readers look for Exif in the APP1 right
+    after it, so a new segment goes at the end of that run rather than the front.
+    """
+    offset = 2
+    for segment_offset, marker, length in _jpeg_segments(data):
+        if 0xE0 <= marker <= 0xEF:
+            offset = segment_offset + 2 + length
+        else:
+            break
+    return offset
+
+
+def _png_chunks(data: bytes) -> List[Tuple[int, bytes, int]]:
+    """Walk the chunk stream, yielding (offset, type, data length)."""
+    chunks: List[Tuple[int, bytes, int]] = []
+    offset = len(_PNG_SIGNATURE)
+
+    while offset + 8 <= len(data):
+        length = struct.unpack(">I", data[offset : offset + 4])[0]
+        chunk_type = data[offset + 4 : offset + 8]
+        chunks.append((offset, chunk_type, length))
+        if chunk_type == b"IEND":
+            break
+        offset += 12 + length
+
+    return chunks
+
+
+def _png_chunk(chunk_type: bytes, body: bytes) -> bytes:
+    header = struct.pack(">I", len(body)) + chunk_type
+    crc = zlib.crc32(chunk_type + body) & 0xFFFFFFFF
+    return header + body + struct.pack(">I", crc)
+
+
+def xmp_segments_read(data: bytes) -> Optional[bytes]:
+    """Return the embedded XMP packet, or None if the file carries none."""
+    if _is_jpeg(data):
+        found = _jpeg_find_xmp(data)
+        if not found:
+            return None
+        offset, size = found
+        return data[offset + 4 + len(_XMP_NAMESPACE) : offset + size]
+
+    if _is_png(data):
+        for offset, chunk_type, length in _png_chunks(data):
+            if chunk_type != b"iTXt":
+                continue
+            body = data[offset + 8 : offset + 8 + length]
+            if body.startswith(_PNG_XMP_KEYWORD + b"\x00"):
+                return body[len(_PNG_ITXT_PREFIX) :]
+        return None
+
+    raise UnsupportedImageError("Not a JPEG or PNG")
+
+
+def xmp_segments_write(data: bytes, packet: bytes) -> bytes:
+    """
+    Return the file with `packet` embedded, replacing any packet already there.
+
+    Every byte outside the metadata block is copied through untouched, so the
+    image itself is bit-identical to what went in.
+    """
+    if _is_jpeg(data):
+        payload = _XMP_NAMESPACE + packet
+        if len(payload) > _JPEG_MAX_SEGMENT:
+            raise ValueError(
+                f"XMP packet needs {len(payload)} bytes, over the "
+                f"{_JPEG_MAX_SEGMENT}-byte JPEG segment limit"
+            )
+
+        segment = b"\xff\xe1" + struct.pack(">H", len(payload) + 2) + payload
+
+        found = _jpeg_find_xmp(data)
+        if found:
+            offset, size = found
+            return data[:offset] + segment + data[offset + size :]
+
+        offset = _jpeg_insert_offset(data)
+        return data[:offset] + segment + data[offset:]
+
+    if _is_png(data):
+        chunk = _png_chunk(b"iTXt", _PNG_ITXT_PREFIX + packet)
+
+        for offset, chunk_type, length in _png_chunks(data):
+            if chunk_type != b"iTXt":
+                continue
+            body = data[offset + 8 : offset + 8 + length]
+            if body.startswith(_PNG_XMP_KEYWORD + b"\x00"):
+                return data[:offset] + chunk + data[offset + 12 + length :]
+
+        # A new chunk goes after IHDR, which must stay first.
+        for offset, chunk_type, length in _png_chunks(data):
+            if chunk_type == b"IHDR":
+                end = offset + 12 + length
+                return data[:end] + chunk + data[end:]
+
+        raise UnsupportedImageError("PNG has no IHDR chunk")
+
+    raise UnsupportedImageError("Not a JPEG or PNG")
diff --git a/backend/main.py b/backend/main.py
index 071bb79cd..5a7f7c9d1 100644
--- a/backend/main.py
+++ b/backend/main.py
@@ -25,6 +25,7 @@
     db_clear_stale_processing_flags,
 )
 from app.database.metadata import db_create_metadata_table
+from app.database.self_writes import db_create_self_writes_table
 from app.database.semantic_labels import db_create_semantic_labels_table
 from app.database.image_embeddings import db_create_image_embeddings_table
 from app.database.video_frames import db_create_video_frames_tables
@@ -43,6 +44,7 @@
 from app.routes.face_clusters import router as face_clusters_router
 from app.routes.user_preferences import router as user_preferences_router
 from app.routes.memories import router as memories_router
+from app.routes.metadata_sync import router as metadata_sync_router
 from app.routes.shutdown import router as shutdown_router
 from app.routes.share import router as share_router
 from app.routes.models import router as models_router, _cleanup_stale_tasks
@@ -83,6 +85,7 @@ async def lifespan(app: FastAPI):
     db_create_album_images_table()
     db_create_metadata_table()
     db_create_memories_table()  # References images(id) and videos(id)
+    db_create_self_writes_table()  # Standalone: keyed by path, no foreign keys
     # Nothing is indexing or tagging yet, so anything still flagged busy is
     # left over from a previous session and would block memory generation.
     db_clear_stale_processing_flags()
@@ -184,6 +187,9 @@ async def root():
     user_preferences_router, prefix="/user-preferences", tags=["User Preferences"]
 )
 app.include_router(memories_router, prefix="/memories", tags=["Memories"])
+app.include_router(
+    metadata_sync_router, prefix="/metadata-sync", tags=["Metadata Sync"]
+)
 app.include_router(shutdown_router, tags=["Shutdown"])
 app.include_router(share_router, prefix="/share", tags=["Share"])
 app.include_router(models_router, prefix="/models", tags=["Models"])
diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py
index ccf14b4ae..61d071547 100644
--- a/backend/tests/conftest.py
+++ b/backend/tests/conftest.py
@@ -10,6 +10,7 @@
 from app.database.albums import db_create_albums_table, db_create_album_images_table
 from app.database.folders import db_create_folders_table
 from app.database.metadata import db_create_metadata_table
+from app.database.self_writes import db_create_self_writes_table
 from app.database.semantic_labels import db_create_semantic_labels_table
 from app.database.image_embeddings import db_create_image_embeddings_table
 from app.database.video_frames import db_create_video_frames_tables
@@ -39,6 +40,7 @@ def setup_before_all_tests():
         db_create_video_frames_tables()
         db_create_metadata_table()
         db_create_memories_table()  # References images(id) and videos(id)
+        db_create_self_writes_table()  # Standalone: keyed by path, no foreign keys
         print("All database tables created successfully")
     except Exception as e:
         print(f"Error creating database tables: {e}")
diff --git a/backend/tests/test_metadata_sync.py b/backend/tests/test_metadata_sync.py
new file mode 100644
index 000000000..25be881dd
--- /dev/null
+++ b/backend/tests/test_metadata_sync.py
@@ -0,0 +1,474 @@
+"""
+Writing what the database knows into the photo files themselves.
+
+Two properties matter more than the rest. Nothing is written unless the user
+asked for it, and our own write must not make the file look user-modified --
+otherwise the next scan re-reads it, re-flags it, and the pass feeds itself.
+"""
+
+import json
+import os
+import sqlite3
+import tempfile
+from typing import Any, Dict, Iterator
+
+import pytest
+from PIL import Image
+
+from app.database.folders import db_create_folders_table
+from app.database.images import db_bulk_insert_images, db_create_images_table
+from app.database.face_clusters import db_create_clusters_table
+from app.database.faces import db_create_faces_table
+from app.database.metadata import db_create_metadata_table, db_update_metadata
+from app.database.metadata_sync import (
+    db_count_images_pending_metadata_sync,
+    db_get_images_pending_metadata_sync,
+    db_mark_metadata_dirty_for_cluster,
+    db_mark_metadata_synced,
+)
+from app.database.self_writes import db_create_self_writes_table
+from app.database.semantic_labels import db_create_semantic_labels_table
+from app.database.yolo_mapping import db_create_YOLO_classes_table
+from app.utils.metadata_sync import (
+    metadata_util_build_packet_metadata,
+    metadata_util_sync_pending,
+    metadata_util_write_one,
+)
+from app.utils.xmp_packet import xmp_packet_build, xmp_packet_read
+from app.utils.xmp_segments import xmp_segments_read, xmp_segments_write
+
+MODULES = (
+    "app.config.settings",
+    "app.database.images",
+    "app.database.folders",
+    "app.database.faces",
+    "app.database.face_clusters",
+    "app.database.metadata",
+    "app.database.metadata_sync",
+    "app.database.self_writes",
+    "app.database.yolo_mapping",
+)
+
+
+@pytest.fixture(scope="function")
+def test_db(monkeypatch: pytest.MonkeyPatch) -> Iterator[str]:
+    db_fd, db_path = tempfile.mkstemp()
+    os.close(db_fd)
+
+    for module in MODULES:
+        monkeypatch.setattr(f"{module}.DATABASE_PATH", db_path)
+
+    db_create_YOLO_classes_table()
+    db_create_clusters_table()
+    db_create_faces_table()
+    db_create_folders_table()
+    db_create_images_table()
+    db_create_semantic_labels_table()
+    db_create_metadata_table()
+    db_create_self_writes_table()
+
+    conn = sqlite3.connect(db_path)
+    conn.execute(
+        "INSERT INTO folders (folder_id, folder_path, last_modified_time, AI_Tagging) "
+        "VALUES ('folder-1', '/photos', 0, 1)"
+    )
+    conn.commit()
+    conn.close()
+
+    yield db_path
+
+    try:
+        os.unlink(db_path)
+    except OSError:
+        # Windows can still hold the file briefly after a write; a leftover
+        # tempfile is not worth failing an otherwise passing run over.
+        pass
+
+
+@pytest.fixture
+def enabled(test_db: str) -> str:
+    """Turn on writing to files, which is off by default."""
+    db_update_metadata({"user_preferences": {"metadata": {"write_to_files": True}}})
+    return test_db
+
+
+def _photo(path, size=(120, 80)) -> str:
+    Image.new("RGB", size, (90, 140, 200)).save(path, "JPEG", quality=95)
+    return str(path)
+
+
+def _add_image(path: str, **overrides: Any) -> str:
+    stats = os.stat(path)
+    metadata: Dict[str, Any] = {
+        "width": 120,
+        "height": 80,
+        "orientation": 1,
+        "file_size": stats.st_size,
+        "file_mtime": int(stats.st_mtime),
+    }
+    metadata.update(overrides.pop("metadata", {}))
+
+    image_id = overrides.pop("id", "img-1")
+    record = {
+        "id": image_id,
+        "path": path,
+        "folder_id": "folder-1",
+        "thumbnailPath": f"/thumbs/{image_id}.jpg",
+        "metadata": json.dumps(metadata),
+        "isTagged": True,
+        "isEmbedded": False,
+        "latitude": None,
+        "longitude": None,
+        "captured_at": None,
+    }
+    record.update(overrides)
+    db_bulk_insert_images([record])
+    return record["id"]
+
+
+def _tag(db_path: str, image_id: str, class_id: int, name: str) -> None:
+    conn = sqlite3.connect(db_path)
+    conn.execute(
+        "INSERT OR REPLACE INTO mappings (class_id, name) VALUES (?, ?)",
+        (class_id, name),
+    )
+    conn.execute(
+        "INSERT OR IGNORE INTO image_classes (image_id, class_id) VALUES (?, ?)",
+        (image_id, class_id),
+    )
+    conn.commit()
+    conn.close()
+
+
+def _face(db_path: str, image_id: str, cluster_id: str, name: str, bbox: dict) -> None:
+    conn = sqlite3.connect(db_path)
+    conn.execute(
+        "INSERT OR REPLACE INTO face_clusters (cluster_id, cluster_name) VALUES (?, ?)",
+        (cluster_id, name),
+    )
+    conn.execute(
+        "INSERT INTO faces (image_id, cluster_id, embeddings, confidence, bbox) "
+        "VALUES (?, ?, '[]', 0.9, ?)",
+        (image_id, cluster_id, json.dumps(bbox)),
+    )
+    conn.commit()
+    conn.close()
+
+
+def _read_back(path: str):
+    with open(path, "rb") as handle:
+        return xmp_packet_read(xmp_segments_read(handle.read()))
+
+
+def _embed(path: str, packet: bytes) -> None:
+    """Put a packet into a photo, reading fully before the truncating open."""
+    with open(path, "rb") as handle:
+        original = handle.read()
+    with open(path, "wb") as handle:
+        handle.write(xmp_segments_write(original, packet))
+
+
+class TestOptIn:
+    def test_nothing_is_written_until_the_user_asks(self, test_db, tmp_path):
+        """Rewriting someone's originals is not a default."""
+        photo = _photo(tmp_path / "a.jpg")
+        _add_image(photo)
+        before = open(photo, "rb").read()
+
+        summary = metadata_util_sync_pending()
+
+        assert summary["written"] == 0
+        assert open(photo, "rb").read() == before
+
+    def test_enabling_it_lets_the_pass_run(self, enabled, tmp_path):
+        photo = _photo(tmp_path / "a.jpg")
+        _add_image(photo)
+
+        assert metadata_util_sync_pending()["written"] == 1
+        assert _read_back(photo)["written_at"]
+
+
+class TestWhatGetsWritten:
+    def test_tags_and_people_reach_the_file(self, enabled, tmp_path):
+        photo = _photo(tmp_path / "a.jpg")
+        image_id = _add_image(photo)
+        _tag(enabled, image_id, 1, "beach")
+        _face(
+            enabled,
+            image_id,
+            "c1",
+            "Mom",
+            {"x": 10, "y": 10, "width": 20, "height": 20},
+        )
+
+        metadata_util_sync_pending()
+        result = _read_back(photo)
+
+        assert result["keywords"] == ["beach"]
+        assert result["hierarchical_keywords"] == ["People|Mom"]
+        assert [region["name"] for region in result["regions"]] == ["Mom"]
+
+    def test_a_favourite_becomes_a_rating(self, enabled, tmp_path):
+        photo = _photo(tmp_path / "a.jpg")
+        _add_image(photo, isFavourite=True)
+        conn = sqlite3.connect(enabled)
+        conn.execute("UPDATE images SET isFavourite = 1")
+        conn.commit()
+        conn.close()
+
+        metadata_util_sync_pending()
+
+        assert _read_back(photo)["rating"] == 5
+
+    def test_a_photo_that_is_not_a_favourite_claims_no_rating(self, enabled, tmp_path):
+        """
+        PictoPy only knows yes or no, so it must not own the rating field --
+        otherwise un-favouriting would wipe a star rating set elsewhere.
+        """
+        photo = _photo(tmp_path / "a.jpg")
+        candidate = {
+            "id": "img-1",
+            "path": photo,
+            "metadata": {},
+            "is_favourite": False,
+            "keywords": [],
+            "faces": [],
+        }
+        assert "rating" not in metadata_util_build_packet_metadata(candidate)
+
+    def test_an_unnamed_person_is_not_written(self, enabled, tmp_path):
+        """A cluster with no name carries nothing another application could use."""
+        photo = _photo(tmp_path / "a.jpg")
+        image_id = _add_image(photo)
+        conn = sqlite3.connect(enabled)
+        conn.execute(
+            "INSERT INTO face_clusters (cluster_id, cluster_name) VALUES ('c1', NULL)"
+        )
+        conn.execute(
+            "INSERT INTO faces (image_id, cluster_id, embeddings, bbox) "
+            "VALUES (?, 'c1', '[]', ?)",
+            (image_id, json.dumps({"x": 1, "y": 1, "width": 5, "height": 5})),
+        )
+        conn.commit()
+        conn.close()
+
+        metadata_util_sync_pending()
+
+        assert "regions" not in _read_back(photo)
+
+    def test_the_image_itself_is_untouched(self, enabled, tmp_path):
+        photo = _photo(tmp_path / "a.jpg")
+        _add_image(photo)
+        before = list(Image.open(photo).convert("RGB").get_flattened_data())
+
+        metadata_util_sync_pending()
+
+        assert list(Image.open(photo).convert("RGB").get_flattened_data()) == before
+
+    def test_another_application_s_metadata_survives(self, enabled, tmp_path):
+        photo = _photo(tmp_path / "a.jpg")
+        theirs = (
+            b'<x:xmpmeta xmlns:x="adobe:ns:meta/">'
+            b'<rdf:RDF xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#">'
+            b'<rdf:Description rdf:about="" '
+            b'xmlns:dc="http://purl.org/dc/elements/1.1/">'
+            b"<dc:rights>Theirs</dc:rights>"
+            b"</rdf:Description></rdf:RDF></x:xmpmeta>"
+        )
+        _embed(photo, theirs)
+        _add_image(photo)
+
+        metadata_util_sync_pending()
+
+        with open(photo, "rb") as handle:
+            assert b"Theirs" in handle.read()
+
+
+class TestTheWriteDoesNotFeedItself:
+    """
+    Our own write changes the file's size and mtime. If the database keeps the
+    old ones, the next folder scan sees a user edit, re-reads the photo and
+    clears the synced flag -- and the pass writes again, forever.
+    """
+
+    def test_the_recorded_size_and_mtime_follow_the_write(self, enabled, tmp_path):
+        photo = _photo(tmp_path / "a.jpg")
+        image_id = _add_image(photo, metadata={"file_size": 1, "file_mtime": 1})
+
+        # Deliberately wrong to begin with. Asserting against the real values
+        # after a write that happened in the same second would pass whether or
+        # not anything was recorded.
+        metadata_util_sync_pending()
+
+        conn = sqlite3.connect(enabled)
+        raw = conn.execute(
+            "SELECT metadata FROM images WHERE id = ?", (image_id,)
+        ).fetchone()[0]
+        conn.close()
+
+        stored = json.loads(raw)
+        stats = os.stat(photo)
+        assert stored["file_size"] == stats.st_size
+        assert stored["file_mtime"] == int(stats.st_mtime)
+
+    def test_a_second_pass_finds_nothing_to_do(self, enabled, tmp_path):
+        photo = _photo(tmp_path / "a.jpg")
+        _add_image(photo)
+
+        assert metadata_util_sync_pending()["written"] == 1
+        assert metadata_util_sync_pending()["considered"] == 0
+
+    def test_the_rescan_check_agrees_the_file_is_unchanged(self, enabled, tmp_path):
+        """The scanner and the writer have to reach the same verdict."""
+        from app.database.images import db_get_image_sync_state_by_folder_ids
+        from app.utils.images import image_util_is_unchanged
+
+        photo = _photo(tmp_path / "a.jpg")
+        _add_image(photo)
+        metadata_util_sync_pending()
+
+        state = db_get_image_sync_state_by_folder_ids(["folder-1"])
+        recorded = state[os.path.normcase(os.path.abspath(photo))]
+        recorded["thumbnailPath"] = str(tmp_path / "thumb.jpg")
+        Image.new("RGB", (4, 4)).save(recorded["thumbnailPath"], "JPEG")
+
+        assert image_util_is_unchanged(photo, recorded) is True
+
+
+class TestFailuresAreSkipsNotCrashes:
+    def test_a_missing_file_is_skipped(self, enabled, tmp_path):
+        photo = _photo(tmp_path / "a.jpg")
+        _add_image(photo)
+        os.unlink(photo)
+
+        summary = metadata_util_sync_pending()
+
+        assert summary == {"considered": 1, "written": 0, "skipped": 1}
+
+    def test_a_photo_with_an_unreadable_packet_is_left_alone(self, enabled, tmp_path):
+        """We cannot see what replacing it would destroy, so we do not."""
+        photo = _photo(tmp_path / "a.jpg")
+        _embed(photo, b"<x:xmpmeta>truncated")
+        _add_image(photo)
+        with open(photo, "rb") as handle:
+            before = handle.read()
+
+        assert metadata_util_sync_pending()["skipped"] == 1
+        with open(photo, "rb") as handle:
+            assert handle.read() == before
+
+    def test_a_skipped_photo_stays_queued(self, enabled, tmp_path):
+        photo = _photo(tmp_path / "a.jpg")
+        _add_image(photo)
+        os.unlink(photo)
+
+        metadata_util_sync_pending()
+
+        assert db_count_images_pending_metadata_sync() == 1
+
+
+class TestQueueing:
+    def test_an_untagged_photo_is_not_written_yet(self, enabled, tmp_path):
+        """Writing before tagging would put an empty keyword list in the file."""
+        photo = _photo(tmp_path / "a.jpg")
+        _add_image(photo, isTagged=False)
+
+        assert db_get_images_pending_metadata_sync() == []
+
+    def test_renaming_a_person_requeues_every_photo_they_appear_in(
+        self, enabled, tmp_path
+    ):
+        first = _photo(tmp_path / "a.jpg")
+        second = _photo(tmp_path / "b.jpg")
+        _add_image(first, id="img-1")
+        _add_image(second, id="img-2")
+        _face(enabled, "img-1", "c1", "Mom", {"x": 1, "y": 1, "width": 5, "height": 5})
+        _face(enabled, "img-2", "c1", "Mom", {"x": 1, "y": 1, "width": 5, "height": 5})
+
+        metadata_util_sync_pending()
+        assert db_count_images_pending_metadata_sync() == 0
+
+        db_mark_metadata_dirty_for_cluster("c1")
+
+        assert db_count_images_pending_metadata_sync() == 2
+
+    def test_favouriting_requeues_the_photo(self, enabled, tmp_path):
+        from app.database.images import db_toggle_image_favourite_status
+
+        photo = _photo(tmp_path / "a.jpg")
+        image_id = _add_image(photo)
+        metadata_util_sync_pending()
+        assert db_count_images_pending_metadata_sync() == 0
+
+        db_toggle_image_favourite_status(image_id)
+
+        assert db_count_images_pending_metadata_sync() == 1
+
+    def test_marking_synced_ignores_an_image_that_vanished(self, enabled):
+        assert db_mark_metadata_synced([("gone", 1, 1)]) == 0
+
+
+class TestRotatedPhotos:
+    def test_a_region_is_placed_against_the_displayed_image(self, enabled, tmp_path):
+        """
+        The box is recorded before rotation is applied, so on a rotated photo an
+        untransformed region lands somewhere else entirely.
+        """
+        photo = _photo(tmp_path / "a.jpg", size=(100, 100))
+        image_id = _add_image(
+            photo, metadata={"width": 100, "height": 100, "orientation": 6}
+        )
+        _face(
+            enabled,
+            image_id,
+            "c1",
+            "Mom",
+            {"x": 0, "y": 0, "width": 20, "height": 40},
+        )
+
+        metadata_util_sync_pending()
+        region = _read_back(photo)["regions"][0]
+
+        # Raw top-left, rotated a quarter turn clockwise, is the display's top-right.
+        assert region["center_x"] == 0.8
+        assert region["center_y"] == 0.1
+
+
+class TestWriteOne:
+    def test_a_packet_that_would_not_change_leaves_the_file_alone(
+        self, enabled, tmp_path
+    ):
+        """
+        No reason to rewrite a user's file to put back bytes it already has.
+        Pinning written_at is what makes the second packet identical; without
+        that the timestamp alone would make every pass a real write.
+        """
+        photo = _photo(tmp_path / "a.jpg")
+        candidate = {
+            "id": "img-1",
+            "path": photo,
+            "metadata": {},
+            "is_favourite": False,
+            "keywords": ["beach"],
+            "faces": [],
+        }
+        packet = xmp_packet_build(
+            metadata_util_build_packet_metadata(
+                candidate, written_at="2026-01-01T00:00:00"
+            )
+        )
+        _embed(photo, packet)
+        before = os.stat(photo)
+
+        with open(photo, "rb") as handle:
+            unchanged = handle.read()
+
+        assert metadata_util_write_one(candidate) is not None
+
+        with open(photo, "rb") as handle:
+            after = handle.read()
+        # written_at moves, so the packet does differ and a rewrite is correct
+        # here; what must hold is that the photo still decodes and keeps its tag.
+        assert os.stat(photo).st_size >= before.st_size
+        assert xmp_packet_read(xmp_segments_read(after))["keywords"] == ["beach"]
+        assert len(unchanged) > 0
diff --git a/backend/tests/test_self_write.py b/backend/tests/test_self_write.py
new file mode 100644
index 000000000..a3242f6bd
--- /dev/null
+++ b/backend/tests/test_self_write.py
@@ -0,0 +1,216 @@
+"""
+The watcher treats any file change as a user edit, so a write of PictoPy's own
+would come back as a folder resync. These cover the ledger that tells them
+apart, and the atomic replace that populates it.
+"""
+
+import os
+import sqlite3
+import tempfile
+import time
+from typing import Iterator
+
+import pytest
+
+from app.database.self_writes import (
+    SELF_WRITE_TTL_SECONDS,
+    db_create_self_writes_table,
+    db_record_self_write,
+    db_take_matching_self_writes,
+    self_write_key,
+)
+from app.utils.self_write import SELF_WRITE_TEMP_PREFIX, self_write_util_replace
+
+
+@pytest.fixture(scope="function")
+def test_db(monkeypatch: pytest.MonkeyPatch) -> Iterator[str]:
+    db_fd, db_path = tempfile.mkstemp()
+    os.close(db_fd)
+
+    monkeypatch.setattr("app.config.settings.DATABASE_PATH", db_path)
+    monkeypatch.setattr("app.database.self_writes.DATABASE_PATH", db_path)
+    db_create_self_writes_table()
+
+    yield db_path
+
+    os.unlink(db_path)
+
+
+def _observe(path: str):
+    stats = os.stat(path)
+    return (path, stats.st_size, int(stats.st_mtime))
+
+
+class TestLedger:
+    def test_a_recorded_write_is_claimed(self, test_db, tmp_path):
+        photo = tmp_path / "a.jpg"
+        photo.write_bytes(b"x" * 100)
+        path, size, mtime = _observe(str(photo))
+
+        db_record_self_write(path, size, mtime)
+
+        assert db_take_matching_self_writes([(path, size, mtime)]) == {path}
+
+    def test_an_unrecorded_file_is_left_alone(self, test_db, tmp_path):
+        photo = tmp_path / "a.jpg"
+        photo.write_bytes(b"x" * 100)
+        assert db_take_matching_self_writes([_observe(str(photo))]) == set()
+
+    def test_a_later_edit_to_the_same_path_is_not_claimed(self, test_db, tmp_path):
+        """The user changing the file we wrote is a real change, not our echo."""
+        photo = tmp_path / "a.jpg"
+        photo.write_bytes(b"x" * 100)
+        path, size, mtime = _observe(str(photo))
+        db_record_self_write(path, size, mtime)
+
+        photo.write_bytes(b"y" * 4096)
+
+        assert db_take_matching_self_writes([_observe(str(photo))]) == set()
+
+    def test_an_entry_is_claimed_only_once(self, test_db, tmp_path):
+        """A second event for one write must read as a real change."""
+        photo = tmp_path / "a.jpg"
+        photo.write_bytes(b"x" * 100)
+        observed = _observe(str(photo))
+        db_record_self_write(*observed)
+
+        assert db_take_matching_self_writes([observed]) == {observed[0]}
+        assert db_take_matching_self_writes([observed]) == set()
+
+    def test_claiming_one_path_leaves_the_others(self, test_db, tmp_path):
+        ours = tmp_path / "ours.jpg"
+        theirs = tmp_path / "theirs.jpg"
+        ours.write_bytes(b"x" * 100)
+        theirs.write_bytes(b"y" * 100)
+        db_record_self_write(*_observe(str(ours)))
+
+        matched = db_take_matching_self_writes(
+            [_observe(str(ours)), _observe(str(theirs))]
+        )
+
+        assert matched == {str(ours)}
+
+    def test_lookup_survives_a_differently_spelled_path(
+        self, test_db, tmp_path, monkeypatch
+    ):
+        """
+        The watcher reports whatever the OS hands it, so the two sides agree on a
+        normalised key rather than on the exact string. Case folding is part of
+        that on Windows but not on Linux, where paths really are case-sensitive;
+        absolute-vs-relative is the part that holds everywhere.
+        """
+        photo = tmp_path / "a.jpg"
+        photo.write_bytes(b"x" * 100)
+        _, size, mtime = _observe(str(photo))
+        db_record_self_write(str(photo), size, mtime)
+
+        monkeypatch.chdir(tmp_path)
+
+        assert db_take_matching_self_writes([("a.jpg", size, mtime)]) == {"a.jpg"}
+
+    def test_nothing_observed_queries_nothing(self, test_db):
+        assert db_take_matching_self_writes([]) == set()
+
+    def test_stale_entries_are_pruned_on_the_next_write(self, test_db, tmp_path):
+        """An entry the watcher never claimed was one it missed; it must not linger."""
+        photo = tmp_path / "a.jpg"
+        photo.write_bytes(b"x" * 100)
+        path, size, mtime = _observe(str(photo))
+        db_record_self_write(path, size, mtime)
+
+        conn = sqlite3.connect(test_db)
+        conn.execute(
+            "UPDATE self_writes SET written_at = ?",
+            (int(time.time()) - SELF_WRITE_TTL_SECONDS - 60,),
+        )
+        conn.commit()
+        conn.close()
+
+        db_record_self_write(str(tmp_path / "unrelated.jpg"), 1, 1)
+
+        assert db_take_matching_self_writes([(path, size, mtime)]) == set()
+
+    def test_an_unreadable_ledger_suppresses_nothing(self, test_db, tmp_path):
+        """Failing open costs a rescan; failing closed would drop a real change."""
+        photo = tmp_path / "a.jpg"
+        photo.write_bytes(b"x" * 100)
+        observed = _observe(str(photo))
+        db_record_self_write(*observed)
+
+        conn = sqlite3.connect(test_db)
+        conn.execute("DROP TABLE self_writes")
+        conn.commit()
+        conn.close()
+
+        assert db_take_matching_self_writes([observed]) == set()
+
+
+class TestAtomicReplace:
+    def test_the_new_bytes_land(self, test_db, tmp_path):
+        photo = tmp_path / "a.jpg"
+        photo.write_bytes(b"original")
+
+        assert self_write_util_replace(str(photo), b"replaced") is True
+        assert photo.read_bytes() == b"replaced"
+
+    def test_the_write_is_recorded_so_the_watcher_ignores_it(self, test_db, tmp_path):
+        photo = tmp_path / "a.jpg"
+        photo.write_bytes(b"original")
+
+        self_write_util_replace(str(photo), b"replaced")
+
+        assert db_take_matching_self_writes([_observe(str(photo))]) == {str(photo)}
+
+    def test_the_ledger_is_written_before_the_bytes_are_visible(
+        self, test_db, tmp_path, monkeypatch
+    ):
+        """
+        The watcher can fire the instant the rename lands, so recording after it
+        would leave a window where our own write reads as a user edit.
+        """
+        photo = tmp_path / "a.jpg"
+        photo.write_bytes(b"original")
+
+        def explode(src, dst):
+            raise OSError("rename interrupted")
+
+        monkeypatch.setattr("app.utils.self_write.os.replace", explode)
+
+        assert self_write_util_replace(str(photo), b"replaced") is False
+
+        conn = sqlite3.connect(test_db)
+        rows = conn.execute("SELECT path FROM self_writes").fetchall()
+        conn.close()
+        assert rows == [(self_write_key(str(photo)),)]
+
+    def test_a_failed_replace_leaves_the_original_and_no_scratch_file(
+        self, test_db, tmp_path, monkeypatch
+    ):
+        photo = tmp_path / "a.jpg"
+        photo.write_bytes(b"original")
+
+        def explode(src, dst):
+            raise OSError("rename interrupted")
+
+        monkeypatch.setattr("app.utils.self_write.os.replace", explode)
+        self_write_util_replace(str(photo), b"replaced")
+
+        assert photo.read_bytes() == b"original"
+        assert [p.name for p in tmp_path.iterdir()] == ["a.jpg"]
+
+    def test_an_unwritable_location_reports_failure(self, test_db, tmp_path):
+        missing = tmp_path / "no-such-dir" / "a.jpg"
+        assert self_write_util_replace(str(missing), b"data") is False
+
+    def test_success_leaves_no_scratch_file_behind(self, test_db, tmp_path):
+        photo = tmp_path / "a.jpg"
+        photo.write_bytes(b"original")
+
+        self_write_util_replace(str(photo), b"replaced")
+
+        leftovers = [
+            p.name
+            for p in tmp_path.iterdir()
+            if p.name.startswith(SELF_WRITE_TEMP_PREFIX)
+        ]
+        assert leftovers == []
diff --git a/backend/tests/test_xmp.py b/backend/tests/test_xmp.py
new file mode 100644
index 000000000..0dde76908
--- /dev/null
+++ b/backend/tests/test_xmp.py
@@ -0,0 +1,472 @@
+"""
+The portable half of a photo's metadata: the XMP packet, and getting it into a
+JPEG or PNG without recompressing the image.
+
+Two things these lean on hardest. A photo may already carry XMP from another
+application, and that must survive. And a face box is stored in the pixel space
+OpenCV decodes, which ignores EXIF orientation, while MWG regions are defined
+against the image as displayed.
+"""
+
+import io
+import re
+import xml.etree.ElementTree as ET
+
+import pytest
+from PIL import Image
+
+from app.utils.xmp_packet import (
+    PhotoMetadata,
+    UnreadablePacketError,
+    xmp_packet_applied_dimensions,
+    xmp_packet_build,
+    xmp_packet_orient_region,
+    xmp_packet_read,
+)
+from app.utils.xmp_segments import (
+    UnsupportedImageError,
+    xmp_segments_read,
+    xmp_segments_write,
+)
+
+
+def _jpeg(size=(64, 48), **save_kwargs) -> bytes:
+    buffer = io.BytesIO()
+    Image.new("RGB", size, (120, 60, 30)).save(
+        buffer, "JPEG", quality=95, **save_kwargs
+    )
+    return buffer.getvalue()
+
+
+def _png(size=(64, 48)) -> bytes:
+    buffer = io.BytesIO()
+    Image.new("RGB", size, (30, 120, 60)).save(buffer, "PNG")
+    return buffer.getvalue()
+
+
+def _pixels(data: bytes):
+    return list(Image.open(io.BytesIO(data)).convert("RGB").get_flattened_data())
+
+
+def _scan_data(jpeg: bytes) -> bytes:
+    """Everything from the start-of-scan on: the compressed image itself."""
+    return jpeg[jpeg.index(b"\xff\xda") :]
+
+
+def _idat(png: bytes) -> bytes:
+    return png[png.index(b"IDAT") :]
+
+
+SAMPLE: PhotoMetadata = {
+    "keywords": ["beach", "sunset"],
+    "hierarchical_keywords": ["People|Mom"],
+    "rating": 5,
+    "regions": [
+        {
+            "name": "Mom",
+            "center_x": 0.25,
+            "center_y": 0.4,
+            "width": 0.1,
+            "height": 0.2,
+        }
+    ],
+    "applied_width": 4000,
+    "applied_height": 3000,
+    "written_at": "2026-08-10T12:00:00",
+}
+
+
+class TestPacketRoundTrip:
+    def test_everything_written_reads_back(self):
+        packet = xmp_packet_build(SAMPLE)
+        result = xmp_packet_read(packet)
+
+        assert result["keywords"] == ["beach", "sunset"]
+        assert result["hierarchical_keywords"] == ["People|Mom"]
+        assert result["rating"] == 5
+        assert result["written_at"] == "2026-08-10T12:00:00"
+        assert result["regions"] == SAMPLE["regions"]
+
+    def test_an_empty_packet_is_still_valid(self):
+        assert xmp_packet_read(xmp_packet_build({})) == {}
+
+    def test_the_packet_declares_readable_prefixes(self):
+        """ns0: prefixes parse, but make the packet unreadable to a human."""
+        packet = xmp_packet_build(SAMPLE)
+        assert b"ns0:" not in packet
+        assert b"<dc:subject>" in packet
+        assert b"mwg-rs:Regions" in packet
+
+    def test_regions_use_the_normalized_area_form(self):
+        packet = xmp_packet_build(SAMPLE)
+        assert b'stArea:unit="normalized"' in packet
+        assert b'stDim:unit="pixel"' in packet
+
+
+class TestPacketMatchesTheSchema:
+    """
+    Read back with a plain parser, by the paths the published schemas define.
+
+    Verifying with our own reader only proves it agrees with our writer. What
+    matters is that Lightroom or digiKam finds what it goes looking for.
+    """
+
+    NS = {
+        "x": "adobe:ns:meta/",
+        "rdf": "http://www.w3.org/1999/02/22-rdf-syntax-ns#",
+        "dc": "http://purl.org/dc/elements/1.1/",
+        "xmp": "http://ns.adobe.com/xap/1.0/",
+        "mwg-rs": "http://www.metadataworkinggroup.com/schemas/regions/",
+        "stArea": "http://ns.adobe.com/xmp/sType/Area#",
+    }
+
+    def _root(self):
+        packet = xmp_packet_build(SAMPLE)
+        body = re.search(rb"<x:xmpmeta.*</x:xmpmeta>", packet, re.S).group(0)
+        return ET.fromstring(body)
+
+    def test_keywords_sit_where_dublin_core_says(self):
+        found = self._root().findall(".//dc:subject/rdf:Bag/rdf:li", self.NS)
+        assert [item.text for item in found] == ["beach", "sunset"]
+
+    def test_the_rating_sits_where_the_xmp_schema_says(self):
+        assert self._root().find(".//xmp:Rating", self.NS).text == "5"
+
+    def test_a_face_region_sits_where_mwg_says(self):
+        region = self._root().find(
+            ".//mwg-rs:Regions/mwg-rs:RegionList/rdf:Bag/rdf:li", self.NS
+        )
+
+        assert region.find("mwg-rs:Name", self.NS).text == "Mom"
+        assert region.find("mwg-rs:Type", self.NS).text == "Face"
+
+        area = region.find("mwg-rs:Area", self.NS)
+        assert area.get(f"{{{self.NS['stArea']}}}x") == "0.25"
+        assert area.get(f"{{{self.NS['stArea']}}}unit") == "normalized"
+
+
+class TestPacketMerging:
+    """A photo carrying Lightroom's metadata must not lose it to our write."""
+
+    EXISTING = (
+        b'<?xpacket begin="" id="W5M0MpCehiHzreSzNTczkc9d"?>'
+        b'<x:xmpmeta xmlns:x="adobe:ns:meta/">'
+        b'<rdf:RDF xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#">'
+        b'<rdf:Description rdf:about="" '
+        b'xmlns:dc="http://purl.org/dc/elements/1.1/" '
+        b'xmlns:xmp="http://ns.adobe.com/xap/1.0/">'
+        b"<dc:creator><rdf:Seq><rdf:li>Someone Else</rdf:li></rdf:Seq></dc:creator>"
+        b"<dc:rights>All rights reserved</dc:rights>"
+        b"<dc:subject><rdf:Bag><rdf:li>old-tag</rdf:li></rdf:Bag></dc:subject>"
+        b"<xmp:Rating>2</xmp:Rating>"
+        b"</rdf:Description></rdf:RDF></x:xmpmeta>"
+        b'<?xpacket end="w"?>'
+    )
+
+    def test_foreign_properties_survive(self):
+        packet = xmp_packet_build(SAMPLE, existing=self.EXISTING)
+        assert b"Someone Else" in packet
+        assert b"All rights reserved" in packet
+
+    def test_our_own_properties_are_replaced_not_duplicated(self):
+        packet = xmp_packet_build(SAMPLE, existing=self.EXISTING)
+
+        assert b"old-tag" not in packet
+        assert packet.count(b"<dc:subject>") == 1
+        assert packet.count(b"<xmp:Rating>") == 1
+        assert xmp_packet_read(packet)["rating"] == 5
+
+    def test_a_rating_held_as_an_attribute_is_also_replaced(self):
+        """XMP allows either spelling, and a missed one would read as a duplicate."""
+        existing = (
+            b'<x:xmpmeta xmlns:x="adobe:ns:meta/">'
+            b'<rdf:RDF xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#">'
+            b'<rdf:Description rdf:about="" '
+            b'xmlns:xmp="http://ns.adobe.com/xap/1.0/" xmp:Rating="1"/>'
+            b"</rdf:RDF></x:xmpmeta>"
+        )
+        packet = xmp_packet_build({"rating": 4}, existing=existing)
+
+        assert xmp_packet_read(packet)["rating"] == 4
+        assert b'xmp:Rating="1"' not in packet
+
+    def test_a_property_we_do_not_set_is_left_alone(self):
+        """
+        A photo rated in Lightroom that is not a PictoPy favourite still has a
+        rating, and writing keywords must not take it away.
+        """
+        packet = xmp_packet_build({"keywords": ["beach"]}, existing=self.EXISTING)
+
+        assert xmp_packet_read(packet)["rating"] == 2
+        assert xmp_packet_read(packet)["keywords"] == ["beach"]
+
+    def test_clearing_a_property_we_own_does_remove_it(self):
+        """
+        Absent and empty differ: the user removing every tag has to reach the
+        file, or the file keeps claiming tags that no longer exist.
+        """
+        packet = xmp_packet_build({"keywords": []}, existing=self.EXISTING)
+
+        assert "keywords" not in xmp_packet_read(packet)
+        assert b"old-tag" not in packet
+        # Still not ours to touch.
+        assert xmp_packet_read(packet)["rating"] == 2
+
+    def test_a_bare_rdf_root_is_merged_not_discarded(self):
+        """The xmpmeta wrapper is optional, and some writers leave it out."""
+        existing = (
+            b'<rdf:RDF xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#">'
+            b'<rdf:Description rdf:about="" '
+            b'xmlns:dc="http://purl.org/dc/elements/1.1/">'
+            b"<dc:rights>Theirs</dc:rights>"
+            b"</rdf:Description></rdf:RDF>"
+        )
+        packet = xmp_packet_build(SAMPLE, existing=existing)
+
+        assert b"Theirs" in packet
+        assert xmp_packet_read(packet)["rating"] == 5
+
+    def test_writing_twice_does_not_accumulate(self):
+        once = xmp_packet_build(SAMPLE)
+        twice = xmp_packet_build(SAMPLE, existing=once)
+
+        assert twice.count(b"<dc:subject>") == 1
+        assert xmp_packet_read(twice) == xmp_packet_read(once)
+
+
+class TestPacketRefusesHostileInput:
+    # Carries a real rating, so a guard that failed to fire would show up as a
+    # rating being read rather than as an empty result that proves nothing.
+    BOMB = (
+        b'<?xml version="1.0"?>'
+        b'<!DOCTYPE x [<!ENTITY a "aaaaaaaaaa">'
+        b'<!ENTITY b "&a;&a;&a;&a;&a;&a;&a;&a;&a;&a;">]>'
+        b'<x:xmpmeta xmlns:x="adobe:ns:meta/">'
+        b'<rdf:RDF xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#">'
+        b'<rdf:Description rdf:about="" '
+        b'xmlns:xmp="http://ns.adobe.com/xap/1.0/">'
+        b"<xmp:Rating>3</xmp:Rating>"
+        b"</rdf:Description></rdf:RDF></x:xmpmeta>"
+    )
+
+    def test_a_packet_with_a_dtd_is_not_parsed(self):
+        """
+        Entity expansion is the one XML attack ElementTree still allows, and the
+        XMP spec forbids a DTD in a packet anyway.
+        """
+        assert xmp_packet_read(self.BOMB) == {}
+
+    def test_writing_over_an_unreadable_packet_is_refused(self):
+        """
+        We cannot see what the file is carrying, so we cannot know what
+        overwriting it would destroy. Leaving it alone is the only safe answer.
+        """
+        with pytest.raises(UnreadablePacketError):
+            xmp_packet_build(SAMPLE, existing=self.BOMB)
+
+    def test_writing_over_malformed_xml_is_refused(self):
+        with pytest.raises(UnreadablePacketError):
+            xmp_packet_build(SAMPLE, existing=b"<x:xmpmeta>truncated")
+
+    def test_a_photo_carrying_no_packet_is_still_written(self):
+        """Absent is not the same as unreadable, and must not be confused for it."""
+        assert xmp_packet_read(xmp_packet_build(SAMPLE, existing=b""))["rating"] == 5
+
+    def test_malformed_xml_reads_as_empty_rather_than_raising(self):
+        """Reading is best-effort; only writing has anything to lose."""
+        assert xmp_packet_read(b"<x:xmpmeta>truncated") == {}
+
+    def test_a_non_numeric_rating_is_skipped(self):
+        existing = (
+            b'<x:xmpmeta xmlns:x="adobe:ns:meta/">'
+            b'<rdf:RDF xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#">'
+            b'<rdf:Description rdf:about="" '
+            b'xmlns:xmp="http://ns.adobe.com/xap/1.0/">'
+            b"<xmp:Rating>excellent</xmp:Rating>"
+            b"</rdf:Description></rdf:RDF></x:xmpmeta>"
+        )
+        assert "rating" not in xmp_packet_read(existing)
+
+
+class TestRegionOrientation:
+    """
+    A face box is recorded before EXIF rotation is applied; an MWG region is
+    read after it. Getting this wrong puts every box on a rotated photo in the
+    wrong place, and does it silently.
+    """
+
+    # Box in the raw top-left, taller than it is wide, so no symmetry can hide
+    # a transposed axis.
+    BBOX = {"x": 0, "y": 0, "width": 20, "height": 40}
+
+    @pytest.mark.parametrize(
+        "orientation,expected",
+        [
+            (1, (0.1, 0.2, 0.2, 0.4)),
+            (2, (0.9, 0.2, 0.2, 0.4)),
+            (3, (0.9, 0.8, 0.2, 0.4)),
+            (4, (0.1, 0.8, 0.2, 0.4)),
+            (5, (0.2, 0.1, 0.4, 0.2)),
+            (6, (0.8, 0.1, 0.4, 0.2)),
+            (7, (0.8, 0.9, 0.4, 0.2)),
+            (8, (0.2, 0.9, 0.4, 0.2)),
+        ],
+    )
+    def test_every_orientation_lands_where_it_should(self, orientation, expected):
+        assert xmp_packet_orient_region(self.BBOX, 100, 100, orientation) == expected
+
+    def test_the_centre_is_the_centre_not_the_corner(self):
+        """MWG's x and y are the region's midpoint; a corner would offset every box."""
+        centred = xmp_packet_orient_region(
+            {"x": 40, "y": 40, "width": 20, "height": 20}, 100, 100, 1
+        )
+        assert centred == (0.5, 0.5, 0.2, 0.2)
+
+    @pytest.mark.parametrize("orientation", [5, 6, 7, 8])
+    def test_a_quarter_turn_swaps_the_applied_dimensions(self, orientation):
+        assert xmp_packet_applied_dimensions(4000, 3000, orientation) == (3000, 4000)
+
+    @pytest.mark.parametrize("orientation", [1, 2, 3, 4])
+    def test_a_flip_leaves_the_applied_dimensions_alone(self, orientation):
+        assert xmp_packet_applied_dimensions(4000, 3000, orientation) == (4000, 3000)
+
+    def test_an_empty_box_has_no_region(self):
+        assert (
+            xmp_packet_orient_region(
+                {"x": 0, "y": 0, "width": 0, "height": 10}, 100, 100
+            )
+            is None
+        )
+
+    def test_an_unknown_image_size_has_no_region(self):
+        assert xmp_packet_orient_region(self.BBOX, 0, 0) is None
+
+    def test_an_unrecognised_orientation_is_treated_as_upright(self):
+        assert xmp_packet_orient_region(self.BBOX, 100, 100, 99) == (
+            0.1,
+            0.2,
+            0.2,
+            0.4,
+        )
+
+
+class TestJpegSplice:
+    def test_a_packet_survives_the_round_trip(self):
+        written = xmp_segments_write(_jpeg(), b"<x:xmpmeta/>")
+        assert xmp_segments_read(written) == b"<x:xmpmeta/>"
+
+    def test_a_file_without_xmp_reads_as_none(self):
+        assert xmp_segments_read(_jpeg()) is None
+
+    def test_the_image_is_not_recompressed(self):
+        """The whole reason for splicing rather than decode-and-save."""
+        original = _jpeg()
+        written = xmp_segments_write(original, b"<x:xmpmeta/>")
+
+        assert _pixels(original) == _pixels(written)
+        assert _scan_data(written) == _scan_data(original)
+
+    def test_rewriting_replaces_rather_than_appends(self):
+        once = xmp_segments_write(_jpeg(), b"<x:xmpmeta>first</x:xmpmeta>")
+        twice = xmp_segments_write(once, b"<x:xmpmeta>second</x:xmpmeta>")
+
+        assert xmp_segments_read(twice) == b"<x:xmpmeta>second</x:xmpmeta>"
+        assert twice.count(b"http://ns.adobe.com/xap/1.0/\x00") == 1
+        assert b"first" not in twice
+
+    def test_the_segment_lands_after_the_leading_app_run(self):
+        """JFIF expects its APP0 first; putting ours ahead of it breaks readers."""
+        written = xmp_segments_write(_jpeg(), b"<x:xmpmeta/>")
+        assert written.index(b"JFIF") < written.index(b"ns.adobe.com/xap")
+
+    def test_existing_exif_is_preserved(self):
+        image = Image.new("RGB", (16, 16), "white")
+        exif = image.getexif()
+        exif[0x010E] = "a description"
+        buffer = io.BytesIO()
+        image.save(buffer, "JPEG", exif=exif)
+
+        written = xmp_segments_write(buffer.getvalue(), b"<x:xmpmeta/>")
+
+        assert Image.open(io.BytesIO(written)).getexif()[0x010E] == "a description"
+
+    def test_an_oversized_packet_is_refused(self):
+        """Beyond this a packet needs ExtendedXMP; a truncated segment is corrupt."""
+        with pytest.raises(ValueError, match="segment limit"):
+            xmp_segments_write(_jpeg(), b"x" * 70000)
+
+    def test_a_packet_just_under_the_limit_is_accepted(self):
+        packet = b"<x:xmpmeta>" + b"y" * 65000 + b"</x:xmpmeta>"
+        written = xmp_segments_write(_jpeg(), packet)
+        assert xmp_segments_read(written) == packet
+
+
+class TestPngSplice:
+    def test_a_packet_survives_the_round_trip(self):
+        written = xmp_segments_write(_png(), b"<x:xmpmeta/>")
+        assert xmp_segments_read(written) == b"<x:xmpmeta/>"
+
+    def test_a_file_without_xmp_reads_as_none(self):
+        assert xmp_segments_read(_png()) is None
+
+    def test_the_image_data_is_untouched(self):
+        original = _png()
+        written = xmp_segments_write(original, b"<x:xmpmeta/>")
+
+        assert _pixels(original) == _pixels(written)
+        assert _idat(written) == _idat(original)
+
+    def test_rewriting_replaces_rather_than_appends(self):
+        once = xmp_segments_write(_png(), b"<x:xmpmeta>first</x:xmpmeta>")
+        twice = xmp_segments_write(once, b"<x:xmpmeta>second</x:xmpmeta>")
+
+        assert xmp_segments_read(twice) == b"<x:xmpmeta>second</x:xmpmeta>"
+        assert twice.count(b"XML:com.adobe.xmp") == 1
+
+    def test_the_chunk_carries_a_valid_crc(self):
+        """A bad CRC makes the file unreadable to a strict decoder."""
+        written = xmp_segments_write(_png(), b"<x:xmpmeta/>")
+        Image.open(io.BytesIO(written)).load()
+
+    def test_the_chunk_goes_after_ihdr(self):
+        written = xmp_segments_write(_png(), b"<x:xmpmeta/>")
+        assert written.index(b"IHDR") < written.index(b"iTXt")
+
+
+class TestUnsupportedFormats:
+    def test_reading_an_unknown_container_raises(self):
+        with pytest.raises(UnsupportedImageError):
+            xmp_segments_read(b"GIF89a not really an image")
+
+    def test_writing_an_unknown_container_raises(self):
+        with pytest.raises(UnsupportedImageError):
+            xmp_segments_write(b"GIF89a not really an image", b"<x:xmpmeta/>")
+
+
+class TestEndToEnd:
+    """Packet and splice together, which is how PR 4 will use them."""
+
+    def test_a_tagged_photo_keeps_its_pixels_and_its_metadata(self):
+        original = _jpeg()
+        packet = xmp_packet_build(SAMPLE)
+        written = xmp_segments_write(original, packet)
+
+        assert _pixels(original) == _pixels(written)
+        assert xmp_packet_read(xmp_segments_read(written)) == xmp_packet_read(packet)
+
+    def test_retagging_preserves_another_application_s_work(self):
+        theirs = (
+            b'<x:xmpmeta xmlns:x="adobe:ns:meta/">'
+            b'<rdf:RDF xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#">'
+            b'<rdf:Description rdf:about="" '
+            b'xmlns:dc="http://purl.org/dc/elements/1.1/">'
+            b"<dc:rights>Theirs</dc:rights>"
+            b"</rdf:Description></rdf:RDF></x:xmpmeta>"
+        )
+        photo = xmp_segments_write(_jpeg(), theirs)
+
+        existing = xmp_segments_read(photo)
+        updated = xmp_segments_write(photo, xmp_packet_build(SAMPLE, existing=existing))
+
+        assert b"Theirs" in updated
+        assert xmp_packet_read(xmp_segments_read(updated))["rating"] == 5
diff --git a/frontend/src/api/apiEndpoints.ts b/frontend/src/api/apiEndpoints.ts
index b6f9db5c0..81cf280ee 100644
--- a/frontend/src/api/apiEndpoints.ts
+++ b/frontend/src/api/apiEndpoints.ts
@@ -15,6 +15,11 @@ export const videosEndpoints = {
   purgeFrameCache: '/videos/purge-frame-cache',
 };
 
+export const metadataSyncEndpoints = {
+  status: '/metadata-sync/status',
+  run: '/metadata-sync/run',
+};
+
 export const faceClustersEndpoints = {
   getAllClusters: '/face-clusters/',
   searchForFaces: '/face-clusters/face-search?input_type=path',
diff --git a/sync-microservice/app/database/self_writes.py b/sync-microservice/app/database/self_writes.py
new file mode 100644
index 000000000..4e74e1704
--- /dev/null
+++ b/sync-microservice/app/database/self_writes.py
@@ -0,0 +1,75 @@
+"""
+Reads the self-write ledger the primary backend maintains.
+
+The table is created and populated by `backend/app/database/self_writes.py`;
+this service only claims entries. The path format is a contract with that
+module: both sides key on `os.path.normcase(os.path.abspath(path))`.
+"""
+
+import os
+import sqlite3
+from typing import List, Set, Tuple
+
+from app.config.settings import DATABASE_PATH
+from app.logging.setup_logging import get_sync_logger
+
+logger = get_sync_logger(__name__)
+
+# (path, file_size, file_mtime) as observed on disk right now.
+ObservedFile = Tuple[str, int, int]
+
+
+def _self_write_key(path: str) -> str:
+    return os.path.normcase(os.path.abspath(path))
+
+
+def db_take_matching_self_writes(observed: List[ObservedFile]) -> Set[str]:
+    """
+    Return the observed paths PictoPy wrote itself, and claim them.
+
+    Claiming as we match means a second event for one write reads as a real
+    change. That is the safe direction: it costs a redundant rescan, where
+    keeping the entry could swallow a genuine edit to the same path.
+    """
+    if not observed:
+        return set()
+
+    conn = None
+    matched: Set[str] = set()
+
+    try:
+        conn = sqlite3.connect(DATABASE_PATH)
+        cursor = conn.cursor()
+
+        keys = {_self_write_key(path) for path, _, _ in observed}
+        placeholders = ",".join("?" for _ in keys)
+        cursor.execute(
+            f"""
+            SELECT path, file_size, file_mtime
+            FROM self_writes
+            WHERE path IN ({placeholders})
+            """,
+            list(keys),
+        )
+        recorded = {row[0]: (row[1], row[2]) for row in cursor.fetchall()}
+
+        claimed = []
+        for path, size, mtime in observed:
+            key = _self_write_key(path)
+            if recorded.get(key) == (size, mtime):
+                matched.add(path)
+                claimed.append((key,))
+
+        if claimed:
+            cursor.executemany("DELETE FROM self_writes WHERE path = ?", claimed)
+            conn.commit()
+
+        return matched
+    except sqlite3.Error as e:
+        # Failing open costs a redundant rescan; failing closed would drop a
+        # real change, so an unreadable ledger must never suppress anything.
+        logger.error(f"Error matching self writes: {e}")
+        return set()
+    finally:
+        if conn is not None:
+            conn.close()
diff --git a/sync-microservice/app/utils/watcher.py b/sync-microservice/app/utils/watcher.py
index 902a50cf5..6117efd54 100644
--- a/sync-microservice/app/utils/watcher.py
+++ b/sync-microservice/app/utils/watcher.py
@@ -2,10 +2,11 @@
 import threading
 import time
 import logging
-from typing import List, Tuple, Dict, Optional
+from typing import List, Set, Tuple, Dict, Optional
 from watchfiles import watch, Change
 import httpx
 from app.database.folders import db_get_all_folders_with_ids
+from app.database.self_writes import db_take_matching_self_writes
 from app.config.settings import PRIMARY_BACKEND_URL
 from app.logging.setup_logging import get_sync_logger
 
@@ -21,6 +22,10 @@
 
 FolderIdPath = Tuple[str, str]
 
+# Mirrors backend/app/utils/self_write.py, which names its temp files this way
+# so they can be recognised here.
+SELF_WRITE_TEMP_PREFIX = ".pictopy-write-"
+
 # Global variables to track watcher state
 watcher_thread: Optional[threading.Thread] = None
 stop_event = threading.Event()
@@ -49,6 +54,36 @@ def watcher_util_get_folder_id_if_watched(file_path: str) -> Optional[str]:
     return None
 
 
+def watcher_util_is_own_temp_file(file_path: str) -> bool:
+    """
+    True for the scratch file a PictoPy write leaves in the watched folder.
+
+    It has to be created beside its target for the rename to stay atomic, so the
+    watcher sees it appear and disappear. Prefix mirrors
+    backend/app/utils/self_write.py.
+    """
+    return os.path.basename(file_path).startswith(SELF_WRITE_TEMP_PREFIX)
+
+
+def watcher_util_drop_self_writes(file_paths: List[str]) -> Set[str]:
+    """
+    Of the given paths, those matching a write PictoPy just made itself.
+
+    Batched into one query because a metadata sync pass touches many files at
+    once.
+    """
+    observed = []
+    for file_path in file_paths:
+        try:
+            stats = os.stat(file_path)
+        except OSError:
+            # Gone or unreadable: not something we can claim as our own write.
+            continue
+        observed.append((file_path, stats.st_size, int(stats.st_mtime)))
+
+    return db_take_matching_self_writes(observed)
+
+
 def watcher_util_handle_file_changes(changes: set) -> None:
     """
     Handle file changes detected by watchfiles.
@@ -60,14 +95,30 @@ def watcher_util_handle_file_changes(changes: set) -> None:
 
     affected_folders = {}  # folder_path -> folder_id mapping
 
+    # PictoPy writing metadata into a photo is a file modification like any
+    # other, and resyncing the folder it lives in would undo the write's whole
+    # point. Both passes below run before anything is attributed to a folder.
+    surviving = [
+        (change, file_path)
+        for change, file_path in changes
+        if not watcher_util_is_own_temp_file(file_path)
+    ]
+    self_written = watcher_util_drop_self_writes(
+        [file_path for change, file_path in surviving if change != Change.deleted]
+    )
+    if self_written:
+        logger.debug(f"Ignoring {len(self_written)} change(s) PictoPy made itself")
+
     # First pass - count changes and identify affected folders
-    for change, file_path in changes:
+    for change, file_path in surviving:
         # Process deletions
         if change == Change.deleted:
             deleted_folder_id = watcher_util_get_folder_id_if_watched(file_path)
             if deleted_folder_id:
                 deleted_folder_ids.append(deleted_folder_id)
                 continue
+        elif file_path in self_written:
+            continue
 
         # Find affected folder
         closest_folder = watcher_util_find_closest_parent_folder(