ci(cloud): pin model catalog adapter source

Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>

.github/workflows/deploy-prod.yml

@@ -0,0 +1,59 @@
+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: b0cf8a0a8679efafc1f2c7c62b7f2ff635dd560e
+
+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

deploy/prod/deploy.sh

@@ -0,0 +1,97 @@
+#!/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
+}
+grep -Eq 'LANGBOT_TELEMETRY_INGEST_TOKEN: .+' <<<"$rendered_compose" || {
+  echo 'Cloud telemetry ingest token must be configured' >&2
+  exit 6
+}
+
+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
+: "${CLOUD_V2_CONTROL_PLANE_TOKEN:?CLOUD_V2_CONTROL_PLANE_TOKEN is required}"
+
+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

@@ -0,0 +1,162 @@
+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_TELEMETRY_INGEST_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:

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

@@ -15,8 +15,17 @@ from ....workspace.collaboration import MembershipPermissionError, WorkspaceColl
 from ....workspace.errors import WorkspaceNotFoundError
 from ....cloud.entitlements import EntitlementUnavailableError
 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
@@ -51,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
 
@@ -95,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
@@ -111,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')
@@ -141,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)
@@ -155,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)'
@@ -268,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/plugins.py

@@ -318,6 +318,13 @@ class PluginsRouterGroup(group.RouterGroup):
         )
         return await operation()
 
+    async def _require_authenticated_plugin_runtime_context(
+        self,
+        request_context: RequestContext,
+    ) -> ExecutionContext:
+        """Fence an authenticated resource request to its injected Workspace."""
+        return await self.ap.plugin_connector.require_workspace_context(request_context)
+
     async def _require_public_plugin_runtime_context(self) -> ExecutionContext:
         """Resolve public assets only for the OSS singleton Workspace.
 
@@ -372,7 +379,7 @@ class PluginsRouterGroup(group.RouterGroup):
             permission=Permission.RESOURCE_VIEW,
         )
         async def _(request_context: RequestContext) -> str:
-            await self.ap.plugin_connector.require_workspace_context(request_context)
+            await self._require_authenticated_plugin_runtime_context(request_context)
             plugins = await self.ap.plugin_connector.list_plugins()
 
             return self.success(data={'plugins': redact_plugin_secrets(plugins)})
@@ -385,7 +392,7 @@ class PluginsRouterGroup(group.RouterGroup):
         )
         async def _(request_context: RequestContext) -> str:
             """Get plugin debug information including debug URL and key"""
-            await self.ap.plugin_connector.require_workspace_context(request_context)
+            await self._require_authenticated_plugin_runtime_context(request_context)
             debug_info = await self.ap.plugin_connector.get_debug_info()
 
             # Get debug URL from config
@@ -428,7 +435,7 @@ class PluginsRouterGroup(group.RouterGroup):
             permission=Permission.RESOURCE_VIEW,
         )
         async def _(author: str, plugin_name: str, request_context: RequestContext) -> str:
-            await self.ap.plugin_connector.require_workspace_context(request_context)
+            await self._require_authenticated_plugin_runtime_context(request_context)
             plugin = await self.ap.plugin_connector.get_plugin_info(author, plugin_name)
             if plugin is None:
                 return self.http_status(404, -1, 'plugin not found')
@@ -469,7 +476,7 @@ class PluginsRouterGroup(group.RouterGroup):
             permission=Permission.RESOURCE_VIEW,
         )
         async def _(author: str, plugin_name: str, request_context: RequestContext) -> quart.Response:
-            await self.ap.plugin_connector.require_workspace_context(request_context)
+            await self._require_authenticated_plugin_runtime_context(request_context)
             plugin = await self.ap.plugin_connector.get_plugin_info(author, plugin_name)
             if plugin is None:
                 return self.http_status(404, -1, 'plugin not found')
@@ -489,7 +496,7 @@ class PluginsRouterGroup(group.RouterGroup):
             permission=Permission.RESOURCE_MANAGE,
         )
         async def _(author: str, plugin_name: str, request_context: RequestContext) -> quart.Response:
-            await self.ap.plugin_connector.require_workspace_context(request_context)
+            await self._require_authenticated_plugin_runtime_context(request_context)
             plugin = await self.ap.plugin_connector.get_plugin_info(author, plugin_name)
             if plugin is None:
                 return self.http_status(404, -1, 'plugin not found')
@@ -506,7 +513,7 @@ class PluginsRouterGroup(group.RouterGroup):
                 )
             except ValueError as exc:
                 return self.http_status(400, -1, str(exc))
-            await self.ap.plugin_connector.require_workspace_context(request_context)
+            await self._require_authenticated_plugin_runtime_context(request_context)
             await self.ap.plugin_connector.set_plugin_config(author, plugin_name, config)
             return self.success(data={})
 
@@ -517,7 +524,7 @@ class PluginsRouterGroup(group.RouterGroup):
             permission=Permission.RESOURCE_VIEW,
         )
         async def _(author: str, plugin_name: str, request_context: RequestContext) -> quart.Response:
-            await self.ap.plugin_connector.require_workspace_context(request_context)
+            await self._require_authenticated_plugin_runtime_context(request_context)
             language = quart.request.args.get('language', 'en')
             readme = await self.ap.plugin_connector.get_plugin_readme(author, plugin_name, language=language)
             return self.success(data={'readme': readme})
@@ -529,7 +536,7 @@ class PluginsRouterGroup(group.RouterGroup):
             permission=Permission.AUDIT_VIEW,
         )
         async def _(author: str, plugin_name: str, request_context: RequestContext) -> quart.Response:
-            await self.ap.plugin_connector.require_workspace_context(request_context)
+            await self._require_authenticated_plugin_runtime_context(request_context)
             try:
                 limit = int(quart.request.args.get('limit', 200))
             except (TypeError, ValueError):
@@ -538,6 +545,44 @@ class PluginsRouterGroup(group.RouterGroup):
             logs = await self.ap.plugin_connector.get_plugin_logs(author, plugin_name, limit=limit, level=level)
             return self.success(data={'logs': logs})
 
+        @self.route(
+            '/<author>/<plugin_name>/authenticated-icon',
+            methods=['GET'],
+            auth_type=group.AuthType.USER_TOKEN_OR_API_KEY,
+            permission=Permission.RESOURCE_VIEW,
+        )
+        async def _(
+            author: str,
+            plugin_name: str,
+            request_context: RequestContext,
+        ) -> quart.Response:
+            await self._require_authenticated_plugin_runtime_context(request_context)
+            icon_data = await self.ap.plugin_connector.get_plugin_icon(author, plugin_name)
+            icon_bytes = await asyncio.to_thread(base64.b64decode, icon_data['plugin_icon_base64'])
+            return quart.Response(icon_bytes, mimetype=icon_data['mime_type'])
+
+        @self.route(
+            '/<author>/<plugin_name>/authenticated-assets/<path:filepath>',
+            methods=['GET'],
+            auth_type=group.AuthType.USER_TOKEN_OR_API_KEY,
+            permission=Permission.RESOURCE_VIEW,
+        )
+        async def _(
+            author: str,
+            plugin_name: str,
+            filepath: str,
+            request_context: RequestContext,
+        ) -> quart.Response:
+            await self._require_authenticated_plugin_runtime_context(request_context)
+            asset_path = _normalize_plugin_asset_path(filepath)
+            if asset_path is None:
+                return quart.Response('Asset not found', status=404)
+            asset_data = await self.ap.plugin_connector.get_plugin_assets(author, plugin_name, asset_path)
+            if not asset_data.get('asset_base64'):
+                return quart.Response('Asset not found', status=404)
+            asset_bytes = await asyncio.to_thread(base64.b64decode, asset_data['asset_base64'])
+            return quart.Response(asset_bytes, mimetype=asset_data['mime_type'])
+
         @self.route(
             '/<author>/<plugin_name>/icon',
             methods=['GET'],
@@ -596,7 +641,7 @@ class PluginsRouterGroup(group.RouterGroup):
         )
         async def _(author: str, plugin_name: str, request_context: RequestContext) -> str:
             """Forward a page API request to the plugin."""
-            await self.ap.plugin_connector.require_workspace_context(request_context)
+            await self._require_authenticated_plugin_runtime_context(request_context)
             data = await quart.request.json
             if not isinstance(data, dict):
                 return self.http_status(400, -1, 'invalid request body')
@@ -625,7 +670,7 @@ class PluginsRouterGroup(group.RouterGroup):
         )
         async def _(request_context: RequestContext) -> str:
             """Get releases from a GitHub repository URL"""
-            await self.ap.plugin_connector.require_workspace_context(request_context)
+            await self._require_authenticated_plugin_runtime_context(request_context)
             data = await quart.request.json
             repo_url = data.get('repo_url', '')
 
@@ -705,7 +750,7 @@ class PluginsRouterGroup(group.RouterGroup):
         )
         async def _(request_context: RequestContext) -> str:
             """Get assets from a specific GitHub release"""
-            await self.ap.plugin_connector.require_workspace_context(request_context)
+            await self._require_authenticated_plugin_runtime_context(request_context)
             data = await quart.request.json
             owner = data.get('owner', '')
             repo = data.get('repo', '')
@@ -901,7 +946,7 @@ class PluginsRouterGroup(group.RouterGroup):
             permission=Permission.RESOURCE_MANAGE,
         )
         async def _(request_context: RequestContext) -> str:
-            await self.ap.plugin_connector.require_workspace_context(request_context)
+            await self._require_authenticated_plugin_runtime_context(request_context)
             file = (await quart.request.files).get('file')
             if file is None:
                 return self.http_status(400, -1, 'file is required')
@@ -942,7 +987,7 @@ class PluginsRouterGroup(group.RouterGroup):
         )
         async def _(request_context: RequestContext) -> str:
             """Upload a file for plugin configuration"""
-            await self.ap.plugin_connector.require_workspace_context(request_context)
+            await self._require_authenticated_plugin_runtime_context(request_context)
             file = (await quart.request.files).get('file')
             if file is None:
                 return self.http_status(400, -1, 'file is required')
@@ -974,7 +1019,7 @@ class PluginsRouterGroup(group.RouterGroup):
         )
         async def _(file_key: str, request_context: RequestContext) -> str:
             """Delete a plugin configuration file"""
-            await self.ap.plugin_connector.require_workspace_context(request_context)
+            await self._require_authenticated_plugin_runtime_context(request_context)
             if not self.ap.storage_mgr.is_scoped_object_key(file_key, expected_owner_type='plugin_config'):
                 return self.http_status(400, -1, 'invalid file key')
 

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')

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/model.py

@@ -5,6 +5,7 @@ import uuid
 import sqlalchemy
 from langbot_plugin.api.entities.builtin.provider import message as provider_message
 
+from ....cloud.model_catalog import LANGBOT_MODELS_PROVIDER_REQUESTER
 from ....core import app
 from ....entity.persistence import model as persistence_model
 from ....entity.persistence import pipeline as persistence_pipeline
@@ -113,6 +114,23 @@ async def _require_workspace_provider(
     return provider
 
 
+def _is_cloud_runtime(ap: app.Application) -> bool:
+    mode = getattr(ap.persistence_mgr, 'mode', None)
+    return getattr(mode, 'value', None) == 'cloud_runtime'
+
+
+async def _assert_cloud_managed_provider_mutable(
+    ap: app.Application,
+    context: TenantContext,
+    provider_uuid: str,
+) -> None:
+    if not _is_cloud_runtime(ap):
+        return
+    provider = await _require_workspace_provider(ap, context, provider_uuid)
+    if provider.get('requester') == LANGBOT_MODELS_PROVIDER_REQUESTER:
+        raise ValueError('LangBot Models is managed by Cloud and cannot be modified')
+
+
 async def _require_runtime_provider(
     ap: app.Application,
     context: TenantContext,
@@ -213,6 +231,7 @@ class LLMModelsService:
                 model_data['provider_uuid'] = provider_uuid
 
         await _require_workspace_provider(self.ap, context, model_data['provider_uuid'])
+        await _assert_cloud_managed_provider_mutable(self.ap, context, model_data['provider_uuid'])
         await _validate_provider_supports(self.ap, context, model_data['provider_uuid'], 'llm')
 
         await self.ap.persistence_mgr.execute_async(sqlalchemy.insert(persistence_model.LLMModel).values(**model_data))
@@ -291,11 +310,17 @@ class LLMModelsService:
 
         return model_dict
 
-    async def update_llm_model(self, context: TenantContext, model_uuid: str, model_data: dict) -> None:
+    async def update_llm_model(
+        self,
+        context: TenantContext,
+        model_uuid: str,
+        model_data: dict,
+    ) -> None:
         """Update an existing LLM model"""
         existing_model = await self.get_llm_model(context, model_uuid, include_secret=True)
         if existing_model is None:
             raise WorkspaceNotFoundError('Model not found')
+        await _assert_cloud_managed_provider_mutable(self.ap, context, existing_model['provider_uuid'])
         model_data = model_data.copy()
         model_data.pop('uuid', None)
         model_data.pop('workspace_uuid', None)
@@ -321,6 +346,7 @@ class LLMModelsService:
 
         provider_uuid = model_data.get('provider_uuid', existing_model['provider_uuid'])
         await _require_workspace_provider(self.ap, context, provider_uuid)
+        await _assert_cloud_managed_provider_mutable(self.ap, context, provider_uuid)
         await _validate_provider_supports(self.ap, context, provider_uuid, 'llm')
 
         result = await self.ap.persistence_mgr.execute_async(
@@ -355,6 +381,11 @@ class LLMModelsService:
 
     async def delete_llm_model(self, context: TenantContext, model_uuid: str) -> None:
         """Delete an LLM model"""
+        if _is_cloud_runtime(self.ap):
+            existing_model = await self.get_llm_model(context, model_uuid, include_secret=True)
+            if existing_model is None:
+                raise WorkspaceNotFoundError('Model not found')
+            await _assert_cloud_managed_provider_mutable(self.ap, context, existing_model['provider_uuid'])
         result = await self.ap.persistence_mgr.execute_async(
             scope_statement(
                 sqlalchemy.delete(persistence_model.LLMModel).where(persistence_model.LLMModel.uuid == model_uuid),
@@ -448,7 +479,10 @@ class EmbeddingModelsService:
         return serialized if include_secret else [_redact_model_secrets(model) for model in serialized]
 
     async def create_embedding_model(
-        self, context: TenantContext, model_data: dict, preserve_uuid: bool = False
+        self,
+        context: TenantContext,
+        model_data: dict,
+        preserve_uuid: bool = False,
     ) -> str:
         """Create a new embedding model"""
         model_data = model_data.copy()
@@ -472,6 +506,7 @@ class EmbeddingModelsService:
                 model_data['provider_uuid'] = provider_uuid
 
         await _require_workspace_provider(self.ap, context, model_data['provider_uuid'])
+        await _assert_cloud_managed_provider_mutable(self.ap, context, model_data['provider_uuid'])
         await _validate_provider_supports(self.ap, context, model_data['provider_uuid'], 'text-embedding')
 
         await self.ap.persistence_mgr.execute_async(
@@ -530,11 +565,17 @@ class EmbeddingModelsService:
 
         return model_dict
 
-    async def update_embedding_model(self, context: TenantContext, model_uuid: str, model_data: dict) -> None:
+    async def update_embedding_model(
+        self,
+        context: TenantContext,
+        model_uuid: str,
+        model_data: dict,
+    ) -> None:
         """Update an existing embedding model"""
         existing_model = await self.get_embedding_model(context, model_uuid, include_secret=True)
         if existing_model is None:
             raise WorkspaceNotFoundError('Model not found')
+        await _assert_cloud_managed_provider_mutable(self.ap, context, existing_model['provider_uuid'])
         model_data = model_data.copy()
         model_data.pop('uuid', None)
         model_data.pop('workspace_uuid', None)
@@ -559,6 +600,7 @@ class EmbeddingModelsService:
 
         provider_uuid = model_data.get('provider_uuid', existing_model['provider_uuid'])
         await _require_workspace_provider(self.ap, context, provider_uuid)
+        await _assert_cloud_managed_provider_mutable(self.ap, context, provider_uuid)
         await _validate_provider_supports(self.ap, context, provider_uuid, 'text-embedding')
 
         result = await self.ap.persistence_mgr.execute_async(
@@ -593,6 +635,11 @@ class EmbeddingModelsService:
 
     async def delete_embedding_model(self, context: TenantContext, model_uuid: str) -> None:
         """Delete an embedding model"""
+        if _is_cloud_runtime(self.ap):
+            existing_model = await self.get_embedding_model(context, model_uuid, include_secret=True)
+            if existing_model is None:
+                raise WorkspaceNotFoundError('Model not found')
+            await _assert_cloud_managed_provider_mutable(self.ap, context, existing_model['provider_uuid'])
         result = await self.ap.persistence_mgr.execute_async(
             scope_statement(
                 sqlalchemy.delete(persistence_model.EmbeddingModel).where(
@@ -685,7 +732,12 @@ class RerankModelsService:
         serialized = [self.ap.persistence_mgr.serialize_model(persistence_model.RerankModel, m) for m in models]
         return serialized if include_secret else [_redact_model_secrets(model) for model in serialized]
 
-    async def create_rerank_model(self, context: TenantContext, model_data: dict, preserve_uuid: bool = False) -> str:
+    async def create_rerank_model(
+        self,
+        context: TenantContext,
+        model_data: dict,
+        preserve_uuid: bool = False,
+    ) -> str:
         """Create a new rerank model"""
         model_data = model_data.copy()
         if not preserve_uuid:
@@ -708,6 +760,7 @@ class RerankModelsService:
                 model_data['provider_uuid'] = provider_uuid
 
         await _require_workspace_provider(self.ap, context, model_data['provider_uuid'])
+        await _assert_cloud_managed_provider_mutable(self.ap, context, model_data['provider_uuid'])
         await _validate_provider_supports(self.ap, context, model_data['provider_uuid'], 'rerank')
 
         await self.ap.persistence_mgr.execute_async(
@@ -766,11 +819,17 @@ class RerankModelsService:
 
         return model_dict
 
-    async def update_rerank_model(self, context: TenantContext, model_uuid: str, model_data: dict) -> None:
+    async def update_rerank_model(
+        self,
+        context: TenantContext,
+        model_uuid: str,
+        model_data: dict,
+    ) -> None:
         """Update an existing rerank model"""
         existing_model = await self.get_rerank_model(context, model_uuid, include_secret=True)
         if existing_model is None:
             raise WorkspaceNotFoundError('Model not found')
+        await _assert_cloud_managed_provider_mutable(self.ap, context, existing_model['provider_uuid'])
         model_data = model_data.copy()
         model_data.pop('uuid', None)
         model_data.pop('workspace_uuid', None)
@@ -795,6 +854,7 @@ class RerankModelsService:
 
         provider_uuid = model_data.get('provider_uuid', existing_model['provider_uuid'])
         await _require_workspace_provider(self.ap, context, provider_uuid)
+        await _assert_cloud_managed_provider_mutable(self.ap, context, provider_uuid)
         await _validate_provider_supports(self.ap, context, provider_uuid, 'rerank')
 
         result = await self.ap.persistence_mgr.execute_async(
@@ -829,6 +889,11 @@ class RerankModelsService:
 
     async def delete_rerank_model(self, context: TenantContext, model_uuid: str) -> None:
         """Delete a rerank model"""
+        if _is_cloud_runtime(self.ap):
+            existing_model = await self.get_rerank_model(context, model_uuid, include_secret=True)
+            if existing_model is None:
+                raise WorkspaceNotFoundError('Model not found')
+            await _assert_cloud_managed_provider_mutable(self.ap, context, existing_model['provider_uuid'])
         result = await self.ap.persistence_mgr.execute_async(
             scope_statement(
                 sqlalchemy.delete(persistence_model.RerankModel).where(

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

@@ -5,6 +5,7 @@ import traceback
 
 import sqlalchemy
 
+from ....cloud.model_catalog import LANGBOT_MODELS_PROVIDER_REQUESTER
 from ....core import app
 from ....entity.persistence import model as persistence_model
 from ....workspace.errors import WorkspaceNotFoundError
@@ -20,6 +21,20 @@ class ModelProviderService:
     def __init__(self, ap: app.Application) -> None:
         self.ap = ap
 
+    def _is_cloud_runtime(self) -> bool:
+        mode = getattr(self.ap.persistence_mgr, 'mode', None)
+        return getattr(mode, 'value', None) == 'cloud_runtime'
+
+    def _system_requester_is_reserved(self, requester: object) -> bool:
+        return self._is_cloud_runtime() and requester == LANGBOT_MODELS_PROVIDER_REQUESTER
+
+    async def _assert_provider_mutable(self, context: TenantContext, provider_uuid: str) -> None:
+        if not self._is_cloud_runtime():
+            return
+        provider = await self.get_provider(context, provider_uuid)
+        if provider is not None and self._system_requester_is_reserved(provider.get('requester')):
+            raise ValueError('LangBot Models is managed by Cloud and cannot be modified')
+
     @staticmethod
     def _normalize_api_keys(api_keys: str | list[str] | tuple[str, ...] | None) -> list[str]:
         if api_keys is None:
@@ -99,6 +114,8 @@ class ModelProviderService:
     async def create_provider(self, context: TenantContext, provider_data: dict) -> str:
         """Create a new provider"""
         provider_data = provider_data.copy()
+        if self._system_requester_is_reserved(provider_data.get('requester')):
+            raise ValueError('space-chat-completions is reserved for the Cloud-managed LangBot Models provider')
         provider_data['uuid'] = str(uuid.uuid4())
         provider_data['workspace_uuid'] = require_workspace_uuid(context)
         provider_data['api_keys'] = self._normalize_api_keys(
@@ -115,7 +132,10 @@ class ModelProviderService:
 
     async def update_provider(self, context: TenantContext, provider_uuid: str, provider_data: dict) -> None:
         """Update an existing provider"""
+        await self._assert_provider_mutable(context, provider_uuid)
         provider_data = provider_data.copy()
+        if self._system_requester_is_reserved(provider_data.get('requester')):
+            raise ValueError('space-chat-completions is reserved for the Cloud-managed LangBot Models provider')
         provider_data.pop('uuid', None)
         provider_data.pop('workspace_uuid', None)
         if 'api_keys' in provider_data:
@@ -145,6 +165,7 @@ class ModelProviderService:
 
     async def delete_provider(self, context: TenantContext, provider_uuid: str) -> None:
         """Delete a provider (only if no models reference it)"""
+        await self._assert_provider_mutable(context, provider_uuid)
         workspace_uuid = require_workspace_uuid(context)
         # Check if any models use this provider
         llm_result = await self.ap.persistence_mgr.execute_async(
@@ -245,6 +266,8 @@ class ModelProviderService:
         api_keys: list,
     ) -> str:
         """Find existing provider or create new one"""
+        if self._system_requester_is_reserved(requester):
+            raise ValueError('space-chat-completions is reserved for the Cloud-managed LangBot Models provider')
         workspace_uuid = require_workspace_uuid(context)
         api_keys = self._normalize_api_keys(restore_secret_placeholders(api_keys, sensitive=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

@@ -13,6 +13,7 @@ from typing import Any, Protocol, runtime_checkable
 from ..workspace.policy import CloudWorkspacePolicy, SingleWorkspacePolicy
 from .directory import DirectoryProjectionProvider, directory_projection_limits_from_config
 from .entitlements import EntitlementProvider, OpenSourceEntitlementProvider
+from .model_catalog import CloudModelCatalogProvider
 
 
 CLOUD_BOOTSTRAP_ENTRY_POINT = 'langbot.cloud_bootstrap'
@@ -50,6 +51,7 @@ class OpenSourceDeployment:
     )
     directory_provider: None = None
     manifest_provider: None = None
+    model_catalog_provider: None = None
     persistence_mode: str = 'oss_compat'
     required_vector_backend: str | None = None
 
@@ -80,6 +82,7 @@ class VerifiedCloudDeployment:
     entitlement_provider: EntitlementProvider
     directory_provider: DirectoryProjectionProvider
     manifest_provider: CloudManifestProvider
+    model_catalog_provider: CloudModelCatalogProvider
     verification_key_id: str
     mode: str = dataclasses.field(default='cloud', init=False)
     workspace_policy: CloudWorkspacePolicy = dataclasses.field(default_factory=CloudWorkspacePolicy, init=False)
@@ -110,6 +113,8 @@ class VerifiedCloudDeployment:
             raise CloudBootstrapError('Verified Cloud bootstrap did not provide a directory adapter')
         if not isinstance(self.manifest_provider, CloudManifestProvider):
             raise CloudBootstrapError('Verified Cloud bootstrap did not provide a Manifest renewal adapter')
+        if not isinstance(self.model_catalog_provider, CloudModelCatalogProvider):
+            raise CloudBootstrapError('Verified Cloud bootstrap did not provide a model catalog adapter')
 
     def validate_instance_config(self, config: dict[str, Any]) -> None:
         try:
@@ -138,8 +143,14 @@ 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', {})
-        if box_config.get('enabled') is not True:
-            raise CloudBootstrapError('Cloud runtime requires box.enabled=true')
+        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('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

@@ -15,12 +15,15 @@ from cryptography.exceptions import InvalidSignature
 from cryptography.hazmat.primitives import serialization
 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
 
+from .support_admin import SupportAdminReplayError, SupportAdminSessionError, hash_grant_jti
+
 if typing.TYPE_CHECKING:
     from ..core.app import Application
 
 
 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
@@ -123,15 +126,60 @@ class SpaceLaunchService:
         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')
         if expected_workspace_uuid is not None and workspace_uuid != expected_workspace_uuid:
             raise SpaceLaunchError('Launch assertion targets another Workspace')
-        await self._consume_jti(_required_string(claims, 'jti'), _required_int(claims, 'exp', minimum=1))
-        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,
         }
+        await self._consume_jti(_required_string(claims, 'jti'), _required_int(claims, 'exp', minimum=1))
+        return result
 
     def _verify_assertion(self, token: str) -> dict[str, typing.Any]:
         if not getattr(getattr(self.ap, 'deployment', None), 'multi_workspace_enabled', False):
@@ -169,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')

src/langbot/pkg/cloud/model_catalog.py

@@ -0,0 +1,284 @@
+from __future__ import annotations
+
+import asyncio
+import uuid
+from datetime import datetime
+from typing import Any, Literal, Protocol, runtime_checkable
+
+import sqlalchemy
+from pydantic import BaseModel, ConfigDict, Field, SecretStr, field_validator
+
+from ..entity.persistence import model as persistence_model
+
+
+LANGBOT_MODELS_PROVIDER_REQUESTER = 'space-chat-completions'
+LANGBOT_MODELS_PROVIDER_NAME = 'LangBot Models'
+_MODEL_RESOURCE_NAMESPACE = uuid.UUID('94c703ca-1df5-4e91-bcd3-74ac65cb7921')
+_SUPPORTED_CATEGORIES = {'chat', 'embedding', 'rerank'}
+_MODEL_TABLES = (
+    persistence_model.LLMModel,
+    persistence_model.EmbeddingModel,
+    persistence_model.RerankModel,
+)
+
+
+class CloudModelCatalogItem(BaseModel):
+    model_config = ConfigDict(extra='forbid', frozen=True)
+
+    uuid: str = Field(min_length=1, max_length=255)
+    model_id: str = Field(min_length=1, max_length=255)
+    category: Literal['chat', 'embedding', 'rerank']
+    llm_abilities: tuple[str, ...] = ()
+    is_featured: bool = False
+    featured_order: int = 0
+
+    @field_validator('llm_abilities')
+    @classmethod
+    def validate_abilities(cls, value: tuple[str, ...]) -> tuple[str, ...]:
+        if any(not item.strip() or len(item) > 64 for item in value):
+            raise ValueError('Model abilities must be non-empty strings of at most 64 characters')
+        if len(set(value)) != len(value):
+            raise ValueError('Model abilities must be unique')
+        return value
+
+
+class CloudWorkspaceModelBilling(BaseModel):
+    model_config = ConfigDict(extra='forbid', frozen=True)
+
+    workspace_uuid: str = Field(min_length=36, max_length=36)
+    owner_account_uuid: str | None = Field(default=None, min_length=36, max_length=36)
+    api_key: SecretStr | None = None
+
+    @field_validator('workspace_uuid')
+    @classmethod
+    def validate_uuid(cls, value: str) -> str:
+        return str(uuid.UUID(value))
+
+    @field_validator('owner_account_uuid')
+    @classmethod
+    def validate_optional_uuid(cls, value: str | None) -> str | None:
+        return None if value is None else str(uuid.UUID(value))
+
+
+class CloudModelCatalogSnapshot(BaseModel):
+    model_config = ConfigDict(extra='forbid', frozen=True)
+
+    instance_uuid: str = Field(min_length=1, max_length=255)
+    generated_at: datetime
+    base_url: str = Field(min_length=1, max_length=512)
+    models: tuple[CloudModelCatalogItem, ...]
+    workspaces: tuple[CloudWorkspaceModelBilling, ...]
+
+    @field_validator('base_url')
+    @classmethod
+    def validate_base_url(cls, value: str) -> str:
+        normalized = value.rstrip('/')
+        if not normalized.startswith('https://'):
+            raise ValueError('Cloud model gateway base URL must use HTTPS')
+        return normalized
+
+    @field_validator('models')
+    @classmethod
+    def validate_models(cls, value: tuple[CloudModelCatalogItem, ...]) -> tuple[CloudModelCatalogItem, ...]:
+        if len(value) > 500:
+            raise ValueError('Cloud model catalog exceeds 500 models')
+        identities = {(item.category, item.uuid) for item in value}
+        if len(identities) != len(value):
+            raise ValueError('Cloud model catalog contains duplicate model identities')
+        return value
+
+    @field_validator('workspaces')
+    @classmethod
+    def validate_workspaces(
+        cls, value: tuple[CloudWorkspaceModelBilling, ...]
+    ) -> tuple[CloudWorkspaceModelBilling, ...]:
+        if len(value) > 10_000:
+            raise ValueError('Cloud model catalog exceeds 10000 Workspaces')
+        identities = {item.workspace_uuid for item in value}
+        if len(identities) != len(value):
+            raise ValueError('Cloud model catalog contains duplicate Workspaces')
+        return value
+
+
+@runtime_checkable
+class CloudModelCatalogProvider(Protocol):
+    async def fetch_model_catalog(self, instance_uuid: str) -> CloudModelCatalogSnapshot:
+        """Fetch and verify the complete model catalog and Workspace billing projection."""
+        ...
+
+
+def system_provider_uuid(workspace_uuid: str) -> str:
+    workspace = str(uuid.UUID(workspace_uuid))
+    return str(uuid.uuid5(_MODEL_RESOURCE_NAMESPACE, f'{workspace}:provider:{LANGBOT_MODELS_PROVIDER_REQUESTER}'))
+
+
+def system_model_uuid(workspace_uuid: str, category: str, upstream_uuid: str) -> str:
+    workspace = str(uuid.UUID(workspace_uuid))
+    if category not in _SUPPORTED_CATEGORIES:
+        raise ValueError(f'Unsupported model category: {category}')
+    if not upstream_uuid:
+        raise ValueError('Upstream model UUID is required')
+    return str(uuid.uuid5(_MODEL_RESOURCE_NAMESPACE, f'{workspace}:model:{category}:{upstream_uuid}'))
+
+
+class CloudModelCatalogSyncService:
+    """Reconcile Space-owned model catalog and Owner billing tokens into every Cloud Workspace."""
+
+    def __init__(
+        self,
+        ap: Any,
+        provider: CloudModelCatalogProvider,
+        instance_uuid: str,
+        *,
+        sync_interval_seconds: float = 3600.0,
+    ) -> None:
+        if not isinstance(provider, CloudModelCatalogProvider):
+            raise TypeError('Cloud model catalog sync requires a CloudModelCatalogProvider')
+        if sync_interval_seconds < 10:
+            raise ValueError('Cloud model catalog sync interval must be at least 10 seconds')
+        self.ap = ap
+        self.provider = provider
+        self.instance_uuid = instance_uuid
+        self.sync_interval_seconds = float(sync_interval_seconds)
+
+    async def initialize(self) -> None:
+        await self.sync_once(reload_runtime=False)
+
+    async def run(self) -> None:
+        while True:
+            await asyncio.sleep(self.sync_interval_seconds)
+            try:
+                await self.sync_once(reload_runtime=True)
+            except asyncio.CancelledError:
+                raise
+            except Exception as exc:
+                self.ap.logger.warning(f'Cloud model catalog synchronization failed: {exc}')
+
+    async def sync_once(self, *, reload_runtime: bool = True) -> dict[str, int]:
+        snapshot = await self.provider.fetch_model_catalog(self.instance_uuid)
+        if snapshot.instance_uuid != self.instance_uuid:
+            raise ValueError('Cloud model catalog targets another LangBot instance')
+
+        bindings = await self.ap.workspace_service.list_active_execution_bindings()
+        billing_by_workspace = {item.workspace_uuid: item for item in snapshot.workspaces}
+        missing = sorted(
+            binding.workspace_uuid for binding in bindings if binding.workspace_uuid not in billing_by_workspace
+        )
+        if missing:
+            raise ValueError(f'Cloud model catalog is missing billing projections for {len(missing)} active Workspaces')
+
+        summary = {'workspaces': 0, 'created': 0, 'updated': 0, 'deleted': 0}
+        changed = False
+        for binding in bindings:
+            counts = await self._sync_workspace(
+                binding.workspace_uuid,
+                snapshot,
+                billing_by_workspace[binding.workspace_uuid],
+            )
+            summary['workspaces'] += 1
+            for key in ('created', 'updated', 'deleted'):
+                summary[key] += counts[key]
+                changed = changed or counts[key] > 0
+
+        if changed and reload_runtime and getattr(self.ap, 'model_mgr', None) is not None:
+            await self.ap.model_mgr.load_models_from_db()
+        if changed:
+            self.ap.logger.info(
+                'Cloud model catalog synchronized '
+                f'({summary["workspaces"]} Workspaces, {len(snapshot.models)} models, '
+                f'created={summary["created"]}, updated={summary["updated"]}, deleted={summary["deleted"]})'
+            )
+        return summary
+
+    async def _sync_workspace(
+        self,
+        workspace_uuid: str,
+        snapshot: CloudModelCatalogSnapshot,
+        billing: CloudWorkspaceModelBilling,
+    ) -> dict[str, int]:
+        counts = {'created': 0, 'updated': 0, 'deleted': 0}
+        provider_uuid = system_provider_uuid(workspace_uuid)
+        desired_keys = [billing.api_key.get_secret_value()] if billing.api_key is not None else []
+
+        async with self.ap.persistence_mgr.tenant_uow(workspace_uuid) as uow:
+            provider = await uow.session.scalar(
+                sqlalchemy.select(persistence_model.ModelProvider).where(
+                    persistence_model.ModelProvider.uuid == provider_uuid
+                )
+            )
+            provider_values = {
+                'workspace_uuid': workspace_uuid,
+                'name': LANGBOT_MODELS_PROVIDER_NAME,
+                'requester': LANGBOT_MODELS_PROVIDER_REQUESTER,
+                'base_url': snapshot.base_url,
+                'api_keys': desired_keys,
+            }
+            if provider is None:
+                provider = persistence_model.ModelProvider(uuid=provider_uuid, **provider_values)
+                uow.session.add(provider)
+                await uow.session.flush()
+                counts['created'] += 1
+            elif self._update_entity(provider, provider_values):
+                counts['updated'] += 1
+
+            existing_by_table: dict[type, dict[str, Any]] = {}
+            for table in _MODEL_TABLES:
+                rows = (
+                    await uow.session.scalars(sqlalchemy.select(table).where(table.provider_uuid == provider_uuid))
+                ).all()
+                existing_by_table[table] = {row.uuid: row for row in rows}
+
+            desired_ids: dict[type, set[str]] = {table: set() for table in _MODEL_TABLES}
+            for item in snapshot.models:
+                table, values = self._model_values(workspace_uuid, provider_uuid, item)
+                model_uuid = system_model_uuid(workspace_uuid, item.category, item.uuid)
+                desired_ids[table].add(model_uuid)
+                existing = existing_by_table[table].get(model_uuid)
+                if existing is None:
+                    uow.session.add(table(uuid=model_uuid, **values))
+                    counts['created'] += 1
+                elif self._update_entity(existing, values):
+                    counts['updated'] += 1
+
+            for table, entities in existing_by_table.items():
+                for model_uuid, entity in entities.items():
+                    if model_uuid not in desired_ids[table]:
+                        await uow.session.delete(entity)
+                        counts['deleted'] += 1
+
+        return counts
+
+    @staticmethod
+    def _update_entity(entity: Any, values: dict[str, Any]) -> bool:
+        changed = False
+        for key, value in values.items():
+            if getattr(entity, key) != value:
+                setattr(entity, key, value)
+                changed = True
+        return changed
+
+    @staticmethod
+    def _model_values(
+        workspace_uuid: str,
+        provider_uuid: str,
+        item: CloudModelCatalogItem,
+    ) -> tuple[type, dict[str, Any]]:
+        ranking = 100 - item.featured_order if item.is_featured else 0
+        common = {
+            'workspace_uuid': workspace_uuid,
+            'name': item.model_id,
+            'provider_uuid': provider_uuid,
+            'extra_args': {},
+            'prefered_ranking': ranking,
+        }
+        if item.category == 'chat':
+            return persistence_model.LLMModel, {
+                **common,
+                'abilities': list(item.llm_abilities),
+                'context_length': None,
+            }
+        if item.category == 'embedding':
+            return persistence_model.EmbeddingModel, common
+        if item.category == 'rerank':
+            return persistence_model.RerankModel, common
+        raise ValueError(f'Unsupported model category: {item.category}')

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,8 +51,10 @@ 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 ..cloud import model_catalog as cloud_model_catalog_module
 from ..api.http.context import ExecutionContext, PrincipalContext, PrincipalType
 
 
@@ -136,16 +138,17 @@ 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
-
+    directory_projection_service: cloud_directory_projection_module.DirectoryProjectionService | None = None
+    cloud_model_catalog_service: cloud_model_catalog_module.CloudModelCatalogSyncService | None = None
     manifest_refresh_service: cloud_bootstrap_module.CloudManifestRefreshService | None = None
 
     entitlement_resolver: cloud_entitlements_module.EntitlementResolver | None = None
 
-    directory_projection_service: cloud_directory_projection_module.DirectoryProjectionService | None = None
-
     vector_db_mgr: vectordb_mgr.VectorDBManager = None
 
     http_ctrl: http_controller.HTTPController = None
@@ -303,6 +306,12 @@ class Application:
                     name='cloud-directory-projection',
                     scopes=[core_entities.LifecycleControlScope.APPLICATION],
                 )
+            if self.cloud_model_catalog_service is not None:
+                self.task_mgr.create_task(
+                    self.cloud_model_catalog_service.run(),
+                    name='cloud-model-catalog-sync',
+                    scopes=[core_entities.LifecycleControlScope.APPLICATION],
+                )
             if self.manifest_refresh_service is not None:
                 self.task_mgr.create_task(
                     self.manifest_refresh_service.run(),

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

@@ -42,9 +42,11 @@ 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
+from ...cloud.model_catalog import CloudModelCatalogSyncService
 from ...api.http.context import ExecutionContext, PrincipalContext, PrincipalType
 from ...api.http.authz import WorkspaceRequiredError
 
@@ -175,11 +177,22 @@ class BuildAppStage(stage.BootingStage):
             # of repeating tenant validation for every manager.
             await workspace_service_inst.prime_startup_execution_bindings()
 
+            if not isinstance(deployment, cloud_bootstrap.VerifiedCloudDeployment):
+                raise RuntimeError('Multi-Workspace runtime requires a verified Cloud deployment')
+            cloud_model_catalog_service = CloudModelCatalogSyncService(
+                ap,
+                deployment.model_catalog_provider,
+                constants.instance_id,
+            )
+            await cloud_model_catalog_service.initialize()
+            ap.cloud_model_catalog_service = cloud_model_catalog_service
+
         ap.workspace_collaboration_service = workspace_collaboration_module.WorkspaceCollaborationService(
             ap,
             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/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/0009_workspace_tenancy_kernel.py

@@ -18,6 +18,17 @@ down_revision = '0008_mcp_resource_prefs'
 branch_labels = None
 depends_on = None
 
+_WORKSPACE_IDENTITY_NAMESPACE = uuid.UUID('8ea04f29-8528-4cc3-bb28-30a838c89d76')
+
+
+def _workspace_uuid_from_instance_id(instance_id: str) -> str:
+    value = instance_id.strip()
+    candidate = value[len('instance_') :] if value.startswith('instance_') else value
+    try:
+        return str(uuid.UUID(candidate))
+    except ValueError:
+        return str(uuid.uuid5(_WORKSPACE_IDENTITY_NAMESPACE, value))
+
 
 def _table_names(conn: sa.Connection) -> set[str]:
     return set(sa.inspect(conn).get_table_names())
@@ -403,7 +414,7 @@ def _bootstrap_default_workspace(conn: sa.Connection) -> None:
                 .values(created_by_account_uuid=owner_account_uuid)
             )
     else:
-        workspace_uuid = str(uuid.uuid4())
+        workspace_uuid = _workspace_uuid_from_instance_id(instance_uuid)
         conn.execute(
             workspaces.insert().values(
                 uuid=workspace_uuid,

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

@@ -0,0 +1,57 @@
+"""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/alembic/versions/0017_oss_workspace_identity.py

@@ -0,0 +1,167 @@
+"""align the OSS Workspace UUID with the persisted instance identity
+
+Revision ID: 0017_oss_workspace_identity
+Revises: 0016_support_admin_sessions
+Create Date: 2026-07-31
+"""
+
+from __future__ import annotations
+
+import uuid
+
+import sqlalchemy as sa
+from alembic import op
+
+
+revision = '0017_oss_workspace_identity'
+down_revision = '0016_support_admin_sessions'
+branch_labels = None
+depends_on = None
+
+_WORKSPACE_IDENTITY_NAMESPACE = uuid.UUID('8ea04f29-8528-4cc3-bb28-30a838c89d76')
+_OSS_WORKSPACE_METADATA_KEY = 'oss_workspace_uuid'
+
+
+def _workspace_uuid_from_instance_id(instance_id: str) -> str:
+    value = instance_id.strip()
+    candidate = value[len('instance_') :] if value.startswith('instance_') else value
+    try:
+        return str(uuid.UUID(candidate))
+    except ValueError:
+        return str(uuid.uuid5(_WORKSPACE_IDENTITY_NAMESPACE, value))
+
+
+def _quote(conn: sa.Connection, identifier: str) -> str:
+    return conn.dialect.identifier_preparer.quote(identifier)
+
+
+def _defer_foreign_keys(conn: sa.Connection, inspector: sa.Inspector, table_names: list[str]) -> None:
+    """Allow the transaction to re-key a connected tenant graph atomically."""
+
+    if conn.dialect.name == 'sqlite':
+        conn.execute(sa.text('PRAGMA defer_foreign_keys = ON'))
+        return
+    if conn.dialect.name != 'postgresql':
+        raise RuntimeError(f'Unsupported Workspace identity migration dialect: {conn.dialect.name}')
+
+    for table_name in table_names:
+        for foreign_key in inspector.get_foreign_keys(table_name):
+            constraint_name = foreign_key.get('name')
+            if not constraint_name:
+                continue
+            conn.execute(
+                sa.text(
+                    f'ALTER TABLE {_quote(conn, table_name)} '
+                    f'ALTER CONSTRAINT {_quote(conn, constraint_name)} DEFERRABLE INITIALLY DEFERRED'
+                )
+            )
+
+
+def _suspend_postgres_rls(
+    conn: sa.Connection,
+    table_names: list[str],
+) -> dict[str, tuple[bool, bool]]:
+    if conn.dialect.name != 'postgresql':
+        return {}
+
+    states: dict[str, tuple[bool, bool]] = {}
+    for table_name in table_names:
+        row = conn.execute(
+            sa.text('SELECT relrowsecurity, relforcerowsecurity FROM pg_class WHERE oid = to_regclass(:table_name)'),
+            {'table_name': table_name},
+        ).one()
+        enabled, forced = bool(row.relrowsecurity), bool(row.relforcerowsecurity)
+        states[table_name] = (enabled, forced)
+        table = _quote(conn, table_name)
+        if forced:
+            conn.execute(sa.text(f'ALTER TABLE {table} NO FORCE ROW LEVEL SECURITY'))
+        if enabled:
+            conn.execute(sa.text(f'ALTER TABLE {table} DISABLE ROW LEVEL SECURITY'))
+    return states
+
+
+def _restore_postgres_rls(conn: sa.Connection, states: dict[str, tuple[bool, bool]]) -> None:
+    for table_name, (enabled, forced) in states.items():
+        table = _quote(conn, table_name)
+        if enabled:
+            conn.execute(sa.text(f'ALTER TABLE {table} ENABLE ROW LEVEL SECURITY'))
+        if forced:
+            conn.execute(sa.text(f'ALTER TABLE {table} FORCE ROW LEVEL SECURITY'))
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    inspector = sa.inspect(conn)
+    table_names = inspector.get_table_names()
+    if 'workspaces' not in table_names:
+        return
+
+    metadata = sa.MetaData()
+    workspaces = sa.Table('workspaces', metadata, autoload_with=conn)
+    local_rows = conn.execute(sa.select(workspaces).where(workspaces.c.source == 'local')).mappings().all()
+    if not local_rows:
+        return
+    if len(local_rows) != 1:
+        raise RuntimeError('Cannot align OSS Workspace identity: expected exactly one local Workspace')
+
+    old_row = dict(local_rows[0])
+    old_uuid = old_row['uuid']
+    canonical_uuid = _workspace_uuid_from_instance_id(old_row['instance_uuid'])
+    if old_uuid == canonical_uuid:
+        return
+    if conn.execute(sa.select(workspaces.c.uuid).where(workspaces.c.uuid == canonical_uuid)).scalar_one_or_none():
+        raise RuntimeError(f'Cannot align OSS Workspace identity: target {canonical_uuid!r} already exists')
+
+    tenant_tables = [
+        table_name
+        for table_name in table_names
+        if table_name == 'workspaces'
+        or 'workspace_uuid' in {column['name'] for column in inspector.get_columns(table_name)}
+    ]
+    rls_states = _suspend_postgres_rls(conn, tenant_tables)
+    try:
+        _defer_foreign_keys(conn, inspector, table_names)
+
+        # Release local source/slug uniqueness while the canonical parent exists
+        # alongside the old parent for the duration of this transaction.
+        temporary_slug = f'__workspace_rekey__{old_uuid}'
+        conn.execute(
+            workspaces.update()
+            .where(workspaces.c.uuid == old_uuid)
+            .values(source='cloud_projection', slug=temporary_slug)
+        )
+        new_row = dict(old_row)
+        new_row['uuid'] = canonical_uuid
+        conn.execute(workspaces.insert().values(**new_row))
+
+        for table_name in tenant_tables:
+            if table_name == 'workspaces':
+                continue
+            table = sa.Table(table_name, metadata, autoload_with=conn, extend_existing=True)
+            conn.execute(table.update().where(table.c.workspace_uuid == old_uuid).values(workspace_uuid=canonical_uuid))
+
+        if 'metadata' in table_names:
+            conn.execute(
+                sa.text('UPDATE metadata SET value = :canonical_uuid WHERE key = :key AND value = :old_uuid'),
+                {
+                    'canonical_uuid': canonical_uuid,
+                    'key': _OSS_WORKSPACE_METADATA_KEY,
+                    'old_uuid': old_uuid,
+                },
+            )
+        conn.execute(workspaces.delete().where(workspaces.c.uuid == old_uuid))
+        if conn.dialect.name == 'postgresql':
+            # Fire deferred FK triggers before ALTER TABLE restores RLS; PostgreSQL
+            # rejects ALTER TABLE while a relation has pending trigger events.
+            conn.execute(sa.text('SET CONSTRAINTS ALL IMMEDIATE'))
+    except Exception:
+        # Alembic owns the transaction. Rollback restores the transactional RLS DDL.
+        raise
+    else:
+        _restore_postgres_rls(conn, rls_states)
+
+
+def downgrade() -> None:
+    # The previous random UUID is intentionally not recoverable. Keeping the
+    # canonical identity preserves every FK and is safe for older application code.
+    pass

src/langbot/pkg/persistence/alembic/versions/0018_merge_launch_replay.py

@@ -0,0 +1,21 @@
+"""merge the published Space launch replay and main migration branches
+
+Revision ID: 0018_merge_launch_replay
+Revises: 0016_space_launch_replay, 0017_oss_workspace_identity
+Create Date: 2026-08-01
+"""
+
+from __future__ import annotations
+
+revision = '0018_merge_launch_replay'
+down_revision = ('0016_space_launch_replay', '0017_oss_workspace_identity')
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    pass
+
+
+def downgrade() -> None:
+    pass

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',

src/langbot/pkg/pipeline/process/handlers/chat.py

@@ -15,6 +15,7 @@ from ....provider import runner as runner_module
 import langbot_plugin.api.entities.events as events
 from ....utils import importutil, constants, runner as runner_utils
 from ....telemetry import features as telemetry_features
+from ....telemetry.identity import workspace_identity
 from ....provider import runners
 import langbot_plugin.api.entities.builtin.provider.session as provider_session
 import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query
@@ -265,7 +266,8 @@ class ChatMessageHandler(handler.MessageHandler):
                         'duration_ms': duration_ms,
                         'model_name': model_name,
                         'version': constants.semantic_version,
-                        'instance_id': constants.instance_id,
+                        **workspace_identity(get_query_execution_context(query)),
+                        'runtime_instance_id': constants.instance_id,
                         'edition': constants.edition,
                         'pipeline_plugins': pipeline_plugins,
                         'features': features,

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/telemetry/heartbeat.py

@@ -27,6 +27,19 @@ if typing.TYPE_CHECKING:
 HEARTBEAT_INTERVAL_SECONDS = 24 * 3600
 
 
+class WorkspaceResourceSnapshot(typing.TypedDict):
+    workspace_uuid: str
+    bot_count: int
+    pipeline_count: int
+    knowledge_base_count: int
+    plugin_count: int
+    mcp_server_count: int
+    extension_count: int
+    skill_count: int
+    adapters: list[str]
+    execution_generation: int
+
+
 async def _count(
     ap: core_app.Application,
     table,
@@ -52,8 +65,64 @@ async def _count(
         return -1
 
 
-async def build_heartbeat_payload(ap: core_app.Application) -> dict:
-    """Collect the anonymous instance profile snapshot."""
+async def _cloud_workspace_resource_counts(ap: core_app.Application, bindings) -> list[WorkspaceResourceSnapshot]:
+    """Summarize already-loaded Cloud registries without per-tenant SQL."""
+    persistence_mgr = ap.persistence_mgr
+    if getattr(getattr(persistence_mgr, 'mode', None), 'value', None) != 'cloud_runtime':
+        return []
+
+    counts: dict[str, WorkspaceResourceSnapshot] = {
+        binding.workspace_uuid: {
+            'workspace_uuid': binding.workspace_uuid,
+            'bot_count': 0,
+            'pipeline_count': 0,
+            'knowledge_base_count': 0,
+            'plugin_count': 0,
+            'mcp_server_count': 0,
+            'extension_count': 0,
+            'skill_count': 0,
+            'adapters': [],
+            'execution_generation': binding.placement_generation,
+        }
+        for binding in bindings
+    }
+
+    adapter_sets: dict[str, set[str]] = {workspace_uuid: set() for workspace_uuid in counts}
+    for key, bot in getattr(ap.platform_mgr, '_bots_by_key', {}).items():
+        if len(key) >= 2 and key[1] in counts:
+            counts[key[1]]['bot_count'] += 1
+            adapter = getattr(bot, 'adapter', None)
+            if adapter is not None and getattr(bot, 'enable', False):
+                adapter_sets[key[1]].add(adapter.__class__.__name__)
+    for key in getattr(ap.pipeline_mgr, '_pipelines_by_key', {}):
+        if len(key) >= 2 and key[1] in counts:
+            counts[key[1]]['pipeline_count'] += 1
+    for key in getattr(ap.rag_mgr, 'knowledge_bases', {}):
+        if len(key) >= 1 and key[0] in counts:
+            counts[key[0]]['knowledge_base_count'] += 1
+    for key in getattr(ap.tool_mgr.mcp_tool_loader, '_sessions', {}):
+        if len(key) >= 2 and key[1] in counts:
+            counts[key[1]]['mcp_server_count'] += 1
+    for workspace_uuid, installations in getattr(ap.plugin_connector, '_workspace_installations', {}).items():
+        if workspace_uuid in counts:
+            counts[workspace_uuid]['plugin_count'] = len(installations)
+    for key, skills in getattr(ap.skill_mgr, '_skills_by_scope', {}).items():
+        if len(key) >= 2 and key[1] in counts:
+            counts[key[1]]['skill_count'] += len(skills)
+
+    for workspace_uuid, resource in counts.items():
+        resource['extension_count'] = resource['plugin_count'] + resource['mcp_server_count']
+        resource['adapters'] = sorted(adapter_sets[workspace_uuid])
+    return list(counts.values())
+
+
+async def build_heartbeat_payload(
+    ap: core_app.Application,
+    *,
+    workspace_uuid: str,
+    workspace_resource: WorkspaceResourceSnapshot | None = None,
+) -> dict:
+    """Collect one anonymous Workspace profile snapshot."""
     from ..entity.persistence import bot as persistence_bot
     from ..entity.persistence import mcp as persistence_mcp
     from ..entity.persistence import pipeline as persistence_pipeline
@@ -136,11 +205,14 @@ async def build_heartbeat_payload(ap: core_app.Application) -> dict:
     except Exception:
         pass
 
+    if workspace_resource is not None:
+        features.update({key: value for key, value in workspace_resource.items() if key != 'workspace_uuid'})
+
     return {
         'event_type': 'instance_heartbeat',
         'query_id': '',
         'version': constants.semantic_version,
-        'instance_id': constants.instance_id,
+        'workspace_uuid': workspace_uuid,
         'instance_create_ts': constants.instance_create_ts,
         'edition': constants.edition,
         'features': features,
@@ -148,14 +220,34 @@ async def build_heartbeat_payload(ap: core_app.Application) -> dict:
     }
 
 
+async def build_heartbeat_payloads(ap: core_app.Application) -> list[dict]:
+    """Build one heartbeat per active Workspace."""
+    bindings = await ap.workspace_service.list_active_execution_bindings()
+    workspace_uuids = sorted({binding.workspace_uuid for binding in bindings})
+    resources = {
+        resource['workspace_uuid']: resource for resource in await _cloud_workspace_resource_counts(ap, bindings)
+    }
+    return [
+        await build_heartbeat_payload(
+            ap,
+            workspace_uuid=workspace_uuid,
+            workspace_resource=resources.get(workspace_uuid),
+        )
+        for workspace_uuid in workspace_uuids
+    ]
+
+
 async def heartbeat_loop(ap: core_app.Application) -> None:
     """Send one heartbeat shortly after startup, then daily."""
     # Small delay so managers (platform, skills, plugins) finish loading first
     await asyncio.sleep(30)
     while True:
         try:
-            payload = await build_heartbeat_payload(ap)
-            await ap.telemetry.start_send_task(payload)
+            for payload in await build_heartbeat_payloads(ap):
+                # Heartbeats are a daily bounded batch, not best-effort query events.
+                # Await each send so the TelemetryManager's 8-task queue cannot drop
+                # Workspaces after the first batch.
+                await ap.telemetry.send(payload)
         except Exception as e:
             try:
                 ap.logger.debug(f'Telemetry heartbeat failed: {e}')

src/langbot/pkg/telemetry/identity.py

@@ -0,0 +1,16 @@
+from __future__ import annotations
+
+import typing
+
+
+class WorkspaceExecutionContext(typing.Protocol):
+    @property
+    def workspace_uuid(self) -> str: ...
+
+
+def workspace_identity(execution_context: WorkspaceExecutionContext) -> dict[str, str]:
+    """Build the canonical telemetry identity for one Workspace execution."""
+    workspace_uuid = execution_context.workspace_uuid.strip()
+    if not workspace_uuid:
+        raise ValueError('Telemetry execution Workspace UUID is empty')
+    return {'workspace_uuid': workspace_uuid}

src/langbot/pkg/telemetry/telemetry.py

@@ -2,7 +2,11 @@ from __future__ import annotations
 
 import asyncio
 import contextlib
+import os
+import typing
+
 import httpx
+
 from ..core import app as core_app
 from ..utils import httpclient
 
@@ -21,7 +25,7 @@ class TelemetryManager:
     def __init__(self, ap: core_app.Application):
         self.ap = ap
 
-        self.telemetry_config = {}
+        self.telemetry_config: dict[str, typing.Any] = {}
         self.send_tasks: list[asyncio.Task] = []
         self._client: httpx.AsyncClient | None = None
 
@@ -131,7 +135,16 @@ class TelemetryManager:
                 async with self._client_context() as client:
                     try:
                         # Use asyncio.wait_for to ensure we always bound the total time
-                        resp = await asyncio.wait_for(client.post(url, json=sanitized), timeout=10 + 1)
+                        telemetry_token = os.getenv('LANGBOT_TELEMETRY_INGEST_TOKEN', '').strip()
+                        if telemetry_token:
+                            request = client.post(
+                                url,
+                                json=sanitized,
+                                headers={'X-LangBot-Telemetry-Token': telemetry_token},
+                            )
+                        else:
+                            request = client.post(url, json=sanitized)
+                        resp = await asyncio.wait_for(request, timeout=10 + 1)
 
                         if resp.status_code >= 400:
                             body = await httpclient.response_text(resp, max_chars=200)
@@ -143,7 +156,8 @@ class TelemetryManager:
                             app_err = False
                             try:
                                 j = await httpclient.parse_json_response(resp)
-                                if isinstance(j, dict) and j.get('code') is not None and int(j.get('code')) >= 400:
+                                app_code = j.get('code') if isinstance(j, dict) else None
+                                if app_code is not None and int(app_code) >= 400:
                                     app_err = True
                                     self.ap.logger.warning(
                                         f'Telemetry post to {url} returned application error code {j.get("code")} - {j.get("msg")}'

src/langbot/pkg/workspace/identity.py

@@ -0,0 +1,20 @@
+from __future__ import annotations
+
+import uuid
+
+
+_INSTANCE_PREFIX = 'instance_'
+_WORKSPACE_IDENTITY_NAMESPACE = uuid.UUID('8ea04f29-8528-4cc3-bb28-30a838c89d76')
+
+
+def workspace_uuid_from_instance_id(instance_id: str) -> str:
+    """Return the stable OSS Workspace UUID for a persisted instance identity."""
+    value = instance_id.strip()
+    if not value:
+        raise ValueError('LangBot instance identity is empty')
+
+    candidate = value[len(_INSTANCE_PREFIX) :] if value.startswith(_INSTANCE_PREFIX) else value
+    try:
+        return str(uuid.UUID(candidate))
+    except ValueError:
+        return str(uuid.uuid5(_WORKSPACE_IDENTITY_NAMESPACE, value))

src/langbot/pkg/workspace/service.py

@@ -30,6 +30,7 @@ from .errors import (
     WorkspaceOwnerAlreadyExistsError,
 )
 from .entities import WorkspaceExecutionBinding
+from .identity import workspace_uuid_from_instance_id
 from .policy import CloudWorkspacePolicy, SingleWorkspacePolicy
 from .repository import WorkspaceRepository
 
@@ -497,7 +498,7 @@ class WorkspaceService:
         created_by_account_uuid: str | None = None,
     ) -> Workspace:
         return Workspace(
-            uuid=str(uuid.uuid4()),
+            uuid=workspace_uuid_from_instance_id(self.instance_uuid),
             instance_uuid=self.instance_uuid,
             name=name,
             slug=slug,

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_migrations.py

@@ -95,6 +95,18 @@ class TestSQLiteMigrationBaseline:
 class TestSQLiteMigrationUpgrade:
     """Tests for upgrade to head workflow."""
 
+    @pytest.mark.asyncio
+    async def test_upgrade_from_published_space_launch_head_to_merged_head(self, sqlite_engine):
+        """A database released at the production-only 0016 head must remain upgradable."""
+        async with sqlite_engine.begin() as conn:
+            await conn.run_sync(Base.metadata.create_all)
+
+        await run_alembic_stamp(sqlite_engine, '0016_space_launch_replay')
+        await run_alembic_upgrade(sqlite_engine, 'head')
+
+        assert await get_alembic_current(sqlite_engine) == _get_script_head()
+        assert _get_script_head() == '0018_merge_launch_replay'
+
     @pytest.mark.asyncio
     async def test_upgrade_from_baseline_to_head(self, sqlite_engine):
         """

tests/integration/persistence/test_workspace_migration.py

@@ -22,6 +22,7 @@ from langbot.pkg.persistence.alembic_runner import (
 from langbot.pkg.utils import constants
 from langbot.pkg.utils import importutil
 from langbot.pkg.workspace.collaboration import normalize_email
+from langbot.pkg.workspace.identity import workspace_uuid_from_instance_id
 
 
 pytestmark = [pytest.mark.integration, pytest.mark.asyncio]
@@ -109,6 +110,7 @@ async def test_legacy_instance_gets_stable_accounts_and_default_workspace(legacy
             .mappings()
             .one()
         )
+        assert workspace['uuid'] == workspace_uuid_from_instance_id('instance_migration_test')
         assert workspace['instance_uuid'] == 'instance_migration_test'
         assert workspace['slug'] == 'default'
         assert workspace['status'] == 'active'
@@ -149,6 +151,53 @@ async def test_workspace_upgrade_is_idempotent_and_preserves_identifiers(legacy_
     assert workspace_uuid_after == workspace_uuid_before
 
 
+async def test_existing_oss_workspace_is_rekeyed_to_instance_identity(tmp_path):
+    engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "workspace-rekey.db"}')
+    instance_id = 'instance_a711d9e4-0953-443f-a0e9-7dd50193a79f'
+    old_workspace_uuid = '11111111-1111-4111-8111-111111111111'
+    canonical_uuid = workspace_uuid_from_instance_id(instance_id)
+    schema = sa.MetaData()
+    sa.Table(
+        'metadata',
+        schema,
+        sa.Column('key', sa.String(255), primary_key=True),
+        sa.Column('value', sa.String(255)),
+    )
+    sa.Table(
+        'workspaces',
+        schema,
+        sa.Column('uuid', sa.String(36), primary_key=True),
+        sa.Column('instance_uuid', sa.String(255), nullable=False),
+        sa.Column('slug', sa.String(255), nullable=False),
+        sa.Column('source', sa.String(32), nullable=False),
+    )
+    sa.Table(
+        'tenant_rows',
+        schema,
+        sa.Column('id', sa.Integer, primary_key=True),
+        sa.Column('workspace_uuid', sa.String(36), sa.ForeignKey('workspaces.uuid'), nullable=False),
+    )
+    async with engine.begin() as conn:
+        await conn.run_sync(schema.create_all)
+        await conn.execute(sa.text("INSERT INTO metadata (key, value) VALUES ('instance_uuid', :value)"), {'value': instance_id})
+        await conn.execute(
+            sa.text("INSERT INTO workspaces (uuid, instance_uuid, slug, source) VALUES (:uuid, :instance, 'default', 'local')"),
+            {'uuid': old_workspace_uuid, 'instance': instance_id},
+        )
+        await conn.execute(
+            sa.text("INSERT INTO tenant_rows (id, workspace_uuid) VALUES (1, :uuid)"),
+            {'uuid': old_workspace_uuid},
+        )
+    await run_alembic_stamp(engine, '0016_support_admin_sessions')
+
+    await run_alembic_upgrade(engine, 'head')
+
+    async with engine.connect() as conn:
+        assert (await conn.execute(sa.text("SELECT uuid FROM workspaces"))).scalar_one() == canonical_uuid
+        assert (await conn.execute(sa.text("SELECT workspace_uuid FROM tenant_rows"))).scalar_one() == canonical_uuid
+    await engine.dispose()
+
+
 async def test_workspace_kernel_upgrade_downgrade_upgrade_round_trip(tmp_path):
     engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "workspace-round-trip.db"}')
     try:
@@ -362,6 +411,47 @@ async def test_persistence_startup_defers_workspace_tables_until_account_upgrade
         await engine.dispose()
 
 
+async def test_oss_workspace_identity_rekeys_fk_graph_and_metadata(tmp_path):
+    engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "workspace-rekey.db"}')
+    try:
+        await _create_legacy_schema(engine)
+        await run_alembic_stamp(engine, '0008_mcp_resource_prefs')
+        await run_alembic_upgrade(engine, '0016_support_admin_sessions')
+
+        async with engine.begin() as conn:
+            old_uuid = await conn.scalar(sa.text("SELECT uuid FROM workspaces WHERE source = 'local'"))
+            instance_uuid = await conn.scalar(sa.text("SELECT instance_uuid FROM workspaces WHERE source = 'local'"))
+            assert old_uuid
+            assert instance_uuid
+            await conn.execute(
+                sa.text(
+                    "INSERT INTO workspace_metadata (workspace_uuid, key, value) "
+                    "VALUES (:workspace_uuid, 'migration_probe', 'present')"
+                ),
+                {'workspace_uuid': old_uuid},
+            )
+            await conn.execute(
+                sa.text(
+                    "INSERT INTO metadata (key, value) VALUES ('oss_workspace_uuid', :workspace_uuid) "
+                    "ON CONFLICT(key) DO UPDATE SET value = excluded.value"
+                ),
+                {'workspace_uuid': old_uuid},
+            )
+
+        await run_alembic_upgrade(engine, 'head')
+        expected_uuid = workspace_uuid_from_instance_id(instance_uuid)
+        async with engine.connect() as conn:
+            assert await conn.scalar(sa.text("SELECT uuid FROM workspaces WHERE source = 'local'")) == expected_uuid
+            assert await conn.scalar(
+                sa.text("SELECT workspace_uuid FROM workspace_metadata WHERE key = 'migration_probe'")
+            ) == expected_uuid
+            assert await conn.scalar(
+                sa.text("SELECT value FROM metadata WHERE key = 'oss_workspace_uuid'")
+            ) == expected_uuid
+    finally:
+        await engine.dispose()
+
+
 async def test_persistence_startup_rejects_instance_uuid_drift(tmp_path, monkeypatch):
     engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "instance-drift.db"}')
     try:

tests/unit_tests/api/service/test_cloud_managed_model_protection.py

@@ -0,0 +1,112 @@
+"""Cloud Runtime write protection for the managed LangBot Models catalog."""
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+
+from langbot.pkg.api.http.service import model as model_service_module
+from langbot.pkg.api.http.service.model import (
+    EmbeddingModelsService,
+    LLMModelsService,
+    RerankModelsService,
+    _assert_cloud_managed_provider_mutable,
+)
+from langbot.pkg.cloud.model_catalog import LANGBOT_MODELS_PROVIDER_REQUESTER
+
+
+WORKSPACE = 'workspace-a'
+PROVIDER = 'managed-provider'
+MODEL = 'managed-model'
+
+
+@pytest.mark.asyncio
+async def test_managed_provider_guard_is_cloud_only(monkeypatch) -> None:
+    async def managed_provider(_ap, _context, provider_uuid):
+        assert provider_uuid == PROVIDER
+        return {'uuid': PROVIDER, 'requester': LANGBOT_MODELS_PROVIDER_REQUESTER}
+
+    monkeypatch.setattr(model_service_module, '_require_workspace_provider', managed_provider)
+    application = SimpleNamespace(persistence_mgr=SimpleNamespace(mode=SimpleNamespace(value='cloud_runtime')))
+
+    with pytest.raises(ValueError, match='managed by Cloud'):
+        await _assert_cloud_managed_provider_mutable(
+            application,
+            WORKSPACE,
+            PROVIDER,
+        )
+
+    application.persistence_mgr.mode.value = 'normal'
+    await _assert_cloud_managed_provider_mutable(
+        application,
+        WORKSPACE,
+        PROVIDER,
+    )
+
+
+@pytest.mark.parametrize(
+    ('service_type', 'create_method', 'model_data'),
+    [
+        (LLMModelsService, 'create_llm_model', {'provider_uuid': PROVIDER, 'name': 'chat', 'abilities': []}),
+        (EmbeddingModelsService, 'create_embedding_model', {'provider_uuid': PROVIDER, 'name': 'embedding'}),
+        (RerankModelsService, 'create_rerank_model', {'provider_uuid': PROVIDER, 'name': 'rerank'}),
+    ],
+)
+@pytest.mark.asyncio
+async def test_all_model_types_reject_creation_under_managed_provider(
+    monkeypatch,
+    service_type,
+    create_method: str,
+    model_data: dict,
+) -> None:
+    guard = AsyncMock(side_effect=ValueError('LangBot Models is managed by Cloud and cannot be modified'))
+    monkeypatch.setattr(model_service_module, '_assert_cloud_managed_provider_mutable', guard)
+    application = SimpleNamespace(
+        persistence_mgr=SimpleNamespace(),
+        provider_service=SimpleNamespace(
+            get_provider=AsyncMock(return_value={'uuid': PROVIDER, 'requester': LANGBOT_MODELS_PROVIDER_REQUESTER})
+        ),
+        model_mgr=None,
+    )
+    service = service_type(application)
+
+    with pytest.raises(ValueError, match='managed by Cloud'):
+        await getattr(service, create_method)(WORKSPACE, model_data)
+
+    guard.assert_awaited_once()
+
+
+@pytest.mark.parametrize(
+    ('service_type', 'get_method', 'write_method', 'payload'),
+    [
+        (LLMModelsService, 'get_llm_model', 'update_llm_model', {'name': 'changed'}),
+        (LLMModelsService, 'get_llm_model', 'delete_llm_model', None),
+        (EmbeddingModelsService, 'get_embedding_model', 'update_embedding_model', {'name': 'changed'}),
+        (EmbeddingModelsService, 'get_embedding_model', 'delete_embedding_model', None),
+        (RerankModelsService, 'get_rerank_model', 'update_rerank_model', {'name': 'changed'}),
+        (RerankModelsService, 'get_rerank_model', 'delete_rerank_model', None),
+    ],
+)
+@pytest.mark.asyncio
+async def test_all_model_types_reject_update_and_delete_for_managed_provider(
+    monkeypatch,
+    service_type,
+    get_method: str,
+    write_method: str,
+    payload: dict | None,
+) -> None:
+    guard = AsyncMock(side_effect=ValueError('LangBot Models is managed by Cloud and cannot be modified'))
+    monkeypatch.setattr(model_service_module, '_assert_cloud_managed_provider_mutable', guard)
+    application = SimpleNamespace(persistence_mgr=SimpleNamespace(mode=SimpleNamespace(value='cloud_runtime')))
+    service = service_type(application)
+    monkeypatch.setattr(
+        service,
+        get_method,
+        AsyncMock(return_value={'uuid': MODEL, 'provider_uuid': PROVIDER, 'extra_args': {}}),
+    )
+
+    args = (WORKSPACE, MODEL) if payload is None else (WORKSPACE, MODEL, payload)
+    with pytest.raises(ValueError, match='managed by Cloud'):
+        await getattr(service, write_method)(*args)
+
+    guard.assert_awaited_once()

tests/unit_tests/api/service/test_provider_service.py

@@ -25,6 +25,7 @@ from langbot.pkg.workspace.errors import WorkspaceNotFoundError
 pytestmark = pytest.mark.asyncio
 
 WORKSPACE_UUID = 'workspace-a'
+SYSTEM_REQUESTER = 'space-chat-completions'
 
 
 def _create_mock_provider(
@@ -1005,3 +1006,56 @@ class TestProviderSecretRoundtrip:
             )
 
         ap.persistence_mgr.execute_async.assert_not_awaited()
+
+
+class TestCloudManagedProviderProtection:
+    @staticmethod
+    def _service() -> ModelProviderService:
+        ap = SimpleNamespace(
+            persistence_mgr=SimpleNamespace(
+                mode=SimpleNamespace(value='cloud_runtime'),
+                execute_async=AsyncMock(),
+            ),
+            model_mgr=SimpleNamespace(),
+        )
+        return ModelProviderService(ap)
+
+    async def test_cloud_rejects_user_created_system_requester(self):
+        service = self._service()
+
+        with pytest.raises(ValueError, match='reserved'):
+            await service.create_provider(
+                WORKSPACE_UUID,
+                {
+                    'name': 'Fake LangBot Models',
+                    'requester': SYSTEM_REQUESTER,
+                    'base_url': 'https://example.invalid/v1',
+                    'api_keys': ['fake'],
+                },
+            )
+
+        with pytest.raises(ValueError, match='reserved'):
+            await service.find_or_create_provider(
+                WORKSPACE_UUID,
+                SYSTEM_REQUESTER,
+                'https://api.langbot.cloud/v1',
+                ['fake'],
+            )
+        service.ap.persistence_mgr.execute_async.assert_not_awaited()
+
+    async def test_cloud_rejects_update_and_delete_of_managed_provider(self):
+        service = self._service()
+        service.get_provider = AsyncMock(
+            return_value={'uuid': 'system-provider', 'requester': SYSTEM_REQUESTER}
+        )
+
+        with pytest.raises(ValueError, match='managed by Cloud'):
+            await service.update_provider(WORKSPACE_UUID, 'system-provider', {'name': 'Renamed'})
+        with pytest.raises(ValueError, match='managed by Cloud'):
+            await service.delete_provider(WORKSPACE_UUID, 'system-provider')
+        service.ap.persistence_mgr.execute_async.assert_not_awaited()
+
+    async def test_oss_does_not_reserve_space_requester(self):
+        ap = SimpleNamespace(persistence_mgr=SimpleNamespace(mode=SimpleNamespace(value='oss_compat')))
+        service = ModelProviderService(ap)
+        assert service._system_requester_is_reserved(SYSTEM_REQUESTER) is False

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/api/test_plugin_runtime_route_fence.py

@@ -46,6 +46,20 @@ def plugin_router_cls():
         yield PluginsRouterGroup
 
 
+@pytest.mark.asyncio
+async def test_authenticated_plugin_resource_fences_injected_workspace_context(plugin_router_cls):
+    connector = SimpleNamespace(
+        require_workspace_context=AsyncMock(return_value=CONTEXT),
+    )
+    router = object.__new__(plugin_router_cls)
+    router.ap = SimpleNamespace(plugin_connector=connector)
+
+    result = await router._require_authenticated_plugin_runtime_context(CONTEXT)
+
+    assert result == CONTEXT
+    connector.require_workspace_context.assert_awaited_once_with(CONTEXT)
+
+
 @pytest.mark.asyncio
 async def test_public_plugin_asset_route_is_disabled_for_multi_workspace_policy(plugin_router_cls):
     connector = SimpleNamespace(

tests/unit_tests/cloud/test_bootstrap.py

@@ -66,6 +66,10 @@ class _Provider:
     def __init__(self):
         self.manifest_provider = _Manifest()
 
+    async def fetch_model_catalog(self, instance_uuid: str):
+        del instance_uuid
+        raise AssertionError('not used by bootstrap contract tests')
+
     def bootstrap(self, *, instance_uuid: str, instance_config: dict):
         del instance_config
         return VerifiedCloudDeployment(
@@ -79,6 +83,7 @@ class _Provider:
             entitlement_provider=_Entitlements(),
             directory_provider=_Directory(),
             manifest_provider=self.manifest_provider,
+            model_catalog_provider=self,
             verification_key_id='root-2026',
         )
 
@@ -228,10 +233,23 @@ 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_model_catalog.py

@@ -0,0 +1,271 @@
+from __future__ import annotations
+
+import logging
+from datetime import UTC, datetime
+from types import SimpleNamespace
+
+import pytest
+import sqlalchemy
+from sqlalchemy.ext.asyncio import create_async_engine
+
+from langbot.pkg.cloud.model_catalog import (
+    CloudModelCatalogSnapshot,
+    CloudModelCatalogSyncService,
+    system_model_uuid,
+    system_provider_uuid,
+)
+from langbot.pkg.entity.persistence.base import Base
+from langbot.pkg.entity.persistence.model import EmbeddingModel, LLMModel, ModelProvider
+from langbot.pkg.entity.persistence.workspace import Workspace
+from langbot.pkg.persistence.mgr import PersistenceManager, PersistenceMode
+
+
+pytestmark = pytest.mark.asyncio
+INSTANCE_UUID = 'instance-model-catalog'
+WORKSPACE_A = '00000000-0000-4000-8000-000000000001'
+WORKSPACE_B = '00000000-0000-4000-8000-000000000002'
+OWNER_A = '10000000-0000-4000-8000-000000000001'
+OWNER_B = '10000000-0000-4000-8000-000000000002'
+
+
+class _CatalogProvider:
+    def __init__(self, snapshot: CloudModelCatalogSnapshot) -> None:
+        self.snapshot = snapshot
+
+    async def fetch_model_catalog(self, instance_uuid: str) -> CloudModelCatalogSnapshot:
+        assert instance_uuid == INSTANCE_UUID
+        return self.snapshot
+
+
+def _snapshot(
+    *,
+    key_a: str | None = 'owner-a-key',
+    model_id: str = 'gpt-test',
+    include_embedding: bool = True,
+) -> CloudModelCatalogSnapshot:
+    models = [
+        {
+            'uuid': 'upstream-chat',
+            'model_id': model_id,
+            'category': 'chat',
+            'llm_abilities': ['chat', 'vision'],
+            'is_featured': True,
+            'featured_order': 7,
+        }
+    ]
+    if include_embedding:
+        models.append(
+            {
+                'uuid': 'upstream-embedding',
+                'model_id': 'embedding-test',
+                'category': 'embedding',
+            }
+        )
+    return CloudModelCatalogSnapshot.model_validate(
+        {
+            'instance_uuid': INSTANCE_UUID,
+            'generated_at': datetime.now(UTC),
+            'base_url': 'https://api.langbot.cloud/v1/',
+            'models': models,
+            'workspaces': [
+                {
+                    'workspace_uuid': WORKSPACE_A,
+                    'owner_account_uuid': OWNER_A,
+                    'api_key': key_a,
+                },
+                {
+                    'workspace_uuid': WORKSPACE_B,
+                    'owner_account_uuid': OWNER_B,
+                    'api_key': 'owner-b-key',
+                },
+            ],
+        }
+    )
+
+
+async def test_catalog_reconciles_every_workspace_idempotently_and_tracks_owner_and_downlisting(tmp_path) -> None:
+    engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "model-catalog.db"}')
+    manager = PersistenceManager(object(), mode=PersistenceMode.CLOUD_RUNTIME)
+    manager.db = SimpleNamespace(get_engine=lambda: engine)
+    bindings = [
+        SimpleNamespace(instance_uuid=INSTANCE_UUID, workspace_uuid=WORKSPACE_A, placement_generation=1),
+        SimpleNamespace(instance_uuid=INSTANCE_UUID, workspace_uuid=WORKSPACE_B, placement_generation=1),
+    ]
+    workspace_service = SimpleNamespace(list_active_execution_bindings=lambda: _async_value(bindings))
+    reload_counter = _AsyncCounter()
+    runtime_reload = SimpleNamespace(load_models_from_db=reload_counter)
+    app = SimpleNamespace(
+        persistence_mgr=manager,
+        workspace_service=workspace_service,
+        model_mgr=runtime_reload,
+        logger=logging.getLogger(__name__),
+    )
+    provider = _CatalogProvider(_snapshot())
+    service = CloudModelCatalogSyncService(app, provider, INSTANCE_UUID)
+
+    try:
+        async with engine.begin() as connection:
+            await connection.run_sync(Base.metadata.create_all)
+            await connection.execute(
+                sqlalchemy.insert(Workspace),
+                [
+                    {
+                        'uuid': WORKSPACE_A,
+                        'instance_uuid': INSTANCE_UUID,
+                        'name': 'A',
+                        'slug': 'a',
+                        'source': 'cloud_projection',
+                    },
+                    {
+                        'uuid': WORKSPACE_B,
+                        'instance_uuid': INSTANCE_UUID,
+                        'name': 'B',
+                        'slug': 'b',
+                        'source': 'cloud_projection',
+                    },
+                ],
+            )
+            await connection.execute(
+                sqlalchemy.insert(ModelProvider).values(
+                    uuid='custom-provider',
+                    workspace_uuid=WORKSPACE_A,
+                    name='Custom',
+                    requester='openai-chat-completions',
+                    base_url='https://custom.example/v1',
+                    api_keys=['custom-key'],
+                )
+            )
+            await connection.execute(
+                sqlalchemy.insert(LLMModel).values(
+                    uuid='custom-model',
+                    workspace_uuid=WORKSPACE_A,
+                    name='custom-model',
+                    provider_uuid='custom-provider',
+                    abilities=['chat'],
+                    extra_args={},
+                    prefered_ranking=0,
+                )
+            )
+
+        first = await service.sync_once()
+        assert first == {'workspaces': 2, 'created': 6, 'updated': 0, 'deleted': 0}
+        assert reload_counter.calls == 1
+
+        async with engine.connect() as connection:
+            providers = (
+                await connection.execute(
+                    sqlalchemy.select(
+                        ModelProvider.uuid,
+                        ModelProvider.workspace_uuid,
+                        ModelProvider.api_keys,
+                    ).where(ModelProvider.requester == 'space-chat-completions')
+                )
+            ).all()
+            assert {item.workspace_uuid for item in providers} == {WORKSPACE_A, WORKSPACE_B}
+            assert {item.uuid for item in providers} == {
+                system_provider_uuid(WORKSPACE_A),
+                system_provider_uuid(WORKSPACE_B),
+            }
+            assert {item.workspace_uuid: item.api_keys for item in providers} == {
+                WORKSPACE_A: ['owner-a-key'],
+                WORKSPACE_B: ['owner-b-key'],
+            }
+            assert await connection.scalar(sqlalchemy.select(sqlalchemy.func.count()).select_from(LLMModel)) == 3
+            assert await connection.scalar(sqlalchemy.select(sqlalchemy.func.count()).select_from(EmbeddingModel)) == 2
+
+        second = await service.sync_once()
+        assert second == {'workspaces': 2, 'created': 0, 'updated': 0, 'deleted': 0}
+        assert reload_counter.calls == 1
+
+        provider.snapshot = _snapshot(
+            key_a='new-owner-key',
+            model_id='gpt-renamed',
+            include_embedding=False,
+        )
+        third = await service.sync_once()
+        assert third == {'workspaces': 2, 'created': 0, 'updated': 3, 'deleted': 2}
+        assert reload_counter.calls == 2
+
+        async with engine.connect() as connection:
+            provider_a_keys = await connection.scalar(
+                sqlalchemy.select(ModelProvider.api_keys).where(ModelProvider.uuid == system_provider_uuid(WORKSPACE_A))
+            )
+            assert provider_a_keys == ['new-owner-key']
+            system_model_names = (
+                (
+                    await connection.execute(
+                        sqlalchemy.select(LLMModel.name).where(
+                            LLMModel.provider_uuid.in_(
+                                [system_provider_uuid(WORKSPACE_A), system_provider_uuid(WORKSPACE_B)]
+                            )
+                        )
+                    )
+                )
+                .scalars()
+                .all()
+            )
+            assert set(system_model_names) == {'gpt-renamed'}
+            assert await connection.scalar(sqlalchemy.select(sqlalchemy.func.count()).select_from(EmbeddingModel)) == 0
+            assert (
+                await connection.scalar(
+                    sqlalchemy.select(sqlalchemy.func.count())
+                    .select_from(ModelProvider)
+                    .where(ModelProvider.uuid == 'custom-provider')
+                )
+                == 1
+            )
+            assert (
+                await connection.scalar(
+                    sqlalchemy.select(sqlalchemy.func.count())
+                    .select_from(LLMModel)
+                    .where(LLMModel.uuid == 'custom-model')
+                )
+                == 1
+            )
+
+        provider.snapshot = _snapshot(key_a=None, model_id='gpt-renamed', include_embedding=False)
+        fourth = await service.sync_once()
+        assert fourth == {'workspaces': 2, 'created': 0, 'updated': 1, 'deleted': 0}
+        assert reload_counter.calls == 3
+        async with engine.connect() as connection:
+            provider_a_keys = await connection.scalar(
+                sqlalchemy.select(ModelProvider.api_keys).where(ModelProvider.uuid == system_provider_uuid(WORKSPACE_A))
+            )
+            assert provider_a_keys == []
+    finally:
+        await engine.dispose()
+
+
+def test_workspace_scoped_ids_are_stable_and_secrets_are_redacted() -> None:
+    assert system_provider_uuid(WORKSPACE_A) == system_provider_uuid(WORKSPACE_A)
+    assert system_provider_uuid(WORKSPACE_A) != system_provider_uuid(WORKSPACE_B)
+    assert system_model_uuid(WORKSPACE_A, 'chat', 'upstream') != system_model_uuid(WORKSPACE_B, 'chat', 'upstream')
+    snapshot = _snapshot()
+    assert 'owner-a-key' not in repr(snapshot)
+
+
+async def test_snapshot_must_cover_every_active_workspace() -> None:
+    snapshot = _snapshot().model_copy(update={'workspaces': _snapshot().workspaces[:1]})
+    app = SimpleNamespace(
+        workspace_service=SimpleNamespace(
+            list_active_execution_bindings=lambda: _async_value(
+                [SimpleNamespace(workspace_uuid=WORKSPACE_A), SimpleNamespace(workspace_uuid=WORKSPACE_B)]
+            )
+        ),
+        logger=logging.getLogger(__name__),
+    )
+    service = CloudModelCatalogSyncService(app, _CatalogProvider(snapshot), INSTANCE_UUID)
+    with pytest.raises(ValueError, match='missing billing projections for 1 active Workspaces'):
+        await service.sync_once()
+
+
+async def _async_value(value):
+    return value
+
+
+class _AsyncCounter:
+    def __init__(self) -> None:
+        self.calls = 0
+
+    async def __call__(self) -> None:
+        self.calls += 1

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
@@ -57,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': {
@@ -86,6 +99,81 @@ async def test_consumes_valid_workspace_launch_assertion_once():
         await service.consume_assertion(token, expected_workspace_uuid=WORKSPACE_UUID)
 
 
+
+
+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['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)
+
+    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()
     now = int(time.time())

tests/unit_tests/telemetry/test_heartbeat.py

@@ -60,10 +60,12 @@ class TestBuildHeartbeatPayload:
     async def test_payload_shape(self):
         heartbeat = get_heartbeat_module()
         ap = make_app()
-        payload = await heartbeat.build_heartbeat_payload(ap)
+        payload = await heartbeat.build_heartbeat_payload(ap, workspace_uuid='workspace-a')
 
         assert payload['event_type'] == 'instance_heartbeat'
         assert payload['query_id'] == ''
+        assert payload['workspace_uuid'] == 'workspace-a'
+        assert 'instance_id' not in payload
         assert 'instance_create_ts' in payload
         assert 'timestamp' in payload
         f = payload['features']
@@ -86,7 +88,7 @@ class TestBuildHeartbeatPayload:
     @pytest.mark.asyncio
     async def test_payload_is_json_serializable(self):
         heartbeat = get_heartbeat_module()
-        payload = await heartbeat.build_heartbeat_payload(make_app())
+        payload = await heartbeat.build_heartbeat_payload(make_app(), workspace_uuid='workspace-a')
         json.dumps(payload)
 
     @pytest.mark.asyncio
@@ -94,7 +96,7 @@ class TestBuildHeartbeatPayload:
         heartbeat = get_heartbeat_module()
         ap = make_app()
         ap.persistence_mgr.execute_async = AsyncMock(side_effect=RuntimeError('db down'))
-        payload = await heartbeat.build_heartbeat_payload(ap)
+        payload = await heartbeat.build_heartbeat_payload(ap, workspace_uuid='workspace-a')
         assert payload['features']['pipeline_count'] == -1
 
     @pytest.mark.asyncio
@@ -106,31 +108,73 @@ class TestBuildHeartbeatPayload:
             side_effect=AssertionError('Cloud heartbeat must not issue per-tenant COUNTs')
         )
         ap.pipeline_mgr = SimpleNamespace(
-            _pipelines_by_key={'pipeline-a': object(), 'pipeline-b': object()},
+            _pipelines_by_key={
+                ('instance-a', 'workspace-a', 'pipeline-a'): object(),
+                ('instance-a', 'workspace-a', 'pipeline-b'): object(),
+            },
         )
+        adapter_a = type('WorkspaceAAdapter', (), {})()
+        adapter_b = type('WorkspaceBAdapter', (), {})()
+        ap.platform_mgr._bots_by_key = {
+            ('instance-a', 'workspace-a', 'bot-a'): SimpleNamespace(enable=True, adapter=adapter_a),
+        }
         ap.tool_mgr = SimpleNamespace(
             mcp_tool_loader=SimpleNamespace(
-                _sessions={'mcp-a': object(), 'mcp-b': object(), 'mcp-c': object()},
+                _sessions={
+                    ('instance-a', 'workspace-a', 1, 'mcp-a'): object(),
+                    ('instance-a', 'workspace-a', 1, 'mcp-b'): object(),
+                    ('instance-a', 'workspace-a', 1, 'mcp-c'): object(),
+                },
             ),
         )
         ap.rag_mgr = SimpleNamespace(
-            knowledge_bases={'kb-a': object()},
+            knowledge_bases={('workspace-a', 'kb-a'): object()},
+        )
+        ap.plugin_connector._workspace_installations = {
+            'workspace-a': {'plugin-a', 'plugin-b'},
+        }
+        ap.skill_mgr._skills_by_scope = {
+            ('instance-a', 'workspace-a', 1): {'skill-a': {}, 'skill-b': {}},
+            ('instance-a', 'workspace-b', 1): {'skill-c': {}},
+        }
+        ap.workspace_service.list_active_execution_bindings = AsyncMock(
+            return_value=[
+                SimpleNamespace(workspace_uuid='workspace-a', placement_generation=7),
+                SimpleNamespace(workspace_uuid='workspace-b', placement_generation=9),
+            ],
+        )
+        ap.platform_mgr._bots_by_key[('instance-a', 'workspace-b', 'bot-b')] = SimpleNamespace(
+            enable=True, adapter=adapter_b
         )
 
-        payload = await heartbeat.build_heartbeat_payload(ap)
+        payloads = await heartbeat.build_heartbeat_payloads(ap)
 
-        features = payload['features']
-        assert features['pipeline_count'] == 2
-        assert features['mcp_server_count'] == 3
-        assert features['knowledge_base_count'] == 1
-        assert features['bot_count'] == 1
+        assert [payload['workspace_uuid'] for payload in payloads] == ['workspace-a', 'workspace-b']
+        assert all('instance_id' not in payload for payload in payloads)
+        by_workspace = {payload['workspace_uuid']: payload['features'] for payload in payloads}
+        assert by_workspace['workspace-a']['pipeline_count'] == 2
+        assert by_workspace['workspace-a']['mcp_server_count'] == 3
+        assert by_workspace['workspace-a']['knowledge_base_count'] == 1
+        assert by_workspace['workspace-a']['bot_count'] == 1
+        assert by_workspace['workspace-a']['plugin_count'] == 2
+        assert by_workspace['workspace-a']['extension_count'] == 5
+        assert by_workspace['workspace-a']['skill_count'] == 2
+        assert by_workspace['workspace-a']['execution_generation'] == 7
+        assert by_workspace['workspace-a']['adapters'] == ['WorkspaceAAdapter']
+        assert by_workspace['workspace-b']['bot_count'] == 1
+        assert by_workspace['workspace-b']['pipeline_count'] == 0
+        assert by_workspace['workspace-b']['skill_count'] == 1
+        assert by_workspace['workspace-b']['execution_generation'] == 9
+        assert by_workspace['workspace-b']['adapters'] == ['WorkspaceBAdapter']
+        assert 'workspace_resources' not in by_workspace['workspace-a']
         ap.persistence_mgr.execute_async.assert_not_awaited()
+        ap.workspace_service.list_active_execution_bindings.assert_awaited_once()
 
     @pytest.mark.asyncio
     async def test_no_user_content_fields(self):
         """The heartbeat must never carry message content / credentials keys."""
         heartbeat = get_heartbeat_module()
-        payload = await heartbeat.build_heartbeat_payload(make_app())
+        payload = await heartbeat.build_heartbeat_payload(make_app(), workspace_uuid='workspace-a')
         flat = json.dumps(payload).lower()
         for forbidden in ('api_key', 'password', 'token', 'message_content'):
             assert forbidden not in flat

tests/unit_tests/telemetry/test_telemetry.py

@@ -569,6 +569,33 @@ class TestHTTPScenarios:
             await manager.send({'query_id': 'test'})
 
 
+class TestTelemetryManagedRuntimeAuthentication:
+    @pytest.mark.asyncio
+    async def test_send_includes_managed_runtime_token_header(self):
+        telemetry = get_telemetry_module()
+        mock_app = Mock()
+        mock_app.logger = Mock()
+        manager = telemetry.TelemetryManager(mock_app)
+        manager.telemetry_config = {'url': 'https://example.com'}
+        captured = {}
+
+        async def mock_post(url, json, headers):
+            captured['headers'] = headers
+            return Mock(status_code=200, text='', json=Mock(return_value={'code': 0}))
+
+        mock_client = Mock()
+        mock_client.post = mock_post
+        mock_client.__aenter__ = AsyncMock(return_value=mock_client)
+        mock_client.__aexit__ = AsyncMock(return_value=None)
+        with (
+            patch.dict('os.environ', {'LANGBOT_TELEMETRY_INGEST_TOKEN': 'managed-runtime-secret'}),
+            patch.object(httpx, 'AsyncClient', return_value=mock_client),
+        ):
+            await manager.send({'event_type': 'instance_heartbeat'})
+
+        assert captured['headers'] == {'X-LangBot-Telemetry-Token': 'managed-runtime-secret'}
+
+
 class TestStartSendTask:
     """Tests for start_send_task() method."""
 

tests/unit_tests/telemetry/test_workspace_identity.py

@@ -0,0 +1,31 @@
+from __future__ import annotations
+
+import uuid
+from types import SimpleNamespace
+
+
+def test_standard_oss_instance_id_aligns_to_embedded_uuid():
+    from langbot.pkg.workspace.identity import workspace_uuid_from_instance_id
+
+    instance_uuid = "a711d9e4-0953-443f-a0e9-7dd50193a79f"
+
+    assert workspace_uuid_from_instance_id(instance_uuid) == instance_uuid
+    assert workspace_uuid_from_instance_id(f"instance_{instance_uuid}") == instance_uuid
+
+
+def test_custom_legacy_instance_id_maps_to_stable_valid_uuid():
+    from langbot.pkg.workspace.identity import workspace_uuid_from_instance_id
+
+    first = workspace_uuid_from_instance_id("instance_migration_test")
+    second = workspace_uuid_from_instance_id("instance_migration_test")
+
+    assert first == second
+    assert str(uuid.UUID(first)) == first
+
+
+def test_query_telemetry_identity_uses_execution_workspace_only():
+    from langbot.pkg.telemetry.identity import workspace_identity
+
+    identity = workspace_identity(SimpleNamespace(workspace_uuid="workspace-a", instance_uuid="instance-a"))
+
+    assert identity == {"workspace_uuid": "workspace-a"}

tests/unit_tests/workspace/test_workspace_service.py

@@ -26,6 +26,7 @@ from langbot.pkg.workspace import (
     WorkspaceOwnerAlreadyExistsError,
     WorkspaceService,
 )
+from langbot.pkg.workspace.identity import workspace_uuid_from_instance_id
 from langbot.pkg.workspace.policy import CloudWorkspacePolicy
 
 
@@ -113,6 +114,7 @@ async def test_ensure_singleton_workspace_is_idempotent(workspace_test_context):
     first = await service.ensure_singleton_workspace()
     second = await service.ensure_singleton_workspace()
 
+    assert first.uuid == workspace_uuid_from_instance_id('instance_service_test')
     assert second.uuid == first.uuid
     async with session_factory() as session:
         assert await session.scalar(sqlalchemy.select(sqlalchemy.func.count()).select_from(Workspace)) == 1

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,8 +28,10 @@ import langbotIcon from '@/app/assets/langbot-logo.webp';
 
 type SpaceOAuthLoginResult = {
   token: string;
-  user: string;
+  user?: string;
   workspace_uuid?: string;
+  principal_type?: 'account' | 'support_admin';
+  actor_account_uuid?: string;
 };
 
 const pendingSpaceOAuthLogins = new Map<
@@ -63,6 +66,10 @@ 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'
@@ -94,6 +101,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 });
@@ -207,8 +224,29 @@ function SpaceOAuthCallbackContent() {
     const errorDescription = searchParams.get('error_description');
     const mode = searchParams.get('mode');
     const state = searchParams.get('state');
-    const workspaceUuid = searchParams.get('workspace_uuid');
-    const launchAssertion = searchParams.get('launch_assertion');
+    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;
 
     if (error) {
       setStatus('error');

web/src/app/home/components/home-sidebar/SidebarDataContext.tsx

@@ -166,6 +166,19 @@ export function SidebarDataProvider({
 
       // Deduplicate plugins by composite key (prefer debug over installed)
       const pluginMap = new Map<string, SidebarEntityItem>();
+      const pluginIconURLs = new Map<string, string>(
+        await Promise.all(
+          pluginsResp.plugins.map(async (plugin) => {
+            const meta = plugin.manifest.manifest.metadata;
+            const author = meta.author ?? '';
+            const name = meta.name;
+            const url = await httpClient
+              .getAuthenticatedPluginIconURL(author, name)
+              .catch(() => '');
+            return [`${author}/${name}`, url] as const;
+          }),
+        ),
+      );
       for (const plugin of pluginsResp.plugins) {
         const meta = plugin.manifest.manifest.metadata;
         const author = meta.author ?? '';
@@ -184,7 +197,7 @@ export function SidebarDataProvider({
         const item: SidebarEntityItem = {
           id: compositeKey,
           name: extractI18nObject(meta.label),
-          iconURL: httpClient.getPluginIconURL(author, name),
+          iconURL: pluginIconURLs.get(compositeKey) || '',
           installSource: plugin.install_source,
           installInfo: plugin.install_info,
           hasUpdate,
@@ -218,7 +231,7 @@ export function SidebarDataProvider({
                 pluginAuthor: author,
                 pluginName: name,
                 pluginLabel: label,
-                pluginIconURL: httpClient.getPluginIconURL(author, name),
+                pluginIconURL: pluginIconURLs.get(`${author}/${name}`) || '',
                 pageId: page.id,
                 path: page.path,
               });

web/src/app/home/knowledge/components/kb-form/KBForm.tsx

@@ -4,6 +4,7 @@ import { useForm } from 'react-hook-form';
 import { zodResolver } from '@hookform/resolvers/zod';
 import { z } from 'zod';
 import { useTranslation } from 'react-i18next';
+import { AuthenticatedPluginIcon } from '@/components/AuthenticatedPluginIcon';
 import { Input } from '@/components/ui/input';
 import EmojiPicker from '@/components/ui/emoji-picker';
 import {
@@ -428,12 +429,9 @@ export default function KBForm({
                             );
                             return (
                               <div className="flex items-center gap-2">
-                                <img
-                                  src={httpClient.getPluginIconURL(
-                                    author,
-                                    name,
-                                  )}
-                                  alt=""
+                                <AuthenticatedPluginIcon
+                                  author={author}
+                                  name={name}
                                   className="h-5 w-5 rounded"
                                 />
                                 <span>
@@ -459,12 +457,9 @@ export default function KBForm({
                               value={engine.plugin_id}
                             >
                               <div className="flex items-center gap-2">
-                                <img
-                                  src={httpClient.getPluginIconURL(
-                                    author,
-                                    name,
-                                  )}
-                                  alt=""
+                                <AuthenticatedPluginIcon
+                                  author={author}
+                                  name={name}
                                   className="h-5 w-5 rounded"
                                 />
                                 <span>{extractI18nObject(engine.name)}</span>

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/home/plugins/components/plugin-installed/ExtensionCardComponent.tsx

@@ -13,7 +13,7 @@ import {
   Puzzle,
 } from 'lucide-react';
 import { getCloudServiceClientSync, systemInfo } from '@/app/infra/http';
-import { httpClient } from '@/app/infra/http/HttpClient';
+import { useAuthenticatedPluginIcon } from '@/hooks/useAuthenticatedPluginResource';
 import { Button } from '@/components/ui/button';
 import { Card } from '@/components/ui/card';
 import {
@@ -39,6 +39,11 @@ export default function ExtensionCardComponent({
   const { t } = useTranslation();
   const [dropdownOpen, setDropdownOpen] = useState(false);
   const [iconFailed, setIconFailed] = useState(false);
+  const authenticatedIcon = useAuthenticatedPluginIcon(
+    cardVO.author,
+    cardVO.name,
+    cardVO.type === 'plugin',
+  );
 
   const FallbackIcon =
     cardVO.type === 'mcp'
@@ -47,8 +52,8 @@ export default function ExtensionCardComponent({
         ? Sparkles
         : Puzzle;
   const iconSrc =
-    cardVO.iconURL || httpClient.getPluginIconURL(cardVO.author, cardVO.name);
-  const showFallback = iconFailed || !iconSrc;
+    cardVO.type === 'plugin' ? authenticatedIcon.url : cardVO.iconURL;
+  const showFallback = iconFailed || authenticatedIcon.error || !iconSrc;
 
   const getTypeLabel = (type: ExtensionType) => {
     switch (type) {

web/src/app/home/plugins/components/plugin-installed/plugin-readme/PluginReadme.tsx

@@ -10,6 +10,74 @@ import rehypeSlug from 'rehype-slug';
 import rehypeAutolinkHeadings from 'rehype-autolink-headings';
 import { getAPILanguageCode } from '@/i18n/I18nProvider';
 import '@/styles/github-markdown.css';
+import { useAuthenticatedPluginAsset } from '@/hooks/useAuthenticatedPluginResource';
+
+function AuthenticatedReadmeImage({
+  author,
+  name,
+  filepath,
+  alt,
+  ...props
+}: {
+  author: string;
+  name: string;
+  filepath: string;
+  alt?: string;
+} & React.ImgHTMLAttributes<HTMLImageElement>) {
+  const { url, error } = useAuthenticatedPluginAsset(author, name, filepath);
+  if (error)
+    return (
+      <span className="text-sm text-muted-foreground">{alt || filepath}</span>
+    );
+  if (!url)
+    return (
+      <span className="inline-block h-6 w-24 animate-pulse rounded bg-muted" />
+    );
+  return (
+    <img
+      src={url}
+      alt={alt || ''}
+      className="max-w-lg h-auto my-4"
+      {...props}
+    />
+  );
+}
+
+function PluginReadmeImage({
+  author,
+  name,
+  src,
+  alt,
+  ...props
+}: {
+  author: string;
+  name: string;
+  src?: string;
+  alt?: string;
+} & React.ImgHTMLAttributes<HTMLImageElement>) {
+  const imageSrc = typeof src === 'string' ? src : '';
+  if (!imageSrc || /^(https?:\/\/|data:)/i.test(imageSrc)) {
+    return (
+      <img
+        src={imageSrc}
+        alt={alt || ''}
+        className="max-w-lg h-auto my-4"
+        {...props}
+      />
+    );
+  }
+  let filepath = imageSrc.replace(/^(\.\/|\/)+/, '');
+  filepath = filepath.replace(/^assets\//, '');
+  return (
+    <AuthenticatedReadmeImage
+      author={author}
+      name={name}
+      filepath={filepath}
+      alt={alt}
+      {...props}
+    />
+  );
+}
 
 export default function PluginReadme({
   pluginAuthor,
@@ -71,49 +139,15 @@ export default function PluginReadme({
                 <ol className="list-decimal">{children}</ol>
               ),
               li: ({ children }) => <li className="ml-4">{children}</li>,
-              img: ({ src, alt, ...props }) => {
-                let imageSrc = src || '';
-
-                if (typeof imageSrc !== 'string') {
-                  return (
-                    <img
-                      src={src}
-                      alt={alt || ''}
-                      className="max-w-full h-auto rounded-lg my-4"
-                      {...props}
-                    />
-                  );
-                }
-
-                if (
-                  imageSrc &&
-                  !imageSrc.startsWith('http://') &&
-                  !imageSrc.startsWith('https://') &&
-                  !imageSrc.startsWith('data:')
-                ) {
-                  imageSrc = imageSrc.replace(/^(\.\/|\/)+/, '');
-
-                  if (!imageSrc.startsWith('assets/')) {
-                    imageSrc = `assets/${imageSrc}`;
-                  }
-
-                  const assetPath = imageSrc.replace(/^assets\//, '');
-                  imageSrc = httpClient.getPluginAssetURL(
-                    pluginAuthor,
-                    pluginName,
-                    assetPath,
-                  );
-                }
-
-                return (
-                  <img
-                    src={imageSrc}
-                    alt={alt || ''}
-                    className="max-w-lg h-auto my-4"
-                    {...props}
-                  />
-                );
-              },
+              img: ({ src, alt, ...props }) => (
+                <PluginReadmeImage
+                  author={pluginAuthor}
+                  name={pluginName}
+                  src={typeof src === 'string' ? src : undefined}
+                  alt={alt}
+                  {...props}
+                />
+              ),
             }}
           >
             {readme}

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

@@ -710,6 +710,32 @@ export class BackendClient extends BaseHttpClient {
     );
   }
 
+  private async getAuthenticatedObjectURL(path: string): Promise<string> {
+    const response = await this.instance.get<Blob>(path, {
+      responseType: 'blob',
+    });
+    return URL.createObjectURL(response.data);
+  }
+
+  public getAuthenticatedPluginAssetURL(
+    author: string,
+    name: string,
+    filepath: string,
+  ): Promise<string> {
+    return this.getAuthenticatedObjectURL(
+      `/api/v1/plugins/${author}/${name}/authenticated-assets/${filepath}`,
+    );
+  }
+
+  public getAuthenticatedPluginIconURL(
+    author: string,
+    name: string,
+  ): Promise<string> {
+    return this.getAuthenticatedObjectURL(
+      `/api/v1/plugins/${author}/${name}/authenticated-icon`,
+    );
+  }
+
   public async pluginPageApi(
     author: string,
     name: string,
@@ -1327,8 +1353,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/src/components/AuthenticatedPluginIcon.tsx

@@ -0,0 +1,32 @@
+import { useAuthenticatedPluginIcon } from '@/hooks/useAuthenticatedPluginResource';
+import { cn } from '@/lib/utils';
+
+export function AuthenticatedPluginIcon({
+  author,
+  name,
+  alt = '',
+  className,
+}: {
+  author: string;
+  name: string;
+  alt?: string;
+  className?: string;
+}) {
+  const icon = useAuthenticatedPluginIcon(
+    author,
+    name,
+    Boolean(author && name),
+  );
+
+  if (!icon.url || icon.error) {
+    return (
+      <span
+        aria-hidden={alt ? undefined : true}
+        aria-label={alt || undefined}
+        className={cn('inline-block bg-muted', className)}
+      />
+    );
+  }
+
+  return <img src={icon.url} alt={alt} className={className} />;
+}

web/src/hooks/useAuthenticatedPluginResource.ts

@@ -0,0 +1,71 @@
+import { useEffect, useState } from 'react';
+import { httpClient } from '@/app/infra/http/HttpClient';
+
+export function useAuthenticatedPluginIcon(
+  author: string,
+  name: string,
+  enabled = true,
+): { url: string; error: boolean } {
+  const [url, setURL] = useState('');
+  const [error, setError] = useState(false);
+
+  useEffect(() => {
+    if (!enabled) {
+      setURL('');
+      setError(false);
+      return;
+    }
+    let active = true;
+    let objectURL = '';
+    setURL('');
+    setError(false);
+    httpClient
+      .getAuthenticatedPluginIconURL(author, name)
+      .then((nextURL) => {
+        objectURL = nextURL;
+        if (active) setURL(nextURL);
+        else URL.revokeObjectURL(nextURL);
+      })
+      .catch(() => {
+        if (active) setError(true);
+      });
+    return () => {
+      active = false;
+      if (objectURL) URL.revokeObjectURL(objectURL);
+    };
+  }, [author, enabled, name]);
+
+  return { url, error };
+}
+
+export function useAuthenticatedPluginAsset(
+  author: string,
+  name: string,
+  filepath: string,
+): { url: string; error: boolean } {
+  const [url, setURL] = useState('');
+  const [error, setError] = useState(false);
+
+  useEffect(() => {
+    let active = true;
+    let objectURL = '';
+    setURL('');
+    setError(false);
+    httpClient
+      .getAuthenticatedPluginAssetURL(author, name, filepath)
+      .then((nextURL) => {
+        objectURL = nextURL;
+        if (active) setURL(nextURL);
+        else URL.revokeObjectURL(nextURL);
+      })
+      .catch(() => {
+        if (active) setError(true);
+      });
+    return () => {
+      active = false;
+      if (objectURL) URL.revokeObjectURL(objectURL);
+    };
+  }, [author, name, filepath]);
+
+  return { url, error };
+}

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

@@ -0,0 +1,18 @@
+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');
+});

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\(\)/);
+});