fix(pipeline): ground local-agent system prompt with current date (#2399 )

The local-agent runner's system prompt is a static string with no template-variable support, so the model had no anchor for "today" and resolved relative time references (e.g. "this quarter", "latest") against whichever period was best represented in training data instead of the real date, sometimes confidently answering with stale information for time-sensitive questions. PreProcessor now appends a short, deterministically-computed "Current date: ..." note to the system prompt on every request for local-agent pipelines, alongside guidance to verify time-sensitive facts with a search tool rather than answering from memory. The existing skill-awareness prompt injection is refactored to share the same append-to-system-prompt helper.
fix(cloud): preserve tenant scope for extension tasks (#2408 )
2026-08-08 20:30:59 +00:00 · 2026-08-08 22:29:54 +08:00 · 2026-08-07 11:32:38 +08:00 · 2026-08-07 11:26:42 +08:00 · 2026-08-07 11:26:06 +08:00 · 2026-08-06 20:39:20 +08:00
165 changed files with 11755 additions and 4459 deletions
@@ -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: 58253c53933f95d81b035fbe2efedb55b6c1a82b
 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
@@ -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
@@ -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:
@@ -14,8 +14,8 @@ services:
    restart: on-failure
    environment:
      - TZ=Asia/Shanghai
-      # Shared with the langbot service and sent only as a WebSocket handshake
+      # Optional. Leave unset on both OSS services, or set the same value on
-      # header. Generate with: openssl rand -hex 32
+      # both to protect the control WebSocket. Generate with: openssl rand -hex 32
      - LANGBOT_PLUGIN_RUNTIME_CONTROL_TOKEN=${LANGBOT_PLUGIN_RUNTIME_CONTROL_TOKEN:-}
      # Process-wide admission for every asyncio.to_thread() call.
      - LANGBOT_BLOCKING_EXECUTOR_MAX_WORKERS=${LANGBOT_BLOCKING_EXECUTOR_MAX_WORKERS:-8}
@@ -77,8 +77,7 @@ services:
    restart: on-failure
    environment:
      - TZ=Asia/Shanghai
-      # Must match langbot_plugin_runtime. Empty/missing values make the
+      # Optional. Leave unset on both OSS services, or match plugin Runtime.
      # external control channel fail closed.
      - LANGBOT_PLUGIN_RUNTIME_CONTROL_TOKEN=${LANGBOT_PLUGIN_RUNTIME_CONTROL_TOKEN:-}
      # Must match the value supplied to langbot_box. The token is sent only
      # in WebSocket handshake headers, never in URLs or action payloads.
@@ -0,0 +1,166 @@
 # LangBot 多租户数据库迁移指南
 ## 概述
 LangBot 从单租户 OSS 架构迁移到多租户 SaaS 架构，需要执行 7 个数据库迁移（0009-0015）。
 ## 迁移序列
 ```
 0009_workspace_tenancy_kernel → 创建 Workspace、成员、邀请表
 0010_scope_tenant_resources → 所有业务表添加 workspace_uuid
 0011_postgres_tenant_rls → PostgreSQL 行级安全策略
 0012_plugin_installation_identity → 插件实例租户绑定
 0013_tenant_pgvector → RAG 向量存储隔离
 0014_cloud_directory_projection → Cloud 控制平面同步
 0015_cloud_core_collaboration → 协作和权限功能
 ```
 ## 执行步骤
 ### 1. 备份（必须）
 ```bash
 # SQLite
 cp ~/.langbot/data/langbot.db ~/.langbot/data/langbot.db.backup-$(date +%Y%m%d)
 # PostgreSQL  
 pg_dump -U langbot_user -d langbot_db -F c -f langbot_backup_$(date +%Y%m%d).dump
 ```
 ### 2. 执行迁移
 ```bash
 # 停止服务
 sudo -S -p '' systemctl stop langbot
 # 执行迁移
 python -m langbot.pkg.persistence.migration upgrade head
 # 验证
 python -m langbot.pkg.persistence.migration current
 # 预期: 0015_cloud_core_collaboration
 # 启动服务
 sudo -S -p '' systemctl start langbot
 ```
 ### 3. OSS 单租户自动迁移
 迁移会自动：
 - 创建默认 Workspace（名称："Default Workspace"）
 - 第一个用户成为 Owner
 - 所有现有资源绑定到该 Workspace
 ### 4. 验证检查
 ```bash
 # 检查 Workspace
 python << EOF
 from langbot.pkg.persistence import manager
 import sqlalchemy as sa
 with manager.engine.connect() as conn:
    ws = conn.execute(sa.text("SELECT uuid, name FROM workspaces LIMIT 1")).first()
    print(f"Workspace: {ws[1]} ({ws[0]})")
    # 检查资源绑定
    bot_count = conn.execute(sa.text(
        f"SELECT COUNT(*) FROM bots WHERE workspace_uuid='{ws[0]}'"
    )).scalar()
    print(f"Bots: {bot_count}")
 EOF
 ```
 ## 回滚方案
 ### 完全回滚（丢失多租户数据）
 ```bash
 # 1. 停止服务
 sudo -S -p '' systemctl stop langbot
 # 2. 恢复备份
 cp ~/.langbot/data/langbot.db.backup-YYYYMMDD ~/.langbot/data/langbot.db
 # 3. 回退代码
 git checkout v4.10.x
 pip install -e .
 # 4. 启动
 sudo -S -p '' systemctl start langbot
 ```
 ### 降级迁移（保留数据但移除多租户）
 ```bash
 # 警告：会移除 Workspace 表但保留资源
 python -m langbot.pkg.persistence.migration downgrade 0008_mcp_resource_prefs
 ```
 ## 常见问题
 ### Q: 迁移后无法登录
 ```bash
 # 检查用户 UUID
 python << EOF
 from langbot.pkg.persistence import manager
 import sqlalchemy as sa
 with manager.engine.connect() as conn:
    users = conn.execute(sa.text("SELECT id, user, uuid, status FROM users")).all()
    for u in users:
        print(f"{u[1]}: UUID={u[2]}, Status={u[3]}")
 EOF
 ```
 ### Q: 资源看不见了
 检查 Workspace 上下文：
 ```bash
 # 前端请求需要带 X-Workspace-ID header
 curl -H "Authorization: Bearer $TOKEN" \
     -H "X-Workspace-ID: $WORKSPACE_UUID" \
     http://localhost:5200/api/v1/platform/bots
 ```
 ### Q: 迁移速度慢
 ```bash
 # SQLite 优化
 sqlite3 ~/.langbot/data/langbot.db << EOF
 PRAGMA journal_mode=WAL;
 PRAGMA synchronous=NORMAL;
 VACUUM;
 EOF
 ```
 ## 性能调优
 ### PostgreSQL 索引
 ```sql
 -- 迁移后创建
 CREATE INDEX CONCURRENTLY idx_model_providers_workspace 
    ON model_providers(workspace_uuid);
 CREATE INDEX CONCURRENTLY idx_bots_workspace 
    ON bots(workspace_uuid);
 CREATE INDEX CONCURRENTLY idx_pipelines_workspace 
    ON pipelines(workspace_uuid);
 ```
 ## 预估时间
 - SQLite < 100MB: 2-5 分钟
 - SQLite 100MB-1GB: 5-15 分钟  
 - PostgreSQL: < 5 分钟（取决于数据量）
 ## 支持
 问题反馈：https://github.com/langbot-app/LangBot/issues
 **版本**: 1.0  
 **最后更新**: 2026-07-30
@@ -1,6 +1,6 @@
 [project]
 name = "langbot"
-version = "4.10.6"
+version = "4.10.7"
 description = "Production-grade platform for building agentic IM bots"
 readme = "README.md"
 license-files = ["LICENSE"]
@@ -71,7 +71,7 @@ dependencies = [
    "chromadb>=1.0.0,<2.0.0",
    "qdrant-client (>=1.15.1,<2.0.0)",
    "pyseekdb==1.1.0.post3",
-    "langbot-plugin @ git+https://github.com/langbot-app/langbot-plugin-sdk.git@1d65ed301a6afc52150a998043f73cd6032c8162",
+    "langbot-plugin @ git+https://github.com/langbot-app/langbot-plugin-sdk.git@9d216208cdfb41f0cb7fcb64632e2a46816d6dc6",
    "asyncpg>=0.30.0",
    "line-bot-sdk>=3.19.0",
    "matrix-nio>=0.25.2",
@@ -32,12 +32,13 @@ The `all` / `box` profile starts three services:
  the LangBot and Box containers. Generate it once with `openssl rand -hex 32`;
  never put it in `box.runtime.endpoint` or commit it to config.
-Every Compose deployment also needs one
+A Compose deployment may optionally set
-`LANGBOT_PLUGIN_RUNTIME_CONTROL_TOKEN` shared by `langbot` and
+`LANGBOT_PLUGIN_RUNTIME_CONTROL_TOKEN` on both `langbot` and
-`langbot_plugin_runtime`. Generate it with `openssl rand -hex 32` and export it
+`langbot_plugin_runtime` when port 5400 needs shared-secret protection. OSS
-before `docker compose up`; the external Plugin Runtime fails closed when the
+defaults to leaving it unset on both sides. If enabled, generate one value with
-token is empty or weak. Kubernetes uses the `langbot-plugin-runtime-control`
+`openssl rand -hex 32`; configuring only one side causes the control connection
-Secret shown in `docker/kubernetes.yaml`.
+to fail. Kubernetes may use the `langbot-plugin-runtime-control` Secret shown in
 `docker/kubernetes.yaml`.
 With Box off, the dashboard/skills list stays visible (read-only) but sandbox
 tools, skill add/edit, and stdio MCP are disabled. Set `box.enabled: false`
@@ -19,7 +19,6 @@ class Permission(enum.StrEnum):
    WORKSPACE_VIEW = 'workspace.view'
    WORKSPACE_UPDATE = 'workspace.update'
    WORKSPACE_DELETE = 'workspace.delete'
    OWNER_TRANSFER = 'owner.transfer'
    MEMBER_VIEW = 'member.view'
    MEMBER_INVITE = 'member.invite'
    MEMBER_UPDATE_ROLE = 'member.update_role'
@@ -49,7 +48,6 @@ _ROLE_PERMISSIONS: typing.Final = types.MappingProxyType(
            if permission
            not in {
                Permission.WORKSPACE_DELETE,
                Permission.OWNER_TRANSFER,
                Permission.BILLING_LINK_MANAGE,
            }
        ),
@@ -74,6 +72,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'
@@ -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)
@@ -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,16 @@ class AuthType(enum.Enum):
    USER_TOKEN_OR_API_KEY = 'user-token-or-api-key'
 _SUPPORT_ADMIN_DENIED_PERMISSIONS = frozenset(
    {
        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 +114,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 +134,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._authenticate_support_admin(token, auth_type)
-                        request_context = await self._resolve_account_context(account, 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 +169,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', '')
+                    elif api_key := quart.request.headers.get('X-API-Key', ''):
                    if api_key:
                        # API key authentication
                        try:
                            request_context = await self._authenticate_api_key(api_key, auth_type)
@@ -155,8 +193,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 +304,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)
@@ -23,8 +23,13 @@ def _storage_owner(context: RequestContext) -> str:
@group.group_class('files', '/api/v1/files')
 class FilesRouterGroup(group.RouterGroup):
    async def initialize(self) -> None:
-        @self.route('/image/<path:image_key>', methods=['GET'], auth_type=group.AuthType.NONE)
+        @self.route(
-        async def _(image_key: str) -> quart.Response:
+            '/image/<path:image_key>',
            methods=['GET'],
            auth_type=group.AuthType.USER_TOKEN_OR_API_KEY,
            permission=Permission.RESOURCE_VIEW,
        )
        async def _(image_key: str, request_context: RequestContext) -> quart.Response:
            image_bytes = await self.ap.storage_mgr.resolve_public_object(
                image_key,
                expected_owner_type='upload_image',
@@ -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)
@@ -128,9 +138,26 @@ class WebSocketChatRouterGroup(group.RouterGroup):
        self,
        request_context: RequestContext,
        token: str,
-    ) -> None:
+    ) -> 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:
@@ -168,6 +195,7 @@ class WebSocketChatRouterGroup(group.RouterGroup):
            entitlement_revision=request_context.entitlement_revision,
        )
        require_permission(current_context, Permission.RUNTIME_OPERATE)
        return current_context
    async def _get_scoped_adapter(self, request_context: RequestContext, pipeline_uuid: str):
        pipeline = await run_in_workspace_uow(
@@ -210,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', {})
@@ -390,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
@@ -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:
@@ -15,7 +15,6 @@ import posixpath
 import sqlalchemy
 from .....core import taskmgr
 from .....core.task_boundary import run_in_workspace_uow
 from .....entity.persistence import plugin as persistence_plugin
 from ...authz import Permission
 from ...context import ExecutionContext, RequestContext
@@ -311,13 +310,22 @@ class PluginsRouterGroup(group.RouterGroup):
    ):
        """Revalidate a captured task context immediately before Runtime I/O."""
-        await run_in_workspace_uow(
+        persistence_mgr = getattr(self.ap, 'persistence_mgr', None)
-            self.ap,
+        tenant_scope = getattr(persistence_mgr, 'tenant_scope', None)
-            execution_context.workspace_uuid,
+        if callable(tenant_scope):
-            lambda: self.ap.plugin_connector.require_workspace_context(execution_context),
+            async with tenant_scope(execution_context.workspace_uuid):
-        )
+                await self.ap.plugin_connector.require_workspace_context(execution_context)
                return await operation()
        await self.ap.plugin_connector.require_workspace_context(execution_context)
        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 +380,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,8 +393,8 @@ 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)
+            execution_context = await self._require_authenticated_plugin_runtime_context(request_context)
-            debug_info = await self.ap.plugin_connector.get_debug_info()
+            debug_info = await self.ap.plugin_connector.get_debug_info(execution_context)
            # Get debug URL from config
            plugin_config = self.ap.instance_config.data.get('plugin', {})
@@ -396,6 +404,7 @@ class PluginsRouterGroup(group.RouterGroup):
                data={
                    'debug_url': debug_url,
                    'plugin_debug_key': debug_info.get('plugin_debug_key', ''),
                    'expires_at': debug_info.get('expires_at', ''),
                }
            )
@@ -428,7 +437,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 +478,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 +498,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 +515,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 +526,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 +538,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 +547,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 +643,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 +672,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 +752,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 +948,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 +989,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 +1021,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')
@@ -2,6 +2,7 @@ from __future__ import annotations
 import quart
 from langbot.pkg.cloud.entitlements import EntitlementFeatureUnavailableError
 from langbot_plugin.box.errors import BoxError
 from ...authz import Permission
@@ -23,6 +24,11 @@ class SkillsRouterGroup(group.RouterGroup):
        async def list_skills(request_context: RequestContext) -> quart.Response:
            try:
                skills = await self.ap.skill_service.list_skills(request_context)
            except EntitlementFeatureUnavailableError:
                # Plans without managed sandbox support have no runnable skills.
                # Treat that capability absence as an empty collection so the
                # shared UI can render normally instead of surfacing a 500.
                return self.success(data={'skills': []})
            except (ValueError, BoxError) as exc:
                return self.http_status(400, -1, str(exc))
            return self.success(data={'skills': skills})
@@ -1,6 +1,7 @@
 import quart
 import argon2
 import asyncio
 import datetime
 import uuid
 from urllib.parse import parse_qs, urlsplit
@@ -218,7 +219,22 @@ class UserRouterGroup(group.RouterGroup):
            try:
                consumed_state = await self.ap.user_service.consume_space_oauth_state_details(state, 'login')
                # Exchange code for tokens
-                token_data = await self.ap.space_service.exchange_oauth_code(code)
+                launch_workspace_uuid = consumed_state.launch_workspace_uuid
                workspace_uuids = [launch_workspace_uuid] if launch_workspace_uuid else []
                workspace_created_ats: dict[str, int] = {}
                if not workspace_uuids and getattr(getattr(self.ap, 'deployment', None), 'mode', 'oss') != 'cloud':
                    binding = await self.ap.workspace_service.get_execution_binding()
                    workspace_uuids = [binding.workspace_uuid]
                    workspace_created_at = binding.workspace_created_at
                    if workspace_created_at is not None:
                        if workspace_created_at.tzinfo is None:
                            workspace_created_at = workspace_created_at.replace(tzinfo=datetime.UTC)
                        workspace_created_ats[binding.workspace_uuid] = int(workspace_created_at.timestamp())
                token_data = await self.ap.space_service.exchange_oauth_code(
                    code,
                    workspace_uuids,
                    workspace_created_ats,
                )
                access_token = token_data.get('access_token')
                refresh_token = token_data.get('refresh_token')
                expires_in = token_data.get('expires_in', 0)
@@ -231,7 +247,6 @@ class UserRouterGroup(group.RouterGroup):
                    access_token, refresh_token, expires_in
                )
                launch_workspace_uuid = consumed_state.launch_workspace_uuid
                if launch_workspace_uuid:
                    try:
                        access = await self.ap.workspace_collaboration_service.resolve_account_workspace(
@@ -285,8 +300,25 @@ class UserRouterGroup(group.RouterGroup):
                request_context.workspace_uuid,
            )
            owner = await self.ap.user_service.get_workspace_owner(access.workspace.uuid)
-            owner_space_bound = bool(owner and owner.space_account_uuid)
+            cloud_mode = getattr(getattr(self.ap, 'deployment', None), 'mode', 'oss') == 'cloud'
-            credits = await self.ap.space_service.get_credits(owner.user) if owner_space_bound else None
+            owner_has_local_space_credentials = bool(owner and owner.space_account_uuid)
            # Cloud Accounts authenticate through LangBot Account, so every projected
            # Workspace owner is already bound even when this Core has no local OAuth
            # token row (model billing uses the owner's control-plane API key).
            owner_space_bound = cloud_mode or owner_has_local_space_credentials
            if cloud_mode:
                catalog_service = getattr(self.ap, 'cloud_model_catalog_service', None)
                credits = (
                    catalog_service.get_workspace_credits(access.workspace.uuid)
                    if catalog_service is not None
                    else None
                )
            else:
                credits = (
                    await self.ap.space_service.get_credits(owner.user)
                    if owner is not None and owner.space_account_uuid
                    else None
                )
            return self.success(
                data={
                    'credits': credits,
@@ -302,8 +334,10 @@ class UserRouterGroup(group.RouterGroup):
                return self.success(data={'initialized': False})
            capabilities = await self.ap.user_service.get_login_capabilities()
-            if getattr(getattr(self.ap, 'deployment', None), 'mode', 'oss') == 'cloud':
+            cloud_mode = getattr(getattr(self.ap, 'deployment', None), 'mode', 'oss') == 'cloud'
            if cloud_mode:
                capabilities['password_login_enabled'] = False
            capabilities['authenticated_invitation_acceptance_enabled'] = cloud_mode
            return self.success(data={'initialized': True, **capabilities})
        @self.route('/set-password', methods=['POST'], auth_type=group.AuthType.USER_TOKEN)
@@ -396,6 +430,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')
@@ -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
@@ -30,12 +30,14 @@ def _workspace_payload(workspace: Workspace) -> dict[str, typing.Any]:
 def _membership_payload(
    membership: WorkspaceMembership,
    *,
    display_name: str,
    email: str,
 ) -> dict[str, typing.Any]:
    return {
        'uuid': membership.uuid,
        'workspace_uuid': membership.workspace_uuid,
        'account_uuid': membership.account_uuid,
        'display_name': display_name,
        'email': email,
        'role': membership.role,
        'status': membership.status,
@@ -94,7 +96,11 @@ class WorkspacesRouterGroup(group.RouterGroup):
                workspaces.append(
                    {
                        'workspace': _workspace_payload(access.workspace),
-                        'membership': _membership_payload(access.membership, email=account.user),
+                        'membership': _membership_payload(
                            access.membership,
                            display_name=account.user,
                            email=account.normalized_email,
                        ),
                        'permissions': sorted(permissions_for_role(access.membership.role)),
                        'placement_generation': access.execution.placement_generation,
                        'plan_name': plan_name,
@@ -120,9 +126,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,10 +135,37 @@ 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,
                            'display_name': 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),
-                    'membership': _membership_payload(membership, email=account.user),
+                    'membership': _membership_payload(
                        membership,
                        display_name=account.user,
                        email=account.normalized_email,
                    ),
                    'permissions': sorted(request_context.workspace.permissions),
                    'placement_generation': request_context.placement_generation,
                    'plan_name': plan_name,
@@ -264,7 +294,8 @@ class WorkspacesRouterGroup(group.RouterGroup):
                data={
                    'member': _membership_payload(
                        member,
-                        email=account.user if account is not None else '',
+                        display_name=account.user if account is not None else '',
                        email=account.normalized_email if account is not None else '',
                    )
                }
            )
@@ -283,7 +314,11 @@ class WorkspacesRouterGroup(group.RouterGroup):
    @staticmethod
    def _member_view_payload(view: WorkspaceMemberView) -> dict[str, typing.Any]:
-        return _membership_payload(view.membership, email=view.email)
+        return _membership_payload(
            view.membership,
            display_name=view.display_name,
            email=view.email,
        )
@group.group_class('invitations', '/api/v1/invitations')
@@ -56,6 +56,14 @@ class KnowledgeService:
        require_workspace_uuid(context)
        # In new architecture, we delegate entirely to RAGManager which uses plugins.
        # Legacy internal KB creation is removed.
        limitation = (
            getattr(getattr(self.ap, 'instance_config', None), 'data', {}).get('system', {}).get('limitation', {})
        )
        max_knowledge_bases = limitation.get('max_knowledge_bases', -1)
        if max_knowledge_bases >= 0:
            knowledge_bases = await self.ap.rag_mgr.get_all_knowledge_base_details(context)
            if len(knowledge_bases) >= max_knowledge_bases:
                raise ValueError(f'Maximum number of knowledge bases ({max_knowledge_bases}) reached')
        knowledge_engine_plugin_id = kb_data.get('knowledge_engine_plugin_id')
        if not knowledge_engine_plugin_id:
@@ -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(
@@ -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))
@@ -331,8 +354,15 @@ class ModelProviderService:
        embedding_models = await self.ap.embedding_models_service.get_embedding_models_by_provider(
            context, provider_uuid
        )
        rerank_service = getattr(self.ap, 'rerank_models_service', None)
        rerank_models = (
            await rerank_service.get_rerank_models_by_provider(context, provider_uuid)
            if rerank_service is not None
            else []
        )
        existing_llm_names = {model['name'] for model in llm_models}
        existing_embedding_names = {model['name'] for model in embedding_models}
        existing_rerank_names = {model['name'] for model in rerank_models}
        filtered_models = []
        for model in scanned_models:
@@ -359,6 +389,8 @@ class ModelProviderService:
                    'already_added': (
                        model_name in existing_embedding_names
                        if scanned_type == 'embedding'
                        else model_name in existing_rerank_names
                        if scanned_type == 'rerank'
                        else model_name in existing_llm_names
                    ),
                }
@@ -12,7 +12,6 @@ from urllib.parse import quote, unquote, urlparse
 import httpx
 from ....cloud.entitlements import EntitlementFeatureUnavailableError
 from ....core import app
 from ....skill.utils import parse_frontmatter
 from ....utils import httpclient
@@ -120,13 +119,7 @@ class SkillService:
        box_service = self._box_service()
        if box_service is None:
            return []
-        try:
+        return [self._serialize_skill(skill) for skill in await box_service.list_skills(execution_context)]
            skills = await box_service.list_skills(execution_context)
        except EntitlementFeatureUnavailableError as error:
            if error.feature == 'managed_sandbox':
                return []
            raise
        return [self._serialize_skill(skill) for skill in skills]
    async def get_skill(self, context: TenantContext, skill_name: str) -> Optional[dict]:
        execution_context = await self._execution_context(context)
@@ -59,6 +59,10 @@ class SpaceService:
        result_list = result.all()
        return result_list[0] if result_list else None
    async def get_valid_access_token(self, user_email: str) -> str | None:
        """Return a current Space bearer, refreshing and persisting it when needed."""
        return await self._ensure_valid_token(user_email)
    async def _ensure_valid_token(self, user_email: str) -> str | None:
        """Ensure access token is valid, refresh if expired. Returns valid access_token or None."""
        user_obj = await self._get_user_by_email(user_email)
@@ -117,7 +121,12 @@ class SpaceService:
            params['state'] = state
        return f'{authorize_url}?{urlencode(params)}'
-    async def exchange_oauth_code(self, code: str) -> typing.Dict:
+    async def exchange_oauth_code(
        self,
        code: str,
        workspace_uuids: list[str] | None = None,
        workspace_created_ats: dict[str, int] | None = None,
    ) -> typing.Dict:
        """Exchange OAuth authorization code for tokens"""
        from langbot.pkg.utils import constants
@@ -127,7 +136,14 @@ class SpaceService:
        session = httpclient.get_session()
        async with session.post(
            f'{space_url}/api/v1/accounts/oauth/token',
-            json={'code': code, 'instance_id': constants.instance_id},
+            json={
                'code': code,
                'instance_id': constants.instance_id,
                # Sending an explicit empty list tells new Space servers not to
                # synthesize a legacy instance-derived Workspace binding.
                'workspace_uuids': workspace_uuids if workspace_uuids is not None else [],
                'workspace_created_ats': workspace_created_ats or {},
            },
        ) as response:
            if response.status != 200:
                error = await httpclient.read_text_limited(response)
@@ -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],
@@ -776,8 +779,27 @@ class UserService:
        local_account = await self.get_user_by_email(user_email)
        if local_account is None:
            raise ValueError('User not found')
-        # Exchange code for tokens
+        # Exchange code for tokens and bind both installation and the active
-        token_data = await self.ap.space_service.exchange_oauth_code(code)
+        # OSS Workspace as independent identities.
        workspace_service = getattr(self.ap, 'workspace_service', None)
        if workspace_service is not None:
            binding = await workspace_service.get_execution_binding()
            created_at = binding.workspace_created_at
            created_ts = (
                int(created_at.replace(tzinfo=datetime.timezone.utc).timestamp())
                if created_at.tzinfo is None
                else int(created_at.timestamp())
            )
            token_data = await self.ap.space_service.exchange_oauth_code(
                code,
                [binding.workspace_uuid],
                {binding.workspace_uuid: created_ts},
            )
        else:
            # Compatibility for early/bootstrap call sites that have not wired
            # WorkspaceService yet; old Space servers still derive the legacy
            # Workspace identity from instance_id when the field is omitted.
            token_data = await self.ap.space_service.exchange_oauth_code(code)
        access_token = token_data.get('access_token')
        refresh_token = token_data.get('refresh_token')
        expires_in = token_data.get('expires_in', 0)
@@ -13,11 +13,12 @@ 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'
 REQUIRED_TENANT_ISOLATION_VERSION = 2
-SUPPORTED_PGVECTOR_DIMENSIONS = frozenset({384, 512, 768, 1024, 1536})
+SUPPORTED_PGVECTOR_DIMENSIONS = frozenset({384, 512, 768, 1024, 1536, 3072})
 class CloudBootstrapError(RuntimeError):
@@ -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:
+        box_enabled = box_config.get('enabled')
-            raise CloudBootstrapError('Cloud runtime requires box.enabled=true')
+        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()
@@ -15,6 +15,7 @@ from ..entity.persistence.cloud_directory import DirectoryProjectionInbox, Direc
 from ..entity.persistence.user import AccountSource, AccountStatus, User
 from ..entity.persistence.workspace import (
    MembershipRole,
    MembershipSource,
    MembershipStatus,
    Workspace,
    WorkspaceExecutionSource,
@@ -358,6 +359,7 @@ class DirectoryProjectionService:
        await self._reconcile_entitlement_snapshot_set(snapshot)
        self._publish_runtime_execution_projection(snapshot.workspaces)
        self._request_model_catalog_sync()
        self._record_batch_cardinality(
            active_workspaces=active_workspace_count,
            workspaces=workspace_count,
@@ -466,6 +468,7 @@ class DirectoryProjectionService:
            returned.values(),
            affected_workspace_uuids=requested,
        )
        self._request_model_catalog_sync()
        self._record_batch_cardinality(
            active_workspaces=active_workspace_count,
            workspaces=workspace_count,
@@ -475,6 +478,14 @@ class DirectoryProjectionService:
            self._record_success()
        self._consumer_cursor = batch.cursor
    def _request_model_catalog_sync(self) -> None:
        """Wake model provisioning after a committed directory change."""
        service = getattr(self.ap, 'cloud_model_catalog_service', None)
        request_sync = getattr(service, 'request_sync', None)
        if callable(request_sync):
            request_sync()
    def _publish_runtime_execution_projection(
        self,
        workspaces: Iterable[DirectoryWorkspace],
@@ -876,15 +887,15 @@ class DirectoryProjectionService:
                        account_uuid=member.account_uuid,
                        role=role,
                        status=status,
                        source=MembershipSource.CLOUD_PROJECTION.value,
                        joined_at=joined_at,
                        projection_revision=member.projection_revision,
                    )
                )
                continue
-            if membership.projection_revision == 0:
+            if membership.source != MembershipSource.CLOUD_PROJECTION.value:
-                # Revision zero is Core-owned collaboration state. Directory
+                # Core-owned collaboration state is never adopted based on
-                # projection seeds memberships, but must not overwrite later
+                # account provenance, revision, or matching account identity.
                # invitation, role, or removal decisions made by Core.
                continue
            if membership.uuid != member.membership_uuid:
                raise DirectoryProjectionUnavailableError('Directory membership UUID changed for one account')
@@ -896,11 +907,12 @@ class DirectoryProjectionService:
                raise DirectoryProjectionUnavailableError('Directory membership revision has conflicting contents')
            membership.role = role
            membership.status = status
            membership.source = MembershipSource.CLOUD_PROJECTION.value
            membership.joined_at = joined_at
            membership.projection_revision = member.projection_revision
        for account_uuid, membership in existing.items():
-            if account_uuid not in included_accounts and membership.projection_revision != 0:
+            if account_uuid not in included_accounts and membership.source == MembershipSource.CLOUD_PROJECTION.value:
                membership.status = MembershipStatus.REMOVED.value
                membership.projection_revision = max(
                    int(membership.projection_revision),
@@ -18,11 +18,19 @@ class EntitlementUnavailableError(RuntimeError):
 class EntitlementFeatureUnavailableError(EntitlementUnavailableError):
-    """Raised when an active entitlement explicitly omits a capability."""
+    """Raised only when an active entitlement does not grant one feature."""
-    def __init__(self, message: str, *, feature: str, entitlement_revision: int | None = None) -> None:
+    def __init__(
-        super().__init__(message, entitlement_revision=entitlement_revision)
+        self,
        feature: str,
        *,
        entitlement_revision: int | None = None,
    ) -> None:
        self.feature = feature
        super().__init__(
            f'Workspace entitlement does not grant {feature}',
            entitlement_revision=entitlement_revision,
        )
 class EntitlementSnapshot(pydantic.BaseModel):
@@ -91,8 +99,7 @@ class EntitlementSnapshot(pydantic.BaseModel):
    def require_feature(self, feature: str) -> None:
        if self.features.get(feature) is not True:
            raise EntitlementFeatureUnavailableError(
-                f'Workspace entitlement does not grant {feature}',
+                feature,
                feature=feature,
                entitlement_revision=self.entitlement_revision,
            )
@@ -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))
+        if kind == SUPPORT_ADMIN_LAUNCH_KIND:
-        return {
+            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:
+        kind = _required_string(claims, 'kind')
-            raise SpaceLaunchError('Launch assertion kind is not workspace.launch')
+        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')
@@ -0,0 +1,337 @@
 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', mode='before')
    @classmethod
    def normalize_missing_abilities(cls, value: Any) -> Any:
        return () if value is None else value
    @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
    credits: int | 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)
        # A tenant UoW commits one Workspace at a time. Keep a durable in-memory
        # convergence marker so a failed runtime reload is retried even when the
        # following database reconciliation is a no-op.
        self._runtime_reload_pending = False
        self._workspace_credits: dict[str, int | None] = {}
        self._sync_requested = asyncio.Event()
    def get_workspace_credits(self, workspace_uuid: str) -> int | None:
        """Return the latest signed owner-credit projection for a Workspace."""
        return self._workspace_credits.get(str(uuid.UUID(workspace_uuid)))
    async def initialize(self) -> None:
        await self.sync_once(reload_runtime=False)
    def request_sync(self) -> None:
        """Wake the catalog loop after a directory Workspace change."""
        self._sync_requested.set()
    async def run(self) -> None:
        while True:
            try:
                await asyncio.wait_for(self._sync_requested.wait(), timeout=self.sync_interval_seconds)
            except TimeoutError:
                pass
            self._sync_requested.clear()
            try:
                await self.sync_once(reload_runtime=True)
            except asyncio.CancelledError:
                raise
            except Exception as exc:
                # Exception messages can contain rendered SQL bound values,
                # including provider API keys. Log only the exception class.
                self.ap.logger.warning(f'Cloud model catalog synchronization failed ({type(exc).__name__})')
    async def sync_once(self, *, reload_runtime: bool = True) -> dict[str, int]:
        summary = {'workspaces': 0, 'created': 0, 'updated': 0, 'deleted': 0}
        snapshot: CloudModelCatalogSnapshot | None = None
        sync_error: Exception | None = None
        reload_error: Exception | None = None
        try:
            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'
                )
            for binding in bindings:
                counts = await self._sync_workspace(
                    binding.workspace_uuid,
                    snapshot,
                    billing_by_workspace[binding.workspace_uuid],
                )
                summary['workspaces'] += 1
                workspace_changed = any(counts[key] > 0 for key in ('created', 'updated', 'deleted'))
                if workspace_changed:
                    # _sync_workspace returns only after its tenant UoW commits.
                    self._runtime_reload_pending = True
                for key in ('created', 'updated', 'deleted'):
                    summary[key] += counts[key]
                self._workspace_credits[binding.workspace_uuid] = billing_by_workspace[binding.workspace_uuid].credits
        except Exception as exc:
            sync_error = exc
        finally:
            model_mgr = getattr(self.ap, 'model_mgr', None)
            if reload_runtime and self._runtime_reload_pending and model_mgr is not None:
                try:
                    await model_mgr.load_models_from_db()
                except Exception as exc:
                    reload_error = exc
                else:
                    self._runtime_reload_pending = False
        if sync_error is not None:
            if reload_error is not None:
                raise sync_error from reload_error
            raise sync_error
        if reload_error is not None:
            raise reload_error
        changed = any(summary[key] > 0 for key in ('created', 'updated', 'deleted'))
        if changed and snapshot is not None:
            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}')
@@ -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)
@@ -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(),
@@ -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)
@@ -41,6 +41,7 @@ _RUNTIME_POLICY_DEFAULTS = {
        }
    },
    'plugin': {
        'connect_timeout_seconds': 180.0,
        'worker': {
            'max_cpus': 1.0,
            'max_memory_mb': 512,
@@ -56,7 +57,7 @@ _RUNTIME_POLICY_DEFAULTS = {
            'restart_failure_window_seconds': 30.0,
            'restart_circuit_open_seconds': 60.0,
            'require_hard_limits': False,
-        }
+        },
    },
    'mcp': {'stdio': {'enabled': True}},
    'monitoring': {
@@ -186,9 +187,14 @@ def _apply_env_overrides_to_config(cfg: dict) -> dict:
                # At the final key
                if key in current:
                    if isinstance(current[key], list):
-                        # Convert comma-separated string to list
+                        # Convert comma-separated values while preserving the
-                        # e.g., SYSTEM__DISABLED_ADAPTERS="aiocqhttp,dingtalk"
+                        # element type declared by a non-empty config default.
-                        current[key] = [item.strip() for item in env_value.split(',') if item.strip()]
+                        items = [item.strip() for item in env_value.split(',') if item.strip()]
                        if current[key]:
                            exemplar = current[key][0]
                            current[key] = [convert_value(item, exemplar) for item in items]
                        else:
                            current[key] = items
                    elif isinstance(current[key], dict):
                        # Skip dict types
                        pass
@@ -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',
        ),
    )
@@ -40,6 +40,11 @@ class MembershipStatus(enum.StrEnum):
    REMOVED = 'removed'
 class MembershipSource(enum.StrEnum):
    LOCAL = 'local'
    CLOUD_PROJECTION = 'cloud_projection'
 class InvitationStatus(enum.StrEnum):
    PENDING = 'pending'
    ACCEPTED = 'accepted'
@@ -151,6 +156,11 @@ class WorkspaceMembership(Base):
        nullable=True,
    )
    joined_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=True)
    source = sqlalchemy.Column(
        sqlalchemy.String(32),
        nullable=False,
        server_default=MembershipSource.LOCAL.value,
    )
    projection_revision = sqlalchemy.Column(sqlalchemy.BigInteger, nullable=False, server_default='0')
    created_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=False, server_default=sqlalchemy.func.now())
    updated_at = sqlalchemy.Column(
@@ -163,6 +173,13 @@ class WorkspaceMembership(Base):
    __table_args__ = (
        sqlalchemy.UniqueConstraint('workspace_uuid', 'account_uuid', name='uq_workspace_membership_account'),
        sqlalchemy.Index('ix_workspace_memberships_account_status', 'account_uuid', 'status'),
        sqlalchemy.Index(
            'uq_workspace_memberships_one_active_owner',
            'workspace_uuid',
            unique=True,
            sqlite_where=sqlalchemy.text("role = 'owner' AND status = 'active'"),
            postgresql_where=sqlalchemy.text("role = 'owner' AND status = 'active'"),
        ),
        sqlalchemy.CheckConstraint(
            "role IN ('owner', 'admin', 'developer', 'operator', 'viewer')",
            name='ck_workspace_memberships_role',
@@ -171,6 +188,10 @@ class WorkspaceMembership(Base):
            "status IN ('active', 'disabled', 'removed')",
            name='ck_workspace_memberships_status',
        ),
        sqlalchemy.CheckConstraint(
            "source IN ('local', 'cloud_projection')",
            name='ck_workspace_memberships_source',
        ),
    )
@@ -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,
@@ -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)
@@ -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)
@@ -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
@@ -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
@@ -0,0 +1,80 @@
 """enforce one active owner per Workspace
 Revision ID: 0019_single_workspace_owner
 Revises: 0018_merge_launch_replay
 Create Date: 2026-08-02
 """
 from __future__ import annotations
 import sqlalchemy as sa
 from alembic import op
 revision = '0019_single_workspace_owner'
 down_revision = '0018_merge_launch_replay'
 branch_labels = None
 depends_on = None
 _INDEX_NAME = 'uq_workspace_memberships_one_active_owner'
 def upgrade() -> None:
    conn = op.get_bind()
    inspector = sa.inspect(conn)
    if 'workspace_memberships' not in inspector.get_table_names():
        return
    # Ownership transfer used to promote a second member without demoting the
    # original owner. Preserve the Workspace creator where possible and demote
    # every historical extra owner before installing the database invariant.
    op.execute(
        sa.text(
            """
            WITH ranked_owners AS (
                SELECT membership.uuid,
                       ROW_NUMBER() OVER (
                           PARTITION BY membership.workspace_uuid
                           ORDER BY
                               CASE
                                   WHEN membership.account_uuid = workspace.created_by_account_uuid THEN 0
                                   ELSE 1
                               END,
                               COALESCE(membership.joined_at, membership.created_at),
                               membership.uuid
                       ) AS owner_rank
                FROM workspace_memberships AS membership
                JOIN workspaces AS workspace
                  ON workspace.uuid = membership.workspace_uuid
                WHERE membership.role = 'owner'
                  AND membership.status = 'active'
            )
            UPDATE workspace_memberships
               SET role = 'admin'
             WHERE uuid IN (
                 SELECT uuid
                   FROM ranked_owners
                  WHERE owner_rank > 1
             )
            """
        )
    )
    # Fresh installations may already have this index because SQLAlchemy
    # metadata is created before Alembic advances the revision marker.
    op.execute(
        sa.text(
            'CREATE UNIQUE INDEX IF NOT EXISTS '
            'uq_workspace_memberships_one_active_owner '
            'ON workspace_memberships (workspace_uuid) '
            "WHERE role = 'owner' AND status = 'active'"
        )
    )
 def downgrade() -> None:
    conn = op.get_bind()
    inspector = sa.inspect(conn)
    if 'workspace_memberships' not in inspector.get_table_names():
        return
    index_names = {index['name'] for index in inspector.get_indexes('workspace_memberships')}
    if _INDEX_NAME in index_names:
        op.drop_index(_INDEX_NAME, table_name='workspace_memberships')
@@ -0,0 +1,43 @@
 """enable 3072-dimensional pgvector embeddings
 Revision ID: 001a_pgvector_dimension_3072
 Revises: 0019_single_workspace_owner
 Create Date: 2026-08-05
 """
 from __future__ import annotations
 import sqlalchemy as sa
 from alembic import op
 revision = '001a_pgvector_dimension_3072'
 down_revision = '0019_single_workspace_owner'
 branch_labels = None
 depends_on = None
 _TABLE = 'langbot_vectors'
 _CHECK = 'ck_langbot_vectors_embedding_dimension_enabled'
 _INDEX = 'ix_langbot_vectors_hnsw_cosine_3072'
 def upgrade() -> None:
    conn = op.get_bind()
    if conn.dialect.name != 'postgresql' or _TABLE not in sa.inspect(conn).get_table_names():
        return
    op.drop_constraint(_CHECK, _TABLE, type_='check')
    op.create_check_constraint(_CHECK, _TABLE, 'embedding_dimension IN (384, 512, 768, 1024, 1536, 3072)')
    op.execute(
        sa.text(
            f'CREATE INDEX {_INDEX} ON {_TABLE} USING hnsw ((embedding::halfvec(3072)) halfvec_cosine_ops) WHERE embedding_dimension = 3072'
        )
    )
 def downgrade() -> None:
    conn = op.get_bind()
    if conn.dialect.name != 'postgresql' or _TABLE not in sa.inspect(conn).get_table_names():
        return
    count = conn.scalar(sa.text(f'SELECT COUNT(*) FROM {_TABLE} WHERE embedding_dimension = 3072'))
    if count:
        raise RuntimeError('Cannot disable 3072-dimensional pgvector while matching embeddings exist')
    op.drop_index(_INDEX, table_name=_TABLE)
    op.drop_constraint(_CHECK, _TABLE, type_='check')
    op.create_check_constraint(_CHECK, _TABLE, 'embedding_dimension IN (384, 512, 768, 1024, 1536)')
@@ -0,0 +1,49 @@
 """add explicit Workspace membership source
 Revision ID: 0020_membership_source
 Revises: 001a_pgvector_dimension_3072
 Create Date: 2026-08-06
 """
 from __future__ import annotations
 import sqlalchemy as sa
 from alembic import op
 revision = '0020_membership_source'
 down_revision = '001a_pgvector_dimension_3072'
 branch_labels = None
 depends_on = None
 _CONSTRAINT_NAME = 'ck_workspace_memberships_source'
 def upgrade() -> None:
    conn = op.get_bind()
    inspector = sa.inspect(conn)
    if 'workspace_memberships' not in inspector.get_table_names():
        return
    if 'source' in {column['name'] for column in inspector.get_columns('workspace_memberships')}:
        return
    # No durable historical field distinguishes Directory-created revision-zero
    # rows from Core invitations. Protect every existing row; production can
    # reclassify separately after UUIDs have been verified against Space.
    with op.batch_alter_table('workspace_memberships') as batch_op:
        batch_op.add_column(sa.Column('source', sa.String(length=32), nullable=False, server_default='local'))
        batch_op.create_check_constraint(
            _CONSTRAINT_NAME,
            "source IN ('local', 'cloud_projection')",
        )
 def downgrade() -> None:
    conn = op.get_bind()
    inspector = sa.inspect(conn)
    if 'workspace_memberships' not in inspector.get_table_names():
        return
    if 'source' not in {column['name'] for column in inspector.get_columns('workspace_memberships')}:
        return
    with op.batch_alter_table('workspace_memberships') as batch_op:
        batch_op.drop_constraint(_CONSTRAINT_NAME, type_='check')
        batch_op.drop_column('source')
@@ -54,6 +54,7 @@ _ALEMBIC_TENANT_TABLES = {
    'workspace_memberships',
    'workspace_invitations',
    'workspace_execution_states',
    'support_admin_temporary_sessions',
    'workspace_metadata',
    'api_keys',
    'bots',
@@ -97,7 +98,7 @@ _WORKSPACE_ALEMBIC_REVISION = '0009_workspace_tenancy'
 _RESOURCE_SCOPE_ALEMBIC_REVISION = '0010_scope_resources'
 _OSS_WORKSPACE_METADATA_KEY = 'oss_workspace_uuid'
 _RELEASE_MIGRATION_ADVISORY_LOCK_ID = 0x4C414E47424F5432
-_PGVECTOR_ALLOWED_DIMENSIONS = (384, 512, 768, 1024, 1536)
+_PGVECTOR_ALLOWED_DIMENSIONS = (384, 512, 768, 1024, 1536, 3072)
 _RUNTIME_SCHEMA = 'public'
 _ALEMBIC_RUNTIME_TABLE = 'alembic_version'
 _RUNTIME_TABLE_PRIVILEGES = frozenset({'SELECT', 'INSERT', 'UPDATE', 'DELETE'})
@@ -1355,14 +1356,16 @@ class PersistenceManager:
            index = by_index.get(index_name)
            index_definition = normalized(None if index is None else index['definition'])
            predicate = normalized(None if index is None else index['predicate'])
            vector_type = 'halfvec' if dimension > 2000 else 'vector'
            operator_class = f'{vector_type}_cosine_ops'
            if (
                index is None
                or index['access_method'] != 'hnsw'
                or index['is_valid'] is not True
                or index['is_ready'] is not True
-                or f'vector({dimension})' not in index_definition
+                or f'{vector_type}({dimension})' not in index_definition
-                or f'(embedding)::vector({dimension})' not in index_definition
+                or f'(embedding)::{vector_type}({dimension})' not in index_definition
-                or 'vector_cosine_ops' not in index_definition
+                or operator_class not in index_definition
                or predicate.strip('() ') != f'embedding_dimension = {dimension}'
            ):
                raise RuntimeError(f'PostgreSQL pgvector ANN index {index_name!r} is invalid')
@@ -13,7 +13,7 @@ import typing
 import sqlalchemy
 import sqlalchemy.ext.asyncio as sqlalchemy_asyncio
 import sqlalchemy.orm as sqlalchemy_orm
-from pgvector.sqlalchemy import Vector
+from pgvector.sqlalchemy import HALFVEC, Vector
 from sqlalchemy.dialects.postgresql.dml import OnConflictDoNothing as PostgreSQLOnConflictDoNothing
 from sqlalchemy.dialects.postgresql.dml import OnConflictDoUpdate as PostgreSQLOnConflictDoUpdate
 from sqlalchemy.dialects.sqlite.dml import OnConflictDoNothing as SQLiteOnConflictDoNothing
@@ -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',
@@ -208,7 +209,7 @@ _ALLOWED_SCOPED_BUILTIN_FUNCTION_TYPES = {
    'now': sqlalchemy.sql.functions.now,
    'sum': sqlalchemy.sql.functions.sum,
 }
-_ALLOWED_SCOPED_GENERIC_FUNCTIONS = frozenset({'length', 'nullif'})
+_ALLOWED_SCOPED_GENERIC_FUNCTIONS = frozenset({'date_trunc', 'length', 'nullif'})
 _ALLOWED_SCOPED_CUSTOM_OPERATORS = frozenset({'<=>'})
 _ALLOWED_SCOPED_STATEMENT_TYPES = (
    sqlalchemy.sql.dml.UpdateBase,
@@ -280,7 +281,7 @@ def _validate_scoped_sql_type(
        return
    seen.add(identity)
-    if type(sql_type) is Vector:
+    if type(sql_type) in {Vector, HALFVEC}:
        return
    if not type(sql_type).__module__.startswith('sqlalchemy.'):
        raise ScopedSessionTransactionError('TenantUnitOfWork does not allow custom SQL types in public statements')
@@ -461,7 +462,7 @@ def _validate_scoped_statement_call(args: tuple[typing.Any, ...], kwargs: dict[s
        if isinstance(element, sqlalchemy.sql.elements.BindParameter) and element.literal_execute:
            raise ScopedSessionTransactionError('TenantUnitOfWork does not allow literal-execute SQL parameters')
-        if isinstance(element, sqlalchemy.sql.elements.Cast) and type(element.type) is not Vector:
+        if isinstance(element, sqlalchemy.sql.elements.Cast) and type(element.type) not in {Vector, HALFVEC}:
            raise ScopedSessionTransactionError(
                'TenantUnitOfWork only allows the trusted pgvector cast used by tenant vector search'
            )
@@ -132,9 +132,7 @@ class Controller:
                            break
-                    if selected_query:  # 找到了
+                    if not selected_query:  # 没找到 说明：没有请求 或者 所有query对应的session都已达到并发上限
                        queries.remove(selected_query)
                    else:  # 没找到 说明：没有请求 或者 所有query对应的session都已达到并发上限
                        await self.ap.query_pool.condition.wait()
                        continue
@@ -41,6 +41,29 @@ class PreProcessor(stage.PipelineStage):
        selected_tool_names = {tool for tool in selected_tools if isinstance(tool, str)}
        return [tool for tool in tools if tool.name in selected_tool_names]
    @staticmethod
    def _append_to_system_prompt(
        messages: list[provider_message.Message],
        addition: str,
    ) -> None:
        """Append text to the first system message, creating one if none exists.
        Handles both plain-string and content-element (list) message bodies.
        """
        if messages and messages[0].role == 'system':
            head = messages[0]
            if isinstance(head.content, str):
                head.content = head.content + addition
            elif isinstance(head.content, list):
                for ce in head.content:
                    if getattr(ce, 'type', None) == 'text':
                        ce.text = (ce.text or '') + addition
                        break
                else:
                    head.content.append(provider_message.ContentElement(type='text', text=addition))
        else:
            messages.insert(0, provider_message.Message(role='system', content=addition.strip()))
    async def process(
        self,
        query: pipeline_query.Query,
@@ -275,6 +298,23 @@ class PreProcessor(stage.PipelineStage):
        query.prompt.messages = event_ctx.event.default_prompt
        query.messages = event_ctx.event.prompt
        # =========== Current date grounding for the local-agent runner ===========
        # local-agent system prompts are static strings with no template-variable
        # support, so without an explicit anchor the LLM resolves relative time
        # references (e.g. "this quarter", "latest", "currently") against whichever
        # period is best represented in its training data instead of the real date,
        # and won't reliably know to double check time-sensitive facts with a tool.
        if selected_runner == 'local-agent':
            date_addition = (
                f'\n\nCurrent date: {datetime.datetime.now().strftime("%Y-%m-%d (%A)")}. '
                'Resolve relative time references (e.g. "today", "this quarter", "latest", '
                '"currently") based on this date, not your training cutoff. For anything '
                'time-sensitive that may have changed since training — stock prices, '
                'financial results, news, current events, exchange rates, or similar — '
                'verify with a search tool if one is available rather than answering from memory.'
            )
            self._append_to_system_prompt(query.prompt.messages, date_addition)
        # =========== Skill awareness for the local-agent runner ===========
        # The actual activation goes through the ``activate`` Tool Call so the
        # LLM doesn't see full SKILL.md instructions until it commits to a
@@ -310,27 +350,7 @@ class PreProcessor(stage.PipelineStage):
                bound_skills=bound_skills,
            )
            if skill_addition:
-                # Append to the first system message; create one if the
+                self._append_to_system_prompt(query.prompt.messages, skill_addition)
                # prompt has none. Handles both plain-string and
                # content-element (list) message bodies.
                if query.prompt.messages and query.prompt.messages[0].role == 'system':
                    head = query.prompt.messages[0]
                    if isinstance(head.content, str):
                        head.content = head.content + skill_addition
                    elif isinstance(head.content, list):
                        appended = False
                        for ce in head.content:
                            if getattr(ce, 'type', None) == 'text':
                                ce.text = (ce.text or '') + skill_addition
                                appended = True
                                break
                        if not appended:
                            head.content.append(provider_message.ContentElement(type='text', text=skill_addition))
                else:
                    query.prompt.messages.insert(
                        0,
                        provider_message.Message(role='system', content=skill_addition.strip()),
                    )
                self.ap.logger.debug(
                    f'Skill index injected into system prompt: '
                    f'pipeline={query.pipeline_uuid} '
@@ -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,
@@ -1,10 +1,12 @@
 from __future__ import annotations
 import typing
 import sqlalchemy
 from .. import handler
 from ... import entities
 from ... import plugin_diagnostics
 from ....entity.persistence.bot import BotAdmin
 import langbot_plugin.api.entities.builtin.provider.message as provider_message
 import langbot_plugin.api.entities.builtin.provider.session as provider_session
 import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query
@@ -24,7 +26,14 @@ class CommandHandler(handler.MessageHandler):
        privilege = 1
-        if f'{query.launcher_type.value}_{query.launcher_id}' in self.ap.instance_config.data['admins']:
+        admins = await self.ap.persistence_mgr.execute_async(
            sqlalchemy.select(BotAdmin).where(
                BotAdmin.bot_uuid == (query.bot_uuid or ''),
                BotAdmin.launcher_type == query.launcher_type.value,
                BotAdmin.launcher_id == str(query.launcher_id),
            )
        )
        if admins.first() is not None:
            privilege = 2
        spt = command_text.split(' ')
@@ -241,7 +241,11 @@ class AiocqhttpMessageConverter(abstract_platform_adapter.AbstractMessageConvert
        async def process_message_data(msg_data, reply_list):
            if msg_data['type'] == 'image':
                image_base64, image_format = await image.qq_image_url_to_base64(msg_data['data']['url'])
-                reply_list.append(platform_message.Image(base64=f'data:image/{image_format};base64,{image_base64}'))
+                reply_list.append(
                    platform_message.Image(
                        url=msg_data['data']['url'], base64=f'data:image/{image_format};base64,{image_base64}'
                    )
                )
            elif msg_data['type'] == 'text':
                reply_list.append(platform_message.Plain(text=msg_data['data']['text']))
@@ -286,7 +290,9 @@ class AiocqhttpMessageConverter(abstract_platform_adapter.AbstractMessageConvert
                    image_msg = platform_message.Face(face_id=face_id, face_name=face_name)
                else:
                    image_base64, image_format = await image.qq_image_url_to_base64(msg.data['url'])
-                    image_msg = platform_message.Image(base64=f'data:image/{image_format};base64,{image_base64}')
+                    image_msg = platform_message.Image(
                        url=msg.data['url'], base64=f'data:image/{image_format};base64,{image_base64}'
                    )
                yiri_msg_list.append(image_msg)
            elif msg.type == 'forward':
                # 暂时不太合理
@@ -764,7 +764,9 @@ class DiscordMessageConverter(abstract_platform_adapter.AbstractMessageConverter
                )
                image_base64 = (await asyncio.to_thread(base64.b64encode, image_data)).decode('utf-8')
                image_format = response.headers['Content-Type']
-                element_list.append(platform_message.Image(base64=f'data:{image_format};base64,{image_base64}'))
+                element_list.append(
                    platform_message.Image(url=attachment.url, base64=f'data:{image_format};base64,{image_base64}')
                )
        return platform_message.MessageChain(element_list)
@@ -102,7 +102,7 @@ class QQOfficialMessageConverter(abstract_platform_adapter.AbstractMessageConver
        yiri_msg_list.append(platform_message.Source(id=message_id, time=datetime.datetime.now()))
        if pic_url is not None:
            base64_url = await image.get_qq_official_image_base64(pic_url=pic_url, content_type=content_type)
-            yiri_msg_list.append(platform_message.Image(base64=base64_url))
+            yiri_msg_list.append(platform_message.Image(url=pic_url, base64=base64_url))
        yiri_msg_list.append(platform_message.Plain(text=message))
        chain = platform_message.MessageChain(yiri_msg_list)
@@ -47,7 +47,7 @@ class SlackMessageConverter(abstract_platform_adapter.AbstractMessageConverter):
        yiri_msg_list.append(platform_message.Source(id=message_id, time=datetime.datetime.now()))
        if pic_url is not None:
            base64_url = await image.get_slack_image_to_base64(pic_url=pic_url, bot_token=bot.bot_token)
-            yiri_msg_list.append(platform_message.Image(base64=base64_url))
+            yiri_msg_list.append(platform_message.Image(url=pic_url, base64=base64_url))
        yiri_msg_list.append(platform_message.Plain(text=message))
        chain = platform_message.MessageChain(yiri_msg_list)
@@ -179,9 +179,15 @@ class TelegramMessageConverter(abstract_platform_adapter.AbstractMessageConverte
                )
                file_format = 'image/jpeg'
            # NOTE: Telegram's file.file_path is a full URL of the form
            # https://api.telegram.org/file/bot<TOKEN>/<path> which embeds the
            # bot token. Unlike the public CDN URLs used by other adapters, it
            # cannot be exposed safely, so only base64 is stored here.
            encoded = await asyncio.to_thread(base64.b64encode, file_bytes)
            message_components.append(
-                platform_message.Image(base64=f'data:{file_format};base64,{encoded.decode("utf-8")}')
+                platform_message.Image(
                    base64=f'data:{file_format};base64,{encoded.decode("utf-8")}',
                )
            )
        if message.voice:
@@ -707,28 +707,37 @@ class WebSocketAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
            if len(listener_tasks) >= 100:
                await self.logger.warning('WebSocket inbound listener capacity reached; dropping message')
                return
-            token = _current_pipeline_uuid.set(pipeline_uuid)
+            listener = typing.cast(
-            try:
+                typing.Callable[[typing.Any, typing.Any], typing.Awaitable[None]],
-                task_manager = getattr(self.ap, 'task_mgr', None)
+                listeners[event.__class__],
-                if task_manager is None or not isinstance(getattr(task_manager, 'tasks', None), list):
+            )
-                    listener_task = asyncio.create_task(listeners[event.__class__](event, callback_adapter))
+
-                else:
+            async def run_listener():
-                    listener_task = task_manager.create_task(
+                token = _current_pipeline_uuid.set(pipeline_uuid)
-                        listeners[event.__class__](event, callback_adapter),
+                try:
-                        kind='websocket-message',
+                    await listener(event, callback_adapter)
-                        name=f'websocket-message-{connection.connection_id}',
+                finally:
-                        scopes=[
+                    _current_pipeline_uuid.reset(token)
-                            core_entities.LifecycleControlScope.APPLICATION,
+
-                            core_entities.LifecycleControlScope.PLATFORM,
+            listener_coro = run_listener()
-                        ],
+            task_manager = getattr(self.ap, 'task_mgr', None)
-                        instance_uuid=connection.instance_uuid,
+            if task_manager is None or not isinstance(getattr(task_manager, 'tasks', None), list):
-                        workspace_uuid=connection.workspace_uuid,
+                listener_task = asyncio.create_task(listener_coro)
-                        placement_generation=connection.placement_generation,
+            else:
-                    ).task
+                listener_task = task_manager.create_task(
-                listener_tasks.add(listener_task)
+                    listener_coro,
-                listener_task.add_done_callback(self._listener_task_done)
+                    kind='websocket-message',
-            finally:
+                    name=f'websocket-message-{connection.connection_id}',
-                _current_pipeline_uuid.reset(token)
+                    scopes=[
                        core_entities.LifecycleControlScope.APPLICATION,
                        core_entities.LifecycleControlScope.PLATFORM,
                    ],
                    instance_uuid=connection.instance_uuid,
                    workspace_uuid=connection.workspace_uuid,
                    placement_generation=connection.placement_generation,
                ).task
            listener_tasks.add(listener_task)
            listener_task.add_done_callback(self._listener_task_done)
    def get_websocket_messages(
        self,
@@ -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),
            )
@@ -133,7 +133,9 @@ class WecomMessageConverter(abstract_platform_adapter.AbstractMessageConverter):
        yiri_msg_list = []
        yiri_msg_list.append(platform_message.Source(id=message_id, time=datetime.datetime.now()))
        image_base64, image_format = await image.get_wecom_image_base64(pic_url=picurl)
-        yiri_msg_list.append(platform_message.Image(base64=f'data:image/{image_format};base64,{image_base64}'))
+        yiri_msg_list.append(
            platform_message.Image(url=picurl, base64=f'data:image/{image_format};base64,{image_base64}')
        )
        chain = platform_message.MessageChain(yiri_msg_list)
        return chain
@@ -6,6 +6,7 @@ import contextlib
 import contextvars
 import hashlib
 import json
 import math
 import time
 import uuid
 from typing import Any
@@ -76,7 +77,7 @@ _GITHUB_ASSET_HOSTS = frozenset(
    }
 )
 _HTTP_REDIRECT_STATUSES = frozenset({301, 302, 303, 307, 308})
-_CONNECT_TIMEOUT_SEC = 30.0
+_DEFAULT_CONNECT_TIMEOUT_SECONDS = 180.0
 _HEARTBEAT_INTERVAL_SEC = 20.0
 _HEARTBEAT_FAILURE_THRESHOLD = 3
 _RECONNECT_MAX_DELAY_SEC = 60.0
@@ -206,6 +207,17 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
        return f'{constants.instance_id}:plugin-runtime'
    @staticmethod
    def _runtime_connect_timeout(plugin_config: dict[str, Any]) -> float:
        value = plugin_config.get('connect_timeout_seconds', _DEFAULT_CONNECT_TIMEOUT_SECONDS)
        if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value) or value <= 0:
            raise ValueError('plugin.connect_timeout_seconds must be a positive number')
        return float(value)
    @staticmethod
    def _runtime_connect_timeout_error(timeout_seconds: float) -> str:
        return f'Plugin runtime did not become ready within {timeout_seconds:g} seconds'
    def _runtime_handler(self) -> handler.RuntimeConnectionHandler:
        runtime_handler = getattr(self, 'handler', None)
        if runtime_handler is None:
@@ -251,6 +263,8 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
    def _control_headers(self, *, allow_generate: bool) -> dict[str, str]:
        if not self._control_token and allow_generate:
            self._control_token = secrets.token_urlsafe(48)
        if not self._control_token:
            return {}
        try:
            self._control_token = validate_runtime_secret(
                self._control_token,
@@ -699,10 +713,13 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
        """
        runtime_handler = self._runtime_handler()
        started_at = time.monotonic()
        async with self._state_lock:
            all_states: dict[str, PluginInstallationDesiredState] = {}
            workspace_installations: dict[str, set[str]] = {}
            workspace_count = 0
            for context in contexts:
                workspace_count += 1
                execution_context = await self._validate_execution_context(context)
                states = await self._load_workspace_desired_states(execution_context)
                installation_ids = {state.binding.installation_uuid for state in states}
@@ -722,6 +739,13 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
                    runtime_handler.unregister_installation_binding(previous.binding)
            self._known_desired_states = all_states
            self._workspace_installations = workspace_installations
            self.ap.logger.info(
                'Shared plugin runtime reconcile completed: workspaces=%d desired_installations=%d '
                'elapsed_seconds=%.3f',
                workspace_count,
                len(all_states),
                time.monotonic() - started_at,
            )
            return result
    async def _validate_execution_context(self, context: TenantContext) -> ExecutionContext:
@@ -817,6 +841,8 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
            runtime_id=self._runtime_id,
        )
        self.worker_policy = self._load_worker_policy()
        plugin_config = self.ap.instance_config.data.get('plugin', {})
        connect_timeout_seconds = self._runtime_connect_timeout(plugin_config)
        async with self._lifecycle_lock:
            if self._closing:
@@ -958,10 +984,12 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
            self._transport_task = asyncio.create_task(task_coro)
            try:
-                await asyncio.wait_for(self._connected.wait(), timeout=_CONNECT_TIMEOUT_SEC)
+                await asyncio.wait_for(self._connected.wait(), timeout=connect_timeout_seconds)
            except asyncio.TimeoutError as exc:
                await self._stop_transport()
-                raise PluginRuntimeNotConnectedError('Plugin runtime did not become ready within 30 seconds') from exc
+                raise PluginRuntimeNotConnectedError(
                    self._runtime_connect_timeout_error(connect_timeout_seconds)
                ) from exc
            if connect_errors:
                await self._stop_transport()
                raise PluginRuntimeNotConnectedError(f'Plugin runtime connection failed: {connect_errors[-1]}')
@@ -1968,11 +1996,11 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
        with runtime_handler.installation_scope(binding):
            return await runtime_handler.handle_page_api(plugin_author, plugin_name, page_id, endpoint, method, body)
-    async def get_debug_info(self) -> dict[str, Any]:
+    async def get_debug_info(self, execution_context: ExecutionContext) -> dict[str, Any]:
        """Get debug information including debug key and WS URL"""
        if not self.is_enable_plugin or not self._runtime_available():
            return {}
-        return await self._runtime_handler().get_debug_info()
+        return await self._runtime_handler().get_debug_info(execution_context)
    async def emit_event(
        self,
@@ -1960,14 +1960,14 @@ class RuntimeConnectionHandler(handler.Handler):
        )
        return result
-    async def get_debug_info(self) -> dict[str, Any]:
+    async def get_debug_info(self, execution_context: ExecutionContext) -> dict[str, Any]:
        """Get debug information including debug key and WS URL"""
-        with self.installation_scope(None):
+        result = await self.call_action(
-            result = await self.call_action(
+            LangBotToRuntimeAction.GET_DEBUG_INFO,
-                LangBotToRuntimeAction.GET_DEBUG_INFO,
+            {},
-                {},
+            timeout=10,
-                timeout=10,
+            action_context=execution_context,
-            )
+        )
        return result
    # ================= RAG Capability Callers (LangBot -> Runtime) =================
@@ -529,6 +529,7 @@ class ModelManager:
            model['uuid']: model
            for model in await self.ap.embedding_models_service.get_embedding_models(context, include_secret=True)
        }
        existing_rerank_models = {m['uuid']: m for m in await self.ap.rerank_models_service.get_rerank_models(context)}
        created = 0
        updated = 0
@@ -597,6 +598,36 @@ class ModelManager:
                        )
                        updated += 1
            elif space_model.category == 'rerank':
                existing = existing_rerank_models.get(space_model.uuid)
                if existing is None:
                    await self.ap.rerank_models_service.create_rerank_model(
                        context,
                        {
                            'uuid': space_model.uuid,
                            'name': space_model.model_id,
                            'provider_uuid': space_model_provider.uuid,
                            'extra_args': {},
                            'prefered_ranking': space_model.featured_order,
                        },
                        preserve_uuid=True,
                    )
                    created += 1
                elif existing.get('provider_uuid') == space_model_provider.uuid:
                    desired = {
                        'name': space_model.model_id,
                        'provider_uuid': space_model_provider.uuid,
                        'prefered_ranking': space_model.featured_order,
                    }
                    if (
                        existing.get('name') != desired['name']
                        or existing.get('prefered_ranking') != desired['prefered_ranking']
                    ):
                        await self.ap.rerank_models_service.update_rerank_model(
                            context, space_model.uuid, dict(desired)
                        )
                        updated += 1
        if created or updated:
            self.ap.logger.info(f'Synced models from LangBot Space: {created} added, {updated} updated.')
@@ -944,16 +944,21 @@ class LiteLLMRequester(requester.ProviderAPIRequester):
        if api_key:
            headers['Authorization'] = f'Bearer {api_key}'
        request_args = dict(extra_args)
        rerank_url = request_args.pop('rerank_url', None)
        rerank_path = request_args.pop('rerank_path', 'rerank')
        payload: dict[str, typing.Any] = {
            'model': model_name,
            'query': query,
            'documents': documents,
            'top_n': top_n,
        }
-        if extra_args:
+        if request_args:
-            payload.update(extra_args)
+            payload.update(request_args)
-        rerank_url = f'{base_url}/rerank'
+        if not rerank_url:
            rerank_url = f'{base_url}/{str(rerank_path).strip("/")}'
        try:
            async with httpx.AsyncClient(
@@ -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,65 @@ async def _count(
        return -1
-async def build_heartbeat_payload(ap: core_app.Application) -> dict:
+async def _cloud_workspace_resource_counts(ap: core_app.Application, bindings) -> list[WorkspaceResourceSnapshot]:
-    """Collect the anonymous instance profile snapshot."""
+    """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_create_ts: int = 0,
    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 +206,16 @@ 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,
        'workspace_create_ts': workspace_create_ts,
        'instance_create_ts': constants.instance_create_ts,
        'edition': constants.edition,
        'features': features,
@@ -148,14 +223,49 @@ async def build_heartbeat_payload(ap: core_app.Application) -> dict:
    }
 def _workspace_created_timestamp(created_at: datetime | None) -> int:
    if created_at is None:
        return 0
    if created_at.tzinfo is None:
        # SQLAlchemy may return persisted UTC values without tzinfo. Never
        # reinterpret them in the host's local timezone.
        created_at = created_at.replace(tzinfo=timezone.utc)
    return int(created_at.timestamp())
 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})
    workspace_create_ts = {
        binding.workspace_uuid: _workspace_created_timestamp(getattr(binding, 'workspace_created_at', None))
        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_create_ts=workspace_create_ts.get(workspace_uuid, 0),
            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)
+            for payload in await build_heartbeat_payloads(ap):
-            await ap.telemetry.start_send_task(payload)
+                # 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}')
@@ -0,0 +1,22 @@
 from __future__ import annotations
 import typing
 class WorkspaceExecutionContext(typing.Protocol):
    @property
    def instance_uuid(self) -> str: ...
    @property
    def workspace_uuid(self) -> str: ...
 def workspace_identity(execution_context: WorkspaceExecutionContext) -> dict[str, str]:
    """Build both first-class telemetry identities for one execution."""
    instance_id = execution_context.instance_uuid.strip()
    workspace_uuid = execution_context.workspace_uuid.strip()
    if not instance_id:
        raise ValueError('Telemetry execution instance ID is empty')
    if not workspace_uuid:
        raise ValueError('Telemetry execution Workspace UUID is empty')
    return {'instance_id': instance_id, 'workspace_uuid': workspace_uuid}
@@ -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,35 @@ 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()
                        headers: dict[str, str] = {}
                        if telemetry_token:
                            headers['X-LangBot-Telemetry-Token'] = telemetry_token
                        else:
                            workspace_uuid = str(sanitized.get('workspace_uuid', '')).strip()
                            user_service = getattr(self.ap, 'user_service', None)
                            if workspace_uuid and user_service is not None:
                                try:
                                    owner = await user_service.get_workspace_owner(workspace_uuid)
                                    owner_email = str(getattr(owner, 'user', '') or '').strip()
                                    space_service = getattr(self.ap, 'space_service', None)
                                    access_token = (
                                        await space_service.get_valid_access_token(owner_email)
                                        if owner_email and space_service is not None
                                        else None
                                    )
                                    access_token = str(access_token or '').strip()
                                    if access_token:
                                        headers['Authorization'] = f'Bearer {access_token}'
                                except Exception:
                                    self.ap.logger.debug(
                                        'Could not resolve authenticated telemetry reporter', exc_info=True
                                    )
                        if headers:
                            request = client.post(url, json=sanitized, headers=headers)
                        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 +175,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")}'
@@ -67,7 +67,7 @@ class VectorDBManager:
                use_business_database = pgvector_config.get('use_business_database', False)
                allowed_dimensions = pgvector_config.get(
                    'allowed_dimensions',
-                    [384, 512, 768, 1024, 1536],
+                    [384, 512, 768, 1024, 1536, 3072],
                )
                common_options = {
                    'use_business_database': use_business_database,
@@ -6,7 +6,7 @@ from collections.abc import AsyncIterator
 from typing import Any
 import sqlalchemy
-from pgvector.sqlalchemy import Vector
+from pgvector.sqlalchemy import HALFVEC, Vector
 from sqlalchemy.dialects.postgresql import insert as postgresql_insert
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
 from sqlalchemy.orm import declarative_base
@@ -18,7 +18,7 @@ from langbot.pkg.vector.vdb import VectorDatabase
 Base = declarative_base()
-DEFAULT_ALLOWED_DIMENSIONS = (384, 512, 768, 1024, 1536)
+DEFAULT_ALLOWED_DIMENSIONS = (384, 512, 768, 1024, 1536, 3072)
 # pgvector schema only stores these metadata fields.
 _PG_SUPPORTED_FIELDS = {'text', 'file_id', 'chunk_uuid'}
@@ -321,7 +321,12 @@ class PgVectorDatabase(VectorDatabase):
        if len(query_embedding) != scope.embedding_dimension:
            raise ValueError(f'Query embedding must have the selected dimension {scope.embedding_dimension}')
-        typed_embedding = sqlalchemy.cast(PgVectorEntry.embedding, Vector(scope.embedding_dimension))
+        typed_embedding = sqlalchemy.cast(
            PgVectorEntry.embedding,
            HALFVEC(scope.embedding_dimension)
            if scope.embedding_dimension > 2000
            else Vector(scope.embedding_dimension),
        )
        distance = typed_embedding.cosine_distance(query_embedding)
        statement = (
            sqlalchemy.select(
@@ -17,6 +17,7 @@ from ..entity.persistence.user import AccountStatus, User
 from ..entity.persistence.workspace import (
    InvitationStatus,
    MembershipRole,
    MembershipSource,
    MembershipStatus,
    Workspace,
    WorkspaceInvitation,
@@ -88,6 +89,7 @@ class ResolvedWorkspaceAccess:
@dataclasses.dataclass(frozen=True, slots=True)
 class WorkspaceMemberView:
    membership: WorkspaceMembership
    display_name: str
    email: str
@@ -294,7 +296,7 @@ class WorkspaceCollaborationService:
        async def operation(active_session: AsyncSession) -> list[WorkspaceMemberView]:
            await self._load_actor(active_session, workspace_uuid, actor)
            statement = (
-                sqlalchemy.select(WorkspaceMembership, User.user)
+                sqlalchemy.select(WorkspaceMembership, User.user, User.normalized_email)
                .join(User, User.uuid == WorkspaceMembership.account_uuid)
                .where(
                    WorkspaceMembership.workspace_uuid == workspace_uuid,
@@ -304,8 +306,12 @@ class WorkspaceCollaborationService:
                .order_by(WorkspaceMembership.created_at, WorkspaceMembership.uuid)
            )
            return [
-                WorkspaceMemberView(membership=membership, email=email)
+                WorkspaceMemberView(
-                for membership, email in (await active_session.execute(statement)).all()
+                    membership=membership,
                    display_name=display_name,
                    email=email,
                )
                for membership, display_name, email in (await active_session.execute(statement)).all()
            ]
        return await self._run(operation, session=session, read_only=True)
@@ -478,6 +484,7 @@ class WorkspaceCollaborationService:
                    account_uuid=account_uuid,
                    role=invitation.role,
                    status=MembershipStatus.ACTIVE.value,
                    source=MembershipSource.LOCAL.value,
                    invited_by_account_uuid=invitation.created_by_account_uuid,
                    joined_at=now,
                    projection_revision=0,
@@ -486,6 +493,7 @@ class WorkspaceCollaborationService:
            elif membership.status != MembershipStatus.ACTIVE.value:
                membership.role = invitation.role
                membership.status = MembershipStatus.ACTIVE.value
                membership.source = MembershipSource.LOCAL.value
                membership.invited_by_account_uuid = invitation.created_by_account_uuid
                membership.joined_at = now
@@ -499,14 +507,14 @@ class WorkspaceCollaborationService:
            return await self._run(operation, session=session)
    @asynccontextmanager
-    async def _invitation_lock(self, token_digest: str):
+    async def _invitation_lock(self, lock_key: str):
-        """Serialize one token while retaining only active lock entries."""
+        """Serialize one token within workspace scope while retaining only active lock entries."""
        async with self._invitation_locks_guard:
-            entry = self._invitation_locks.get(token_digest)
+            entry = self._invitation_locks.get(lock_key)
            if entry is None:
                entry = _InvitationLockEntry(lock=asyncio.Lock())
-                self._invitation_locks[token_digest] = entry
+                self._invitation_locks[lock_key] = entry
            entry.users += 1
        await entry.lock.acquire()
@@ -517,7 +525,7 @@ class WorkspaceCollaborationService:
            async with self._invitation_locks_guard:
                entry.users -= 1
                if entry.users == 0:
-                    self._invitation_locks.pop(token_digest, None)
+                    self._invitation_locks.pop(lock_key, None)
    async def revoke_invitation(
        self,
@@ -606,6 +614,8 @@ class WorkspaceCollaborationService:
    ) -> WorkspaceMembership:
        if role not in {item.value for item in MembershipRole}:
            raise MembershipPermissionError('Unknown Workspace role')
        if role == MembershipRole.OWNER.value:
            raise MembershipPermissionError('Workspace ownership cannot be transferred')
        async def operation(active_session: AsyncSession) -> WorkspaceMembership:
            await self._require_active_workspace(active_session, workspace_uuid)
@@ -617,8 +627,8 @@ class WorkspaceCollaborationService:
                target_account_uuid,
            )
            self._require_can_manage_target(persisted_actor, target, new_role=role)
-            if target.role == MembershipRole.OWNER.value and role != MembershipRole.OWNER.value:
+            if target.role == MembershipRole.OWNER.value:
-                await self._require_another_owner(active_session, workspace_uuid, target.account_uuid)
+                raise LastOwnerError('The Workspace owner cannot be removed or demoted')
            target.role = role
            await active_session.flush()
            return target
@@ -644,7 +654,7 @@ class WorkspaceCollaborationService:
            )
            self._require_can_manage_target(persisted_actor, target)
            if target.role == MembershipRole.OWNER.value:
-                await self._require_another_owner(active_session, workspace_uuid, target.account_uuid)
+                raise LastOwnerError('The Workspace owner cannot be removed or demoted')
            target.status = MembershipStatus.REMOVED.value
            await active_session.flush()
            return target
@@ -751,26 +761,6 @@ class WorkspaceCollaborationService:
            raise WorkspaceNotFoundError('Workspace not found')
        return persisted_actor
    async def _require_another_owner(
        self,
        session: AsyncSession,
        workspace_uuid: str,
        excluded_account_uuid: str,
    ) -> None:
        owners = (
            await session.scalars(
                sqlalchemy.select(WorkspaceMembership)
                .where(
                    WorkspaceMembership.workspace_uuid == workspace_uuid,
                    WorkspaceMembership.status == MembershipStatus.ACTIVE.value,
                    WorkspaceMembership.role == MembershipRole.OWNER.value,
                )
                .with_for_update()
            )
        ).all()
        if not any(owner.account_uuid != excluded_account_uuid for owner in owners):
            raise LastOwnerError('The last Workspace owner cannot be removed or demoted')
    def _require_actor_workspace(self, actor: WorkspaceMembership, workspace_uuid: str) -> None:
        if actor.workspace_uuid != workspace_uuid or actor.status != MembershipStatus.ACTIVE.value:
            raise WorkspaceNotFoundError('Workspace not found')
@@ -1,5 +1,6 @@
 from __future__ import annotations
 import datetime
 from dataclasses import dataclass
@@ -12,3 +13,4 @@ class WorkspaceExecutionBinding:
    placement_generation: int
    write_fenced: bool
    state: str
    workspace_created_at: datetime.datetime | None = None
@@ -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))
@@ -11,6 +11,7 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 from ..entity.persistence.workspace import (
    MembershipRole,
    MembershipSource,
    MembershipStatus,
    Workspace,
    WorkspaceExecutionSource,
@@ -30,6 +31,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
@@ -282,6 +284,7 @@ class WorkspaceService:
                placement_generation=execution_state.active_generation,
                write_fenced=execution_state.write_fenced,
                state=execution_state.state,
                workspace_created_at=workspace.created_at,
            )
        binding = await self._run(operation, session=session)
@@ -449,6 +452,7 @@ class WorkspaceService:
                account_uuid=account_uuid,
                role=MembershipRole.OWNER.value,
                status=MembershipStatus.ACTIVE.value,
                source=MembershipSource.LOCAL.value,
                joined_at=joined_at,
                projection_revision=0,
            )
@@ -456,6 +460,7 @@ class WorkspaceService:
        else:
            membership.role = MembershipRole.OWNER.value
            membership.status = MembershipStatus.ACTIVE.value
            membership.source = MembershipSource.LOCAL.value
            membership.joined_at = membership.joined_at or joined_at
        if workspace.created_by_account_uuid is None:
@@ -497,7 +502,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,
@@ -105,6 +105,7 @@ system:
        max_bots: -1
        max_pipelines: -1
        max_extensions: -1
        max_knowledge_bases: -1
        # When set to a non-empty string, every pipeline is forced to use this
        # Box sandbox-scope template regardless of its own configuration, and
        # the per-pipeline "Sandbox Scope" selector is locked in the web UI.
@@ -200,7 +201,7 @@ vdb:
        # keep this false when deliberately using an external pgvector DB.
        use_business_database: false
        # Release migrations create one partial ANN index per enabled value.
-        allowed_dimensions: [384, 512, 768, 1024, 1536]
+        allowed_dimensions: [384, 512, 768, 1024, 1536, 3072]
        host: '127.0.0.1'
        port: 5433
        database: 'langbot'
@@ -244,6 +245,8 @@ storage:
        max_concurrency: 16
 plugin:
    enable: true
    # Maximum time for the Runtime transport, handshake, and desired-state replay.
    connect_timeout_seconds: 180.0
    runtime_ws_url: 'ws://langbot_plugin_runtime:5400/control/ws'
    enable_marketplace: true
    display_plugin_debug_url: 'ws://localhost:5401/plugin/debug/ws'
@@ -106,7 +106,12 @@ async def plugin_security_api(plugin_module):
    application.plugin_connector.require_workspace_context = AsyncMock()
    application.plugin_connector.list_plugins = AsyncMock(return_value=[raw_plugin])
    application.plugin_connector.get_plugin_info = AsyncMock(return_value=raw_plugin)
-    application.plugin_connector.get_debug_info = AsyncMock(return_value={'plugin_debug_key': 'runtime-debug-secret'})
+    application.plugin_connector.get_debug_info = AsyncMock(
        return_value={
            'plugin_debug_key': 'runtime-debug-secret',
            'expires_at': '2026-08-04T12:00:00Z',
        }
    )
    application.plugin_connector.get_plugin_logs = AsyncMock(return_value=['private runtime line'])
    application.plugin_connector.set_plugin_config = AsyncMock()
@@ -232,8 +237,9 @@ async def test_debug_key_requires_resource_manage_permission(plugin_security_api
    assert (await allowed.get_json())['data'] == {
        'debug_url': 'http://localhost:5401',
        'plugin_debug_key': 'runtime-debug-secret',
        'expires_at': '2026-08-04T12:00:00Z',
    }
-    application.plugin_connector.get_debug_info.assert_awaited_once_with()
+    application.plugin_connector.get_debug_info.assert_awaited_once()
@pytest.mark.asyncio
@@ -0,0 +1,79 @@
 """Skills API behavior when a workspace plan has no managed sandbox."""
 from __future__ import annotations
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, Mock
 import pytest
 import quart
 from langbot.pkg.api.http.controller.groups.skills import SkillsRouterGroup
 from langbot.pkg.cloud.entitlements import (
    EntitlementFeatureUnavailableError,
    EntitlementUnavailableError,
 )
 pytestmark = pytest.mark.integration
 WORKSPACE_UUID = '11111111-1111-4111-8111-111111111111'
@pytest.fixture
 async def skills_api():
    account = SimpleNamespace(uuid='owner-account', user='owner@example.com')
    access = SimpleNamespace(
        workspace=SimpleNamespace(uuid=WORKSPACE_UUID),
        membership=SimpleNamespace(uuid='member-owner', role='owner', projection_revision=1),
        execution=SimpleNamespace(instance_uuid='instance-a', placement_generation=1),
    )
    application = Mock()
    application.deployment = SimpleNamespace(multi_workspace_enabled=False)
    application.persistence_mgr = SimpleNamespace(tenant_uow=None)
    application.user_service.get_authenticated_account = AsyncMock(return_value=account)
    application.workspace_collaboration_service.resolve_account_workspace = AsyncMock(return_value=access)
    application.skill_service.list_skills = AsyncMock(
        side_effect=EntitlementFeatureUnavailableError(
            'managed_sandbox',
            entitlement_revision=1,
        )
    )
    quart_app = quart.Quart(__name__)
    router = SkillsRouterGroup(application, quart_app)
    await router.initialize()
    return application, quart_app.test_client()
@pytest.mark.asyncio
 async def test_list_skills_is_empty_when_plan_has_no_managed_sandbox(skills_api):
    application, client = skills_api
    response = await client.get(
        '/api/v1/skills',
        headers={
            'Authorization': 'Bearer owner-token',
            'X-Workspace-Id': WORKSPACE_UUID,
        },
    )
    assert response.status_code == 200
    payload = await response.get_json()
    assert payload['data'] == {'skills': []}
    application.skill_service.list_skills.assert_awaited_once()
@pytest.mark.asyncio
 async def test_list_skills_does_not_hide_other_entitlement_failures(skills_api):
    application, client = skills_api
    application.skill_service.list_skills.side_effect = EntitlementUnavailableError(
        'Workspace entitlement revision rolled back'
    )
    response = await client.get(
        '/api/v1/skills',
        headers={
            'Authorization': 'Bearer owner-token',
            'X-Workspace-Id': WORKSPACE_UUID,
        },
    )
    assert response.status_code == 500
@@ -9,6 +9,8 @@ Run: uv run pytest tests/integration/api/test_smoke.py -q
 from __future__ import annotations
 from types import SimpleNamespace
 import pytest
 from unittest.mock import MagicMock, AsyncMock, Mock
@@ -304,12 +306,34 @@ class TestUserInitEndpoint:
        data = await response.get_json()
        assert data['data'] == {
            'initialized': True,
            'authenticated_invitation_acceptance_enabled': False,
            'password_login_enabled': True,
            'space_login_enabled': False,
        }
        fake_api_app.user_service.get_login_capabilities.assert_awaited_once_with()
        fake_api_app.user_service.get_first_user.assert_not_awaited()
    @pytest.mark.asyncio
    async def test_account_info_enables_authenticated_invitation_acceptance_in_cloud(
        self, quart_test_client, fake_api_app
    ):
        fake_api_app.deployment = SimpleNamespace(mode='cloud')
        fake_api_app.user_service.is_initialized.return_value = True
        fake_api_app.user_service.get_login_capabilities = AsyncMock(
            return_value={'password_login_enabled': True, 'space_login_enabled': True}
        )
        response = await quart_test_client.get('/api/v1/user/account-info')
        assert response.status_code == 200
        data = await response.get_json()
        assert data['data'] == {
            'initialized': True,
            'authenticated_invitation_acceptance_enabled': True,
            'password_login_enabled': False,
            'space_login_enabled': True,
        }
    @pytest.mark.asyncio
    async def test_recovery_key_resets_any_existing_account(self, quart_test_client, fake_api_app, monkeypatch):
        fake_api_app.user_service.is_initialized.return_value = True
@@ -0,0 +1,452 @@
 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.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
        )
@@ -2,6 +2,7 @@
 from __future__ import annotations
 import datetime
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, Mock
 from urllib.parse import parse_qs, urlsplit
@@ -14,6 +15,7 @@ from langbot.pkg.api.http.controller.groups.user import UserRouterGroup
 pytestmark = pytest.mark.integration
 WORKSPACE_UUID = '11111111-1111-4111-8111-111111111111'
 WORKSPACE_CREATED_AT = datetime.datetime(2026, 1, 2, 3, 4, 5, tzinfo=datetime.UTC)
@pytest.fixture
@@ -58,6 +60,12 @@ async def space_oauth_api():
        return_value={'account_uuid': 'account-a', 'workspace_uuid': WORKSPACE_UUID}
    )
    application.workspace_collaboration_service.resolve_account_workspace = AsyncMock(return_value=access)
    application.workspace_service.get_execution_binding = AsyncMock(
        return_value=SimpleNamespace(
            workspace_uuid=WORKSPACE_UUID,
            workspace_created_at=WORKSPACE_CREATED_AT,
        )
    )
    application.space_service.get_oauth_authorize_url = Mock(
        side_effect=lambda redirect_uri, state: f'https://space.example/authorize?state={state}'
    )
@@ -234,7 +242,11 @@ async def test_login_callback_requires_and_consumes_server_state(space_oauth_api
    assert response.status_code == 200
    assert (await response.get_json())['data']['token'] == 'space-login-token'
    application.user_service.consume_space_oauth_state_details.assert_awaited_once_with('opaque-login-state', 'login')
-    application.space_service.exchange_oauth_code.assert_awaited_once_with('oauth-code')
+    application.space_service.exchange_oauth_code.assert_awaited_once_with(
        'oauth-code',
        [WORKSPACE_UUID],
        {WORKSPACE_UUID: int(WORKSPACE_CREATED_AT.timestamp())},
    )
@pytest.mark.asyncio
@@ -270,11 +282,12 @@ 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},
    )
    payload = await response.get_json()
    assert response.status_code == 200
-    assert (await response.get_json())['data'] == {
+    assert payload['data'] == {
        'credits': 25000,
        'owner_space_bound': True,
        'is_workspace_owner': True,
@@ -282,6 +295,31 @@ async def test_space_credits_are_resolved_from_workspace_owner(space_oauth_api):
    application.space_service.get_credits.assert_awaited_once_with('owner@example.com')
@pytest.mark.asyncio
 async def test_cloud_workspace_owner_is_always_space_bound_after_login(space_oauth_api):
    application, client = space_oauth_api
    application.deployment.mode = 'cloud'
    application.user_service.get_workspace_owner = AsyncMock(return_value=None)
    application.space_service.get_credits = AsyncMock()
    application.cloud_model_catalog_service = SimpleNamespace(
        get_workspace_credits=lambda workspace_uuid: 25000 if workspace_uuid == WORKSPACE_UUID else None
    )
    response = await client.get(
        '/api/v1/user/space-credits',
        headers={'Authorization': 'Bearer account-token', 'X-Workspace-Id': WORKSPACE_UUID},
    )
    payload = await response.get_json()
    assert response.status_code == 200
    assert payload['data'] == {
        'credits': 25000,
        'owner_space_bound': True,
        'is_workspace_owner': True,
    }
    application.space_service.get_credits.assert_not_awaited()
@pytest.mark.asyncio
 async def test_bind_callback_uses_opaque_state_and_never_treats_it_as_jwt(space_oauth_api):
    application, client = space_oauth_api
@@ -188,6 +188,7 @@ async def test_owner_invites_second_account_and_secret_is_not_persisted(workspac
    workspace_uuid = current['workspace']['uuid']
    assert current['membership']['role'] == 'owner'
    assert 'member.invite' in current['permissions']
    assert 'owner.transfer' not in current['permissions']
    invite_response = await client.post(
        f'/api/v1/workspaces/{workspace_uuid}/invitations',
@@ -263,6 +264,14 @@ async def test_owner_invites_second_account_and_secret_is_not_persisted(workspac
    assert member_current['membership']['role'] == 'viewer'
    assert 'member.invite' not in member_current['permissions']
    transfer_response = await client.patch(
        f'/api/v1/workspaces/{workspace_uuid}/members/{member_current["membership"]["account_uuid"]}',
        headers=_auth(owner_token, workspace_uuid),
        json={'role': 'owner'},
    )
    assert transfer_response.status_code == 403
    assert (await transfer_response.get_json())['code'] == 'permission_denied'
    forbidden_invite = await client.post(
        f'/api/v1/workspaces/{workspace_uuid}/invitations',
        headers=_auth(member_token, workspace_uuid),
@@ -272,6 +281,31 @@ async def test_owner_invites_second_account_and_secret_is_not_persisted(workspac
    assert (await forbidden_invite.get_json())['code'] == 'permission_denied'
 async def test_workspace_member_list_returns_display_name_and_email(workspace_api):
    _, client, engine, owner_token = workspace_api
    current_response = await client.get('/api/v1/workspaces/current', headers=_auth(owner_token))
    current = (await current_response.get_json())['data']
    workspace_uuid = current['workspace']['uuid']
    owner_uuid = current['membership']['account_uuid']
    async with engine.begin() as connection:
        await connection.execute(
            sqlalchemy.update(User).where(User.uuid == owner_uuid).values(user='Owner Display Name')
        )
    response = await client.get(
        f'/api/v1/workspaces/{workspace_uuid}/members',
        headers=_auth(owner_token, workspace_uuid),
    )
    assert response.status_code == 200
    members = (await response.get_json())['data']['members']
    assert len(members) == 1
    assert members[0]['display_name'] == 'Owner Display Name'
    assert members[0]['email'] == 'owner@example.com'
 async def test_oss_invitation_accept_requires_logout_before_registration(workspace_api):
    _, client, _, owner_token = workspace_api
@@ -0,0 +1,70 @@
 from __future__ import annotations
 import pytest
 import sqlalchemy as sa
 from sqlalchemy.ext.asyncio import create_async_engine
 from langbot.pkg.persistence.alembic_runner import run_alembic_stamp, run_alembic_upgrade
@pytest.mark.asyncio
 async def test_membership_source_migration_backfills_existing_rows_as_local_and_enforces_constraint(tmp_path):
    engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "membership-source.db"}')
    try:
        async with engine.begin() as connection:
            await connection.execute(
                sa.text(
                    """
                    CREATE TABLE workspace_memberships (
                        uuid VARCHAR(36) PRIMARY KEY,
                        workspace_uuid VARCHAR(36) NOT NULL,
                        account_uuid VARCHAR(36) NOT NULL,
                        role VARCHAR(32) NOT NULL,
                        status VARCHAR(32) NOT NULL,
                        projection_revision BIGINT NOT NULL DEFAULT 0,
                        created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
                        updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
                    )
                    """
                )
            )
            await connection.execute(
                sa.text(
                    """
                    INSERT INTO workspace_memberships
                        (uuid, workspace_uuid, account_uuid, role, status, projection_revision)
                    VALUES
                        ('00000000-0000-4000-8000-000000000001', 'workspace', 'local-account',
                         'viewer', 'active', 0),
                        ('00000000-0000-4000-8000-000000000002', 'workspace', 'cloud-account',
                         'viewer', 'active', 0)
                    """
                )
            )
        await run_alembic_stamp(engine, '0019_single_workspace_owner')
        await run_alembic_upgrade(engine, 'head')
        async with engine.connect() as connection:
            rows = (
                await connection.execute(sa.text('SELECT uuid, source FROM workspace_memberships ORDER BY uuid'))
            ).all()
            columns = await connection.run_sync(
                lambda sync_connection: {
                    column['name']: column
                    for column in sa.inspect(sync_connection).get_columns('workspace_memberships')
                }
            )
        assert rows == [
            ('00000000-0000-4000-8000-000000000001', 'local'),
            ('00000000-0000-4000-8000-000000000002', 'local'),
        ]
        assert columns['source']['nullable'] is False
        with pytest.raises(sa.exc.IntegrityError):
            async with engine.begin() as connection:
                await connection.execute(
                    sa.text("UPDATE workspace_memberships SET source = 'guessed-from-user-source'")
                )
    finally:
        await engine.dispose()
@@ -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() == '0020_membership_source'
    @pytest.mark.asyncio
    async def test_upgrade_from_baseline_to_head(self, sqlite_engine):
        """
@@ -85,6 +85,32 @@ async def clean_database(postgres_engine: AsyncEngine):
    await clean()
 async def test_upgrade_adds_3072_dimension_index_and_constraint(
    postgres_engine: AsyncEngine,
    clean_database,
 ) -> None:
    async with postgres_engine.begin() as conn:
        await conn.execute(text('CREATE EXTENSION IF NOT EXISTS vector'))
        await conn.run_sync(Base.metadata.create_all)
    await run_alembic_stamp(postgres_engine, '0010_scope_resources')
    await run_alembic_upgrade(postgres_engine, 'head')
    async with postgres_engine.connect() as conn:
        constraint = await conn.scalar(
            text(
                'SELECT pg_get_constraintdef(oid) FROM pg_constraint '
                "WHERE conrelid = 'langbot_vectors'::regclass "
                "AND conname = 'ck_langbot_vectors_embedding_dimension_enabled'"
            )
        )
        assert '3072' in constraint
        index_definition = await conn.scalar(
            text("SELECT indexdef FROM pg_indexes WHERE indexname = 'ix_langbot_vectors_hnsw_cosine_3072'")
        )
        assert 'halfvec(3072)' in index_definition
        assert 'halfvec_cosine_ops' in index_definition
 async def test_legacy_upgrade_temporarily_suspends_and_restores_source_rls_for_unprivileged_owner(
    postgres_url: str,
    postgres_engine: AsyncEngine,
@@ -92,7 +92,7 @@ def _application(postgres_url: str, *, runtime_role: str = 'langbot_runtime_not_
                    'use': 'pgvector',
                    'pgvector': {
                        'use_business_database': True,
-                        'allowed_dimensions': [384, 512, 768, 1024, 1536],
+                        'allowed_dimensions': [384, 512, 768, 1024, 1536, 3072],
                    },
                },
            }
@@ -0,0 +1,100 @@
 from __future__ import annotations
 import pytest
 import sqlalchemy as sa
 from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
 from langbot.pkg.entity.persistence.base import Base
 from langbot.pkg.entity.persistence.user import User
 from langbot.pkg.entity.persistence.workspace import Workspace, WorkspaceMembership
 from langbot.pkg.persistence.alembic_runner import run_alembic_stamp, run_alembic_upgrade
@pytest.mark.asyncio
 async def test_single_owner_migration_demotes_historical_extra_owner_and_installs_unique_index(tmp_path):
    engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "single-owner.db"}')
    try:
        async with engine.begin() as connection:
            await connection.run_sync(Base.metadata.create_all)
            await connection.execute(sa.text('DROP INDEX uq_workspace_memberships_one_active_owner'))
        session_factory = async_sessionmaker(engine, expire_on_commit=False)
        workspace_uuid = '00000000-0000-4000-8000-000000000001'
        creator_uuid = '00000000-0000-4000-8000-000000000010'
        promoted_uuid = '00000000-0000-4000-8000-000000000020'
        async with session_factory() as session:
            session.add_all(
                [
                    User(
                        uuid=creator_uuid,
                        user='creator@example.test',
                        normalized_email='creator@example.test',
                        password='hash',
                        account_type='local',
                    ),
                    User(
                        uuid=promoted_uuid,
                        user='promoted@example.test',
                        normalized_email='promoted@example.test',
                        password='hash',
                        account_type='local',
                    ),
                    Workspace(
                        uuid=workspace_uuid,
                        instance_uuid='instance-test',
                        name='Workspace',
                        slug='workspace',
                        type='team',
                        status='active',
                        source='local',
                        created_by_account_uuid=creator_uuid,
                    ),
                    WorkspaceMembership(
                        uuid='00000000-0000-4000-8000-000000000100',
                        workspace_uuid=workspace_uuid,
                        account_uuid=creator_uuid,
                        role='owner',
                        status='active',
                    ),
                    WorkspaceMembership(
                        uuid='00000000-0000-4000-8000-000000000200',
                        workspace_uuid=workspace_uuid,
                        account_uuid=promoted_uuid,
                        role='owner',
                        status='active',
                    ),
                ]
            )
            await session.commit()
        await run_alembic_stamp(engine, '0018_merge_launch_replay')
        await run_alembic_upgrade(engine, 'head')
        async with engine.connect() as connection:
            roles = dict(
                (
                    await connection.execute(
                        sa.text(
                            'SELECT account_uuid, role FROM workspace_memberships '
                            'WHERE workspace_uuid = :workspace_uuid ORDER BY account_uuid'
                        ),
                        {'workspace_uuid': workspace_uuid},
                    )
                ).all()
            )
            assert roles == {creator_uuid: 'owner', promoted_uuid: 'admin'}
            indexes = await connection.run_sync(
                lambda sync_connection: {
                    index['name'] for index in sa.inspect(sync_connection).get_indexes('workspace_memberships')
                }
            )
            assert 'uq_workspace_memberships_one_active_owner' in indexes
        with pytest.raises(sa.exc.IntegrityError):
            async with engine.begin() as connection:
                await connection.execute(
                    sa.text("UPDATE workspace_memberships SET role = 'owner' WHERE account_uuid = :account_uuid"),
                    {'account_uuid': promoted_uuid},
                )
    finally:
        await engine.dispose()
@@ -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:
@@ -27,10 +27,9 @@ def test_owner_has_every_fixed_permission():
    assert ctx.workspace.permissions == frozenset(permission.value for permission in authz.Permission)
-def test_admin_cannot_transfer_owner_delete_workspace_or_link_billing():
+def test_admin_cannot_delete_workspace_or_link_billing():
    ctx = _context(authz.WorkspaceRole.ADMIN)
    assert not authz.has_permission(ctx, authz.Permission.OWNER_TRANSFER)
    assert not authz.has_permission(ctx, authz.Permission.WORKSPACE_DELETE)
    assert not authz.has_permission(ctx, authz.Permission.BILLING_LINK_MANAGE)
    assert authz.has_permission(ctx, authz.Permission.MEMBER_INVITE)
@@ -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()
@@ -34,6 +34,7 @@ class _Rows:
 def _app():
    return SimpleNamespace(
        logger=Mock(),
        instance_config=SimpleNamespace(data={}),
        rag_mgr=SimpleNamespace(
            get_all_knowledge_base_details=AsyncMock(return_value=[]),
            get_knowledge_base_details=AsyncMock(return_value=None),
@@ -119,6 +120,21 @@ async def test_create_validates_schema_and_binds_context():
    )
@pytest.mark.asyncio
 async def test_create_enforces_workspace_knowledge_base_limit():
    app = _app()
    app.instance_config.data = {'system': {'limitation': {'max_knowledge_bases': 2}}}
    app.rag_mgr.get_all_knowledge_base_details.return_value = [{'uuid': 'kb-a'}, {'uuid': 'kb-b'}]
    service = KnowledgeService(app)
    with pytest.raises(ValueError, match=r'Maximum number of knowledge bases \(2\) reached'):
        await service.create_knowledge_base(
            CONTEXT,
            {'knowledge_engine_plugin_id': 'author/engine'},
        )
    app.rag_mgr.create_knowledge_base.assert_not_awaited()
@pytest.mark.asyncio
 async def test_update_rejects_guessed_uuid_and_scopes_reload():
    app = _app()
@@ -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(
@@ -833,6 +834,44 @@ class TestModelProviderServiceScanProviderModels:
        assert len(result['models']) == 1
        assert result['models'][0]['type'] == 'llm'
    async def test_scan_provider_marks_existing_rerank_model(self):
        """Rerank scan results use the rerank service when computing already_added."""
        ap = SimpleNamespace()
        ap.persistence_mgr = SimpleNamespace()
        ap.model_mgr = SimpleNamespace()
        ap.llm_model_service = SimpleNamespace()
        ap.embedding_models_service = SimpleNamespace()
        ap.rerank_models_service = SimpleNamespace()
        provider = _create_mock_provider(provider_uuid='rerank-scan-uuid')
        ap.persistence_mgr.execute_async = AsyncMock(return_value=_create_mock_result([], first_item=provider))
        ap.persistence_mgr.serialize_model = Mock(
            return_value={
                'uuid': 'rerank-scan-uuid',
                'name': 'New API',
                'requester': 'new-api-chat-completions',
                'base_url': 'https://new-api.example.com/v1',
                'api_keys': ['key'],
            }
        )
        runtime_provider = Mock()
        runtime_provider.token_mgr.get_token.return_value = 'token'
        runtime_provider.requester.scan_models = AsyncMock(
            return_value={'models': [{'id': 'Qwen3-Reranker-8B', 'type': 'rerank'}]}
        )
        ap.model_mgr.load_provider = AsyncMock(return_value=runtime_provider)
        ap.llm_model_service.get_llm_models_by_provider = AsyncMock(return_value=[])
        ap.embedding_models_service.get_embedding_models_by_provider = AsyncMock(return_value=[])
        ap.rerank_models_service.get_rerank_models_by_provider = AsyncMock(return_value=[{'name': 'Qwen3-Reranker-8B'}])
        result = await ModelProviderService(ap).scan_provider_models(
            WORKSPACE_UUID, 'rerank-scan-uuid', model_type='rerank'
        )
        assert result['models'][0]['type'] == 'rerank'
        assert result['models'][0]['already_added'] is True
    async def test_scan_provider_not_implemented_raises_error(self):
        """Raises ValueError when scan not implemented."""
        # Setup
@@ -967,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
@@ -25,6 +25,7 @@ import time
 from langbot.pkg.api.http.service.space import SpaceService
 from langbot.pkg.entity.persistence.user import User
 from langbot.pkg.utils import constants
 pytestmark = pytest.mark.asyncio
@@ -573,10 +574,20 @@ class TestSpaceServiceExchangeOAuthCode:
            mock_session_obj.post.return_value.__aexit__ = AsyncMock(return_value=None)
            # Execute
-            result = await service.exchange_oauth_code('auth_code')
+            result = await service.exchange_oauth_code(
                'auth_code',
                ['workspace-1'],
                {'workspace-1': 1_700_000_000},
            )
        # Verify
        assert result['access_token'] == 'new_access_token'
        assert mock_session_obj.post.call_args.kwargs['json'] == {
            'code': 'auth_code',
            'instance_id': constants.instance_id,
            'workspace_uuids': ['workspace-1'],
            'workspace_created_ats': {'workspace-1': 1_700_000_000},
        }
    async def test_exchange_oauth_code_api_error(self):
        """Raises ValueError on API error."""
@@ -756,7 +767,7 @@ class TestSpaceServiceGetModels:
                            'uuid': 'uuid-2',
                            'model_id': 'model-2',
                            'provider': 'provider-2',
-                            'category': 'chat',
+                            'category': 'rerank',
                            'status': 'active',
                        },
                    ]
@@ -778,6 +789,7 @@ class TestSpaceServiceGetModels:
        # Verify
        assert len(result) == 2
        assert result[1].category == 'rerank'
    async def test_get_models_api_error(self):
        """Raises ValueError on API error."""
@@ -418,6 +418,7 @@ class TestUserServiceGenerateJwtToken:
        assert token is not None
 class TestUserServiceVerifyJwtToken:
    """Tests for verify_jwt_token method."""
@@ -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] = {}
@@ -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(
@@ -110,24 +124,37 @@ async def test_background_plugin_operation_refences_captured_generation(plugin_r
@pytest.mark.asyncio
-async def test_background_plugin_operation_revalidates_inside_short_tenant_uow(plugin_router_cls):
+async def test_background_plugin_operation_revalidates_and_runs_inside_tenant_uow(plugin_router_cls):
    scopes = []
    active_scope = None
    transaction_active = False
    @asynccontextmanager
-    async def tenant_uow(workspace_uuid):
+    async def tenant_scope(workspace_uuid):
        nonlocal active_scope
        scopes.append(workspace_uuid)
-        yield
+        active_scope = workspace_uuid
        try:
            yield
        finally:
            active_scope = None
    connector = SimpleNamespace(
        require_workspace_context=AsyncMock(side_effect=lambda context: context),
    )
-    operation = AsyncMock(return_value='done')
+
    async def operation():
        assert active_scope == CONTEXT.workspace_uuid
        assert transaction_active is False
        return 'done'
    router = object.__new__(plugin_router_cls)
    router.ap = SimpleNamespace(
        plugin_connector=connector,
        persistence_mgr=SimpleNamespace(
            mode=SimpleNamespace(value='cloud_runtime'),
-            tenant_uow=tenant_uow,
+            tenant_scope=tenant_scope,
        ),
    )
@@ -136,4 +163,3 @@ async def test_background_plugin_operation_revalidates_inside_short_tenant_uow(p
    assert result == 'done'
    assert scopes == [CONTEXT.workspace_uuid]
    connector.require_workspace_context.assert_awaited_once_with(CONTEXT)
    operation.assert_awaited_once()
@@ -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',
        )
@@ -103,7 +108,7 @@ def _cloud_config() -> dict:
            'use': 'pgvector',
            'pgvector': {
                'use_business_database': True,
-                'allowed_dimensions': [384, 768, 1536],
+                'allowed_dimensions': [384, 768, 1536, 3072],
            },
        },
        'mcp': {'stdio': {'enabled': False}},
@@ -211,7 +216,6 @@ async def test_cloud_directory_capacity_contract_is_fail_closed(directory_config
    [
        ({'use_business_database': False, 'allowed_dimensions': [1536]}, 'use_business_database=true'),
        ({'use_business_database': True, 'allowed_dimensions': []}, 'allowed_dimensions'),
        ({'use_business_database': True, 'allowed_dimensions': [3072]}, 'allowed_dimensions'),
        ({'use_business_database': True, 'allowed_dimensions': [True]}, 'allowed_dimensions'),
    ],
 )
@@ -228,10 +232,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'),
        (
@@ -181,6 +181,39 @@ def _delta(
    )
 async def test_directory_delta_requests_model_catalog_sync_after_commit(projection_context):
    application, _session_factory = projection_context
    request_sync = Mock()
    application.cloud_model_catalog_service = SimpleNamespace(request_sync=request_sync)
    event = DirectoryEvent(
        cursor=2,
        uuid='20000000-0000-4000-8000-000000000002',
        aggregate_uuid=WORKSPACE_UUID,
        event_type='directory.changed',
        revision=2,
        payload={'workspace_uuid': WORKSPACE_UUID, 'directory_revision': 2},
        created_at=datetime.datetime(2026, 7, 24, 12, 30, tzinfo=datetime.UTC),
    )
    batch = DirectoryEventBatch(
        instance_uuid=INSTANCE_UUID,
        after_cursor=1,
        cursor=2,
        high_water_cursor=2,
        events=[event],
    )
    service = DirectoryProjectionService(
        application,
        _Provider([_snapshot(1)], [batch], [_delta(workspaces=[_workspace(revision=2)])]),
        INSTANCE_UUID,
    )
    await service.initialize()
    request_sync.reset_mock()
    await service.sync_once()
    request_sync.assert_called_once_with()
 async def test_initial_snapshot_projects_core_owned_rows(projection_context):
    application, session_factory = projection_context
    reconcile_execution_projection = Mock()
@@ -1023,7 +1056,7 @@ async def test_snapshot_for_another_instance_is_rejected(projection_context):
        await service.initialize()
-async def test_core_owned_membership_survives_directory_updates_and_omission(projection_context):
+async def test_directory_revision_zero_membership_is_adopted(projection_context):
    application, session_factory = projection_context
    service = DirectoryProjectionService(application, _Provider([_snapshot(1)]), INSTANCE_UUID)
    await service.initialize()
@@ -1034,29 +1067,125 @@ async def test_core_owned_membership_survives_directory_updates_and_omission(pro
            membership.role = 'viewer'
            membership.status = 'active'
            membership.projection_revision = 0
            session.add(
                WorkspaceMembership(
                    uuid=SECOND_MEMBERSHIP_UUID,
                    workspace_uuid=WORKSPACE_UUID,
                    account_uuid='20000000-0000-0000-0000-000000000099',
                    role='viewer',
                    status='active',
                    joined_at=membership.joined_at,
                    projection_revision=0,
                )
            )
    projected_member = _member(revision=2).model_copy(update={'role': 'owner', 'membership_status': 'removed'})
    projected_workspace = _workspace(revision=2).model_copy(update={'members': (projected_member,)})
    await service.apply_snapshot(_snapshot(2, workspaces=[projected_workspace]))
    async with session_factory() as session:
-        memberships = {
+        membership = await session.scalar(sqlalchemy.select(WorkspaceMembership))
-            membership.uuid: membership
+    assert membership.source == 'cloud_projection'
-            for membership in (await session.scalars(sqlalchemy.select(WorkspaceMembership))).all()
+    assert membership.role == 'owner'
-        }
+    assert membership.status == 'removed'
-    assert memberships[MEMBERSHIP_UUID].role == 'viewer'
+    assert membership.projection_revision == 2
-    assert memberships[MEMBERSHIP_UUID].status == 'active'
+
-    assert memberships[MEMBERSHIP_UUID].projection_revision == 0
+
-    assert memberships[SECOND_MEMBERSHIP_UUID].status == 'active'
+async def test_directory_revision_zero_membership_omitted_from_snapshot_is_removed(projection_context):
-    assert memberships[SECOND_MEMBERSHIP_UUID].projection_revision == 0
+    application, session_factory = projection_context
    service = DirectoryProjectionService(application, _Provider([_snapshot(1)]), INSTANCE_UUID)
    await service.initialize()
    historical_account_uuid = '20000000-0000-0000-0000-000000000099'
    async with session_factory() as session:
        async with session.begin():
            membership = await session.scalar(sqlalchemy.select(WorkspaceMembership))
            session.add(
                User(
                    uuid=historical_account_uuid,
                    user='Historical Space Member',
                    normalized_email='historical@example.com',
                    password='',
                    status='active',
                    source='cloud_projection',
                    projection_revision=1,
                    account_type='space',
                    space_account_uuid=historical_account_uuid,
                )
            )
            session.add(
                WorkspaceMembership(
                    uuid=SECOND_MEMBERSHIP_UUID,
                    workspace_uuid=WORKSPACE_UUID,
                    account_uuid=historical_account_uuid,
                    role='viewer',
                    status='active',
                    source='cloud_projection',
                    joined_at=membership.joined_at,
                    projection_revision=0,
                )
            )
    await service.apply_snapshot(_snapshot(2))
    async with session_factory() as session:
        historical = await session.get(WorkspaceMembership, SECOND_MEMBERSHIP_UUID)
    assert historical.status == 'removed'
    assert historical.projection_revision == 2
 async def test_cloud_account_core_invitation_membership_survives_directory_omission(projection_context):
    application, session_factory = projection_context
    service = DirectoryProjectionService(application, _Provider([_snapshot(1)]), INSTANCE_UUID)
    await service.initialize()
    invited_account_uuid = '20000000-0000-0000-0000-000000000098'
    async with session_factory() as session:
        async with session.begin():
            projected_membership = await session.scalar(sqlalchemy.select(WorkspaceMembership))
            session.add(
                User(
                    uuid=invited_account_uuid,
                    user='Invited Cloud Account',
                    normalized_email='invited-cloud@example.com',
                    password='',
                    status='active',
                    source='cloud_projection',
                    projection_revision=1,
                    account_type='space',
                    space_account_uuid=invited_account_uuid,
                )
            )
            session.add(
                WorkspaceMembership(
                    uuid=SECOND_MEMBERSHIP_UUID,
                    workspace_uuid=WORKSPACE_UUID,
                    account_uuid=invited_account_uuid,
                    role='viewer',
                    status='active',
                    source='local',
                    joined_at=projected_membership.joined_at,
                    projection_revision=0,
                )
            )
    await service.apply_snapshot(_snapshot(2))
    async with session_factory() as session:
        membership = await session.get(WorkspaceMembership, SECOND_MEMBERSHIP_UUID)
    assert membership.source == 'local'
    assert membership.status == 'active'
    assert membership.projection_revision == 0
 async def test_directory_does_not_adopt_local_membership_with_different_uuid_for_same_cloud_account(projection_context):
    application, session_factory = projection_context
    service = DirectoryProjectionService(application, _Provider([_snapshot(1)]), INSTANCE_UUID)
    await service.initialize()
    async with session_factory() as session:
        async with session.begin():
            membership = await session.scalar(sqlalchemy.select(WorkspaceMembership))
            membership.uuid = SECOND_MEMBERSHIP_UUID
            membership.source = 'local'
            membership.projection_revision = 0
    projected_member = _member(revision=2).model_copy(update={'role': 'owner', 'membership_status': 'removed'})
    projected_workspace = _workspace(revision=2).model_copy(update={'members': (projected_member,)})
    await service.apply_snapshot(_snapshot(2, workspaces=[projected_workspace]))
    async with session_factory() as session:
        membership = await session.get(WorkspaceMembership, SECOND_MEMBERSHIP_UUID)
    assert membership.source == 'local'
    assert membership.role == 'developer'
    assert membership.status == 'active'
    assert membership.projection_revision == 0
@@ -0,0 +1,466 @@
 from __future__ import annotations
 import asyncio
 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,
                    'credits': 25000,
                },
                {
                    'workspace_uuid': WORKSPACE_B,
                    'owner_account_uuid': OWNER_B,
                    'api_key': 'owner-b-key',
                    'credits': 5000,
                },
            ],
        }
    )
 async def test_catalog_snapshot_treats_null_model_abilities_as_empty() -> None:
    payload = _snapshot().model_dump(mode='json')
    payload['models'][0]['llm_abilities'] = None
    snapshot = CloudModelCatalogSnapshot.model_validate(payload)
    assert snapshot.models[0].llm_abilities == ()
 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
        assert service.get_workspace_credits(WORKSPACE_A) == 25000
        assert service.get_workspace_credits(WORKSPACE_B) == 5000
        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 test_periodic_sync_discovers_workspace_created_after_startup_cache_release(tmp_path) -> None:
    engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "model-catalog-new-workspace.db"}')
    manager = PersistenceManager(object(), mode=PersistenceMode.CLOUD_RUNTIME)
    manager.db = SimpleNamespace(get_engine=lambda: engine)
    startup_bindings = [
        SimpleNamespace(instance_uuid=INSTANCE_UUID, workspace_uuid=WORKSPACE_A, placement_generation=1)
    ]
    live_bindings = [
        *startup_bindings,
        SimpleNamespace(instance_uuid=INSTANCE_UUID, workspace_uuid=WORKSPACE_B, placement_generation=1),
    ]
    class _WorkspaceService:
        startup_released = False
        async def list_active_execution_bindings(self):
            return list(live_bindings if self.startup_released else startup_bindings)
        def release_startup_execution_bindings(self):
            self.startup_released = True
    workspace_service = _WorkspaceService()
    app = SimpleNamespace(
        persistence_mgr=manager,
        workspace_service=workspace_service,
        model_mgr=SimpleNamespace(load_models_from_db=_AsyncCounter()),
        logger=logging.getLogger(__name__),
    )
    service = CloudModelCatalogSyncService(app, _CatalogProvider(_snapshot()), 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 service.initialize()
        workspace_service.release_startup_execution_bindings()
        await service.sync_once()
        async with engine.connect() as connection:
            provider_b = await connection.scalar(
                sqlalchemy.select(ModelProvider).where(ModelProvider.uuid == system_provider_uuid(WORKSPACE_B))
            )
        assert provider_b is not None
    finally:
        await engine.dispose()
 async def test_catalog_run_wakes_immediately_when_directory_changes() -> None:
    sync_started = asyncio.Event()
    class _WakeService(CloudModelCatalogSyncService):
        async def sync_once(self, *, reload_runtime: bool = True):
            del reload_runtime
            sync_started.set()
            return {'workspaces': 0, 'created': 0, 'updated': 0, 'deleted': 0}
    app = SimpleNamespace(logger=logging.getLogger(__name__))
    service = _WakeService(app, _CatalogProvider(_snapshot()), INSTANCE_UUID, sync_interval_seconds=3600)
    task = asyncio.create_task(service.run())
    try:
        await asyncio.sleep(0)
        service.request_sync()
        await asyncio.wait_for(sync_started.wait(), timeout=0.2)
    finally:
        task.cancel()
        with pytest.raises(asyncio.CancelledError):
            await task
 async def _async_value(value):
    return value
 class _AsyncCounter:
    def __init__(self) -> None:
        self.calls = 0
    async def __call__(self) -> None:
        self.calls += 1
 async def test_partial_workspace_failure_reloads_already_committed_changes() -> None:
    bindings = [
        SimpleNamespace(workspace_uuid=WORKSPACE_A),
        SimpleNamespace(workspace_uuid=WORKSPACE_B),
    ]
    reload_counter = _AsyncCounter()
    app = SimpleNamespace(
        workspace_service=SimpleNamespace(list_active_execution_bindings=lambda: _async_value(bindings)),
        model_mgr=SimpleNamespace(load_models_from_db=reload_counter),
        logger=logging.getLogger(__name__),
    )
    service = CloudModelCatalogSyncService(app, _CatalogProvider(_snapshot()), INSTANCE_UUID)
    calls = 0
    async def sync_workspace(*_args):
        nonlocal calls
        calls += 1
        if calls == 1:
            return {'created': 1, 'updated': 0, 'deleted': 0}
        raise RuntimeError('second Workspace failed')
    service._sync_workspace = sync_workspace  # type: ignore[method-assign]
    with pytest.raises(RuntimeError, match='second Workspace failed'):
        await service.sync_once()
    assert service.get_workspace_credits(WORKSPACE_A) == 25000
    assert service.get_workspace_credits(WORKSPACE_B) is None
    assert reload_counter.calls == 1
 async def test_failed_runtime_reload_is_retried_after_noop_sync() -> None:
    bindings = [SimpleNamespace(workspace_uuid=WORKSPACE_A)]
    class _FlakyReload:
        def __init__(self) -> None:
            self.calls = 0
        async def __call__(self) -> None:
            self.calls += 1
            if self.calls == 1:
                raise RuntimeError('reload failed')
    runtime_reload = _FlakyReload()
    app = SimpleNamespace(
        workspace_service=SimpleNamespace(list_active_execution_bindings=lambda: _async_value(bindings)),
        model_mgr=SimpleNamespace(load_models_from_db=runtime_reload),
        logger=logging.getLogger(__name__),
    )
    service = CloudModelCatalogSyncService(app, _CatalogProvider(_snapshot()), INSTANCE_UUID)
    calls = 0
    async def sync_workspace(*_args):
        nonlocal calls
        calls += 1
        if calls == 1:
            return {'created': 1, 'updated': 0, 'deleted': 0}
        return {'created': 0, 'updated': 0, 'deleted': 0}
    service._sync_workspace = sync_workspace  # type: ignore[method-assign]
    with pytest.raises(RuntimeError, match='reload failed'):
        await service.sync_once()
    summary = await service.sync_once()
    assert summary == {'workspaces': 1, 'created': 0, 'updated': 0, 'deleted': 0}
    assert runtime_reload.calls == 2
 async def test_background_sync_log_redacts_exception_message(caplog) -> None:
    secret = 'owner-secret-api-key'
    attempted = asyncio.Event()
    class _FailingProvider:
        async def fetch_model_catalog(self, instance_uuid: str) -> CloudModelCatalogSnapshot:
            del instance_uuid
            attempted.set()
            raise RuntimeError(f'database parameters include {secret}')
    app = SimpleNamespace(logger=logging.getLogger(__name__))
    service = CloudModelCatalogSyncService(app, _FailingProvider(), INSTANCE_UUID)
    service.sync_interval_seconds = 0.001
    task = asyncio.create_task(service.run())
    try:
        await asyncio.wait_for(attempted.wait(), timeout=1)
        await asyncio.sleep(0.01)
    finally:
        task.cancel()
        with pytest.raises(asyncio.CancelledError):
            await task
    assert secret not in caplog.text
    assert 'Cloud model catalog synchronization failed (RuntimeError)' in caplog.text
@@ -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())
@@ -152,6 +152,19 @@ class TestApplyEnvOverridesToConfig:
        assert result['system']['disabled_adapters'] == ['aiocqhttp', 'dingtalk', 'telegram']
    def test_override_integer_list_preserves_item_type(self):
        """Comma-separated overrides inherit the existing list item type."""
        load_config = get_load_config_module()
        cfg = {'vdb': {'pgvector': {'allowed_dimensions': [384, 512]}}}
        env = {'VDB__PGVECTOR__ALLOWED_DIMENSIONS': '384,512,768'}
        with patch.dict(os.environ, env, clear=True):
            result = load_config._apply_env_overrides_to_config(cfg)
        assert result['vdb']['pgvector']['allowed_dimensions'] == [384, 512, 768]
        assert all(isinstance(item, int) for item in result['vdb']['pgvector']['allowed_dimensions'])
    def test_override_list_value_empty_items(self):
        """Test that empty items in comma-separated list are filtered."""
        load_config = get_load_config_module()
@@ -306,6 +319,7 @@ class TestApplyEnvOverridesToConfig:
        load_config = get_load_config_module()
        cfg = {
            'plugin': {
                'connect_timeout_seconds': 30.0,
                'worker': {
                    'max_cpus': 1.0,
                    'max_memory_mb': 512,
@@ -316,11 +330,12 @@ class TestApplyEnvOverridesToConfig:
                    'restart_failure_threshold': 8,
                    'restart_failure_window_seconds': 30.0,
                    'restart_circuit_open_seconds': 60.0,
-                }
+                },
            },
            'mcp': {'stdio': {'enabled': True}},
        }
        env = {
            'PLUGIN__CONNECT_TIMEOUT_SECONDS': '180',
            'PLUGIN__WORKER__MAX_CPUS': '2.5',
            'PLUGIN__WORKER__MAX_MEMORY_MB': '1024',
            'PLUGIN__WORKER__MAX_PIDS': '64',
@@ -336,6 +351,7 @@ class TestApplyEnvOverridesToConfig:
        with patch.dict(os.environ, env, clear=True):
            result = load_config._apply_env_overrides_to_config(cfg)
        assert result['plugin']['connect_timeout_seconds'] == 180.0
        assert result['plugin']['worker'] == {
            'max_cpus': 2.5,
            'max_memory_mb': 1024,
@@ -380,6 +396,14 @@ class TestApplyEnvOverridesToConfig:
        assert isinstance(result['plugin']['worker']['max_memory_mb'], int)
        assert result['mcp']['stdio']['enabled'] is False
    def test_runtime_policy_defaults_add_typed_plugin_connect_timeout(self):
        load_config = get_load_config_module()
        completed = load_config._complete_runtime_policy_defaults({'plugin': {'enable': True}})
        assert completed['plugin']['connect_timeout_seconds'] == 180.0
        assert isinstance(completed['plugin']['connect_timeout_seconds'], float)
    def test_webhook_prefix_override(self):
        """Test overriding webhook_prefix via environment variable."""
        load_config = get_load_config_module()
@@ -7,7 +7,7 @@ from types import SimpleNamespace
 import pytest
 import sqlalchemy as sa
-from pgvector.sqlalchemy import Vector
+from pgvector.sqlalchemy import HALFVEC, Vector
 from sqlalchemy.dialects.postgresql import insert as postgresql_insert
 from sqlalchemy.dialects.sqlite import insert as sqlite_insert
 from sqlalchemy.ext.asyncio import create_async_engine
@@ -961,11 +961,13 @@ async def test_scoped_session_rejects_raw_or_unapproved_sql(
        sa.select(sa.func.coalesce(sa.func.sum(sa.literal(1)), sa.literal(0))),
        sa.select(
            sa.func.now(),
            sa.func.date_trunc('hour', sa.column('timestamp')),
            sa.func.length(sa.literal('value')),
            sa.func.nullif(sa.literal('value'), sa.literal('')),
        ),
        sa.select(sa.column('embedding').op('<=>')(sa.literal([0.1]))),
        sa.select(sa.cast(sa.column('embedding'), Vector(384))),
        sa.select(sa.cast(sa.column('embedding'), HALFVEC(3072))),
        sa.insert(sa.table('rows', sa.column('id'))).values(id=1),
        _multi_value_statement(value=1),
        _on_conflict_statement(update_value=sa.func.coalesce(sa.literal(1), sa.literal(0))),
@@ -158,17 +158,21 @@ class TestCommandHandlerReal:
    @pytest.mark.asyncio
    async def test_admin_privilege_check(self, fake_app, mock_event_ctx, mock_execute_factory):
-        """Admin users get privilege level 2."""
+        """A per-bot admin from the database is marked as admin in command events."""
        from langbot_plugin.api.entities.builtin.provider.session import LauncherTypes
        command = get_command_handler()
-        fake_app.instance_config.data = {'admins': ['person_12345']}
+        admin_result = Mock()
        admin_result.first.return_value = Mock()
        fake_app.persistence_mgr.execute_async = AsyncMock(return_value=admin_result)
        fake_app.instance_config.data = {}
        fake_app.plugin_connector.emit_event = AsyncMock(return_value=mock_event_ctx)
        fake_app.cmd_mgr.execute = mock_execute_factory()
        handler = command.CommandHandler(fake_app)
        query = command_query('status')
        query.bot_uuid = 'bot-1'
        query.launcher_type = LauncherTypes.PERSON
        query.launcher_id = 12345
@@ -176,23 +180,28 @@ class TestCommandHandlerReal:
        async for result in handler.handle(query):
            results.append(result)
        fake_app.persistence_mgr.execute_async.assert_awaited_once()
        call_args = fake_app.plugin_connector.emit_event.call_args
        event = call_args[0][0]
        assert event.is_admin is True
    @pytest.mark.asyncio
    async def test_non_admin_privilege_check(self, fake_app, mock_event_ctx, mock_execute_factory):
-        """Non-admin users get privilege level 1."""
+        """A launcher absent from the per-bot admin table is not an admin."""
        from langbot_plugin.api.entities.builtin.provider.session import LauncherTypes
        command = get_command_handler()
-        fake_app.instance_config.data = {'admins': ['person_12345']}
+        admin_result = Mock()
        admin_result.first.return_value = None
        fake_app.persistence_mgr.execute_async = AsyncMock(return_value=admin_result)
        fake_app.instance_config.data = {}
        fake_app.plugin_connector.emit_event = AsyncMock(return_value=mock_event_ctx)
        fake_app.cmd_mgr.execute = mock_execute_factory()
        handler = command.CommandHandler(fake_app)
        query = command_query('status')
        query.bot_uuid = 'bot-1'
        query.launcher_type = LauncherTypes.PERSON
        query.launcher_id = 67890
@@ -200,6 +209,7 @@ class TestCommandHandlerReal:
        async for result in handler.handle(query):
            results.append(result)
        fake_app.persistence_mgr.execute_async.assert_awaited_once()
        call_args = fake_app.plugin_connector.emit_event.call_args
        event = call_args[0][0]
        assert event.is_admin is False
@@ -11,6 +11,7 @@ from sqlalchemy.ext.asyncio import create_async_engine
 from langbot.pkg.persistence.mgr import PersistenceManager, PersistenceMode
 from langbot.pkg.persistence.tenant_uow import PersistenceScopeKind
 from langbot.pkg.pipeline.controller import Controller
 from langbot.pkg.pipeline.pool import QueryPool
 from langbot.pkg.workspace.errors import WorkspaceGenerationMismatchError
@@ -143,3 +144,31 @@ async def test_controller_revalidates_generation_before_running_pipeline(
    runtime_pipeline.run.assert_awaited_once_with(sample_query)
    query_pool.remove_query.assert_awaited_once_with(sample_query)
    session._semaphore.release.assert_called_once_with()
@pytest.mark.asyncio
 async def test_controller_schedules_query_without_removing_it_twice(mock_app, sample_query):
    query_pool = QueryPool()
    query_pool.queries.append(sample_query)
    mock_app.query_pool = query_pool
    mock_app.sess_mgr.get_session = AsyncMock(return_value=SimpleNamespace(_semaphore=asyncio.Semaphore(1)))
    scheduler_errors: list[str] = []
    def stop_on_scheduler_error(message):
        scheduler_errors.append(str(message))
        raise asyncio.CancelledError
    def stop_after_scheduling(process_coro, **_kwargs):
        process_coro.close()
        raise asyncio.CancelledError
    mock_app.logger.error.side_effect = stop_on_scheduler_error
    mock_app.task_mgr.create_task.side_effect = stop_after_scheduling
    controller = Controller(mock_app)
    with pytest.raises(asyncio.CancelledError):
        await controller.consumer()
    assert scheduler_errors == []
    assert query_pool.queries == []
@@ -488,3 +488,115 @@ class TestPreProcessorToolSelection:
        result = await stage.process(query, 'PreProcessor')
        assert [tool.name for tool in result.new_query.use_funcs] == ['plugin_tool']
 class TestPreProcessorDateGrounding:
    """Tests for current-date injection into the local-agent system prompt."""
    @pytest.mark.asyncio
    async def test_local_agent_appends_date_to_existing_system_message(self):
        """Date grounding text should be appended to an existing system prompt."""
        preproc = get_preproc_module()
        app = FakeApp()
        mock_session = make_session()
        app.sess_mgr.get_session = AsyncMock(return_value=mock_session)
        mock_conversation = Mock()
        mock_conversation.prompt = Mock(messages=[])
        mock_conversation.prompt.copy = Mock(return_value=Mock(messages=[]))
        mock_conversation.messages = []
        mock_conversation.uuid = None
        app.sess_mgr.get_conversation = AsyncMock(return_value=mock_conversation)
        app.model_mgr.get_model_by_uuid = AsyncMock(return_value=None)
        app.tool_mgr.get_all_tools = AsyncMock(return_value=[])
        from langbot_plugin.api.entities.builtin.provider import message as provider_message
        system_message = provider_message.Message(role='system', content='You are a helpful assistant.')
        mock_event_ctx = Mock()
        mock_event_ctx.event = Mock(default_prompt=[system_message], prompt=[])
        app.plugin_connector.emit_event = AsyncMock(return_value=mock_event_ctx)
        stage = preproc.PreProcessor(app)
        query = text_query('hello')
        result = await stage.process(query, 'PreProcessor')
        messages = result.new_query.prompt.messages
        assert len(messages) == 1
        assert messages[0].role == 'system'
        assert messages[0].content.startswith('You are a helpful assistant.')
        assert 'Current date:' in messages[0].content
    @pytest.mark.asyncio
    async def test_local_agent_creates_system_message_when_none_exists(self):
        """A system message should be created when the prompt has none."""
        preproc = get_preproc_module()
        app = FakeApp()
        mock_session = make_session()
        app.sess_mgr.get_session = AsyncMock(return_value=mock_session)
        mock_conversation = Mock()
        mock_conversation.prompt = Mock(messages=[])
        mock_conversation.prompt.copy = Mock(return_value=Mock(messages=[]))
        mock_conversation.messages = []
        mock_conversation.uuid = None
        app.sess_mgr.get_conversation = AsyncMock(return_value=mock_conversation)
        app.model_mgr.get_model_by_uuid = AsyncMock(return_value=None)
        app.tool_mgr.get_all_tools = AsyncMock(return_value=[])
        mock_event_ctx = Mock()
        mock_event_ctx.event = Mock(default_prompt=[], prompt=[])
        app.plugin_connector.emit_event = AsyncMock(return_value=mock_event_ctx)
        stage = preproc.PreProcessor(app)
        query = text_query('hello')
        result = await stage.process(query, 'PreProcessor')
        messages = result.new_query.prompt.messages
        assert len(messages) == 1
        assert messages[0].role == 'system'
        assert 'Current date:' in messages[0].content
    @pytest.mark.asyncio
    async def test_non_local_agent_runner_skips_date_injection(self):
        """Runners other than local-agent should not get the date addition."""
        preproc = get_preproc_module()
        app = FakeApp()
        mock_session = make_session()
        app.sess_mgr.get_session = AsyncMock(return_value=mock_session)
        mock_conversation = Mock()
        mock_conversation.prompt = Mock(messages=[])
        mock_conversation.prompt.copy = Mock(return_value=Mock(messages=[]))
        mock_conversation.messages = []
        mock_conversation.uuid = None
        app.sess_mgr.get_conversation = AsyncMock(return_value=mock_conversation)
        app.model_mgr.get_model_by_uuid = AsyncMock(return_value=None)
        app.tool_mgr.get_all_tools = AsyncMock(return_value=[])
        mock_event_ctx = Mock()
        mock_event_ctx.event = Mock(default_prompt=[], prompt=[])
        app.plugin_connector.emit_event = AsyncMock(return_value=mock_event_ctx)
        stage = preproc.PreProcessor(app)
        query = text_query('hello')
        query.pipeline_config = {
            'ai': {
                'runner': {'runner': 'dify-service-api'},
                'local-agent': {'model': {'primary': '', 'fallbacks': []}, 'prompt': 'default'},
            },
            'output': {'misc': {'at-sender': False}},
            'trigger': {'misc': {}},
        }
        result = await stage.process(query, 'PreProcessor')
        assert result.new_query.prompt.messages == []
@@ -1,7 +1,8 @@
 """Tests for Telegram Dify form callback helpers."""
 import json
-from unittest.mock import AsyncMock, MagicMock
+from contextlib import asynccontextmanager
 from unittest.mock import AsyncMock, MagicMock, patch
 import pytest
 from telegram import ForceReply
@@ -9,8 +10,10 @@ from telegram import ForceReply
 import langbot_plugin.api.entities.builtin.platform.entities as platform_entities
 import langbot_plugin.api.entities.builtin.platform.events as platform_events
 import langbot_plugin.api.entities.builtin.platform.message as platform_message
 from langbot.pkg.platform.sources.telegram import (
    TelegramAdapter,
    TelegramMessageConverter,
    _decode_telegram_base64_limited,
    _telegram_form_action_from_callback,
    _telegram_select_field_options,
@@ -26,6 +29,70 @@ def test_telegram_base64_decode_is_bounded(monkeypatch):
        _decode_telegram_base64_limited('A' * 12)
 TELEGRAM_BOT_TOKEN = '123456789:AAExampleBotTokenThatMustNotLeak'
 TELEGRAM_FILE_URL = f'https://api.telegram.org/file/bot{TELEGRAM_BOT_TOKEN}/photos/file_0.jpg'
@pytest.mark.asyncio
 async def test_telegram_photo_does_not_expose_bot_token_in_image_url():
    """Regression test for the Telegram bot-token leak.
    telegram.Bot builds file.file_path as
    https://api.telegram.org/file/bot<TOKEN>/<path>, embedding the bot token.
    The converter must not copy that URL into Image.url, or the token leaks to
    the monitoring DB, dashboard and every installed plugin via the message
    chain. Only base64 (which carries no token) may be stored.
    """
    tg_file = MagicMock()
    tg_file.file_path = TELEGRAM_FILE_URL
    photo_size = MagicMock()
    photo_size.get_file = AsyncMock(return_value=tg_file)
    message = MagicMock()
    message.text = None
    message.caption = None
    message.photo = [photo_size]
    message.voice = None
    message.document = None
    response = MagicMock()
    response.headers = {}
    async def iter_chunked(_chunk_size):
        yield b'\xff\xd8\xff\xe0jpeg-bytes'
    response.content.iter_chunked = iter_chunked
    @asynccontextmanager
    async def fake_get(url):
        yield response
    fake_session = MagicMock()
    fake_session.get = fake_get
    with patch(
        'langbot.pkg.platform.sources.telegram.httpclient.get_session',
        return_value=fake_session,
    ):
        chain = await TelegramMessageConverter.target2yiri(message, MagicMock(), 'bot-account')
    images = [c for c in chain if isinstance(c, platform_message.Image)]
    assert len(images) == 1
    image = images[0]
    # The token-bearing URL must not be retained anywhere on the component.
    assert not image.url
    assert image.base64 is not None
    assert image.base64.startswith('data:image/jpeg;base64,')
    # Belt-and-suspenders: the token must not appear in the serialized chain
    # (this is what gets persisted to the monitoring DB and sent to plugins).
    serialized = json.dumps(chain.model_dump(), ensure_ascii=False)
    assert TELEGRAM_BOT_TOKEN not in serialized
    assert 'api.telegram.org/file/bot' not in serialized
 def _select_form_data() -> dict:
    return {
        '_current_input_field': 'choice',
--- a/Show More
+++ b/Show More
Author	SHA1	Message	Date
leonoxo	22c389edc1	fix(pipeline): ground local-agent system prompt with current date (#2399 ) The local-agent runner's system prompt is a static string with no template-variable support, so the model had no anchor for "today" and resolved relative time references (e.g. "this quarter", "latest") against whichever period was best represented in training data instead of the real date, sometimes confidently answering with stale information for time-sensitive questions. PreProcessor now appends a short, deterministically-computed "Current date: ..." note to the system prompt on every request for local-agent pipelines, alongside guidance to verify time-sensitive facts with a search tool rather than answering from memory. The existing skill-awareness prompt injection is refactored to share the same append-to-system-prompt helper.	2026-08-08 22:29:54 +08:00
Hyu	78068db9c8	fix(cloud): preserve tenant scope for extension tasks (#2408 ) * fix(cloud): preserve tenant scope for extension tasks * ci: retrigger extension scope checks --------- Co-authored-by: Chan <dadachann@users.noreply.github.com>	2026-08-07 11:32:38 +08:00
Hyu	7dc9dafb7c	fix(plugin): validate runtime timeout before startup (#2407 ) Co-authored-by: Chan <dadachann@users.noreply.github.com>	2026-08-07 11:26:42 +08:00
Hyu	4bd899e77b	fix(cloud): track workspace membership provenance (#2406 ) * fix(cloud): converge legacy revision-zero members * fix(cloud): track workspace membership provenance * test(persistence): track current migration head --------- Co-authored-by: Chan <dadachann@users.noreply.github.com>	2026-08-07 11:26:06 +08:00
Hyu	ddb6dbf593	fix(cloud): tolerate slow plugin runtime reconciliation (#2405 ) Co-authored-by: Chan <dadachann@users.noreply.github.com>	2026-08-06 20:39:20 +08:00
Hyu	f59343fd5b	fix: provision Cloud models after Workspace activation (#2403 ) Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>	2026-08-05 23:26:23 +08:00
Hyu	211710e24c	fix: allow trusted halfvec tenant search casts (#2402 ) Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>	2026-08-05 22:09:01 +08:00
Hyu	cdd5c6589c	fix: support 3072-dimensional knowledge embeddings (#2401 ) Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>	2026-08-05 21:19:57 +08:00
Hyu	3b4698463c	chore: pin runtime control binding SDK (#2400 ) Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>	2026-08-05 15:12:40 +08:00
Hyu	edd6cad449	chore: pin fenced debug runtime sessions (#2396 ) Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>	2026-08-04 19:59:51 +08:00
Hyu	d78546967c	feat: expose Workspace-scoped rotating plugin debug keys (#2395 ) * feat: add Workspace-scoped rotating plugin debug tokens * chore: pin formatted Workspace debug runtime * chore: pin merged Workspace debug runtime * chore: pin tenant-safe debug runtime --------- Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>	2026-08-04 19:20:06 +08:00
Hyu	c08bfc8ced	feat: report independent instance and workspace identities (#2394 ) * feat: report independent instance and workspace identities * test: include workspace in OAuth callback fixture * ci: pin production cloud adapter to Space release * fix: preserve authenticated Workspace telemetry attribution * ci: pin production cloud adapter to final Space release --------- Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>	2026-08-04 17:27:23 +08:00
Hyu	7820949d3a	fix(workspace): show member emails (#2393 ) Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>	2026-08-04 11:48:43 +08:00
Hyu	e263a5d1d7	fix(web): use natural tooltip wrapping (#2391 ) Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>	2026-08-03 21:26:55 +08:00
Hyu	6bad7bcffc	fix(web): keep extension market navigable at quota (#2390 ) Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>	2026-08-03 20:10:46 +08:00
Hyu	f0b2c103c1	fix(web): disable quota-reached create actions (#2389 ) * fix(web): disable quota-reached create actions * fix(web): close quota review gaps --------- Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>	2026-08-03 19:08:47 +08:00
Hyu	3101c9be6a	[verified] fix: harden OSS and Cloud workspace UI (#2387 ) Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>	2026-08-03 13:27:38 +08:00
Hyu	1e6e4c0ca7	fix(cloud): show owner model balance and enforce single owner (#2384 ) (#2385 ) * fix(cloud): show owner model balance and enforce single owner * fix(migrations): create owner index idempotently --------- Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>	2026-08-03 02:14:43 +08:00
Hyu	408c8031d4	Merge pull request #2383 from langbot-app/sync/cloud-fixes-to-master merge: sync Cloud production fixes to master	2026-08-02 17:24:48 +08:00
dadachann	0e6cca4690	merge: sync Cloud production fixes to master	2026-08-02 09:19:41 +00:00
dadachann	a7a7218afe	fix(cloud): accept invitations with current account	2026-08-02 09:08:49 +00:00
Hyu	a67728c163	fix cloud monitoring and invitation sign-in (#2381 ) Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>	2026-08-02 16:39:14 +08:00
Hyu	e9c9e896c6	fix(cloud): preserve pipeline routing in debug chat (#2380 ) Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>	2026-08-02 01:49:17 +08:00
Hyu	e2331c4967	fix(cloud): restore plugins and pipeline execution (#2379 ) Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>	2026-08-02 01:34:11 +08:00
Hyu	0ccbcd5f5f	fix(migrations): preserve published Cloud revision head (#2375 ) Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>	2026-08-02 00:56:20 +08:00
Hyu	c5aada494d	fix(cloud): treat workspace owners as Space-bound (#2378 ) Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>	2026-08-02 00:33:46 +08:00
Hyu	e36e3aaea8	fix(cloud): accept null model abilities (#2377 ) Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>	2026-08-01 18:50:47 +08:00
Hyu	d64278ab3f	feat(cloud): provision workspace model catalog (#2376 ) * feat(cloud): provision workspace model catalog * ci(cloud): pin model catalog adapter source * fix: make cloud model catalog sync recoverable * ci: pin cloud adapter source for release --------- Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>	2026-08-01 18:16:02 +08:00
dadachann	161ea9b3eb	fix(cloud): restore fragment-based Space launch callback	2026-08-01 04:41:44 +00:00
dadachann	c7d14676fc	fix(cloud): restore disabled Box production mode	2026-08-01 04:11:17 +00:00
dadachann	2456bf1350	fix(migrations): preserve published Cloud revision head	2026-07-31 18:35:21 +00:00
dadachann	05a941ff16	Merge remote-tracking branch 'origin/deploy/prod' into release/cloud-monitoring-prod	2026-07-31 18:18:46 +00:00
dadachann	0330788d14	chore(prod): release workspace monitoring	2026-07-31 18:18:46 +00:00
Hyu	d8ab0ba567	feat(telemetry): report workspace execution generation (#2374 ) Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>	2026-08-01 02:10:59 +08:00
Hyu	e3832ca536	style: format workspace identity modules (#2372 ) Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>	2026-07-31 23:46:03 +08:00
Hyu	5d9fd15671	feat: use workspace identity for telemetry (#2371 ) Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>	2026-07-31 23:35:14 +08:00
Hyu	404e3466d9	feat(cloud): add scoped support admin sessions (#2369 ) * feat(cloud): add scoped support admin sessions * style(web): format support admin session changes * fix(cloud): isolate support adapter sessions * fix(cloud): authenticate plugin assets and report workspace resources --------- Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>	2026-07-31 17:41:55 +08:00
Constantine	9df021eb8f	fix(telegram): stop leaking bot token via Image.url (#2366 ) Telegram file.file_path is a full URL of the form https://api.telegram.org/file/bot<TOKEN>/<path> that embeds the bot token. Since #2362 this URL was copied into Image.url, so the token was serialized into the message chain and thereby persisted to the monitoring database, shown in the dashboard, and forwarded to every installed plugin via event dispatch. Anyone with dashboard or plugin access could recover the token and take full control of the bot. Unlike the public CDN URLs used by the other adapters changed in #2362, Telegram file URLs are only usable with the embedded token, so there is no safe URL to expose. Store base64 only (as before #2362); the vision path already relies solely on base64, so nothing downstream changes. Add a regression test asserting the token never appears in the converted Image or the serialized message chain. Co-authored-by: Constantine1916 <Constantine1916@users.noreply.github.com>	2026-07-31 17:25:19 +08:00
Hyu	98d0dba6d4	fix(web): restore locale key parity (#2370 ) Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>	2026-07-31 17:07:04 +08:00
Hyu	5ec2371879	fix(config): preserve typed list environment overrides (#2367 ) Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>	2026-07-31 16:06:09 +08:00
dadachann	473ba573a3	fix(cloud): retain launch replay records through clock skew	2026-07-30 21:37:06 +00:00
dadachann	93dbd3541e	fix(cloud): make direct launch replay-safe	2026-07-30 21:02:38 +00:00
dadachann	a5a26f81ee	feat(auth): accept direct Space launch assertions	2026-07-30 20:05:17 +00:00
dadachann	92d9db8f95	chore(plugin): pin SDK 0.5.0	2026-07-30 20:05:17 +00:00
dadachann	59db012594	feat(cloud): enforce workspace resource quotas	2026-07-30 18:48:42 +00:00
dadachann	88f328066b	fix(cloud): keep runtime sdk ahead of plugin dependencies	2026-07-30 16:42:01 +00:00
dadachann	d155d9d5a8	fix(cloud): allow explicitly disabled box runtime	2026-07-30 15:42:49 +00:00
dadachann	dd95545309	fix(prod): configure shared Box runtime	2026-07-30 15:17:17 +00:00
dadachann	9066c25729	fix(deploy): let migration own exact runtime ACLs	2026-07-30 14:58:31 +00:00
dadachann	6d2e9d3d72	fix(deploy): do not start disabled Box runtime	2026-07-30 14:48:51 +00:00
dadachann	a0b85e11fd	fix(deploy): keep runtime role schema read-only	2026-07-30 14:44:57 +00:00
dadachann	122d8fa659	fix(deploy): retry transient image pull failures	2026-07-30 14:34:20 +00:00
dadachann	ace8cc67f2	fix(config): preserve typed list environment overrides	2026-07-30 14:26:41 +00:00
dadachann	52c0772806	fix(cloud): pin Space production URL and deployment health	2026-07-30 14:21:04 +00:00
dadachann	d5044c2f1e	fix(ci): authenticate cloud adapter checkout	2026-07-30 14:08:58 +00:00
dadachann	7baa89254c	ops(cloud): deploy exact production stack to jp09	2026-07-30 13:58:37 +00:00
RockChinQ	e1ac5e0fc8	feat(tenancy): add Workspace multi-tenant foundation (#2353 ) * Document multi-tenant workspace architecture * Add OSS and commercial workspace boundaries * docs: redesign multi-tenant workspace architecture * feat(tenancy): implement workspace isolation * docs(tenancy): record verification evidence * docs(tenancy): revise single-instance SaaS topology * docs(tenancy): refine architecture options * docs: finalize cloud v2 multi-tenant decisions * feat(tenancy): establish cloud isolation foundations * feat(tenancy): harden shared cloud runtime boundaries * docs(tenancy): record final isolation verification * fix(tenancy): close isolation and permission gaps * docs(tenancy): record final isolation verification * feat(tenancy): connect cloud workspace control plane * fix(build): install git for pinned SDK * docs(cloud): update control plane verification * chore: update multi-tenant SDK pin * fix(cloud): skip legacy model sync during startup * test(cloud): preserve minimal model manager fixtures * fix(cloud): preserve authenticated account context * fix(cloud): reuse authenticated account for user info * feat(cloud): complete Workspace settings navigation * test(web): cover Workspace dropdown menu * feat(web): place workspace controls in sidebar * refactor(web): streamline workspace controls * style(web): format workspace layout test * fix(cloud): surface runtime and workspace plan status * fix(plugin): keep runtime identity stable across restarts * fix(ui): widen and center workspace switcher * fix(ui): hide roles from workspace switcher * fix(ui): align workspace switcher with sidebar entries * feat(workspace): add in-product collaboration and direct Cloud launch * style: format collaboration changes * fix(workspace): bind collaboration APIs to tenant UoW * fix(cloud): preserve Core-owned collaboration state * test(cloud): require Space identity for invite registration * feat(cloud): complete secure invitation experience * style(web): format invitation flows * fix(cloud): recover box runtime without unscoped skill reload * feat(oss): enforce invitation account and owner billing flows * style: format OSS account service * test(oss): cover invitation logout handoff * fix(oss): resolve workspace owner in scoped session * feat(cloud): harden multi-tenant runtime resources * fix(cloud): bound runtime restart storms * fix(cloud): eliminate periodic runtime CPU spikes * fix(cloud): enforce instance capacity ceilings * fix(cloud): scope public login capability discovery * fix(cloud): bound tenant maintenance and monitoring work * fix(runtime): bound tenant resource amplification * fix(deps): pin green multi-tenant plugin SDK * fix(cloud): handle unavailable skill capability * fix(security): require authentication for image file endpoint (H-2) - Changed /api/v1/files/image from AuthType.NONE to USER_TOKEN_OR_API_KEY - Added Permission.RESOURCE_VIEW requirement - Prevents unauthenticated cross-tenant file access via leaked keys - Fixes HIGH severity finding from multi-tenant security review docs: add comprehensive database migration guide - Complete migration steps for OSS → multi-tenant - Backup, execution, verification procedures - Rollback scenarios and recovery plans - Performance tuning recommendations * test: add comprehensive cross-tenant isolation tests Added 7 critical test scenarios for multi-tenant boundaries: - Cross-tenant bot access prevention - Viewer role read-only enforcement - Removed member immediate access revocation - Model provider credential isolation - WebSocket message isolation - Invitation token workspace scoping - Multi-workspace context validation These tests address P0-2 coverage gaps for: - workspaces.py (membership & invitation flows) - user.py (authentication & authorization) - websocket_chat.py (real-time isolation) - plugins.py (resource access control) docs: finalize database migration guide * fix(security): resolve M-1, M-2, M-3 security findings M-1: WebSocket authorization TOCTOU race (FIXED) - Changed _revalidate_websocket_authorization to return RequestContext - Ensures validated context is used immediately without race window - Prevents removed members from sending messages during revalidation gap M-2: Model Manager cache workspace isolation (VERIFIED) - Confirmed _CacheKey already uses 4-tuple: (instance, workspace, generation, resource) - Cache is properly scoped per workspace, no cross-tenant leakage possible - No code change needed, documented as working correctly M-3: Invitation lock workspace scoping (FIXED) - Changed lock key from token_digest to workspace_uuid:token_digest - Prevents DoS where attacker locks token in Workspace A to block Workspace B - Locks now isolated per workspace All MEDIUM severity findings from security review now resolved. * fix(cloud): unblock tenant CI and enforce knowledge quotas * fix(tenancy): scope rerank model sync --------- Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>	2026-07-30 21:43:35 +08:00
douxt	463b120923	feat(platform): pass original image URL to Image component in 6 adapters (#2362 ) Preserve the platform CDN URL in Image.url alongside base64 data, enabling plugins to use ContentElement.from_image_url() for direct vision API access without redundant local download. - aiocqhttp: use msg_data["data"]["url"] and msg.data["url"] - discord: use attachment.url - telegram: use file.file_path - slack: use pic_url - wecom: use picurl - qqofficial: use pic_url Satori adapter already follows this pattern (satori.py:168). The change is purely additive — base64 is preserved for backward compatibility, and get_bytes() priority (url → base64 → path) ensures plugins can choose the optimal path. Closes #2355 Co-authored-by: douxt <8429023+douxt@users.noreply.github.com> Co-authored-by: Claude <noreply@anthropic.com>	2026-07-28 23:57:00 +08:00
Hyu	3ca724d18e	fix: use per-bot admins for command events (#2359 ) * fix: use per-bot admins for command events * style: format rerank provider changes --------- Co-authored-by: dadachann <185672915+dadachann@users.noreply.github.com>	2026-07-26 17:02:51 +08:00
Hyu	dd8d1007a1	feat: support Space rerank models (#2358 ) Co-authored-by: chan <dadachann@users.noreply.github.com>	2026-07-26 16:37:34 +08:00