diff --git a/app/assets/scanner.py b/app/assets/scanner.py
index eacf86867..0666e87a9 100644
--- a/app/assets/scanner.py
+++ b/app/assets/scanner.py
@@ -397,7 +397,15 @@ def get_unenriched_assets_for_roots(
             .where(AssetContent.is_missing.is_(False))
         )
         if compute_hashes:
-            query = query.where(AssetContent.hash.is_(None))
+            # A split-created record has a hash but NULL metadata; it must still
+            # enrich. Widen the hash-mode branch so a missing hash OR missing
+            # metadata makes a row a candidate.
+            query = query.where(
+                sa.or_(
+                    AssetContent.hash.is_(None),
+                    Asset.system_metadata.is_(None),
+                )
+            )
         else:
             query = query.where(Asset.system_metadata.is_(None))
         rows = sess.execute(query.order_by(Asset.id)).all()
@@ -449,9 +457,14 @@ def enrich_asset(
         if metadata:
             mime_type = metadata.content_type
 
+    content = session.get(AssetContent, content_id)
+
     digest: str | None = None
     stored_hash: str | None = None
-    if compute_hash:
+    # A split-created record already carries the hash of its current bytes; only
+    # hash when content has none, so a metadata-only enrich never re-hashes a
+    # large file that is already identified.
+    if compute_hash and content is not None and content.hash is None:
         try:
             snapshot = snapshot_hash(file_path)
             if snapshot is None:
@@ -466,7 +479,6 @@ def enrich_asset(
             logging.warning("Failed to hash %s: %s", file_path, e)
 
     # Optimistic guard: discard results if content changed during enrichment.
-    content = session.get(AssetContent, content_id)
     record = session.get(Asset, record_id)
     if content is None or record is None or content.mtime_ns != initial_mtime_ns:
         session.rollback()
diff --git a/app/assets/scanner_changes.py b/app/assets/scanner_changes.py
index 71bfbff5d..317a982c5 100644
--- a/app/assets/scanner_changes.py
+++ b/app/assets/scanner_changes.py
@@ -107,6 +107,12 @@ def detect_content_change(
     if hashing_is_enabled:
         queue_pending_verification(content.id)
         return
+    if content.size_bytes == stat_result.st_size:
+        # User identity rule: a same-size mtime bump (rsync, cloud sync, backup
+        # restore) is the same file. Without a hash to prove a content change,
+        # treat it as a no-op rather than split and destroy the user's tags and
+        # metadata. Hash mode above defers to the snapshot hash instead.
+        return
     split_content(session, content, stat_result, hash_value=None)
 
 
diff --git a/tests-unit/assets_test/services/test_detection_gate.py b/tests-unit/assets_test/services/test_detection_gate.py
index 44acd18b9..ebf7f2d31 100644
--- a/tests-unit/assets_test/services/test_detection_gate.py
+++ b/tests-unit/assets_test/services/test_detection_gate.py
@@ -51,7 +51,7 @@ def _stored_hash(path: Path) -> str:
     return to_stored_hash(digest)
 
 
-def test_off_mode_touch_splits(session, temp_dir: Path):
+def test_off_mode_same_size_touch_does_not_split(session, temp_dir: Path):
     input_root = temp_dir / "input"
     input_root.mkdir()
     path = input_root / "touched.bin"
@@ -66,6 +66,31 @@ def test_off_mode_touch_splits(session, temp_dir: Path):
         sync_prefixes_with_filesystem(session, [str(input_root)])
     session.commit()
 
+    # User identity rule: a same-size mtime bump (rsync, cloud sync, backup
+    # restore) is the same file. Without a hash to prove a content change, OFF
+    # mode must not split and destroy the record's tags/metadata.
+    contents = list(session.scalars(select(AssetContent)))
+    assert len(contents) == 1
+    assert session.get(AssetContent, old_content.id).is_missing is False
+
+
+def test_off_mode_size_change_splits(session, temp_dir: Path):
+    input_root = temp_dir / "input"
+    input_root.mkdir()
+    path = input_root / "grown.bin"
+    path.write_bytes(b"small")
+    old_content, _ = _seed_content(session, path, hash_value="historical")
+    path.write_bytes(b"a decidedly larger set of bytes")
+
+    with (
+        patch("folder_paths.get_input_directory", return_value=str(input_root)),
+        patch("app.assets.scanner.mode.hashing_enabled", return_value=False),
+    ):
+        sync_prefixes_with_filesystem(session, [str(input_root)])
+    session.commit()
+
+    # A genuine change (mtime AND size both moved) still splits: the old bytes
+    # are retired and a fresh replacement takes the live path.
     contents = list(session.scalars(select(AssetContent).order_by(AssetContent.created_at)))
     assert len(contents) == 2
     assert session.get(AssetContent, old_content.id).is_missing is True
diff --git a/tests-unit/assets_test/services/test_split_policy.py b/tests-unit/assets_test/services/test_split_policy.py
new file mode 100644
index 000000000..46b83147a
--- /dev/null
+++ b/tests-unit/assets_test/services/test_split_policy.py
@@ -0,0 +1,280 @@
+"""Unified split policy: identity by (path, size), NULL-metadata replacements.
+
+Three properties are locked here:
+
+1. Enrich candidacy (hash mode). The hash-mode enrich predicate must return a
+   split-created record — hash set, ``system_metadata`` NULL — so its metadata
+   is filled in a later pass. A fully enriched record (hash + metadata) must
+   not be returned.
+
+2. Split gate. ``detect_content_change`` splits only when *both* ``mtime_ns``
+   and ``size_bytes`` change. A bare mtime bump at the same size (rsync, cloud
+   sync, backup restore) is the same file and must not split — user tags and
+   metadata survive on the live record. ``mtime`` unchanged stays the existing
+   Ruling #10 early return.
+
+3. Replacement shape. Both split sites (``scanner_changes.split_content`` and
+   ``hash_mode_state.drain_transition_queue``) create the replacement record
+   with real SQL NULL ``system_metadata`` — byte-derived metadata from the old
+   bytes is never carried onto the new bytes.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterator
+from contextlib import contextmanager
+from dataclasses import dataclass
+from pathlib import Path
+from unittest.mock import patch
+
+import sqlalchemy as sa
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+import pytest
+
+from app.assets.database.models import Asset, AssetContent
+from app.assets.database.queries import create_content, create_record
+from app.assets.database.queries.records import fetch_record_tags
+from app.assets.helpers import to_stored_hash
+from app.assets.scanner import get_unenriched_assets_for_roots
+from app.assets.scanner_changes import detect_content_change
+from app.assets.services.hash_mode_state import (
+    clear_transition_queue,
+    drain_transition_queue,
+    enqueue_transition_work,
+)
+from app.assets.services.snapshot_hash import snapshot_hash
+
+
+@dataclass(frozen=True, slots=True)
+class _FakeStat:
+    """Minimal stand-in for ``os.stat_result`` (only size + mtime are read)."""
+
+    st_size: int
+    st_mtime_ns: int
+
+
+@contextmanager
+def _reuse_session(session: Session) -> Iterator[Session]:
+    """Hand the seeded session to scanner.create_session without closing it."""
+    yield session
+
+
+def _raw_system_metadata(session: Session, record_id: str) -> object:
+    """Read the stored column value with no ORM/JSON type processing."""
+    return session.execute(
+        sa.text("SELECT system_metadata FROM assets WHERE id = :id"),
+        {"id": record_id},
+    ).scalar()
+
+
+def _candidates_under(session: Session, temp_dir: Path, *, compute_hashes: bool) -> set[str]:
+    with (
+        patch("app.assets.scanner.create_session", lambda: _reuse_session(session)),
+        patch(
+            "app.assets.scanner.get_scan_prefixes_for_root",
+            return_value=[str(temp_dir)],
+        ),
+    ):
+        rows = get_unenriched_assets_for_roots(("models",), compute_hashes=compute_hashes)
+    return {row.record_id for row in rows}
+
+
+@pytest.fixture(autouse=True)
+def _transition_queue_isolation() -> Iterator[None]:
+    clear_transition_queue()
+    yield
+    clear_transition_queue()
+
+
+@pytest.fixture(autouse=True)
+def _input_base_is_temp_dir(temp_dir: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """Make temp_dir a recognized input base so path-derived tags resolve.
+
+    ``split_content`` and ``drain_transition_queue`` derive replacement tags from
+    the path; a bare ``/tmp`` path is under no known root and would raise.
+    """
+    monkeypatch.setattr("folder_paths.get_input_directory", lambda: str(temp_dir))
+
+
+# --- (i) + (ii): hash-mode enrich candidacy ---------------------------------
+
+
+def test_split_record_is_enrich_candidate_in_hash_mode(
+    session: Session, temp_dir: Path
+) -> None:
+    # Given a split-created record: hash set, system_metadata NULL
+    path = temp_dir / "split.safetensors"
+    content = create_content(session, str(path), hash="blake3:deadbeef")
+    record = create_record(session, content.id, path.name)
+    session.commit()
+    record_id = record.id
+
+    # When enrichment candidates are queried in hash mode
+    candidates = _candidates_under(session, temp_dir, compute_hashes=True)
+
+    # Then the split record is a candidate: it still needs metadata
+    assert record_id in candidates
+
+
+def test_enriched_record_is_not_enrich_candidate_in_hash_mode(
+    session: Session, temp_dir: Path
+) -> None:
+    # Given a fully enriched record: hash set AND system_metadata set
+    path = temp_dir / "enriched.safetensors"
+    content = create_content(session, str(path), hash="blake3:deadbeef")
+    record = create_record(
+        session, content.id, path.name, system_metadata={"architecture": "flux"}
+    )
+    session.commit()
+    record_id = record.id
+
+    # When enrichment candidates are queried in hash mode
+    candidates = _candidates_under(session, temp_dir, compute_hashes=True)
+
+    # Then a fully enriched record is not returned
+    assert record_id not in candidates
+
+
+# --- (iii) + (iv) + (v): the split gate -------------------------------------
+
+
+def test_same_size_mtime_bump_does_not_split(session: Session, temp_dir: Path) -> None:
+    # Given a live record carrying user tags and metadata
+    path = temp_dir / "touched.safetensors"
+    content = create_content(session, str(path), hash=None, size_bytes=100, mtime_ns=1000)
+    record = create_record(
+        session, content.id, path.name, tags=["keepme"], system_metadata={"k": "v"}
+    )
+    session.commit()
+    content_id, record_id = content.id, record.id
+
+    # When a same-size mtime bump is observed (OFF mode)
+    detect_content_change(
+        session, content, _FakeStat(st_size=100, st_mtime_ns=2000), hashing_is_enabled=False
+    )
+    session.commit()
+    session.expire_all()
+
+    # Then nothing splits: one live row, tags and metadata survive
+    live = session.get(AssetContent, content_id)
+    assert live is not None and live.is_missing is False
+    rows_at_path = list(
+        session.scalars(select(AssetContent).where(AssetContent.path == str(path)))
+    )
+    assert len(rows_at_path) == 1
+    surviving = session.get(Asset, record_id)
+    assert surviving.system_metadata == {"k": "v"}
+    tags = fetch_record_tags(session, record_id)
+    assert "keepme" in tags
+    assert "missing" not in tags
+
+
+def test_mtime_and_size_change_splits_with_null_metadata(
+    session: Session, temp_dir: Path
+) -> None:
+    # Given a live record carrying user tags and metadata
+    path = temp_dir / "grown.safetensors"
+    content = create_content(session, str(path), hash=None, size_bytes=100, mtime_ns=1000)
+    create_record(
+        session, content.id, path.name, tags=["oldtag"], system_metadata={"k": "v"}
+    )
+    session.commit()
+    old_content_id = content.id
+
+    # When both mtime AND size change (genuine content change, OFF mode)
+    detect_content_change(
+        session, content, _FakeStat(st_size=200, st_mtime_ns=2000), hashing_is_enabled=False
+    )
+    session.commit()
+    session.expire_all()
+
+    # Then the old content is retired and a fresh replacement takes its place
+    assert session.get(AssetContent, old_content_id).is_missing is True
+    live = session.scalar(
+        select(AssetContent).where(
+            AssetContent.path == str(path), AssetContent.is_missing.is_(False)
+        )
+    )
+    assert live is not None and live.id != old_content_id
+    assert live.size_bytes == 200 and live.mtime_ns == 2000
+    new_record = session.scalar(select(Asset).where(Asset.content_id == live.id))
+    assert new_record is not None
+
+    # ... the replacement carries real NULL metadata (never the old bytes' meta)
+    assert new_record.system_metadata is None
+    assert _raw_system_metadata(session, new_record.id) is None
+    # ... a genuine change drops the old user tags
+    assert "oldtag" not in fetch_record_tags(session, new_record.id)
+    # ... and the fresh replacement is immediately enrichable
+    assert new_record.id in _candidates_under(session, temp_dir, compute_hashes=True)
+
+
+def test_mtime_unchanged_size_changed_does_not_split(
+    session: Session, temp_dir: Path
+) -> None:
+    # Given a live record
+    path = temp_dir / "weird.safetensors"
+    content = create_content(session, str(path), hash=None, size_bytes=100, mtime_ns=1000)
+    create_record(session, content.id, path.name, tags=["keepme"])
+    session.commit()
+    content_id = content.id
+
+    # When size drifts but mtime is unchanged (Ruling #10 undefined territory)
+    detect_content_change(
+        session, content, _FakeStat(st_size=999, st_mtime_ns=1000), hashing_is_enabled=False
+    )
+    session.commit()
+    session.expire_all()
+
+    # Then the existing early return holds: no split
+    assert session.get(AssetContent, content_id).is_missing is False
+    rows_at_path = list(
+        session.scalars(select(AssetContent).where(AssetContent.path == str(path)))
+    )
+    assert len(rows_at_path) == 1
+
+
+# --- (vi): hash_mode_state split site produces the same replacement shape ----
+
+
+def test_transition_drain_split_replacement_has_null_metadata(
+    session: Session, temp_dir: Path
+) -> None:
+    # Given a live hashed record carrying user tags and metadata
+    path = temp_dir / "changed.bin"
+    path.write_bytes(b"old bytes")
+    old_snapshot = snapshot_hash(str(path))
+    assert old_snapshot is not None
+    old_digest, _ = old_snapshot
+    stat = path.stat()
+    old_content = create_content(
+        session, str(path), to_stored_hash(old_digest), stat.st_size, stat.st_mtime_ns
+    )
+    old_content_id = old_content.id
+    create_record(
+        session, old_content_id, "changed.bin", tags=["oldtag"], system_metadata={"k": "v"}
+    )
+    # ... whose bytes then change to a different hash
+    path.write_bytes(b"different new bytes")
+
+    # When the OFF->ON transition drains and splits the changed row
+    enqueue_transition_work(session, "off_to_on")
+    drain_transition_queue(session)
+    session.commit()
+    session.expire_all()
+
+    # Then the replacement has the same shape as the scanner split: NULL metadata
+    assert session.get(AssetContent, old_content_id).is_missing is True
+    live = session.scalar(
+        select(AssetContent).where(
+            AssetContent.path == str(path), AssetContent.is_missing.is_(False)
+        )
+    )
+    assert live is not None and live.id != old_content_id
+    new_record = session.scalar(select(Asset).where(Asset.content_id == live.id))
+    assert new_record is not None
+    assert new_record.system_metadata is None
+    assert _raw_system_metadata(session, new_record.id) is None
+    assert "oldtag" not in fetch_record_tags(session, new_record.id)