fix(cloud): isolate support adapter sessions

.github/workflows/deploy-prod.yml

@@ -1,78 +0,0 @@
-name: Build and deploy production
-
-on:
-  push:
-    branches: [deploy/prod]
-  workflow_dispatch:
-
-permissions:
-  contents: read
-
-concurrency:
-  group: langbot-production
-  cancel-in-progress: false
-
-env:
-  CORE_IMAGE: ${{ secrets.DOCKER_USERNAME }}/langbot
-  CLOUD_IMAGE: ${{ secrets.DOCKER_USERNAME }}/langbot-cloud-core
-  SPACE_REF: e1b261dac45e886efc667b1096a4ec493c6a6111
-
-jobs:
-  build-and-deploy:
-    runs-on: ubuntu-latest
-    environment: production
-    steps:
-      - uses: actions/checkout@v4
-      - uses: docker/setup-buildx-action@v3
-      - uses: docker/login-action@v3
-        with:
-          username: ${{ secrets.DOCKER_USERNAME }}
-          password: ${{ secrets.DOCKER_PASSWORD }}
-      - name: Build exact Core image
-        uses: docker/build-push-action@v6
-        with:
-          context: .
-          push: true
-          tags: |
-            ${{ env.CORE_IMAGE }}:prod-${{ github.sha }}
-            ${{ env.CORE_IMAGE }}:deploy-prod
-          cache-from: type=gha,scope=core-prod
-          cache-to: type=gha,mode=max,scope=core-prod
-      - name: Checkout production Cloud adapter
-        uses: actions/checkout@v4
-        with:
-          repository: langbot-app/langbot-space
-          ref: ${{ env.SPACE_REF }}
-          token: ${{ secrets.CLA_PAT }}
-          path: .space
-      - name: Build exact Cloud Core image
-        uses: docker/build-push-action@v6
-        with:
-          context: .space
-          file: .space/Dockerfile.cloud
-          push: true
-          build-args: LANGBOT_CORE_IMAGE=${{ env.CORE_IMAGE }}:prod-${{ github.sha }}
-          tags: |
-            ${{ env.CLOUD_IMAGE }}:prod-${{ github.sha }}
-            ${{ env.CLOUD_IMAGE }}:deploy-prod
-          cache-from: type=gha,scope=cloud-core-prod
-          cache-to: type=gha,mode=max,scope=cloud-core-prod
-      - name: Configure SSH
-        env:
-          SSH_KEY: ${{ secrets.JP09_SSH_KEY }}
-          KNOWN_HOSTS: ${{ secrets.JP09_KNOWN_HOSTS }}
-        run: |
-          install -m 700 -d ~/.ssh
-          install -m 600 /dev/null ~/.ssh/id_ed25519
-          printf '%s\n' "$SSH_KEY" > ~/.ssh/id_ed25519
-          printf '%s\n' "$KNOWN_HOSTS" > ~/.ssh/known_hosts
-      - name: Upload release manifest and deploy
-        env:
-          HOST: ${{ secrets.JP09_HOST }}
-          USER: ${{ secrets.JP09_USER }}
-          PORT: ${{ secrets.JP09_PORT }}
-        run: |
-          remote="$USER@$HOST"
-          ssh -p "$PORT" "$remote" 'install -d -m 700 /opt/langbot-cloud-prod'
-          scp -P "$PORT" deploy/prod/docker-compose.yml deploy/prod/deploy.sh "$remote:/opt/langbot-cloud-prod/"
-          ssh -p "$PORT" "$remote" "chmod 700 /opt/langbot-cloud-prod/deploy.sh && /opt/langbot-cloud-prod/deploy.sh prod-${GITHUB_SHA}"

deploy/prod/deploy.sh

@@ -1,92 +0,0 @@
-#!/usr/bin/env bash
-set -Eeuo pipefail
-
-cd /opt/langbot-cloud-prod
-TAG=${1:?usage: deploy.sh prod-<40-char-sha>}
-[[ "$TAG" =~ ^prod-[0-9a-f]{40}$ ]] || { echo 'invalid immutable image tag' >&2; exit 2; }
-[[ -s .env ]] || { echo '/opt/langbot-cloud-prod/.env is missing' >&2; exit 3; }
-
-rendered_compose=$(docker compose config)
-grep -Fq 'LANGBOT_SPACE_CONTROL_PLANE_URL: https://space.langbot.app' <<<"$rendered_compose" || {
-  echo 'Cloud control-plane URL must be https://space.langbot.app' >&2
-  exit 4
-}
-grep -Fq 'SPACE__URL: https://space.langbot.app' <<<"$rendered_compose" || {
-  echo 'Cloud user-facing Space URL must be https://space.langbot.app' >&2
-  exit 5
-}
-
-update_env() {
-  local key=$1 value=$2
-  python3 - "$key" "$value" <<'PY'
-from pathlib import Path
-import os
-import sys
-
-path = Path('.env')
-key, value = sys.argv[1:]
-lines = path.read_text().splitlines()
-updated = False
-for index, line in enumerate(lines):
-    if line.startswith(f'{key}='):
-        lines[index] = f'{key}={value}'
-        updated = True
-        break
-if not updated:
-    lines.append(f'{key}={value}')
-temporary = Path('.env.tmp')
-temporary.write_text('\n'.join(lines) + '\n')
-os.chmod(temporary, 0o600)
-temporary.replace(path)
-PY
-}
-update_env LANGBOT_IMAGE_TAG "$TAG"
-set -a
-. ./.env
-set +a
-
-for attempt in 1 2 3 4 5; do
-  if docker compose pull postgres redis migrate plugin-runtime core; then
-    break
-  fi
-  if [ "$attempt" -eq 5 ]; then
-    echo "docker compose pull failed after $attempt attempts" >&2
-    exit 1
-  fi
-  delay=$((attempt * 10))
-  echo "docker compose pull failed (attempt $attempt/5); retrying in ${delay}s" >&2
-  sleep "$delay"
-done
-docker compose up -d postgres redis
-for _ in $(seq 1 60); do
-  if docker compose exec -T postgres pg_isready -U langbot_operator -d langbot >/dev/null 2>&1; then break; fi
-  sleep 2
-done
-docker compose exec -T postgres pg_isready -U langbot_operator -d langbot >/dev/null
-
-docker compose exec -T postgres psql -v ON_ERROR_STOP=1 -U langbot_operator -d langbot \
-  -v runtime_password="$POSTGRES_RUNTIME_PASSWORD" <<'SQL'
-SELECT format('CREATE ROLE langbot_runtime LOGIN PASSWORD %L', :'runtime_password')
-WHERE NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'langbot_runtime')\gexec
-ALTER ROLE langbot_runtime PASSWORD :'runtime_password';
-GRANT CONNECT ON DATABASE langbot TO langbot_runtime;
-REVOKE CREATE ON SCHEMA public FROM PUBLIC, langbot_runtime;
-REVOKE ALL PRIVILEGES ON ALL TABLES IN SCHEMA public FROM langbot_runtime;
-REVOKE ALL PRIVILEGES ON ALL SEQUENCES IN SCHEMA public FROM langbot_runtime;
-ALTER DEFAULT PRIVILEGES FOR ROLE langbot_operator IN SCHEMA public REVOKE ALL ON TABLES FROM langbot_runtime;
-ALTER DEFAULT PRIVILEGES FOR ROLE langbot_operator IN SCHEMA public REVOKE ALL ON SEQUENCES FROM langbot_runtime;
-GRANT USAGE ON SCHEMA public TO langbot_runtime;
-SQL
-
-docker compose --profile tools run --rm migrate
-
-docker compose up -d --remove-orphans plugin-runtime core
-for _ in $(seq 1 90); do
-  if docker compose exec -T core python -c 'import urllib.request; urllib.request.urlopen("http://127.0.0.1:5300/healthz", timeout=3)' >/dev/null 2>&1; then
-    docker compose ps
-    exit 0
-  fi
-  sleep 2
-done
-docker compose logs --tail=200 core plugin-runtime >&2
-exit 1

deploy/prod/docker-compose.yml

@@ -1,161 +0,0 @@
-services:
-  postgres:
-    image: pgvector/pgvector:pg17
-    container_name: langbot-cloud-postgres
-    restart: unless-stopped
-    environment:
-      POSTGRES_DB: langbot
-      POSTGRES_USER: langbot_operator
-      POSTGRES_PASSWORD: ${POSTGRES_OPERATOR_PASSWORD}
-    volumes:
-      - postgres-data:/var/lib/postgresql/data
-    healthcheck:
-      test: [CMD-SHELL, "pg_isready -U langbot_operator -d langbot"]
-      interval: 5s
-      timeout: 5s
-      retries: 30
-    networks: [internal]
-
-  redis:
-    image: redis:7.4-alpine
-    container_name: langbot-cloud-redis
-    restart: unless-stopped
-    command: [redis-server, --appendonly, "yes", --requirepass, "${REDIS_PASSWORD}"]
-    volumes:
-      - redis-data:/data
-    healthcheck:
-      test: [CMD-SHELL, "redis-cli -a \"$${REDIS_PASSWORD}\" ping | grep PONG"]
-      interval: 5s
-      timeout: 5s
-      retries: 20
-    environment:
-      REDIS_PASSWORD: ${REDIS_PASSWORD}
-    networks: [internal]
-
-  migrate:
-    image: rockchin/langbot-cloud-core:${LANGBOT_IMAGE_TAG}
-    profiles: [tools]
-    command: [uv, run, langbot, migrate, --cloud]
-    environment: &core-env
-      TZ: Asia/Shanghai
-      SYSTEM__INSTANCE_ID: ${CLOUD_V2_INSTANCE_UUID}
-      SYSTEM__EDITION: cloud
-      SYSTEM__RECOVERY_KEY: ${SYSTEM_RECOVERY_KEY}
-      SYSTEM__JWT__SECRET: ${JWT_SECRET}
-      SYSTEM__LIMITATION__MAX_BOTS: "2"
-      SYSTEM__LIMITATION__MAX_PIPELINES: "3"
-      SYSTEM__LIMITATION__MAX_EXTENSIONS: "3"
-      SYSTEM__LIMITATION__MAX_KNOWLEDGE_BASES: "2"
-      API__WEBHOOK_PREFIX: https://cloud.langbot.app
-      API__WEBUI_URL: https://cloud.langbot.app
-      WORKSPACE__INVITATIONS__PUBLIC_WEB_URL: https://cloud.langbot.app
-      DATABASE__USE: postgresql
-      DATABASE__POSTGRESQL__URL: postgresql+asyncpg://langbot_runtime:${POSTGRES_RUNTIME_PASSWORD}@postgres:5432/langbot
-      DATABASE__CLOUD_MIGRATION__OPERATOR_DSN_ENV: LANGBOT_CLOUD_MIGRATION_DSN
-      LANGBOT_CLOUD_MIGRATION_DSN: postgresql://langbot_operator:${POSTGRES_OPERATOR_PASSWORD}@postgres:5432/langbot
-      VDB__USE: pgvector
-      VDB__PGVECTOR__USE_BUSINESS_DATABASE: "true"
-      VDB__PGVECTOR__ALLOWED_DIMENSIONS: "384,512,768,1024,1536"
-      PLUGIN__ENABLE: "true"
-      PLUGIN__RUNTIME_WS_URL: ws://plugin-runtime:5400/control/ws
-      PLUGIN__DISPLAY_PLUGIN_DEBUG_URL: wss://cloud.langbot.app/plugin/debug/ws
-      PLUGIN__WORKER__MAX_CPUS: "0.25"
-      PLUGIN__WORKER__MAX_MEMORY_MB: "256"
-      PLUGIN__WORKER__MAX_PIDS: "128"
-      PLUGIN__WORKER__MAX_WORKERS: "16"
-      PLUGIN__WORKER__MAX_TOTAL_CPUS: "4.0"
-      PLUGIN__WORKER__MAX_TOTAL_MEMORY_MB: "4096"
-      PLUGIN__WORKER__REQUIRE_HARD_LIMITS: "true"
-      LANGBOT_PLUGIN_RUNTIME_CONTROL_TOKEN: ${PLUGIN_RUNTIME_CONTROL_TOKEN}
-      # Cloud v2 currently grants no managed Box capability. Keep the shared
-      # runtime deployed but disable Core integration until a hard-quota-capable
-      # backend can satisfy the fail-closed Cloud readiness contract.
-      BOX__ENABLED: "false"
-      BOX__BACKEND: nsjail
-      BOX__RUNTIME__ENDPOINT: ws://box:5410
-      BOX__ADMISSION__REQUIRED: "true"
-      BOX__ADMISSION__LOGICAL_SESSION_ID: global
-      BOX__ADMISSION__REQUIRED_BACKEND: nsjail
-      BOX__ADMISSION__MAX_SESSIONS: "1"
-      BOX__ADMISSION__MAX_MANAGED_PROCESSES: "0"
-      BOX__ADMISSION__CPUS: "0.25"
-      BOX__ADMISSION__MEMORY_MB: "256"
-      BOX__ADMISSION__WORKSPACE_QUOTA_MB: "256"
-      BOX__LOCAL__HOST_ROOT: /app/data/box
-      BOX__LOCAL__DEFAULT_WORKSPACE: /app/data/box
-      BOX__LOCAL__ALLOWED_MOUNT_ROOTS: /app/data/box
-      LANGBOT_BOX_CONTROL_TOKEN: ${BOX_CONTROL_TOKEN}
-      MCP__STDIO__ENABLED: "false"
-      LANGBOT_SPACE_CONTROL_PLANE_URL: https://space.langbot.app
-      LANGBOT_SPACE_CONTROL_PLANE_TOKEN: ${CLOUD_V2_CONTROL_PLANE_TOKEN}
-      LANGBOT_SPACE_CONTROL_PLANE_PUBLIC_KEY: ${CLOUD_V2_MANIFEST_PUBLIC_KEY}
-      LANGBOT_SPACE_CONTROL_PLANE_KEY_ID: ${CLOUD_V2_MANIFEST_KEY_ID}
-      SPACE__URL: https://space.langbot.app
-    depends_on:
-      postgres: {condition: service_healthy}
-    networks: [internal]
-
-  plugin-runtime:
-    image: rockchin/langbot:${LANGBOT_IMAGE_TAG}
-    container_name: langbot-cloud-plugin-runtime
-    restart: unless-stopped
-    command: [uv, run, python, -m, langbot_plugin.cli.__init__, rt]
-    environment:
-      LANGBOT_PLUGIN_RUNTIME_CONTROL_TOKEN: ${PLUGIN_RUNTIME_CONTROL_TOKEN}
-    volumes:
-      - plugin-data:/app/data
-      - /sys/fs/cgroup:/sys/fs/cgroup:rw
-    cgroup: host
-    privileged: true
-    expose: ["5400"]
-    networks: [internal]
-
-  box:
-    image: rockchin/langbot:${LANGBOT_IMAGE_TAG}
-    container_name: langbot-cloud-box
-    restart: unless-stopped
-    command: [uv, run, lbp, box, --host, 0.0.0.0, --ws-control-port, "5410"]
-    environment:
-      LANGBOT_BOX_CONTROL_TOKEN: ${BOX_CONTROL_TOKEN}
-      LANGBOT_BOX_ROOT: /app/data/box
-    volumes:
-      - box-data:/app/data/box
-      - /sys/fs/cgroup:/sys/fs/cgroup:rw
-    cgroup: host
-    privileged: true
-    expose: ["5410"]
-    networks: [internal]
-
-  core:
-    image: rockchin/langbot-cloud-core:${LANGBOT_IMAGE_TAG}
-    container_name: langbot-cloud-core
-    restart: unless-stopped
-    environment: *core-env
-    volumes:
-      - core-data:/app/data
-      - box-data:/app/data/box
-    depends_on:
-      postgres: {condition: service_healthy}
-      redis: {condition: service_healthy}
-      plugin-runtime: {condition: service_started}
-      box: {condition: service_started}
-    expose: ["5300"]
-    healthcheck:
-      test: [CMD-SHELL, "python -c 'import urllib.request; urllib.request.urlopen(\"http://127.0.0.1:5300/healthz\", timeout=3)'" ]
-      interval: 10s
-      timeout: 5s
-      retries: 30
-      start_period: 30s
-    networks: [internal, shared-network]
-
-networks:
-  internal:
-  shared-network:
-    external: true
-
-volumes:
-  postgres-data:
-  redis-data:
-  plugin-data:
-  box-data:
-  core-data:

docs/multi-tenant/implementation-decisions.md

@@ -81,10 +81,10 @@ This log records implementation choices made while delivering the Workspace arch
 - Decision: The MCP ASGI mount authenticates the API key once, binds an immutable per-request `RequestContext`, and every tool checks a fixed permission before calling tenant services with that same context.
 - Reason: Authenticating the transport without propagating Workspace identity into tool calls would leave the direct service path globally scoped.
 
-### Released SDK protocol is pinned from PyPI
+### Unreleased SDK protocol is pinned reproducibly without publishing
 
-- Decision: The SDK tenancy protocol is released as `langbot-plugin==0.5.0` and LangBot pins that exact registry version.
-- Reason: The final PyPI release contains the complete tenant action context and shared Runtime hardening, while the exact version pin keeps production installs reproducible.
+- Decision: The SDK tenancy protocol is versioned as 0.4.18. This task does not create a GitHub release or publish PyPI because the user authorized pushing code, not a package release. After the SDK feature branch is final, LangBot's feature branch temporarily pins the exact pushed SDK Git commit. Before merging to master, the release gate is to publish `langbot-plugin==0.4.18` and replace the Git pin with the registry pin.
+- Reason: The current registry release does not contain the complete tenant action context and shared Runtime hardening. An exact Git commit is reproducible and keeps the feature branch testable without expanding release authority.
 
 ### Cloud directory writes stay outside Core
 

pyproject.toml

@@ -71,7 +71,7 @@ dependencies = [
     "chromadb>=1.0.0,<2.0.0",
     "qdrant-client (>=1.15.1,<2.0.0)",
     "pyseekdb==1.1.0.post3",
-    "langbot-plugin==0.5.0",
+    "langbot-plugin @ git+https://github.com/langbot-app/langbot-plugin-sdk.git@1d65ed301a6afc52150a998043f73cd6032c8162",
     "asyncpg>=0.30.0",
     "line-bot-sdk>=3.19.0",
     "matrix-nio>=0.25.2",

src/langbot/pkg/api/http/authz.py

@@ -74,6 +74,11 @@ class AuthorizationError(Exception):
     error_code = 'forbidden'
 
 
+class AuthenticationDeniedError(AuthorizationError):
+    status_code = 401
+    error_code = 'invalid_authentication'
+
+
 class WorkspaceRequiredError(AuthorizationError):
     status_code = 400
     error_code = 'workspace_required'

src/langbot/pkg/api/http/context.py

@@ -9,6 +9,7 @@ class PrincipalType(enum.StrEnum):
 
     ACCOUNT = 'account'
     API_KEY = 'api_key'
+    SUPPORT_ADMIN = 'support_admin'
     SYSTEM = 'system'
     PUBLIC_BOT = 'public_bot'
 
@@ -19,7 +20,9 @@ class PrincipalContext:
 
     principal_type: PrincipalType
     account_uuid: str | None = None
+    actor_account_uuid: str | None = None
     api_key_uuid: str | None = None
+    support_session_id: str | None = None
 
 
 @dataclasses.dataclass(frozen=True, slots=True)

src/langbot/pkg/api/http/controller/group.py

@@ -14,10 +14,18 @@ from ....utils import bounded_executor
 from ....workspace.collaboration import MembershipPermissionError, WorkspaceCollaborationError
 from ....workspace.errors import WorkspaceNotFoundError
 from ....cloud.entitlements import EntitlementUnavailableError
-from ....cloud.quotas import WorkspaceQuotaExceededError
 from ....core.errors import TaskCapacityError
-from ..authz import AuthorizationError, Permission, permissions_for_role, require_permission
+from ..authz import (
+    AuthenticationDeniedError,
+    AuthorizationError,
+    Permission,
+    PermissionDeniedError,
+    WorkspaceRequiredError,
+    permissions_for_role,
+    require_permission,
+)
 from ..context import PrincipalContext, PrincipalType, RequestContext, WorkspaceContext
+from ....cloud.support_admin import SupportAdminSessionError
 
 if typing.TYPE_CHECKING:
     from ....core.app import Application
@@ -52,6 +60,17 @@ class AuthType(enum.Enum):
     USER_TOKEN_OR_API_KEY = 'user-token-or-api-key'
 
 
+_SUPPORT_ADMIN_DENIED_PERMISSIONS = frozenset(
+    {
+        Permission.OWNER_TRANSFER.value,
+        Permission.MEMBER_VIEW.value,
+        Permission.MEMBER_INVITE.value,
+        Permission.MEMBER_UPDATE_ROLE.value,
+        Permission.MEMBER_REMOVE.value,
+    }
+)
+
+
 class RouterGroup(abc.ABC):
     name: str
 
@@ -96,6 +115,10 @@ class RouterGroup(abc.ABC):
                         return self.http_status(401, -1, 'No valid user token provided')
 
                     try:
+                        if self._is_support_admin_token(token):
+                            raise AuthenticationDeniedError(
+                                'Support admin tokens cannot be refreshed or used on account endpoints'
+                            )
                         account, user_email = await self._authenticate_account(token)
                         # Account-token routes deliberately stop before Workspace
                         # selection. They may bootstrap a selector, but cannot
@@ -112,8 +135,13 @@ class RouterGroup(abc.ABC):
                         return self.http_status(401, -1, 'No valid user token provided')
 
                     try:
-                        account, user_email = await self._authenticate_account(token)
-                        request_context = await self._resolve_account_context(account, auth_type)
+                        request_context = await self._authenticate_support_admin(token, auth_type)
+                        if request_context is not None:
+                            self._require_support_admin_route_allowed(rule, f, permission)
+                            user_email = None
+                        else:
+                            account, user_email = await self._authenticate_account(token)
+                            request_context = await self._resolve_account_context(account, auth_type)
                         if permission is not None:
                             if request_context is None:
                                 raise AuthorizationError('Workspace authorization is unavailable')
@@ -142,10 +170,20 @@ class RouterGroup(abc.ABC):
                         return self._auth_error_response(e)
 
                 elif auth_type == AuthType.USER_TOKEN_OR_API_KEY:
+                    token = quart.request.headers.get('Authorization', '').replace('Bearer ', '')
+                    if token and self._is_support_admin_token(token):
+                        try:
+                            request_context = await self._authenticate_support_admin(token, auth_type)
+                            if request_context is None:
+                                raise AuthenticationDeniedError('Invalid support admin token')
+                            self._require_support_admin_route_allowed(rule, f, permission)
+                            if permission is not None:
+                                require_permission(request_context, permission)
+                            self._inject_handler_context(f, kwargs, None, request_context)
+                        except Exception as e:
+                            return self._auth_error_response(e)
                     # Try API key first (check X-API-Key header)
-                    api_key = quart.request.headers.get('X-API-Key', '')
-
-                    if api_key:
+                    elif api_key := quart.request.headers.get('X-API-Key', ''):
                         # API key authentication
                         try:
                             request_context = await self._authenticate_api_key(api_key, auth_type)
@@ -156,8 +194,6 @@ class RouterGroup(abc.ABC):
                             return self._auth_error_response(e)
                     else:
                         # Try user token authentication (Authorization header)
-                        token = quart.request.headers.get('Authorization', '').replace('Bearer ', '')
-
                         if not token:
                             return self.http_status(
                                 401, -1, 'No valid authentication provided (user token or API key required)'
@@ -220,8 +256,6 @@ class RouterGroup(abc.ABC):
                         return self.http_status(403, e.code, str(e))
                     if isinstance(e, WorkspaceCollaborationError):
                         return self.http_status(400, e.code, str(e))
-                    if isinstance(e, WorkspaceQuotaExceededError):
-                        return self.http_status(409, e.error_code, str(e))
                     if isinstance(e, TaskCapacityError):
                         return self.http_status(429, 'task_capacity_exceeded', str(e))
                     if isinstance(
@@ -271,10 +305,83 @@ class RouterGroup(abc.ABC):
             raise ValueError('User not found')
         return account, account.user
 
+    def _is_support_admin_token(self, token: str) -> bool:
+        service = getattr(self.ap, 'support_admin_session_service', None)
+        detector = getattr(service, 'is_support_admin_token', None)
+        return callable(detector) and detector(token) is True
+
+    async def _authenticate_support_admin(
+        self,
+        token: str,
+        auth_type: AuthType,
+        *,
+        workspace_uuid: str | None = None,
+        request_id: str | None = None,
+    ) -> RequestContext | None:
+        service = getattr(self.ap, 'support_admin_session_service', None)
+        detector = getattr(service, 'is_support_admin_token', None)
+        if service is None or not callable(detector) or detector(token) is not True:
+            return None
+
+        requested_workspace_uuid = (
+            workspace_uuid if workspace_uuid is not None else quart.request.headers.get('X-Workspace-Id')
+        )
+        if not requested_workspace_uuid:
+            raise WorkspaceRequiredError('Support admin token requires an explicit Workspace selector')
+        try:
+            identity = await service.authenticate_token(
+                token,
+                requested_workspace_uuid=requested_workspace_uuid,
+            )
+        except SupportAdminSessionError as exc:
+            raise AuthenticationDeniedError(str(exc)) from exc
+
+        entitlement_revision = await self._resolve_entitlement_revision(
+            identity.instance_uuid,
+            identity.workspace_uuid,
+        )
+        request_context = RequestContext(
+            instance_uuid=identity.instance_uuid,
+            placement_generation=identity.placement_generation,
+            request_id=request_id or self.request_id(),
+            auth_type=auth_type.value,
+            principal=PrincipalContext(
+                principal_type=PrincipalType.SUPPORT_ADMIN,
+                actor_account_uuid=identity.actor_account_uuid,
+                support_session_id=identity.grant_jti_hash,
+            ),
+            workspace=WorkspaceContext(
+                workspace_uuid=identity.workspace_uuid,
+                membership_uuid=None,
+                role='owner',
+                permissions=permissions_for_role('owner') - _SUPPORT_ADMIN_DENIED_PERMISSIONS,
+                membership_revision=0,
+            ),
+            entitlement_revision=entitlement_revision,
+        )
+        quart.g.request_context = request_context
+        quart.g.workspace_membership = None
+        return request_context
+
+    @staticmethod
+    def _require_support_admin_route_allowed(
+        rule: str,
+        handler: RouteCallable,
+        permission: Permission | str | None,
+    ) -> None:
+        parameters = inspect.signature(handler).parameters
+        if rule.startswith('/api/v1/user/') or 'account' in parameters or 'user_email' in parameters:
+            raise AuthenticationDeniedError('Support admin tokens are not permitted on account endpoints')
+        permission_value = permission.value if isinstance(permission, Permission) else permission
+        if permission_value in _SUPPORT_ADMIN_DENIED_PERMISSIONS:
+            raise PermissionDeniedError(permission_value)
+
     async def _resolve_account_context(
         self,
         account: typing.Any,
         auth_type: AuthType,
+        *,
+        token: str | None = None,
     ) -> RequestContext | None:
         collaboration_service = getattr(self.ap, 'workspace_collaboration_service', None)
         account_uuid = getattr(account, 'uuid', None)

src/langbot/pkg/api/http/controller/groups/pipelines/websocket_chat.py

@@ -97,6 +97,16 @@ class WebSocketChatRouterGroup(group.RouterGroup):
         if not token or not workspace_uuid:
             raise ValueError('Authentication is required')
 
+        support_context = await self._authenticate_support_admin(
+            token,
+            group.AuthType.USER_TOKEN,
+            workspace_uuid=workspace_uuid,
+            request_id=quart.websocket.headers.get('X-Request-Id') or str(uuid.uuid4()),
+        )
+        if support_context is not None:
+            require_permission(support_context, Permission.RUNTIME_OPERATE)
+            return support_context, token
+
         account, _ = await self._authenticate_account(token)
         account_uuid = getattr(account, 'uuid', None)
         collaboration_service = getattr(self.ap, 'workspace_collaboration_service', None)
@@ -131,6 +141,23 @@ class WebSocketChatRouterGroup(group.RouterGroup):
     ) -> RequestContext:
         """Recheck revocable account, membership, permission, and placement state."""
 
+        if request_context.principal.principal_type == PrincipalType.SUPPORT_ADMIN:
+            current_context = await self._authenticate_support_admin(
+                token,
+                group.AuthType.USER_TOKEN,
+                workspace_uuid=request_context.workspace_uuid,
+                request_id=request_context.request_id,
+            )
+            if current_context is None or current_context.principal != request_context.principal:
+                raise ValueError('WebSocket support admin session changed')
+            if (
+                current_context.instance_uuid != request_context.instance_uuid
+                or current_context.placement_generation != request_context.placement_generation
+            ):
+                raise ValueError('WebSocket authorization changed')
+            require_permission(current_context, Permission.RUNTIME_OPERATE)
+            return current_context
+
         account, _ = await self._authenticate_account(token)
         account_uuid = getattr(account, 'uuid', None)
         if account_uuid != request_context.account_uuid:
@@ -211,6 +238,7 @@ class WebSocketChatRouterGroup(group.RouterGroup):
                     scope=WebSocketScope.from_context(request_context),
                     pipeline_uuid=pipeline_uuid,
                     session_type=session_type,
+                    trigger_principal=request_context.principal,
                     metadata={'user_agent': quart.websocket.headers.get('User-Agent', '')},
                     send_queue_size=(
                         self.ap.instance_config.data.get('system', {})
@@ -391,7 +419,7 @@ class WebSocketChatRouterGroup(group.RouterGroup):
                         )
                     elif message_type == 'message':
                         try:
-                            await self._revalidate_websocket_authorization(request_context, token)
+                            request_context = await self._revalidate_websocket_authorization(request_context, token)
                         except Exception:
                             await connection.send_queue.put({'type': 'error', 'message': 'Unauthorized'})
                             break

src/langbot/pkg/api/http/controller/groups/platform/adapters.py

@@ -22,6 +22,7 @@ class _AdapterSessionScope:
     principal_type: str
     account_uuid: str | None
     api_key_uuid: str | None
+    support_session_id: str | None
 
     @classmethod
     def from_request_context(cls, request_context: RequestContext) -> '_AdapterSessionScope':
@@ -33,6 +34,7 @@ class _AdapterSessionScope:
             principal_type=principal.principal_type.value,
             account_uuid=principal.account_uuid,
             api_key_uuid=principal.api_key_uuid,
+            support_session_id=principal.support_session_id,
         )
 
     def matches(self, request_context: RequestContext) -> bool:

src/langbot/pkg/api/http/controller/groups/user.py

@@ -396,6 +396,19 @@ class UserRouterGroup(group.RouterGroup):
                 launch_assertion,
                 expected_workspace_uuid=workspace_uuid,
             )
+            if launch.get('launch_mode') == 'support_admin':
+                token = launch.get('support_admin_token')
+                if not token:
+                    raise SpaceLaunchError('Support admin launch session was not issued')
+                return self.success(
+                    data={
+                        'token': token,
+                        'workspace_uuid': launch['workspace_uuid'],
+                        'principal_type': 'support_admin',
+                        'actor_account_uuid': launch['actor_account_uuid'],
+                    }
+                )
+
             account = await self.ap.user_service.get_user_by_uuid(launch['account_uuid'])
             if account is None:
                 raise SpaceLaunchError('Launch Account is not projected into Core')
@@ -410,7 +423,6 @@ class UserRouterGroup(group.RouterGroup):
                     'token': token,
                     'user': account.user,
                     'workspace_uuid': access.workspace.uuid,
-                    'return_path': launch.get('return_path', '/home'),
                 }
             )
         except SpaceLaunchError:

src/langbot/pkg/api/http/controller/groups/workspaces.py

@@ -5,7 +5,7 @@ import typing
 import quart
 
 from ...authz import Permission, permissions_for_role
-from ...context import RequestContext
+from ...context import PrincipalType, RequestContext
 from ...service.user import AccountExistsLoginRequiredError, ControlPlaneDirectoryRequiredError
 from .....entity.persistence.workspace import Workspace, WorkspaceInvitation, WorkspaceMembership
 from .....entity.persistence.workspace import WorkspaceSource
@@ -120,9 +120,6 @@ class WorkspacesRouterGroup(group.RouterGroup):
         @self.route('/current', methods=['GET'], permission=Permission.WORKSPACE_VIEW)
         async def _(request_context: RequestContext) -> typing.Any:
             membership = quart.g.workspace_membership
-            account = await self.ap.user_service.get_user_by_uuid(request_context.account_uuid)
-            if account is None:
-                return self.http_status(401, 'invalid_authentication', 'Account not found')
             workspace = await self.ap.workspace_service.get_workspace(request_context.workspace_uuid)
             plan_name: str | None = None
             resolver = getattr(self.ap, 'entitlement_resolver', None)
@@ -132,6 +129,28 @@ class WorkspacesRouterGroup(group.RouterGroup):
                     minimum_revision=request_context.entitlement_revision,
                 )
                 plan_name = entitlement.plan_name
+            if request_context.principal.principal_type == PrincipalType.SUPPORT_ADMIN:
+                return self.success(
+                    data={
+                        'workspace': _workspace_payload(workspace),
+                        'membership': {
+                            'uuid': None,
+                            'workspace_uuid': request_context.workspace_uuid,
+                            'account_uuid': None,
+                            'email': None,
+                            'role': 'owner',
+                            'status': 'active',
+                            'joined_at': None,
+                            'created_at': None,
+                        },
+                        'permissions': sorted(request_context.workspace.permissions),
+                        'placement_generation': request_context.placement_generation,
+                        'plan_name': plan_name,
+                    }
+                )
+            account = await self.ap.user_service.get_user_by_uuid(request_context.account_uuid)
+            if account is None:
+                return self.http_status(401, 'invalid_authentication', 'Account not found')
             return self.success(
                 data={
                     'workspace': _workspace_payload(workspace),

src/langbot/pkg/api/http/service/bot.py

@@ -4,7 +4,6 @@ import uuid
 import sqlalchemy
 
 from ....core import app
-from ....cloud.quotas import require_resource_capacity, resolve_workspace_quota
 from ....entity.persistence import bot as persistence_bot
 from ....entity.persistence import pipeline as persistence_pipeline
 from ....workspace.errors import WorkspaceNotFoundError
@@ -102,21 +101,20 @@ class BotService:
     async def create_bot(self, context: TenantContext, bot_data: dict) -> str:
         """Create bot"""
         workspace_uuid = require_workspace_uuid(context)
+        # Check limitation
         limitation = self.ap.instance_config.data.get('system', {}).get('limitation', {})
-        quota = await resolve_workspace_quota(
-            self.ap,
-            workspace_uuid,
-            'bots.max',
-            fallback=limitation.get('max_bots', -1),
-        )
+        max_bots = limitation.get('max_bots', -1)
+        if max_bots >= 0:
+            existing_bots = await self.get_bots(context)
+            if len(existing_bots) >= max_bots:
+                raise ValueError(f'Maximum number of bots ({max_bots}) reached')
 
         # TODO: 检查配置信息格式
         bot_data = bot_data.copy()
         bot_data['uuid'] = str(uuid.uuid4())
         bot_data['workspace_uuid'] = workspace_uuid
 
-        # Preserve the legacy flat-row result shape for this optional lookup;
-        # quota admission and insertion below still share one transaction.
+        # bind the most recently updated pipeline if any exist
         result = await self.ap.persistence_mgr.execute_async(
             scope_statement(
                 sqlalchemy.select(persistence_pipeline.LegacyPipeline),
@@ -131,25 +129,7 @@ class BotService:
             bot_data['use_pipeline_uuid'] = pipeline.uuid
             bot_data['use_pipeline_name'] = pipeline.name
 
-        async def persist(execute) -> None:
-            await require_resource_capacity(
-                execute,
-                workspace_uuid=workspace_uuid,
-                model=persistence_bot.Bot,
-                quota=quota,
-                resource_name='bots',
-            )
-
-            await execute(sqlalchemy.insert(persistence_bot.Bot).values(bot_data))
-
-        tenant_uow = getattr(self.ap.persistence_mgr, 'tenant_uow', None)
-        if quota.requires_transaction_lock:
-            if not callable(tenant_uow):
-                raise RuntimeError('Cloud bot quota enforcement requires transactional persistence')
-            async with tenant_uow(workspace_uuid) as uow:
-                await persist(uow.execute)
-        else:
-            await persist(self.ap.persistence_mgr.execute_async)
+        await self.ap.persistence_mgr.execute_async(sqlalchemy.insert(persistence_bot.Bot).values(bot_data))
 
         bot = await self.get_bot(context, bot_data['uuid'], include_secret=True)
 

src/langbot/pkg/api/http/service/user.py

@@ -400,7 +400,10 @@ class UserService:
 
         return await self.generate_jwt_token(user_obj)
 
-    async def generate_jwt_token(self, account: user.User | str) -> str:
+    async def generate_jwt_token(
+        self,
+        account: user.User | str,
+    ) -> str:
         jwt_secret = self.ap.instance_config.data['system']['jwt']['secret']
         jwt_expire = self.ap.instance_config.data['system']['jwt']['expire']
 
@@ -413,7 +416,7 @@ class UserService:
                 # Lightweight unit-test and bootstrap callers may not have persistence wired.
                 account_obj = None
 
-        payload = {
+        payload: dict[str, typing.Any] = {
             'user': user_email,
             'iss': self._jwt_identity()[0],
             'aud': self._jwt_identity()[1],

src/langbot/pkg/cloud/bootstrap.py

@@ -138,14 +138,8 @@ class VerifiedCloudDeployment:
         if plugin_worker.get('require_hard_limits') is not True:
             raise CloudBootstrapError('Cloud Runtime requires plugin.worker.require_hard_limits=true')
         box_config = config.get('box', {})
-        box_enabled = box_config.get('enabled')
-        if box_enabled is False:
-            # Explicitly disabling Box removes the sandbox surface entirely and
-            # therefore does not weaken tenant isolation. Validate the strict
-            # runtime/admission contract only when the surface is enabled.
-            return
-        if box_enabled is not True:
-            raise CloudBootstrapError('Cloud runtime requires box.enabled to be an explicit boolean')
+        if box_config.get('enabled') is not True:
+            raise CloudBootstrapError('Cloud runtime requires box.enabled=true')
         if box_config.get('backend') != 'nsjail':
             raise CloudBootstrapError('Cloud runtime requires box.backend=nsjail')
         runtime_endpoint = str(box_config.get('runtime', {}).get('endpoint', '') or '').strip()

src/langbot/pkg/cloud/launch.py

@@ -3,11 +3,9 @@ from __future__ import annotations
 import asyncio
 import base64
 import binascii
-import datetime
 import hashlib
 import heapq
 import json
-import math
 import os
 import time
 import typing
@@ -16,10 +14,8 @@ from collections.abc import Callable, Iterable
 from cryptography.exceptions import InvalidSignature
 from cryptography.hazmat.primitives import serialization
 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
-import sqlalchemy
-from sqlalchemy.dialects.postgresql import insert as pg_insert
 
-from ..entity.persistence.cloud_directory import SpaceLaunchAssertionConsumption
+from .support_admin import SupportAdminReplayError, SupportAdminSessionError, hash_grant_jti
 
 if typing.TYPE_CHECKING:
     from ..core.app import Application
@@ -27,6 +23,7 @@ if typing.TYPE_CHECKING:
 
 CONTROL_PLANE_TYP = 'langbot-control-plane+jwt'
 LAUNCH_KIND = 'workspace.launch'
+SUPPORT_ADMIN_LAUNCH_KIND = 'workspace.support_admin_launch'
 EXPECTED_ISSUER = 'langbot-space'
 EXPECTED_AUDIENCE = 'langbot-cloud-runtime'
 _CONSUMED_JTI_MAX_ENTRIES = 4096
@@ -125,30 +122,66 @@ class SpaceLaunchService:
         *,
         expected_workspace_uuid: str | None = None,
     ) -> dict[str, str]:
-        claims, clock_skew_seconds = self._verify_assertion(assertion)
+        claims = self._verify_assertion(assertion)
         payload = claims.get('payload')
         if not isinstance(payload, dict):
             raise SpaceLaunchError('Launch assertion payload must be a JSON object')
-        account_uuid = _required_string(payload, 'account_uuid')
+        kind = _required_string(claims, 'kind')
         workspace_uuid = _required_string(payload, 'workspace_uuid')
-        return_path = _required_string(payload, 'return_path')
-        if (
-            not return_path.startswith('/')
-            or return_path.startswith('//')
-            or any(character in return_path for character in ('\\', '\r', '\n', '\t'))
-        ):
-            raise SpaceLaunchError('Launch assertion return path is invalid')
         if expected_workspace_uuid is not None and workspace_uuid != expected_workspace_uuid:
             raise SpaceLaunchError('Launch assertion targets another Workspace')
-        replay_retention_expires_at = _required_int(claims, 'exp', minimum=1) + math.ceil(clock_skew_seconds)
-        await self._consume_jti(_required_string(claims, 'jti'), replay_retention_expires_at)
-        return {
+        if kind == SUPPORT_ADMIN_LAUNCH_KIND:
+            if 'account_uuid' in payload:
+                raise SpaceLaunchError('Admin launch assertion must not identify a customer Account')
+            if payload.get('launch_mode') != 'support_admin' or payload.get('principal_type') != 'support_admin':
+                raise SpaceLaunchError('Admin launch principal must be support_admin')
+            actor_account_uuid = _required_string(payload, 'actor_account_uuid')
+            if _required_string(payload, 'effective_role') != 'owner':
+                raise SpaceLaunchError('Admin launch effective role must be owner')
+            issued_at = _required_int(claims, 'iat')
+            expires_at = _required_int(claims, 'exp', minimum=1)
+            if expires_at - issued_at > 90:
+                raise SpaceLaunchError('Admin launch assertion lifetime exceeds 90 seconds')
+            grant_jti_hash = hash_grant_jti(_required_string(claims, 'jti'))
+            result = {
+                'workspace_uuid': workspace_uuid,
+                'launch_mode': 'support_admin',
+                'actor_account_uuid': actor_account_uuid,
+                'effective_role': 'owner',
+                'grant_jti_hash': grant_jti_hash,
+            }
+            support_service = getattr(self.ap, 'support_admin_session_service', None)
+            if support_service is None or not callable(getattr(support_service, 'consume_launch_grant', None)):
+                raise SpaceLaunchError('Durable support admin session service is unavailable')
+            try:
+                support_session = await support_service.consume_launch_grant(
+                    grant_jti_hash=grant_jti_hash,
+                    workspace_uuid=workspace_uuid,
+                    actor_account_uuid=actor_account_uuid,
+                )
+            except SupportAdminReplayError as exc:
+                raise SpaceLaunchError('Launch assertion has already been consumed') from exc
+            except SupportAdminSessionError as exc:
+                raise SpaceLaunchError(str(exc)) from exc
+            result['support_admin_token'] = support_session.token
+            self.ap.logger.info(
+                'cloud_support_admin_launch_consumed actor_account_uuid=%s workspace_uuid=%s',
+                result['actor_account_uuid'],
+                workspace_uuid,
+            )
+            return result
+
+        if payload.get('launch_mode') is not None:
+            raise SpaceLaunchError('Launch assertion mode is unsupported')
+        account_uuid = _required_string(payload, 'account_uuid')
+        result = {
             'account_uuid': account_uuid,
             'workspace_uuid': workspace_uuid,
-            'return_path': return_path,
         }
+        await self._consume_jti(_required_string(claims, 'jti'), _required_int(claims, 'exp', minimum=1))
+        return result
 
-    def _verify_assertion(self, token: str) -> tuple[dict[str, typing.Any], float]:
+    def _verify_assertion(self, token: str) -> dict[str, typing.Any]:
         if not getattr(getattr(self.ap, 'deployment', None), 'multi_workspace_enabled', False):
             raise SpaceLaunchError('Space direct launch requires verified Cloud mode')
         public_key, key_id, clock_skew_seconds = self._trust_config()
@@ -184,8 +217,9 @@ class SpaceLaunchService:
             raise SpaceLaunchError('Launch assertion subject targets another instance')
         if _required_string(claims, 'instance_uuid') != instance_uuid:
             raise SpaceLaunchError('Launch assertion instance UUID does not match this Core')
-        if _required_string(claims, 'kind') != LAUNCH_KIND:
-            raise SpaceLaunchError('Launch assertion kind is not workspace.launch')
+        kind = _required_string(claims, 'kind')
+        if kind not in {LAUNCH_KIND, SUPPORT_ADMIN_LAUNCH_KIND}:
+            raise SpaceLaunchError('Launch assertion kind is not supported')
 
         issued_at = _required_int(claims, 'iat')
         not_before = _required_int(claims, 'nbf')
@@ -199,7 +233,7 @@ class SpaceLaunchService:
             raise SpaceLaunchError('Launch assertion is expired')
         if expires_at <= max(issued_at, not_before):
             raise SpaceLaunchError('Launch assertion expiry must follow issue time')
-        return claims, clock_skew_seconds
+        return claims
 
     def _trust_config(self) -> tuple[Ed25519PublicKey, str, float]:
         data = getattr(getattr(self.ap, 'instance_config', None), 'data', {}) or {}
@@ -229,39 +263,19 @@ class SpaceLaunchService:
     async def _consume_jti(self, jti: str, expires_at: int) -> None:
         digest = hashlib.sha256(jti.encode('utf-8')).hexdigest()
         now = int(self._wall_time())
-        persistence_mgr = getattr(self.ap, 'persistence_mgr', None)
-        instance_uuid = str(self.ap.workspace_service.instance_uuid)
-        if persistence_mgr is not None:
-            expires_at_datetime = datetime.datetime.fromtimestamp(expires_at, tz=datetime.timezone.utc)
-            now_datetime = datetime.datetime.fromtimestamp(now, tz=datetime.timezone.utc)
-            async with persistence_mgr.directory_projection_uow(instance_uuid) as uow:
-                await uow.session.execute(
-                    sqlalchemy.delete(SpaceLaunchAssertionConsumption).where(
-                        SpaceLaunchAssertionConsumption.instance_uuid == instance_uuid,
-                        SpaceLaunchAssertionConsumption.expires_at < now_datetime,
-                    )
-                )
-                statement = (
-                    pg_insert(SpaceLaunchAssertionConsumption)
-                    .values(instance_uuid=instance_uuid, jti=digest, expires_at=expires_at_datetime)
-                    .on_conflict_do_nothing(index_elements=['instance_uuid', 'jti'])
-                    .returning(SpaceLaunchAssertionConsumption.jti)
-                )
-                result = await uow.session.execute(statement)
-                if result.scalar_one_or_none() is None:
-                    raise SpaceLaunchError('Launch assertion has already been consumed')
-            return
-
-        # Lightweight unit-test and OSS compatibility fallback. Verified Cloud
-        # runtime always supplies the durable PostgreSQL persistence manager.
         async with self._replay_lock:
             self._prune_consumed_jtis(now)
             if digest in self._consumed_jtis:
                 raise SpaceLaunchError('Launch assertion has already been consumed')
             if len(self._consumed_jtis) >= _CONSUMED_JTI_MAX_ENTRIES:
+                # Evicting a still-valid digest would make a signed launch
+                # assertion replayable. Bound memory by failing closed instead.
                 raise SpaceLaunchError('Launch assertion replay cache capacity reached')
             self._consumed_jtis[digest] = expires_at
-            heapq.heappush(self._consumed_jti_expiry_heap, (expires_at, digest))
+            heapq.heappush(
+                self._consumed_jti_expiry_heap,
+                (expires_at, digest),
+            )
 
     def _prune_consumed_jtis(self, now: int) -> None:
         while self._consumed_jti_expiry_heap:

src/langbot/pkg/cloud/quotas.py

@@ -1,82 +0,0 @@
-from __future__ import annotations
-
-from dataclasses import dataclass
-from typing import Any, Awaitable, Callable
-
-import sqlalchemy
-
-from ..entity.persistence import workspace as persistence_workspace
-from .entitlements import EntitlementResolver
-
-
-Execute = Callable[[Any], Awaitable[Any]]
-
-
-class WorkspaceQuotaExceededError(ValueError):
-    """A stable business error raised when a workspace has no free slots."""
-
-    error_code = 'workspace_quota_exceeded'
-
-    def __init__(self, resource_name: str, limit: int) -> None:
-        self.resource_name = resource_name
-        self.limit = limit
-        super().__init__(f'Maximum number of {resource_name} ({limit}) reached')
-
-
-@dataclass(frozen=True, slots=True)
-class WorkspaceQuota:
-    limit: int
-    requires_transaction_lock: bool
-
-
-async def resolve_workspace_quota(
-    ap: Any,
-    workspace_uuid: str,
-    limit_name: str,
-    *,
-    fallback: int = -1,
-) -> WorkspaceQuota:
-    """Resolve a plan-agnostic Cloud limit while preserving OSS configuration."""
-
-    resolver = getattr(ap, 'entitlement_resolver', None)
-    if isinstance(resolver, EntitlementResolver):
-        snapshot = await resolver.resolve(workspace_uuid)
-        return WorkspaceQuota(
-            limit=snapshot.limit(limit_name),
-            requires_transaction_lock=True,
-        )
-    return WorkspaceQuota(limit=fallback, requires_transaction_lock=False)
-
-
-async def lock_workspace_for_quota(execute: Execute, workspace_uuid: str) -> None:
-    """Serialize quota checks on the durable Workspace row within one transaction."""
-
-    result = await execute(
-        sqlalchemy.select(persistence_workspace.Workspace.uuid)
-        .where(persistence_workspace.Workspace.uuid == workspace_uuid)
-        .with_for_update()
-    )
-    if result.first() is None:
-        raise ValueError('Workspace does not exist')
-
-
-async def require_resource_capacity(
-    execute: Execute,
-    *,
-    workspace_uuid: str,
-    model: type,
-    quota: WorkspaceQuota,
-    resource_name: str,
-    workspace_locked: bool = False,
-) -> None:
-    if quota.limit < 0:
-        return
-    if quota.requires_transaction_lock and not workspace_locked:
-        await lock_workspace_for_quota(execute, workspace_uuid)
-    result = await execute(
-        sqlalchemy.select(sqlalchemy.func.count())
-        .select_from(model)
-        .where(model.workspace_uuid == workspace_uuid)
-    )
-    if int(result.scalar_one()) >= quota.limit:
-        raise WorkspaceQuotaExceededError(resource_name, quota.limit)

src/langbot/pkg/cloud/support_admin.py

@@ -0,0 +1,248 @@
+from __future__ import annotations
+
+import dataclasses
+import datetime
+import hashlib
+import re
+import time
+import typing
+
+import jwt
+from sqlalchemy.exc import IntegrityError
+
+from ..entity.persistence.support_admin import SupportAdminTemporarySession
+from ..workspace.errors import WorkspaceError
+
+if typing.TYPE_CHECKING:
+    from ..core.app import Application
+
+
+SUPPORT_ADMIN_TOKEN_TYP = 'langbot-support-admin+jwt'
+SUPPORT_ADMIN_TOKEN_KIND = 'support_admin.session'
+SUPPORT_ADMIN_EFFECTIVE_ROLE = 'owner'
+SUPPORT_ADMIN_MAX_TOKEN_SECONDS = 300
+_SHA256_HEX = re.compile(r'^[0-9a-f]{64}$')
+
+
+class SupportAdminSessionError(ValueError):
+    """Raised when a support-admin session or token is not admissible."""
+
+
+class SupportAdminReplayError(SupportAdminSessionError):
+    """Raised when a launch grant JTI has already been consumed."""
+
+
+@dataclasses.dataclass(frozen=True, slots=True)
+class IssuedSupportAdminSession:
+    token: str
+    grant_jti_hash: str
+    workspace_uuid: str
+    actor_account_uuid: str
+    issued_at: datetime.datetime
+    expires_at: datetime.datetime
+
+
+@dataclasses.dataclass(frozen=True, slots=True)
+class SupportAdminSessionIdentity:
+    grant_jti_hash: str
+    workspace_uuid: str
+    actor_account_uuid: str
+    instance_uuid: str
+    placement_generation: int
+
+
+def hash_grant_jti(jti: str) -> str:
+    return hashlib.sha256(jti.encode('utf-8')).hexdigest()
+
+
+class SupportAdminSessionService:
+    """Issue and validate temporary Workspace-scoped support-admin sessions."""
+
+    def __init__(
+        self,
+        ap: Application,
+        *,
+        wall_time: typing.Callable[[], float] = time.time,
+    ) -> None:
+        self.ap = ap
+        self._wall_time = wall_time
+
+    async def consume_launch_grant(
+        self,
+        *,
+        grant_jti_hash: str,
+        workspace_uuid: str,
+        actor_account_uuid: str,
+    ) -> IssuedSupportAdminSession:
+        self._validate_grant_hash(grant_jti_hash)
+        if not workspace_uuid or not actor_account_uuid:
+            raise SupportAdminSessionError('Support admin session requires an actor and Workspace')
+
+        issued_at = self._utcnow()
+        expires_at = issued_at + datetime.timedelta(seconds=SUPPORT_ADMIN_MAX_TOKEN_SECONDS)
+        tenant_uow = getattr(self.ap.persistence_mgr, 'tenant_uow', None)
+        if not callable(tenant_uow):
+            raise SupportAdminSessionError('Support admin sessions require tenant persistence')
+
+        try:
+            async with tenant_uow(workspace_uuid) as uow:
+                await self.ap.workspace_service.get_execution_binding(workspace_uuid, session=uow.session)
+                uow.session.add(
+                    SupportAdminTemporarySession(
+                        grant_jti_hash=grant_jti_hash,
+                        workspace_uuid=workspace_uuid,
+                        actor_account_uuid=actor_account_uuid,
+                        issued_at=issued_at,
+                        expires_at=expires_at,
+                    )
+                )
+                await uow.session.flush()
+        except IntegrityError as exc:
+            raise SupportAdminReplayError('Launch assertion has already been consumed') from exc
+        except WorkspaceError as exc:
+            raise SupportAdminSessionError('Workspace is unavailable for support access') from exc
+
+        return IssuedSupportAdminSession(
+            token=self._encode_token(
+                grant_jti_hash=grant_jti_hash,
+                workspace_uuid=workspace_uuid,
+                actor_account_uuid=actor_account_uuid,
+                issued_at=issued_at,
+                expires_at=expires_at,
+            ),
+            grant_jti_hash=grant_jti_hash,
+            workspace_uuid=workspace_uuid,
+            actor_account_uuid=actor_account_uuid,
+            issued_at=issued_at,
+            expires_at=expires_at,
+        )
+
+    def is_support_admin_token(self, token: str) -> bool:
+        """Return True only for compact JWTs marked as support-admin tokens."""
+
+        if not isinstance(token, str) or token.count('.') != 2:
+            return False
+        try:
+            header = jwt.get_unverified_header(token)
+        except jwt.PyJWTError:
+            return False
+        if header.get('typ') == SUPPORT_ADMIN_TOKEN_TYP:
+            return True
+        try:
+            payload = jwt.decode(token, options={'verify_signature': False})
+        except jwt.PyJWTError:
+            return False
+        return payload.get('kind') == SUPPORT_ADMIN_TOKEN_KIND
+
+    async def authenticate_token(
+        self,
+        token: str,
+        *,
+        requested_workspace_uuid: str | None,
+    ) -> SupportAdminSessionIdentity:
+        if not self.is_support_admin_token(token):
+            raise SupportAdminSessionError('Not a support admin token')
+        workspace_uuid = (requested_workspace_uuid or '').strip()
+        if not workspace_uuid:
+            raise SupportAdminSessionError('Support admin token requires an explicit Workspace selector')
+
+        jwt_secret = self.ap.instance_config.data['system']['jwt']['secret']
+        try:
+            payload = jwt.decode(
+                token,
+                jwt_secret,
+                algorithms=['HS256'],
+                issuer='langbot-core',
+                audience=self._audience(workspace_uuid),
+                options={'require': ['exp', 'iat', 'nbf', 'iss', 'aud']},
+            )
+        except jwt.PyJWTError as exc:
+            raise SupportAdminSessionError('Invalid support admin token') from exc
+        self._validate_payload(payload, workspace_uuid)
+        grant_jti_hash = payload['grant_jti_hash']
+        actor_account_uuid = payload['actor_account_uuid']
+
+        tenant_uow = getattr(self.ap.persistence_mgr, 'tenant_uow', None)
+        if not callable(tenant_uow):
+            raise SupportAdminSessionError('Support admin sessions require tenant persistence')
+
+        now = self._utcnow()
+        async with tenant_uow(workspace_uuid) as uow:
+            session = await uow.session.get(SupportAdminTemporarySession, grant_jti_hash)
+            if (
+                session is None
+                or session.workspace_uuid != workspace_uuid
+                or session.actor_account_uuid != actor_account_uuid
+                or session.revoked_at is not None
+                or session.expires_at <= now
+            ):
+                raise SupportAdminSessionError('Support admin session is inactive')
+            binding = await self.ap.workspace_service.get_execution_binding(workspace_uuid, session=uow.session)
+            session.last_used_at = now
+            await uow.session.flush()
+
+        return SupportAdminSessionIdentity(
+            grant_jti_hash=grant_jti_hash,
+            workspace_uuid=workspace_uuid,
+            actor_account_uuid=actor_account_uuid,
+            instance_uuid=binding.instance_uuid,
+            placement_generation=binding.placement_generation,
+        )
+
+    async def revoke_session(self, grant_jti_hash: str, workspace_uuid: str) -> None:
+        self._validate_grant_hash(grant_jti_hash)
+        now = self._utcnow()
+        async with self.ap.persistence_mgr.tenant_uow(workspace_uuid) as uow:
+            row = await uow.session.get(SupportAdminTemporarySession, grant_jti_hash)
+            if row is not None and row.revoked_at is None:
+                row.revoked_at = now
+
+    def _encode_token(
+        self,
+        *,
+        grant_jti_hash: str,
+        workspace_uuid: str,
+        actor_account_uuid: str,
+        issued_at: datetime.datetime,
+        expires_at: datetime.datetime,
+    ) -> str:
+        jwt_secret = self.ap.instance_config.data['system']['jwt']['secret']
+        payload: dict[str, typing.Any] = {
+            'kind': SUPPORT_ADMIN_TOKEN_KIND,
+            'iss': 'langbot-core',
+            'aud': self._audience(workspace_uuid),
+            'sub': f'support-admin:{actor_account_uuid}',
+            'iat': issued_at,
+            'nbf': issued_at,
+            'exp': expires_at,
+            'actor_account_uuid': actor_account_uuid,
+            'workspace_uuid': workspace_uuid,
+            'effective_role': SUPPORT_ADMIN_EFFECTIVE_ROLE,
+            'grant_jti_hash': grant_jti_hash,
+        }
+        return jwt.encode(payload, jwt_secret, algorithm='HS256', headers={'typ': SUPPORT_ADMIN_TOKEN_TYP})
+
+    def _validate_payload(self, payload: dict[str, typing.Any], workspace_uuid: str) -> None:
+        if payload.get('kind') != SUPPORT_ADMIN_TOKEN_KIND:
+            raise SupportAdminSessionError('Invalid support admin token kind')
+        if payload.get('workspace_uuid') != workspace_uuid:
+            raise SupportAdminSessionError('Support admin session is scoped to another Workspace')
+        if payload.get('effective_role') != SUPPORT_ADMIN_EFFECTIVE_ROLE:
+            raise SupportAdminSessionError('Invalid support admin token role')
+        actor_account_uuid = payload.get('actor_account_uuid')
+        if not isinstance(actor_account_uuid, str) or not actor_account_uuid.strip():
+            raise SupportAdminSessionError('Invalid support admin actor')
+        grant_jti_hash = payload.get('grant_jti_hash')
+        if not isinstance(grant_jti_hash, str) or not _SHA256_HEX.match(grant_jti_hash):
+            raise SupportAdminSessionError('Invalid support admin grant')
+
+    def _audience(self, workspace_uuid: str) -> str:
+        return f'langbot-support-admin:{self.ap.workspace_service.instance_uuid}:{workspace_uuid}'
+
+    @staticmethod
+    def _validate_grant_hash(grant_jti_hash: str) -> None:
+        if not _SHA256_HEX.match(grant_jti_hash):
+            raise SupportAdminSessionError('Invalid support admin grant')
+
+    def _utcnow(self) -> datetime.datetime:
+        return datetime.datetime.fromtimestamp(self._wall_time(), datetime.UTC).replace(tzinfo=None)

src/langbot/pkg/core/app.py

@@ -51,6 +51,7 @@ from ..workspace import collaboration as workspace_collaboration_module
 from ..workspace import invitation_delivery as invitation_delivery_module
 from ..cloud import bootstrap as cloud_bootstrap_module
 from ..cloud import launch as cloud_launch_module
+from ..cloud import support_admin as cloud_support_admin_module
 from ..cloud import directory_projection as cloud_directory_projection_module
 from ..cloud import entitlements as cloud_entitlements_module
 from ..api.http.context import ExecutionContext, PrincipalContext, PrincipalType
@@ -136,6 +137,8 @@ class Application:
 
     space_launch_service: cloud_launch_module.SpaceLaunchService = None
 
+    support_admin_session_service: cloud_support_admin_module.SupportAdminSessionService = None
+
     deployment: cloud_bootstrap_module.OpenSourceDeployment | cloud_bootstrap_module.VerifiedCloudDeployment = None
 
     deployment_admission: cloud_bootstrap_module.DeploymentAdmissionGuard = None

src/langbot/pkg/core/stages/build_app.py

@@ -42,6 +42,7 @@ from ...workspace import collaboration as workspace_collaboration_module
 from ...workspace import invitation_delivery as invitation_delivery_module
 from ...cloud import bootstrap as cloud_bootstrap
 from ...cloud import launch as cloud_launch_module
+from ...cloud import support_admin as cloud_support_admin_module
 from ...cloud.directory import directory_projection_limits_from_config
 from ...cloud.directory_projection import DirectoryProjectionService
 from ...cloud.entitlements import EntitlementResolver
@@ -180,6 +181,7 @@ class BuildAppStage(stage.BootingStage):
             workspace_service_inst,
         )
         ap.invitation_delivery_service = invitation_delivery_module.InvitationDeliveryService(ap)
+        ap.support_admin_session_service = cloud_support_admin_module.SupportAdminSessionService(ap)
         ap.space_launch_service = cloud_launch_module.SpaceLaunchService(ap)
 
         user_service_inst = user_service.UserService(ap)

src/langbot/pkg/core/stages/load_config.py

@@ -186,14 +186,9 @@ def _apply_env_overrides_to_config(cfg: dict) -> dict:
                 # At the final key
                 if key in current:
                     if isinstance(current[key], list):
-                        # Convert comma-separated values while preserving the
-                        # element type declared by a non-empty config default.
-                        items = [item.strip() for item in env_value.split(',') if item.strip()]
-                        if current[key]:
-                            exemplar = current[key][0]
-                            current[key] = [convert_value(item, exemplar) for item in items]
-                        else:
-                            current[key] = items
+                        # Convert comma-separated string to list
+                        # e.g., SYSTEM__DISABLED_ADAPTERS="aiocqhttp,dingtalk"
+                        current[key] = [item.strip() for item in env_value.split(',') if item.strip()]
                     elif isinstance(current[key], dict):
                         # Skip dict types
                         pass

src/langbot/pkg/entity/persistence/cloud_directory.py

@@ -67,26 +67,3 @@ class DirectoryProjectionInbox(Base):
             name='ck_directory_projection_inbox_fingerprint',
         ),
     )
-
-
-class SpaceLaunchAssertionConsumption(Base):
-    """Durable, instance-scoped replay ledger for signed Space launch assertions."""
-
-    __tablename__ = 'space_launch_assertion_consumptions'
-
-    instance_uuid = sqlalchemy.Column(sqlalchemy.String(255), primary_key=True)
-    jti = sqlalchemy.Column(sqlalchemy.String(255), primary_key=True)
-    expires_at = sqlalchemy.Column(sqlalchemy.DateTime(timezone=True), nullable=False)
-    consumed_at = sqlalchemy.Column(
-        sqlalchemy.DateTime(timezone=True),
-        nullable=False,
-        server_default=sqlalchemy.func.now(),
-    )
-
-    __table_args__ = (
-        sqlalchemy.Index(
-            'ix_space_launch_assertion_consumptions_expiry',
-            'instance_uuid',
-            'expires_at',
-        ),
-    )

src/langbot/pkg/entity/persistence/support_admin.py

@@ -0,0 +1,35 @@
+from __future__ import annotations
+
+import sqlalchemy
+
+from .base import Base
+
+
+class SupportAdminTemporarySession(Base):
+    """Temporary support-admin Workspace access session."""
+
+    __tablename__ = 'support_admin_temporary_sessions'
+
+    grant_jti_hash = sqlalchemy.Column(sqlalchemy.String(64), primary_key=True)
+    workspace_uuid = sqlalchemy.Column(
+        sqlalchemy.String(36),
+        sqlalchemy.ForeignKey('workspaces.uuid', ondelete='CASCADE'),
+        nullable=False,
+    )
+    actor_account_uuid = sqlalchemy.Column(sqlalchemy.String(36), nullable=False)
+    issued_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=False)
+    expires_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=False)
+    revoked_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=True)
+    last_used_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=True)
+
+    __table_args__ = (
+        sqlalchemy.Index(
+            'ix_support_admin_sessions_workspace_expiry',
+            'workspace_uuid',
+            'expires_at',
+        ),
+        sqlalchemy.CheckConstraint(
+            'length(grant_jti_hash) = 64',
+            name='ck_support_admin_sessions_grant_jti_hash',
+        ),
+    )

src/langbot/pkg/persistence/alembic/versions/0016_space_launch_replay.py

@@ -1,57 +0,0 @@
-"""add durable replay protection for signed Space launch assertions
-
-Revision ID: 0016_space_launch_replay
-Revises: 0015_cloud_core_collab
-Create Date: 2026-07-31
-"""
-
-from __future__ import annotations
-
-import sqlalchemy as sa
-from alembic import op
-
-revision = '0016_space_launch_replay'
-down_revision = '0015_cloud_core_collab'
-branch_labels = None
-depends_on = None
-
-_TABLE = 'space_launch_assertion_consumptions'
-_POLICY = 'langbot_directory_projection'
-_SETTING = "NULLIF(current_setting('langbot.directory_instance_uuid', true), '')"
-
-
-def upgrade() -> None:
-    conn = op.get_bind()
-    if _TABLE not in set(sa.inspect(conn).get_table_names()):
-        op.create_table(
-            _TABLE,
-            sa.Column('instance_uuid', sa.String(255), nullable=False),
-            sa.Column('jti', sa.String(255), nullable=False),
-            sa.Column('expires_at', sa.DateTime(timezone=True), nullable=False),
-            sa.Column('consumed_at', sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
-            sa.PrimaryKeyConstraint('instance_uuid', 'jti'),
-        )
-        op.create_index(
-            'ix_space_launch_assertion_consumptions_expiry',
-            _TABLE,
-            ['instance_uuid', 'expires_at'],
-            unique=False,
-        )
-    if conn.dialect.name == 'postgresql':
-        table = conn.dialect.identifier_preparer.quote(_TABLE)
-        policy = conn.dialect.identifier_preparer.quote(_POLICY)
-        expression = f'instance_uuid::text = {_SETTING}'
-        op.execute(sa.text(f'ALTER TABLE {table} ENABLE ROW LEVEL SECURITY'))
-        op.execute(sa.text(f'ALTER TABLE {table} FORCE ROW LEVEL SECURITY'))
-        op.execute(sa.text(f'DROP POLICY IF EXISTS {policy} ON {table}'))
-        op.execute(
-            sa.text(
-                f'CREATE POLICY {policy} ON {table} AS PERMISSIVE FOR ALL TO PUBLIC '
-                f'USING ({expression}) WITH CHECK ({expression})'
-            )
-        )
-
-
-def downgrade() -> None:
-    if _TABLE in set(sa.inspect(op.get_bind()).get_table_names()):
-        op.drop_table(_TABLE)

src/langbot/pkg/persistence/alembic/versions/0016_support_admin_sessions.py

@@ -0,0 +1,88 @@
+"""add temporary support-admin sessions
+
+Revision ID: 0016_support_admin_sessions
+Revises: 0015_cloud_core_collab
+Create Date: 2026-07-31
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+
+revision = '0016_support_admin_sessions'
+down_revision = '0015_cloud_core_collab'
+branch_labels = None
+depends_on = None
+
+
+_TABLE_NAME = 'support_admin_temporary_sessions'
+_POLICY_NAME = 'langbot_workspace_isolation'
+_TENANT_SETTING = 'langbot.workspace_uuid'
+
+
+def _setting(name: str) -> str:
+    return f"NULLIF(current_setting('{name}', true), '')"
+
+
+def _quote(conn: sa.Connection, identifier: str) -> str:
+    return conn.dialect.identifier_preparer.quote(identifier)
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    existing_tables = set(sa.inspect(conn).get_table_names())
+    if _TABLE_NAME not in existing_tables:
+        op.create_table(
+            _TABLE_NAME,
+            sa.Column('grant_jti_hash', sa.String(64), nullable=False),
+            sa.Column(
+                'workspace_uuid',
+                sa.String(36),
+                sa.ForeignKey('workspaces.uuid', ondelete='CASCADE'),
+                nullable=False,
+            ),
+            sa.Column('actor_account_uuid', sa.String(36), nullable=False),
+            sa.Column('issued_at', sa.DateTime(), nullable=False),
+            sa.Column('expires_at', sa.DateTime(), nullable=False),
+            sa.Column('revoked_at', sa.DateTime(), nullable=True),
+            sa.Column('last_used_at', sa.DateTime(), nullable=True),
+            sa.CheckConstraint(
+                'length(grant_jti_hash) = 64',
+                name='ck_support_admin_sessions_grant_jti_hash',
+            ),
+            sa.PrimaryKeyConstraint('grant_jti_hash'),
+        )
+        op.create_index(
+            'ix_support_admin_sessions_workspace_expiry',
+            _TABLE_NAME,
+            ['workspace_uuid', 'expires_at'],
+            unique=False,
+        )
+
+    if conn.dialect.name != 'postgresql':
+        return
+
+    table = _quote(conn, _TABLE_NAME)
+    policy = _quote(conn, _POLICY_NAME)
+    expression = f'workspace_uuid::text = {_setting(_TENANT_SETTING)}'
+    op.execute(sa.text(f'ALTER TABLE {table} ENABLE ROW LEVEL SECURITY'))
+    op.execute(sa.text(f'ALTER TABLE {table} FORCE ROW LEVEL SECURITY'))
+    op.execute(sa.text(f'DROP POLICY IF EXISTS {policy} ON {table}'))
+    op.execute(
+        sa.text(
+            f'CREATE POLICY {policy} ON {table} AS PERMISSIVE FOR ALL TO PUBLIC '
+            f'USING ({expression}) WITH CHECK ({expression})'
+        )
+    )
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+    if conn.dialect.name == 'postgresql':
+        table = _quote(conn, _TABLE_NAME)
+        policy = _quote(conn, _POLICY_NAME)
+        op.execute(sa.text(f'DROP POLICY IF EXISTS {policy} ON {table}'))
+    op.drop_index('ix_support_admin_sessions_workspace_expiry', table_name=_TABLE_NAME)
+    op.drop_table(_TABLE_NAME)

src/langbot/pkg/persistence/mgr.py

@@ -54,6 +54,7 @@ _ALEMBIC_TENANT_TABLES = {
     'workspace_memberships',
     'workspace_invitations',
     'workspace_execution_states',
+    'support_admin_temporary_sessions',
     'workspace_metadata',
     'api_keys',
     'bots',

src/langbot/pkg/persistence/tenant_uow.py

@@ -43,6 +43,7 @@ TENANT_TABLE_COLUMNS: dict[str, str] = {
     'workspace_memberships': 'workspace_uuid',
     'workspace_invitations': 'workspace_uuid',
     'workspace_execution_states': 'workspace_uuid',
+    'support_admin_temporary_sessions': 'workspace_uuid',
     'workspace_metadata': 'workspace_uuid',
     'api_keys': 'workspace_uuid',
     'bots': 'workspace_uuid',
@@ -75,7 +76,6 @@ TENANT_TABLE_COLUMNS: dict[str, str] = {
 DIRECTORY_PROJECTION_TABLE_COLUMNS: dict[str, str] = {
     'directory_projection_states': 'instance_uuid',
     'directory_projection_inbox': 'instance_uuid',
-    'space_launch_assertion_consumptions': 'instance_uuid',
 }
 
 DIRECTORY_PROJECTED_TENANT_TABLES = frozenset(

src/langbot/pkg/platform/sources/websocket_manager.py

@@ -9,7 +9,7 @@ from datetime import datetime
 
 import pydantic
 
-from ...api.http.context import ExecutionContext
+from ...api.http.context import ExecutionContext, PrincipalContext
 
 logger = logging.getLogger(__name__)
 _SESSION_FILTER_UNSET = object()
@@ -95,6 +95,9 @@ class WebSocketConnection(pydantic.BaseModel):
     metadata: dict = pydantic.Field(default_factory=dict)
     """连接元数据（可存储额外信息）"""
 
+    trigger_principal: PrincipalContext | None = None
+    """Authenticated principal that opened this dashboard connection."""
+
     @property
     def scope(self) -> WebSocketScope:
         return WebSocketScope(
@@ -112,6 +115,7 @@ class WebSocketConnection(pydantic.BaseModel):
             workspace_uuid=self.workspace_uuid,
             placement_generation=self.placement_generation,
             pipeline_uuid=self.pipeline_uuid,
+            trigger_principal=self.trigger_principal,
         )
 
 
@@ -138,6 +142,7 @@ class WebSocketConnectionManager:
         pipeline_uuid: str,
         session_type: str,
         metadata: dict | None = None,
+        trigger_principal: PrincipalContext | None = None,
         session_id: str | None = None,
         send_queue_size: int = _DEFAULT_SEND_QUEUE_SIZE,
         max_connections: int = 1024,
@@ -174,6 +179,7 @@ class WebSocketConnectionManager:
                 session_id=session_id,
                 websocket=websocket,
                 metadata=metadata or {},
+                trigger_principal=trigger_principal,
                 send_queue=asyncio.Queue(maxsize=send_queue_size),
             )
 

src/langbot/pkg/plugin/connector.py

@@ -19,11 +19,6 @@ from urllib.parse import urljoin, urlparse
 from langbot_plugin.api.entities.builtin.pipeline.query import provider_session
 
 from ..core import app
-from ..cloud.quotas import (
-    lock_workspace_for_quota,
-    require_resource_capacity,
-    resolve_workspace_quota,
-)
 from . import handler
 from .archive import inspect_plugin_archive_metadata
 from .github import (
@@ -1300,11 +1295,6 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
         install_info: dict[str, Any],
         artifact_digest: str,
     ) -> tuple[InstallationBinding, str | None, bool]:
-        quota = await resolve_workspace_quota(
-            self.ap,
-            execution_context.workspace_uuid,
-            'plugins.max',
-        )
         safe_install_info = {
             key: value
             for key, value in install_info.items()
@@ -1326,19 +1316,9 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
         )
 
         async def persist(execute):
-            if quota.requires_transaction_lock:
-                await lock_workspace_for_quota(execute, execution_context.workspace_uuid)
             result = await execute(statement)
             setting = result.first()
             if setting is None:
-                await require_resource_capacity(
-                    execute,
-                    workspace_uuid=execution_context.workspace_uuid,
-                    model=persistence_plugin.PluginSetting,
-                    quota=quota,
-                    resource_name='plugins',
-                    workspace_locked=quota.requires_transaction_lock,
-                )
                 installation_uuid = str(uuid.uuid4())
                 runtime_revision = 1
                 previous_digest = None
@@ -1393,8 +1373,6 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
             )
 
         tenant_uow = getattr(self.ap.persistence_mgr, 'tenant_uow', None)
-        if quota.requires_transaction_lock and not callable(tenant_uow):
-            raise RuntimeError('Cloud plugin quota enforcement requires transactional persistence')
         if callable(tenant_uow):
             async with tenant_uow(execution_context.workspace_uuid) as uow:
                 return await persist(uow.execute)

tests/integration/api/test_support_admin_launch.py

@@ -0,0 +1,453 @@
+from __future__ import annotations
+
+import base64
+import datetime
+import json
+import logging
+import time
+import uuid
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+import sqlalchemy
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from quart import Quart
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+from langbot.pkg.api.http.authz import Permission
+from langbot.pkg.api.http.context import PrincipalType, RequestContext
+from langbot.pkg.api.http.controller import group
+from langbot.pkg.api.http.controller.groups.pipelines.websocket_chat import WebSocketChatRouterGroup
+from langbot.pkg.api.http.controller.groups.user import UserRouterGroup
+from langbot.pkg.cloud.launch import SpaceLaunchError, SpaceLaunchService
+from langbot.pkg.cloud.support_admin import SupportAdminSessionService
+from langbot.pkg.entity.persistence.base import Base
+from langbot.pkg.entity.persistence.support_admin import SupportAdminTemporarySession
+from langbot.pkg.entity.persistence.user import User
+from langbot.pkg.entity.persistence.workspace import (
+    Workspace,
+    WorkspaceExecutionState,
+    WorkspaceMembership,
+)
+from langbot.pkg.workspace.service import WorkspaceService
+
+
+pytestmark = [pytest.mark.integration, pytest.mark.asyncio]
+
+
+INSTANCE_UUID = 'instance-support-admin'
+WORKSPACE_UUID = '10000000-0000-4000-8000-000000000001'
+OTHER_WORKSPACE_UUID = '10000000-0000-4000-8000-000000000002'
+ACTOR_ACCOUNT_UUID = '20000000-0000-4000-8000-000000000001'
+KEY_ID = 'support-admin-key-1'
+
+
+def _base64url(raw: bytes) -> str:
+    return base64.urlsafe_b64encode(raw).rstrip(b'=').decode('ascii')
+
+
+def _sign(private_key: Ed25519PrivateKey, claims: dict, *, key_id: str = KEY_ID) -> str:
+    header = {'alg': 'EdDSA', 'kid': key_id, 'typ': 'langbot-control-plane+jwt'}
+    encoded_header = _base64url(json.dumps(header, separators=(',', ':')).encode('utf-8'))
+    encoded_claims = _base64url(json.dumps(claims, separators=(',', ':')).encode('utf-8'))
+    signing_input = f'{encoded_header}.{encoded_claims}'
+    return f'{signing_input}.{_base64url(private_key.sign(signing_input.encode("ascii")))}'
+
+
+def _admin_claims(*, now: int, jti: str | None = None, workspace_uuid: str = WORKSPACE_UUID) -> dict:
+    return {
+        'iss': 'langbot-space',
+        'aud': 'langbot-cloud-runtime',
+        'sub': f'langbot-instance:{INSTANCE_UUID}',
+        'jti': jti or str(uuid.uuid4()),
+        'iat': now,
+        'nbf': now - 5,
+        'exp': now + 90,
+        'instance_uuid': INSTANCE_UUID,
+        'kind': 'workspace.support_admin_launch',
+        'payload': {
+            'workspace_uuid': workspace_uuid,
+            'launch_mode': 'support_admin',
+            'principal_type': 'support_admin',
+            'actor_account_uuid': ACTOR_ACCOUNT_UUID,
+            'effective_role': 'owner',
+        },
+    }
+
+
+@group.group_class('support_admin_probe', '/api/v1/support-admin-probe')
+class SupportAdminProbeGroup(group.RouterGroup):
+    async def initialize(self) -> None:
+        @self.route('/user-token', auth_type=group.AuthType.USER_TOKEN, permission=Permission.WORKSPACE_VIEW)
+        async def _(request_context: RequestContext) -> str:
+            return self.success(data=_context_payload(request_context))
+
+        @self.route(
+            '/member-operation',
+            auth_type=group.AuthType.USER_TOKEN,
+            permission=Permission.MEMBER_VIEW,
+        )
+        async def member_operation(request_context: RequestContext) -> str:
+            return self.success(data=_context_payload(request_context))
+
+        @self.route(
+            '/user-token-or-api-key',
+            auth_type=group.AuthType.USER_TOKEN_OR_API_KEY,
+            permission=Permission.WORKSPACE_VIEW,
+        )
+        async def _(request_context: RequestContext) -> str:
+            return self.success(data=_context_payload(request_context))
+
+
+def _context_payload(request_context: RequestContext) -> dict:
+    return {
+        'principal_type': request_context.principal.principal_type.value,
+        'actor_account_uuid': request_context.principal.actor_account_uuid,
+        'account_uuid': request_context.principal.account_uuid,
+        'role': request_context.workspace.role,
+        'membership_uuid': request_context.workspace.membership_uuid,
+        'permissions': sorted(request_context.workspace.permissions),
+    }
+
+
+class _TenantUow:
+    def __init__(self, engine):
+        self._engine = engine
+        self.session = None
+        self._transaction = None
+
+    async def __aenter__(self):
+        session_factory = async_sessionmaker(self._engine, expire_on_commit=False)
+        self.session = session_factory()
+        self._transaction = await self.session.begin()
+        return self
+
+    async def __aexit__(self, exc_type, exc, traceback):
+        try:
+            if exc_type is None:
+                await self._transaction.commit()
+            else:
+                await self._transaction.rollback()
+        finally:
+            await self.session.close()
+
+
+class _TenantScope:
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc, traceback):
+        return False
+
+
+class _PersistenceManager:
+    def __init__(self, engine):
+        self._engine = engine
+        self.mode = SimpleNamespace(value='oss_compat')
+
+    def get_db_engine(self):
+        return self._engine
+
+    def tenant_uow(self, workspace_uuid: str):
+        del workspace_uuid
+        return _TenantUow(self._engine)
+
+    def tenant_scope(self, workspace_uuid: str):
+        del workspace_uuid
+        return _TenantScope()
+
+
+@pytest.fixture
+async def support_admin_api(tmp_path):
+    private_key = Ed25519PrivateKey.generate()
+    public_key = private_key.public_key().public_bytes(
+        encoding=serialization.Encoding.Raw,
+        format=serialization.PublicFormat.Raw,
+    )
+    engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "support-admin.db"}')
+    async with engine.begin() as connection:
+        await connection.run_sync(
+            Base.metadata.create_all,
+            tables=[
+                User.__table__,
+                Workspace.__table__,
+                WorkspaceExecutionState.__table__,
+                WorkspaceMembership.__table__,
+                SupportAdminTemporarySession.__table__,
+            ],
+        )
+        for workspace_uuid, slug in (
+            (WORKSPACE_UUID, 'support-admin-a'),
+            (OTHER_WORKSPACE_UUID, 'support-admin-b'),
+        ):
+            await connection.execute(
+                sqlalchemy.insert(Workspace).values(
+                    uuid=workspace_uuid,
+                    instance_uuid=INSTANCE_UUID,
+                    name=slug,
+                    slug=slug,
+                    type='team',
+                    status='active',
+                    source='cloud_projection',
+                    projection_revision=1,
+                )
+            )
+            await connection.execute(
+                sqlalchemy.insert(WorkspaceExecutionState).values(
+                    workspace_uuid=workspace_uuid,
+                    instance_uuid=INSTANCE_UUID,
+                    active_generation=1,
+                    state='active',
+                    write_fenced=False,
+                    source='cloud',
+                    desired_state_revision=1,
+                )
+            )
+
+    app = SimpleNamespace()
+    app.persistence_mgr = _PersistenceManager(engine)
+    app.instance_config = SimpleNamespace(
+        data={
+            'system': {
+                'jwt': {'secret': 'support-admin-secret', 'expire': 3600},
+                'websocket_retention': {},
+            },
+            'space': {
+                'launch': {
+                    'control_plane_public_key': _base64url(public_key),
+                }
+            },
+            'api': {'global_api_key': ''},
+        }
+    )
+    app.logger = logging.getLogger('support-admin-test')
+    app.deployment = SimpleNamespace(mode='cloud', multi_workspace_enabled=True, verification_key_id=KEY_ID)
+    app.directory_projection_service = SimpleNamespace(require_ready=lambda: None)
+    app.workspace_service = WorkspaceService(app, instance_uuid=INSTANCE_UUID)
+    app.entitlement_resolver = SimpleNamespace(
+        instance_uuid=INSTANCE_UUID,
+        resolve=AsyncMock(return_value=SimpleNamespace(entitlement_revision=7)),
+    )
+    app.support_admin_session_service = SupportAdminSessionService(app)
+    app.space_launch_service = SpaceLaunchService(app)
+    app.user_service = SimpleNamespace()
+    app.user_service.get_authenticated_account = AsyncMock(side_effect=AssertionError('normal account auth used'))
+    app.user_service.verify_jwt_token = AsyncMock(side_effect=AssertionError('normal token verification used'))
+    app.user_service.get_user_by_email = AsyncMock(side_effect=AssertionError('user lookup used'))
+    app.apikey_service = SimpleNamespace()
+    app.apikey_service.authenticate_api_key = AsyncMock(
+        return_value=SimpleNamespace(
+            instance_uuid=INSTANCE_UUID,
+            workspace_uuid=OTHER_WORKSPACE_UUID,
+            placement_generation=1,
+            api_key_uuid='api-key',
+            permissions=frozenset(permission.value for permission in Permission),
+        )
+    )
+
+    quart_app = Quart(__name__)
+    await UserRouterGroup(app, quart_app).initialize()
+    await SupportAdminProbeGroup(app, quart_app).initialize()
+
+    yield app, quart_app.test_client(), engine, private_key
+    await engine.dispose()
+
+
+async def _issue_support_token(app, private_key: Ed25519PrivateKey, *, jti: str | None = None) -> dict[str, str]:
+    launch = await app.space_launch_service.consume_assertion(
+        _sign(private_key, _admin_claims(now=int(time.time()), jti=jti)),
+        expected_workspace_uuid=WORKSPACE_UUID,
+    )
+    return launch
+
+
+def _auth(token: str, workspace_uuid: str = WORKSPACE_UUID) -> dict[str, str]:
+    return {'Authorization': f'Bearer {token}', 'X-Workspace-Id': workspace_uuid}
+
+
+async def test_support_admin_membership_only_routes_are_denied(support_admin_api):
+    app, client, _engine, private_key = support_admin_api
+    launch = await _issue_support_token(app, private_key)
+
+    response = await client.get(
+        '/api/v1/support-admin-probe/member-operation',
+        headers=_auth(launch['support_admin_token']),
+    )
+
+    assert response.status_code == 403
+    assert (await response.get_json())['code'] == 'permission_denied'
+
+
+async def test_support_admin_check_token_is_rejected(support_admin_api):
+    app, client, _engine, private_key = support_admin_api
+    launch = await _issue_support_token(app, private_key)
+
+    response = await client.get('/api/v1/user/check-token', headers=_auth(launch['support_admin_token']))
+
+    assert response.status_code == 401
+    assert (await response.get_json())['code'] == 'invalid_authentication'
+
+
+async def test_support_admin_cross_workspace_denied_for_user_token_and_or_api_key(support_admin_api):
+    app, client, _engine, private_key = support_admin_api
+    launch = await _issue_support_token(app, private_key)
+
+    missing_selector = await client.get(
+        '/api/v1/support-admin-probe/user-token',
+        headers={'Authorization': f'Bearer {launch["support_admin_token"]}'},
+    )
+    user_response = await client.get(
+        '/api/v1/support-admin-probe/user-token',
+        headers=_auth(launch['support_admin_token'], OTHER_WORKSPACE_UUID),
+    )
+    either_response = await client.get(
+        '/api/v1/support-admin-probe/user-token-or-api-key',
+        headers={
+            **_auth(launch['support_admin_token'], OTHER_WORKSPACE_UUID),
+            'X-API-Key': 'valid-api-key',
+        },
+    )
+
+    assert missing_selector.status_code == 400
+    assert user_response.status_code == 401
+    assert either_response.status_code == 401
+    app.apikey_service.authenticate_api_key.assert_not_awaited()
+
+
+async def test_support_admin_request_context_has_actor_owner_and_no_membership(support_admin_api):
+    app, client, engine, private_key = support_admin_api
+    before_count = await _membership_count(engine)
+    launch = await _issue_support_token(app, private_key)
+
+    response = await client.get(
+        '/api/v1/support-admin-probe/user-token',
+        headers=_auth(launch['support_admin_token']),
+    )
+
+    assert response.status_code == 200
+    data = (await response.get_json())['data']
+    permissions = set(data.pop('permissions'))
+    assert Permission.WORKSPACE_VIEW.value in permissions
+    assert Permission.RESOURCE_MANAGE.value in permissions
+    assert not permissions.intersection(
+        {
+            Permission.OWNER_TRANSFER.value,
+            Permission.MEMBER_VIEW.value,
+            Permission.MEMBER_INVITE.value,
+            Permission.MEMBER_UPDATE_ROLE.value,
+            Permission.MEMBER_REMOVE.value,
+        }
+    )
+    assert data == {
+        'principal_type': PrincipalType.SUPPORT_ADMIN.value,
+        'actor_account_uuid': ACTOR_ACCOUNT_UUID,
+        'account_uuid': None,
+        'role': 'owner',
+        'membership_uuid': None,
+    }
+    assert await _membership_count(engine) == before_count
+
+
+async def test_support_admin_missing_workspace_is_controlled_launch_failure(support_admin_api):
+    app, _client, engine, private_key = support_admin_api
+    async with engine.begin() as connection:
+        await connection.execute(
+            sqlalchemy.delete(WorkspaceExecutionState).where(WorkspaceExecutionState.workspace_uuid == WORKSPACE_UUID)
+        )
+
+    with pytest.raises(SpaceLaunchError, match='unavailable'):
+        await _issue_support_token(app, private_key)
+
+
+async def test_support_admin_launch_replay_is_durable_across_service_instances(support_admin_api):
+    app, _client, _engine, private_key = support_admin_api
+    jti = str(uuid.uuid4())
+
+    await _issue_support_token(app, private_key, jti=jti)
+    second_service = SpaceLaunchService(app)
+
+    with pytest.raises(SpaceLaunchError, match='already been consumed'):
+        await second_service.consume_assertion(
+            _sign(private_key, _admin_claims(now=int(time.time()), jti=jti)),
+            expected_workspace_uuid=WORKSPACE_UUID,
+        )
+
+
+async def test_support_admin_persisted_expiry_and_revocation_are_enforced(support_admin_api):
+    app, client, engine, private_key = support_admin_api
+    launch = await _issue_support_token(app, private_key)
+    token = launch['support_admin_token']
+
+    async with engine.begin() as connection:
+        await connection.execute(
+            sqlalchemy.update(SupportAdminTemporarySession)
+            .where(SupportAdminTemporarySession.grant_jti_hash == launch['grant_jti_hash'])
+            .values(expires_at=datetime.datetime.now(datetime.UTC).replace(tzinfo=None) - datetime.timedelta(minutes=1))
+        )
+    expired = await client.get('/api/v1/support-admin-probe/user-token', headers=_auth(token))
+    assert expired.status_code == 401
+
+    second = await _issue_support_token(app, private_key)
+    async with engine.begin() as connection:
+        await connection.execute(
+            sqlalchemy.update(SupportAdminTemporarySession)
+            .where(SupportAdminTemporarySession.grant_jti_hash == second['grant_jti_hash'])
+            .values(revoked_at=datetime.datetime.now(datetime.UTC).replace(tzinfo=None))
+        )
+    revoked = await client.get('/api/v1/support-admin-probe/user-token', headers=_auth(second['support_admin_token']))
+    assert revoked.status_code == 401
+
+
+async def test_support_admin_websocket_preserves_actor_and_revalidates(support_admin_api):
+    app, _client, _engine, private_key = support_admin_api
+    launch = await _issue_support_token(app, private_key)
+    captured_contexts = []
+
+    class Adapter:
+        async def handle_websocket_message(self, connection, data):
+            del data
+            captured_contexts.append(connection.execution_context)
+            await connection.send_queue.put({'type': 'handled'})
+            connection.is_active = False
+
+    app.pipeline_service = SimpleNamespace(get_pipeline=AsyncMock(return_value=SimpleNamespace(uuid='pipeline-1')))
+    app.platform_mgr = SimpleNamespace(
+        get_websocket_proxy_bot=AsyncMock(return_value=SimpleNamespace(adapter=Adapter()))
+    )
+
+    quart_app = Quart(__name__)
+    await WebSocketChatRouterGroup(app, quart_app).initialize()
+
+    async with quart_app.test_client().websocket('/api/v1/pipelines/pipeline-1/ws/connect') as websocket:
+        await websocket.send(
+            json.dumps(
+                {
+                    'type': 'authenticate',
+                    'token': launch['support_admin_token'],
+                    'workspace_uuid': WORKSPACE_UUID,
+                }
+            )
+        )
+        connected = json.loads(await websocket.receive())
+        assert connected['type'] == 'connected'
+        await websocket.send(json.dumps({'type': 'message', 'message': [{'type': 'text', 'text': 'hi'}]}))
+        handled = json.loads(await websocket.receive())
+        assert handled['type'] == 'handled'
+
+    assert captured_contexts
+    principal = captured_contexts[0].trigger_principal
+    assert principal is not None
+    assert principal.principal_type == PrincipalType.SUPPORT_ADMIN
+    assert principal.actor_account_uuid == ACTOR_ACCOUNT_UUID
+
+
+async def _membership_count(engine) -> int:
+    async with engine.connect() as connection:
+        return int(
+            await connection.scalar(
+                sqlalchemy.select(sqlalchemy.func.count()).select_from(WorkspaceMembership),
+            )
+            or 0
+        )

tests/integration/api/test_user_space_oauth.py

@@ -270,7 +270,7 @@ async def test_space_credits_are_resolved_from_workspace_owner(space_oauth_api):
 
     response = await client.get(
         '/api/v1/user/space-credits',
-        headers={'Authorization': 'Bearer account-token', 'X-Workspace-UUID': WORKSPACE_UUID},
+        headers={'Authorization': 'Bearer account-token', 'X-Workspace-Id': WORKSPACE_UUID},
     )
 
     assert response.status_code == 200

tests/integration/persistence/test_workspace_quota_postgres.py

@@ -1,137 +0,0 @@
-"""PostgreSQL integration coverage for durable workspace quota locking.
-
-Run with TEST_POSTGRES_URL=postgresql+asyncpg://... pytest ...
-"""
-
-from __future__ import annotations
-
-import asyncio
-import os
-import uuid
-
-import pytest
-import sqlalchemy as sa
-from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
-from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
-
-from langbot.pkg.cloud import quotas as quota_module
-from langbot.pkg.cloud.quotas import WorkspaceQuota, WorkspaceQuotaExceededError, require_resource_capacity
-
-
-pytestmark = [pytest.mark.integration, pytest.mark.slow, pytest.mark.asyncio]
-
-
-class _Base(DeclarativeBase):
-    pass
-
-
-class _Workspace(_Base):
-    __tablename__ = 'quota_integration_workspaces'
-
-    uuid: Mapped[str] = mapped_column(sa.String(36), primary_key=True)
-
-
-class _Resource(_Base):
-    __tablename__ = 'quota_integration_resources'
-
-    uuid: Mapped[str] = mapped_column(sa.String(36), primary_key=True)
-    workspace_uuid: Mapped[str] = mapped_column(
-        sa.String(36),
-        sa.ForeignKey('quota_integration_workspaces.uuid', ondelete='CASCADE'),
-        nullable=False,
-        index=True,
-    )
-
-
-@pytest.fixture
-async def quota_postgres(monkeypatch):
-    url = os.environ.get('TEST_POSTGRES_URL')
-    if not url:
-        pytest.skip('TEST_POSTGRES_URL not set')
-    if url.startswith('postgresql://'):
-        url = url.replace('postgresql://', 'postgresql+asyncpg://', 1)
-
-    engine = create_async_engine(url, pool_size=5, max_overflow=0)
-    monkeypatch.setattr(quota_module.persistence_workspace, 'Workspace', _Workspace)
-    async with engine.begin() as connection:
-        await connection.run_sync(_Base.metadata.drop_all)
-        await connection.run_sync(_Base.metadata.create_all)
-    try:
-        yield url, engine
-    finally:
-        async with engine.begin() as connection:
-            await connection.run_sync(_Base.metadata.drop_all)
-        await engine.dispose()
-
-
-async def test_workspace_row_lock_is_atomic_isolated_and_survives_pool_restart(quota_postgres) -> None:
-    url, engine = quota_postgres
-    workspace_a = str(uuid.uuid4())
-    workspace_b = str(uuid.uuid4())
-    quota = WorkspaceQuota(limit=1, requires_transaction_lock=True)
-    sessions = async_sessionmaker(engine, expire_on_commit=False)
-
-    async with engine.begin() as connection:
-        await connection.execute(sa.insert(_Workspace), [{'uuid': workspace_a}, {'uuid': workspace_b}])
-
-    lock_acquired = asyncio.Event()
-    release_first = asyncio.Event()
-
-    async def admit(workspace_uuid: str, *, hold: bool = False) -> None:
-        async with sessions() as session:
-            async with session.begin():
-                await require_resource_capacity(
-                    session.execute,
-                    workspace_uuid=workspace_uuid,
-                    model=_Resource,
-                    quota=quota,
-                    resource_name='resources',
-                )
-                if hold:
-                    lock_acquired.set()
-                    await release_first.wait()
-                await session.execute(
-                    sa.insert(_Resource).values(uuid=str(uuid.uuid4()), workspace_uuid=workspace_uuid)
-                )
-
-    first = asyncio.create_task(admit(workspace_a, hold=True))
-    await asyncio.wait_for(lock_acquired.wait(), timeout=2)
-    same_workspace = asyncio.create_task(admit(workspace_a))
-    other_workspace = asyncio.create_task(admit(workspace_b))
-
-    await asyncio.wait_for(other_workspace, timeout=2)
-    assert not same_workspace.done(), 'same-workspace transaction bypassed SELECT FOR UPDATE'
-
-    release_first.set()
-    await first
-    with pytest.raises(WorkspaceQuotaExceededError, match=r'Maximum number of resources \(1\) reached'):
-        await same_workspace
-
-    async with sessions() as session:
-        counts = dict(
-            (
-                await session.execute(
-                    sa.select(_Resource.workspace_uuid, sa.func.count())
-                    .group_by(_Resource.workspace_uuid)
-                    .order_by(_Resource.workspace_uuid)
-                )
-            ).all()
-        )
-    assert counts == {workspace_a: 1, workspace_b: 1}
-
-    await engine.dispose()
-    restarted_engine = create_async_engine(url, pool_size=2, max_overflow=0)
-    restarted_sessions = async_sessionmaker(restarted_engine, expire_on_commit=False)
-    try:
-        async with restarted_sessions() as session:
-            async with session.begin():
-                with pytest.raises(WorkspaceQuotaExceededError):
-                    await require_resource_capacity(
-                        session.execute,
-                        workspace_uuid=workspace_a,
-                        model=_Resource,
-                        quota=quota,
-                        resource_name='resources',
-                    )
-    finally:
-        await restarted_engine.dispose()

tests/unit_tests/api/http/test_internal_error_responses.py

@@ -9,7 +9,6 @@ import quart
 
 from langbot.pkg.api.http.controller import group
 from langbot.pkg.api.http.controller.groups.webhooks import WebhookRouterGroup
-from langbot.pkg.cloud.quotas import WorkspaceQuotaExceededError
 from langbot.pkg.utils.bounded_executor import (
     BlockingWorkCapacityError,
     current_blocking_work_scope,
@@ -49,16 +48,6 @@ class _BlockingCapacityRouterGroup(group.RouterGroup):
             raise BlockingWorkCapacityError('Workspace blocking executor capacity reached')
 
 
-class _QuotaRouterGroup(group.RouterGroup):
-    name = 'quota-test'
-    path = '/quota-test'
-
-    async def initialize(self) -> None:
-        @self.route('', methods=['POST'], auth_type=group.AuthType.NONE)
-        async def _():
-            raise WorkspaceQuotaExceededError('bots', 2)
-
-
 class _InvalidAccountRouterGroup(group.RouterGroup):
     name = 'invalid-account-test'
     path = '/invalid-account-test'
@@ -141,20 +130,6 @@ async def test_blocking_work_capacity_maps_to_retryable_http_response():
     }
 
 
-async def test_workspace_quota_maps_to_stable_conflict_response():
-    application = SimpleNamespace(logger=Mock())
-    quart_app = quart.Quart(__name__)
-    await _QuotaRouterGroup(application, quart_app).initialize()
-
-    response = await quart_app.test_client().post('/quota-test')
-
-    assert response.status_code == 409
-    assert await response.get_json() == {
-        'code': 'workspace_quota_exceeded',
-        'msg': 'Maximum number of bots (2) reached',
-    }
-
-
 async def test_public_webhook_carries_scope_without_holding_database_session():
     class ScopeOnlyPersistenceManager:
         mode = SimpleNamespace(value='cloud_runtime')

tests/unit_tests/api/service/test_bot_service.py

@@ -311,9 +311,10 @@ class TestBotServiceCreateBot:
         ap.platform_mgr = SimpleNamespace()
         ap.platform_mgr.load_bot = AsyncMock()
 
-        # Mock the atomic count query to report 2 existing bots.
-        mock_result = _create_mock_result()
-        mock_result.scalar_one = Mock(return_value=2)
+        # Mock get_bots to return 2 bots already
+        bot1 = _create_mock_bot(bot_uuid='uuid-1')
+        bot2 = _create_mock_bot(bot_uuid='uuid-2')
+        mock_result = _create_mock_result([bot1, bot2])
         ap.persistence_mgr.execute_async = AsyncMock(return_value=mock_result)
         ap.persistence_mgr.serialize_model = Mock(return_value={'uuid': 'uuid-1', 'name': 'Bot 1'})
 

tests/unit_tests/api/service/test_cloud_bot_quota_concurrency.py

@@ -1,129 +0,0 @@
-from __future__ import annotations
-
-import asyncio
-from collections import defaultdict
-from types import SimpleNamespace
-from unittest.mock import AsyncMock
-
-import pytest
-import sqlalchemy
-
-from langbot.pkg.api.http.service.bot import BotService
-from langbot.pkg.cloud.entitlements import EntitlementResolver, EntitlementSnapshot
-
-
-INSTANCE_UUID = 'cloud-instance'
-WORKSPACE_A = '11111111-1111-1111-1111-111111111111'
-WORKSPACE_B = '22222222-2222-2222-2222-222222222222'
-
-
-class _Provider:
-    async def get_workspace_entitlement(self, workspace_uuid: str) -> EntitlementSnapshot:
-        return EntitlementSnapshot(
-            instance_uuid=INSTANCE_UUID,
-            workspace_uuid=workspace_uuid,
-            entitlement_revision=1,
-            status='active',
-            not_before=0,
-            expires_at=4_102_444_800,
-            features={},
-            limits={'bots.max': 2},
-        )
-
-
-class _Result:
-    def __init__(self, *, first=None, scalar=None) -> None:
-        self._first = first
-        self._scalar = scalar
-
-    def first(self):
-        return self._first
-
-    def scalar_one(self):
-        return self._scalar
-
-
-class _TenantUow:
-    def __init__(self, manager: '_Persistence', workspace_uuid: str) -> None:
-        self.manager = manager
-        self.workspace_uuid = workspace_uuid
-        self.lock = manager.locks[workspace_uuid]
-
-    async def __aenter__(self):
-        await self.lock.acquire()
-        return self
-
-    async def __aexit__(self, exc_type, exc, tb):
-        self.lock.release()
-
-    async def execute(self, statement):
-        sql = str(statement)
-        if isinstance(statement, sqlalchemy.sql.dml.Insert):
-            assert statement.table.name == 'bots'
-            self.manager.bots[self.workspace_uuid].append(statement.compile().params)
-            return _Result()
-        if 'FROM workspaces' in sql:
-            assert statement._for_update_arg is not None
-            self.manager.workspace_locks_seen += 1
-            return _Result(first=(self.workspace_uuid,))
-        if 'count(' in sql.lower() and 'FROM bots' in sql:
-            return _Result(scalar=len(self.manager.bots[self.workspace_uuid]))
-        if 'FROM legacy_pipelines' in sql:
-            return _Result(first=None)
-        raise AssertionError(f'unexpected statement: {sql}')
-
-
-class _Persistence:
-    def __init__(self) -> None:
-        self.locks = defaultdict(asyncio.Lock)
-        self.bots = defaultdict(list)
-        self.workspace_locks_seen = 0
-
-    def tenant_uow(self, workspace_uuid: str) -> _TenantUow:
-        return _TenantUow(self, workspace_uuid)
-
-    async def execute_async(self, statement):
-        assert 'FROM legacy_pipelines' in str(statement)
-        return _Result(first=None)
-
-
-async def _service(manager: _Persistence) -> BotService:
-    resolver = EntitlementResolver(INSTANCE_UUID, _Provider())
-    await resolver.reconcile_active_workspaces({WORKSPACE_A, WORKSPACE_B})
-    ap = SimpleNamespace(
-        entitlement_resolver=resolver,
-        persistence_mgr=manager,
-        instance_config=SimpleNamespace(data={'system': {'limitation': {'max_bots': 99}}}),
-        platform_mgr=SimpleNamespace(load_bot=AsyncMock()),
-    )
-    service = BotService(ap)
-    service.get_bot = AsyncMock(return_value={'uuid': 'created'})
-    return service
-
-
-@pytest.mark.asyncio
-async def test_cloud_bot_quota_is_atomic_isolated_and_persists_across_service_restart() -> None:
-    manager = _Persistence()
-    service = await _service(manager)
-
-    async def create(workspace_uuid: str, index: int):
-        return await service.create_bot(workspace_uuid, {'name': f'bot-{index}'})
-
-    results = await asyncio.gather(
-        *(create(WORKSPACE_A, index) for index in range(8)),
-        *(create(WORKSPACE_B, index) for index in range(8)),
-        return_exceptions=True,
-    )
-
-    successes = [result for result in results if isinstance(result, str)]
-    failures = [result for result in results if isinstance(result, ValueError)]
-    assert len(successes) == 4
-    assert len(failures) == 12
-    assert len(manager.bots[WORKSPACE_A]) == 2
-    assert len(manager.bots[WORKSPACE_B]) == 2
-    assert manager.workspace_locks_seen == 16
-
-    restarted_service = await _service(manager)
-    with pytest.raises(ValueError, match=r'Maximum number of bots \(2\) reached'):
-        await restarted_service.create_bot(WORKSPACE_A, {'name': 'after-restart'})
-    assert len(manager.bots[WORKSPACE_A]) == 2

tests/unit_tests/api/service/test_user_service.py

@@ -418,6 +418,7 @@ class TestUserServiceGenerateJwtToken:
         assert token is not None
 
 
+
 class TestUserServiceVerifyJwtToken:
     """Tests for verify_jwt_token method."""
 

tests/unit_tests/api/test_adapter_session_scoping.py

@@ -149,6 +149,36 @@ async def test_session_scope_matches_exact_tenant_placement_and_principal():
     assert sessions == {}
 
 
+async def test_support_admin_sessions_are_scoped_to_the_persisted_grant():
+    def support_context(grant_jti_hash: str) -> RequestContext:
+        return RequestContext(
+            instance_uuid='instance-test',
+            placement_generation=1,
+            request_id='request-test',
+            auth_type='support-admin',
+            principal=PrincipalContext(
+                principal_type=PrincipalType.SUPPORT_ADMIN,
+                actor_account_uuid='support-actor',
+                support_session_id=grant_jti_hash,
+            ),
+            workspace=WorkspaceContext(
+                workspace_uuid='workspace-a',
+                membership_uuid=None,
+                role='owner',
+                permissions=frozenset({'resource.manage'}),
+            ),
+        )
+
+    first_context = support_context('a' * 64)
+    second_context = support_context('b' * 64)
+    sessions: dict[str, dict] = {'session-test': {'status': 'waiting'}}
+    _bind_session_scope(sessions['session-test'], first_context)
+
+    assert _get_owned_session(sessions, 'session-test', second_context) is None
+    assert _pop_owned_session(sessions, 'session-test', second_context) is None
+    assert _get_owned_session(sessions, 'session-test', first_context) is sessions['session-test']
+
+
 async def test_session_capacity_evicts_oldest_session_in_same_workspace():
     owner_context = _request_context()
     sessions: dict[str, dict] = {}

tests/unit_tests/cloud/test_bootstrap.py

@@ -228,23 +228,10 @@ async def test_cloud_pgvector_contract_is_fail_closed(pgvector_config, message):
         )
 
 
-async def test_cloud_runtime_allows_explicitly_disabled_box():
-    config = _cloud_config()
-    config['box']['enabled'] = False
-
-    deployment = await resolve_deployment(
-        instance_uuid='instance-a',
-        instance_config=config,
-        entry_points=lambda: _EntryPoints([_EntryPoint(_Provider())]),
-        now=1_000,
-    )
-
-    assert isinstance(deployment, VerifiedCloudDeployment)
-
-
 @pytest.mark.parametrize(
     ('mutate', 'message'),
     [
+        (lambda config: config['box'].update(enabled=False), 'box.enabled=true'),
         (lambda config: config['box'].update(backend='docker'), 'box.backend=nsjail'),
         (lambda config: config['box']['runtime'].update(endpoint=''), 'box.runtime.endpoint'),
         (

tests/unit_tests/cloud/test_space_launch.py

@@ -11,6 +11,7 @@ from cryptography.hazmat.primitives import serialization
 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
 
 from langbot.pkg.cloud.launch import SpaceLaunchError, SpaceLaunchService
+from langbot.pkg.cloud.support_admin import SupportAdminReplayError
 
 
 pytestmark = pytest.mark.asyncio
@@ -48,7 +49,6 @@ def _claims(*, now: int, jti: str | None = None, workspace_uuid: str = WORKSPACE
         'payload': {
             'account_uuid': ACCOUNT_UUID,
             'workspace_uuid': workspace_uuid,
-            'return_path': '/',
         },
     }
 
@@ -58,9 +58,21 @@ def _service(private_key: Ed25519PrivateKey, *, now: int) -> SpaceLaunchService:
         encoding=serialization.Encoding.Raw,
         format=serialization.PublicFormat.Raw,
     )
+    consumed: set[str] = set()
+
+    class DurableSupportAdminService:
+        async def consume_launch_grant(self, **kwargs):
+            grant_hash = kwargs['grant_jti_hash']
+            if grant_hash in consumed:
+                raise SupportAdminReplayError('already consumed')
+            consumed.add(grant_hash)
+            return SimpleNamespace(token='support-admin-token')
+
     app = SimpleNamespace(
         deployment=SimpleNamespace(multi_workspace_enabled=True, verification_key_id=KEY_ID),
         workspace_service=SimpleNamespace(instance_uuid=INSTANCE_UUID),
+        logger=SimpleNamespace(info=lambda *args, **kwargs: None),
+        support_admin_session_service=DurableSupportAdminService(),
         instance_config=SimpleNamespace(
             data={
                 'space': {
@@ -82,30 +94,85 @@ async def test_consumes_valid_workspace_launch_assertion_once():
 
     launch = await service.consume_assertion(token, expected_workspace_uuid=WORKSPACE_UUID)
 
-    assert launch == {
-        'account_uuid': ACCOUNT_UUID,
-        'workspace_uuid': WORKSPACE_UUID,
-        'return_path': '/',
-    }
+    assert launch == {'account_uuid': ACCOUNT_UUID, 'workspace_uuid': WORKSPACE_UUID}
     with pytest.raises(SpaceLaunchError, match='already been consumed'):
         await service.consume_assertion(token, expected_workspace_uuid=WORKSPACE_UUID)
 
 
-async def test_consumed_assertion_remains_blocked_through_clock_skew_window():
+
+
+async def test_consumes_admin_owner_launch_once_and_validates_claims():
     private_key = Ed25519PrivateKey.generate()
     now = int(time.time())
     service = _service(private_key, now=now)
     claims = _claims(now=now)
-    claims['iat'] = now - 10
-    claims['nbf'] = now - 10
-    claims['exp'] = now - 1
+    claims['kind'] = 'workspace.support_admin_launch'
+    claims['payload'].update(
+        {
+            'launch_mode': 'support_admin',
+            'principal_type': 'support_admin',
+            'actor_account_uuid': '33333333-3333-4333-8333-333333333333',
+            'effective_role': 'owner',
+        }
+    )
+    claims['payload'].pop('account_uuid')
     token = _sign(private_key, claims)
 
-    await service.consume_assertion(token, expected_workspace_uuid=WORKSPACE_UUID)
+    launch = await service.consume_assertion(token, expected_workspace_uuid=WORKSPACE_UUID)
 
+    assert launch == {
+        'workspace_uuid': WORKSPACE_UUID,
+        'launch_mode': 'support_admin',
+        'actor_account_uuid': '33333333-3333-4333-8333-333333333333',
+        'effective_role': 'owner',
+        'grant_jti_hash': launch['grant_jti_hash'],
+        'support_admin_token': 'support-admin-token',
+    }
     with pytest.raises(SpaceLaunchError, match='already been consumed'):
         await service.consume_assertion(token, expected_workspace_uuid=WORKSPACE_UUID)
 
+    invalid = _claims(now=now)
+    invalid['kind'] = 'workspace.support_admin_launch'
+    invalid['payload'].update(
+        {
+            'launch_mode': 'support_admin',
+            'principal_type': 'support_admin',
+            'actor_account_uuid': '33333333-3333-4333-8333-333333333333',
+            'effective_role': 'member',
+        }
+    )
+    invalid['payload'].pop('account_uuid')
+    with pytest.raises(SpaceLaunchError, match='effective role'):
+        await service.consume_assertion(_sign(private_key, invalid), expected_workspace_uuid=WORKSPACE_UUID)
+
+    too_long = _claims(now=now)
+    too_long['kind'] = 'workspace.support_admin_launch'
+    too_long['exp'] = now + 91
+    too_long['payload'].update(
+        {
+            'launch_mode': 'support_admin',
+            'principal_type': 'support_admin',
+            'actor_account_uuid': '33333333-3333-4333-8333-333333333333',
+            'effective_role': 'owner',
+        }
+    )
+    too_long['payload'].pop('account_uuid')
+    with pytest.raises(SpaceLaunchError, match='lifetime exceeds 90 seconds'):
+        await service.consume_assertion(_sign(private_key, too_long), expected_workspace_uuid=WORKSPACE_UUID)
+
+    impersonating = _claims(now=now)
+    impersonating['kind'] = 'workspace.support_admin_launch'
+    impersonating['payload'].update(
+        {
+            'launch_mode': 'support_admin',
+            'principal_type': 'support_admin',
+            'actor_account_uuid': '33333333-3333-4333-8333-333333333333',
+            'effective_role': 'owner',
+        }
+    )
+    with pytest.raises(SpaceLaunchError, match='customer Account'):
+        await service.consume_assertion(_sign(private_key, impersonating), expected_workspace_uuid=WORKSPACE_UUID)
+
 
 async def test_replay_cache_does_not_scan_all_live_assertions(monkeypatch):
     private_key = Ed25519PrivateKey.generate()
@@ -183,15 +250,3 @@ async def test_rejects_invalid_signature_and_non_cloud_mode():
     oss_service.ap.deployment.multi_workspace_enabled = False
     with pytest.raises(SpaceLaunchError, match='verified Cloud mode'):
         await oss_service.consume_assertion(token, expected_workspace_uuid=WORKSPACE_UUID)
-
-
-@pytest.mark.asyncio
-async def test_rejects_unsafe_signed_return_path() -> None:
-    private_key = Ed25519PrivateKey.generate()
-    now = int(time.time())
-    service = _service(private_key, now=now)
-    claims = _claims(now=now)
-    claims['payload']['return_path'] = '//evil.example'
-
-    with pytest.raises(SpaceLaunchError, match='return path'):
-        await service.consume_assertion(_sign(private_key, claims))

tests/unit_tests/cloud/test_workspace_quotas.py

@@ -1,102 +0,0 @@
-from __future__ import annotations
-
-from types import SimpleNamespace
-from unittest.mock import AsyncMock, Mock
-
-import pytest
-import sqlalchemy
-
-from langbot.pkg.cloud.entitlements import EntitlementResolver, EntitlementSnapshot
-from langbot.pkg.cloud.quotas import (
-    WorkspaceQuota,
-    require_resource_capacity,
-    resolve_workspace_quota,
-)
-from langbot.pkg.entity.persistence.bot import Bot
-
-
-WORKSPACE_UUID = '11111111-1111-1111-1111-111111111111'
-INSTANCE_UUID = 'cloud-instance'
-
-
-class _Provider:
-    def __init__(self, limits: dict[str, int]) -> None:
-        self.limits = limits
-
-    async def get_workspace_entitlement(self, workspace_uuid: str) -> EntitlementSnapshot:
-        return EntitlementSnapshot(
-            instance_uuid=INSTANCE_UUID,
-            workspace_uuid=workspace_uuid,
-            entitlement_revision=1,
-            status='active',
-            not_before=0,
-            expires_at=4_102_444_800,
-            features={},
-            limits=self.limits,
-            plan_name='test',
-        )
-
-
-async def _resolver(limits: dict[str, int]) -> EntitlementResolver:
-    resolver = EntitlementResolver(INSTANCE_UUID, _Provider(limits))
-    await resolver.reconcile_active_workspaces({WORKSPACE_UUID})
-    return resolver
-
-
-@pytest.mark.asyncio
-async def test_resolve_workspace_quota_uses_signed_cloud_limit() -> None:
-    ap = SimpleNamespace(entitlement_resolver=await _resolver({'bots.max': 2}))
-
-    quota = await resolve_workspace_quota(ap, WORKSPACE_UUID, 'bots.max', fallback=99)
-
-    assert quota == WorkspaceQuota(limit=2, requires_transaction_lock=True)
-
-
-@pytest.mark.asyncio
-async def test_resolve_workspace_quota_preserves_oss_fallback() -> None:
-    quota = await resolve_workspace_quota(SimpleNamespace(), WORKSPACE_UUID, 'bots.max', fallback=7)
-
-    assert quota == WorkspaceQuota(limit=7, requires_transaction_lock=False)
-
-
-@pytest.mark.asyncio
-async def test_require_resource_capacity_locks_workspace_before_counting() -> None:
-    statements: list[object] = []
-    lock_result = Mock()
-    lock_result.first.return_value = (WORKSPACE_UUID,)
-    count_result = Mock()
-    count_result.scalar_one.return_value = 1
-    execute = AsyncMock(side_effect=[lock_result, count_result])
-
-    await require_resource_capacity(
-        execute,
-        workspace_uuid=WORKSPACE_UUID,
-        model=Bot,
-        quota=WorkspaceQuota(limit=2, requires_transaction_lock=True),
-        resource_name='bots',
-    )
-
-    statements.extend(call.args[0] for call in execute.await_args_list)
-    assert len(statements) == 2
-    assert isinstance(statements[0], sqlalchemy.sql.Select)
-    assert statements[0]._for_update_arg is not None
-    assert 'workspaces' in str(statements[0])
-    assert 'count' in str(statements[1]).lower()
-
-
-@pytest.mark.asyncio
-async def test_require_resource_capacity_rejects_at_boundary() -> None:
-    lock_result = Mock()
-    lock_result.first.return_value = (WORKSPACE_UUID,)
-    count_result = Mock()
-    count_result.scalar_one.return_value = 2
-    execute = AsyncMock(side_effect=[lock_result, count_result])
-
-    with pytest.raises(ValueError, match=r'Maximum number of bots \(2\) reached'):
-        await require_resource_capacity(
-            execute,
-            workspace_uuid=WORKSPACE_UUID,
-            model=Bot,
-            quota=WorkspaceQuota(limit=2, requires_transaction_lock=True),
-            resource_name='bots',
-        )

tests/unit_tests/core/test_load_config.py

@@ -152,19 +152,6 @@ class TestApplyEnvOverridesToConfig:
 
         assert result['system']['disabled_adapters'] == ['aiocqhttp', 'dingtalk', 'telegram']
 
-    def test_override_integer_list_preserves_item_type(self):
-        """Comma-separated overrides inherit the existing list item type."""
-        load_config = get_load_config_module()
-
-        cfg = {'vdb': {'pgvector': {'allowed_dimensions': [384, 512]}}}
-        env = {'VDB__PGVECTOR__ALLOWED_DIMENSIONS': '384,512,768'}
-
-        with patch.dict(os.environ, env, clear=True):
-            result = load_config._apply_env_overrides_to_config(cfg)
-
-        assert result['vdb']['pgvector']['allowed_dimensions'] == [384, 512, 768]
-        assert all(isinstance(item, int) for item in result['vdb']['pgvector']['allowed_dimensions'])
-
     def test_override_list_value_empty_items(self):
         """Test that empty items in comma-separated list are filtered."""
         load_config = get_load_config_module()

tests/unit_tests/plugin/test_cloud_plugin_quota_concurrency.py

@@ -1,191 +0,0 @@
-from __future__ import annotations
-
-import asyncio
-from collections import defaultdict
-from types import SimpleNamespace
-
-import pytest
-import sqlalchemy
-
-from langbot.pkg.api.http.context import ExecutionContext
-from langbot.pkg.cloud.entitlements import EntitlementResolver, EntitlementSnapshot
-from langbot.pkg.cloud.quotas import WorkspaceQuotaExceededError
-from langbot.pkg.plugin.connector import PluginRuntimeConnector
-from langbot_plugin.runtime.plugin.mgr import PluginInstallSource
-
-
-INSTANCE_UUID = 'cloud-instance'
-WORKSPACE_A = '11111111-1111-1111-1111-111111111111'
-WORKSPACE_B = '22222222-2222-2222-2222-222222222222'
-
-
-class _Provider:
-    async def get_workspace_entitlement(self, workspace_uuid: str) -> EntitlementSnapshot:
-        return EntitlementSnapshot(
-            instance_uuid=INSTANCE_UUID,
-            workspace_uuid=workspace_uuid,
-            entitlement_revision=1,
-            status='active',
-            not_before=0,
-            expires_at=4_102_444_800,
-            features={},
-            limits={'plugins.max': 3},
-        )
-
-
-class _Result:
-    def __init__(self, *, first=None, scalar=None) -> None:
-        self._first = first
-        self._scalar = scalar
-
-    def first(self):
-        return self._first
-
-    def scalar_one(self):
-        return self._scalar
-
-
-class _TenantUow:
-    def __init__(self, manager: '_Persistence', workspace_uuid: str) -> None:
-        self.manager = manager
-        self.workspace_uuid = workspace_uuid
-        self.lock = manager.locks[workspace_uuid]
-
-    async def __aenter__(self):
-        await self.lock.acquire()
-        return self
-
-    async def __aexit__(self, exc_type, exc, tb):
-        self.lock.release()
-
-    async def execute(self, statement):
-        sql = str(statement)
-        params = statement.compile().params
-        if isinstance(statement, sqlalchemy.sql.dml.Insert):
-            assert statement.table.name == 'plugin_settings'
-            key = (params['plugin_author'], params['plugin_name'])
-            self.manager.plugins[self.workspace_uuid][key] = dict(params)
-            return _Result()
-        if isinstance(statement, sqlalchemy.sql.dml.Update):
-            return _Result()
-        if 'FROM workspaces' in sql:
-            assert statement._for_update_arg is not None
-            self.manager.workspace_locks_seen += 1
-            return _Result(first=(self.workspace_uuid,))
-        if 'count(' in sql.lower() and 'FROM plugin_settings' in sql:
-            return _Result(scalar=len(self.manager.plugins[self.workspace_uuid]))
-        if 'FROM plugin_settings' in sql:
-            author = next(value for name, value in params.items() if 'plugin_author' in name)
-            name = next(value for param, value in params.items() if 'plugin_name' in param)
-            row = self.manager.plugins[self.workspace_uuid].get((author, name))
-            if row is None:
-                return _Result(first=None)
-            return _Result(
-                first=SimpleNamespace(
-                    installation_uuid=row['installation_uuid'],
-                    runtime_revision=row['runtime_revision'],
-                    artifact_digest=row['artifact_digest'],
-                    install_info=row['install_info'],
-                )
-            )
-        raise AssertionError(f'unexpected statement: {sql}')
-
-
-class _Persistence:
-    def __init__(self) -> None:
-        self.locks = defaultdict(asyncio.Lock)
-        self.plugins = defaultdict(dict)
-        self.workspace_locks_seen = 0
-
-    def tenant_uow(self, workspace_uuid: str) -> _TenantUow:
-        return _TenantUow(self, workspace_uuid)
-
-
-async def _connector(manager: _Persistence) -> PluginRuntimeConnector:
-    resolver = EntitlementResolver(INSTANCE_UUID, _Provider())
-    await resolver.reconcile_active_workspaces({WORKSPACE_A, WORKSPACE_B})
-    connector = object.__new__(PluginRuntimeConnector)
-    connector.ap = SimpleNamespace(entitlement_resolver=resolver, persistence_mgr=manager)
-    return connector
-
-
-def _context(workspace_uuid: str) -> ExecutionContext:
-    return ExecutionContext(
-        instance_uuid=INSTANCE_UUID,
-        workspace_uuid=workspace_uuid,
-        placement_generation=1,
-        entitlement_revision=1,
-    )
-
-
-@pytest.mark.asyncio
-async def test_cloud_plugin_quota_is_atomic_isolated_and_persists_across_connector_restart() -> None:
-    manager = _Persistence()
-    connector = await _connector(manager)
-
-    async def install(workspace_uuid: str, index: int):
-        return await connector._persist_installation_package(
-            _context(workspace_uuid),
-            plugin_author='test-author',
-            plugin_name=f'plugin-{index}',
-            install_source=PluginInstallSource.MARKETPLACE,
-            install_info={'author': 'test-author', 'name': f'plugin-{index}'},
-            artifact_digest=f'{index:064x}',
-        )
-
-    results = await asyncio.gather(
-        *(install(WORKSPACE_A, index) for index in range(10)),
-        *(install(WORKSPACE_B, index) for index in range(10)),
-        return_exceptions=True,
-    )
-
-    successes = [result for result in results if isinstance(result, tuple)]
-    failures = [result for result in results if isinstance(result, WorkspaceQuotaExceededError)]
-    assert len(successes) == 6
-    assert len(failures) == 14
-    assert len(manager.plugins[WORKSPACE_A]) == 3
-    assert len(manager.plugins[WORKSPACE_B]) == 3
-    assert manager.workspace_locks_seen == 20
-
-    restarted_connector = await _connector(manager)
-    with pytest.raises(WorkspaceQuotaExceededError, match=r'Maximum number of plugins \(3\) reached'):
-        await restarted_connector._persist_installation_package(
-            _context(WORKSPACE_A),
-            plugin_author='test-author',
-            plugin_name='after-restart',
-            install_source=PluginInstallSource.MARKETPLACE,
-            install_info={},
-            artifact_digest='f' * 64,
-        )
-    assert len(manager.plugins[WORKSPACE_A]) == 3
-
-    installed_name = next(iter(manager.plugins[WORKSPACE_A]))[1]
-
-    async def reinstall():
-        return await restarted_connector._persist_installation_package(
-            _context(WORKSPACE_A),
-            plugin_author='test-author',
-            plugin_name=installed_name,
-            install_source=PluginInstallSource.MARKETPLACE,
-            install_info={'author': 'test-author', 'name': installed_name, 'revision': 2},
-            artifact_digest='e' * 64,
-        )
-
-    reinstall_results = await asyncio.gather(reinstall(), reinstall())
-    assert all(result[2] is True for result in reinstall_results)
-
-    mixed_results = await asyncio.gather(
-        reinstall(),
-        restarted_connector._persist_installation_package(
-            _context(WORKSPACE_A),
-            plugin_author='test-author',
-            plugin_name='new-at-capacity',
-            install_source=PluginInstallSource.MARKETPLACE,
-            install_info={},
-            artifact_digest='d' * 64,
-        ),
-        return_exceptions=True,
-    )
-    assert isinstance(mixed_results[0], tuple)
-    assert isinstance(mixed_results[1], WorkspaceQuotaExceededError)
-    assert len(manager.plugins[WORKSPACE_A]) == 3

uv.lock

@@ -2116,7 +2116,7 @@ requires-dist = [
     { name = "ebooklib", specifier = ">=0.18" },
     { name = "gewechat-client", specifier = ">=0.1.5" },
     { name = "html2text", specifier = ">=2024.2.26" },
-    { name = "langbot-plugin", specifier = "==0.5.0" },
+    { name = "langbot-plugin", git = "https://github.com/langbot-app/langbot-plugin-sdk.git?rev=1d65ed301a6afc52150a998043f73cd6032c8162" },
     { name = "langchain", specifier = ">=1.3.9" },
     { name = "langchain-core", specifier = ">=1.3.3" },
     { name = "langchain-text-splitters", specifier = ">=1.1.2" },
@@ -2182,8 +2182,8 @@ dev = [
 
 [[package]]
 name = "langbot-plugin"
-version = "0.5.0"
-source = { registry = "https://pypi.org/simple" }
+version = "0.4.18"
+source = { git = "https://github.com/langbot-app/langbot-plugin-sdk.git?rev=1d65ed301a6afc52150a998043f73cd6032c8162#1d65ed301a6afc52150a998043f73cd6032c8162" }
 dependencies = [
     { name = "aiofiles" },
     { name = "aiohttp" },
@@ -2203,10 +2203,6 @@ dependencies = [
     { name = "watchdog" },
     { name = "websockets" },
 ]
-sdist = { url = "https://files.pythonhosted.org/packages/5c/09/697037dea617a235c9b3df85174badfb44886e3c67149a928049839a959a/langbot_plugin-0.5.0.tar.gz", hash = "sha256:9b81fa0f73cde1fe199a746e03ad891f8ced8f4fa0d05aeb6d78bfe7d755a976", size = 464033, upload-time = "2026-07-30T19:22:57.503Z" }
-wheels = [
-    { url = "https://files.pythonhosted.org/packages/d2/70/93e544c3a120953f4871266db9ef84ea9a58daf01d493fafa7c45674ad36/langbot_plugin-0.5.0-py3-none-any.whl", hash = "sha256:2b078db96d869de55d08304465974f51765e3cfe582ff79b32f09161292fb877", size = 300758, upload-time = "2026-07-30T19:22:56.248Z" },
-]
 
 [[package]]
 name = "langchain"

web/src/app/auth/space/callback/page.tsx

@@ -3,6 +3,7 @@ import { useNavigate, useSearchParams } from 'react-router-dom';
 import { httpClient } from '@/app/infra/http/HttpClient';
 import {
   beginAuthenticatedSession,
+  beginSupportAdminSession,
   bootstrapWorkspaceSession,
   getPendingInvitationToken,
 } from '@/app/infra/http';
@@ -27,9 +28,10 @@ import langbotIcon from '@/app/assets/langbot-logo.webp';
 
 type SpaceOAuthLoginResult = {
   token: string;
-  user: string;
+  user?: string;
   workspace_uuid?: string;
-  return_path?: string;
+  principal_type?: 'account' | 'support_admin';
+  actor_account_uuid?: string;
 };
 
 const pendingSpaceOAuthLogins = new Map<
@@ -64,10 +66,6 @@ function SpaceOAuthCallbackContent() {
   const [searchParams] = useSearchParams();
   const { t } = useTranslation();
   const isMountedRef = useRef(true);
-  const directLaunchFragmentRef = useRef<{
-    workspaceUuid: string | null;
-    launchAssertion: string | null;
-  } | null>(null);
 
   const [status, setStatus] = useState<
     'loading' | 'confirm' | 'success' | 'error'
@@ -99,6 +97,16 @@ function SpaceOAuthCallbackContent() {
           return;
         }
 
+        if (response.principal_type === 'support_admin') {
+          if (!response.workspace_uuid) {
+            throw new Error('Support admin launch did not include a Workspace');
+          }
+          beginSupportAdminSession(response.token, response.workspace_uuid);
+          await bootstrapWorkspaceSession();
+          navigate('/home', { replace: true });
+          return;
+        }
+
         beginAuthenticatedSession(response.token, response.user);
         if (getPendingInvitationToken()) {
           navigate('/invitations/accept', { replace: true });
@@ -111,13 +119,7 @@ function SpaceOAuthCallbackContent() {
           throw new Error('No Workspace is available for this Account');
         }
         if (response.workspace_uuid) {
-          const returnPath =
-            typeof response.return_path === 'string' &&
-            response.return_path.startsWith('/') &&
-            !response.return_path.startsWith('//')
-              ? response.return_path
-              : '/home';
-          navigate(returnPath, { replace: true });
+          navigate('/home', { replace: true });
           return;
         }
         setStatus('success');
@@ -218,29 +220,8 @@ function SpaceOAuthCallbackContent() {
     const errorDescription = searchParams.get('error_description');
     const mode = searchParams.get('mode');
     const state = searchParams.get('state');
-    if (directLaunchFragmentRef.current === null) {
-      const fragmentParams = new URLSearchParams(
-        window.location.hash.startsWith('#')
-          ? window.location.hash.slice(1)
-          : window.location.hash,
-      );
-      directLaunchFragmentRef.current = {
-        workspaceUuid: fragmentParams.get('workspace_uuid'),
-        launchAssertion: fragmentParams.get('launch_assertion'),
-      };
-      if (window.location.hash) {
-        window.history.replaceState(
-          null,
-          '',
-          `${window.location.pathname}${window.location.search}`,
-        );
-      }
-    }
-    const workspaceUuid =
-      directLaunchFragmentRef.current.workspaceUuid ??
-      searchParams.get('workspace_uuid');
-    const launchAssertion =
-      directLaunchFragmentRef.current.launchAssertion;
+    const workspaceUuid = searchParams.get('workspace_uuid');
+    const launchAssertion = searchParams.get('launch_assertion');
 
     if (error) {
       setStatus('error');

web/src/app/home/layout.tsx

@@ -17,6 +17,7 @@ import {
   bootstrapWorkspaceSession,
   systemInfo,
   initializeSystemInfo,
+  isSupportAdminSession,
   useCurrentWorkspace,
 } from '@/app/infra/http';
 import { useNavigate, useLocation } from 'react-router-dom';
@@ -156,7 +157,7 @@ export default function HomeLayout({
   // selected Workspace's wizard state.
   useEffect(() => {
     if (!identityReady) return;
-    if (systemInfo.wizard_status === 'none') {
+    if (systemInfo?.wizard_status === 'none' && !isSupportAdminSession()) {
       navigate('/wizard', { replace: true });
     }
   }, [identityReady, navigate]);

web/src/app/infra/http/BackendClient.ts

@@ -1327,8 +1327,10 @@ export class BackendClient extends BaseHttpClient {
     launchAssertion?: string,
   ): Promise<{
     token: string;
-    user: string;
+    user?: string;
     workspace_uuid?: string;
+    principal_type?: 'account' | 'support_admin';
+    actor_account_uuid?: string;
   }> {
     const response = await this.instance.post(
       '/api/v1/user/space/callback',

web/src/app/infra/http/index.ts

@@ -217,10 +217,34 @@ export function beginAuthenticatedSession(
   if (typeof window === 'undefined') return;
   localStorage.removeItem('token');
   localStorage.removeItem('userEmail');
+  localStorage.removeItem('authPrincipalType');
   localStorage.setItem('token', token);
   if (userEmail) localStorage.setItem('userEmail', userEmail);
 }
 
+export function beginSupportAdminSession(
+  token: string,
+  workspaceUuid: string,
+): void {
+  userInfo = null;
+  clearWorkspaceSelection();
+  clearWorkspaceBootstrapSnapshot();
+
+  if (typeof window === 'undefined') return;
+  localStorage.removeItem('token');
+  localStorage.removeItem('userEmail');
+  localStorage.setItem('token', token);
+  localStorage.setItem('authPrincipalType', 'support_admin');
+  setActiveWorkspaceUuid(workspaceUuid);
+}
+
+export function isSupportAdminSession(): boolean {
+  return (
+    typeof window !== 'undefined' &&
+    localStorage.getItem('authPrincipalType') === 'support_admin'
+  );
+}
+
 async function initializeSelectedWorkspace(
   workspaceUuid: string,
   workspaces: WorkspaceBootstrapEntry[],
@@ -252,6 +276,27 @@ async function initializeSelectedWorkspace(
 export async function bootstrapWorkspaceSession(
   options: WorkspaceBootstrapOptions = {},
 ): Promise<WorkspaceBootstrapResult> {
+  if (isSupportAdminSession()) {
+    const selectedWorkspaceUuid = getActiveWorkspaceUuid();
+    if (!selectedWorkspaceUuid) {
+      throw new Error('Support admin session is missing its Workspace scope');
+    }
+    if (
+      options.preferredWorkspaceUuid &&
+      options.preferredWorkspaceUuid !== selectedWorkspaceUuid
+    ) {
+      throw new Error('Support admin session cannot change Workspace scope');
+    }
+    await initializeWorkspaceInfo();
+    const workspace = getCurrentWorkspaceSnapshot();
+    if (!workspace || workspace.workspace.uuid !== selectedWorkspaceUuid) {
+      clearWorkspaceSelection();
+      throw new Error('Support admin Workspace scope could not be initialized');
+    }
+    clearWorkspaceBootstrapSnapshot();
+    return { status: 'ready', workspace, workspaces: [] };
+  }
+
   if (options.resetSelection) {
     clearWorkspaceSelection();
     clearWorkspaceBootstrapSnapshot();
@@ -339,6 +384,9 @@ export const clearUserInfo = (): void => {
   userInfo = null;
   clearWorkspaceSelection();
   clearWorkspaceBootstrapSnapshot();
+  if (typeof window !== 'undefined') {
+    localStorage.removeItem('authPrincipalType');
+  }
 };
 
 export {

web/tests/unit/space-launch-callback.test.mjs

@@ -1,24 +0,0 @@
-import assert from 'node:assert/strict';
-import fs from 'node:fs';
-import test from 'node:test';
-
-const source = fs.readFileSync(
-  new URL('../../src/app/auth/space/callback/page.tsx', import.meta.url),
-  'utf8',
-);
-
-test('direct launch assertion is fragment-only and removed before exchange', () => {
-  assert.doesNotMatch(source, /searchParams\.get\(['"]launch_assertion['"]\)/);
-  const readIndex = source.indexOf("fragmentParams.get('launch_assertion')");
-  const clearIndex = source.indexOf('window.history.replaceState');
-  const exchangeIndex = source.indexOf('handleOAuthCallback(', clearIndex);
-  assert.ok(readIndex >= 0, 'fragment assertion read is missing');
-  assert.ok(clearIndex > readIndex, 'URL fragment is not cleared after copying the assertion');
-  assert.ok(exchangeIndex > clearIndex, 'assertion exchange starts before the fragment is cleared');
-});
-
-test('direct launch honors only a local signed return path', () => {
-  assert.match(source, /response\.return_path\.startsWith\(['"]\/['"]\)/);
-  assert.match(source, /!response\.return_path\.startsWith\(['"]\/\/['"]\)/);
-  assert.match(source, /navigate\(returnPath, \{ replace: true \}\)/);
-});

web/tests/unit/support-admin-session.test.mjs

@@ -0,0 +1,42 @@
+import assert from 'node:assert/strict';
+import fs from 'node:fs';
+import path from 'node:path';
+import test from 'node:test';
+import { fileURLToPath } from 'node:url';
+
+const root = path.resolve(
+  path.dirname(fileURLToPath(import.meta.url)),
+  '../..',
+);
+const read = (file) => fs.readFileSync(path.join(root, file), 'utf8');
+
+test('support-admin launch stores a scoped principal instead of starting an Account session', () => {
+  const callback = read('src/app/auth/space/callback/page.tsx');
+  assert.match(callback, /response\.principal_type === 'support_admin'/);
+  assert.match(
+    callback,
+    /beginSupportAdminSession\(response\.token, response\.workspace_uuid\)/,
+  );
+});
+
+test('support-admin workspace bootstrap never calls Account bootstrap', () => {
+  const source = read('src/app/infra/http/index.ts');
+  assert.match(source, /export function beginSupportAdminSession/);
+  assert.match(source, /export function isSupportAdminSession\(\): boolean/);
+  const supportBranch = source.indexOf('if (isSupportAdminSession())');
+  const accountBootstrap = source.indexOf(
+    'backendClient.getWorkspaceBootstrap()',
+  );
+  assert.ok(supportBranch >= 0);
+  assert.ok(accountBootstrap > supportBranch);
+  assert.match(
+    source.slice(supportBranch, accountBootstrap),
+    /initializeWorkspaceInfo\([\s\S]*status: 'ready'/,
+  );
+  assert.match(
+    source,
+    /localStorage\.setItem\('authPrincipalType', 'support_admin'\)/,
+  );
+  const homeLayout = read('src/app/home/layout.tsx');
+  assert.match(homeLayout, /!isSupportAdminSession\(\)/);
+});