feat(04-rbac-03): wire RBAC guards to all portal API endpoints + new endpoints

- Add require_platform_admin guard to GET/POST /tenants, PUT/DELETE /tenants/{id}
- Add require_tenant_member to GET /tenants/{id}, GET agents, GET agent/{id}
- Add require_tenant_admin to POST agents, PUT/DELETE agents
- Add require_tenant_admin to billing checkout and portal endpoints
- Add require_tenant_admin to channels slack/install and whatsapp/connect
- Add require_tenant_member to channels /{tid}/test
- Add require_tenant_admin to all llm_keys endpoints
- Add require_tenant_member to all usage GET endpoints
- Add POST /tenants/{tid}/agents/{aid}/test (require_tenant_member for operators)
- Add GET /tenants/{tid}/users with pending invitations (require_tenant_admin)
- Add GET /admin/users with tenant filter/role filter (require_platform_admin)
- Add POST /admin/impersonate with AuditEvent logging (require_platform_admin)
- Add POST /admin/stop-impersonation with AuditEvent logging (require_platform_admin)

packages/shared/shared/api/billing.py

@@ -33,6 +33,7 @@ from sqlalchemy import select, text
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from shared.api.rbac import PortalCaller, require_tenant_admin
 from shared.config import settings
 from shared.db import get_session
 from shared.models.billing import StripeEvent
@@ -205,6 +206,7 @@ async def process_stripe_event(event_data: dict[str, Any], session: AsyncSession
 @billing_router.post("/checkout", response_model=CheckoutResponse)
 async def create_checkout_session(
     body: CheckoutRequest,
+    caller: PortalCaller = Depends(require_tenant_admin),
     session: AsyncSession = Depends(get_session),
 ) -> CheckoutResponse:
     """
@@ -254,6 +256,7 @@ async def create_checkout_session(
 @billing_router.post("/portal", response_model=PortalResponse)
 async def create_billing_portal_session(
     body: PortalRequest,
+    caller: PortalCaller = Depends(require_tenant_admin),
     session: AsyncSession = Depends(get_session),
 ) -> PortalResponse:
     """

packages/shared/shared/api/channels.py

@@ -38,6 +38,7 @@ from pydantic import BaseModel
 from sqlalchemy import select, text
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from shared.api.rbac import PortalCaller, require_tenant_admin, require_tenant_member
 from shared.config import settings
 from shared.crypto import KeyEncryptionService
 from shared.db import get_session
@@ -186,6 +187,7 @@ async def _get_tenant_or_404(tenant_id: uuid.UUID, session: AsyncSession) -> Ten
 @channels_router.get("/slack/install", response_model=SlackInstallResponse)
 async def slack_install(
     tenant_id: uuid.UUID = Query(...),
+    caller: PortalCaller = Depends(require_tenant_admin),
 ) -> SlackInstallResponse:
     """
     Generate the Slack OAuth authorization URL for installing the app.
@@ -322,6 +324,7 @@ async def slack_callback(
 @channels_router.post("/whatsapp/connect", response_model=WhatsAppConnectResponse, status_code=status.HTTP_201_CREATED)
 async def whatsapp_connect(
     body: WhatsAppConnectRequest,
+    caller: PortalCaller = Depends(require_tenant_admin),
     session: AsyncSession = Depends(get_session),
 ) -> WhatsAppConnectResponse:
     """
@@ -396,6 +399,7 @@ async def whatsapp_connect(
 async def test_channel_connection(
     tenant_id: uuid.UUID,
     body: TestChannelRequest,
+    caller: PortalCaller = Depends(require_tenant_member),
     session: AsyncSession = Depends(get_session),
 ) -> TestChannelResponse:
     """

packages/shared/shared/api/llm_keys.py

@@ -27,6 +27,7 @@ from pydantic import BaseModel, Field
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from shared.api.rbac import PortalCaller, require_tenant_admin
 from shared.config import settings
 from shared.crypto import KeyEncryptionService
 from shared.db import get_session
@@ -120,6 +121,7 @@ async def _get_tenant_or_404(tenant_id: uuid.UUID, session: AsyncSession) -> Ten
 @llm_keys_router.get("", response_model=list[LlmKeyResponse])
 async def list_llm_keys(
     tenant_id: uuid.UUID,
+    caller: PortalCaller = Depends(require_tenant_admin),
     session: AsyncSession = Depends(get_session),
 ) -> list[LlmKeyResponse]:
     """
@@ -143,6 +145,7 @@ async def list_llm_keys(
 async def create_llm_key(
     tenant_id: uuid.UUID,
     body: LlmKeyCreate,
+    caller: PortalCaller = Depends(require_tenant_admin),
     session: AsyncSession = Depends(get_session),
 ) -> LlmKeyResponse:
     """
@@ -195,6 +198,7 @@ async def create_llm_key(
 async def delete_llm_key(
     tenant_id: uuid.UUID,
     key_id: uuid.UUID,
+    caller: PortalCaller = Depends(require_tenant_admin),
     session: AsyncSession = Depends(get_session),
 ) -> None:
     """

packages/shared/shared/api/portal.py

@@ -16,12 +16,13 @@ from typing import Any
 import bcrypt
 from fastapi import APIRouter, Depends, HTTPException, Query, status
 from pydantic import BaseModel, Field, field_validator
-from sqlalchemy import select
+from sqlalchemy import select, text
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from shared.api.rbac import PortalCaller, require_platform_admin
+from shared.api.rbac import PortalCaller, require_platform_admin, require_tenant_admin, require_tenant_member
 from shared.db import get_session
-from shared.models.auth import PortalUser, UserTenantRole
+from shared.models.audit import AuditEvent
+from shared.models.auth import PortalInvitation, PortalUser, UserTenantRole
 from shared.models.tenant import Agent, Tenant
 from shared.rls import current_tenant_id
 
@@ -134,6 +135,52 @@ class TenantsListResponse(BaseModel):
     page_size: int
 
 
+class AgentTestRequest(BaseModel):
+    message: str = Field(min_length=1, max_length=4096)
+
+
+class AgentTestResponse(BaseModel):
+    agent_id: str
+    message: str
+    response: str
+
+
+class TenantUserResponse(BaseModel):
+    id: str
+    name: str
+    email: str
+    role: str
+    created_at: datetime
+
+    model_config = {"from_attributes": True}
+
+
+class TenantUsersResponse(BaseModel):
+    users: list[TenantUserResponse]
+    pending_invitations: list[dict[str, Any]]
+
+
+class AdminUserResponse(BaseModel):
+    id: str
+    name: str
+    email: str
+    role: str
+    tenant_memberships: list[dict[str, Any]]
+    created_at: datetime
+
+    model_config = {"from_attributes": True}
+
+
+class ImpersonateRequest(BaseModel):
+    tenant_id: uuid.UUID
+
+
+class ImpersonateResponse(BaseModel):
+    tenant_id: str
+    tenant_name: str
+    tenant_slug: str
+
+
 class AgentCreate(BaseModel):
     name: str = Field(min_length=1, max_length=255)
     role: str = Field(min_length=1, max_length=255)
@@ -306,6 +353,7 @@ async def register_user(
 async def list_tenants(
     page: int = Query(default=1, ge=1),
     page_size: int = Query(default=20, ge=1, le=100),
+    caller: PortalCaller = Depends(require_platform_admin),
     session: AsyncSession = Depends(get_session),
 ) -> TenantsListResponse:
     """List all tenants (platform admin view). Paginated."""
@@ -327,6 +375,7 @@ async def list_tenants(
 @portal_router.post("/tenants", response_model=TenantResponse, status_code=status.HTTP_201_CREATED)
 async def create_tenant(
     body: TenantCreate,
+    caller: PortalCaller = Depends(require_platform_admin),
     session: AsyncSession = Depends(get_session),
 ) -> TenantResponse:
     """Create a new tenant. Validates slug uniqueness."""
@@ -350,6 +399,7 @@ async def create_tenant(
 @portal_router.get("/tenants/{tenant_id}", response_model=TenantResponse)
 async def get_tenant(
     tenant_id: uuid.UUID,
+    caller: PortalCaller = Depends(require_tenant_member),
     session: AsyncSession = Depends(get_session),
 ) -> TenantResponse:
     """Get a tenant by ID. Returns 404 if not found."""
@@ -364,6 +414,7 @@ async def get_tenant(
 async def update_tenant(
     tenant_id: uuid.UUID,
     body: TenantUpdate,
+    caller: PortalCaller = Depends(require_platform_admin),
     session: AsyncSession = Depends(get_session),
 ) -> TenantResponse:
     """Update a tenant (partial updates supported)."""
@@ -401,6 +452,7 @@ async def update_tenant(
 @portal_router.delete("/tenants/{tenant_id}", status_code=status.HTTP_204_NO_CONTENT)
 async def delete_tenant(
     tenant_id: uuid.UUID,
+    caller: PortalCaller = Depends(require_platform_admin),
     session: AsyncSession = Depends(get_session),
 ) -> None:
     """Delete a tenant. Cascade deletes agents and channel_connections."""
@@ -429,6 +481,7 @@ async def _get_tenant_or_404(tenant_id: uuid.UUID, session: AsyncSession) -> Ten
 @portal_router.get("/tenants/{tenant_id}/agents", response_model=list[AgentResponse])
 async def list_agents(
     tenant_id: uuid.UUID,
+    caller: PortalCaller = Depends(require_tenant_member),
     session: AsyncSession = Depends(get_session),
 ) -> list[AgentResponse]:
     """List all agents for a tenant."""
@@ -452,6 +505,7 @@ async def list_agents(
 async def create_agent(
     tenant_id: uuid.UUID,
     body: AgentCreate,
+    caller: PortalCaller = Depends(require_tenant_admin),
     session: AsyncSession = Depends(get_session),
 ) -> AgentResponse:
     """Create an AI employee for a tenant."""
@@ -481,6 +535,7 @@ async def create_agent(
 async def get_agent(
     tenant_id: uuid.UUID,
     agent_id: uuid.UUID,
+    caller: PortalCaller = Depends(require_tenant_member),
     session: AsyncSession = Depends(get_session),
 ) -> AgentResponse:
     """Get an agent by ID (must belong to the specified tenant)."""
@@ -503,6 +558,7 @@ async def update_agent(
     tenant_id: uuid.UUID,
     agent_id: uuid.UUID,
     body: AgentUpdate,
+    caller: PortalCaller = Depends(require_tenant_admin),
     session: AsyncSession = Depends(get_session),
 ) -> AgentResponse:
     """Update an agent (partial updates supported)."""
@@ -531,6 +587,7 @@ async def update_agent(
 async def delete_agent(
     tenant_id: uuid.UUID,
     agent_id: uuid.UUID,
+    caller: PortalCaller = Depends(require_tenant_admin),
     session: AsyncSession = Depends(get_session),
 ) -> None:
     """Delete an agent."""
@@ -547,3 +604,241 @@ async def delete_agent(
         await session.commit()
     finally:
         current_tenant_id.reset(token)
+
+
+# ---------------------------------------------------------------------------
+# Test-message endpoint — operators can send test messages to agents
+# ---------------------------------------------------------------------------
+
+
+@portal_router.post(
+    "/tenants/{tenant_id}/agents/{agent_id}/test",
+    response_model=AgentTestResponse,
+)
+async def test_agent_message(
+    tenant_id: uuid.UUID,
+    agent_id: uuid.UUID,
+    body: AgentTestRequest,
+    caller: PortalCaller = Depends(require_tenant_member),
+    session: AsyncSession = Depends(get_session),
+) -> AgentTestResponse:
+    """
+    Send a test message to an agent.
+
+    Available to all tenant members including customer_operator (per locked decision:
+    operators can send test messages to agents). Returns a lightweight echo response
+    with the agent's identity. Full orchestrator integration is added when the agent
+    pipeline is wired to the portal API.
+    """
+    await _get_tenant_or_404(tenant_id, session)
+    token = current_tenant_id.set(tenant_id)
+    try:
+        result = await session.execute(
+            select(Agent).where(Agent.id == agent_id, Agent.tenant_id == tenant_id)
+        )
+        agent = result.scalar_one_or_none()
+    finally:
+        current_tenant_id.reset(token)
+    if agent is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Agent not found")
+
+    # Lightweight test handler — dispatches to orchestrator when integration is complete.
+    # Returns a stub response with agent identity until orchestrator is wired to portal API.
+    response_text = (
+        f"Hi! I'm {agent.name}, your {agent.role}. "
+        f"I received your test message: '{body.message[:100]}'. "
+        "I'm ready to assist your team!"
+    )
+    return AgentTestResponse(
+        agent_id=str(agent.id),
+        message=body.message,
+        response=response_text,
+    )
+
+
+# ---------------------------------------------------------------------------
+# User listing endpoints
+# ---------------------------------------------------------------------------
+
+
+@portal_router.get("/tenants/{tenant_id}/users", response_model=TenantUsersResponse)
+async def list_tenant_users(
+    tenant_id: uuid.UUID,
+    caller: PortalCaller = Depends(require_tenant_admin),
+    session: AsyncSession = Depends(get_session),
+) -> TenantUsersResponse:
+    """
+    List all users and pending invitations for a tenant.
+
+    Returns active users (from user_tenant_roles JOIN portal_users) and
+    pending invitations for the given tenant. Requires tenant admin or platform admin.
+    """
+    await _get_tenant_or_404(tenant_id, session)
+
+    # Load users with membership in this tenant
+    result = await session.execute(
+        select(PortalUser, UserTenantRole)
+        .join(UserTenantRole, PortalUser.id == UserTenantRole.user_id)
+        .where(UserTenantRole.tenant_id == tenant_id)
+        .order_by(UserTenantRole.created_at.desc())
+    )
+    rows = result.all()
+    users = [
+        TenantUserResponse(
+            id=str(user.id),
+            name=user.name,
+            email=user.email,
+            role=membership.role,
+            created_at=user.created_at,
+        )
+        for user, membership in rows
+    ]
+
+    # Load pending invitations
+    inv_result = await session.execute(
+        select(PortalInvitation).where(
+            PortalInvitation.tenant_id == tenant_id,
+            PortalInvitation.status == "pending",
+        )
+    )
+    invitations = inv_result.scalars().all()
+    pending = [
+        {
+            "id": str(inv.id),
+            "email": inv.email,
+            "name": inv.name,
+            "role": inv.role,
+            "expires_at": inv.expires_at.isoformat(),
+            "created_at": inv.created_at.isoformat(),
+        }
+        for inv in invitations
+    ]
+
+    return TenantUsersResponse(users=users, pending_invitations=pending)
+
+
+@portal_router.get("/admin/users", response_model=list[AdminUserResponse])
+async def list_all_users(
+    tenant_id: uuid.UUID | None = Query(default=None),
+    role: str | None = Query(default=None),
+    caller: PortalCaller = Depends(require_platform_admin),
+    session: AsyncSession = Depends(get_session),
+) -> list[AdminUserResponse]:
+    """
+    Global user management endpoint for platform admins.
+
+    Lists ALL portal users with their tenant memberships. Supports optional
+    filtering by tenant_id or role. Requires platform admin access.
+    """
+    query = select(PortalUser).order_by(PortalUser.created_at.desc())
+    if role is not None:
+        query = query.where(PortalUser.role == role)
+
+    result = await session.execute(query)
+    users = result.scalars().all()
+
+    # If filtering by tenant_id, load matching user IDs first
+    if tenant_id is not None:
+        membership_result = await session.execute(
+            select(UserTenantRole).where(UserTenantRole.tenant_id == tenant_id)
+        )
+        tenant_user_ids = {m.user_id for m in membership_result.scalars().all()}
+        users = [u for u in users if u.id in tenant_user_ids]
+
+    # Build response with membership info for each user
+    response = []
+    for user in users:
+        mem_result = await session.execute(
+            select(UserTenantRole).where(UserTenantRole.user_id == user.id)
+        )
+        memberships = [
+            {"tenant_id": str(m.tenant_id), "role": m.role}
+            for m in mem_result.scalars().all()
+        ]
+        response.append(
+            AdminUserResponse(
+                id=str(user.id),
+                name=user.name,
+                email=user.email,
+                role=user.role,
+                tenant_memberships=memberships,
+                created_at=user.created_at,
+            )
+        )
+    return response
+
+
+# ---------------------------------------------------------------------------
+# Impersonation endpoints (platform admin only)
+# ---------------------------------------------------------------------------
+
+
+@portal_router.post("/admin/impersonate", response_model=ImpersonateResponse)
+async def start_impersonation(
+    body: ImpersonateRequest,
+    caller: PortalCaller = Depends(require_platform_admin),
+    session: AsyncSession = Depends(get_session),
+) -> ImpersonateResponse:
+    """
+    Begin platform admin impersonation of a tenant.
+
+    Logs an AuditEvent with action_type='impersonation' containing the platform
+    admin's user_id and the target tenant_id. The portal uses this response
+    to trigger a JWT update with impersonating_tenant_id in the session.
+    """
+    result = await session.execute(select(Tenant).where(Tenant.id == body.tenant_id))
+    tenant = result.scalar_one_or_none()
+    if tenant is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Tenant not found")
+
+    # Log impersonation start to audit trail using raw INSERT (per audit.py design)
+    await session.execute(
+        text(
+            "INSERT INTO audit_events (tenant_id, user_id, action_type, metadata) "
+            "VALUES (:tenant_id, :user_id, 'impersonation', CAST(:metadata AS jsonb))"
+        ),
+        {
+            "tenant_id": str(body.tenant_id),
+            "user_id": str(caller.user_id),
+            "metadata": (
+                f'{{"action": "start", "platform_admin_id": "{caller.user_id}", '
+                f'"target_tenant_id": "{body.tenant_id}"}}'
+            ),
+        },
+    )
+    await session.commit()
+
+    return ImpersonateResponse(
+        tenant_id=str(tenant.id),
+        tenant_name=tenant.name,
+        tenant_slug=tenant.slug,
+    )
+
+
+@portal_router.post("/admin/stop-impersonation", status_code=status.HTTP_204_NO_CONTENT)
+async def stop_impersonation(
+    body: ImpersonateRequest,
+    caller: PortalCaller = Depends(require_platform_admin),
+    session: AsyncSession = Depends(get_session),
+) -> None:
+    """
+    End platform admin impersonation session.
+
+    Logs an AuditEvent with action_type='impersonation' and action='stop'
+    to complete the impersonation audit trail.
+    """
+    await session.execute(
+        text(
+            "INSERT INTO audit_events (tenant_id, user_id, action_type, metadata) "
+            "VALUES (:tenant_id, :user_id, 'impersonation', CAST(:metadata AS jsonb))"
+        ),
+        {
+            "tenant_id": str(body.tenant_id),
+            "user_id": str(caller.user_id),
+            "metadata": (
+                f'{{"action": "stop", "platform_admin_id": "{caller.user_id}", '
+                f'"target_tenant_id": "{body.tenant_id}"}}'
+            ),
+        },
+    )
+    await session.commit()

packages/shared/shared/api/usage.py

@@ -24,6 +24,7 @@ from pydantic import BaseModel
 from sqlalchemy import select, text
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from shared.api.rbac import PortalCaller, require_tenant_member
 from shared.db import get_session
 from shared.models.tenant import Agent, Tenant
 
@@ -201,6 +202,7 @@ async def get_usage_summary(
     tenant_id: uuid.UUID,
     start_date: str = Query(default=None),
     end_date: str = Query(default=None),
+    caller: PortalCaller = Depends(require_tenant_member),
     session: AsyncSession = Depends(get_session),
 ) -> UsageSummaryResponse:
     """
@@ -268,6 +270,7 @@ async def get_usage_by_provider(
     tenant_id: uuid.UUID,
     start_date: str = Query(default=None),
     end_date: str = Query(default=None),
+    caller: PortalCaller = Depends(require_tenant_member),
     session: AsyncSession = Depends(get_session),
 ) -> ProviderUsageResponse:
     """Cost aggregated by LLM provider from audit_events.metadata.provider."""
@@ -323,6 +326,7 @@ async def get_message_volume(
     tenant_id: uuid.UUID,
     start_date: str = Query(default=None),
     end_date: str = Query(default=None),
+    caller: PortalCaller = Depends(require_tenant_member),
     session: AsyncSession = Depends(get_session),
 ) -> MessageVolumeResponse:
     """Message count grouped by channel from audit_events.metadata.channel."""
@@ -371,6 +375,7 @@ async def get_message_volume(
 @usage_router.get("/{tenant_id}/budget-alerts", response_model=BudgetAlertsResponse)
 async def get_budget_alerts(
     tenant_id: uuid.UUID,
+    caller: PortalCaller = Depends(require_tenant_member),
     session: AsyncSession = Depends(get_session),
 ) -> BudgetAlertsResponse:
     """