fix(oidc): serialize create_user() across workers

create_user() (password-user creation) now takes _interprocess_auth_lock
with reload-before-save, preventing a concurrent set_oidc_user_admin()
from losing a newly-created password user.

Introduces _create_user_locked() internal helper so setup() (which
already holds the inter-process lock) can create users without a
nested fcntl.flock deadlock.

Also changed _auth_intraprocess_lock to threading.RLock to support
safe nesting patterns across mutation methods.

Regression: test_create_user_survives_concurrent_set_oidc_user_admin
(109 total, 0 failures).

core/auth.py

@@ -85,7 +85,11 @@ RESERVED_USERNAMES = frozenset({INTERNAL_TOOL_USER, "api", "demo", "system"})
 # *other* processes — two threads in the same process calling flock(LOCK_EX)
 # on the same file both succeed immediately.  This lock closes that gap so the
 # critical section is serialised across both threads and workers.
-_auth_intraprocess_lock = threading.Lock()
+#
+# RLock (reentrant) so a mutation method that acquires the inter-process lock
+# can safely call another mutation method that also acquires it (e.g. setup()
+# calling create_user()).
+_auth_intraprocess_lock = threading.RLock()
 
 
 def normalize_known_username(users: Dict[str, Any], username: str | None) -> Optional[str]:
@@ -318,28 +322,41 @@ class AuthManager:
             self._load()
             if self.is_configured:
                 return False
-            return self.create_user(username, password, is_admin=True)
+            # _create_user_locked assumes both locks are already held,
+            # avoiding a nested fcntl.flock deadlock (flock is not
+            # reentrant across different file descriptors).
+            return self._create_user_locked(username, password, is_admin=True)
 
     def create_user(self, username: str, password: str, is_admin: bool = False) -> bool:
-        """Create a new user account."""
+        """Create a new user account.
+
+        Serialised across workers via the shared inter-process lock so a
+        concurrent OIDC admin sync cannot lose a newly-created user.
+        """
         username = username.strip().lower()
         if not username:
             return False
         if username in RESERVED_USERNAMES:
             logger.warning("Refused to create reserved username '%s'", username)
             return False
-        with self._config_lock:
-            if username in self.users:
-                return False
-            if "users" not in self._config:
-                self._config["users"] = {}
-            self._config["users"][username] = {
-                "password_hash": _hash_password(password),
-                "created": time.time(),
-                "is_admin": is_admin,
-                "privileges": dict(ADMIN_PRIVILEGES if is_admin else DEFAULT_PRIVILEGES),
-            }
-            self._save()
+        with self._interprocess_auth_lock(), self._config_lock:
+            self._load()
+            return self._create_user_locked(username, password, is_admin)
+
+    def _create_user_locked(self, username: str, password: str, is_admin: bool) -> bool:
+        """Internal helper — caller must hold _interprocess_auth_lock
+        and _config_lock.  Does not reload (caller did that)."""
+        if username in self._config.get("users", {}):
+            return False
+        if "users" not in self._config:
+            self._config["users"] = {}
+        self._config["users"][username] = {
+            "password_hash": _hash_password(password),
+            "created": time.time(),
+            "is_admin": is_admin,
+            "privileges": dict(ADMIN_PRIVILEGES if is_admin else DEFAULT_PRIVILEGES),
+        }
+        self._save()
         logger.info(f"Created user '{username}' (admin={is_admin})")
         return True
 

tests/test_oidc_auth.py

@@ -701,3 +701,58 @@ class TestInterprocessFirstAdminSerialisation:
         assert f1.decrypt(token) == b"hello"
         token = f1.encrypt(b"world")
         assert f0.decrypt(token) == b"world"
+
+    def test_create_user_survives_concurrent_set_oidc_user_admin(self, tmp_path):
+        """create_user() (password user) must not be lost when a concurrent
+        set_oidc_user_admin() executes on another manager.  Both operations
+        now take the inter-process lock, so the second operation must reload
+        and see the first operation's result."""
+        from core.auth import AuthManager
+        import threading
+
+        auth_path = str(tmp_path / "auth.json")
+
+        # Manager A: create an OIDC user (alice) then make her admin so the
+        # no-op short-circuit in set_oidc_user_admin doesn't trigger.
+        mgr_a = AuthManager(auth_path)
+        alice = mgr_a.create_user_oidc("alice", "sub-a", "https://idp.example.com")
+        assert alice == "alice"
+        mgr_a._config["users"]["alice"]["is_admin"] = True
+        mgr_a._save()
+
+        # Manager B: a separate instance with the same auth file.
+        mgr_b = AuthManager(auth_path)
+
+        results = {}
+        barrier = threading.Barrier(2, timeout=5)
+
+        def do_create_user():
+            barrier.wait()
+            ok = mgr_a.create_user("bob", "password123")
+            results["create"] = ok
+
+        def do_sync_admin():
+            barrier.wait()
+            ok = mgr_b.set_oidc_user_admin("alice", False)
+            results["sync"] = ok
+
+        t_create = threading.Thread(target=do_create_user)
+        t_sync = threading.Thread(target=do_sync_admin)
+        t_create.start()
+        t_sync.start()
+        t_create.join()
+        t_sync.join()
+
+        assert results.get("create") is True, "create_user must succeed"
+        assert results.get("sync") is True, "set_oidc_user_admin must succeed"
+
+        # Reload both — bob and alice must both exist.
+        mgr_a._load()
+        mgr_b._load()
+        users_a = mgr_a._config.get("users", {})
+        users_b = mgr_b._config.get("users", {})
+        assert "bob" in users_a, f"bob must survive concurrent admin sync; users: {list(users_a)}"
+        assert "bob" in users_b, f"bob must survive concurrent admin sync; users: {list(users_b)}"
+        assert "alice" in users_a
+        assert "alice" in users_b
+        assert not users_a["alice"].get("is_admin"), "alice must be demoted"