feat(authz): split the deployment-operator bit from tenant-scoped admin

core/switch_core/authz.py

@@ -8,8 +8,14 @@
   - The subject of every decision is a **user** (`Principal`). Agent-initiated
     requests resolve to the agent's owner (see the protocol service's
     `_resolve_acting_identity`); an agent inherits exactly its owner's
-    permissions.
-  - `User.role == "admin"` is a global bypass.
+    permissions in the tenant the request is bound to — never more than the
+    owner holds there.
+  - `Principal.is_admin` means "may administer the tenant bound to this
+    request" — the deployment operator bypass (`User.role == "admin"`, global,
+    never granted by anything self-service) OR an `owner`/`admin` membership
+    row for this user in the bound tenant (`tenant_members.role`, granted per
+    tenant, e.g. by creating it). `administers_tenant` below computes it; a
+    caller with a database session should use that rather than re-deriving it.
   - Owned entities carry a nullable owner plus independent `read_visibility`
     and `write_visibility` ("public" | "private").
 
@@ -38,12 +44,30 @@ class Principal:
 
     `id` may be ``None`` for an agent with no owner; such a principal owns
     nothing and is limited to public access.
+
+    `is_admin` is tenant-scoped, not global — see `administers_tenant`.
     """
 
     id: str | None
     is_admin: bool
 
 
+# `tenant_members.role` is a checked string, not an enum type (`db/models.py`).
+TENANT_ADMIN_ROLES = ("owner", "admin")
+
+
+def administers_tenant(*, is_operator: bool, tenant_role: str | None) -> bool:
+    """Whether a caller may administer the tenant they hold `tenant_role` in.
+
+    True for a deployment operator (the global bypass), regardless of
+    membership — an operator administers every tenant. Otherwise true only if
+    the membership role itself is `owner` or `admin`; a plain `member`, or no
+    membership row at all, is false. This is the one function that should ever
+    turn a role into an admin bit — see `Principal.is_admin`.
+    """
+    return is_operator or tenant_role in TENANT_ADMIN_ROLES
+
+
 @runtime_checkable
 class Authorizable(Protocol):
     """Structural interface for any entity `can` arbitrates over.

core/switch_core/bridges/agent/protocol/service.py

@@ -2308,13 +2308,9 @@ async def create_moderation_room(
         )
 
         async with self.session_factory() as session:
-            agent = await self.agent_store.get(session, agent_id)
-            if agent is None:
-                raise ValueError(f"Unknown agent: {agent_id}")
-            owner_is_admin = False
-            if agent.owner_id is not None:
-                owner = await session.get(User, agent.owner_id)
-                owner_is_admin = owner is not None and owner.role == "admin"
+            agent, _owner_id, owner_is_admin = await self._resolve_acting_identity(
+                session, agent_id
+            )
             group_id = await self._resolve_group_name(session, group_name)
         if (reference_ids or package_ids) and agent.owner_id is None:
             raise ValueError(
@@ -2463,14 +2459,23 @@ async def _resolve_acting_identity(
 
         Returns ``(agent, owner_id, owner_is_admin)``. Used by moderation
         methods that perform resource-access checks on the agent's behalf.
+
+        ``owner_is_admin`` is the owner's tenant-scoped administrative bit
+        (``UserStore.administers``), not the global operator flag alone: the
+        agent must not gain more than its owner holds in the tenant this
+        request is bound to. `session` is already scoped there by the time an
+        agent-bridge request reaches this method, so the read is the owner's
+        membership in the same tenant the agent itself belongs to.
         """
         agent = await self.agent_store.get(session, agent_id)
         if agent is None:
             raise ValueError(f"Unknown agent: {agent_id}")
         owner_is_admin = False
         if agent.owner_id is not None:
             owner = await session.get(User, agent.owner_id)
-            owner_is_admin = owner is not None and owner.role == "admin"
+            owner_is_admin = owner is not None and await self.user_store.administers(
+                session, owner
+            )
         return agent, agent.owner_id, owner_is_admin
 
     async def _require_room_action(

core/switch_core/db/stores/user_store.py

@@ -6,12 +6,14 @@
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from switch_core.authz import administers_tenant
 from switch_core.db.models import (
     OidcIdentity,
     TenantMember,
     User,
     require_tenant_id,
 )
+from switch_core.tenant_context import current_tenant_id
 
 logger = logging.getLogger(__name__)
 
@@ -297,3 +299,40 @@ async def _link_identity(
     async def get_all(self, session: AsyncSession) -> list[User]:
         result = await session.execute(select(User))
         return list(result.scalars().all())
+
+    async def tenant_role(
+        self, session: AsyncSession, tenant_id: str, user_id: str
+    ) -> str | None:
+        """`user_id`'s membership role in `tenant_id`, or None if not a member.
+
+        `TenantMember` is addressed by its whole primary key, so this is a
+        plain `session.get` rather than a query — same shape as
+        `ensure_membership`'s write.
+        """
+        membership = await session.get(TenantMember, (tenant_id, user_id))
+        return membership.role if membership is not None else None
+
+    async def administers(self, session: AsyncSession, user: User) -> bool:
+        """Whether `user` may administer the tenant bound to `session`'s
+        context — the operator bypass, or an owner/admin membership in it.
+
+        See `authz.administers_tenant`, which this composes with a read of the
+        one membership row that can answer "in *this* tenant".
+
+        Raises:
+            RuntimeError: no tenant is bound. The question has no
+                tenant-independent answer: half of it is a membership row that
+                cannot be read without one. Answering on the operator bit
+                alone would quietly demote a workspace owner to a plain
+                member, and the symptom — a 403 on their own workspace — says
+                nothing about why.
+        """
+        tenant_id = current_tenant_id()
+        if tenant_id is None:
+            raise RuntimeError(
+                "administers requires a bound tenant; whether someone may "
+                "administer a workspace is only answerable about a particular "
+                "one"
+            )
+        role = await self.tenant_role(session, tenant_id, user.id)
+        return administers_tenant(is_operator=user.role == "admin", tenant_role=role)

core/switch_core/gateway/agents.py

@@ -26,7 +26,7 @@
 from switch_core.db.stores.agent_store import AgentStore
 from switch_core.db.stores.room_store import RoomStore
 from switch_core.db.stores.user_store import UserStore
-from switch_core.gateway.auth import get_current_user
+from switch_core.gateway.auth import get_current_user, get_tenant_is_admin
 from switch_core.gateway.dependencies import (
     get_agent_store,
     get_protocol,
@@ -74,12 +74,16 @@ async def delete_agent_by_name(
     agent_store: Annotated[AgentStore, Depends(get_agent_store)],
     protocol: Annotated[ProtocolService, Depends(get_protocol)],
     user: Annotated[User, Depends(get_current_user)],
+    is_admin: Annotated[bool, Depends(get_tenant_is_admin)],
 ) -> dict[str, bool]:
     agent = await agent_store.get_by_name(session, agent_name)
     if agent is None:
         raise HTTPException(status_code=404, detail=f"Agent not found: {agent_name}")
     try:
-        require_manage(Principal(user.id, user.role == "admin"), agent.owner_id)
+        require_manage(
+            Principal(user.id, is_admin),
+            agent.owner_id,
+        )
     except PermissionError:
         raise HTTPException(
             status_code=403,
@@ -101,12 +105,16 @@ async def delete_agent(
     agent_store: Annotated[AgentStore, Depends(get_agent_store)],
     protocol: Annotated[ProtocolService, Depends(get_protocol)],
     user: Annotated[User, Depends(get_current_user)],
+    is_admin: Annotated[bool, Depends(get_tenant_is_admin)],
 ) -> dict[str, bool]:
     agent = await agent_store.get(session, agent_id)
     if agent is None:
         raise HTTPException(status_code=404, detail=f"Agent not found: {agent_id}")
     try:
-        require_manage(Principal(user.id, user.role == "admin"), agent.owner_id)
+        require_manage(
+            Principal(user.id, is_admin),
+            agent.owner_id,
+        )
     except PermissionError:
         raise HTTPException(
             status_code=403,
@@ -192,6 +200,7 @@ async def register_known_subagents(
     protocol: Annotated[ProtocolService, Depends(get_protocol)],
     session: Annotated[AsyncSession, Depends(get_session)],
     agent_store: Annotated[AgentStore, Depends(get_agent_store)],
+    is_admin: Annotated[bool, Depends(get_tenant_is_admin)],
 ) -> RegisterKnownSubagentsResponse:
     """Register many Claude Code subagents under one parent agent (session-authed).
 
@@ -216,7 +225,10 @@ async def register_known_subagents(
             status_code=404, detail=f"Parent agent not found: {req.parent_agent_id}"
         )
     try:
-        require_manage(Principal(user.id, user.role == "admin"), parent.owner_id)
+        require_manage(
+            Principal(user.id, is_admin),
+            parent.owner_id,
+        )
     except PermissionError as exc:
         raise HTTPException(
             status_code=403,
@@ -299,6 +311,7 @@ async def update_agent_options(
     session: Annotated[AsyncSession, Depends(get_session)],
     agent_store: Annotated[AgentStore, Depends(get_agent_store)],
     user: Annotated[User, Depends(get_current_user)],
+    is_admin: Annotated[bool, Depends(get_tenant_is_admin)],
 ) -> AgentSummary:
     """Replace a known-agent's options.
 
@@ -317,7 +330,10 @@ async def update_agent_options(
         raise HTTPException(status_code=404, detail=f"Agent not found: {agent_id}")
 
     try:
-        require_manage(Principal(user.id, user.role == "admin"), agent.owner_id)
+        require_manage(
+            Principal(user.id, is_admin),
+            agent.owner_id,
+        )
     except PermissionError:
         raise HTTPException(
             status_code=403,
@@ -351,6 +367,7 @@ async def update_agent_icon(
     session: Annotated[AsyncSession, Depends(get_session)],
     agent_store: Annotated[AgentStore, Depends(get_agent_store)],
     user: Annotated[User, Depends(get_current_user)],
+    is_admin: Annotated[bool, Depends(get_tenant_is_admin)],
 ) -> AgentSummary:
     """Set, change, or clear an agent's icon (CHOO-2171).
 
@@ -370,7 +387,10 @@ async def update_agent_icon(
         raise HTTPException(status_code=404, detail=f"Agent not found: {agent_id}")
 
     try:
-        require_manage(Principal(user.id, user.role == "admin"), agent.owner_id)
+        require_manage(
+            Principal(user.id, is_admin),
+            agent.owner_id,
+        )
     except PermissionError:
         raise HTTPException(
             status_code=403,
@@ -404,6 +424,7 @@ async def update_agent_display_name(
     session: Annotated[AsyncSession, Depends(get_session)],
     agent_store: Annotated[AgentStore, Depends(get_agent_store)],
     user: Annotated[User, Depends(get_current_user)],
+    is_admin: Annotated[bool, Depends(get_tenant_is_admin)],
 ) -> AgentSummary:
     """Set, change, or clear an agent's human display name.
 
@@ -420,7 +441,10 @@ async def update_agent_display_name(
         raise HTTPException(status_code=404, detail=f"Agent not found: {agent_id}")
 
     try:
-        require_manage(Principal(user.id, user.role == "admin"), agent.owner_id)
+        require_manage(
+            Principal(user.id, is_admin),
+            agent.owner_id,
+        )
     except PermissionError:
         raise HTTPException(
             status_code=403,
@@ -498,6 +522,7 @@ async def update_addressing_policy(
     user_store: Annotated[UserStore, Depends(get_user_store)],
     protocol: Annotated[ProtocolService, Depends(get_protocol)],
     user: Annotated[User, Depends(get_current_user)],
+    is_admin: Annotated[bool, Depends(get_tenant_is_admin)],
 ) -> AgentDetail:
     """Set or clear an agent's scoped addressing policy (CHOO-1585).
 
@@ -509,7 +534,10 @@ async def update_addressing_policy(
         raise HTTPException(status_code=404, detail=f"Agent not found: {agent_id}")
 
     try:
-        require_manage(Principal(user.id, user.role == "admin"), agent.owner_id)
+        require_manage(
+            Principal(user.id, is_admin),
+            agent.owner_id,
+        )
     except PermissionError:
         raise HTTPException(
             status_code=403,

core/switch_core/gateway/auth.py

@@ -234,17 +234,60 @@ async def get_current_user(
 async def require_admin(
     user: Annotated[User, Depends(get_current_user)],
 ) -> User:
+    """Raise 403 unless `user` is a deployment operator.
+
+    ``User.role == "admin"`` is deliberately global and deliberately not
+    self-service: it names the person who runs the server, not a role any
+    tenant can grant. Gate deployment-wide actions on this — creating or
+    listing every user in the deployment — never a single tenant's resources.
+    Those use ``require_tenant_admin``.
+    """
     if user.role != "admin":
         raise HTTPException(status_code=403, detail="Admin access required")
     return user
 
 
+async def get_tenant_is_admin(
+    session: Annotated[AsyncSession, Depends(get_session)],
+    user: Annotated[User, Depends(get_current_user)],
+    user_store: Annotated[UserStore, Depends(get_user_store)],
+) -> bool:
+    """Whether `user` may administer the tenant this request is bound to.
+
+    This is the boolean every ``Principal.is_admin`` in a gateway request
+    should be built from — see ``UserStore.administers`` for what it actually
+    checks.
+
+    Also callable directly (not just as a FastAPI dependency) from any
+    handler or service function that already holds a bound `session`, the
+    caller's `User`, and a `UserStore`.
+    """
+    return await user_store.administers(session, user)
+
+
+async def require_tenant_admin(
+    user: Annotated[User, Depends(get_current_user)],
+    is_admin: Annotated[bool, Depends(get_tenant_is_admin)],
+) -> User:
+    """Raise 403 unless `user` may administer the tenant this request is bound to.
+
+    Unlike ``require_admin``, this is granted by ``tenant_members.role`` as
+    well as the operator bit — use it for routes that manage one tenant's
+    resources (a collaboration bridge, a room) rather than the deployment
+    itself.
+    """
+    if not is_admin:
+        raise HTTPException(status_code=403, detail="Tenant admin access required")
+    return user
+
+
 async def require_room_access(
     session: AsyncSession,
     room_store: RoomStore,
     room_id: str,
     user: User,
     action: Action,
+    is_admin: bool,
 ) -> Room:
     """Load a room (404 if missing) and authorize `action` for `user`,
     raising HTTP 403 if denied.
@@ -253,12 +296,21 @@ async def require_room_access(
     id (attaching references, linking rooms, …) so they cannot operate on a
     room the caller lacks access to. Mirrors the protocol layer's
     ``_require_room_action``.
+
+    `is_admin` comes from ``get_tenant_is_admin``; it is a parameter rather
+    than a read of its own so that a route cannot end up authorizing against
+    a different admin bit than the one its own dependencies resolved.
+
+    This is the widest of the tenant-admin gates: `authz.can` short-circuits
+    on the admin bit, so a workspace owner or admin passes it for every room
+    in their workspace, private ones included. See the phase 2 design note on
+    what that grant covers.
     """
     room = await room_store.get(session, room_id)
     if room is None:
         raise HTTPException(status_code=404, detail="Room not found")
     try:
-        require(Principal(user.id, user.role == "admin"), action, room)
+        require(Principal(user.id, is_admin), action, room)
     except PermissionError as e:
         raise HTTPException(status_code=403, detail=str(e)) from e
     return room