Merge remote-tracking branch 'origin/dev/4.11.x' into dev/4.11.x

# Conflicts:
#	uv.lock
#	web/src/i18n/locales/en-US.ts
#	web/src/i18n/locales/ja-JP.ts
#	web/src/i18n/locales/zh-Hans.ts
#	web/src/i18n/locales/zh-Hant.ts

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

@@ -15,6 +15,7 @@ from ....workspace.collaboration import MembershipPermissionError, WorkspaceColl
 from ....workspace.errors import WorkspaceNotFoundError
 from ....cloud.entitlements import EntitlementUnavailableError
 from ....core.errors import TaskCapacityError
+from ....provider.modelmgr.codex_errors import CodexProviderError
 from ..authz import (
     AuthenticationDeniedError,
     AuthorizationError,
@@ -249,6 +250,8 @@ class RouterGroup(abc.ABC):
                     return await f(*args, **kwargs)
 
                 except Exception as e:  # 自动 500
+                    if isinstance(e, CodexProviderError):
+                        return self.http_status(e.status_code, e.error_code, str(e))
                     if isinstance(e, AuthorizationError):
                         return self.http_status(e.status_code, e.error_code, str(e))
                     if isinstance(e, WorkspaceNotFoundError):

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

@@ -47,6 +47,21 @@ class FilesRouterGroup(group.RouterGroup):
 
             return quart.Response(image_bytes, mimetype=mime_type)
 
+        @self.route(
+            '/media/<filename>',
+            methods=['GET'],
+            auth_type=group.AuthType.NONE,
+        )
+        async def get_media_file(filename: str) -> quart.Response:
+            media = await self.ap.storage_mgr.media_cache.get_media(filename)
+            if media is None:
+                return quart.Response('Media not found or expired', status=404)
+            media_bytes, mime_type = media
+            headers = {
+                'Cache-Control': 'public, max-age=2592000, immutable',
+            }
+            return quart.Response(media_bytes, mimetype=mime_type, headers=headers)
+
         @self.route(
             '/images',
             methods=['POST'],

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

@@ -5,6 +5,7 @@ import quart
 
 from ...authz import Permission
 from ...context import RequestContext
+from ...service.monitoring_traffic import get_traffic_series
 from .. import group
 
 
@@ -218,6 +219,7 @@ class MonitoringRouterGroup(group.RouterGroup):
             pipeline_ids = quart.request.args.getlist('pipelineId')
             start_time_str = quart.request.args.get('startTime')
             end_time_str = quart.request.args.get('endTime')
+            user_query = quart.request.args.get('userQuery')
             is_active_str = quart.request.args.get('isActive')
             limit = int(quart.request.args.get('limit', 100))
             offset = int(quart.request.args.get('offset', 0))
@@ -237,6 +239,7 @@ class MonitoringRouterGroup(group.RouterGroup):
                 pipeline_ids=pipeline_ids if pipeline_ids else None,
                 start_time=start_time,
                 end_time=end_time,
+                user_query=user_query,
                 is_active=is_active,
                 limit=limit,
                 offset=offset,
@@ -375,6 +378,14 @@ class MonitoringRouterGroup(group.RouterGroup):
 
             return self.success(
                 data={
+                    'traffic': await get_traffic_series(
+                        self.ap,
+                        request_context,
+                        bot_ids=bot_ids or None,
+                        pipeline_ids=pipeline_ids or None,
+                        start_time=start_time,
+                        end_time=end_time,
+                    ),
                     'overview': overview,
                     'messages': messages,
                     'llmCalls': llm_calls,
@@ -396,7 +407,15 @@ class MonitoringRouterGroup(group.RouterGroup):
         @self.route('/sessions/<session_id>/analysis', methods=['GET'], permission=Permission.RESOURCE_VIEW)
         async def get_session_analysis(session_id: str, request_context: RequestContext) -> str:
             """Get detailed analysis for a specific session"""
-            analysis = await self.ap.monitoring_service.get_session_analysis(request_context, session_id)
+            start_time = parse_iso_datetime(quart.request.args.get('startTime'))
+            end_time = parse_iso_datetime(quart.request.args.get('endTime'))
+            analysis = await self.ap.monitoring_service.get_session_analysis(
+                request_context,
+                session_id,
+                start_time=start_time,
+                end_time=end_time,
+                bot_id=quart.request.args.get('botId'),
+            )
 
             # Always return success with the analysis data
             # The frontend will handle the 'found: false' case

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

@@ -199,7 +199,9 @@ class EmbedRouterGroup(group.RouterGroup):
 
             base_url = quart.request.host_url.rstrip('/')
             webhook_prefix = self.ap.instance_config.data.get('api', {}).get('webhook_prefix', '')
-            if webhook_prefix:
+            # The wizard previews the currently connected backend, which can
+            # differ from the public address used by external website embeds.
+            if webhook_prefix and quart.request.args.get('preview') != 'wizard':
                 base_url = webhook_prefix.rstrip('/')
 
             if not re.match(r'^https?://[a-zA-Z0-9._:/-]+$', base_url):

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

@@ -0,0 +1,33 @@
+"""User-confirmed migration only; deliberately not API-key/MCP accessible."""
+
+import quart
+
+from ... import group
+from ....authz import Permission
+from ....context import RequestContext
+from ....service.pipeline_migration import PipelineMigrationService, MigrationError
+
+
+@group.group_class('pipeline_migration', '/api/v1/pipelines/_/migration')
+class PipelineMigrationRouterGroup(group.RouterGroup):
+    async def initialize(self):
+        self.service = PipelineMigrationService(self.ap)
+
+        @self.route(
+            '/preview', methods=['GET'], auth_type=group.AuthType.USER_TOKEN, permission=Permission.RESOURCE_VIEW
+        )
+        async def preview(request_context: RequestContext):
+            try:
+                return self.success(data=await self.service.preview(request_context))
+            except MigrationError as exc:
+                return self.http_status(exc.status_code, -1, exc.code)
+
+        @self.route(
+            '/execute', methods=['POST'], auth_type=group.AuthType.USER_TOKEN, permission=Permission.RESOURCE_MANAGE
+        )
+        async def execute(request_context: RequestContext):
+            try:
+                body = await quart.request.get_json(silent=True)
+                return self.success(data=await self.service.execute(request_context, body))
+            except MigrationError as exc:
+                return self.http_status(exc.status_code, -1, exc.code)

src/langbot/pkg/api/http/controller/groups/provider/models.py

@@ -4,6 +4,7 @@ from langbot.pkg.provider.modelmgr import errors as provider_errors
 from ....authz import Permission, has_permission
 from ....context import RequestContext
 from ... import group
+from .query import resolve_include_secret
 
 
 MODEL_TEST_ERROR_CODE = 'model_test_failed'
@@ -20,7 +21,12 @@ class LLMModelsRouterGroup(group.RouterGroup):
         )
         async def _(request_context: RequestContext) -> str:
             provider_uuid = quart.request.args.get('provider_uuid')
-            include_secret = has_permission(request_context, Permission.PROVIDER_SECRET_MANAGE)
+            include_secret, error = resolve_include_secret(
+                quart.request.args.get('include_secret'),
+                permitted=has_permission(request_context, Permission.PROVIDER_SECRET_MANAGE),
+            )
+            if error:
+                return self.http_status(400, -1, error)
             if provider_uuid:
                 models = await self.ap.llm_model_service.get_llm_models_by_provider(
                     request_context,
@@ -57,10 +63,16 @@ class LLMModelsRouterGroup(group.RouterGroup):
             permission=Permission.RESOURCE_VIEW,
         )
         async def _(model_uuid: str, request_context: RequestContext) -> str:
+            include_secret, error = resolve_include_secret(
+                quart.request.args.get('include_secret'),
+                permitted=has_permission(request_context, Permission.PROVIDER_SECRET_MANAGE),
+            )
+            if error:
+                return self.http_status(400, -1, error)
             model = await self.ap.llm_model_service.get_llm_model(
                 request_context,
                 model_uuid,
-                include_secret=has_permission(request_context, Permission.PROVIDER_SECRET_MANAGE),
+                include_secret=include_secret,
             )
             if model is None:
                 return self.http_status(404, -1, 'model not found')
@@ -106,7 +118,7 @@ class LLMModelsRouterGroup(group.RouterGroup):
                     model_uuid,
                     await quart.request.json,
                 )
-            except (provider_errors.RequesterError, ValueError) as exc:
+            except provider_errors.RequesterError as exc:
                 return self.http_status(400, MODEL_TEST_ERROR_CODE, str(exc))
             return self.success()
 
@@ -122,7 +134,12 @@ class EmbeddingModelsRouterGroup(group.RouterGroup):
         )
         async def _(request_context: RequestContext) -> str:
             provider_uuid = quart.request.args.get('provider_uuid')
-            include_secret = has_permission(request_context, Permission.PROVIDER_SECRET_MANAGE)
+            include_secret, error = resolve_include_secret(
+                quart.request.args.get('include_secret'),
+                permitted=has_permission(request_context, Permission.PROVIDER_SECRET_MANAGE),
+            )
+            if error:
+                return self.http_status(400, -1, error)
             if provider_uuid:
                 models = await self.ap.embedding_models_service.get_embedding_models_by_provider(
                     request_context,
@@ -159,10 +176,16 @@ class EmbeddingModelsRouterGroup(group.RouterGroup):
             permission=Permission.RESOURCE_VIEW,
         )
         async def _(model_uuid: str, request_context: RequestContext) -> str:
+            include_secret, error = resolve_include_secret(
+                quart.request.args.get('include_secret'),
+                permitted=has_permission(request_context, Permission.PROVIDER_SECRET_MANAGE),
+            )
+            if error:
+                return self.http_status(400, -1, error)
             model = await self.ap.embedding_models_service.get_embedding_model(
                 request_context,
                 model_uuid,
-                include_secret=has_permission(request_context, Permission.PROVIDER_SECRET_MANAGE),
+                include_secret=include_secret,
             )
             if model is None:
                 return self.http_status(404, -1, 'model not found')
@@ -208,7 +231,7 @@ class EmbeddingModelsRouterGroup(group.RouterGroup):
                     model_uuid,
                     await quart.request.json,
                 )
-            except (provider_errors.RequesterError, ValueError) as exc:
+            except provider_errors.RequesterError as exc:
                 return self.http_status(400, MODEL_TEST_ERROR_CODE, str(exc))
             return self.success()
 
@@ -224,7 +247,12 @@ class RerankModelsRouterGroup(group.RouterGroup):
         )
         async def _(request_context: RequestContext) -> str:
             provider_uuid = quart.request.args.get('provider_uuid')
-            include_secret = has_permission(request_context, Permission.PROVIDER_SECRET_MANAGE)
+            include_secret, error = resolve_include_secret(
+                quart.request.args.get('include_secret'),
+                permitted=has_permission(request_context, Permission.PROVIDER_SECRET_MANAGE),
+            )
+            if error:
+                return self.http_status(400, -1, error)
             if provider_uuid:
                 models = await self.ap.rerank_models_service.get_rerank_models_by_provider(
                     request_context,
@@ -261,10 +289,16 @@ class RerankModelsRouterGroup(group.RouterGroup):
             permission=Permission.RESOURCE_VIEW,
         )
         async def _(model_uuid: str, request_context: RequestContext) -> str:
+            include_secret, error = resolve_include_secret(
+                quart.request.args.get('include_secret'),
+                permitted=has_permission(request_context, Permission.PROVIDER_SECRET_MANAGE),
+            )
+            if error:
+                return self.http_status(400, -1, error)
             model = await self.ap.rerank_models_service.get_rerank_model(
                 request_context,
                 model_uuid,
-                include_secret=has_permission(request_context, Permission.PROVIDER_SECRET_MANAGE),
+                include_secret=include_secret,
             )
             if model is None:
                 return self.http_status(404, -1, 'model not found')
@@ -310,6 +344,6 @@ class RerankModelsRouterGroup(group.RouterGroup):
                     model_uuid,
                     await quart.request.json,
                 )
-            except (provider_errors.RequesterError, ValueError) as exc:
+            except provider_errors.RequesterError as exc:
                 return self.http_status(400, MODEL_TEST_ERROR_CODE, str(exc))
             return self.success()

src/langbot/pkg/api/http/controller/groups/provider/providers.py

@@ -3,11 +3,86 @@ import quart
 from ....authz import Permission, has_permission
 from ....context import RequestContext
 from ... import group
+from .query import resolve_include_secret
 
 
 @group.group_class('models/providers', '/api/v1/provider/providers')
 class ModelProvidersRouterGroup(group.RouterGroup):
     async def initialize(self) -> None:
+        # Subscription authorization is an interactive, browser-user-only surface.
+        @self.route(
+            '/<provider_uuid>/codex/status',
+            methods=['GET'],
+            auth_type=group.AuthType.USER_TOKEN,
+            permission=Permission.PROVIDER_SECRET_MANAGE,
+        )
+        async def codex_status(provider_uuid: str, request_context: RequestContext):
+            try:
+                return self.success(
+                    data=await self.ap.provider_service.codex_auth.status(request_context, provider_uuid)
+                )
+            except ValueError as exc:
+                return self.http_status(400, -1, str(exc))
+
+        @self.route(
+            '/<provider_uuid>/codex/device',
+            methods=['POST'],
+            auth_type=group.AuthType.USER_TOKEN,
+            permission=Permission.PROVIDER_SECRET_MANAGE,
+        )
+        async def codex_device(provider_uuid: str, request_context: RequestContext):
+            try:
+                return self.success(
+                    data=await self.ap.provider_service.codex_auth.start(request_context, provider_uuid)
+                )
+            except ValueError as exc:
+                return self.http_status(400, -1, str(exc))
+
+        @self.route(
+            '/<provider_uuid>/codex/device/poll',
+            methods=['POST'],
+            auth_type=group.AuthType.USER_TOKEN,
+            permission=Permission.PROVIDER_SECRET_MANAGE,
+        )
+        async def codex_poll(provider_uuid: str, request_context: RequestContext):
+            body = await quart.request.get_json()
+            if not isinstance(body, dict):
+                return self.http_status(400, -1, 'JSON object required')
+            try:
+                return self.success(
+                    data=await self.ap.provider_service.codex_auth.poll(
+                        request_context, provider_uuid, body.get('authorization_id')
+                    )
+                )
+            except ValueError as exc:
+                return self.http_status(400, -1, str(exc))
+
+        @self.route(
+            '/<provider_uuid>/codex/auth',
+            methods=['DELETE'],
+            auth_type=group.AuthType.USER_TOKEN,
+            permission=Permission.PROVIDER_SECRET_MANAGE,
+        )
+        async def codex_disconnect(provider_uuid: str, request_context: RequestContext):
+            try:
+                await self.ap.provider_service.codex_auth.disconnect(request_context, provider_uuid)
+                return self.success()
+            except ValueError as exc:
+                return self.http_status(400, -1, str(exc))
+
+        @self.route(
+            '/<provider_uuid>/codex/device/<authorization_id>',
+            methods=['DELETE'],
+            auth_type=group.AuthType.USER_TOKEN,
+            permission=Permission.PROVIDER_SECRET_MANAGE,
+        )
+        async def codex_cancel(provider_uuid: str, authorization_id: str, request_context: RequestContext):
+            try:
+                await self.ap.provider_service.codex_auth.cancel(request_context, provider_uuid, authorization_id)
+                return self.success()
+            except ValueError as exc:
+                return self.http_status(400, -1, str(exc))
+
         @self.route(
             '',
             methods=['GET'],
@@ -15,9 +90,15 @@ class ModelProvidersRouterGroup(group.RouterGroup):
             permission=Permission.RESOURCE_VIEW,
         )
         async def _(request_context: RequestContext) -> str:
+            include_secret, error = resolve_include_secret(
+                quart.request.args.get('include_secret'),
+                permitted=has_permission(request_context, Permission.PROVIDER_SECRET_MANAGE),
+            )
+            if error:
+                return self.http_status(400, -1, error)
             providers = await self.ap.provider_service.get_providers(
                 request_context,
-                include_secret=has_permission(request_context, Permission.PROVIDER_SECRET_MANAGE),
+                include_secret=include_secret,
             )
             for provider in providers:
                 counts = await self.ap.provider_service.get_provider_model_counts(request_context, provider['uuid'])
@@ -47,10 +128,16 @@ class ModelProvidersRouterGroup(group.RouterGroup):
             permission=Permission.RESOURCE_VIEW,
         )
         async def _(provider_uuid: str, request_context: RequestContext) -> str:
+            include_secret, error = resolve_include_secret(
+                quart.request.args.get('include_secret'),
+                permitted=has_permission(request_context, Permission.PROVIDER_SECRET_MANAGE),
+            )
+            if error:
+                return self.http_status(400, -1, error)
             provider = await self.ap.provider_service.get_provider(
                 request_context,
                 provider_uuid,
-                include_secret=has_permission(request_context, Permission.PROVIDER_SECRET_MANAGE),
+                include_secret=include_secret,
             )
             if provider is None:
                 return self.http_status(404, -1, 'provider not found')
@@ -82,7 +169,15 @@ class ModelProvidersRouterGroup(group.RouterGroup):
         )
         async def _(provider_uuid: str, request_context: RequestContext) -> str:
             try:
-                await self.ap.provider_service.delete_provider(request_context, provider_uuid)
+                cascade_values = quart.request.args.getlist('cascade')
+                if cascade_values:
+                    if len(cascade_values) != 1 or cascade_values[0] not in ('true', 'false'):
+                        return self.http_status(400, -1, 'cascade must be a single true or false value')
+                    await self.ap.provider_service.delete_provider(
+                        request_context, provider_uuid, cascade=cascade_values[0] == 'true'
+                    )
+                else:
+                    await self.ap.provider_service.delete_provider(request_context, provider_uuid)
                 return self.success()
             except ValueError as e:
                 return self.http_status(400, -1, str(e))

src/langbot/pkg/api/http/controller/groups/provider/query.py

@@ -0,0 +1,15 @@
+from __future__ import annotations
+
+
+def resolve_include_secret(raw_value: str | None, *, permitted: bool) -> tuple[bool, str | None]:
+    """Resolve the optional secret projection query parameter."""
+
+    if raw_value is None:
+        return permitted, None
+
+    value = raw_value.strip().lower()
+    if value == 'false':
+        return False, None
+    if value == 'true':
+        return permitted, None
+    return False, 'include_secret must be either true or false'

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

@@ -7,14 +7,116 @@ from .. import group
 from .....utils import constants
 from .....entity.persistence.metadata import WorkspaceMetadata
 from ...authz import Permission
-from ...context import RequestContext
+from ...context import PrincipalType, RequestContext
 from .....provider.tools.loaders.mcp_policy import stdio_mcp_enabled
 from .....workspace.invitation_delivery import InvitationDeliveryService
 
 
+SYSTEM_CAPABILITY_OPERATIONS = (
+    'bot.list',
+    'bot.get',
+    'bot.create',
+    'bot.update',
+    'bot.delete',
+    'pipeline.list',
+    'pipeline.get',
+    'pipeline.create',
+    'pipeline.update',
+    'pipeline.delete',
+    'pipeline.copy',
+    'task.list',
+    'task.get',
+    'knowledge_base.list',
+    'knowledge_base.get',
+    'knowledge_base.create',
+    'knowledge_base.update',
+    'knowledge_base.delete',
+    'knowledge_base.file.list',
+    'knowledge_base.file.store',
+    'knowledge_base.file.delete',
+    'knowledge_base.retrieve',
+    'file.document.upload',
+    'plugin.install.github',
+    'plugin.install.marketplace',
+    'plugin.install.local',
+    'plugin.upgrade',
+    'plugin.get',
+    'plugin.list',
+    'plugin.config.get',
+    'plugin.config.update',
+    'plugin.logs',
+    'plugin.delete',
+    'provider.list',
+    'provider.get',
+    'provider.create',
+    'provider.update',
+    'provider.delete',
+    'provider.scan_models',
+    'model.llm.list',
+    'model.llm.get',
+    'model.llm.create',
+    'model.llm.update',
+    'model.llm.delete',
+    'model.llm.test',
+    'model.embedding.list',
+    'model.embedding.get',
+    'model.embedding.create',
+    'model.embedding.update',
+    'model.embedding.delete',
+    'model.embedding.test',
+    'model.rerank.list',
+    'model.rerank.get',
+    'model.rerank.create',
+    'model.rerank.update',
+    'model.rerank.delete',
+    'model.rerank.test',
+    'skill.list',
+    'skill.get',
+    'skill.create',
+    'skill.update',
+    'skill.delete',
+    'skill.files.list',
+    'skill.files.read',
+    'skill.files.write',
+    'skill.preview',
+    'skill.install.github',
+    'skill.install.upload',
+    'mcp_server.list',
+    'mcp_server.get',
+    'mcp_server.create',
+    'mcp_server.update',
+    'mcp_server.delete',
+    'mcp_server.resources',
+    'mcp_server.resource_templates',
+    'mcp_server.resource_read',
+    'mcp_server.logs',
+    'mcp_server.test',
+)
+
+
 @group.group_class('system', '/api/v1/system')
 class SystemRouterGroup(group.RouterGroup):
     async def initialize(self) -> None:
+        @self.route('/context', methods=['GET'], auth_type=group.AuthType.API_KEY)
+        async def _(request_context: RequestContext) -> str:
+            return self.success(
+                data={
+                    'instance_uuid': request_context.instance_uuid,
+                    'workspace_uuid': request_context.workspace_uuid,
+                    'api_key_id': request_context.principal.api_key_uuid,
+                    'permissions': sorted(request_context.workspace.permissions),
+                }
+            )
+
+        @self.route('/capabilities', methods=['GET'], auth_type=group.AuthType.API_KEY)
+        async def _() -> str:
+            return self.success(
+                data={
+                    'schema_version': 1,
+                    'operations': {operation: {'supported': True} for operation in SYSTEM_CAPABILITY_OPERATIONS},
+                }
+            )
+
         @self.route('/info', methods=['GET'], auth_type=group.AuthType.NONE)
         async def _() -> str:
             # Read wizard_status and wizard_progress from metadata table
@@ -238,7 +340,7 @@ class SystemRouterGroup(group.RouterGroup):
         @self.route(
             '/tasks',
             methods=['GET'],
-            auth_type=group.AuthType.USER_TOKEN,
+            auth_type=group.AuthType.USER_TOKEN_OR_API_KEY,
             permission=Permission.RESOURCE_VIEW,
         )
         async def _(request_context: RequestContext) -> str:
@@ -257,18 +359,23 @@ class SystemRouterGroup(group.RouterGroup):
                     instance_uuid=request_context.instance_uuid,
                     workspace_uuid=request_context.workspace_uuid,
                     placement_generation=request_context.placement_generation,
+                    public=request_context.principal.principal_type == PrincipalType.API_KEY,
                 )
             )
 
         @self.route(
             '/tasks/<task_id>',
             methods=['GET'],
-            auth_type=group.AuthType.USER_TOKEN,
+            auth_type=group.AuthType.USER_TOKEN_OR_API_KEY,
             permission=Permission.RESOURCE_VIEW,
         )
         async def _(task_id: str, request_context: RequestContext) -> str:
+            try:
+                task_index = int(task_id)
+            except (TypeError, ValueError):
+                return self.http_status(404, 404, 'Task not found')
             task = self.ap.task_mgr.get_task_by_id(
-                int(task_id),
+                task_index,
                 instance_uuid=request_context.instance_uuid,
                 workspace_uuid=request_context.workspace_uuid,
                 placement_generation=request_context.placement_generation,
@@ -277,6 +384,8 @@ class SystemRouterGroup(group.RouterGroup):
             if task is None:
                 return self.http_status(404, 404, 'Task not found')
 
+            if request_context.principal.principal_type == PrincipalType.API_KEY:
+                return self.success(data=task.to_public_dict())
             return self.success(data=task.to_dict())
 
         @self.route(

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

@@ -1,7 +1,12 @@
+from __future__ import annotations
+
 import quart
 import argon2
 import asyncio
 import datetime
+import hmac
+import time
+import typing
 import uuid
 from urllib.parse import parse_qs, urlsplit
 
@@ -11,6 +16,33 @@ from ...context import RequestContext
 from .....cloud.launch import SpaceLaunchError
 from ...service.user import ControlPlaneDirectoryRequiredError, PublicRegistrationClosedError
 
+# Fixed-window admission quota for the unauthenticated reset-password endpoint (#2392).
+# The admission check and slot bump share ONE synchronous critical section with no await
+# points, so concurrent bursts within a single event loop cannot slip past accounting.
+# Every admitted attempt consumes quota (regardless of success), which throttles both the
+# legacy 24-bit keyspace exhaustion and brute-force on modern high-entropy keys.
+# NOTE: this state is process-local; multi-worker deployments need a shared limiter upstream.
+_MAX_RESET_ATTEMPTS_PER_WINDOW = 5
+_RESET_WINDOW_SECONDS = 15 * 60
+
+_reset_password_state: dict = {'window_started_at': 0.0, 'attempts': 0}
+
+
+def _admit_reset_attempt(now: float) -> bool:
+    """Atomically reserve one reset-password admission slot.
+
+    Must stay await-free: running to completion without suspension makes the
+    check-and-increment atomic under the single-threaded event loop.
+    """
+    st = _reset_password_state
+    if now - st['window_started_at'] >= _RESET_WINDOW_SECONDS:
+        st['window_started_at'] = now
+        st['attempts'] = 0
+    if st['attempts'] >= _MAX_RESET_ATTEMPTS_PER_WINDOW:
+        return False
+    st['attempts'] += 1
+    return True
+
 
 @group.group_class('user', '/api/v1/user')
 class UserRouterGroup(group.RouterGroup):
@@ -35,6 +67,22 @@ class UserRouterGroup(group.RouterGroup):
 
         return redirect_uri
 
+    def _extract_origin_and_rp_id(self, json_data: dict[str, typing.Any] | None = None) -> tuple[str, str]:
+        origin = ''
+        if json_data and isinstance(json_data, dict):
+            origin = json_data.get('origin', '')
+        if not origin:
+            origin = quart.request.headers.get('Origin', '')
+        if not origin:
+            origin = quart.request.headers.get('Referer', '')
+        if not origin:
+            origin = quart.request.url_root.rstrip('/')
+
+        parsed = urlsplit(origin)
+        rp_id = parsed.hostname or 'localhost'
+        clean_origin = f'{parsed.scheme}://{parsed.netloc}' if parsed.scheme and parsed.netloc else origin.rstrip('/')
+        return clean_origin, rp_id
+
     async def initialize(self) -> None:
         @self.route('/init', methods=['GET', 'POST'], auth_type=group.AuthType.NONE)
         async def _() -> str:
@@ -81,6 +129,12 @@ class UserRouterGroup(group.RouterGroup):
 
         @self.route('/reset-password', methods=['POST'], auth_type=group.AuthType.NONE)
         async def _() -> str:
+            # Admit (or reject) BEFORE touching the body or any service call (#2392):
+            # rejecting requests never reach the slow path, and quota accounting happens
+            # synchronously at entry, closing the post-await race of burst requests.
+            if not _admit_reset_attempt(time.monotonic()):
+                return self.http_status(429, -1, 'Too many attempts, try again later')
+
             json_data = await quart.request.json
 
             user_email = json_data['user']
@@ -98,7 +152,18 @@ class UserRouterGroup(group.RouterGroup):
             if user_obj is None:
                 return self.http_status(400, -1, 'User not found')
 
-            if recovery_key != self.ap.instance_config.data['system']['recovery_key']:
+            stored_key = self.ap.instance_config.data['system']['recovery_key']
+            try:
+                key_matches = (
+                    isinstance(recovery_key, str)
+                    and isinstance(stored_key, str)
+                    and hmac.compare_digest(recovery_key.encode(), stored_key.encode())
+                )
+            except UnicodeEncodeError:
+                # JSON can contain lone surrogates, which are not valid UTF-8.
+                key_matches = False
+
+            if not key_matches:
                 return self.http_status(403, -1, 'Invalid recovery key')
 
             await self.ap.user_service.reset_password(user_email, new_password)
@@ -186,6 +251,9 @@ class UserRouterGroup(group.RouterGroup):
             json_data = await quart.request.json
             code = json_data.get('code')
             state = json_data.get('state')
+            redirect_uri = json_data.get('redirect_uri') or (
+                quart.request.url_root.rstrip('/') + '/auth/space/callback'
+            )
             launch_assertion = json_data.get('launch_assertion')
             workspace_uuid = json_data.get('workspace_uuid')
 
@@ -199,8 +267,11 @@ class UserRouterGroup(group.RouterGroup):
                 return self.fail(1, 'Missing authorization code')
             if not state:
                 return self.fail(1, 'Missing state parameter')
+            if not str(code).startswith('v4_'):
+                return self.fail(1, 'Unsupported Space OAuth code contract')
 
             try:
+                redirect_uri = self._validate_space_redirect_uri(str(redirect_uri), bind=False)
                 consumed_state = await self.ap.user_service.consume_space_oauth_state_details(state, 'login')
                 # Exchange code for tokens
                 launch_workspace_uuid = consumed_state.launch_workspace_uuid
@@ -218,24 +289,36 @@ class UserRouterGroup(group.RouterGroup):
                     code,
                     workspace_uuids,
                     workspace_created_ats,
+                    redirect_uri=redirect_uri,
                 )
                 access_token = token_data.get('access_token')
                 refresh_token = token_data.get('refresh_token')
                 expires_in = token_data.get('expires_in', 0)
+                cloud_workspace_uuid = token_data.get('cloud_workspace_uuid')
 
                 if not access_token:
                     return self.fail(1, 'Failed to get access token from Space')
 
-                # Authenticate and create/update local user
+                cloud_mode = getattr(getattr(self.ap, 'deployment', None), 'mode', 'oss') == 'cloud'
+                if cloud_mode and launch_workspace_uuid and launch_workspace_uuid != cloud_workspace_uuid:
+                    return self.fail(1, 'Space OAuth Workspace binding mismatch')
+                target_workspace_uuid = launch_workspace_uuid or cloud_workspace_uuid
+                if cloud_mode:
+                    if not target_workspace_uuid:
+                        return self.fail(1, 'Space OAuth response is missing the Cloud Workspace binding')
+                    await self.ap.directory_projection_service.reconcile_workspaces((target_workspace_uuid,))
+
+                # Authenticate only after the signed, exact Workspace delta has
+                # established the Account and membership runtime shadow rows.
                 jwt_token, user_obj = await self.ap.user_service.authenticate_space_user(
                     access_token, refresh_token, expires_in
                 )
 
-                if launch_workspace_uuid:
+                if target_workspace_uuid:
                     try:
                         access = await self.ap.workspace_collaboration_service.resolve_account_workspace(
                             user_obj.uuid,
-                            launch_workspace_uuid,
+                            target_workspace_uuid,
                         )
                     except Exception:
                         self.ap.logger.warning('Rejected Space OAuth launch for unauthorized Workspace')
@@ -323,6 +406,8 @@ class UserRouterGroup(group.RouterGroup):
                 capabilities['password_login_enabled'] = False
             capabilities['authenticated_invitation_acceptance_enabled'] = cloud_mode
             capabilities['invitation_registration_enabled'] = not cloud_mode
+            capabilities['passkey_login_enabled'] = True
+            capabilities['passkey_supported'] = True
             return self.success(data={'initialized': True, **capabilities})
 
         @self.route('/set-password', methods=['POST'], auth_type=group.AuthType.USER_TOKEN)
@@ -367,12 +452,17 @@ class UserRouterGroup(group.RouterGroup):
             json_data = await quart.request.json
             code = json_data.get('code')
             state = json_data.get('state')
+            redirect_uri = json_data.get('redirect_uri') or (
+                quart.request.url_root.rstrip('/') + '/auth/space/callback?mode=bind'
+            )
 
             if not code:
                 return self.http_status(400, -1, 'Missing authorization code')
 
             if not state:
                 return self.http_status(400, -1, 'Missing state parameter')
+            if not str(code).startswith('v4_'):
+                return self.http_status(400, -1, 'Unsupported Space OAuth code contract')
 
             try:
                 user_obj = await self.ap.user_service.consume_space_oauth_state(state, 'bind')
@@ -385,7 +475,10 @@ class UserRouterGroup(group.RouterGroup):
                 return self.http_status(400, -1, 'Only local accounts can bind to Space')
 
             try:
-                updated_user = await self.ap.user_service.bind_space_account(user_obj.user, code)
+                redirect_uri = self._validate_space_redirect_uri(str(redirect_uri), bind=True)
+                updated_user = await self.ap.user_service.bind_space_account(
+                    user_obj.user, code, redirect_uri=redirect_uri
+                )
                 jwt_token = await self.ap.user_service.generate_jwt_token(updated_user)
                 return self.success(
                     data={
@@ -405,6 +498,182 @@ class UserRouterGroup(group.RouterGroup):
             except Exception:
                 raise
 
+        @self.route('/passkey/register/options', methods=['POST'], auth_type=group.AuthType.USER_TOKEN)
+        async def _(user_email: str) -> str:
+            """Generate WebAuthn registration options for current account."""
+            allow_modify_login_info = self.ap.instance_config.data.get('system', {}).get(
+                'allow_modify_login_info', True
+            )
+            if not allow_modify_login_info:
+                return self.http_status(403, -1, 'Modifying login info is disabled')
+
+            user_obj = await self.ap.user_service.get_user_by_email(user_email)
+            if user_obj is None:
+                return self.http_status(404, -1, 'User not found')
+
+            json_data = (await quart.request.json) or {}
+            origin, rp_id = self._extract_origin_and_rp_id(json_data)
+
+            try:
+                options, challenge_token = await self.ap.user_service.generate_passkey_registration_options(
+                    account_uuid=user_obj.uuid,
+                    rp_id=rp_id,
+                    origin=origin,
+                    rp_name='LangBot',
+                )
+                return self.success(data={'options': options, 'challenge_token': challenge_token})
+            except Exception as e:
+                return self.fail(1, str(e))
+
+        @self.route('/passkey/register/verify', methods=['POST'], auth_type=group.AuthType.USER_TOKEN)
+        async def _(user_email: str) -> str:
+            """Verify WebAuthn registration response and save credential."""
+            allow_modify_login_info = self.ap.instance_config.data.get('system', {}).get(
+                'allow_modify_login_info', True
+            )
+            if not allow_modify_login_info:
+                return self.http_status(403, -1, 'Modifying login info is disabled')
+
+            user_obj = await self.ap.user_service.get_user_by_email(user_email)
+            if user_obj is None:
+                return self.http_status(404, -1, 'User not found')
+
+            json_data = await quart.request.json
+            challenge_token = json_data.get('challenge_token')
+            credential = json_data.get('credential') or json_data.get('response')
+            name = json_data.get('name')
+
+            if not challenge_token or not credential:
+                return self.fail(1, 'Missing challenge_token or credential')
+
+            try:
+                cred = await self.ap.user_service.verify_and_save_passkey_registration(
+                    challenge_token=challenge_token,
+                    credential_data=credential,
+                    name=name,
+                )
+                return self.success(
+                    data={
+                        'uuid': cred.uuid,
+                        'name': cred.name,
+                        'created_at': cred.created_at.isoformat() if cred.created_at else None,
+                    }
+                )
+            except Exception as e:
+                return self.fail(1, str(e))
+
+        @self.route('/passkey/auth/options', methods=['POST'], auth_type=group.AuthType.NONE)
+        async def _() -> str:
+            """Generate WebAuthn authentication options for passkey login."""
+            json_data = (await quart.request.json) or {}
+            email = json_data.get('email')
+            origin, rp_id = self._extract_origin_and_rp_id(json_data)
+
+            try:
+                options, challenge_token = await self.ap.user_service.generate_passkey_authentication_options(
+                    rp_id=rp_id,
+                    origin=origin,
+                    email=email,
+                )
+                return self.success(data={'options': options, 'challenge_token': challenge_token})
+            except Exception as e:
+                return self.fail(1, str(e))
+
+        @self.route('/passkey/auth/verify', methods=['POST'], auth_type=group.AuthType.NONE)
+        async def _() -> str:
+            """Verify WebAuthn authentication response and log in."""
+            json_data = await quart.request.json
+            challenge_token = json_data.get('challenge_token')
+            credential = json_data.get('credential') or json_data.get('response')
+
+            if not challenge_token or not credential:
+                return self.fail(1, 'Missing challenge_token or credential')
+
+            try:
+                token, user_obj = await self.ap.user_service.verify_passkey_authentication(
+                    challenge_token=challenge_token,
+                    credential_data=credential,
+                )
+                return self.success(
+                    data={
+                        'token': token,
+                        'user': user_obj.user,
+                    }
+                )
+            except Exception as e:
+                return self.fail(1, str(e))
+
+        @self.route('/passkeys', methods=['GET'], auth_type=group.AuthType.USER_TOKEN)
+        async def _(user_email: str) -> str:
+            """List registered passkeys for the current user."""
+            user_obj = await self.ap.user_service.get_user_by_email(user_email)
+            if user_obj is None:
+                return self.http_status(404, -1, 'User not found')
+
+            passkeys = await self.ap.user_service.get_user_passkeys(user_obj.uuid)
+            return self.success(
+                data=[
+                    {
+                        'uuid': pk.uuid,
+                        'name': pk.name,
+                        'aaguid': pk.aaguid,
+                        'transports': pk.transports,
+                        'backed_up': pk.backed_up,
+                        'created_at': pk.created_at.isoformat() if pk.created_at else None,
+                        'last_used_at': pk.last_used_at.isoformat() if pk.last_used_at else None,
+                    }
+                    for pk in passkeys
+                ]
+            )
+
+        @self.route('/passkey/<passkey_uuid>', methods=['PATCH'], auth_type=group.AuthType.USER_TOKEN)
+        async def _(user_email: str, passkey_uuid: str) -> str:
+            """Rename a registered passkey."""
+            allow_modify_login_info = self.ap.instance_config.data.get('system', {}).get(
+                'allow_modify_login_info', True
+            )
+            if not allow_modify_login_info:
+                return self.http_status(403, -1, 'Modifying login info is disabled')
+
+            user_obj = await self.ap.user_service.get_user_by_email(user_email)
+            if user_obj is None:
+                return self.http_status(404, -1, 'User not found')
+
+            json_data = await quart.request.json
+            name = (json_data.get('name') or '').strip()
+            if not name:
+                return self.fail(1, 'Passkey name cannot be empty')
+
+            updated = await self.ap.user_service.rename_user_passkey(
+                account_uuid=user_obj.uuid,
+                passkey_uuid=passkey_uuid,
+                new_name=name,
+            )
+            if not updated:
+                return self.http_status(404, -1, 'Passkey not found')
+            return self.success(data={'uuid': updated.uuid, 'name': updated.name})
+
+        @self.route('/passkey/<passkey_uuid>', methods=['DELETE'], auth_type=group.AuthType.USER_TOKEN)
+        async def _(user_email: str, passkey_uuid: str) -> str:
+            """Delete/revoke a registered passkey."""
+            allow_modify_login_info = self.ap.instance_config.data.get('system', {}).get(
+                'allow_modify_login_info', True
+            )
+            if not allow_modify_login_info:
+                return self.http_status(403, -1, 'Modifying login info is disabled')
+
+            user_obj = await self.ap.user_service.get_user_by_email(user_email)
+            if user_obj is None:
+                return self.http_status(404, -1, 'User not found')
+
+            deleted = await self.ap.user_service.delete_user_passkey(
+                account_uuid=user_obj.uuid,
+                passkey_uuid=passkey_uuid,
+            )
+            if not deleted:
+                return self.http_status(404, -1, 'Passkey not found')
+            return self.success()
+
     async def _handle_space_direct_launch(
         self,
         launch_assertion: str,
@@ -428,6 +697,10 @@ class UserRouterGroup(group.RouterGroup):
                     }
                 )
 
+            projection_service = self.ap.directory_projection_service
+            if projection_service is None:
+                raise SpaceLaunchError('Cloud directory projection is unavailable')
+            await projection_service.reconcile_workspaces((launch['workspace_uuid'],))
             account = await self.ap.user_service.get_user_by_uuid(launch['account_uuid'])
             if account is None:
                 raise SpaceLaunchError('Launch Account is not projected into Core')

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

@@ -529,16 +529,20 @@ class AgentService:
         return config, component_ref, descriptor.supported_event_patterns
 
     async def get_processor_runs(self, context, processor_id, *, before_id=None):
-        """Read only this Workspace's explicitly created processor instance."""
+        """Read only this Workspace's Agent or plugin processor runs."""
         from ....agent.runner.run_ledger_store import RunLedgerStore
 
         processor = await self.get_agent(context, processor_id)
-        if processor is None or processor.get('kind') != AGENT_KIND_EVENT_PROCESSOR:
-            raise ValueError('Event processor not found')
+        if processor is None or processor.get('kind') not in {'agent', AGENT_KIND_EVENT_PROCESSOR}:
+            raise ValueError('Processor not found')
         store = RunLedgerStore(self.ap.persistence_mgr.get_db_engine())
         items, cursor, has_more, total = await store.list_runs(
             workspace_id=require_workspace_uuid(context),
-            binding_id=f'event_processor:{processor_id}',
+            **(
+                {'binding_id': f'event_processor:{processor_id}'}
+                if processor.get('kind') == AGENT_KIND_EVENT_PROCESSOR
+                else {'agent_id': processor_id}
+            ),
             before_id=before_id,
         )
         return {'items': items, 'next_cursor': cursor, 'has_more': has_more, 'total': total}
@@ -548,14 +552,22 @@ class AgentService:
         from ....agent.runner.run_ledger_store import RunLedgerStore
 
         processor = await self.get_agent(context, processor_id)
-        if processor is None or processor.get('kind') != AGENT_KIND_EVENT_PROCESSOR:
-            raise ValueError('Event processor not found')
+        if processor is None or processor.get('kind') not in {'agent', AGENT_KIND_EVENT_PROCESSOR}:
+            raise ValueError('Processor not found')
         store = RunLedgerStore(self.ap.persistence_mgr.get_db_engine())
         run = await store.get_run(run_id)
         if (
             run is None
             or run.get('workspace_id') != require_workspace_uuid(context)
-            or run.get('binding_id') != f'event_processor:{processor_id}'
+            or not (
+                run.get('binding_id') == f'event_processor:{processor_id}'
+                if processor.get('kind') == AGENT_KIND_EVENT_PROCESSOR
+                else run.get('agent_id') == processor_id
+                or (
+                    run.get('agent_id') is None
+                    and str(run.get('binding_id', '')).startswith((f'agent_{processor_id}_', f'debug:{processor_id}:'))
+                )
+            )
         ):
             raise ValueError('Processor run not found')
         items, next_cursor, _, has_more = await store.page_run_events(

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

@@ -80,16 +80,40 @@ class MaintenanceService:
             DEFAULT_LOG_RETENTION_DAYS,
             'storage.cleanup.log_retention_days',
         )
+        media_cfg = self.ap.instance_config.data.get('storage', {}).get('media_cache', {})
+        media_retention_days = self._positive_int(
+            media_cfg.get('retention_days'),
+            30,
+            'storage.media_cache.retention_days',
+        )
+        media_max_size_mb = self._non_negative_int(
+            media_cfg.get('max_size_mb'),
+            0,
+            'storage.media_cache.max_size_mb',
+        )
+        media_cache = getattr(getattr(self.ap, 'storage_mgr', None), 'media_cache', None)
+        is_singleton = await self._is_oss_singleton(context)
+        media_cleanup = (
+            await media_cache.cleanup(
+                media_retention_days,
+                media_max_size_mb,
+            )
+            if media_cache is not None and is_singleton
+            else {}
+        )
 
-        return {
+        result = {
             'uploaded_files': await self._cleanup_expired_uploaded_files(context, upload_retention_days),
             'log_files': await asyncio.to_thread(
                 self._cleanup_expired_log_files,
                 log_retention_days,
             )
-            if await self._is_oss_singleton(context)
+            if is_singleton
             else 0,
         }
+        if media_cache is not None and is_singleton:
+            result['media_files'] = media_cleanup.get('expired_deleted', 0) + media_cleanup.get('size_deleted', 0)
+        return result
 
     async def get_storage_analysis(self, context: TenantContext) -> dict[str, Any]:
         require_workspace_uuid(context)
@@ -573,6 +597,17 @@ class MaintenanceService:
             count += len(files)
         return count
 
+    def _non_negative_int(self, value: Any, default: int, name: str) -> int:
+        try:
+            parsed = int(value)
+        except (TypeError, ValueError):
+            self.ap.logger.warning(f'Invalid {name}: {value!r}, using {default}')
+            return default
+        if parsed < 0:
+            self.ap.logger.warning(f'{name} must be non-negative: {value!r}, using {default}')
+            return default
+        return parsed
+
     def _positive_int(self, value: Any, default: int, name: str) -> int:
         try:
             parsed = int(value)

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

@@ -446,15 +446,19 @@ class MCPService:
             persisted_session = runtime_mcp_session
 
             async def _refresh_and_report() -> None:
-                needs_start = persisted_session.status == MCPSessionStatus.ERROR or persisted_session.session is None
-                if needs_start:
-                    await persisted_session.start()
-                else:
-                    try:
-                        await persisted_session.refresh()
-                    except Exception:
+                try:
+                    needs_start = (
+                        persisted_session.status == MCPSessionStatus.ERROR or persisted_session.session is None
+                    )
+                    if needs_start:
                         await persisted_session.start()
-                ctx.metadata['runtime_info'] = persisted_session.get_runtime_info_dict()
+                    else:
+                        try:
+                            await persisted_session.refresh()
+                        except Exception:
+                            await persisted_session.start()
+                finally:
+                    ctx.metadata['runtime_info'] = persisted_session.get_runtime_info_dict()
 
             coroutine = _refresh_and_report()
         else:
@@ -471,8 +475,11 @@ class MCPService:
             async def _run_and_cleanup() -> None:
                 try:
                     await test_session.start()
-                    ctx.metadata['runtime_info'] = test_session.get_runtime_info_dict()
                 finally:
+                    # start() raises for a failed connection. Preserve the
+                    # terminal runtime state so the UI can render actionable
+                    # failure phases such as OAuth-required.
+                    ctx.metadata['runtime_info'] = test_session.get_runtime_info_dict()
                     try:
                         await test_session.shutdown()
                     except Exception as exc:

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

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import re
 import uuid
 import datetime
 import functools
@@ -29,6 +30,19 @@ _DEFAULT_CLEANUP_BATCHES_PER_TABLE = 4
 _HARD_MAX_CLEANUP_BATCHES_PER_TABLE = 100
 
 
+def _normalize_user_id(value: str | int | None) -> str | None:
+    """Convert numeric platform IDs before binding a VARCHAR with asyncpg.
+
+    Opaque string IDs (including whitespace and leading zeros) and missing
+    IDs must remain unchanged. Do not silently stringify unsupported objects.
+    """
+    if value is None or isinstance(value, str):
+        return value
+    if isinstance(value, int) and not isinstance(value, bool):
+        return str(value)
+    raise TypeError('user_id must be a string, integer, or None')
+
+
 def _workspace_transaction(method):
     """Run an explicit service entrypoint in one Workspace transaction."""
 
@@ -281,19 +295,21 @@ class MonitoringService:
         for _batch_number in range(max_batches):
 
             async def delete_batch() -> tuple[int, int]:
+                key_columns = list(model_cls.__table__.primary_key.columns)
                 select_result = await self.ap.persistence_mgr.execute_async(
-                    sqlalchemy.select(pk_column)
+                    sqlalchemy.select(*key_columns)
                     .where(model_cls.workspace_uuid == workspace_uuid, ts_column < cutoff)
                     .limit(batch_size)
                 )
-                pk_values = list(select_result.scalars().all())
+                pk_values = [tuple(row) for row in select_result.all()]
                 if not pk_values:
                     return 0, 0
 
                 delete_result = await self.ap.persistence_mgr.execute_async(
                     sqlalchemy.delete(model_cls).where(
                         model_cls.workspace_uuid == workspace_uuid,
-                        pk_column.in_(pk_values),
+                        sqlalchemy.tuple_(*key_columns).in_(pk_values),
+                        ts_column < cutoff,
                     )
                 )
                 return len(pk_values), int(delete_result.rowcount or 0)
@@ -402,6 +418,32 @@ class MonitoringService:
 
     # ========== Recording Methods ==========
 
+    def _sanitize_message_content(self, content: str) -> str:
+        """Strip raw base64 data to protect database storage from unbounded bloating."""
+        if not content or len(content) < 10000 or (';base64,' not in content and 'data:image/' not in content):
+            return content
+        try:
+            data = json.loads(content)
+
+            def _strip_node(node):
+                if isinstance(node, list):
+                    return [_strip_node(x) for x in node]
+                if isinstance(node, dict):
+                    res = dict(node)
+                    if res.get('type') == 'Image' and res.get('base64'):
+                        res['base64'] = None
+                    for k, v in list(res.items()):
+                        if isinstance(v, (list, dict)):
+                            res[k] = _strip_node(v)
+                    return res
+                return node
+
+            return json.dumps(_strip_node(data), ensure_ascii=False)
+        except Exception:
+            return re.sub(
+                r'data:image/[a-zA-Z0-9.+_-]+;base64,[\sA-Za-z0-9+/=]{1000,}', '[base64 image omitted]', content
+            )
+
     @_workspace_transaction
     async def record_message(
         self,
@@ -415,7 +457,7 @@ class MonitoringService:
         status: str = 'success',
         level: str = 'info',
         platform: str | None = None,
-        user_id: str | None = None,
+        user_id: str | int | None = None,
         user_name: str | None = None,
         runner_name: str | None = None,
         variables: str | None = None,
@@ -424,6 +466,7 @@ class MonitoringService:
         """Record a message"""
         workspace_uuid = self._require_write_context(context)
         message_id = str(uuid.uuid4())
+        message_content = self._sanitize_message_content(message_content)
         message_data = {
             'id': message_id,
             'workspace_uuid': workspace_uuid,
@@ -437,7 +480,7 @@ class MonitoringService:
             'status': status,
             'level': level,
             'platform': platform,
-            'user_id': user_id,
+            'user_id': _normalize_user_id(user_id),
             'user_name': user_name,
             'runner_name': runner_name,
             'variables': variables,
@@ -610,7 +653,7 @@ class MonitoringService:
         pipeline_id: str,
         pipeline_name: str,
         platform: str | None = None,
-        user_id: str | None = None,
+        user_id: str | int | None = None,
         user_name: str | None = None,
     ) -> None:
         """Record a new session"""
@@ -622,17 +665,29 @@ class MonitoringService:
             'bot_name': bot_name,
             'pipeline_id': pipeline_id,
             'pipeline_name': pipeline_name,
-            'message_count': 0,
+            'message_count': 1,
             'start_time': datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None),
             'last_activity': datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None),
             'is_active': True,
             'platform': platform,
-            'user_id': user_id,
+            'user_id': _normalize_user_id(user_id),
             'user_name': user_name,
         }
 
+        model = persistence_monitoring.MonitoringSession
+        dialect = self.ap.persistence_mgr.get_db_engine().dialect.name
+        insert = postgresql_dialect.insert if dialect == 'postgresql' else sqlite_dialect.insert
+        statement = insert(model).values(session_data)
         await self.ap.persistence_mgr.execute_async(
-            sqlalchemy.insert(persistence_monitoring.MonitoringSession).values(session_data)
+            statement.on_conflict_do_update(
+                index_elements=['workspace_uuid', 'bot_id', 'session_id'],
+                set_={
+                    'message_count': model.message_count + 1,
+                    'last_activity': statement.excluded.last_activity,
+                    'pipeline_id': statement.excluded.pipeline_id,
+                    'pipeline_name': statement.excluded.pipeline_name,
+                },
+            )
         )
 
     @_workspace_transaction
@@ -642,6 +697,7 @@ class MonitoringService:
         session_id: str,
         pipeline_id: str | None = None,
         pipeline_name: str | None = None,
+        bot_id: str | None = None,
     ) -> bool:
         """Update session last activity time and increment message count.
 
@@ -651,6 +707,9 @@ class MonitoringService:
             True if session was found and updated, False if session doesn't exist.
         """
         workspace_uuid = self._require_write_context(context)
+        bot_id = bot_id if bot_id is not None else context.bot_uuid
+        if not bot_id:
+            raise ValueError('Session activity requires a bot_id')
         update_values = {
             'last_activity': datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None),
             'message_count': persistence_monitoring.MonitoringSession.message_count + 1,
@@ -667,6 +726,7 @@ class MonitoringService:
             .where(
                 persistence_monitoring.MonitoringSession.workspace_uuid == workspace_uuid,
                 persistence_monitoring.MonitoringSession.session_id == session_id,
+                persistence_monitoring.MonitoringSession.bot_id == bot_id,
             )
             .values(update_values)
         )
@@ -769,13 +829,13 @@ class MonitoringService:
             message_conditions.append(persistence_monitoring.MonitoringMessage.timestamp >= start_time)
             llm_conditions.append(persistence_monitoring.MonitoringLLMCall.timestamp >= start_time)
             embedding_conditions.append(persistence_monitoring.MonitoringEmbeddingCall.timestamp >= start_time)
-            session_conditions.append(persistence_monitoring.MonitoringSession.start_time >= start_time)
+            session_conditions.append(persistence_monitoring.MonitoringSession.last_activity >= start_time)
 
         if end_time:
             message_conditions.append(persistence_monitoring.MonitoringMessage.timestamp <= end_time)
             llm_conditions.append(persistence_monitoring.MonitoringLLMCall.timestamp <= end_time)
             embedding_conditions.append(persistence_monitoring.MonitoringEmbeddingCall.timestamp <= end_time)
-            session_conditions.append(persistence_monitoring.MonitoringSession.start_time <= end_time)
+            session_conditions.append(persistence_monitoring.MonitoringSession.last_activity <= end_time)
 
         # Total messages
         message_query = sqlalchemy.select(sqlalchemy.func.count(persistence_monitoring.MonitoringMessage.id))
@@ -1257,6 +1317,7 @@ class MonitoringService:
         pipeline_ids: list[str] | None = None,
         start_time: datetime.datetime | None = None,
         end_time: datetime.datetime | None = None,
+        user_query: str | None = None,
         is_active: bool | None = None,
         limit: int = 100,
         offset: int = 0,
@@ -1271,9 +1332,17 @@ class MonitoringService:
         if pipeline_ids:
             conditions.append(persistence_monitoring.MonitoringSession.pipeline_id.in_(pipeline_ids))
         if start_time:
-            conditions.append(persistence_monitoring.MonitoringSession.start_time >= start_time)
+            conditions.append(persistence_monitoring.MonitoringSession.last_activity >= start_time)
         if end_time:
-            conditions.append(persistence_monitoring.MonitoringSession.start_time <= end_time)
+            conditions.append(persistence_monitoring.MonitoringSession.last_activity <= end_time)
+        if user_query and user_query.strip():
+            user_pattern = f'%{user_query.strip()}%'
+            conditions.append(
+                sqlalchemy.or_(
+                    persistence_monitoring.MonitoringSession.user_id.ilike(user_pattern),
+                    persistence_monitoring.MonitoringSession.user_name.ilike(user_pattern),
+                )
+            )
         if is_active is not None:
             conditions.append(persistence_monitoring.MonitoringSession.is_active == is_active)
 
@@ -1365,6 +1434,9 @@ class MonitoringService:
         self,
         context: TenantContext,
         session_id: str,
+        start_time: datetime.datetime | None = None,
+        end_time: datetime.datetime | None = None,
+        bot_id: str | None = None,
     ) -> dict:
         """Get bounded session details with full statistics computed in SQL."""
         workspace_uuid = require_workspace_uuid(context)
@@ -1374,8 +1446,13 @@ class MonitoringService:
             persistence_monitoring.MonitoringSession.workspace_uuid == workspace_uuid,
             persistence_monitoring.MonitoringSession.session_id == session_id,
         )
-        session_result = await self.ap.persistence_mgr.execute_async(session_query)
-        session_row = session_result.first()
+        if bot_id is not None:
+            session_query = session_query.where(persistence_monitoring.MonitoringSession.bot_id == bot_id)
+        session_result = await self.ap.persistence_mgr.execute_async(session_query.limit(2))
+        session_rows = session_result.all()
+        if len(session_rows) > 1:
+            return {'session_id': session_id, 'found': False, 'ambiguous': True}
+        session_row = session_rows[0] if session_rows else None
 
         if not session_row:
             return {
@@ -1384,6 +1461,7 @@ class MonitoringService:
             }
 
         session = session_row[0] if isinstance(session_row, tuple) else session_row
+        bot_id = session.bot_id
 
         message_stats_result = await self.ap.persistence_mgr.execute_async(
             sqlalchemy.select(
@@ -1411,6 +1489,7 @@ class MonitoringService:
             ).where(
                 persistence_monitoring.MonitoringMessage.workspace_uuid == workspace_uuid,
                 persistence_monitoring.MonitoringMessage.session_id == session_id,
+                persistence_monitoring.MonitoringMessage.bot_id == bot_id,
             )
         )
         message_stats = message_stats_result.one()
@@ -1449,6 +1528,7 @@ class MonitoringService:
             ).where(
                 persistence_monitoring.MonitoringLLMCall.workspace_uuid == workspace_uuid,
                 persistence_monitoring.MonitoringLLMCall.session_id == session_id,
+                persistence_monitoring.MonitoringLLMCall.bot_id == bot_id,
             )
         )
         llm_stats = llm_stats_result.one()
@@ -1475,15 +1555,22 @@ class MonitoringService:
             ).where(
                 persistence_monitoring.MonitoringToolCall.workspace_uuid == workspace_uuid,
                 persistence_monitoring.MonitoringToolCall.session_id == session_id,
+                persistence_monitoring.MonitoringToolCall.bot_id == bot_id,
             )
         )
         tool_stats = tool_stats_result.one()
+        tool_conditions = [
+            persistence_monitoring.MonitoringToolCall.workspace_uuid == workspace_uuid,
+            persistence_monitoring.MonitoringToolCall.session_id == session_id,
+            persistence_monitoring.MonitoringToolCall.bot_id == bot_id,
+        ]
+        if start_time is not None:
+            tool_conditions.append(persistence_monitoring.MonitoringToolCall.timestamp >= start_time)
+        if end_time is not None:
+            tool_conditions.append(persistence_monitoring.MonitoringToolCall.timestamp <= end_time)
         tool_query = (
             sqlalchemy.select(persistence_monitoring.MonitoringToolCall)
-            .where(
-                persistence_monitoring.MonitoringToolCall.workspace_uuid == workspace_uuid,
-                persistence_monitoring.MonitoringToolCall.session_id == session_id,
-            )
+            .where(*tool_conditions)
             .order_by(persistence_monitoring.MonitoringToolCall.timestamp.asc())
             .limit(detail_limit + 1)
         )
@@ -1504,6 +1591,7 @@ class MonitoringService:
             .where(
                 persistence_monitoring.MonitoringError.workspace_uuid == workspace_uuid,
                 persistence_monitoring.MonitoringError.session_id == session_id,
+                persistence_monitoring.MonitoringError.bot_id == bot_id,
             )
             .order_by(persistence_monitoring.MonitoringError.timestamp.desc())
             .limit(detail_limit + 1)
@@ -1988,9 +2076,9 @@ class MonitoringService:
         if pipeline_ids:
             conditions.append(persistence_monitoring.MonitoringSession.pipeline_id.in_(pipeline_ids))
         if start_time:
-            conditions.append(persistence_monitoring.MonitoringSession.start_time >= start_time)
+            conditions.append(persistence_monitoring.MonitoringSession.last_activity >= start_time)
         if end_time:
-            conditions.append(persistence_monitoring.MonitoringSession.start_time <= end_time)
+            conditions.append(persistence_monitoring.MonitoringSession.last_activity <= end_time)
 
         query = sqlalchemy.select(persistence_monitoring.MonitoringSession).order_by(
             persistence_monitoring.MonitoringSession.last_activity.desc()
@@ -2024,6 +2112,7 @@ class MonitoringService:
 
     # ========== Feedback Methods ==========
 
+    @_workspace_transaction
     async def record_feedback(
         self,
         context: ExecutionContext,
@@ -2038,7 +2127,7 @@ class MonitoringService:
         session_id: str | None = None,
         message_id: str | None = None,
         stream_id: str | None = None,
-        user_id: str | None = None,
+        user_id: str | int | None = None,
         platform: str | None = None,
     ) -> str | None:
         """Record user feedback (like/dislike) from AI Bot conversation.
@@ -2094,7 +2183,7 @@ class MonitoringService:
             'session_id': session_id,
             'message_id': message_id,
             'stream_id': stream_id,
-            'user_id': user_id,
+            'user_id': _normalize_user_id(user_id),
             'platform': platform,
         }
         dialect_name = self.ap.persistence_mgr.get_db_engine().dialect.name

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

@@ -0,0 +1,83 @@
+"""Bounded traffic aggregation, independent of record-list pagination."""
+
+from __future__ import annotations
+
+import datetime
+import typing
+
+import sqlalchemy
+
+from ....entity.persistence.monitoring import MonitoringLLMCall, MonitoringMessage
+from .tenant import TenantContext, require_workspace_uuid
+
+if typing.TYPE_CHECKING:
+    from ....core.app import Application
+
+MAX_TRAFFIC_POINTS = 1000
+
+
+async def get_traffic_series(
+    ap: Application,
+    context: TenantContext,
+    *,
+    bot_ids: list[str] | None = None,
+    pipeline_ids: list[str] | None = None,
+    start_time: datetime.datetime | None = None,
+    end_time: datetime.datetime | None = None,
+) -> dict:
+    """Count all matching records in UTC buckets, returning at most 1000 points."""
+    workspace_uuid = require_workspace_uuid(context)
+    bucket = 'hour' if start_time and end_time and end_time - start_time <= datetime.timedelta(days=7) else 'day'
+    step = datetime.timedelta(hours=1) if bucket == 'hour' else datetime.timedelta(days=1)
+    postgres = ap.persistence_mgr.get_db_engine().dialect.name == 'postgresql'
+    points: dict[datetime.datetime, dict[str, int]] = {}
+    truncated = False
+    for model, field in ((MonitoringMessage, 'messages'), (MonitoringLLMCall, 'llm_calls')):
+        timestamp = model.timestamp
+        if postgres:
+            time_bucket = sqlalchemy.func.date_trunc(bucket, timestamp)
+        else:
+            pattern = '%Y-%m-%dT%H:00:00' if bucket == 'hour' else '%Y-%m-%dT00:00:00'
+            time_bucket = sqlalchemy.func.strftime(pattern, timestamp)
+        conditions = [model.workspace_uuid == workspace_uuid]
+        if bot_ids:
+            conditions.append(model.bot_id.in_(bot_ids))
+        if pipeline_ids:
+            conditions.append(model.pipeline_id.in_(pipeline_ids))
+        if start_time is not None:
+            conditions.append(timestamp >= start_time)
+        if end_time is not None:
+            conditions.append(timestamp <= end_time)
+        statement = (
+            sqlalchemy.select(time_bucket.label('bucket'), sqlalchemy.func.count(model.id).label('count'))
+            .where(*conditions)
+            .group_by(time_bucket)
+            .order_by(time_bucket)
+            .limit(MAX_TRAFFIC_POINTS + 1)
+        )
+        result = await ap.persistence_mgr.execute_async(statement)
+        rows = result.all()
+        truncated = truncated or len(rows) > MAX_TRAFFIC_POINTS
+        for timestamp_value, count in rows[:MAX_TRAFFIC_POINTS]:
+            key = (
+                datetime.datetime.fromisoformat(timestamp_value)
+                if isinstance(timestamp_value, str)
+                else timestamp_value
+            )
+            points.setdefault(key, {'messages': 0, 'llm_calls': 0})[field] = int(count)
+
+    def floor(value: datetime.datetime) -> datetime.datetime:
+        return value.replace(minute=0, second=0, microsecond=0, **({'hour': 0} if bucket == 'day' else {}))
+
+    first = floor(start_time) if start_time is not None else min(points, default=None)
+    last = floor(end_time) if end_time is not None else max(points, default=None)
+    series = []
+    if first is not None and last is not None:
+        cursor = first
+        while cursor <= last and len(series) < MAX_TRAFFIC_POINTS:
+            series.append(
+                {'timestamp': cursor.isoformat() + 'Z', **points.get(cursor, {'messages': 0, 'llm_calls': 0})}
+            )
+            cursor += step
+        truncated = truncated or cursor <= last
+    return {'bucket': bucket, 'points': series, 'truncated': truncated}

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

@@ -13,7 +13,7 @@ from ....pipeline.extension_preferences import (
     validate_extension_preferences,
 )
 from ....workspace.errors import WorkspaceNotFoundError
-from .secrets import contains_secret_placeholder, redact_secrets, restore_secret_placeholders
+from .secrets import redact_secrets, restore_secret_placeholders
 from .tenant import TenantContext, require_workspace_uuid, scope_statement
 
 
@@ -213,12 +213,24 @@ class PipelineService:
             pipeline_data.pop(protected_field, None)
 
         if 'config' in pipeline_data:
-            current_config = None
-            if contains_secret_placeholder(pipeline_data['config']):
-                current_pipeline = await self.get_pipeline(context, pipeline_uuid, include_secret=True)
-                if current_pipeline is None:
-                    raise WorkspaceNotFoundError('Pipeline not found')
-                current_config = current_pipeline.get('config', {})
+            current_pipeline = await self.get_pipeline(context, pipeline_uuid, include_secret=True)
+            if current_pipeline is None:
+                raise WorkspaceNotFoundError('Pipeline not found')
+            current_config = current_pipeline.get('config', {})
+            old_ai = current_config.get('ai', {}) if isinstance(current_config, dict) else {}
+            old_runner = old_ai.get('runner') if isinstance(old_ai, dict) else None
+            legacy_binding = (
+                isinstance(old_runner, str)
+                and bool(old_runner)
+                or isinstance(old_runner, dict)
+                and bool(old_runner.get('runner'))
+                and not old_runner.get('id')
+            )
+            new_config = pipeline_data['config']
+            new_ai = new_config.get('ai', {}) if isinstance(new_config, dict) else {}
+            new_runner = new_ai.get('runner') if isinstance(new_ai, dict) else None
+            if legacy_binding and isinstance(new_runner, dict) and new_runner.get('id'):
+                raise ValueError('manual_migration_required')
             pipeline_data['config'] = restore_secret_placeholders(
                 pipeline_data['config'],
                 current_config if current_config is not None else {},

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

@@ -0,0 +1,936 @@
+"""Explicit, tenant-scoped migration. Preview never contacts plugin runtimes.
+
+The task handle is process-local. Original snapshots and pending activation are
+persistent; a refreshed preview is required after observation loss or restart.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import copy
+import hashlib
+import hmac
+import json
+import math
+import time
+import secrets
+import sys
+import uuid
+
+from langbot_plugin.runtime.plugin.mgr import PluginInstallSource
+
+import httpx
+import sqlalchemy as sa
+
+from ..authz import Permission, permissions_for_role, require_permission
+from ..context import ExecutionContext, PrincipalType, RequestContext
+from ....agent.runner.config_resolver import RunnerConfigResolver
+from ....agent.runner import config_schema
+from ....agent.runner.resource_builder import AgentResourceBuilder
+from ....agent.runner.resource_policy import ResourcePolicyProjector
+from ....entity.persistence.model import LLMModel, RerankModel, EmbeddingModel
+from ....entity.persistence.rag import KnowledgeBase
+from ....entity.persistence.mcp import MCPServer
+from ....core.taskmgr import TaskContext
+from ....entity.persistence.pipeline import LegacyPipeline
+from ....entity.persistence.agent_interaction import AgentInteraction
+from ....persistence.pipeline_admission import lock_pipeline_admission
+from ....entity.persistence.pipeline_migration import PipelineMigrationSnapshot
+from ....entity.persistence.plugin import PluginSetting
+from ....entity.persistence.user import User
+from ....entity.persistence.workspace import Workspace, WorkspaceMembership, WorkspaceExecutionState
+from ....plugin.errors import (
+    PluginRuntimeNotConnectedError,
+    MarketplacePluginVersionNotFoundError,
+    PluginInstallationFailedError,
+)
+from ....pipeline.legacy_config_migration import PLANNER_VERSION, plan_legacy_pipeline
+
+
+class MigrationInstallContext(TaskContext):
+    """Publish only structured installer progress, never upstream logs or URLs."""
+
+    ACTIONS = {
+        'downloading plugin package': 'downloading',
+        'validating plugin package': 'validating',
+        'preparing plugin installation': 'preparing',
+        'installing plugin dependencies': 'installing_deps',
+        'waiting for plugin initialization': 'activating',
+        'refreshing plugin components': 'refreshing',
+        'plugin installed': 'done',
+    }
+    METRICS = ('progress_percent', 'download_current', 'download_total', 'download_speed', 'deps_total')
+
+    def __init__(self, target):
+        super().__init__()
+        self.progress = {
+            **target,
+            'status': 'installing',
+            'stage': 'checking',
+            'code': None,
+            'started_at': time.time(),
+            'finished_at': None,
+            'steps': [],
+        }
+        self._step('checking')
+
+    def _step(self, stage):
+        steps = self.progress['steps']
+        if steps and steps[-1]['stage'] == stage:
+            return
+        now = time.time()
+        if steps:
+            steps[-1]['finished_at'] = now
+        steps.append({'stage': stage, 'started_at': now, 'finished_at': None})
+        self.progress['stage'] = stage
+
+    def set_current_action(self, action):
+        super().set_current_action(action)
+        stage = self.ACTIONS.get(action)
+        if stage:
+            self._step(stage)
+        self.publish()
+
+    def publish(self):
+        for key in self.METRICS:
+            value = self.metadata.get(key)
+            if type(value) in (int, float) and math.isfinite(value) and value >= 0:
+                self.progress[key] = min(value, 100) if key == 'progress_percent' else value
+        self.progress['updated_at'] = time.time()
+
+    def finish(self, code=None):
+        self.publish()
+        self.progress.update(status='failed' if code else 'completed', code=code, finished_at=time.time())
+        self.progress['steps'][-1]['finished_at'] = self.progress['finished_at']
+        if not code:
+            self.progress['progress_percent'] = 100
+
+    async def observe(self):
+        while True:
+            self.publish()
+            await asyncio.sleep(0.2)
+
+
+class MigrationError(ValueError):
+    """Only constant, credential-free codes cross the API/task boundary."""
+
+    def __init__(self, code: str, status_code: int = 409):
+        super().__init__(code)
+        self.code = code
+        self.status_code = status_code
+
+
+def _failure_code(exc: Exception) -> str:
+    if isinstance(exc, MigrationError):
+        return exc.code
+    if isinstance(exc, PluginRuntimeNotConnectedError):
+        return 'plugin_runtime_unavailable'
+    return 'migration_failed'
+
+
+def validate_execute_request(body) -> list[dict]:
+    if not isinstance(body, dict) or set(body) != {'confirmed', 'items'} or body['confirmed'] is not True:
+        raise MigrationError('confirmation_required', 400)
+    items = body['items']
+    if not isinstance(items, list) or not 1 <= len(items) <= 50:
+        raise MigrationError('invalid_selection', 400)
+    seen = set()
+    for item in items:
+        if not isinstance(item, dict) or set(item) != {'pipeline_uuid', 'preview_token'}:
+            raise MigrationError('invalid_selection', 400)
+        for key, maximum in [('pipeline_uuid', 255), ('preview_token', 256)]:
+            value = item[key]
+            if not isinstance(value, str) or not value.strip() or len(value) > maximum or value != value.strip():
+                raise MigrationError('invalid_selection', 400)
+        if item['pipeline_uuid'] in seen:
+            raise MigrationError('invalid_selection', 400)
+        seen.add(item['pipeline_uuid'])
+    return copy.deepcopy(items)
+
+
+def _json(value):
+    return json.dumps(value, sort_keys=True, separators=(',', ':'), ensure_ascii=False, default=str)
+
+
+def _fingerprint(row):
+    # SQL timestamps are neither monotonic revisions nor portable CAS values.
+    return hashlib.sha256(
+        _json(
+            {k: v for k, v in row.items() if not k.startswith('_') and k not in ('created_at', 'updated_at')}
+        ).encode()
+    ).hexdigest()
+
+
+class PipelineMigrationService:
+    def __init__(self, ap):
+        self.ap = ap
+        self.pm = ap.persistence_mgr
+        # Loss of this process key only invalidates previews, never snapshots.
+        self._token_key = secrets.token_bytes(32)
+        self._all_tasks: set[str] = set()
+
+    async def _binding(self, ctx, session=None):
+        binding = await self.ap.workspace_service.get_execution_binding(
+            ctx.workspace_uuid, expected_generation=ctx.placement_generation, session=session
+        )
+        if binding.instance_uuid != ctx.instance_uuid:
+            raise MigrationError('placement_changed')
+
+    async def _authorize(self, ctx, session=None):
+        require_permission(ctx, Permission.RESOURCE_MANAGE)
+        if ctx.principal.principal_type != PrincipalType.ACCOUNT or not ctx.account_uuid:
+            raise MigrationError('authorization_changed', 403)
+        status = (await self.pm.execute_async(sa.select(User.status).where(User.uuid == ctx.account_uuid))).scalar()
+        if status != 'active':
+            raise MigrationError('authorization_changed', 403)
+        access = await self.ap.workspace_collaboration_service.resolve_account_workspace(
+            ctx.account_uuid, ctx.workspace_uuid, session=session
+        )
+        if (
+            access.workspace.uuid != ctx.workspace_uuid
+            or access.membership.uuid != ctx.workspace.membership_uuid
+            or access.membership.projection_revision != ctx.workspace.membership_revision
+            or Permission.RESOURCE_MANAGE.value not in permissions_for_role(access.membership.role)
+            or access.execution.instance_uuid != ctx.instance_uuid
+            or access.execution.placement_generation != ctx.placement_generation
+        ):
+            raise MigrationError('authorization_changed', 403)
+        await self._binding(ctx, session)
+
+    async def _rows(self, ctx, pipeline_uuid=None):
+        # Read DB-native text for JSON CAS: never compare PostgreSQL JSON using =,
+        # and never assume Python's JSON whitespace/key order matches stored text.
+        table = LegacyPipeline.__table__
+        stmt = sa.select(
+            table,
+            *[sa.cast(table.c[k], sa.Text).label('_raw_' + k) for k in ('config', 'stages', 'extensions_preferences')],
+        )
+        stmt = stmt.where(table.c.workspace_uuid == ctx.workspace_uuid)
+        if pipeline_uuid is not None:
+            stmt = stmt.where(table.c.uuid == pipeline_uuid)
+        return [dict(r._mapping) for r in (await self.pm.execute_async(stmt.order_by(table.c.uuid))).all()]
+
+    async def _plugin(self, ctx, plan):
+        target = plan.get('target_plugin')
+        if not target:
+            return None
+        t = PluginSetting.__table__
+        row = (
+            await self.pm.execute_async(
+                sa.select(
+                    t.c.enabled, t.c.installation_uuid, t.c.artifact_digest, t.c.runtime_revision, t.c.install_info
+                ).where(
+                    t.c.workspace_uuid == ctx.workspace_uuid,
+                    t.c.plugin_author == target['author'],
+                    t.c.plugin_name == target['name'],
+                )
+            )
+        ).first()
+        return dict(row._mapping) if row else None
+
+    def _native_interactions(self, ctx, pipeline_uuid):
+        """Inspect the old in-process cache without importing/pruning it.
+
+        Native forms have no durable handoff to the plugin runner. Operators
+        must finish them in the original runtime before migrating; never infer
+        completion from the cache TTL or submit/cancel a form here.
+        """
+        module = sys.modules.get('langbot.pkg.provider.runners.difysvapi')
+        if module is None:
+            return []
+        cache = getattr(module, '_PENDING_FORMS', None)
+        if not isinstance(cache, dict):
+            return ['unavailable']
+        observed = []
+        for key, forms in cache.items():
+            if not isinstance(key, tuple) or len(key) != 8:
+                if forms:
+                    observed.append('unscoped')
+                continue
+            instance, workspace, _, _, pipeline, *_ = key
+            # Inspect ownership before touching any form contents. Empty scope
+            # cannot prove this is another tenant; block without disclosing it.
+            if instance and instance != ctx.instance_uuid or workspace and workspace != ctx.workspace_uuid:
+                continue
+            if pipeline and pipeline != pipeline_uuid:
+                continue
+            if forms:
+                observed.append(hashlib.sha256(_json(key).encode()).hexdigest())
+        return sorted(observed)
+
+    async def _interaction_state(self, ctx, pipeline_uuid):
+        """Read only correlation/status, never request/submission/token data.
+
+        The real table has workspace_id (not workspace_uuid), and no instance
+        column. Execution placement is verified by the caller; processor_id is
+        the globally unique pipeline UUID. NULL ownership on that exact UUID
+        is uncertainty, not permission to query other Workspaces.
+        """
+        if self.pm.get_db_engine().dialect.name not in {'sqlite', 'postgresql'}:
+            return {'boundary': 'unavailable'}, True
+        t = AgentInteraction.__table__
+        try:
+            records = (
+                await self.pm.execute_async(
+                    sa.select(t.c.id, t.c.status, t.c.updated_at, t.c.workspace_id)
+                    .where(
+                        t.c.processor_type == 'pipeline',
+                        t.c.processor_id == pipeline_uuid,
+                        sa.or_(
+                            t.c.workspace_id == ctx.workspace_uuid, t.c.workspace_id.is_(None), t.c.workspace_id == ''
+                        ),
+                    )
+                    .order_by(t.c.id)
+                )
+            ).all()
+            durable = [tuple(r) for r in records]
+            blocked = any(r.status not in {'submitted', 'cancelled', 'expired', 'delivery_failed'} for r in records)
+        except Exception:
+            # Missing schema, permissions, or a failed store must not be treated
+            # as an empty store. No original diagnostic crosses the API.
+            durable, blocked = ['unavailable'], True
+        native = self._native_interactions(ctx, pipeline_uuid)
+        return {'durable': durable, 'native': native}, blocked or bool(native)
+
+    async def _plan(self, ctx, row):
+        plan = plan_legacy_pipeline(row['config'], row['extensions_preferences'])
+        pending = None
+        if plan['state'] == 'already_current':
+            t = PipelineMigrationSnapshot.__table__
+            snapshots = (
+                await self.pm.execute_async(
+                    sa.select(t).where(
+                        t.c.workspace_uuid == ctx.workspace_uuid,
+                        t.c.pipeline_uuid == row['uuid'],
+                        t.c.state == 'activation_pending',
+                        t.c.target_fingerprint == _fingerprint(row),
+                        t.c.planner_version == str(PLANNER_VERSION),
+                    )
+                )
+            ).all()
+            if snapshots:
+                pending = dict(snapshots[0]._mapping)
+                source = pending['source_snapshot']
+                plan = plan_legacy_pipeline(source['config'], source['extensions_preferences'])
+                candidate = {**row, 'config': plan.get('config')}
+                if plan['state'] != 'ready' or _fingerprint(candidate) != _fingerprint(row):
+                    raise MigrationError('planner_changed')
+        facts = await self._plugin(ctx, plan)
+        state = 'activation_pending' if pending else plan['state']
+        blockers = list(plan['blockers'])
+        if state in ('ready', 'activation_pending'):
+            if facts is None:
+                state = 'needs_plugin'
+                blockers.append({'code': 'plugin_missing'})
+            elif facts['enabled'] is not True:
+                state = 'blocked'
+                blockers.append({'code': 'plugin_disabled'})
+        if plan['state'] == 'ready':
+            observation, interaction_blocked = await self._interaction_state(ctx, row['uuid'])
+            plan['_interaction_state'] = observation
+            if interaction_blocked:
+                state = 'blocked'
+                blockers.append({'code': 'runtime.pending_interaction'})
+        return plan, facts, pending, state, blockers
+
+    def _token(self, ctx, row, plan, facts, pending):
+        claim = [
+            ctx.instance_uuid,
+            ctx.workspace_uuid,
+            ctx.placement_generation,
+            ctx.account_uuid,
+            ctx.workspace.membership_uuid,
+            ctx.workspace.membership_revision,
+            str(PLANNER_VERSION),
+            _fingerprint(row),
+            plan,
+            facts,
+            pending['uuid'] if pending else None,
+        ]
+        # An authenticated opaque digest discloses neither credentials nor configs.
+        return hmac.new(self._token_key, _json(claim).encode(), hashlib.sha256).hexdigest()
+
+    async def preview(self, ctx: RequestContext):
+        require_permission(ctx, Permission.RESOURCE_VIEW)
+        async with self.pm.tenant_scope(ctx.workspace_uuid):
+            await self._binding(ctx)
+            items = []
+            for row in await self._rows(ctx):
+                plan, facts, pending, state, blockers = await self._plan(ctx, row)
+                item = {
+                    k: copy.deepcopy(plan[k])
+                    for k in ('legacy_runner', 'target_runner_id', 'target_plugin', 'changed_paths', 'warnings')
+                }
+                item.update(
+                    pipeline_uuid=row['uuid'],
+                    name=row['name'],
+                    state=state,
+                    blockers=blockers,
+                    preview_token=self._token(ctx, row, plan, facts, pending)
+                    if state in ('ready', 'activation_pending')
+                    else None,
+                )
+                items.append(item)
+            return {
+                'planner_version': str(PLANNER_VERSION),
+                'workspace_uuid': ctx.workspace_uuid,
+                'items': items,
+                'total': len(items),
+            }
+
+    async def _selected(self, ctx, selection, *, data_only=False):
+        rows = await self._rows(ctx, selection['pipeline_uuid'])
+        if not rows:
+            raise MigrationError('pipeline_not_found', 404)
+        row = rows[0]
+        plan, facts, pending, state, blockers = await self._plan(ctx, row)
+        token = self._token(ctx, row, plan, facts, pending)
+        if not hmac.compare_digest(token, selection['preview_token']):
+            raise MigrationError('preview_stale')
+        if state not in ('ready', 'activation_pending') and not (
+            data_only
+            and plan['state'] == 'ready'
+            and all(b['code'] in ('plugin_missing', 'plugin_disabled') for b in blockers)
+        ):
+            raise MigrationError('migration_blocked')
+        return row, plan, facts, pending
+
+    async def execute(self, ctx: RequestContext, body):
+        require_permission(ctx, Permission.RESOURCE_MANAGE)
+        if isinstance(body, dict) and body.get('all') is True:
+            if (
+                set(body) != {'confirmed', 'all', 'install_plugins'}
+                or body['confirmed'] is not True
+                or not isinstance(body['install_plugins'], bool)
+            ):
+                raise MigrationError('confirmation_required', 400)
+            return await self._execute_all(ctx, install_plugins=body['install_plugins'])
+        items = validate_execute_request(body)
+        async with self.pm.tenant_scope(ctx.workspace_uuid):
+            await self._authorize(ctx)
+            # Preflight the entire selection before creating tasks or contacting runtimes.
+            for selection in items:
+                await self._selected(ctx, selection)
+        task_context = TaskContext.new()
+        task_context.metadata = {
+            'kind': 'pipeline_migration',
+            'results': [{'pipeline_uuid': i['pipeline_uuid'], 'state': 'pending', 'code': None} for i in items],
+        }
+        task = self.ap.task_mgr.create_user_task(
+            self._run(ctx, ExecutionContext.from_request(ctx), items, task_context),
+            kind='pipeline_migration',
+            name='pipeline_migration',
+            context=task_context,
+            instance_uuid=ctx.instance_uuid,
+            workspace_uuid=ctx.workspace_uuid,
+            placement_generation=ctx.placement_generation,
+        )
+        return {'task_id': task.id}
+
+    async def _execute_all(self, ctx, *, install_plugins):
+        # Capture all sources on admission. Plugin installation must not silently
+        # include later edits or pipelines created while the task is running.
+        if ctx.workspace_uuid in self._all_tasks:
+            raise MigrationError('migration_running')
+        self._all_tasks.add(ctx.workspace_uuid)
+        try:
+            async with self.pm.tenant_scope(ctx.workspace_uuid):
+                await self._authorize(ctx)
+                sources = []
+                for row in await self._rows(ctx):
+                    _, _, _, state, _ = await self._plan(ctx, row)
+                    if state not in ('already_current', 'not_legacy'):
+                        sources.append(row)
+            if not sources:
+                raise MigrationError('nothing_to_migrate')
+            task_context = TaskContext.new()
+            task_context.metadata = {
+                'kind': 'pipeline_migration',
+                'phase': 'installing' if install_plugins else 'migrating',
+                'installations': [],
+                'results': [{'pipeline_uuid': row['uuid'], 'state': 'pending', 'code': None} for row in sources],
+            }
+            task = self.ap.task_mgr.create_user_task(
+                self._run_all(
+                    ctx, ExecutionContext.from_request(ctx), sources, task_context, install_plugins=install_plugins
+                ),
+                kind='pipeline_migration',
+                name='pipeline_migration',
+                context=task_context,
+                instance_uuid=ctx.instance_uuid,
+                workspace_uuid=ctx.workspace_uuid,
+                placement_generation=ctx.placement_generation,
+            )
+            return {'task_id': task.id, 'pipeline_uuids': [row['uuid'] for row in sources]}
+        except BaseException:
+            self._all_tasks.discard(ctx.workspace_uuid)
+            raise
+
+    async def _run_all(self, ctx, execution, sources, task_context, *, install_plugins):
+        installed = {}
+        try:
+            async with self.pm.tenant_scope(execution.workspace_uuid):
+                for source, result in zip(sources, task_context.metadata['results']):
+                    if result['state'] != 'pending':
+                        continue
+                    try:
+                        await self._authorize(ctx)
+                        current = await self._rows(ctx, source['uuid'])
+                        if not current or _fingerprint(current[0]) != _fingerprint(source):
+                            raise MigrationError('preview_stale')
+                        plan, facts, pending, state, blockers = await self._plan(ctx, current[0])
+                        hard_blockers = [b for b in blockers if b['code'] not in ('plugin_missing', 'plugin_disabled')]
+                        if hard_blockers:
+                            raise MigrationError(hard_blockers[0]['code'])
+                        if state in ('already_current', 'not_legacy'):
+                            result.update(state='already_current', code=None)
+                            continue
+                        target = plan.get('target_plugin')
+                        if not target:
+                            raise MigrationError('migration_blocked')
+                        if not install_plugins:
+                            selection = {
+                                'pipeline_uuid': source['uuid'],
+                                'preview_token': self._token(ctx, current[0], plan, facts, pending),
+                            }
+                            await self._run(ctx, execution, [selection], task_context, results=[result], data_only=True)
+                            continue
+                        key = (target['author'], target['name'], target['version'])
+                        if key not in installed:
+                            runners = await self.ap.runner_registry.list_runners(
+                                execution, use_cache=False, usage='agent'
+                            )
+                            descriptor = next((r for r in runners if r.id == plan['target_runner_id']), None)
+                            ready = (
+                                facts
+                                and facts['enabled']
+                                and descriptor
+                                and descriptor.plugin_version == target['version']
+                            )
+                            if not ready:
+                                task_context.metadata['phase'] = 'installing'
+                                install_context = MigrationInstallContext(target)
+                                task_context.metadata['installations'].append(install_context.progress)
+                                observer = asyncio.create_task(install_context.observe())
+                                install_code = 'operation_cancelled'
+                                try:
+                                    await self._authorize(ctx)
+                                    await self.ap.plugin_connector.require_workspace_context(execution)
+                                    await self.ap.plugin_connector.install_plugin(
+                                        PluginInstallSource.MARKETPLACE,
+                                        {
+                                            'plugin_author': target['author'],
+                                            'plugin_name': target['name'],
+                                            'plugin_version': target['version'],
+                                        },
+                                        task_context=install_context,
+                                    )
+                                    install_code = None
+                                except PluginRuntimeNotConnectedError:
+                                    install_code = 'plugin_runtime_unavailable'
+                                except MarketplacePluginVersionNotFoundError:
+                                    install_code = 'plugin_version_unavailable'
+                                except httpx.TimeoutException:
+                                    install_code = 'plugin_download_timeout'
+                                except httpx.HTTPStatusError as exc:
+                                    install_code = (
+                                        'plugin_version_unavailable'
+                                        if exc.response.status_code == 404
+                                        else 'plugin_marketplace_unavailable'
+                                    )
+                                except httpx.RequestError:
+                                    install_code = 'plugin_download_failed'
+                                except PluginInstallationFailedError as exc:
+                                    install_code = {
+                                        'dependency_prepare_failed': 'dependency_prepare_failed',
+                                        'worker_launch_failed': 'plugin_launch_failed',
+                                    }.get(exc.error_code, 'plugin_install_failed')
+                                except Exception:
+                                    install_code = 'plugin_install_failed'
+                                finally:
+                                    observer.cancel()
+                                    try:
+                                        await observer
+                                    except asyncio.CancelledError:
+                                        pass
+                                    install_context.finish(install_code)
+                                installed[key] = install_code
+                            else:
+                                installed[key] = None
+                        if installed[key]:
+                            raise MigrationError(installed[key])
+                        await self._authorize(ctx)
+                        current = await self._rows(ctx, source['uuid'])
+                        if not current or _fingerprint(current[0]) != _fingerprint(source):
+                            raise MigrationError('preview_stale')
+                        plan, facts, pending, state, blockers = await self._plan(ctx, current[0])
+                        if state not in ('ready', 'activation_pending'):
+                            raise MigrationError(blockers[0]['code'] if blockers else 'migration_blocked')
+                        selection = {
+                            'pipeline_uuid': source['uuid'],
+                            'preview_token': self._token(ctx, current[0], plan, facts, pending),
+                        }
+                        task_context.metadata['phase'] = 'migrating'
+                        await self._run(ctx, execution, [selection], task_context, results=[result])
+                    except Exception as exc:
+                        result.update(
+                            state='blocked' if isinstance(exc, MigrationError) else 'failed',
+                            code=_failure_code(exc),
+                        )
+                    if any(
+                        r['code'] in ('operation_cancelled', 'commit_outcome_unknown')
+                        for r in task_context.metadata['results']
+                    ):
+                        break
+        finally:
+            for result in task_context.metadata['results']:
+                if result['state'] == 'pending':
+                    result.update(state='failed', code='operation_cancelled')
+            task_context.metadata['phase'] = 'finished'
+            self._all_tasks.discard(ctx.workspace_uuid)
+        return task_context.metadata
+
+    async def _verify_runtime(self, execution, plan):
+        runners = await self.ap.runner_registry.list_runners(execution, use_cache=False, usage='agent')
+        descriptor = next((r for r in runners if r.id == plan['target_runner_id']), None)
+        if descriptor is None or 'agent' not in descriptor.usages:
+            raise MigrationError('runner_unavailable')
+        if descriptor.plugin_version != plan['target_plugin']['version']:
+            raise MigrationError('plugin_version_incompatible')
+        config = plan['config']
+        RunnerConfigResolver.validate_pipeline_config(config)
+        runner_config = config['ai']['runner_config'][plan['target_runner_id']]
+        schema = {item['name']: item for item in descriptor.config_schema}
+        # Host fields have their own scope/capability checks below; plugin
+        # schema defaults must never implicitly authorize a foreign resource.
+        host_fields = {
+            'enable-all-tools': 'boolean',
+            'tools': 'array',
+            'knowledge-bases': 'knowledge-base-multi-selector',
+            'mcp-resources': 'array',
+            'mcp-resource-agent-read-enabled': 'boolean',
+        }
+        for name, value in runner_config.items():
+            field = schema.get(name)
+            if field is None and name in host_fields:
+                field = {'type': host_fields[name]}
+            if field is None:
+                raise MigrationError('runner_schema_incompatible')
+            if value is None and field.get('nullable') is True:
+                continue
+            kind = config_schema.normalize_schema_item_type(field.get('type'))
+            valid = (
+                (
+                    kind
+                    in (
+                        'string',
+                        'text',
+                        'password',
+                        'secret',
+                        'select',
+                        'llm-model-selector',
+                        'embedding-model-selector',
+                        'rerank-model-selector',
+                    )
+                    and isinstance(value, str)
+                )
+                or (kind == 'boolean' and type(value) is bool)
+                or (kind in ('integer', 'int') and type(value) is int)
+                or (kind in ('float', 'number') and type(value) in (int, float))
+                or (
+                    kind
+                    in (
+                        'array',
+                        'multi-select',
+                        'knowledge-base-selector',
+                        'knowledge-base-multi-selector',
+                        'prompt-editor',
+                    )
+                    and isinstance(value, list)
+                )
+                or (kind in ('object', 'json') and type(value) is dict)
+                or (kind == 'array[string]' and type(value) is list and all(type(v) is str for v in value))
+                or (
+                    kind == 'model-fallback-selector'
+                    and (
+                        isinstance(value, str)
+                        or isinstance(value, dict)
+                        and set(value) <= {'primary', 'fallbacks', 'reasoning'}
+                        and isinstance(value.get('primary'), str)
+                        and isinstance(value.get('fallbacks', []), list)
+                        and all(isinstance(v, str) for v in value.get('fallbacks', []))
+                        and isinstance(value.get('reasoning', {}), dict)
+                    )
+                )
+            )
+            if not valid:
+                raise MigrationError('runner_schema_incompatible')
+            if kind == 'select' and field.get('options') and field.get('allow_custom') is not True:
+                if value not in [o.get('value', o.get('name')) for o in field['options']]:
+                    raise MigrationError('runner_schema_incompatible')
+        if any(
+            f.get('required') and name not in runner_config and f.get('default') is None for name, f in schema.items()
+        ):
+            raise MigrationError('runner_schema_incompatible')
+        await self._verify_resources(execution, descriptor, runner_config)
+        return _json(
+            [
+                descriptor.id,
+                descriptor.plugin_version,
+                descriptor.usages,
+                descriptor.config_schema,
+                getattr(descriptor, 'capabilities', None),
+                getattr(descriptor, 'permissions', None),
+            ]
+        )
+
+    async def _verify_resources(self, execution, descriptor, runner_config):
+        """Fail closed on references the scoped Host cannot authorize.
+
+        This runs only on explicit execution, outside the mutation transaction.
+        Resource use is still reauthorized by the Host on every eventual run.
+        """
+
+        async def require_row(model, identifier, *, column=None):
+            column = model.uuid if column is None else column
+            found = (
+                await self.pm.execute_async(
+                    sa.select(model.uuid).where(
+                        model.workspace_uuid == execution.workspace_uuid,
+                        column == identifier,
+                    )
+                )
+            ).first()
+            if found is None:
+                raise MigrationError('runner_resource_unavailable')
+
+        permissions = getattr(descriptor, 'permissions', None)
+        for model_type, model_uuid in config_schema.iter_config_model_refs(descriptor, runner_config):
+            allowed = set(getattr(permissions, 'models', []))
+            if not (allowed & ({'rerank'} if model_type == 'rerank' else {'invoke', 'stream'})):
+                raise MigrationError('runner_resource_unavailable')
+            await require_row(RerankModel if model_type == 'rerank' else LLMModel, model_uuid)
+        for field in config_schema.iter_schema_items(descriptor, {'embedding-model-selector'}):
+            value = runner_config.get(field['name'])
+            if value and value not in config_schema.NONE_SENTINELS:
+                await require_row(EmbeddingModel, value)
+
+        kb_ids = runner_config.get('knowledge-bases', [])
+        if not isinstance(kb_ids, list) or any(not isinstance(v, str) or not v for v in kb_ids):
+            raise MigrationError('runner_resource_unavailable')
+        if kb_ids:
+            if not config_schema.uses_host_knowledge_bases(descriptor) or 'retrieve' not in getattr(
+                permissions, 'knowledge_bases', []
+            ):
+                raise MigrationError('runner_resource_unavailable')
+            for kb_id in kb_ids:
+                await require_row(KnowledgeBase, kb_id)
+
+        tool_names = runner_config.get('tools', [])
+        if not isinstance(tool_names, list) or any(not isinstance(v, str) or not v for v in tool_names):
+            raise MigrationError('runner_resource_unavailable')
+        tool_grant = (
+            runner_config.get('enable-all-tools', False)
+            or tool_names
+            or runner_config.get('mcp-resource-agent-read-enabled', False)
+        )
+        if tool_grant:
+            if not config_schema.uses_host_tools(descriptor) or 'call' not in getattr(permissions, 'tools', []):
+                raise MigrationError('runner_resource_unavailable')
+            policy = ResourcePolicyProjector.from_runner_config(runner_config)
+            tools = await AgentResourceBuilder(self.ap)._build_tools_from_binding(
+                execution,
+                permissions,
+                policy,
+                descriptor,
+                runner_config,
+            )
+            if not set(tool_names) <= {tool['tool_name'] for tool in tools}:
+                raise MigrationError('runner_resource_unavailable')
+
+        for attachment in runner_config.get('mcp-resources', []):
+            if not isinstance(attachment, dict):
+                raise MigrationError('runner_resource_unavailable')
+            identifier = attachment.get('server_uuid') or attachment.get('server_id')
+            name = attachment.get('server_name')
+            if identifier:
+                await require_row(MCPServer, identifier)
+            elif name:
+                await require_row(MCPServer, name, column=MCPServer.name)
+            else:
+                raise MigrationError('runner_resource_unavailable')
+
+    async def _lock(self, ctx, pipeline_uuid, session):
+        # Every pending producer shares this row and revalidates its captured
+        # configuration/conversation authority after acquiring admission.
+        if self.pm.get_db_engine().dialect.name not in {'sqlite', 'postgresql'}:
+            raise MigrationError('runtime.pending_interaction')
+        await lock_pipeline_admission(session, ctx.workspace_uuid, pipeline_uuid)
+        # Hold revocable directory rows through commit (no network in this UoW).
+        await session.execute(sa.select(User.uuid).where(User.uuid == ctx.account_uuid).with_for_update())
+        for model, condition in [
+            (Workspace, Workspace.uuid == ctx.workspace_uuid),
+            (WorkspaceExecutionState, WorkspaceExecutionState.workspace_uuid == ctx.workspace_uuid),
+            (
+                WorkspaceMembership,
+                sa.and_(
+                    WorkspaceMembership.workspace_uuid == ctx.workspace_uuid,
+                    WorkspaceMembership.account_uuid == ctx.account_uuid,
+                ),
+            ),
+            (PluginSetting, PluginSetting.workspace_uuid == ctx.workspace_uuid),
+        ]:
+            await session.execute(sa.select(model.__table__).where(condition).with_for_update())
+
+    async def _commit(self, ctx, selection, expected_facts, candidate, *, data_only=False):
+        async with self.pm.tenant_uow(ctx.workspace_uuid) as uow:
+            await self._lock(ctx, selection['pipeline_uuid'], uow.session)
+            await self._authorize(ctx, uow.session)
+            row, plan, facts, pending = await self._selected(ctx, selection, data_only=data_only)
+            if facts != expected_facts:
+                raise MigrationError('plugin_changed')
+            if pending:
+                return pending['uuid'], pending['target_fingerprint']
+            target = {**row, 'config': candidate['config']}
+            t = LegacyPipeline.__table__
+            cas = [t.c.workspace_uuid == ctx.workspace_uuid, t.c.uuid == row['uuid']]
+            for key in ('config', 'stages', 'extensions_preferences'):
+                cas.append(sa.cast(t.c[key], sa.Text) == row['_raw_' + key])
+            result = await self.pm.execute_async(sa.update(t).where(*cas).values(config=candidate['config']))
+            if result.rowcount != 1:
+                raise MigrationError('preview_stale')
+            snapshot_uuid = str(uuid.uuid4())
+            target_fingerprint = _fingerprint(target)
+            await self.pm.execute_async(
+                sa.insert(PipelineMigrationSnapshot).values(
+                    uuid=snapshot_uuid,
+                    workspace_uuid=ctx.workspace_uuid,
+                    pipeline_uuid=row['uuid'],
+                    source_fingerprint=_fingerprint(row),
+                    target_fingerprint=target_fingerprint,
+                    source_snapshot=json.loads(_json({k: v for k, v in row.items() if not k.startswith('_')})),
+                    planner_version=str(PLANNER_VERSION),
+                    state='activation_pending',
+                )
+            )
+            if self._native_interactions(ctx, row['uuid']):
+                raise MigrationError('runtime.pending_interaction')
+        return snapshot_uuid, target_fingerprint
+
+    async def _activate(self, ctx, selection, snapshot_uuid, target_fingerprint, candidate, plan, expected_facts):
+        # Re-read the committed target under a short lock before synchronous
+        # publication. A later editor must never be replaced by our old candidate.
+        async with self.pm.tenant_uow(ctx.workspace_uuid) as uow:
+            await self._lock(ctx, selection['pipeline_uuid'], uow.session)
+            await self._authorize(ctx, uow.session)
+            rows = await self._rows(ctx, selection['pipeline_uuid'])
+            if not rows or _fingerprint(rows[0]) != target_fingerprint:
+                raise MigrationError('activation_source_changed')
+            if await self._plugin(ctx, plan) != expected_facts:
+                raise MigrationError('plugin_changed')
+            _, interaction_blocked = await self._interaction_state(ctx, selection['pipeline_uuid'])
+            if interaction_blocked:
+                raise MigrationError('runtime.pending_interaction')
+            # Native cache inspection above is the last await before publication
+            # and session invalidation; no native producer can interleave here.
+            self.ap.pipeline_mgr.publish_pipeline(candidate)
+            for session in self.ap.sess_mgr.session_list:
+                conversation = session.using_conversation
+                if (
+                    conversation is not None
+                    and conversation.pipeline_uuid == selection['pipeline_uuid']
+                    and getattr(session, 'workspace_uuid', None) == ctx.workspace_uuid
+                ):
+                    session.using_conversation = None
+            t = PipelineMigrationSnapshot.__table__
+            await self.pm.execute_async(
+                sa.update(t)
+                .where(t.c.workspace_uuid == ctx.workspace_uuid, t.c.uuid == snapshot_uuid)
+                .values(state='active')
+            )
+
+    async def _run(self, ctx, execution, items, task_context, results=None, *, data_only=False):
+        async with self.pm.tenant_scope(execution.workspace_uuid):
+            for selection, result in zip(items, results if results is not None else task_context.metadata['results']):
+                committed = False
+                commit_attempted = False
+                try:
+                    await self._authorize(ctx)
+                    row, plan, facts, pending = await self._selected(ctx, selection, data_only=data_only)
+                    runtime_schema = None if data_only else await self._verify_runtime(execution, plan)
+                    RunnerConfigResolver.validate_pipeline_config(plan['config'])
+                    candidate_entity = {k: copy.deepcopy(v) for k, v in row.items() if not k.startswith('_')}
+                    candidate_entity['config'] = copy.deepcopy(plan['config'])
+                    runtime = await self.ap.pipeline_mgr.prepare_pipeline(execution, copy.deepcopy(candidate_entity))
+                    if not data_only and await self._verify_runtime(execution, plan) != runtime_schema:
+                        raise MigrationError('runner_schema_changed')
+                    # Runtime awaits are over. Recheck authorization/source/plugin
+                    # facts under database locks, then atomically journal and CAS.
+                    commit_attempted = True
+                    snapshot_uuid, target_fingerprint = await self._commit(
+                        ctx, selection, facts, candidate_entity, **({'data_only': True} if data_only else {})
+                    )
+                    committed = True
+                    await self._activate(ctx, selection, snapshot_uuid, target_fingerprint, runtime, plan, facts)
+                    result.update(state='migrated', code='data_only' if data_only else None)
+                except (Exception, asyncio.CancelledError) as exc:
+                    cancelled = isinstance(exc, asyncio.CancelledError)
+                    reconciliation_cancel = None
+                    code = 'operation_cancelled' if cancelled else _failure_code(exc)
+                    if commit_attempted and not committed and not isinstance(exc, MigrationError):
+                        # A commit can succeed while acknowledgement/connection
+                        # cleanup fails, including cancellation. Make one read
+                        # after the UoW has unwound, in this task's tenant scope;
+                        # never shield/retry through another cancellation.
+                        try:
+                            t = PipelineMigrationSnapshot.__table__
+                            conditions = [
+                                t.c.workspace_uuid == ctx.workspace_uuid,
+                                t.c.pipeline_uuid == selection['pipeline_uuid'],
+                                t.c.source_fingerprint
+                                == (pending['source_fingerprint'] if pending else _fingerprint(row)),
+                                t.c.target_fingerprint == _fingerprint(candidate_entity),
+                                t.c.planner_version == str(PLANNER_VERSION),
+                            ]
+                            if pending:
+                                # A retry starts from the target, not the legacy
+                                # source. Only the original snapshot proves this
+                                # durable migration; an unrelated journal cannot.
+                                conditions.extend(
+                                    [
+                                        t.c.uuid == pending['uuid'],
+                                        t.c.target_fingerprint == pending['target_fingerprint'],
+                                    ]
+                                )
+                            found = (await self.pm.execute_async(sa.select(t.c.uuid).where(*conditions))).first()
+                            committed = found is not None
+                        except asyncio.CancelledError as cancel_exc:
+                            reconciliation_cancel = cancel_exc
+                            cancelled = True
+                            code = 'commit_outcome_unknown'
+                        except Exception:
+                            code = 'commit_outcome_unknown'
+                    state = (
+                        # Unknown is deliberately recovery-needed, not a claim
+                        # of rollback or proof that a snapshot was committed.
+                        'activation_pending'
+                        if committed or code == 'commit_outcome_unknown'
+                        else 'stale'
+                        if code == 'preview_stale'
+                        else 'blocked'
+                        if isinstance(exc, MigrationError)
+                        else 'failed'
+                    )
+                    result.update(state=state, code=code)
+                    if cancelled:
+                        for unstarted in task_context.metadata['results']:
+                            if unstarted['state'] == 'pending':
+                                unstarted.update(state='failed', code='operation_cancelled')
+                        if reconciliation_cancel is not None:
+                            # Preserve cancellation propagation during recovery,
+                            # but record safe outcomes before leaving the task.
+                            raise reconciliation_cancel
+                        return task_context.metadata
+                    # Never attach the original exception, traceback or plugin
+                    # diagnostic to TaskContext: upstream errors can contain keys.
+            return task_context.metadata

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

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 import uuid
 import traceback
 
@@ -7,8 +8,10 @@ import sqlalchemy
 
 from ....cloud.model_catalog import LANGBOT_MODELS_PROVIDER_REQUESTER
 from ....core import app
+from ....core.task_boundary import create_detached_task
 from ....entity.persistence import model as persistence_model
 from ....workspace.errors import WorkspaceNotFoundError
+from ....provider.modelmgr.codex_auth import CodexAuth, REQUESTER as CODEX_REQUESTER, validate_config
 from .secrets import contains_secret_placeholder, redact_secrets, restore_secret_placeholders
 from .tenant import TenantContext, require_workspace_uuid, scope_statement
 
@@ -20,6 +23,8 @@ class ModelProviderService:
 
     def __init__(self, ap: app.Application) -> None:
         self.ap = ap
+        self.codex_auth = CodexAuth(ap)
+        self._deletion_tasks: set[asyncio.Task[None]] = set()
 
     def _is_cloud_runtime(self) -> bool:
         mode = getattr(self.ap.persistence_mgr, 'mode', None)
@@ -116,14 +121,30 @@ class ModelProviderService:
         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')
+        validate_config(provider_data)
         provider_data['uuid'] = str(uuid.uuid4())
         provider_data['workspace_uuid'] = require_workspace_uuid(context)
         provider_data['api_keys'] = self._normalize_api_keys(
             restore_secret_placeholders(provider_data.get('api_keys'), sensitive=True)
         )
-        await self.ap.persistence_mgr.execute_async(
-            sqlalchemy.insert(persistence_model.ModelProvider).values(**provider_data)
-        )
+        if provider_data.get('requester') == CODEX_REQUESTER:
+            async with self.ap.persistence_mgr.tenant_uow(provider_data['workspace_uuid']):
+                await self.ap.persistence_mgr.execute_async(
+                    sqlalchemy.insert(persistence_model.ModelProvider).values(**provider_data)
+                )
+                await self.ap.persistence_mgr.execute_async(
+                    sqlalchemy.insert(persistence_model.CodexCredential).values(
+                        workspace_uuid=provider_data['workspace_uuid'],
+                        provider_uuid=provider_data['uuid'],
+                        payload={},
+                        version=0,
+                        lease_until=0,
+                    )
+                )
+        else:
+            await self.ap.persistence_mgr.execute_async(
+                sqlalchemy.insert(persistence_model.ModelProvider).values(**provider_data)
+            )
 
         # load to runtime
         runtime_provider = await self.ap.model_mgr.load_provider(context, provider_data)
@@ -138,6 +159,17 @@ class ModelProviderService:
             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 {'requester', 'base_url', 'api_keys'} & provider_data.keys():
+            current = await self.get_provider(context, provider_uuid, include_secret=True)
+            if current is None:
+                raise WorkspaceNotFoundError('Provider not found')
+            if CODEX_REQUESTER in (current.get('requester'), provider_data.get('requester')):
+                if provider_data.get('requester', current.get('requester')) != current.get('requester'):
+                    raise ValueError('Create a separate provider to change the ChatGPT authentication type')
+                merged = {**current, **provider_data}
+                validate_config(merged)
+                provider_data['base_url'] = merged['base_url']
+                provider_data['api_keys'] = []
         if 'api_keys' in provider_data:
             submitted_keys = provider_data.get('api_keys')
             if contains_secret_placeholder(submitted_keys, sensitive=True):
@@ -163,60 +195,107 @@ class ModelProviderService:
             raise WorkspaceNotFoundError('Provider not found')
         await self.ap.model_mgr.reload_provider(context, provider_uuid)
 
-    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)
+    async def delete_provider(self, context: TenantContext, provider_uuid: str, cascade: bool = False) -> None:
+        """Delete a provider, optionally deleting all its Workspace-scoped models."""
         workspace_uuid = require_workspace_uuid(context)
-        # Check if any models use this provider
-        llm_result = await self.ap.persistence_mgr.execute_async(
-            scope_statement(
-                sqlalchemy.select(persistence_model.LLMModel).where(
-                    persistence_model.LLMModel.provider_uuid == provider_uuid
-                ),
-                persistence_model.LLMModel,
-                workspace_uuid,
-            )
+        persistence = self.ap.persistence_mgr
+        model_types = (
+            (persistence_model.LLMModel, 'LLM', 'remove_llm_model'),
+            (persistence_model.EmbeddingModel, 'Embedding', 'remove_embedding_model'),
+            (persistence_model.RerankModel, 'Rerank', 'remove_rerank_model'),
         )
-        if llm_result.first() is not None:
-            raise ValueError('Cannot delete provider: LLM models still reference it')
-
-        embedding_result = await self.ap.persistence_mgr.execute_async(
-            scope_statement(
-                sqlalchemy.select(persistence_model.EmbeddingModel).where(
-                    persistence_model.EmbeddingModel.provider_uuid == provider_uuid
-                ),
-                persistence_model.EmbeddingModel,
-                workspace_uuid,
+        deleted_models: list[tuple[str, list[str]]] = []
+        async with persistence.tenant_uow(workspace_uuid):
+            # Check ownership before touching children. Lock the provider on PostgreSQL
+            # so concurrent model inserts cannot race the reference check/deletion.
+            provider_result = await persistence.execute_async(
+                scope_statement(
+                    sqlalchemy.select(persistence_model.ModelProvider.requester)
+                    .where(persistence_model.ModelProvider.uuid == provider_uuid)
+                    .with_for_update(),
+                    persistence_model.ModelProvider,
+                    workspace_uuid,
+                )
             )
-        )
-        if embedding_result.first() is not None:
-            raise ValueError('Cannot delete provider: Embedding models still reference it')
+            provider = provider_result.first()
+            if provider is None:
+                raise WorkspaceNotFoundError('Provider not found')
+            if self._system_requester_is_reserved(provider.requester):
+                raise ValueError('LangBot Models is managed by Cloud and cannot be modified')
 
-        rerank_result = await self.ap.persistence_mgr.execute_async(
-            scope_statement(
-                sqlalchemy.select(persistence_model.RerankModel).where(
-                    persistence_model.RerankModel.provider_uuid == provider_uuid
-                ),
-                persistence_model.RerankModel,
-                workspace_uuid,
+            for model_type, label, remover in model_types:
+                result = await persistence.execute_async(
+                    scope_statement(
+                        sqlalchemy.select(model_type.uuid).where(model_type.provider_uuid == provider_uuid),
+                        model_type,
+                        workspace_uuid,
+                    )
+                )
+                model_uuids = list(result.scalars())
+                if model_uuids and not cascade:
+                    raise ValueError(f'Cannot delete provider: {label} models still reference it')
+                if model_uuids:
+                    # Model services have no pipeline/KB deletion side effects: they
+                    # delete the scoped row and evict its runtime cache. Defer eviction
+                    # here rather than calling those services before our commit.
+                    await persistence.execute_async(
+                        scope_statement(
+                            sqlalchemy.delete(model_type).where(model_type.provider_uuid == provider_uuid),
+                            model_type,
+                            workspace_uuid,
+                        )
+                    )
+                    deleted_models.append((remover, model_uuids))
+
+            # Explicit cleanup also works on legacy SQLite connections without FK
+            # enforcement; never load or serialize the private credential payload.
+            await persistence.execute_async(
+                scope_statement(
+                    sqlalchemy.delete(persistence_model.CodexCredential).where(
+                        persistence_model.CodexCredential.provider_uuid == provider_uuid
+                    ),
+                    persistence_model.CodexCredential,
+                    workspace_uuid,
+                )
             )
-        )
-        if rerank_result.first() is not None:
-            raise ValueError('Cannot delete provider: Rerank models still reference it')
-
-        result = await self.ap.persistence_mgr.execute_async(
-            scope_statement(
-                sqlalchemy.delete(persistence_model.ModelProvider).where(
-                    persistence_model.ModelProvider.uuid == provider_uuid
-                ),
-                persistence_model.ModelProvider,
-                workspace_uuid,
+            result = await persistence.execute_async(
+                scope_statement(
+                    sqlalchemy.delete(persistence_model.ModelProvider).where(
+                        persistence_model.ModelProvider.uuid == provider_uuid
+                    ),
+                    persistence_model.ModelProvider,
+                    workspace_uuid,
+                )
             )
-        )
-        if getattr(result, 'rowcount', None) == 0:
-            raise WorkspaceNotFoundError('Provider not found')
+            if result.rowcount == 0:
+                raise WorkspaceNotFoundError('Provider not found')
 
-        await self.ap.model_mgr.remove_provider(context, provider_uuid)
+        async def remove_runtime() -> None:
+            async with persistence.tenant_scope(workspace_uuid):
+                for remover, model_uuids in deleted_models:
+                    for model_uuid in model_uuids:
+                        await getattr(self.ap.model_mgr, remover)(context, model_uuid)
+                # This also closes the requester's HTTP client; models go first.
+                await self.ap.model_mgr.remove_provider(context, provider_uuid)
+
+        if persistence.current_session() is None:
+            await remove_runtime()
+        else:
+            # A nested UoW has not committed yet. Reuse the rollback-cancelled gate
+            # and detached context boundary instead of evicting uncommitted data.
+            task = create_detached_task(
+                remove_runtime(),
+                after_commit_manager=persistence,
+                workspace_uuid=workspace_uuid,
+            )
+            self._deletion_tasks.add(task)
+
+            def completed(task: asyncio.Task[None]) -> None:
+                self._deletion_tasks.discard(task)
+                if not task.cancelled() and task.exception() is not None:
+                    self.ap.logger.error('Failed to remove deleted provider runtime', exc_info=task.exception())
+
+            task.add_done_callback(completed)
 
     async def get_provider_model_counts(self, context: TenantContext, provider_uuid: str) -> dict:
         """Get count of models using this provider"""

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

@@ -119,7 +119,7 @@ class SpaceService:
 
         space_config = self._get_space_config()
         authorize_url = space_config['oauth_authorize_url']
-        params = {'redirect_uri': redirect_uri}
+        params = {'redirect_uri': redirect_uri, 'code_contract': 'redirect-v1'}
         if state:
             params['state'] = state
         return f'{authorize_url}?{urlencode(params)}'
@@ -129,6 +129,8 @@ class SpaceService:
         code: str,
         workspace_uuids: list[str] | None = None,
         workspace_created_ats: dict[str, int] | None = None,
+        *,
+        redirect_uri: str = '',
     ) -> typing.Dict:
         """Exchange OAuth authorization code for tokens"""
         from langbot.pkg.utils import constants
@@ -141,6 +143,7 @@ class SpaceService:
             f'{space_url}/api/v1/accounts/oauth/token',
             json={
                 'code': code,
+                'redirect_uri': redirect_uri,
                 'instance_id': constants.instance_id,
                 # Sending an explicit empty list tells new Space servers not to
                 # synthesize a legacy instance-derived Workspace binding.

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

@@ -4,6 +4,7 @@ import sqlalchemy
 import argon2
 import jwt
 import datetime
+import json
 import typing
 import asyncio
 import dataclasses
@@ -12,10 +13,19 @@ import hashlib
 import secrets
 import time
 import uuid
+import webauthn
+from webauthn.helpers import bytes_to_base64url, base64url_to_bytes
+from webauthn.helpers.structs import (
+    AuthenticatorSelectionCriteria,
+    PublicKeyCredentialDescriptor,
+    ResidentKeyRequirement,
+    UserVerificationRequirement,
+)
 
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
 from ....entity.persistence import user
+from ....entity.persistence import passkey
 from ....entity.persistence.workspace import MembershipRole, MembershipStatus, WorkspaceMembership
 from ....utils import constants
 from ....entity.errors import account as account_errors
@@ -29,6 +39,9 @@ if typing.TYPE_CHECKING:
 _SPACE_OAUTH_STATE_MAX_ENTRIES = 4096
 _SPACE_OAUTH_STATE_HEAP_COMPACT_FLOOR = 64
 _SPACE_OAUTH_STATE_HEAP_MAX_MULTIPLIER = 4
+_PASSKEY_CHALLENGE_MAX_ENTRIES = 4096
+_PASSKEY_CHALLENGE_HEAP_COMPACT_FLOOR = 64
+_PASSKEY_CHALLENGE_HEAP_MAX_MULTIPLIER = 4
 
 
 class AccountExistsLoginRequiredError(ValueError):
@@ -54,6 +67,17 @@ class SpaceOAuthStateConsumption:
     launch_workspace_uuid: str | None = None
 
 
+@dataclasses.dataclass(frozen=True, slots=True)
+class PasskeyChallengeData:
+    challenge: bytes
+    purpose: typing.Literal['register', 'auth']
+    rp_id: str
+    origin: str
+    expires_at: float
+    account_uuid: str | None = None
+    user_email: str | None = None
+
+
 class UserService:
     ap: Application
     _create_user_lock: asyncio.Lock
@@ -65,6 +89,9 @@ class UserService:
         self._space_oauth_state_lock = asyncio.Lock()
         self._space_oauth_states: dict[str, tuple[str, str | None, float, str | None]] = {}
         self._space_oauth_state_expiry_heap: list[tuple[float, str]] = []
+        self._passkey_challenge_lock = asyncio.Lock()
+        self._passkey_challenges: dict[str, PasskeyChallengeData] = {}
+        self._passkey_challenge_expiry_heap: list[tuple[float, str]] = []
 
     @staticmethod
     def _space_oauth_state_digest(state: str) -> str:
@@ -774,7 +801,7 @@ class UserService:
             f'email:{normalized_email}',
         )
 
-    async def bind_space_account(self, user_email: str, code: str) -> user.User:
+    async def bind_space_account(self, user_email: str, code: str, *, redirect_uri: str = '') -> user.User:
         """Bind Space account to existing local account"""
         local_account = await self.get_user_by_email(user_email)
         if local_account is None:
@@ -794,12 +821,13 @@ class UserService:
                 code,
                 [binding.workspace_uuid],
                 {binding.workspace_uuid: created_ts},
+                redirect_uri=redirect_uri,
             )
         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)
+            token_data = await self.ap.space_service.exchange_oauth_code(code, redirect_uri=redirect_uri)
         access_token = token_data.get('access_token')
         refresh_token = token_data.get('refresh_token')
         expires_in = token_data.get('expires_in', 0)
@@ -849,3 +877,309 @@ class UserService:
         await self._update_space_provider_for_account(local_account, api_key)
 
         return await self.get_user_by_email(space_email)
+
+    def _prune_passkey_challenges(self, now: float) -> None:
+        while self._passkey_challenge_expiry_heap:
+            expires_at, token = self._passkey_challenge_expiry_heap[0]
+            entry = self._passkey_challenges.get(token)
+            if entry is None or entry.expires_at != expires_at:
+                heapq.heappop(self._passkey_challenge_expiry_heap)
+                continue
+            if expires_at > now:
+                break
+            heapq.heappop(self._passkey_challenge_expiry_heap)
+            self._passkey_challenges.pop(token, None)
+
+        max_heap_entries = max(
+            _PASSKEY_CHALLENGE_HEAP_COMPACT_FLOOR,
+            len(self._passkey_challenges) * _PASSKEY_CHALLENGE_HEAP_MAX_MULTIPLIER,
+        )
+        if len(self._passkey_challenge_expiry_heap) > max_heap_entries:
+            self._passkey_challenge_expiry_heap[:] = [
+                (entry.expires_at, token) for token, entry in self._passkey_challenges.items()
+            ]
+            heapq.heapify(self._passkey_challenge_expiry_heap)
+
+    async def issue_passkey_challenge(
+        self,
+        purpose: typing.Literal['register', 'auth'],
+        rp_id: str,
+        origin: str,
+        *,
+        account_uuid: str | None = None,
+        user_email: str | None = None,
+        ttl_seconds: int = 300,
+    ) -> tuple[str, bytes]:
+        now = time.monotonic()
+        challenge_bytes = secrets.token_bytes(32)
+        challenge_token = secrets.token_urlsafe(32)
+        expires_at = now + ttl_seconds
+
+        async with self._passkey_challenge_lock:
+            self._prune_passkey_challenges(now)
+            while len(self._passkey_challenges) >= _PASSKEY_CHALLENGE_MAX_ENTRIES:
+                if not self._passkey_challenge_expiry_heap:
+                    break
+                _, oldest_token = heapq.heappop(self._passkey_challenge_expiry_heap)
+                self._passkey_challenges.pop(oldest_token, None)
+
+            self._passkey_challenges[challenge_token] = PasskeyChallengeData(
+                challenge=challenge_bytes,
+                purpose=purpose,
+                rp_id=rp_id,
+                origin=origin,
+                expires_at=expires_at,
+                account_uuid=account_uuid,
+                user_email=user_email,
+            )
+            heapq.heappush(self._passkey_challenge_expiry_heap, (expires_at, challenge_token))
+
+        return challenge_token, challenge_bytes
+
+    async def consume_passkey_challenge(
+        self,
+        challenge_token: str,
+        purpose: typing.Literal['register', 'auth'],
+    ) -> PasskeyChallengeData:
+        now = time.monotonic()
+        async with self._passkey_challenge_lock:
+            self._prune_passkey_challenges(now)
+            data = self._passkey_challenges.pop(challenge_token, None)
+
+        if data is None or data.expires_at < now:
+            raise ValueError('Invalid or expired passkey challenge')
+        if data.purpose != purpose:
+            raise ValueError('Passkey challenge purpose mismatch')
+        return data
+
+    async def get_user_passkeys(self, account_uuid: str) -> list[passkey.PasskeyCredential]:
+        statement = (
+            sqlalchemy.select(passkey.PasskeyCredential)
+            .where(passkey.PasskeyCredential.account_uuid == account_uuid)
+            .order_by(passkey.PasskeyCredential.created_at.desc())
+        )
+        async with self._session_factory()() as session:
+            result = await session.scalars(statement)
+            return list(result.all())
+
+    async def get_passkey_by_credential_id(self, credential_id: str) -> passkey.PasskeyCredential | None:
+        statement = sqlalchemy.select(passkey.PasskeyCredential).where(
+            passkey.PasskeyCredential.credential_id == credential_id
+        )
+        async with self._session_factory()() as session:
+            return await session.scalar(statement)
+
+    async def get_passkey_by_uuid(self, passkey_uuid: str) -> passkey.PasskeyCredential | None:
+        statement = sqlalchemy.select(passkey.PasskeyCredential).where(passkey.PasskeyCredential.uuid == passkey_uuid)
+        async with self._session_factory()() as session:
+            return await session.scalar(statement)
+
+    async def generate_passkey_registration_options(
+        self,
+        account_uuid: str,
+        rp_id: str,
+        origin: str,
+        rp_name: str = 'LangBot',
+    ) -> tuple[dict[str, typing.Any], str]:
+        account = await self.get_user_by_uuid(account_uuid)
+        if account is None:
+            raise ValueError('User not found')
+        self._require_active_account(account)
+
+        challenge_token, challenge_bytes = await self.issue_passkey_challenge(
+            purpose='register',
+            rp_id=rp_id,
+            origin=origin,
+            account_uuid=account_uuid,
+            user_email=account.user,
+        )
+
+        existing_passkeys = await self.get_user_passkeys(account_uuid)
+        exclude_credentials = [
+            PublicKeyCredentialDescriptor(id=base64url_to_bytes(pk.credential_id)) for pk in existing_passkeys
+        ]
+
+        options = webauthn.generate_registration_options(
+            rp_id=rp_id,
+            rp_name=rp_name,
+            user_name=account.user,
+            user_id=account.uuid.encode('utf-8'),
+            user_display_name=account.user,
+            challenge=challenge_bytes,
+            exclude_credentials=exclude_credentials or None,
+            authenticator_selection=AuthenticatorSelectionCriteria(
+                resident_key=ResidentKeyRequirement.PREFERRED,
+            ),
+        )
+
+        options_dict = json.loads(webauthn.options_to_json(options))
+        return options_dict, challenge_token
+
+    async def verify_and_save_passkey_registration(
+        self,
+        challenge_token: str,
+        credential_data: dict[str, typing.Any] | str,
+        name: str | None = None,
+    ) -> passkey.PasskeyCredential:
+        challenge_data = await self.consume_passkey_challenge(challenge_token, 'register')
+        if not challenge_data.account_uuid:
+            raise ValueError('Registration challenge must be bound to an account')
+
+        verification = webauthn.verify_registration_response(
+            credential=credential_data,
+            expected_challenge=challenge_data.challenge,
+            expected_rp_id=challenge_data.rp_id,
+            expected_origin=challenge_data.origin,
+            require_user_verification=False,
+        )
+
+        cred_id_str = bytes_to_base64url(verification.credential_id)
+        pub_key_str = bytes_to_base64url(verification.credential_public_key)
+
+        transports = None
+        if isinstance(credential_data, dict):
+            resp = credential_data.get('response', {})
+            if isinstance(resp, dict) and 'transports' in resp:
+                t_list = resp.get('transports')
+                if isinstance(t_list, list):
+                    transports = ','.join(str(x) for x in t_list)
+
+        credential_name = (name or '').strip()
+        if not credential_name:
+            credential_name = f'Passkey ({datetime.datetime.now().strftime("%Y-%m-%d %H:%M")})'
+
+        record = passkey.PasskeyCredential(
+            uuid=str(uuid.uuid4()),
+            account_uuid=challenge_data.account_uuid,
+            name=credential_name,
+            credential_id=cred_id_str,
+            public_key=pub_key_str,
+            sign_count=verification.sign_count,
+            aaguid=verification.aaguid,
+            transports=transports,
+            backed_up=verification.credential_backed_up,
+        )
+
+        async with self._session_factory()() as session:
+            async with session.begin():
+                session.add(record)
+                await session.flush()
+                await session.refresh(record)
+                return record
+
+    async def generate_passkey_authentication_options(
+        self,
+        rp_id: str,
+        origin: str,
+        email: str | None = None,
+    ) -> tuple[dict[str, typing.Any], str]:
+        challenge_token, challenge_bytes = await self.issue_passkey_challenge(
+            purpose='auth',
+            rp_id=rp_id,
+            origin=origin,
+            user_email=email,
+        )
+
+        allow_credentials: list[PublicKeyCredentialDescriptor] | None = None
+        if email:
+            user_obj = await self.get_user_by_email(email)
+            if user_obj:
+                user_passkeys = await self.get_user_passkeys(user_obj.uuid)
+                if user_passkeys:
+                    allow_credentials = [
+                        PublicKeyCredentialDescriptor(id=base64url_to_bytes(pk.credential_id)) for pk in user_passkeys
+                    ]
+
+        options = webauthn.generate_authentication_options(
+            rp_id=rp_id,
+            challenge=challenge_bytes,
+            allow_credentials=allow_credentials or None,
+            user_verification=UserVerificationRequirement.PREFERRED,
+        )
+
+        options_dict = json.loads(webauthn.options_to_json(options))
+        return options_dict, challenge_token
+
+    async def verify_passkey_authentication(
+        self,
+        challenge_token: str,
+        credential_data: dict[str, typing.Any] | str,
+    ) -> tuple[str, user.User]:
+        challenge_data = await self.consume_passkey_challenge(challenge_token, 'auth')
+
+        raw_id = credential_data.get('id') if isinstance(credential_data, dict) else None
+        if not raw_id:
+            raise ValueError('Missing credential id')
+
+        stored_credential = await self.get_passkey_by_credential_id(raw_id)
+        if stored_credential is None:
+            raise ValueError('Passkey credential not recognized')
+
+        user_obj = await self.get_user_by_uuid(stored_credential.account_uuid)
+        if user_obj is None:
+            raise ValueError('Associated user not found')
+        self._require_active_account(user_obj)
+
+        verification = webauthn.verify_authentication_response(
+            credential=credential_data,
+            expected_challenge=challenge_data.challenge,
+            expected_rp_id=challenge_data.rp_id,
+            expected_origin=challenge_data.origin,
+            credential_public_key=base64url_to_bytes(stored_credential.public_key),
+            credential_current_sign_count=stored_credential.sign_count,
+            require_user_verification=False,
+        )
+
+        async with self._session_factory()() as session:
+            async with session.begin():
+                record = await session.scalar(
+                    sqlalchemy.select(passkey.PasskeyCredential).where(
+                        passkey.PasskeyCredential.id == stored_credential.id
+                    )
+                )
+                if record:
+                    record.sign_count = verification.new_sign_count
+                    record.last_used_at = datetime.datetime.now()
+                    record.backed_up = verification.credential_backed_up
+
+        token = await self.generate_jwt_token(user_obj)
+        return token, user_obj
+
+    async def rename_user_passkey(
+        self,
+        account_uuid: str,
+        passkey_uuid: str,
+        new_name: str,
+    ) -> passkey.PasskeyCredential | None:
+        async with self._session_factory()() as session:
+            async with session.begin():
+                record = await session.scalar(
+                    sqlalchemy.select(passkey.PasskeyCredential).where(
+                        passkey.PasskeyCredential.uuid == passkey_uuid,
+                        passkey.PasskeyCredential.account_uuid == account_uuid,
+                    )
+                )
+                if record is None:
+                    return None
+                record.name = new_name
+                await session.flush()
+                await session.refresh(record)
+                return record
+
+    async def delete_user_passkey(
+        self,
+        account_uuid: str,
+        passkey_uuid: str,
+    ) -> bool:
+        async with self._session_factory()() as session:
+            async with session.begin():
+                record = await session.scalar(
+                    sqlalchemy.select(passkey.PasskeyCredential).where(
+                        passkey.PasskeyCredential.uuid == passkey_uuid,
+                        passkey.PasskeyCredential.account_uuid == account_uuid,
+                    )
+                )
+                if record is None:
+                    return False
+                await session.delete(record)
+                return True

src/langbot/pkg/api/mcp/server.py

@@ -227,14 +227,16 @@ class LangBotMCPServer:
             return _dump(await ap.agent_service.get_agent_metadata(context))
 
         @mcp.tool(
-            description='List one Event processor instance run history; use before_id to page older runs. '
+            description='List one Agent or plugin processor run history; use before_id to page older runs. '
             'created_at_ms, started_at_ms and finished_at_ms are Host lifecycle times in epoch milliseconds.'
         )
         async def list_processor_runs(processor_uuid: str, before_id: int | None = None) -> str:
             context = _authorized(Permission.RESOURCE_VIEW)
             return _dump(await ap.agent_service.get_processor_runs(context, processor_uuid, before_id=before_id))
 
-        @mcp.tool(description='Read logs and action results for an Event processor run; page using after_sequence.')
+        @mcp.tool(
+            description='Read logs and action results for an Agent or plugin processor run; page using after_sequence.'
+        )
         async def get_processor_run_events(
             processor_uuid: str,
             run_id: str,