diff --git a/docs/merge-unification-flow.md b/docs/merge-unification-flow.md
index a776a27..35a88f9 100644
--- a/docs/merge-unification-flow.md
+++ b/docs/merge-unification-flow.md
@@ -31,6 +31,7 @@ only an unverified email, the merge is refused with `422 Unverified email`.
 ```
 merge(survivor S, duplicate D, actor A):
   reject if S == D                         -> 400 SameUser
+  acquire shared user-operation locks for S and D
   load S, D (must exist)                   -> 404 UserNotFound
   reject if S or D disabled                -> 409 InactiveAccount
   decision = decide_match(S, D, explicit)
@@ -55,6 +56,7 @@ merge(survivor S, duplicate D, actor A):
   disable D                                       # D can never authenticate again
   audit "duplicate_tombstoned"
   audit "merge_completed" {moved_*, conflicts}
+  release shared user-operation locks
   return MergeResult{..., audit_id}
 ```
 
@@ -72,6 +74,25 @@ The duplicate is **not deleted**. Its Keycloak user attribute
 (`enabled: false`). This preserves forensic history and lets any stale reference
 resolve to the survivor.
 
+### SCIM/merge serialization invariant
+
+`PUT /scim/v2/Users/{id}` performs a full Keycloak user-representation write and
+can set `active: true`. Its tombstone check and replacement PUT therefore execute
+inside the **same user-operation lock** used by the complete merge transaction.
+A merge cannot create `merged_into_user_id` between those two Admin API calls,
+and SCIM cannot wipe a newly-created tombstone or reactivate the duplicate.
+
+Standalone deployments use a dedicated SQLite sidecar lock database and hold a
+`BEGIN IMMEDIATE` transaction for the complete critical section. This provides a
+crash-safe mutex shared by every worker/process using the same database path;
+process death closes the connection and releases the lock. The current backend
+serializes all user mutations conservatively rather than risking a multi-user
+deadlock. Lock acquisition waits up to 10 seconds, then returns retryable HTTP
+`503` without performing a partial mutation. A clustered Postgres deployment
+must provide the same `UserOperationLocks` contract (for example, ordered
+advisory locks) and wire one shared instance into both the merge service and SCIM
+router.
+
 ## Audit
 
 Every step emits an immutable `account_merge_audit` event sharing one
diff --git a/services/account_unification/app/api.py b/services/account_unification/app/api.py
index 3e8119b..fe042a7 100644
--- a/services/account_unification/app/api.py
+++ b/services/account_unification/app/api.py
@@ -13,6 +13,7 @@
 )
 from .models import FederatedIdentity, MergeRequest, MergeResult, UserAccount
 from .service import UnificationService
+from .user_locks import UserOperationLockTimeout
 
 router = APIRouter()
 
@@ -75,6 +76,11 @@ def merge_accounts(
         raise HTTPException(status_code=409, detail=str(exc)) from exc
     except InactiveAccountError as exc:
         raise HTTPException(status_code=409, detail=str(exc)) from exc
+    except UserOperationLockTimeout as exc:
+        raise HTTPException(
+            status_code=503,
+            detail="one of the requested accounts is being modified; retry",
+        ) from exc
 
 
 @router.get("/merges/{audit_id}/audit", tags=["merge"])
diff --git a/services/account_unification/app/healthcheck.py b/services/account_unification/app/healthcheck.py
index 673fffc..22a1148 100644
--- a/services/account_unification/app/healthcheck.py
+++ b/services/account_unification/app/healthcheck.py
@@ -7,16 +7,49 @@
 
 import json
 import sys
+import urllib.parse
 import urllib.request
 
 DEFAULT_URL = "http://127.0.0.1:8099/healthz"
+_ALLOWED_SCHEMES = frozenset({"http", "https"})
+
+
+class _HttpOnlyRedirectHandler(urllib.request.HTTPRedirectHandler):
+    """Drop redirects whose target scheme is not HTTP(S)."""
+
+    def redirect_request(self, req, fp, code, msg, headers, newurl):  # noqa: ANN001
+        """Return an HTTP(S) redirect request, or reject another scheme."""
+        if urllib.parse.urlsplit(newurl).scheme.lower() not in _ALLOWED_SCHEMES:
+            return None
+        return super().redirect_request(req, fp, code, msg, headers, newurl)
+
+
+def _build_http_only_opener() -> urllib.request.OpenerDirector:
+    """Build an opener that can speak only HTTP(S), without file/FTP handlers."""
+    opener = urllib.request.OpenerDirector()
+    opener.add_handler(urllib.request.HTTPHandler())
+    opener.add_handler(urllib.request.HTTPSHandler())
+    opener.add_handler(_HttpOnlyRedirectHandler())
+    opener.add_handler(urllib.request.HTTPErrorProcessor())
+    return opener
+
+
+def _open_health_url(url: str):  # noqa: ANN202
+    """Open an HTTP(S) health URL through the restricted opener."""
+    return _build_http_only_opener().open(url, timeout=5)
 
 
 def main(url: str = DEFAULT_URL) -> int:
     """Check the configured health endpoint and return a shell status code."""
+    scheme = urllib.parse.urlsplit(url).scheme.lower()
+    if scheme not in _ALLOWED_SCHEMES:
+        print(f"healthcheck failed: unsupported URL scheme {scheme!r}", file=sys.stderr)
+        return 1
     try:
-        # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected -- container healthcheck against a hardcoded loopback default (127.0.0.1); any override is a deployment-controlled target, not user input.
-        with urllib.request.urlopen(url, timeout=5) as response:  # noqa: S310
+        # The initial and redirected schemes are constrained to HTTP(S), and the
+        # opener carries no handler for file, FTP, or data URLs.
+        # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected
+        with _open_health_url(url) as response:  # noqa: S310
             body = json.loads(response.read().decode("utf-8"))
     except Exception as exc:  # pragma: no cover - network failure path
         print(f"healthcheck failed: {exc}", file=sys.stderr)
diff --git a/services/account_unification/app/keycloak_client.py b/services/account_unification/app/keycloak_client.py
index 7d6ab05..4ab2d57 100644
--- a/services/account_unification/app/keycloak_client.py
+++ b/services/account_unification/app/keycloak_client.py
@@ -108,6 +108,10 @@ def set_user_attribute(self, user_id: str, key: str, value: str) -> None:
         """Set one single-valued user attribute."""
         ...
 
+    def get_user_attribute(self, user_id: str, key: str) -> str | None:
+        """Return one single-valued user attribute, or ``None`` if unset."""
+        ...
+
 
 class HttpAdminApi:
     """httpx-backed :class:`AdminApi` for a live Keycloak instance.
@@ -339,6 +343,18 @@ def set_user_attribute(self, user_id: str, key: str, value: str) -> None:
             f"/admin/realms/{self._realm}/users/{user_id}", {"attributes": attributes}
         )
 
+    def get_user_attribute(self, user_id: str, key: str) -> str | None:
+        """Return one single-valued Keycloak user attribute, or ``None``.
+
+        Keycloak stores attributes as string lists; this returns the first
+        element (or a bare string, defensively), or ``None`` when unset.
+        """
+        data = self._get(f"/admin/realms/{self._realm}/users/{user_id}")
+        value = (data.get("attributes") or {}).get(key)
+        if isinstance(value, list):
+            return value[0] if value else None
+        return value if isinstance(value, str) else None
+
     # -- transport ---------------------------------------------------------
     def _get(self, path: str, params: dict | None = None) -> dict | list:
         """Issue an authenticated GET and parse JSON."""
diff --git a/services/account_unification/app/main.py b/services/account_unification/app/main.py
index ddff44a..355bc09 100644
--- a/services/account_unification/app/main.py
+++ b/services/account_unification/app/main.py
@@ -20,6 +20,7 @@
 from .keycloak_client import HttpAdminApi
 from .scim import scim_router
 from .service import UnificationService
+from .user_locks import SqliteUserOperationLocks
 
 
 def build_service(app: FastAPI) -> None:
@@ -39,10 +40,22 @@ def build_service(app: FastAPI) -> None:
     # a Postgres-backed sink writing account_merge_audit.
     audit_path = descriptor.sqlite_path or "account_unification.db"
     audit = AuditLogger(SqliteAuditSink(audit_path))
+    # Use a dedicated sidecar database so the serialization transaction never
+    # blocks config reads/writes or audit persistence. Every service worker that
+    # shares this path also shares the same crash-safe SQLite mutex.
+    user_operation_locks = SqliteUserOperationLocks(
+        f"{audit_path}.user-operation-locks.sqlite3"
+    )
 
-    app.state.unification_service = UnificationService(api, audit, config)
+    app.state.unification_service = UnificationService(
+        api,
+        audit,
+        config,
+        user_operation_locks,
+    )
     app.state.audit_logger = audit
     app.state.keycloak_api = api
+    app.state.user_operation_locks = user_operation_locks
     app.state.ready = True
 
 
diff --git a/services/account_unification/app/scim.py b/services/account_unification/app/scim.py
index 2cb7609..4c0b9fb 100644
--- a/services/account_unification/app/scim.py
+++ b/services/account_unification/app/scim.py
@@ -20,6 +20,11 @@
 
 from .keycloak_client import AdminApi
 from .models import UserAccount
+from .service import TOMBSTONE_ATTRIBUTE_KEY
+from .user_locks import (
+    UserOperationLocks,
+    UserOperationLockTimeout,
+)
 
 SCIM_USER_SCHEMA = "urn:ietf:params:scim:schemas:core:2.0:User"
 SCIM_LIST_SCHEMA = "urn:ietf:params:scim:api:messages:2.0:ListResponse"
@@ -38,6 +43,14 @@ def get_provisioner(request: Request) -> AdminApi:
     return api
 
 
+def get_user_operation_locks(request: Request) -> UserOperationLocks:
+    """Return the shared lock manager used by both SCIM and merge paths."""
+    locks = getattr(request.app.state, "user_operation_locks", None)
+    if locks is None:  # pragma: no cover - only when misconfigured / test wiring
+        raise HTTPException(status_code=503, detail="user operation locks not wired")
+    return locks
+
+
 def _scim_error(status: int, detail: str) -> HTTPException:
     """Build a SCIM-shaped HTTP error."""
     return HTTPException(
@@ -186,15 +199,35 @@ def replace_user(
     user_id: str,
     resource: dict[str, Any],
     provisioner: AdminApi = Depends(get_provisioner),
+    user_operation_locks: UserOperationLocks = Depends(get_user_operation_locks),
 ) -> Response:
     """Replace a provisioned user from a SCIM PUT request."""
     try:
-        provisioner.get_user(user_id)
-    except KeyError as exc:
-        raise _scim_error(404, f"user '{user_id}' not found") from exc
-    account = _to_user_account(resource, user_id=user_id)
-    provisioner.replace_user(user_id, account)
-    return _scim_response(_to_scim_resource(provisioner.get_user(user_id)))
+        with user_operation_locks.hold(user_id):
+            try:
+                provisioner.get_user(user_id)
+            except KeyError as exc:
+                raise _scim_error(404, f"user '{user_id}' not found") from exc
+            # A merged-away duplicate is tombstoned (disabled + a
+            # merged_into_user_id pointer) so it can never authenticate again.
+            # Keep the check and the full replacement PUT under the same lock
+            # used by merge_accounts; otherwise merge can create the tombstone
+            # between these two Admin API calls and SCIM can wipe it again.
+            if provisioner.get_user_attribute(user_id, TOMBSTONE_ATTRIBUTE_KEY):
+                raise _scim_error(
+                    409,
+                    f"user '{user_id}' has been merged into another account "
+                    "and cannot be modified",
+                )
+            account = _to_user_account(resource, user_id=user_id)
+            provisioner.replace_user(user_id, account)
+            replaced = provisioner.get_user(user_id)
+    except UserOperationLockTimeout as exc:
+        raise _scim_error(
+            503,
+            f"user '{user_id}' is being modified; retry the request",
+        ) from exc
+    return _scim_response(_to_scim_resource(replaced))
 
 
 @scim_router.patch("/Users/{user_id}")
diff --git a/services/account_unification/app/service.py b/services/account_unification/app/service.py
index 3354c6f..16c42d8 100644
--- a/services/account_unification/app/service.py
+++ b/services/account_unification/app/service.py
@@ -27,11 +27,13 @@
 from .matching import decide_match, have_matching_verified_email
 from .models import (
     FederatedIdentity,
+    MatchReason,
     MergeConflict,
     MergeRequest,
     MergeResult,
     UserAccount,
 )
+from .user_locks import UserOperationLocks
 
 # Keycloak user attribute stamped on a tombstoned duplicate (two-word snake_case).
 TOMBSTONE_ATTRIBUTE_KEY = "merged_into_user_id"
@@ -46,11 +48,13 @@ def __init__(
         api: AdminApi,
         audit: AuditLogger,
         config: ServiceConfig,
+        user_operation_locks: UserOperationLocks,
     ) -> None:
-        """Create a service around admin API, audit, and config dependencies."""
+        """Create a service around admin, audit, config, and lock dependencies."""
         self._api = api
         self._audit = audit
         self._config = config
+        self._user_operation_locks = user_operation_locks
 
     # -- (a) inspect identities -------------------------------------------
     def get_account(self, user_id: str) -> UserAccount:
@@ -70,6 +74,18 @@ def merge_accounts(self, request: MergeRequest) -> MergeResult:
         if request.survivor_user_id == request.duplicate_user_id:
             raise SameUserError("survivor and duplicate are the same account")
 
+        # The complete merge, including the final tombstone write, shares the
+        # same duplicate-user lock as SCIM replacement. This closes the TOCTOU
+        # window where SCIM could pass its tombstone check, then overwrite a
+        # concurrently-created tombstone with an active user representation.
+        with self._user_operation_locks.hold(
+            request.survivor_user_id,
+            request.duplicate_user_id,
+        ):
+            return self._merge_accounts_locked(request)
+
+    def _merge_accounts_locked(self, request: MergeRequest) -> MergeResult:
+        """Perform a merge while both participating user IDs are serialized."""
         survivor = self._load_user(request.survivor_user_id)
         duplicate = self._load_user(request.duplicate_user_id)
         survivor.federated_identities = self._api.list_federated_identities(
@@ -85,19 +101,28 @@ def merge_accounts(self, request: MergeRequest) -> MergeResult:
         decision = decide_match(
             survivor, duplicate, explicit_link=request.explicit_link
         )
-        # Guard: refuse when the accounts only coincide on an unverified email.
-        # (decide_match already refuses to *call* that a verified match; here we
-        # produce the precise error for the operator + audit trail.)
-        if not decision.matched:
-            same_email = (
-                (survivor.email or "").strip().lower()
-                == (duplicate.email or "").strip().lower()
-                and bool(survivor.email)
+        # Hard rule (enforced here per docs/merge-unification-flow.md and the
+        # MergeRequest.explicit_link contract): never merge when the only tie is
+        # an UNVERIFIED email. An unverified address is attacker-registerable,
+        # so even an operator's explicit_link assertion must not promote a shared
+        # unverified email into a merge — only a strong tie (exact idp subject or
+        # a mutually verified email) justifies it. This guard therefore runs
+        # regardless of decision.matched, catching the explicit_link path too.
+        shares_unverified_email = (
+            bool((survivor.email or "").strip())
+            and (survivor.email or "").strip().lower()
+            == (duplicate.email or "").strip().lower()
+            and not have_matching_verified_email(survivor, duplicate)
+        )
+        strong_tie = decision.reason in (
+            MatchReason.EXACT_IDP_SUBJECT,
+            MatchReason.VERIFIED_EMAIL,
+        )
+        if shares_unverified_email and not strong_tie:
+            raise UnverifiedEmailMergeError(
+                "refusing merge: accounts share only an UNVERIFIED email"
             )
-            if same_email and not have_matching_verified_email(survivor, duplicate):
-                raise UnverifiedEmailMergeError(
-                    "refusing merge: accounts share only an UNVERIFIED email"
-                )
+        if not decision.matched:
             raise NoMatchError(decision.detail or "no matching rule satisfied")
 
         audit_id = self._audit.new_correlation_id()
@@ -181,8 +206,10 @@ def _move_federated_identities(
                     duplicate.user_id, link.identity_provider
                 )
                 self._audit.emit(
-                    audit_id=audit_id, event_type="federated_identity_conflict",
-                    actor=actor, survivor_user_id=survivor.user_id,
+                    audit_id=audit_id,
+                    event_type="federated_identity_conflict",
+                    actor=actor,
+                    survivor_user_id=survivor.user_id,
                     duplicate_user_id=duplicate.user_id,
                     payload={"identifier": identifier, "resolution": "survivor_wins"},
                 )
@@ -193,8 +220,11 @@ def _move_federated_identities(
             )
             moved.append(identifier)
             self._audit.emit(
-                audit_id=audit_id, event_type="federated_identity_moved", actor=actor,
-                survivor_user_id=survivor.user_id, duplicate_user_id=duplicate.user_id,
+                audit_id=audit_id,
+                event_type="federated_identity_moved",
+                actor=actor,
+                survivor_user_id=survivor.user_id,
+                duplicate_user_id=duplicate.user_id,
                 payload={"identifier": identifier},
             )
         return moved
@@ -219,7 +249,9 @@ def _move_role_mappings(
                 # survivor-wins: survivor already has it; drop the duplicate's.
                 self._api.remove_role_mapping(duplicate.user_id, role)
                 self._audit.emit(
-                    audit_id=audit_id, event_type="role_mapping_conflict", actor=actor,
+                    audit_id=audit_id,
+                    event_type="role_mapping_conflict",
+                    actor=actor,
                     survivor_user_id=survivor.user_id,
                     duplicate_user_id=duplicate.user_id,
                     payload={"identifier": identifier, "resolution": "survivor_wins"},
@@ -229,10 +261,16 @@ def _move_role_mappings(
             self._api.remove_role_mapping(duplicate.user_id, role)
             moved.append(identifier)
             self._audit.emit(
-                audit_id=audit_id, event_type="role_mapping_moved", actor=actor,
-                survivor_user_id=survivor.user_id, duplicate_user_id=duplicate.user_id,
-                payload={"identifier": identifier, "role_name": role.role_name,
-                         "client_id": role.client_id},
+                audit_id=audit_id,
+                event_type="role_mapping_moved",
+                actor=actor,
+                survivor_user_id=survivor.user_id,
+                duplicate_user_id=duplicate.user_id,
+                payload={
+                    "identifier": identifier,
+                    "role_name": role.role_name,
+                    "client_id": role.client_id,
+                },
             )
         return moved
 
@@ -252,8 +290,10 @@ def _move_group_memberships(
                 )
                 self._api.remove_group_membership(duplicate.user_id, group)
                 self._audit.emit(
-                    audit_id=audit_id, event_type="group_membership_conflict",
-                    actor=actor, survivor_user_id=survivor.user_id,
+                    audit_id=audit_id,
+                    event_type="group_membership_conflict",
+                    actor=actor,
+                    survivor_user_id=survivor.user_id,
                     duplicate_user_id=duplicate.user_id,
                     payload={"identifier": identifier, "resolution": "survivor_wins"},
                 )
@@ -262,8 +302,11 @@ def _move_group_memberships(
             self._api.remove_group_membership(duplicate.user_id, group)
             moved.append(identifier)
             self._audit.emit(
-                audit_id=audit_id, event_type="group_membership_moved", actor=actor,
-                survivor_user_id=survivor.user_id, duplicate_user_id=duplicate.user_id,
+                audit_id=audit_id,
+                event_type="group_membership_moved",
+                actor=actor,
+                survivor_user_id=survivor.user_id,
+                duplicate_user_id=duplicate.user_id,
                 payload={"identifier": identifier},
             )
         return moved
@@ -277,8 +320,11 @@ def _tombstone(self, duplicate, survivor, audit_id, actor) -> None:
         )
         self._api.deactivate_user(duplicate.user_id)
         self._audit.emit(
-            audit_id=audit_id, event_type="duplicate_tombstoned", actor=actor,
-            survivor_user_id=survivor.user_id, duplicate_user_id=duplicate.user_id,
+            audit_id=audit_id,
+            event_type="duplicate_tombstoned",
+            actor=actor,
+            survivor_user_id=survivor.user_id,
+            duplicate_user_id=duplicate.user_id,
             payload={"attribute_key": TOMBSTONE_ATTRIBUTE_KEY},
         )
 
diff --git a/services/account_unification/app/user_locks.py b/services/account_unification/app/user_locks.py
new file mode 100644
index 0000000..0e940aa
--- /dev/null
+++ b/services/account_unification/app/user_locks.py
@@ -0,0 +1,137 @@
+"""Cross-path serialization for mutations of Keycloak user records.
+
+SCIM replacement and account merge both write complete or partial Keycloak user
+representations. They must share one lock boundary so a merge cannot tombstone a
+duplicate between SCIM's tombstone check and its replacement PUT.
+
+The standalone runtime uses :class:`SqliteUserOperationLocks`, backed by a
+dedicated sidecar SQLite database. ``BEGIN IMMEDIATE`` provides a crash-safe,
+cross-process mutex for every service worker sharing that database file. The
+current implementation intentionally serializes all user mutations rather than
+risking a multi-key deadlock; the public interface remains user-ID keyed so a
+future Postgres advisory-lock implementation can safely increase concurrency.
+"""
+from __future__ import annotations
+
+import sqlite3
+import threading
+from contextlib import contextmanager
+from typing import ContextManager, Iterator, Protocol
+
+
+class UserOperationLockTimeout(RuntimeError):
+    """Raised when a shared user-operation lock cannot be acquired in time."""
+
+
+class UserOperationLocks(Protocol):
+    """Serialize mutations that involve one or more Keycloak user IDs."""
+
+    def hold(self, *user_ids: str) -> ContextManager[None]:
+        """Return a context manager holding the requested user-operation locks."""
+        ...
+
+
+def _normalise_user_ids(user_ids: tuple[str, ...]) -> tuple[str, ...]:
+    """Return unique, non-empty user IDs in deterministic acquisition order."""
+    ordered = tuple(sorted(set(user_ids)))
+    if not ordered or any(not user_id for user_id in ordered):
+        raise ValueError("at least one non-empty user ID is required")
+    return ordered
+
+
+class InMemoryUserOperationLocks:
+    """Process-local keyed lock manager for tests and explicit single-worker use."""
+
+    def __init__(self) -> None:
+        """Create an empty keyed re-entrant lock registry."""
+        self._registry_guard = threading.Lock()
+        self._locks: dict[str, threading.RLock] = {}
+
+    @contextmanager
+    def hold(self, *user_ids: str) -> Iterator[None]:
+        """Hold all requested user locks in stable order to avoid deadlocks."""
+        ordered_ids = _normalise_user_ids(user_ids)
+        with self._registry_guard:
+            locks = [
+                self._locks.setdefault(user_id, threading.RLock())
+                for user_id in ordered_ids
+            ]
+        for lock in locks:
+            lock.acquire()
+        try:
+            yield
+        finally:
+            for lock in reversed(locks):
+                lock.release()
+
+
+class SqliteUserOperationLocks:
+    """Cross-process mutex backed by a dedicated SQLite sidecar database.
+
+    SQLite permits only one writer holding a ``BEGIN IMMEDIATE`` transaction.
+    Every manager instance pointed at the same database file therefore shares a
+    crash-safe mutex: process termination closes the connection and releases the
+    lock automatically. This is deliberately coarser than the user-ID-keyed
+    protocol, but it fully serializes the SCIM and merge critical sections for
+    the supported SQLite deployment without introducing a lease-expiry race.
+    """
+
+    _SCHEMA = """
+    CREATE TABLE IF NOT EXISTS user_operation_lock_state (
+        lock_name TEXT PRIMARY KEY,
+        requested_user_ids TEXT NOT NULL
+    );
+    """
+
+    def __init__(self, database_path: str, *, timeout_seconds: float = 10.0) -> None:
+        """Create a manager using ``database_path`` and an acquisition timeout."""
+        if not database_path:
+            raise ValueError("database_path is required")
+        if timeout_seconds <= 0:
+            raise ValueError("timeout_seconds must be positive")
+        self._database_path = database_path
+        self._timeout_seconds = timeout_seconds
+        self._initialize()
+
+    def _connect(self) -> sqlite3.Connection:
+        """Open one autocommit connection configured with the lock timeout."""
+        return sqlite3.connect(
+            self._database_path,
+            timeout=self._timeout_seconds,
+            isolation_level=None,
+        )
+
+    def _initialize(self) -> None:
+        """Create the sidecar schema before requests begin competing for it."""
+        connection = self._connect()
+        try:
+            connection.execute(self._SCHEMA)
+        finally:
+            connection.close()
+
+    @contextmanager
+    def hold(self, *user_ids: str) -> Iterator[None]:
+        """Hold the shared SQLite mutex for the complete user mutation."""
+        ordered_ids = _normalise_user_ids(user_ids)
+        connection = self._connect()
+        try:
+            try:
+                connection.execute("BEGIN IMMEDIATE")
+            except sqlite3.OperationalError as exc:
+                if "locked" in str(exc).lower():
+                    raise UserOperationLockTimeout(
+                        "timed out waiting for another user mutation to finish"
+                    ) from exc
+                raise
+            connection.execute(
+                "INSERT INTO user_operation_lock_state "
+                "(lock_name, requested_user_ids) VALUES ('global', ?) "
+                "ON CONFLICT(lock_name) DO UPDATE SET "
+                "requested_user_ids = excluded.requested_user_ids",
+                (",".join(ordered_ids),),
+            )
+            yield
+        finally:
+            if connection.in_transaction:
+                connection.rollback()
+            connection.close()
diff --git a/services/account_unification/tests/conftest.py b/services/account_unification/tests/conftest.py
index 3deba76..64d7ecf 100644
--- a/services/account_unification/tests/conftest.py
+++ b/services/account_unification/tests/conftest.py
@@ -11,6 +11,7 @@
 from app.audit import AuditLogger, InMemoryAuditSink  # noqa: E402
 from app.config import ServiceConfig  # noqa: E402
 from app.service import UnificationService  # noqa: E402
+from app.user_locks import InMemoryUserOperationLocks  # noqa: E402
 
 from .mock_keycloak import MockKeycloakAdminApi  # noqa: E402
 
@@ -42,8 +43,16 @@ def config() -> ServiceConfig:
     )
 
 
+@pytest.fixture
+def user_operation_locks() -> InMemoryUserOperationLocks:
+    return InMemoryUserOperationLocks()
+
+
 @pytest.fixture
 def service(
-    api: MockKeycloakAdminApi, audit: AuditLogger, config: ServiceConfig
+    api: MockKeycloakAdminApi,
+    audit: AuditLogger,
+    config: ServiceConfig,
+    user_operation_locks: InMemoryUserOperationLocks,
 ) -> UnificationService:
-    return UnificationService(api, audit, config)
+    return UnificationService(api, audit, config, user_operation_locks)
diff --git a/services/account_unification/tests/mock_keycloak.py b/services/account_unification/tests/mock_keycloak.py
index f844391..26f9c38 100644
--- a/services/account_unification/tests/mock_keycloak.py
+++ b/services/account_unification/tests/mock_keycloak.py
@@ -148,3 +148,7 @@ def set_user_attribute(self, user_id: str, key: str, value: str) -> None:
             self.users[user_id] = self.users[user_id].model_copy(
                 update={"external_id": value}
             )
+
+    def get_user_attribute(self, user_id: str, key: str) -> str | None:
+        self.calls.append(f"get_user_attribute:{user_id}:{key}")
+        return self.attributes.get((user_id, key))
diff --git a/services/account_unification/tests/test_api.py b/services/account_unification/tests/test_api.py
index 51c5693..d9ce523 100644
--- a/services/account_unification/tests/test_api.py
+++ b/services/account_unification/tests/test_api.py
@@ -9,11 +9,16 @@
 
 
 @pytest.fixture
-def client(api, audit, config):
+def client(api, audit, config, user_operation_locks):
     from app.service import UnificationService
 
     app = create_app(wire=False)
-    app.state.unification_service = UnificationService(api, audit, config)
+    app.state.unification_service = UnificationService(
+        api,
+        audit,
+        config,
+        user_operation_locks,
+    )
     app.state.audit_logger = audit
     app.state.keycloak_api = api
     with TestClient(app) as test_client:
diff --git a/services/account_unification/tests/test_audit.py b/services/account_unification/tests/test_audit.py
index a12de6d..6777200 100644
--- a/services/account_unification/tests/test_audit.py
+++ b/services/account_unification/tests/test_audit.py
@@ -7,6 +7,7 @@
 from app.config import ServiceConfig
 from app.models import FederatedIdentity, MergeRequest, RoleMapping
 from app.service import UnificationService
+from app.user_locks import InMemoryUserOperationLocks
 
 from .mock_keycloak import MockKeycloakAdminApi
 
@@ -70,7 +71,12 @@ def test_sqlite_audit_sink_persists(tmp_path):
             keycloak_client_id="svc",
             keycloak_client_secret="secret",
         )
-        service = UnificationService(api, audit, config)
+        service = UnificationService(
+            api,
+            audit,
+            config,
+            InMemoryUserOperationLocks(),
+        )
         result = service.merge_accounts(
             MergeRequest(
                 survivor_user_id="survivor",
diff --git a/services/account_unification/tests/test_healthcheck.py b/services/account_unification/tests/test_healthcheck.py
index e538b16..60e4ff1 100644
--- a/services/account_unification/tests/test_healthcheck.py
+++ b/services/account_unification/tests/test_healthcheck.py
@@ -1,6 +1,8 @@
 """Container healthcheck command behavior."""
 from __future__ import annotations
 
+import urllib.request
+
 from app import healthcheck
 
 
@@ -19,32 +21,72 @@ def read(self) -> bytes:
 
 
 def test_healthcheck_returns_zero_for_ok_status(monkeypatch, capsys):
-    def fake_urlopen(url: str, *, timeout: int) -> _Response:
+    def fake_open(url: str) -> _Response:
         assert url == "http://service/healthz"
-        assert timeout == 5
         return _Response(b'{"status":"ok"}')
 
-    monkeypatch.setattr(healthcheck.urllib.request, "urlopen", fake_urlopen)
+    monkeypatch.setattr(healthcheck, "_open_health_url", fake_open)
 
     assert healthcheck.main("http://service/healthz") == 0
     assert capsys.readouterr().out == "ok\n"
 
 
 def test_healthcheck_returns_one_for_non_ok_status(monkeypatch, capsys):
-    def fake_urlopen(url: str, *, timeout: int) -> _Response:
+    def fake_open(url: str) -> _Response:
         return _Response(b'{"status":"starting"}')
 
-    monkeypatch.setattr(healthcheck.urllib.request, "urlopen", fake_urlopen)
+    monkeypatch.setattr(healthcheck, "_open_health_url", fake_open)
 
     assert healthcheck.main("http://service/healthz") == 1
     assert "not ready" in capsys.readouterr().err
 
 
 def test_healthcheck_returns_one_for_request_error(monkeypatch, capsys):
-    def fake_urlopen(url: str, *, timeout: int) -> _Response:
+    def fake_open(url: str) -> _Response:
         raise OSError("connection refused")
 
-    monkeypatch.setattr(healthcheck.urllib.request, "urlopen", fake_urlopen)
+    monkeypatch.setattr(healthcheck, "_open_health_url", fake_open)
 
     assert healthcheck.main("http://service/healthz") == 1
     assert "healthcheck failed: connection refused" in capsys.readouterr().err
+
+
+def test_healthcheck_rejects_non_http_scheme(monkeypatch, capsys):
+    """A non-HTTP(S) URL is rejected before urllib ever opens it."""
+    def fail_open(*args: object, **kwargs: object) -> _Response:
+        raise AssertionError("the opener must not run for a rejected scheme")
+
+    monkeypatch.setattr(healthcheck, "_open_health_url", fail_open)
+
+    assert healthcheck.main("file:///etc/passwd") == 1
+    assert "unsupported URL scheme 'file'" in capsys.readouterr().err
+
+
+def test_healthcheck_opener_drops_non_http_redirect_target():
+    """A ``http:// -> ftp://`` redirect is dropped, and no ftp/file handler exists."""
+    handler = healthcheck._HttpOnlyRedirectHandler()
+    dropped = handler.redirect_request(
+        urllib.request.Request("http://127.0.0.1:8099/healthz"),
+        None,
+        302,
+        "Found",
+        {},
+        "ftp://127.0.0.1/secret",
+    )
+    assert dropped is None
+    # A same-scheme redirect is still honoured (returns a Request, not None).
+    kept = handler.redirect_request(
+        urllib.request.Request("http://127.0.0.1:8099/healthz"),
+        None,
+        302,
+        "Found",
+        {},
+        "http://127.0.0.1:8099/ready",
+    )
+    assert kept is not None
+    # The opener carries no protocol handler that could open ftp/file targets.
+    opener = healthcheck._build_http_only_opener()
+    assert not any(
+        type(h).__name__ in {"FTPHandler", "FileHandler", "DataHandler"}
+        for h in opener.handlers
+    )
diff --git a/services/account_unification/tests/test_merge.py b/services/account_unification/tests/test_merge.py
index 1acfd12..de00138 100644
--- a/services/account_unification/tests/test_merge.py
+++ b/services/account_unification/tests/test_merge.py
@@ -35,9 +35,13 @@ def test_merge_moves_links_roles_groups_and_tombstones(service, api):
         email="jane@corp.com",
         is_email_verified=True,
         federated_identities=[
-            FederatedIdentity(identity_provider="employer-adfs", external_user_id="jane@corp")
+            FederatedIdentity(
+                identity_provider="employer-adfs", external_user_id="jane@corp"
+            )
+        ],
+        role_mappings=[
+            RoleMapping(role_id="r-s", role_name="viewer", client_id="naruon")
         ],
-        role_mappings=[RoleMapping(role_id="r-s", role_name="viewer", client_id="naruon")],
         group_memberships=[GroupMembership(group_id="g-org", group_path="/org")],
     )
     api.create_test_user(
@@ -45,9 +49,13 @@ def test_merge_moves_links_roles_groups_and_tombstones(service, api):
         email="jane@corp.com",
         is_email_verified=True,
         federated_identities=[
-            FederatedIdentity(identity_provider="google", external_user_id="jane@gmail")
+            FederatedIdentity(
+                identity_provider="google", external_user_id="jane@gmail"
+            )
+        ],
+        role_mappings=[
+            RoleMapping(role_id="r-d", role_name="editor", client_id="clearfolio")
         ],
-        role_mappings=[RoleMapping(role_id="r-d", role_name="editor", client_id="clearfolio")],
         group_memberships=[GroupMembership(group_id="g-proj", group_path="/pg-erd")],
     )
 
@@ -55,93 +63,127 @@ def test_merge_moves_links_roles_groups_and_tombstones(service, api):
 
     assert result.match_reason is MatchReason.VERIFIED_EMAIL
     assert result.duplicate_tombstoned is True
-    # survivor gained the duplicate's external identity...
-    survivor_idps = {f.identity_provider for f in api.list_federated_identities("survivor")}
+    survivor_idps = {
+        identity.identity_provider
+        for identity in api.list_federated_identities("survivor")
+    }
     assert survivor_idps == {"employer-adfs", "google"}
-    # ...its client role...
-    survivor_roles = {r.role_name for r in api.list_role_mappings("survivor")}
+    survivor_roles = {role.role_name for role in api.list_role_mappings("survivor")}
     assert survivor_roles == {"viewer", "editor"}
-    # ...and its group.
-    survivor_groups = {g.group_id for g in api.list_group_memberships("survivor")}
+    survivor_groups = {
+        group.group_id for group in api.list_group_memberships("survivor")
+    }
     assert survivor_groups == {"g-org", "g-proj"}
-    # duplicate is emptied + tombstoned + disabled.
     assert api.list_federated_identities("dup") == []
     assert "dup" in api.deactivated
     assert api.attributes[("dup", TOMBSTONE_ATTRIBUTE_KEY)] == "survivor"
 
 
 def test_merge_by_exact_idp_subject(service, api):
-    shared = FederatedIdentity(identity_provider="employer-adfs", external_user_id="jane@corp")
+    shared = FederatedIdentity(
+        identity_provider="employer-adfs", external_user_id="jane@corp"
+    )
     api.create_test_user("survivor", email="a@x.com", federated_identities=[shared])
     api.create_test_user("dup", email="b@y.com", federated_identities=[shared])
     result = service.merge_accounts(_merge())
     assert result.match_reason is MatchReason.EXACT_IDP_SUBJECT
-    # shared link stays on survivor exactly once (survivor-wins conflict).
-    assert [f.external_user_id for f in api.list_federated_identities("survivor")] == ["jane@corp"]
-    assert any(c.kind == "federated_identity" for c in result.conflicts)
+    assert [
+        identity.external_user_id
+        for identity in api.list_federated_identities("survivor")
+    ] == ["jane@corp"]
+    assert any(conflict.kind == "federated_identity" for conflict in result.conflicts)
 
 
 def test_federated_identity_provider_conflict_is_survivor_wins(service, api):
-    # Same provider alias, different external subject: Keycloak allows only one
-    # link per provider, so survivor-wins keeps the survivor's.
     api.create_test_user(
-        "survivor", email="j@x.com", is_email_verified=True,
+        "survivor",
+        email="j@x.com",
+        is_email_verified=True,
         federated_identities=[
-            FederatedIdentity(identity_provider="employer-adfs", external_user_id="jane@corp")
+            FederatedIdentity(
+                identity_provider="employer-adfs", external_user_id="jane@corp"
+            )
         ],
     )
     api.create_test_user(
-        "dup", email="j@x.com", is_email_verified=True,
+        "dup",
+        email="j@x.com",
+        is_email_verified=True,
         federated_identities=[
-            FederatedIdentity(identity_provider="employer-adfs", external_user_id="jane2@corp")
+            FederatedIdentity(
+                identity_provider="employer-adfs", external_user_id="jane2@corp"
+            )
         ],
     )
     result = service.merge_accounts(_merge())
     survivor_links = api.list_federated_identities("survivor")
-    assert [f.external_user_id for f in survivor_links] == ["jane@corp"]
-    assert any(c.kind == "federated_identity" for c in result.conflicts)
+    assert [identity.external_user_id for identity in survivor_links] == ["jane@corp"]
+    assert any(conflict.kind == "federated_identity" for conflict in result.conflicts)
 
 
 def test_role_conflict_is_survivor_wins(service, api):
     api.create_test_user(
-        "survivor", email="j@x.com", is_email_verified=True,
-        role_mappings=[RoleMapping(role_id="r-s", role_name="admin", client_id="naruon")],
+        "survivor",
+        email="j@x.com",
+        is_email_verified=True,
+        role_mappings=[
+            RoleMapping(role_id="r-s", role_name="admin", client_id="naruon")
+        ],
     )
     api.create_test_user(
-        "dup", email="j@x.com", is_email_verified=True,
-        role_mappings=[RoleMapping(role_id="r-d", role_name="admin", client_id="naruon")],
+        "dup",
+        email="j@x.com",
+        is_email_verified=True,
+        role_mappings=[
+            RoleMapping(role_id="r-d", role_name="admin", client_id="naruon")
+        ],
     )
     result = service.merge_accounts(_merge())
     survivor_roles = api.list_role_mappings("survivor")
-    # only the survivor's admin role on naruon survives.
     assert len(survivor_roles) == 1
     assert survivor_roles[0].role_id == "r-s"
-    assert any(c.kind == "role_mapping" and c.resolution == "survivor_wins" for c in result.conflicts)
+    assert any(
+        conflict.kind == "role_mapping"
+        and conflict.resolution == "survivor_wins"
+        for conflict in result.conflicts
+    )
 
 
 def test_realm_role_moves(service, api):
     api.create_test_user("survivor", email="j@x.com", is_email_verified=True)
     api.create_test_user(
-        "dup", email="j@x.com", is_email_verified=True,
-        role_mappings=[RoleMapping(role_id="r-realm", role_name="ecosystem-user", client_id=None)],
+        "dup",
+        email="j@x.com",
+        is_email_verified=True,
+        role_mappings=[
+            RoleMapping(
+                role_id="r-realm", role_name="ecosystem-user", client_id=None
+            )
+        ],
     )
     result = service.merge_accounts(_merge())
     assert "realm:ecosystem-user" in result.moved_role_mappings
-    assert any(r.client_id is None for r in api.list_role_mappings("survivor"))
+    assert any(
+        role.client_id is None for role in api.list_role_mappings("survivor")
+    )
 
 
 def test_group_conflict_is_survivor_wins(service, api):
     api.create_test_user(
-        "survivor", email="j@x.com", is_email_verified=True,
+        "survivor",
+        email="j@x.com",
+        is_email_verified=True,
         group_memberships=[GroupMembership(group_id="g1", group_path="/owners")],
     )
     api.create_test_user(
-        "dup", email="j@x.com", is_email_verified=True,
+        "dup",
+        email="j@x.com",
+        is_email_verified=True,
         group_memberships=[GroupMembership(group_id="g1", group_path="/owners")],
     )
     result = service.merge_accounts(_merge())
     assert len(api.list_group_memberships("survivor")) == 1
-    assert any(c.kind == "group_membership" for c in result.conflicts)
+    assert any(conflict.kind == "group_membership" for conflict in result.conflicts)
 
 
 def test_refuse_merge_on_unverified_email(service, api):
@@ -149,7 +191,6 @@ def test_refuse_merge_on_unverified_email(service, api):
     api.create_test_user("dup", email="jane@corp.com", is_email_verified=False)
     with pytest.raises(UnverifiedEmailMergeError):
         service.merge_accounts(_merge())
-    # nothing mutated: duplicate not tombstoned.
     assert "dup" not in api.deactivated
 
 
@@ -168,6 +209,22 @@ def test_explicit_link_allows_merge_without_shared_signal(service, api):
     assert result.duplicate_tombstoned
 
 
+def test_explicit_link_cannot_override_shared_unverified_email(service, api):
+    api.create_test_user("survivor", email="jane@corp.com", is_email_verified=False)
+    api.create_test_user("dup", email="jane@corp.com", is_email_verified=False)
+    with pytest.raises(UnverifiedEmailMergeError):
+        service.merge_accounts(_merge(explicit=True))
+    assert "dup" not in api.deactivated
+
+
+def test_explicit_link_cannot_override_case_variant_unverified_email(service, api):
+    api.create_test_user("survivor", email="Jane@Corp.com", is_email_verified=True)
+    api.create_test_user("dup", email="jane@corp.com", is_email_verified=False)
+    with pytest.raises(UnverifiedEmailMergeError):
+        service.merge_accounts(_merge(explicit=True))
+    assert "dup" not in api.deactivated
+
+
 def test_refuse_self_merge(service, api):
     api.create_test_user("same", email="a@x.com", is_email_verified=True)
     with pytest.raises(SameUserError):
diff --git a/services/account_unification/tests/test_scim.py b/services/account_unification/tests/test_scim.py
index bdcef85..9b4d30d 100644
--- a/services/account_unification/tests/test_scim.py
+++ b/services/account_unification/tests/test_scim.py
@@ -1,16 +1,83 @@
 """Inbound SCIM 2.0 provisioning shim -> Keycloak Admin API."""
 from __future__ import annotations
 
+import threading
+from concurrent.futures import ThreadPoolExecutor
+from contextlib import contextmanager
+
 import pytest
 from fastapi.testclient import TestClient
 
 from app.main import create_app
+from app.models import MergeRequest
+from app.service import TOMBSTONE_ATTRIBUTE_KEY, UnificationService
+
+from .mock_keycloak import MockKeycloakAdminApi
+
+
+class _TestUserOperationLocks:
+    """Small keyed lock manager used to prove cross-path serialization."""
+
+    def __init__(self) -> None:
+        self._guard = threading.Lock()
+        self._locks: dict[str, threading.RLock] = {}
+
+    @contextmanager
+    def hold(self, *user_ids: str):
+        """Hold all requested user locks in stable order."""
+        ordered_ids = sorted(set(user_ids))
+        with self._guard:
+            locks = [self._locks.setdefault(user_id, threading.RLock()) for user_id in ordered_ids]
+        for lock in locks:
+            lock.acquire()
+        try:
+            yield
+        finally:
+            for lock in reversed(locks):
+                lock.release()
+
+
+class _BlockingReplaceApi(MockKeycloakAdminApi):
+    """Pause SCIM replacement after its tombstone check to expose the race."""
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.replace_started = threading.Event()
+        self.allow_replace = threading.Event()
+        self.tombstone_started = threading.Event()
+
+    def replace_user(self, user_id, user) -> None:
+        """Wait until the test permits the full Keycloak representation PUT."""
+        self.replace_started.set()
+        if not self.allow_replace.wait(timeout=5):
+            raise AssertionError("test did not release the blocked SCIM replacement")
+        super().replace_user(user_id, user)
+        # Keycloak's full user-representation PUT can remove attributes omitted
+        # from the payload and re-enable the account via SCIM's active=true.
+        self.attributes = {
+            attribute: value
+            for attribute, value in self.attributes.items()
+            if attribute[0] != user_id
+        }
+        self.deactivated.discard(user_id)
+
+    def set_user_attribute(self, user_id: str, key: str, value: str) -> None:
+        """Signal when merge begins writing the duplicate tombstone."""
+        if user_id == "dup" and key == TOMBSTONE_ATTRIBUTE_KEY:
+            self.tombstone_started.set()
+        super().set_user_attribute(user_id, key, value)
+
+
+@pytest.fixture
+def user_operation_locks():
+    return _TestUserOperationLocks()
 
 
 @pytest.fixture
-def client(api):
+def client(api, user_operation_locks):
     app = create_app(wire=False)
     app.state.keycloak_api = api
+    app.state.user_operation_locks = user_operation_locks
     with TestClient(app) as test_client:
         yield test_client
 
@@ -82,6 +149,80 @@ def test_scim_replace_updates_user(client, api):
     assert api.get_user(created["id"]).email == "jane.doe@corp.com"
 
 
+def test_scim_replace_refuses_to_resurrect_a_tombstoned_duplicate(client, api):
+    """A merged-away (tombstoned) duplicate must not be re-enabled via SCIM PUT.
+
+    After a merge the duplicate is disabled and carries a merged_into_user_id
+    pointer. A routine upstream full-sync PUT (``active`` defaults to true) must
+    be refused with 409, leaving the duplicate disabled with its survivor pointer
+    intact -- never silently reactivated.
+    """
+    created = client.post("/scim/v2/Users", json=_scim_user()).json()
+    dup_id = created["id"]
+    # Simulate the post-merge tombstone state (service._tombstone does exactly this).
+    api.set_user_attribute(dup_id, "merged_into_user_id", "survivor-id")
+    api.deactivate_user(dup_id)
+
+    response = client.put(f"/scim/v2/Users/{dup_id}", json=_scim_user())
+
+    assert response.status_code == 409
+    assert dup_id in api.deactivated
+    assert api.get_user(dup_id).state == "disabled"
+    assert api.get_user_attribute(dup_id, "merged_into_user_id") == "survivor-id"
+
+
+def test_scim_replace_is_serialized_with_concurrent_merge(config, audit):
+    """A concurrent merge cannot slip between SCIM's tombstone check and PUT."""
+    api = _BlockingReplaceApi()
+    locks = _TestUserOperationLocks()
+    service = UnificationService(
+        api,
+        audit,
+        config,
+        user_operation_locks=locks,
+    )
+    app = create_app(wire=False)
+    app.state.keycloak_api = api
+    app.state.user_operation_locks = locks
+    app.state.unification_service = service
+    api.create_test_user(
+        "survivor", email="jane@corp.com", is_email_verified=True
+    )
+    api.create_test_user("dup", email="jane@corp.com", is_email_verified=True)
+    merge_invoked = threading.Event()
+
+    def run_merge():
+        merge_invoked.set()
+        return service.merge_accounts(
+            MergeRequest(
+                survivor_user_id="survivor",
+                duplicate_user_id="dup",
+                actor="admin@cwl",
+            )
+        )
+
+    with TestClient(app) as test_client, ThreadPoolExecutor(max_workers=2) as executor:
+        scim_future = executor.submit(
+            test_client.put,
+            "/scim/v2/Users/dup",
+            json=_scim_user(username="dup"),
+        )
+        assert api.replace_started.wait(timeout=2)
+        merge_future = executor.submit(run_merge)
+        assert merge_invoked.wait(timeout=2)
+
+        merge_was_serialized = not api.tombstone_started.wait(timeout=0.25)
+        api.allow_replace.set()
+        response = scim_future.result(timeout=5)
+        merge_result = merge_future.result(timeout=5)
+
+    assert merge_was_serialized
+    assert response.status_code == 200
+    assert merge_result.duplicate_tombstoned is True
+    assert api.get_user("dup").state == "disabled"
+    assert api.get_user_attribute("dup", TOMBSTONE_ATTRIBUTE_KEY) == "survivor"
+
+
 def test_scim_patch_deactivates_user(client, api):
     created = client.post("/scim/v2/Users", json=_scim_user()).json()
     response = client.patch(
diff --git a/services/account_unification/tests/test_user_locks.py b/services/account_unification/tests/test_user_locks.py
new file mode 100644
index 0000000..97f0271
--- /dev/null
+++ b/services/account_unification/tests/test_user_locks.py
@@ -0,0 +1,71 @@
+"""Shared user-operation serialization for merge and SCIM writes."""
+from __future__ import annotations
+
+import threading
+from concurrent.futures import ThreadPoolExecutor
+
+import pytest
+
+from app.user_locks import (
+    InMemoryUserOperationLocks,
+    SqliteUserOperationLocks,
+    UserOperationLockTimeout,
+)
+
+
+def _assert_overlapping_operation_waits(first_manager, second_manager) -> None:
+    first_entered = threading.Event()
+    release_first = threading.Event()
+    second_entered = threading.Event()
+
+    def hold_first() -> None:
+        with first_manager.hold("survivor", "dup"):
+            first_entered.set()
+            assert release_first.wait(timeout=5)
+
+    def hold_second() -> None:
+        assert first_entered.wait(timeout=5)
+        with second_manager.hold("dup"):
+            second_entered.set()
+
+    with ThreadPoolExecutor(max_workers=2) as executor:
+        first_future = executor.submit(hold_first)
+        second_future = executor.submit(hold_second)
+        assert first_entered.wait(timeout=2)
+        was_serialized = not second_entered.wait(timeout=0.25)
+        release_first.set()
+        first_future.result(timeout=5)
+        second_future.result(timeout=5)
+
+    assert was_serialized
+    assert second_entered.is_set()
+
+
+def test_in_memory_locks_serialize_overlapping_user_ids():
+    manager = InMemoryUserOperationLocks()
+    _assert_overlapping_operation_waits(manager, manager)
+
+
+def test_sqlite_locks_serialize_distinct_manager_instances(tmp_path):
+    database_path = str(tmp_path / "user-operation-locks.sqlite3")
+    first_manager = SqliteUserOperationLocks(database_path)
+    second_manager = SqliteUserOperationLocks(database_path)
+    _assert_overlapping_operation_waits(first_manager, second_manager)
+
+
+def test_sqlite_lock_timeout_is_explicit_and_retryable(tmp_path):
+    database_path = str(tmp_path / "user-operation-locks.sqlite3")
+    first_manager = SqliteUserOperationLocks(database_path)
+    impatient_manager = SqliteUserOperationLocks(database_path, timeout_seconds=0.05)
+
+    with first_manager.hold("dup"):
+        with pytest.raises(UserOperationLockTimeout):
+            with impatient_manager.hold("dup"):
+                pytest.fail("contending operation unexpectedly acquired the lock")
+
+
+def test_lock_manager_rejects_empty_user_ids():
+    manager = InMemoryUserOperationLocks()
+    with pytest.raises(ValueError):
+        with manager.hold(""):
+            pytest.fail("empty user ID unexpectedly acquired a lock")