fix(auth): propagate invite role to users.role + recognise FO-1b admin taxonomy (#253)

Multi-layer mismatch was silently 403'ing freshly-bootstrapped founders
on every admin-gated route. Trace:

1. INSERT_USER / SELECT_USER_BY_USERNAME didn't touch the users.role
   column. New rows got the DB default ('user'); reads dropped the
   column entirely, so /auth/me always reported "role": "user" even
   when the row was correct.
2. ensure_user (the claim helper) didn't forward the invite role
   to store.create_user. Even with the storage layer fixed, the
   founder bootstrap path stamped role='user' for every claimer.
3. require_admin matched only the legacy role string 'admin'. The
   FO-1b invite taxonomy uses 'enterprise_admin' / 'l2_admin' /
   'user' — so a correctly-rolled enterprise_admin user still 403'd
   on every admin route.

Fix: thread role through INSERT_USER + SELECT_USER_BY_USERNAME +
store.create_user + ensure_user + claim_invite_route. Centralise the
admin-equivalence check in auth.is_admin_role() and use it in
require_admin plus the five existing `role == "admin"` sites in
crosstalk_routes + activity_routes so the legacy and new
taxonomies stay in sync.

New test: test_claim_persists_invite_role — invites a user with
role=enterprise_admin, claims it, asserts /auth/me reports the
correct role. Prevents the regression that produced this bug.

Author: dwinter3

server/backend/src/cq_server/activity_routes.py

@@ -37,7 +37,7 @@
 from pydantic import BaseModel
 
 from .activity import EVENT_TYPES
-from .auth import get_current_user, scope_filter
+from .auth import get_current_user, is_admin_role, scope_filter
 from .deps import get_store
 from .store._sqlite import SqliteStore
 
@@ -170,7 +170,7 @@ async def list_activity(
     #   - admin: pass through whatever they sent (including None for
     #     "all personas in this Enterprise")
     #   - non-admin: pin to their own username regardless of input
-    if role == "admin":
+    if is_admin_role(role):
         effective_persona = persona
     else:
         effective_persona = username

server/backend/src/cq_server/auth.py

@@ -381,6 +381,27 @@ async def get_current_user(
     return caller.username
 
 
+# Roles that grant admin-equivalent privilege.
+#   - ``admin``           — legacy global admin (pre-FO-1b)
+#   - ``enterprise_admin`` — Enterprise-scoped admin (FO-1b invite taxonomy)
+#   - ``l2_admin``        — single-L2-scoped admin (FO-1b invite taxonomy)
+# All three pass ``require_admin`` until per-Enterprise / per-L2 scoping
+# lands. Without this set the founder bootstrap path produces a
+# role=enterprise_admin user who then 403s on every admin route, since
+# require_admin only matched the legacy string ``admin``.
+_ADMIN_ROLES: frozenset[str] = frozenset({"admin", "enterprise_admin", "l2_admin"})
+
+
+def is_admin_role(role: str | None) -> bool:
+    """Return True when ``role`` grants admin-equivalent privilege.
+
+    Centralises the legacy/new role mapping so callers in
+    crosstalk_routes / activity_routes / etc. don't drift apart from
+    ``require_admin``.
+    """
+    return role in _ADMIN_ROLES if role is not None else False
+
+
 async def require_admin(
     request: Request,
     username: str = Depends(get_current_user),
@@ -391,12 +412,14 @@ async def require_admin(
     Returns the username on success; 401 on missing/invalid JWT (raised
     by the chained ``get_current_user`` dep), 403 when the caller is
     authenticated but not an admin. Admin-ness is global in v1 — there
-    is no per-Enterprise scoping yet (see plan doc, Lane D).
+    is no per-Enterprise scoping yet (see plan doc, Lane D). Accepts
+    the legacy ``admin`` role plus the FO-1b invite roles
+    ``enterprise_admin`` / ``l2_admin`` via ``_ADMIN_ROLES``.
     """
     user = await store.get_user(username)
     if user is None:
         raise HTTPException(status_code=401, detail="User not found")
-    if user.get("role") != "admin":
+    if not is_admin_role(user.get("role")):
         raise HTTPException(status_code=403, detail="Admin role required")
     return username
 

server/backend/src/cq_server/crosstalk_routes.py

@@ -58,7 +58,7 @@
 from pydantic import BaseModel, Field
 
 from .activity_logger import log_activity
-from .auth import get_current_user, scope_filter
+from .auth import get_current_user, is_admin_role, scope_filter
 from .deps import get_store
 from .store._sqlite import SqliteStore
 
@@ -388,7 +388,7 @@ async def reply_on_thread(
         raise HTTPException(status_code=404, detail="Thread not found")
     if thread["status"] != "open":
         raise HTTPException(status_code=409, detail="Thread is closed")
-    if username not in thread["participants"] and role != "admin":
+    if username not in thread["participants"] and not is_admin_role(role):
         raise HTTPException(status_code=403, detail="Not a participant")
 
     # Pick the recipient — for two-party threads, it's the other participant.
@@ -459,7 +459,7 @@ async def list_threads(
         username=username,
         tenant_enterprise=read_ent,
         tenant_group=read_grp,
-        is_admin=(role == "admin"),
+        is_admin=is_admin_role(role),
         limit=limit,
     )
     items = [ThreadSummary(**r) for r in rows]
@@ -480,7 +480,7 @@ async def get_thread(
     thread = await store.get_crosstalk_thread(thread_id=thread_id, tenant_enterprise=read_ent, tenant_group=read_grp)
     if thread is None:
         raise HTTPException(status_code=404, detail="Thread not found")
-    if username not in thread["participants"] and role != "admin":
+    if username not in thread["participants"] and not is_admin_role(role):
         raise HTTPException(status_code=403, detail="Not a participant")
 
     msgs = await store.list_crosstalk_messages(
@@ -510,7 +510,7 @@ async def close_thread(
     thread = await store.get_crosstalk_thread(thread_id=thread_id, tenant_enterprise=read_ent, tenant_group=read_grp)
     if thread is None:
         raise HTTPException(status_code=404, detail="Thread not found")
-    if username not in thread["participants"] and role != "admin":
+    if username not in thread["participants"] and not is_admin_role(role):
         raise HTTPException(status_code=403, detail="Not a participant")
 
     won = await store.close_crosstalk_thread(

server/backend/src/cq_server/invite_routes.py

@@ -393,6 +393,7 @@ async def claim_invite_route(
         username=canonical_username,
         password=request.password,
         email=metadata.email,
+        role=metadata.role,
     )
 
     outcome = claim_invite(store, token=token, claiming_user_id=user_id)

server/backend/src/cq_server/invites.py

@@ -413,13 +413,19 @@ async def ensure_user(
     username: str,
     password: str,
     email: str,
+    role: str = "user",
 ) -> int:
     """Create-or-fetch the user record for an invite-claimer.
 
     If a user with this ``username`` already exists, the existing row's
-    id is returned (no password rotation, no email mutation — that's
-    out of scope for FO-1b). Otherwise we hash the password and insert,
-    then return the new user's id.
+    id is returned (no password rotation, no email mutation, no role
+    rotation — that's out of scope for FO-1b). Otherwise we hash the
+    password, insert with the supplied role, and return the new user's id.
+
+    The ``role`` argument comes from the invite (FO-1b's role taxonomy:
+    ``enterprise_admin`` / ``l2_admin`` / ``user``). Without this wiring
+    the founder bootstrap path produced a ``role='user'`` admin who then
+    got 403'd from every admin-gated endpoint.
 
     The ``email`` argument is currently unused by the create path
     (FO-1a's ``users.email`` column is additive but ``create_user``
@@ -431,7 +437,7 @@ async def ensure_user(
     existing = await store.get_user(username)
     if existing is not None:
         return int(existing["id"])
-    await store.create_user(username, hash_password(password))
+    await store.create_user(username, hash_password(password), role=role)
     fresh = await store.get_user(username)
     if fresh is None:  # pragma: no cover — race that breaks the world
         raise RuntimeError("user creation succeeded but lookup returned None")

server/backend/src/cq_server/store/_queries.py

@@ -225,11 +225,12 @@ def select_list_units(
 # --- users ------------------------------------------------------------------
 
 INSERT_USER: TextClause = text(
-    "INSERT INTO users (username, password_hash, created_at) VALUES (:username, :password_hash, :created_at)"
+    "INSERT INTO users (username, password_hash, role, created_at) "
+    "VALUES (:username, :password_hash, :role, :created_at)"
 )
 
 SELECT_USER_BY_USERNAME: TextClause = text(
-    "SELECT id, username, password_hash, created_at FROM users WHERE username = :username"
+    "SELECT id, username, password_hash, role, created_at FROM users WHERE username = :username"
 )
 
 # --- api_keys ---------------------------------------------------------------

server/backend/src/cq_server/store/_sqlite.py

@@ -191,8 +191,8 @@ async def create_api_key(
             expires_at=expires_at,
         )
 
-    async def create_user(self, username: str, password_hash: str) -> None:
-        await self._run_sync(self._create_user_sync, username, password_hash)
+    async def create_user(self, username: str, password_hash: str, role: str = "user") -> None:
+        await self._run_sync(self._create_user_sync, username, password_hash, role)
 
     # --- WebAuthn / passkey credentials (FO-1a, #191) ---------------------
 
@@ -1580,15 +1580,20 @@ def _create_api_key_sync(
             "revoked_at": None,
         }
 
-    def _create_user_sync(self, username: str, password_hash: str) -> None:
+    def _create_user_sync(self, username: str, password_hash: str, role: str = "user") -> None:
         from ._queries import INSERT_USER
 
         created_at = datetime.now(UTC).isoformat()
         try:
             with self._engine.begin() as conn:
                 conn.execute(
                     INSERT_USER,
-                    {"username": username, "password_hash": password_hash, "created_at": created_at},
+                    {
+                        "username": username,
+                        "password_hash": password_hash,
+                        "role": role,
+                        "created_at": created_at,
+                    },
                 )
         except IntegrityError as e:
             if e.orig is not None:

server/backend/tests/test_invites.py

@@ -449,6 +449,32 @@ def test_get_metadata_then_claim(
         assert "cq_session" in claim_resp.cookies
         assert claim_resp.cookies["cq_session"] == claim_body["token"]
 
+    def test_claim_persists_invite_role(
+        self,
+        client: TestClient,
+        mock_sender: MockEmailSender,
+    ) -> None:
+        # The invite role (enterprise_admin / l2_admin / user) must round-trip
+        # to users.role — otherwise the founder bootstrap path produces a
+        # role=user admin who 403s on every admin-gated route.
+        client.post(
+            "/api/v1/admin/invites",
+            json={"email": "founder@example.com", "role": "enterprise_admin"},
+            headers=_admin_headers(client),
+        )
+        token = mock_sender.sent[0].jwt
+
+        claim_resp = client.post(
+            f"/api/v1/invites/{token}/claim",
+            json={"password": "password123"},
+        )
+        assert claim_resp.status_code == 200, claim_resp.text
+
+        # /auth/me must reflect the enterprise_admin role.
+        me_resp = client.get("/api/v1/auth/me")
+        assert me_resp.status_code == 200, me_resp.text
+        assert me_resp.json()["role"] == "enterprise_admin"
+
     def test_double_claim_returns_409(
         self,
         client: TestClient,