diff --git a/.github/workflows/build-dev-image.yaml b/.github/workflows/build-dev-image.yaml index 062a72d01..623a9bf99 100644 --- a/.github/workflows/build-dev-image.yaml +++ b/.github/workflows/build-dev-image.yaml @@ -7,23 +7,42 @@ on: jobs: build-dev-image: runs-on: ubuntu-latest - # 如果是tag则跳过 if: ${{ !startsWith(github.ref, 'refs/tags/') }} + permissions: + contents: read steps: - name: Checkout - uses: actions/checkout@v2 + uses: actions/checkout@v4 with: persist-credentials: false - - name: Generate Tag - id: generate_tag + - name: Set up Docker Buildx + uses: docker/setup-buildx-action@v3 + + - name: Generate image metadata + id: image + shell: bash run: | - # 获取分支名称，把/替换为- - echo ${{ github.ref }} | sed 's/refs\/heads\///g' | sed 's/\//-/g' - echo ::set-output name=tag::$(echo ${{ github.ref }} | sed 's/refs\/heads\///g' | sed 's/\//-/g') - - name: Login to Registry - run: docker login --username=${{ secrets.DOCKER_USERNAME }} --password ${{ secrets.DOCKER_PASSWORD }} - - name: Build Docker Image - run: | - docker buildx create --name mybuilder --use - docker build -t rockchin/langbot:${{ steps.generate_tag.outputs.tag }} . --push + set -euo pipefail + branch_tag="${GITHUB_REF#refs/heads/}" + branch_tag="${branch_tag//\//-}" + echo "branch_tag=${branch_tag}" >> "$GITHUB_OUTPUT" + echo "sha_tag=sha-${GITHUB_SHA}" >> "$GITHUB_OUTPUT" + + - name: Login to Docker Hub + uses: docker/login-action@v3 + with: + username: ${{ secrets.DOCKER_USERNAME }} + password: ${{ secrets.DOCKER_PASSWORD }} + + - name: Build and push immutable Core image + uses: docker/build-push-action@v6 + with: + context: . + push: true + tags: | + rockchin/langbot:${{ steps.image.outputs.branch_tag }} + rockchin/langbot:${{ steps.image.outputs.sha_tag }} + labels: | + org.opencontainers.image.revision=${{ github.sha }} + org.opencontainers.image.source=${{ github.server_url }}/${{ github.repository }} diff --git a/Dockerfile b/Dockerfile index c8569fb95..1d1351c72 100644 --- a/Dockerfile +++ b/Dockerfile @@ -62,7 +62,7 @@ RUN apt-get update \ && apt-get install -y --no-install-recommends nodejs \ && rm -f /tmp/nodesource_setup.sh \ && python -m pip install --no-cache-dir uv \ - && uv sync \ + && uv sync --extra seekdb \ && apt-get purge -y --auto-remove curl git gnupg \ && rm -rf /var/lib/apt/lists/* \ && touch /.dockerenv diff --git a/docs/PYPI_INSTALLATION.md b/docs/PYPI_INSTALLATION.md index 1144d5cb3..6b55aed59 100644 --- a/docs/PYPI_INSTALLATION.md +++ b/docs/PYPI_INSTALLATION.md @@ -10,6 +10,19 @@ uvx langbot This will automatically download and run the latest version of LangBot. +SeekDB support is optional and is not installed by the command above. If you +want to use the SeekDB vector database or the built-in SeekDB embedding model, +run LangBot with the `seekdb` extra: + +```bash +uvx --from 'langbot[seekdb]@latest' langbot +``` + +The extra includes native dependencies whose supported operating systems may +be narrower than LangBot's. In particular, the current Apple Silicon wheels +require macOS 15 or later. The default Chroma backend does not have this +requirement. + ## Install with pip/uv You can also install LangBot as a regular Python package: @@ -20,6 +33,10 @@ pip install langbot # Using uv uv pip install langbot + +# Include optional SeekDB support +pip install 'langbot[seekdb]' +# or: uv pip install 'langbot[seekdb]' ``` Then run it: @@ -101,7 +118,7 @@ uvx langbot ## System Requirements -- Python 3.10.1 or higher +- Python 3.11 or higher (lower than Python 4) - Operating System: Linux, macOS, or Windows ## Differences from Source Installation diff --git a/docs/SEEKDB_INTEGRATION.md b/docs/SEEKDB_INTEGRATION.md index b5ae7f9c3..a38eb9f08 100644 --- a/docs/SEEKDB_INTEGRATION.md +++ b/docs/SEEKDB_INTEGRATION.md @@ -16,12 +16,20 @@ This document describes how to use OceanBase SeekDB as the vector database backe ## Installation -SeekDB support is automatically included when you install LangBot. The required dependency `pyseekdb` is listed in `pyproject.toml`. +SeekDB is an optional LangBot feature. A normal LangBot installation uses +Chroma by default and does not install `pyseekdb` or its native bindings. -If you need to install it manually: +Choose the command that matches how you run LangBot: ```bash -pip install pyseekdb +# PyPI / uvx +uvx --from 'langbot[seekdb]@latest' langbot + +# Installed package +pip install 'langbot[seekdb]' + +# Source checkout +uv sync --extra seekdb ``` ## ⚠️ Platform Compatibility @@ -30,31 +38,36 @@ pip install pyseekdb | Platform | Status | Notes | |----------|--------|-------| -| Linux | ✅ Supported | Full embedded mode support via `pylibseekdb` | -| macOS | ❌ Not Supported | `pylibseekdb` is Linux-only; use server mode instead | -| Windows | ❌ Not Supported | `pylibseekdb` is Linux-only; use server mode instead | +| Linux x86_64 / ARM64 | ✅ Supported | Full embedded mode support via `pylibseekdb` | +| macOS 15+ on Apple Silicon | ✅ Supported | Requires the macOS ARM64 `pylibseekdb` wheel | +| macOS 14 or earlier on Apple Silicon | ❌ Not currently supported | The published native wheel requires macOS 15+; follow [oceanbase/seekdb#1324](https://github.com/oceanbase/seekdb/issues/1324) | +| macOS on Intel | ❌ Not currently supported | No embedded binding is selected by `pyseekdb` | +| Windows | ❌ Not currently supported | No Windows `pylibseekdb` wheel is published | -**Important**: Embedded mode requires the `pylibseekdb` library, which is only available on Linux. If you're on macOS or Windows, you must use server mode. +**Important**: Embedded mode requires a compatible `pylibseekdb` wheel. Do not +force-install or retag a wheel built for a newer macOS release: the bundled +binaries also declare macOS 15 as their minimum deployment target. ### Server Mode (Docker) | Platform | Status | Notes | |----------|--------|-------| | Linux | ✅ Supported | Full Docker support | -| macOS | ⚠️ Known Issue | Docker container initialization failure - [See Issue #36](https://github.com/oceanbase/seekdb/issues/36) | -| Windows | ⚠️ Untested | Should work but not yet tested | - -**macOS Users**: Currently, SeekDB Docker containers have an initialization issue on macOS ([oceanbase/seekdb#36](https://github.com/oceanbase/seekdb/issues/36)). Until this is resolved, we recommend: -- Using ChromaDB or Qdrant as alternatives -- Connecting to a remote SeekDB server on Linux if available +| macOS | ✅ Supported by Docker Desktop | The previous slow-disk startup issue was fixed upstream in [oceanbase/seekdb#36](https://github.com/oceanbase/seekdb/issues/36) | +| Windows | ⚠️ Depends on the container runtime | Use a Linux container and follow the upstream image documentation | ### Server Mode (Remote Connection) | Platform | Status | Notes | |----------|--------|-------| -| All Platforms | ✅ Supported | Connect to SeekDB running on a remote Linux server | +| Linux | ✅ Supported | Install the `seekdb` extra and connect to the remote server | +| macOS 15+ on Apple Silicon | ✅ Supported | Install the `seekdb` extra and connect to the remote server | +| macOS 14 or earlier on Apple Silicon | ⚠️ Blocked by upstream packaging | `pyseekdb` currently requires the unavailable native wheel even for server-only use; follow [#1324](https://github.com/oceanbase/seekdb/issues/1324) | +| macOS on Intel / Windows | ✅ Server mode only | Embedded bindings are not available | -**Recommendation for macOS/Windows users**: Deploy SeekDB on a Linux server and connect via server mode configuration. +Remote server mode does not use embedded storage at runtime. However, whether +the Python client can be installed still depends on `pyseekdb`'s package +metadata for the current platform. ## Configuration @@ -170,22 +183,23 @@ Key methods: ### Import Error -If you see: `ImportError: pyseekdb is not installed` +If you see: `SeekDB support is not installed` Solution: ```bash -pip install pyseekdb +uv sync --extra seekdb +# or: uvx --from 'langbot[seekdb]@latest' langbot ``` -### Embedded Mode Error on macOS/Windows +### Embedded Mode Is Unavailable on the Current Platform **Error**: ``` RuntimeError: Embedded Client is not available because pylibseekdb is not available. -Please install pylibseekdb (Linux only) or use RemoteServerClient (host/port) instead. ``` -**Cause**: `pylibseekdb` is only available on Linux platforms. +**Cause**: No compatible `pylibseekdb` wheel is installed for the current OS, +CPU architecture, Python version, and macOS deployment target. **Solution**: Use server mode instead: 1. Deploy SeekDB on a Linux server or VM @@ -208,29 +222,6 @@ vdb: use: chroma # or qdrant ``` -### Docker Container Fails on macOS - -**Symptoms**: -```bash -docker run -d -p 2881:2881 oceanbase/seekdb:latest -# Container exits immediately with code 30 -``` - -**Error in logs**: -``` -[ERROR] Code: Agent.SeekDB.Not.Exists -Message: initialize failed: init agent failed: SeekDB not exists in current directory. -``` - -**Cause**: This is a known issue with SeekDB Docker containers on macOS. See [oceanbase/seekdb#36](https://github.com/oceanbase/seekdb/issues/36). - -**Status**: Under investigation by OceanBase team. - -**Workaround Options**: -1. **Use alternatives**: ChromaDB or Qdrant work perfectly on macOS -2. **Remote server**: Deploy SeekDB on a Linux server and connect remotely -3. **Wait for fix**: Monitor the GitHub issue for updates - ### Connection Error (Server Mode) If SeekDB server is not reachable, check: diff --git a/pyproject.toml b/pyproject.toml index 34d48c53d..93f7d817b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -69,7 +69,6 @@ dependencies = [ "langchain-text-splitters>=1.1.2", "chromadb>=1.0.0,<2.0.0", "qdrant-client (>=1.15.1,<2.0.0)", - "pyseekdb==1.1.0.post3", "langbot-plugin==0.5.3", "asyncpg>=0.30.0", "line-bot-sdk>=3.19.0", @@ -107,6 +106,11 @@ classifiers = [ "Topic :: Communications :: Chat", ] +[project.optional-dependencies] +seekdb = [ + "pyseekdb==1.1.0.post3", +] + [project.urls] Homepage = "https://langbot.app" Documentation = "https://docs.langbot.app" diff --git a/src/langbot/libs/wecom_ai_bot_api/ws_client.py b/src/langbot/libs/wecom_ai_bot_api/ws_client.py index 516e7e389..5ca557b4d 100644 --- a/src/langbot/libs/wecom_ai_bot_api/ws_client.py +++ b/src/langbot/libs/wecom_ai_bot_api/ws_client.py @@ -48,6 +48,14 @@ CMD_RESPOND_MSG = 'aibot_respond_msg' CMD_RESPOND_WELCOME = 'aibot_respond_welcome_msg' CMD_RESPOND_UPDATE = 'aibot_respond_update_msg' CMD_SEND_MSG = 'aibot_send_msg' +# Media upload protocol (3 steps: init -> chunk * N -> finish). The +# command names below match the WeCom AI Bot long-connection protocol. +CMD_UPLOAD_INIT = 'aibot_upload_media_init' +CMD_UPLOAD_CHUNK = 'aibot_upload_media_chunk' +CMD_UPLOAD_FINISH = 'aibot_upload_media_finish' + +# Default upload chunk size: 512 KB before base64 encoding. +_UPLOAD_CHUNK_SIZE = 512 * 1024 _DEDUP_CACHE_MAX = 4096 _STREAM_CACHE_MAX = 1024 @@ -499,6 +507,145 @@ class WecomBotWsClient: body['chatid'] = chat_id return await self._send_reply(req_id, body, cmd=CMD_SEND_MSG) + # ------------------------------------------------------------------ + # Media upload (image / voice / file) + # ------------------------------------------------------------------ + + async def upload_media( + self, + data: bytes, + filename: str = 'attachment', + media_type: str = 'file', + ) -> Optional[dict]: + """Upload *data* to the WeCom AI Bot CDN and return the parsed ACK. + + Implements the three-step protocol documented for the WeCom + AI Bot: + + 1. ``aibot_upload_media_init`` — declare media type, file name, + size, MD5 and chunk count; receive ``upload_id``. + 2. ``aibot_upload_media_chunk`` — send each chunk (base64-encoded + bytes) until done; receive per-chunk ACK. + 3. ``aibot_upload_media_finish`` — finalize the upload; receive + ``media_id``. + + Returns a dict with the final ``media_id`` (and the raw + ``finish`` ACK) on success, or ``None`` on any failure. The + caller is expected to ignore the result and continue + gracefully — the framework will keep working without media + delivery. + """ + import base64 as _b64 + import hashlib as _hl + + if not data: + return None + + file_size = len(data) + file_md5 = _hl.md5(data).hexdigest() + total_chunks = (file_size + _UPLOAD_CHUNK_SIZE - 1) // _UPLOAD_CHUNK_SIZE + if total_chunks == 0: + total_chunks = 1 + + # Step 1: init. + init_req_id = _generate_req_id(CMD_UPLOAD_INIT) + init_body = { + 'type': media_type, + 'filename': filename, + 'total_size': file_size, + 'total_chunks': total_chunks, + 'md5': file_md5, + } + init_ack = await self._send_reply( + init_req_id, + init_body, + cmd=CMD_UPLOAD_INIT, + ) + if not init_ack or init_ack.get('errcode', 0) != 0: + await self.logger.warning(f'upload_media init failed: ack={init_ack!r}') + return None + upload_id = ( + init_ack.get('upload_id') + or init_ack.get('body', {}).get('upload_id') + or init_ack.get('data', {}).get('upload_id') + ) + if not upload_id: + await self.logger.warning(f'upload_media init returned no upload_id: ack={init_ack!r}') + return None + + # Step 2: chunks. + for index in range(total_chunks): + start = index * _UPLOAD_CHUNK_SIZE + end = min(start + _UPLOAD_CHUNK_SIZE, file_size) + chunk_bytes = data[start:end] + chunk_req_id = _generate_req_id(CMD_UPLOAD_CHUNK) + chunk_body = { + 'upload_id': upload_id, + 'chunk_index': index, + 'base64_data': _b64.b64encode(chunk_bytes).decode('ascii'), + } + chunk_ack = await self._send_reply( + chunk_req_id, + chunk_body, + cmd=CMD_UPLOAD_CHUNK, + ) + if not chunk_ack or chunk_ack.get('errcode', 0) != 0: + await self.logger.warning(f'upload_media chunk {index} failed: ack={chunk_ack!r}') + return None + + # Step 3: finish. + finish_req_id = _generate_req_id(CMD_UPLOAD_FINISH) + finish_body = {'upload_id': upload_id} + finish_ack = await self._send_reply( + finish_req_id, + finish_body, + cmd=CMD_UPLOAD_FINISH, + ) + if not finish_ack or finish_ack.get('errcode', 0) != 0: + await self.logger.warning(f'upload_media finish failed: ack={finish_ack!r}') + return None + + media_id = ( + finish_ack.get('media_id') + or finish_ack.get('body', {}).get('media_id') + or finish_ack.get('data', {}).get('media_id') + ) + if not media_id: + await self.logger.warning(f'upload_media finish returned no media_id: ack={finish_ack!r}') + return None + return {'media_id': media_id, 'ack': finish_ack} + + async def _reply_media( + self, + req_id: str, + media_id: str, + kind: str, + ) -> Optional[dict]: + """Send a media reply (image / voice / file) referencing *media_id*. + + ``kind`` is one of ``'image'``, ``'voice'``, ``'file'``. Uses + the standard ``aibot_respond_msg`` command with a per-kind + body key (matches the convention documented for the WeCom + AI Bot SDK). + """ + if kind not in {'image', 'voice', 'file'}: + await self.logger.warning(f'_reply_media called with unknown kind={kind!r}') + return None + body = { + 'msgtype': kind, + kind: {'media_id': media_id}, + } + return await self._send_reply(req_id, body, cmd=CMD_RESPOND_MSG) + + async def reply_image(self, req_id: str, media_id: str) -> Optional[dict]: + return await self._reply_media(req_id, media_id, 'image') + + async def reply_file(self, req_id: str, media_id: str) -> Optional[dict]: + return await self._reply_media(req_id, media_id, 'file') + + async def reply_voice(self, req_id: str, media_id: str) -> Optional[dict]: + return await self._reply_media(req_id, media_id, 'voice') + async def push_stream_chunk(self, msg_id: str, content: str, is_final: bool = False) -> bool: """Push a streaming chunk for a given message ID. diff --git a/src/langbot/pkg/api/http/controller/groups/user.py b/src/langbot/pkg/api/http/controller/groups/user.py index 6b42d4ea3..5568047ca 100644 --- a/src/langbot/pkg/api/http/controller/groups/user.py +++ b/src/langbot/pkg/api/http/controller/groups/user.py @@ -322,6 +322,7 @@ class UserRouterGroup(group.RouterGroup): if cloud_mode: capabilities['password_login_enabled'] = False capabilities['authenticated_invitation_acceptance_enabled'] = cloud_mode + capabilities['invitation_registration_enabled'] = not cloud_mode return self.success(data={'initialized': True, **capabilities}) @self.route('/set-password', methods=['POST'], auth_type=group.AuthType.USER_TOKEN) diff --git a/src/langbot/pkg/box/service.py b/src/langbot/pkg/box/service.py index 9b9ee5993..e9246e4c8 100644 --- a/src/langbot/pkg/box/service.py +++ b/src/langbot/pkg/box/service.py @@ -1225,8 +1225,9 @@ class BoxService: async def _read_outbox_via_exec(self, query: pipeline_query.Query) -> list[dict]: """Fallback: read the outbox over the exec channel (E2B / remote). - Note: exec stdout is truncated by ``output_limit_chars``, so this path - only reliably transfers small files. The host path is preferred. + Uses ``client.execute`` directly (bypassing ``_serialize_result``) + so stdout is NOT truncated by ``output_limit_chars`` - the raw + base64 payload can be far larger than the 4000-char display limit. """ import json as _json @@ -1280,14 +1281,22 @@ class BoxService: ' break\n' 'print(json.dumps(out))\n' ) - result = await self.execute_tool( - {'command': f"python3 - <<'LBPY'\n{script}\nLBPY", 'timeout_sec': 120}, - query, - ) - if not result.get('ok'): + spec_payload: dict = { + 'cmd': f"python3 - <<'LBPY'\n{script}\nLBPY", + 'timeout_sec': 120, + 'session_id': self.resolve_box_session_id(query), + } + if 'extra_mounts' not in spec_payload: + spec_payload['extra_mounts'] = self.build_skill_extra_mounts(query) + try: + spec = self.build_spec(spec_payload) + result = await self.client.execute(spec) + except Exception: + return [] + if not result.ok: return [] try: - return _json.loads(str(result.get('stdout') or '').strip().splitlines()[-1]) + return _json.loads(str(result.stdout or '').strip().splitlines()[-1]) except Exception: return [] diff --git a/src/langbot/pkg/command/cmdmgr.py b/src/langbot/pkg/command/cmdmgr.py index 7560f7c43..31f28dfdf 100644 --- a/src/langbot/pkg/command/cmdmgr.py +++ b/src/langbot/pkg/command/cmdmgr.py @@ -3,6 +3,7 @@ from __future__ import annotations import typing import inspect +from ..api.http.context import ExecutionContext from ..core import app from . import operator from ..utils import importutil @@ -66,7 +67,14 @@ class CommandManager: require_context = getattr(self.ap.plugin_connector, 'require_workspace_context', None) if require_context is not None: - result = require_context(context) + result = require_context( + ExecutionContext( + instance_uuid=context.instance_uuid, + workspace_uuid=context.workspace_uuid, + placement_generation=context.placement_generation, + query_uuid=context.query_uuid, + ) + ) if inspect.isawaitable(result): await result diff --git a/src/langbot/pkg/core/app.py b/src/langbot/pkg/core/app.py index 013f59b74..6ca8e513c 100644 --- a/src/langbot/pkg/core/app.py +++ b/src/langbot/pkg/core/app.py @@ -315,11 +315,36 @@ class Application: async def initialize(self): pass + async def _initialize_plugin_runtime(self) -> None: + try: + await self.plugin_connector.initialize() + except asyncio.CancelledError: + raise + except Exception as exc: + self.logger.warning(f'Plugin runtime unavailable during startup; reconnecting in background: {exc}') + self.plugin_connector.schedule_reconnect() + + def _start_plugin_runtime_initialization(self) -> asyncio.Task | None: + task = getattr(self, '_plugin_runtime_initialization_task', None) + if task is not None and not task.done(): + return task + # This is application lifecycle work, not a request side effect. It must + # not wait on PersistenceManager's after-commit gate at boot. + task = asyncio.create_task( + self._initialize_plugin_runtime(), + name='plugin-runtime-initialization', + ) + self._plugin_runtime_initialization_task = task + return task + async def run(self): self.event_loop_monitor.start() try: - if self.directory_projection_service is not None: - self.task_mgr.create_task( + if ( + self.directory_projection_service is not None + and getattr(self, 'directory_projection_task', None) is None + ): + self.directory_projection_task = self.task_mgr.create_task( self.directory_projection_service.run(), name='cloud-directory-projection', scopes=[core_entities.LifecycleControlScope.APPLICATION], @@ -336,7 +361,6 @@ class Application: name='cloud-manifest-refresh', scopes=[core_entities.LifecycleControlScope.APPLICATION], ) - await self.plugin_connector.initialize_plugins() # 后续可能会允许动态重启其他任务 # 故为了防止程序在非 Ctrl-C 情况下退出，这里创建一个不会结束的协程 @@ -362,6 +386,7 @@ class Application: name='http-api-controller', scopes=[core_entities.LifecycleControlScope.APPLICATION], ) + self._start_plugin_runtime_initialization() # Telemetry instance heartbeat (startup + daily); respects # space.disable_telemetry via TelemetryManager.send(). @@ -543,6 +568,11 @@ class Application: if self.task_mgr is not None: self.task_mgr.cancel_by_scope(core_entities.LifecycleControlScope.APPLICATION) + plugin_runtime_task = getattr(self, '_plugin_runtime_initialization_task', None) + if plugin_runtime_task is not None and not plugin_runtime_task.done(): + plugin_runtime_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await plugin_runtime_task with contextlib.suppress(Exception): await self.event_loop_monitor.stop() mcp_mount = getattr(self.http_ctrl, 'mcp_mount', None) diff --git a/src/langbot/pkg/core/stages/build_app.py b/src/langbot/pkg/core/stages/build_app.py index 8f0f21842..a6984a728 100644 --- a/src/langbot/pkg/core/stages/build_app.py +++ b/src/langbot/pkg/core/stages/build_app.py @@ -1,6 +1,6 @@ from __future__ import annotations -from .. import stage, app +from .. import stage, app, entities as core_entities from ...utils import version, proxy, constants from ...pipeline import pool, controller, pipelinemgr from ...pipeline import aggregator as message_aggregator @@ -297,14 +297,17 @@ class BuildAppStage(stage.BootingStage): async def runtime_disconnect_callback(connector: plugin_connector.PluginRuntimeConnector) -> None: connector.schedule_reconnect() + if ap.directory_projection_service is not None: + # Keep the projection fresh while shared Runtime cold restore runs. + # BuildApp initializes the connector before Application.run() starts + # its long-lived tasks, so start the single refresh task here. + ap.directory_projection_task = ap.task_mgr.create_task( + ap.directory_projection_service.run(), + name='cloud-directory-projection', + scopes=[core_entities.LifecycleControlScope.APPLICATION], + ) + plugin_connector_inst = plugin_connector.PluginRuntimeConnector(ap, runtime_disconnect_callback) - try: - await plugin_connector_inst.initialize() - except Exception as exc: - # Keep the API/UI available while an external or managed runtime is - # starting, then recover in the background with bounded backoff. - ap.logger.warning(f'Plugin runtime unavailable during startup; reconnecting in background: {exc}') - plugin_connector_inst.schedule_reconnect() ap.plugin_connector = plugin_connector_inst workspace_service_inst.release_startup_execution_bindings() diff --git a/src/langbot/pkg/persistence/mgr.py b/src/langbot/pkg/persistence/mgr.py index 7b444e2a6..f6ee713c8 100644 --- a/src/langbot/pkg/persistence/mgr.py +++ b/src/langbot/pkg/persistence/mgr.py @@ -177,7 +177,6 @@ class PersistenceManager: await self._validate_cloud_runtime() return - self._enable_sqlite_foreign_keys() if self.mode == PersistenceMode.RELEASE_MIGRATION: async with self._release_migration_lock(): await self._initialize_managed_schema() @@ -185,6 +184,7 @@ class PersistenceManager: return await self._initialize_managed_schema() + await self._enable_sqlite_foreign_keys_after_migration() if self.mode == PersistenceMode.OSS_COMPAT: await self.write_space_model_providers() @@ -328,6 +328,17 @@ class PersistenceManager: sqlalchemy.event.listen(self.get_db_engine().sync_engine, 'begin', set_oss_tenant_scope) self._oss_tenant_scope_listener_installed = True + async def _enable_sqlite_foreign_keys_after_migration(self) -> None: + """Enable SQLite FK enforcement only after table-rebuilding migrations.""" + engine = self.get_db_engine() + if engine.dialect.name != 'sqlite': + return + await engine.dispose() + self._enable_sqlite_foreign_keys() + # Dispose again so every runtime connection is opened through the new + # listener instead of reusing a pre-migration pooled connection. + await engine.dispose() + def _enable_sqlite_foreign_keys(self) -> None: """Enable SQLite FK enforcement for every pooled runtime connection.""" engine = self.get_db_engine() diff --git a/src/langbot/pkg/persistence/sqlite_migration_backup.py b/src/langbot/pkg/persistence/sqlite_migration_backup.py index 5e1f7e683..9004b7fcc 100644 --- a/src/langbot/pkg/persistence/sqlite_migration_backup.py +++ b/src/langbot/pkg/persistence/sqlite_migration_backup.py @@ -12,6 +12,7 @@ import re import secrets import sqlite3 import tempfile +import time import typing from sqlalchemy.ext.asyncio import AsyncEngine @@ -117,8 +118,19 @@ def _write_manifest(backup: SQLiteMigrationBackup, status: str, **extra: typing. temporary_path.unlink(missing_ok=True) -def _fsync_file(path: pathlib.Path) -> None: - descriptor = os.open(path, os.O_RDONLY) +def _fsync_file(path: pathlib.Path, *, reopen_attempts: int = 20) -> None: + """Sync a file, tolerating delayed visibility after replace on bind mounts.""" + + descriptor: int | None = None + for attempt in range(reopen_attempts): + try: + descriptor = os.open(path, os.O_RDONLY) + break + except FileNotFoundError: + if attempt + 1 >= reopen_attempts: + raise + time.sleep(0.05) + assert descriptor is not None try: os.fsync(descriptor) finally: diff --git a/src/langbot/pkg/pipeline/wrapper/wrapper.py b/src/langbot/pkg/pipeline/wrapper/wrapper.py index 50db693d4..eff976bfa 100644 --- a/src/langbot/pkg/pipeline/wrapper/wrapper.py +++ b/src/langbot/pkg/pipeline/wrapper/wrapper.py @@ -158,6 +158,18 @@ class ResponseWrapper(stage.PipelineStage): result_type=entities.ResultType.CONTINUE, new_query=query, ) + elif ( + isinstance(result, provider_message.MessageChunk) and result.is_final and not result.tool_calls + ): + # Final streaming chunk with no text content but + # possibly carrying sandbox outbox attachments. + reply_chain = platform_message.MessageChain([]) + await self._append_outbound_attachments(query, reply_chain) + query.resp_message_chain.append(reply_chain) + yield entities.StageProcessResult( + result_type=entities.ResultType.CONTINUE, + new_query=query, + ) if result.tool_calls is not None and len(result.tool_calls) > 0: # 有函数调用 function_names = [tc.function.name for tc in result.tool_calls] diff --git a/src/langbot/pkg/platform/sources/wecombot.py b/src/langbot/pkg/platform/sources/wecombot.py index e00bb9eba..3e650519a 100644 --- a/src/langbot/pkg/platform/sources/wecombot.py +++ b/src/langbot/pkg/platform/sources/wecombot.py @@ -3,8 +3,10 @@ import typing import asyncio import time import traceback +import base64 import datetime + import langbot_plugin.api.definition.abstract.platform.adapter as abstract_platform_adapter import langbot_plugin.api.entities.builtin.platform.message as platform_message import langbot_plugin.api.entities.builtin.platform.events as platform_events @@ -24,11 +26,24 @@ from langbot.libs.wecom_ai_bot_api.ws_client import WecomBotWsClient class WecomBotMessageConverter(abstract_platform_adapter.AbstractMessageConverter): @staticmethod async def yiri2target(message_chain: platform_message.MessageChain): - content = '' + """Convert a MessageChain into a list of component dicts. + + Each dict has a ``type`` key (``'text'``, ``'image'``, + ``'voice'``, ``'file'``). Text items carry ``text``; media + items carry ``base64`` (may include a ``data:...;base64,`` + prefix) and optionally ``name``. + """ + items: list[dict] = [] for msg in message_chain: if type(msg) is platform_message.Plain: - content += msg.text - return content + items.append({'type': 'text', 'text': msg.text}) + elif type(msg) is platform_message.Image: + items.append({'type': 'image', 'base64': msg.base64 or ''}) + elif type(msg) is platform_message.Voice: + items.append({'type': 'voice', 'base64': msg.base64 or ''}) + elif type(msg) is platform_message.File: + items.append({'type': 'file', 'base64': msg.base64 or '', 'name': msg.name or ''}) + return items @staticmethod async def target2yiri(event: WecomBotEvent, bot_name: str = ''): @@ -362,13 +377,76 @@ class WecomBotAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): } ) + @staticmethod + def _join_text_components(items: list[dict]) -> str: + """Concatenate ``text`` items in order, leaving media items alone.""" + return ''.join(item['text'] for item in items if item.get('type') == 'text') + + @staticmethod + def _iter_media_components(items: list[dict]): + """Yield non-text items in order.""" + for item in items: + if item.get('type') in {'image', 'voice', 'file'}: + yield item + + @staticmethod + async def _send_media( + bot, + req_id: str, + item: dict, + ) -> bool: + """Upload *item* to the WeCom AI Bot CDN and send it as a media reply. + + Returns True on success. Falls back to a no-op (with a warning log) + if the SDK does not yet implement ``upload_media`` / + ``reply_image`` / ``reply_file`` / ``reply_voice`` — the framework + will keep working, just without image delivery. + """ + kind = item.get('type') + upload = getattr(bot, 'upload_media', None) + if upload is None: + return False + b64_text = item.get('base64') or '' + if not b64_text: + return False + if b64_text.startswith('data:') and ',' in b64_text: + b64_text = b64_text.split(',', 1)[1] + try: + data = base64.b64decode(b64_text, validate=False) + except Exception: + return False + if not data: + return False + try: + upload_result = await upload(data, item.get('name') or f'attachment.{kind}', media_type=kind) + except Exception: + return False + media_id = getattr(upload_result, 'media_id', None) or ( + isinstance(upload_result, dict) and upload_result.get('media_id') + ) + if not media_id: + return False + reply_fn = { + 'image': getattr(bot, 'reply_image', None), + 'file': getattr(bot, 'reply_file', None), + 'voice': getattr(bot, 'reply_voice', None), + }.get(kind) + if reply_fn is None: + return False + try: + await reply_fn(req_id, media_id) + return True + except Exception: + return False + async def reply_message( self, message_source: platform_events.MessageEvent, message: platform_message.MessageChain, quote_origin: bool = False, ): - content = await self.message_converter.yiri2target(message) + items = await self.message_converter.yiri2target(message) + text = self._join_text_components(items) _ws_mode = not self.config.get('enable-webhook', False) event = message_source.source_platform_object @@ -382,7 +460,7 @@ class WecomBotAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): else: chat_id = str(message_source.sender.id) try: - await self.bot.send_message(chat_id, content) + await self.bot.send_message(chat_id, text) except Exception: await self.logger.error( f'WeComBot: proactive reply for synthetic event failed: {traceback.format_exc()}' @@ -396,12 +474,15 @@ class WecomBotAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): if _ws_mode: req_id = event.get('req_id', '') if isinstance(event, dict) else getattr(event, 'req_id', '') - if req_id: - await self.bot.reply_text(req_id, content) - else: - await self.bot.set_message(event.message_id, content) + if text: + if req_id: + await self.bot.reply_text(req_id, text) + else: + await self.bot.set_message(event.message_id, text) + for item in self._iter_media_components(items): + await self._send_media(self.bot, req_id, item) else: - await self.bot.set_message(event.message_id, content) + await self.bot.set_message(event.message_id, text) async def reply_message_chunk( self, @@ -411,7 +492,8 @@ class WecomBotAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): quote_origin: bool = False, is_final: bool = False, ): - content = await self.message_converter.yiri2target(message) + items = await self.message_converter.yiri2target(message) + text = self._join_text_components(items) _ws_mode = not self.config.get('enable-webhook', False) # Synthetic events (e.g. button-click triggered form resume) have @@ -420,7 +502,7 @@ class WecomBotAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): # of the stream/reply path. spo = message_source.source_platform_object if spo is None: - return await self._handle_synthetic_chunk(message_source, bot_message, content, is_final, _ws_mode) + return await self._handle_synthetic_chunk(message_source, bot_message, text, is_final, _ws_mode) msg_id = spo.message_id @@ -452,7 +534,7 @@ class WecomBotAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): form_data.get('actions', []) or [], ) except Exception: - fallback = content or '（人工输入）' + fallback = text or '（人工输入）' if _ws_mode: event = message_source.source_platform_object req_id = event.get('req_id', '') if isinstance(event, dict) else getattr(event, 'req_id', '') @@ -463,17 +545,22 @@ class WecomBotAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): return {'stream': False, 'form': True, 'fallback': True} if _ws_mode: - success = await self.bot.push_stream_chunk(msg_id, content, is_final=is_final) + success = await self.bot.push_stream_chunk(msg_id, text, is_final=is_final) if not success and is_final: event = message_source.source_platform_object req_id = event.get('req_id', '') if req_id: - await self.bot.reply_text(req_id, content) + await self.bot.reply_text(req_id, text) + if is_final: + event = message_source.source_platform_object + req_id = event.get('req_id', '') + for item in self._iter_media_components(items): + await self._send_media(self.bot, req_id, item) return {'stream': success} else: - success = await self.bot.push_stream_chunk(msg_id, content, is_final=is_final) + success = await self.bot.push_stream_chunk(msg_id, text, is_final=is_final) if not success and is_final: - await self.bot.set_message(msg_id, content) + await self.bot.set_message(msg_id, text) return {'stream': success} async def is_stream_output_supported(self) -> bool: @@ -627,8 +714,9 @@ class WecomBotAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter): async def send_message(self, target_type, target_id, message): _ws_mode = not self.config.get('enable-webhook', False) if _ws_mode: - content = await self.message_converter.yiri2target(message) - await self.bot.send_message(target_id, content) + items = await self.message_converter.yiri2target(message) + text = self._join_text_components(items) + await self.bot.send_message(target_id, text) else: pass diff --git a/src/langbot/pkg/plugin/connector.py b/src/langbot/pkg/plugin/connector.py index c36dfb7c9..5fb89bf68 100644 --- a/src/langbot/pkg/plugin/connector.py +++ b/src/langbot/pkg/plugin/connector.py @@ -701,7 +701,13 @@ class PluginRuntimeConnector(ManagedRuntimeConnector): } self._known_desired_states.update({state.binding.installation_uuid: state for state in desired_states}) - result = await runtime_handler.reconcile_plugin_installations(tuple(self._known_desired_states.values())) + reconcile_timeout_seconds = max( + 300.0, self._runtime_connect_timeout(self.ap.instance_config.data.get('plugin', {})) + ) + result = await runtime_handler.reconcile_plugin_installations( + tuple(self._known_desired_states.values()), + timeout=reconcile_timeout_seconds, + ) await self._repair_reconcile_missing_artifacts(self._known_desired_states, result) self._record_reconcile_failures(self._known_desired_states, result) @@ -736,7 +742,13 @@ class PluginRuntimeConnector(ManagedRuntimeConnector): if state.binding.installation_uuid in all_states: raise ValueError('Duplicate plugin installation UUID across projected Workspaces') all_states[state.binding.installation_uuid] = state - result = await runtime_handler.reconcile_plugin_installations(tuple(all_states.values())) + reconcile_timeout_seconds = max( + 300.0, self._runtime_connect_timeout(self.ap.instance_config.data.get('plugin', {})) + ) + result = await runtime_handler.reconcile_plugin_installations( + tuple(all_states.values()), + timeout=reconcile_timeout_seconds, + ) await self._repair_reconcile_missing_artifacts(all_states, result) self._record_reconcile_failures(all_states, result) for installation_uuid, previous in tuple(self._known_desired_states.items()): diff --git a/src/langbot/pkg/plugin/handler.py b/src/langbot/pkg/plugin/handler.py index 6491eb073..7e59c399c 100644 --- a/src/langbot/pkg/plugin/handler.py +++ b/src/langbot/pkg/plugin/handler.py @@ -13,6 +13,8 @@ from dataclasses import dataclass import pydantic import sqlalchemy +import sqlalchemy.dialects.postgresql +import sqlalchemy.dialects.sqlite from langbot_plugin.runtime.io import handler from langbot_plugin.runtime.io.connection import Connection @@ -832,6 +834,19 @@ class RuntimeConnectionHandler(handler.Handler): return f'{identity.plugin_author}/{identity.plugin_name}' raise ValueError(f'Unsupported binary storage owner_type {owner_type!r}') + @staticmethod + def _legacy_binary_storage_key( + action_context: ActionContext, + *, + owner_type: str, + owner: str, + key: str, + ) -> str: + """Return the pre-tenancy key shape for a row already scoped to this Workspace.""" + + legacy_owner = action_context.workspace_uuid if owner_type == 'workspace' else owner + return f'{owner_type}:{legacy_owner}:{key}' + @classmethod def _binary_storage_key( cls, @@ -1661,25 +1676,82 @@ class RuntimeConnectionHandler(handler.Handler): .where(persistence_bstorage.BinaryStorage.workspace_uuid == action_context.workspace_uuid) .where(persistence_bstorage.BinaryStorage.unique_key == unique_key) ) + storage = result.first() + if storage is None: + legacy_key = self._legacy_binary_storage_key( + action_context, + owner_type=owner_type, + owner=owner, + key=key, + ) + result = await self.ap.persistence_mgr.execute_async( + sqlalchemy.select(persistence_bstorage.BinaryStorage) + .where(persistence_bstorage.BinaryStorage.workspace_uuid == action_context.workspace_uuid) + .where(persistence_bstorage.BinaryStorage.unique_key == legacy_key) + .where(persistence_bstorage.BinaryStorage.key == key) + .where(persistence_bstorage.BinaryStorage.owner_type == owner_type) + .where(persistence_bstorage.BinaryStorage.owner == owner) + ) + storage = result.first() + if storage is not None: + update_result = await self.ap.persistence_mgr.execute_async( + sqlalchemy.update(persistence_bstorage.BinaryStorage) + .where(persistence_bstorage.BinaryStorage.workspace_uuid == action_context.workspace_uuid) + .where(persistence_bstorage.BinaryStorage.unique_key == legacy_key) + .where(persistence_bstorage.BinaryStorage.key == key) + .where(persistence_bstorage.BinaryStorage.owner_type == owner_type) + .where(persistence_bstorage.BinaryStorage.owner == owner) + .values(unique_key=unique_key, value=value) + ) + if update_result.rowcount: + return handler.ActionResponse.success(data={}) + canonical_update = await self.ap.persistence_mgr.execute_async( + sqlalchemy.update(persistence_bstorage.BinaryStorage) + .where(persistence_bstorage.BinaryStorage.workspace_uuid == action_context.workspace_uuid) + .where(persistence_bstorage.BinaryStorage.unique_key == unique_key) + .where(persistence_bstorage.BinaryStorage.key == key) + .where(persistence_bstorage.BinaryStorage.owner_type == owner_type) + .where(persistence_bstorage.BinaryStorage.owner == owner) + .values(value=value) + ) + if canonical_update.rowcount: + return handler.ActionResponse.success(data={}) + storage = None - if result.first() is not None: + if storage is not None: await self.ap.persistence_mgr.execute_async( sqlalchemy.update(persistence_bstorage.BinaryStorage) .where(persistence_bstorage.BinaryStorage.workspace_uuid == action_context.workspace_uuid) .where(persistence_bstorage.BinaryStorage.unique_key == unique_key) + .where(persistence_bstorage.BinaryStorage.key == key) + .where(persistence_bstorage.BinaryStorage.owner_type == owner_type) + .where(persistence_bstorage.BinaryStorage.owner == owner) .values(value=value) ) - else: - await self.ap.persistence_mgr.execute_async( - sqlalchemy.insert(persistence_bstorage.BinaryStorage).values( - workspace_uuid=action_context.workspace_uuid, - unique_key=unique_key, - key=key, - owner_type=owner_type, - owner=owner, - value=value, - ) + return handler.ActionResponse.success(data={}) + + dialect_name = self.ap.persistence_mgr.get_db_engine().dialect.name + insert = { + 'postgresql': sqlalchemy.dialects.postgresql.insert, + 'sqlite': sqlalchemy.dialects.sqlite.insert, + }.get(dialect_name) + if insert is None: + return handler.ActionResponse.error(message=f'Unsupported storage database dialect: {dialect_name}') + await self.ap.persistence_mgr.execute_async( + insert(persistence_bstorage.BinaryStorage) + .values( + workspace_uuid=action_context.workspace_uuid, + unique_key=unique_key, + key=key, + owner_type=owner_type, + owner=owner, + value=value, ) + .on_conflict_do_update( + index_elements=['workspace_uuid', 'unique_key'], + set_={'value': value}, + ) + ) return handler.ActionResponse.success( data={}, @@ -1722,6 +1794,29 @@ class RuntimeConnectionHandler(handler.Handler): ) storage = result.first() + if storage is None: + legacy_key = self._legacy_binary_storage_key( + action_context, + owner_type=owner_type, + owner=owner, + key=key, + ) + result = await self.ap.persistence_mgr.execute_async( + sqlalchemy.select(persistence_bstorage.BinaryStorage) + .where(persistence_bstorage.BinaryStorage.workspace_uuid == action_context.workspace_uuid) + .where(persistence_bstorage.BinaryStorage.unique_key == legacy_key) + .where(persistence_bstorage.BinaryStorage.key == key) + .where(persistence_bstorage.BinaryStorage.owner_type == owner_type) + .where(persistence_bstorage.BinaryStorage.owner == owner) + ) + storage = result.first() + if storage is None: + retry_result = await self.ap.persistence_mgr.execute_async( + sqlalchemy.select(persistence_bstorage.BinaryStorage) + .where(persistence_bstorage.BinaryStorage.workspace_uuid == action_context.workspace_uuid) + .where(persistence_bstorage.BinaryStorage.unique_key == unique_key) + ) + storage = retry_result.first() if storage is None: return handler.ActionResponse.error( message=f'Storage with key {key} not found', @@ -1768,10 +1863,19 @@ class RuntimeConnectionHandler(handler.Handler): message=str(e), ) + legacy_key = self._legacy_binary_storage_key( + action_context, + owner_type=owner_type, + owner=owner, + key=key, + ) await self.ap.persistence_mgr.execute_async( sqlalchemy.delete(persistence_bstorage.BinaryStorage) .where(persistence_bstorage.BinaryStorage.workspace_uuid == action_context.workspace_uuid) - .where(persistence_bstorage.BinaryStorage.unique_key == unique_key) + .where(persistence_bstorage.BinaryStorage.unique_key.in_((unique_key, legacy_key))) + .where(persistence_bstorage.BinaryStorage.key == key) + .where(persistence_bstorage.BinaryStorage.owner_type == owner_type) + .where(persistence_bstorage.BinaryStorage.owner == owner) ) return handler.ActionResponse.success( @@ -1810,7 +1914,7 @@ class RuntimeConnectionHandler(handler.Handler): return handler.ActionResponse.success( data={ - 'keys': result.scalars().all(), + 'keys': list(dict.fromkeys(result.scalars().all())), }, ) @@ -2477,13 +2581,15 @@ class RuntimeConnectionHandler(handler.Handler): async def reconcile_plugin_installations( self, installations: tuple[PluginInstallationDesiredState, ...], + *, + timeout: float = 300, ) -> dict[str, Any]: request = ReconcilePluginInstallationsRequest(installations=installations) with self.installation_scope(None): return await self.call_action( LangBotToRuntimeAction.RECONCILE_PLUGIN_INSTALLATIONS, request.model_dump(), - timeout=300, + timeout=timeout, ) async def apply_plugin_installation( diff --git a/src/langbot/pkg/provider/modelmgr/requesters/seekdbembed.py b/src/langbot/pkg/provider/modelmgr/requesters/seekdbembed.py index 4b881dbfe..ff7d1d642 100644 --- a/src/langbot/pkg/provider/modelmgr/requesters/seekdbembed.py +++ b/src/langbot/pkg/provider/modelmgr/requesters/seekdbembed.py @@ -24,7 +24,10 @@ class SeekDBEmbedding(requester.ProviderAPIRequester): try: import pyseekdb except ImportError: - raise ImportError('pyseekdb is not installed. Install it with: pip install pyseekdb') + raise ImportError( + "SeekDB support is not installed. Install LangBot with the 'seekdb' extra: " + "uv sync --extra seekdb (source) or uvx --from 'langbot[seekdb]@latest' langbot (PyPI)." + ) self._embedding_function = pyseekdb.get_default_embedding_function() diff --git a/src/langbot/pkg/vector/vdbs/seekdb.py b/src/langbot/pkg/vector/vdbs/seekdb.py index fc82298e0..5be28b458 100644 --- a/src/langbot/pkg/vector/vdbs/seekdb.py +++ b/src/langbot/pkg/vector/vdbs/seekdb.py @@ -42,7 +42,10 @@ class SeekDBVectorDatabase(VectorDatabase): def __init__(self, ap: app.Application): if not SEEKDB_AVAILABLE: - raise ImportError('pyseekdb is not installed. Install it with: pip install pyseekdb') + raise ImportError( + "SeekDB support is not installed. Install LangBot with the 'seekdb' extra: " + "uv sync --extra seekdb (source) or uvx --from 'langbot[seekdb]@latest' langbot (PyPI)." + ) self.ap = ap config = self.ap.instance_config.data['vdb']['seekdb'] diff --git a/src/langbot/templates/config.yaml b/src/langbot/templates/config.yaml index a39261f77..c4ff914ba 100644 --- a/src/langbot/templates/config.yaml +++ b/src/langbot/templates/config.yaml @@ -181,6 +181,11 @@ vdb: host: localhost port: 6333 api_key: '' + # SeekDB is optional. Native/package installs need the `seekdb` extra: + # `uv sync --extra seekdb` (source) or + # `uvx --from 'langbot[seekdb]@latest' langbot` (PyPI). + # The official Docker image already includes it. + # Embedded-mode platform support depends on the native pylibseekdb wheels. seekdb: mode: embedded # 'embedded' or 'server' # Embedded mode options: diff --git a/tests/integration/api/test_smoke.py b/tests/integration/api/test_smoke.py index dfd5054f2..642efaf2b 100644 --- a/tests/integration/api/test_smoke.py +++ b/tests/integration/api/test_smoke.py @@ -307,6 +307,7 @@ class TestUserInitEndpoint: assert data['data'] == { 'initialized': True, 'authenticated_invitation_acceptance_enabled': False, + 'invitation_registration_enabled': True, 'password_login_enabled': True, 'space_login_enabled': False, } @@ -330,6 +331,28 @@ class TestUserInitEndpoint: assert data['data'] == { 'initialized': True, 'authenticated_invitation_acceptance_enabled': True, + 'invitation_registration_enabled': False, + 'password_login_enabled': False, + 'space_login_enabled': True, + } + + @pytest.mark.asyncio + async def test_account_info_enables_local_invitation_registration_for_oauth_only_oss( + self, quart_test_client, fake_api_app + ): + fake_api_app.user_service.is_initialized.return_value = True + fake_api_app.user_service.get_login_capabilities = AsyncMock( + return_value={'password_login_enabled': False, 'space_login_enabled': True} + ) + + response = await quart_test_client.get('/api/v1/user/account-info') + + assert response.status_code == 200 + data = await response.get_json() + assert data['data'] == { + 'initialized': True, + 'authenticated_invitation_acceptance_enabled': False, + 'invitation_registration_enabled': True, 'password_login_enabled': False, 'space_login_enabled': True, } diff --git a/tests/integration/api/test_user_space_oauth.py b/tests/integration/api/test_user_space_oauth.py index 9d22e5b3c..be0a9f021 100644 --- a/tests/integration/api/test_user_space_oauth.py +++ b/tests/integration/api/test_user_space_oauth.py @@ -312,6 +312,29 @@ async def test_space_credits_are_resolved_from_workspace_owner(space_oauth_api): application.space_service.get_credits.assert_awaited_once_with('owner@example.com') +@pytest.mark.asyncio +async def test_oss_local_only_owner_requires_space_binding_for_langbot_models(space_oauth_api): + application, client = space_oauth_api + application.user_service.get_workspace_owner = AsyncMock( + return_value=SimpleNamespace(user='owner@example.com', space_account_uuid=None) + ) + application.space_service.get_credits = AsyncMock() + + response = await client.get( + '/api/v1/user/space-credits', + headers={'Authorization': 'Bearer account-token', 'X-Workspace-Id': WORKSPACE_UUID}, + ) + payload = await response.get_json() + + assert response.status_code == 200 + assert payload['data'] == { + 'credits': None, + 'owner_space_bound': False, + 'is_workspace_owner': True, + } + application.space_service.get_credits.assert_not_awaited() + + @pytest.mark.asyncio async def test_cloud_workspace_owner_is_always_space_bound_after_login(space_oauth_api): application, client = space_oauth_api diff --git a/tests/integration/persistence/resource_migration_support.py b/tests/integration/persistence/resource_migration_support.py index f3eea0f30..afb05ae15 100644 --- a/tests/integration/persistence/resource_migration_support.py +++ b/tests/integration/persistence/resource_migration_support.py @@ -81,6 +81,7 @@ async def create_legacy_resource_schema(engine, *, instance_uuid: str) -> None: sa.Column('key', sa.String(255), nullable=False), sa.Column('owner_type', sa.String(255), nullable=False), sa.Column('owner', sa.String(255), nullable=False), + sa.Column('value', sa.LargeBinary, nullable=False), ) mcp_servers = _uuid_table( metadata, @@ -210,7 +211,13 @@ async def create_legacy_resource_schema(engine, *, instance_uuid: str) -> None: await conn.execute(bots.insert().values(uuid='bot-1', name='bot', updated_at=now)) await conn.execute(bot_admins.insert().values(bot_uuid='bot-1', launcher_type='person', launcher_id='owner')) await conn.execute( - binary_storages.insert().values(unique_key='plugin:demo:key', key='key', owner_type='plugin', owner='demo') + binary_storages.insert().values( + unique_key='plugin:demo:key', + key='key', + owner_type='plugin', + owner='demo', + value=b'legacy-plugin-value', + ) ) await conn.execute(mcp_servers.insert().values(uuid='mcp-1', name='shared-name', enable=True, updated_at=now)) await conn.execute(model_providers.insert().values(uuid='provider-1', name='provider', requester='openai')) diff --git a/tests/integration/persistence/test_resource_tenancy_migration.py b/tests/integration/persistence/test_resource_tenancy_migration.py index 6ae288061..8bc3797a1 100644 --- a/tests/integration/persistence/test_resource_tenancy_migration.py +++ b/tests/integration/persistence/test_resource_tenancy_migration.py @@ -76,6 +76,26 @@ async def test_legacy_sqlite_resources_are_backfilled_and_contracted(tmp_path): ) assert legacy_kb['collection_id'] == 'collection-1' assert legacy_kb['legacy_vector_collection'] == 1 + legacy_binary_storage = ( + ( + await conn.execute( + sa.text( + 'SELECT workspace_uuid, unique_key, key, owner_type, owner, value ' + "FROM binary_storages WHERE owner_type = 'plugin' AND owner = 'demo'" + ) + ) + ) + .mappings() + .one() + ) + assert legacy_binary_storage == { + 'workspace_uuid': workspace_uuid, + 'unique_key': 'plugin:demo:key', + 'key': 'key', + 'owner_type': 'plugin', + 'owner': 'demo', + 'value': b'legacy-plugin-value', + } assert ( await conn.scalar( sa.text( @@ -209,8 +229,8 @@ async def test_sqlite_scoped_keys_allow_cross_workspace_but_reject_same_workspac await conn.execute( sa.text( 'INSERT INTO binary_storages ' - '(workspace_uuid, unique_key, key, owner_type, owner) ' - "VALUES (:workspace_uuid, 'plugin:demo:key', 'key', 'plugin', 'demo')" + '(workspace_uuid, unique_key, key, owner_type, owner, value) ' + "VALUES (:workspace_uuid, 'plugin:demo:key', 'key', 'plugin', 'demo', X'')" ), {'workspace_uuid': second_workspace_uuid}, ) diff --git a/tests/integration/persistence/test_sqlite_migration_backup.py b/tests/integration/persistence/test_sqlite_migration_backup.py index dab50dd96..e808cb3fa 100644 --- a/tests/integration/persistence/test_sqlite_migration_backup.py +++ b/tests/integration/persistence/test_sqlite_migration_backup.py @@ -2,6 +2,7 @@ from __future__ import annotations import json import logging +import os import pathlib import sqlite3 @@ -9,7 +10,7 @@ import pytest import sqlalchemy as sa from sqlalchemy.ext.asyncio import create_async_engine -from langbot.pkg.persistence import alembic_runner +from langbot.pkg.persistence import alembic_runner, sqlite_migration_backup from langbot.pkg.persistence.mgr import PersistenceManager from .resource_migration_support import create_legacy_resource_schema @@ -105,3 +106,31 @@ async def test_failed_tenancy_migration_restores_backup_and_revision( assert await alembic_runner.get_alembic_current(engine) == alembic_runner.get_alembic_head() finally: await engine.dispose() + + +async def test_backup_retries_transient_reopen_failure_after_replace(tmp_path, monkeypatch): + database_path = tmp_path / 'legacy-bind-mount.db' + engine = create_async_engine(f'sqlite+aiosqlite:///{database_path}') + real_open = os.open + transient_failures = 0 + + def transient_open(path, flags, *args, **kwargs): + nonlocal transient_failures + candidate = pathlib.Path(path) + if candidate.suffix == '.sqlite3' and candidate.parent.name == 'migration-backups' and transient_failures == 0: + transient_failures += 1 + raise FileNotFoundError(2, 'simulated delayed bind-mount visibility', str(candidate)) + return real_open(path, flags, *args, **kwargs) + + try: + await create_legacy_resource_schema(engine, instance_uuid='backup-bind-mount') + await alembic_runner.run_alembic_stamp(engine, '0008_mcp_resource_prefs') + monkeypatch.setattr(sqlite_migration_backup.os, 'open', transient_open) + + await _manager(engine)._run_alembic_migrations() + + assert transient_failures == 1 + assert await alembic_runner.get_alembic_current(engine) == alembic_runner.get_alembic_head() + assert len(_manifest_payloads(tmp_path / 'migration-backups')) == 2 + finally: + await engine.dispose() diff --git a/tests/integration/persistence/test_workspace_migration.py b/tests/integration/persistence/test_workspace_migration.py index a686f59e8..2a24cb025 100644 --- a/tests/integration/persistence/test_workspace_migration.py +++ b/tests/integration/persistence/test_workspace_migration.py @@ -220,13 +220,17 @@ async def test_existing_oss_workspace_is_rekeyed_to_instance_identity(tmp_path): ) async with engine.begin() as conn: await conn.run_sync(schema.create_all) - await conn.execute(sa.text("INSERT INTO metadata (key, value) VALUES ('instance_uuid', :value)"), {'value': instance_id}) await conn.execute( - sa.text("INSERT INTO workspaces (uuid, instance_uuid, slug, source) VALUES (:uuid, :instance, 'default', 'local')"), + sa.text("INSERT INTO metadata (key, value) VALUES ('instance_uuid', :value)"), {'value': instance_id} + ) + await conn.execute( + sa.text( + "INSERT INTO workspaces (uuid, instance_uuid, slug, source) VALUES (:uuid, :instance, 'default', 'local')" + ), {'uuid': old_workspace_uuid, 'instance': instance_id}, ) await conn.execute( - sa.text("INSERT INTO tenant_rows (id, workspace_uuid) VALUES (1, :uuid)"), + sa.text('INSERT INTO tenant_rows (id, workspace_uuid) VALUES (1, :uuid)'), {'uuid': old_workspace_uuid}, ) await run_alembic_stamp(engine, '0016_support_admin_sessions') @@ -234,8 +238,8 @@ async def test_existing_oss_workspace_is_rekeyed_to_instance_identity(tmp_path): await run_alembic_upgrade(engine, 'head') async with engine.connect() as conn: - assert (await conn.execute(sa.text("SELECT uuid FROM workspaces"))).scalar_one() == canonical_uuid - assert (await conn.execute(sa.text("SELECT workspace_uuid FROM tenant_rows"))).scalar_one() == canonical_uuid + assert (await conn.execute(sa.text('SELECT uuid FROM workspaces'))).scalar_one() == canonical_uuid + assert (await conn.execute(sa.text('SELECT workspace_uuid FROM tenant_rows'))).scalar_one() == canonical_uuid await engine.dispose() @@ -452,6 +456,45 @@ async def test_persistence_startup_defers_workspace_tables_until_account_upgrade await engine.dispose() +async def test_persistence_startup_preserves_legacy_workspace_membership_with_foreign_keys( + tmp_path, + monkeypatch, +): + database_path = tmp_path / 'startup-foreign-keys.db' + engine = create_async_engine(f'sqlite+aiosqlite:///{database_path}') + try: + await _create_legacy_schema(engine) + await run_alembic_stamp(engine, '0008_mcp_resource_prefs') + finally: + await engine.dispose() + + monkeypatch.setattr(constants, 'instance_id', 'instance_migration_test') + application = type('Application', (), {})() + application.logger = logging.getLogger('workspace-startup-foreign-keys-test') + application.instance_config = type( + 'InstanceConfig', + (), + {'data': {'database': {'use': 'sqlite', 'sqlite': {'path': str(database_path)}}}}, + )() + manager = PersistenceManager(application) + + await manager.initialize() + try: + async with manager.get_db_engine().connect() as conn: + workspace = ( + (await conn.execute(sa.text("SELECT * FROM workspaces WHERE source = 'local'"))).mappings().one() + ) + membership = (await conn.execute(sa.text('SELECT * FROM workspace_memberships'))).mappings().one() + foreign_keys = await conn.scalar(sa.text('PRAGMA foreign_keys')) + + assert workspace['created_by_account_uuid'] == membership['account_uuid'] + assert membership['role'] == 'owner' + assert membership['status'] == 'active' + assert foreign_keys == 1 + finally: + await manager.shutdown() + + async def test_oss_workspace_identity_rekeys_fk_graph_and_metadata(tmp_path): engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "workspace-rekey.db"}') try: @@ -466,7 +509,7 @@ async def test_oss_workspace_identity_rekeys_fk_graph_and_metadata(tmp_path): assert instance_uuid await conn.execute( sa.text( - "INSERT INTO workspace_metadata (workspace_uuid, key, value) " + 'INSERT INTO workspace_metadata (workspace_uuid, key, value) ' "VALUES (:workspace_uuid, 'migration_probe', 'present')" ), {'workspace_uuid': old_uuid}, @@ -474,7 +517,7 @@ async def test_oss_workspace_identity_rekeys_fk_graph_and_metadata(tmp_path): await conn.execute( sa.text( "INSERT INTO metadata (key, value) VALUES ('oss_workspace_uuid', :workspace_uuid) " - "ON CONFLICT(key) DO UPDATE SET value = excluded.value" + 'ON CONFLICT(key) DO UPDATE SET value = excluded.value' ), {'workspace_uuid': old_uuid}, ) @@ -483,12 +526,16 @@ async def test_oss_workspace_identity_rekeys_fk_graph_and_metadata(tmp_path): expected_uuid = workspace_uuid_from_instance_id(instance_uuid) async with engine.connect() as conn: assert await conn.scalar(sa.text("SELECT uuid FROM workspaces WHERE source = 'local'")) == expected_uuid - assert await conn.scalar( - sa.text("SELECT workspace_uuid FROM workspace_metadata WHERE key = 'migration_probe'") - ) == expected_uuid - assert await conn.scalar( - sa.text("SELECT value FROM metadata WHERE key = 'oss_workspace_uuid'") - ) == expected_uuid + assert ( + await conn.scalar( + sa.text("SELECT workspace_uuid FROM workspace_metadata WHERE key = 'migration_probe'") + ) + == expected_uuid + ) + assert ( + await conn.scalar(sa.text("SELECT value FROM metadata WHERE key = 'oss_workspace_uuid'")) + == expected_uuid + ) finally: await engine.dispose() diff --git a/tests/unit_tests/box/test_box_service.py b/tests/unit_tests/box/test_box_service.py index 9c78f771c..ce6e7bc5f 100644 --- a/tests/unit_tests/box/test_box_service.py +++ b/tests/unit_tests/box/test_box_service.py @@ -2163,25 +2163,38 @@ class TestInboundOutboundRoundTrip: calls = [] - async def fake_execute_tool(parameters, q): - calls.append(parameters['command']) - if 'os.scandir' in parameters['command']: - return { - 'ok': True, - 'stdout': '[{"name": "out.png", "b64": "QUJD"}]', - 'stderr': '', - } + async def fake_client_execute(spec): + cmd = spec.cmd + calls.append(cmd) + if 'os.scandir' in cmd: + return BoxExecutionResult( + session_id='s', + backend_name='test', + status=BoxExecutionStatus.COMPLETED, + exit_code=0, + stdout='[{"name": "out.png", "b64": "QUJD"}]', + duration_ms=10, + ) # the rm -rf cleanup call - return {'ok': True, 'stdout': '', 'stderr': ''} + return BoxExecutionResult( + session_id='s', + backend_name='test', + status=BoxExecutionStatus.COMPLETED, + exit_code=0, + stdout='', + duration_ms=10, + ) - service.execute_tool = AsyncMock(side_effect=fake_execute_tool) + service.client.execute = AsyncMock(side_effect=fake_client_execute) + service.execute_tool = AsyncMock(return_value={'ok': True, 'stdout': '', 'stderr': ''}) attachments = await service.collect_outbound_attachments(query) assert len(attachments) == 1 assert attachments[0]['type'] == 'Image' assert attachments[0]['name'] == 'out.png' # cleanup (rm -rf) must have been issued after a successful collection - assert any('rm -rf' in c for c in calls) + service.execute_tool.assert_awaited_once() + assert 'rm -rf' in service.execute_tool.await_args.args[0]['command'] @pytest.mark.asyncio async def test_collect_outbound_empty_still_clears(self): @@ -2193,16 +2206,33 @@ class TestInboundOutboundRoundTrip: calls = [] - async def fake_execute_tool(parameters, q): - calls.append(parameters['command']) - if 'os.scandir' in parameters['command']: - return {'ok': True, 'stdout': '[]', 'stderr': ''} - return {'ok': True, 'stdout': '', 'stderr': ''} + async def fake_client_execute(spec): + cmd = spec.cmd + calls.append(cmd) + if 'os.scandir' in cmd: + return BoxExecutionResult( + session_id='s', + backend_name='test', + status=BoxExecutionStatus.COMPLETED, + exit_code=0, + stdout='[]', + duration_ms=10, + ) + return BoxExecutionResult( + session_id='s', + backend_name='test', + status=BoxExecutionStatus.COMPLETED, + exit_code=0, + stdout='', + duration_ms=10, + ) - service.execute_tool = AsyncMock(side_effect=fake_execute_tool) + service.client.execute = AsyncMock(side_effect=fake_client_execute) + service.execute_tool = AsyncMock(return_value={'ok': True, 'stdout': '', 'stderr': ''}) assert await service.collect_outbound_attachments(query) == [] # cleanup (rm -rf) is issued unconditionally now - assert any('rm -rf' in c for c in calls) + service.execute_tool.assert_awaited_once() + assert 'rm -rf' in service.execute_tool.await_args.args[0]['command'] @pytest.mark.asyncio async def test_passthrough_noop_when_unavailable(self): diff --git a/tests/unit_tests/command/test_cmdmgr.py b/tests/unit_tests/command/test_cmdmgr.py index ade27cf48..afc251a13 100644 --- a/tests/unit_tests/command/test_cmdmgr.py +++ b/tests/unit_tests/command/test_cmdmgr.py @@ -11,6 +11,7 @@ from unittest.mock import AsyncMock, Mock from langbot.pkg.command import operator from langbot.pkg.command.cmdmgr import CommandManager +from langbot.pkg.api.http.context import ExecutionContext from tests.factories import FakeApp, command_query import langbot_plugin.api.entities.builtin.provider.session as provider_session @@ -393,6 +394,32 @@ class TestCommandManagerInternalExecute: assert len(results) == 1 assert results[0].text == 'plugin response' + @pytest.mark.asyncio + async def test_execute_selects_workspace_with_trusted_context(self): + """Plugin command discovery receives the typed runtime scope.""" + + fake_app = FakeApp() + mgr = CommandManager(fake_app) + mgr.cmd_list = [] + fake_app.plugin_connector.require_workspace_context = AsyncMock() + fake_app.plugin_connector.list_commands = AsyncMock(return_value=[]) + + ctx = self._create_context(command='help') + ctx.instance_uuid = 'instance-a' + ctx.workspace_uuid = 'workspace-a' + ctx.placement_generation = 4 + ctx.query_uuid = 'query-a' + + async for _ in mgr._execute(ctx, mgr.cmd_list): + pass + + selected = fake_app.plugin_connector.require_workspace_context.await_args.args[0] + assert isinstance(selected, ExecutionContext) + assert selected.instance_uuid == 'instance-a' + assert selected.workspace_uuid == 'workspace-a' + assert selected.placement_generation == 4 + assert selected.query_uuid == 'query-a' + @pytest.mark.asyncio async def test_execute_with_bound_plugins(self): """_execute passes bound_plugins to plugin connector.""" diff --git a/tests/unit_tests/core/test_app_shutdown.py b/tests/unit_tests/core/test_app_shutdown.py index ad6dcfad6..2c953aaac 100644 --- a/tests/unit_tests/core/test_app_shutdown.py +++ b/tests/unit_tests/core/test_app_shutdown.py @@ -144,3 +144,39 @@ async def test_runtime_resource_stats_are_aggregate_and_constant_time() -> None: assert stats['models']['providers'] == 1 assert stats['runtimes']['plugin_installations'] == 1 assert stats['runtimes']['plugin_runtime_connected'] is True + + +@pytest.mark.asyncio +async def test_start_plugin_runtime_initialization_bypasses_after_commit_gate() -> None: + app = Application() + app.plugin_connector = SimpleNamespace(initialize=AsyncMock()) + app.task_mgr = SimpleNamespace(create_task=AsyncMock()) + + task = app._start_plugin_runtime_initialization() + await task + + app.plugin_connector.initialize.assert_awaited_once_with() + app.task_mgr.create_task.assert_not_called() + + +@pytest.mark.asyncio +async def test_shutdown_cancels_plugin_runtime_initialization_task() -> None: + app = Application() + app._plugin_runtime_initialization_task = asyncio.create_task(asyncio.sleep(60)) + app.task_mgr = SimpleNamespace(cancel_by_scope=lambda *_: None, tasks=[]) + app.event_loop_monitor = SimpleNamespace(stop=AsyncMock()) + app.http_ctrl = SimpleNamespace(mcp_mount=None) + app.platform_mgr = None + app.tool_mgr = None + app.model_mgr = None + app.box_service = None + app.plugin_connector = None + app.telemetry = None + app.vector_db_mgr = None + app.storage_mgr = None + app.persistence_mgr = SimpleNamespace(db=SimpleNamespace(engine=SimpleNamespace(dispose=AsyncMock()))) + app.deployment = None + + await app.shutdown() + + assert app._plugin_runtime_initialization_task.cancelled() diff --git a/tests/unit_tests/platform/test_wecombot_media.py b/tests/unit_tests/platform/test_wecombot_media.py new file mode 100644 index 000000000..8bbff53e1 --- /dev/null +++ b/tests/unit_tests/platform/test_wecombot_media.py @@ -0,0 +1,127 @@ +import base64 + +import pytest + +import langbot.pkg.core.app # noqa: F401 +import langbot_plugin.api.entities.builtin.platform.message as platform_message +from langbot.libs.wecom_ai_bot_api.ws_client import _UPLOAD_CHUNK_SIZE, WecomBotWsClient +from langbot.pkg.platform.sources.wecombot import WecomBotAdapter, WecomBotMessageConverter + + +class Logger: + def __init__(self): + self.warnings = [] + self.errors = [] + + async def warning(self, message): + self.warnings.append(message) + + async def error(self, message): + self.errors.append(message) + + async def info(self, message): + return None + + +class UploadClient(WecomBotWsClient): + def __init__(self): + super().__init__(bot_id='bot', secret='secret', logger=Logger()) + self.frames = [] + + async def _send_reply(self, req_id: str, body: dict, cmd: str = 'aibot_respond_msg'): + self.frames.append((cmd, body)) + if cmd == 'aibot_upload_media_init': + return {'errcode': 0, 'body': {'upload_id': 'upload-1'}} + if cmd == 'aibot_upload_media_finish': + return {'errcode': 0, 'body': {'media_id': 'media-1'}} + return {'errcode': 0} + + +class Bot: + def __init__(self): + self.calls = [] + + async def upload_media(self, data, filename='attachment', media_type='file'): + self.calls.append(('upload_media', media_type, filename, data)) + return {'media_id': 'media-1'} + + async def reply_text(self, req_id, content): + self.calls.append(('reply_text', req_id, content)) + + async def reply_image(self, req_id, media_id): + self.calls.append(('reply_image', req_id, media_id)) + + async def send_message(self, target_id, content): + self.calls.append(('send_message', target_id, content)) + + +def make_adapter(bot): + return WecomBotAdapter.model_construct( + bot=bot, + config={'enable-webhook': False}, + logger=Logger(), + message_converter=WecomBotMessageConverter(), + ) + + +@pytest.mark.asyncio +async def test_ws_client_upload_media_uses_chunk_protocol(): + client = UploadClient() + data = b'a' * (_UPLOAD_CHUNK_SIZE + 1) + + upload_result = await client.upload_media(data, 'image.png', media_type='image') + + assert upload_result['media_id'] == 'media-1' + assert [cmd for cmd, _ in client.frames] == [ + 'aibot_upload_media_init', + 'aibot_upload_media_chunk', + 'aibot_upload_media_chunk', + 'aibot_upload_media_finish', + ] + init_body = client.frames[0][1] + assert init_body['type'] == 'image' + assert init_body['filename'] == 'image.png' + assert init_body['total_size'] == len(data) + assert init_body['total_chunks'] == 2 + assert client.frames[1][1]['chunk_index'] == 0 + assert base64.b64decode(client.frames[1][1]['base64_data']) == b'a' * _UPLOAD_CHUNK_SIZE + assert client.frames[2][1]['chunk_index'] == 1 + assert base64.b64decode(client.frames[2][1]['base64_data']) == b'a' + + +@pytest.mark.asyncio +async def test_reply_message_uploads_and_replies_image_media(): + bot = Bot() + adapter = make_adapter(bot) + png_data = b'\x89PNG\r\n\x1a\nimage' + image_b64 = base64.b64encode(png_data).decode('utf-8') + chain = platform_message.MessageChain([platform_message.Image(base64=f'data:image/png;base64,{image_b64}')]) + + items = await WecomBotMessageConverter.yiri2target(chain) + await adapter._send_media(bot, 'req-1', items[0]) + + assert bot.calls == [ + ('upload_media', 'image', 'attachment.image', png_data), + ('reply_image', 'req-1', 'media-1'), + ] + + +@pytest.mark.asyncio +async def test_send_message_sends_text_and_skips_proactive_image(): + bot = Bot() + adapter = make_adapter(bot) + jpg_data = b'\xff\xd8\xffimage' + image_b64 = base64.b64encode(jpg_data).decode('utf-8') + chain = platform_message.MessageChain( + [ + platform_message.Plain(text='before'), + platform_message.Image(base64=f'data:image/jpeg;base64,{image_b64}'), + platform_message.Plain(text='after'), + ] + ) + + await adapter.send_message('group', 'chat-1', chain) + + assert bot.calls == [ + ('send_message', 'chat-1', 'beforeafter'), + ] diff --git a/tests/unit_tests/plugin/test_connector_reconcile.py b/tests/unit_tests/plugin/test_connector_reconcile.py index b390b6c5d..0a230e656 100644 --- a/tests/unit_tests/plugin/test_connector_reconcile.py +++ b/tests/unit_tests/plugin/test_connector_reconcile.py @@ -107,6 +107,19 @@ def shared_connector( return connector +@pytest.mark.asyncio +async def test_shared_reconcile_uses_configured_cold_start_timeout(): + binding = execution_binding("workspace-a") + setting = plugin_setting("01", "a" * 64) + connector = shared_connector([[binding]], {"workspace-a": [setting]}) + connector.ap.instance_config.data["plugin"]["connect_timeout_seconds"] = 900 + connector.handler = runtime_handler() + + await connector._prepare_connected_runtime() + + assert connector.handler.reconcile_plugin_installations.await_args.kwargs["timeout"] == 900 + + @pytest.mark.asyncio async def test_shared_reconnect_replays_two_workspaces_and_removes_missing_projection(): binding_a = execution_binding('workspace-a') @@ -150,7 +163,7 @@ async def test_empty_projected_workspaces_do_not_retain_installation_sets(): assert connector._workspace_installations == {} assert connector._known_desired_states == {} - connector.handler.reconcile_plugin_installations.assert_awaited_once_with(()) + connector.handler.reconcile_plugin_installations.assert_awaited_once_with((), timeout=300.0) @pytest.mark.asyncio diff --git a/tests/unit_tests/plugin/test_handler.py b/tests/unit_tests/plugin/test_handler.py index d6779a11a..860d16989 100644 --- a/tests/unit_tests/plugin/test_handler.py +++ b/tests/unit_tests/plugin/test_handler.py @@ -81,6 +81,18 @@ async def test_reconcile_plugin_installations_allows_cloud_cold_start_to_finish( assert runtime_handler.call_action.await_args.kwargs['timeout'] == 300 +@pytest.mark.asyncio +async def test_reconcile_plugin_installations_accepts_configured_cold_start_timeout(): + runtime_handler = make_handler(SimpleNamespace()) + runtime_handler.call_action = AsyncMock(return_value={}) + binding = next(iter(runtime_handler._installation_bindings.values()))[0] + desired = PluginInstallationDesiredState(binding=binding, enabled=True) + + await runtime_handler.reconcile_plugin_installations((desired,), timeout=900) + + assert runtime_handler.call_action.await_args.kwargs["timeout"] == 900 + + class TestHandlerQueryVariables: """Tests for handler query variable logic.""" diff --git a/tests/unit_tests/plugin/test_handler_actions.py b/tests/unit_tests/plugin/test_handler_actions.py index 83c2af24a..be3784c4c 100644 --- a/tests/unit_tests/plugin/test_handler_actions.py +++ b/tests/unit_tests/plugin/test_handler_actions.py @@ -277,6 +277,7 @@ class TestSetBinaryStorage: }, } mock_app.persistence_mgr = Mock() + mock_app.persistence_mgr.get_db_engine.return_value = SimpleNamespace(dialect=SimpleNamespace(name='sqlite')) mock_app.persistence_mgr.execute_async = AsyncMock(return_value=make_result()) mock_app.logger = Mock() return mock_app @@ -313,8 +314,8 @@ class TestSetBinaryStorage: ) assert response.code == 0 - assert app.persistence_mgr.execute_async.await_count == 2 - insert_params = compiled_params(app.persistence_mgr.execute_async.await_args_list[1].args[0]) + assert app.persistence_mgr.execute_async.await_count == 3 + insert_params = compiled_params(app.persistence_mgr.execute_async.await_args_list[2].args[0]) assert insert_params['workspace_uuid'] == 'workspace-a' assert insert_params['unique_key'] == canonical_binary_key( 'plugin', @@ -344,6 +345,69 @@ class TestSetBinaryStorage: assert expected_key in update_params.values() assert update_params['value'] == b'new' + @pytest.mark.asyncio + async def test_adopts_legacy_storage_before_updating(self, app): + """A migrated pre-tenancy row is updated in place rather than duplicated.""" + runtime_handler = make_handler(app) + legacy_storage = SimpleNamespace(unique_key='plugin:test-author/test-plugin:test-key') + adopted = SimpleNamespace(rowcount=1) + app.persistence_mgr.execute_async.side_effect = [ + make_result(), + make_result(legacy_storage), + adopted, + ] + + response = await runtime_handler.actions[RuntimeToLangBotAction.SET_BINARY_STORAGE.value](self.payload(b'new')) + + assert response.code == 0 + assert app.persistence_mgr.execute_async.await_count == 3 + adoption_params = compiled_params(app.persistence_mgr.execute_async.await_args_list[2].args[0]) + expected_key = canonical_binary_key('plugin', 'test-author/test-plugin', 'test-key') + assert expected_key in adoption_params.values() + assert adoption_params['value'] == b'new' + + @pytest.mark.asyncio + async def test_legacy_adoption_race_updates_winning_canonical_row(self, app): + runtime_handler = make_handler(app) + legacy_storage = SimpleNamespace(unique_key='plugin:test-author/test-plugin:test-key') + lost_race = SimpleNamespace(rowcount=0) + canonical_winner = SimpleNamespace(rowcount=1) + app.persistence_mgr.execute_async.side_effect = [ + make_result(), + make_result(legacy_storage), + lost_race, + canonical_winner, + ] + + response = await runtime_handler.actions[RuntimeToLangBotAction.SET_BINARY_STORAGE.value](self.payload(b'new')) + + assert response.code == 0 + assert app.persistence_mgr.execute_async.await_count == 4 + winner_update = compiled_params(app.persistence_mgr.execute_async.await_args_list[3].args[0]) + assert canonical_binary_key('plugin', 'test-author/test-plugin', 'test-key') in winner_update.values() + assert winner_update['value'] == b'new' + + @pytest.mark.asyncio + async def test_legacy_adoption_lost_to_delete_inserts_new_value(self, app): + runtime_handler = make_handler(app) + legacy_storage = SimpleNamespace(unique_key='plugin:test-author/test-plugin:test-key') + lost_race = SimpleNamespace(rowcount=0) + app.persistence_mgr.execute_async.side_effect = [ + make_result(), + make_result(legacy_storage), + lost_race, + SimpleNamespace(rowcount=0), + make_result(), + ] + + response = await runtime_handler.actions[RuntimeToLangBotAction.SET_BINARY_STORAGE.value](self.payload(b'new')) + + assert response.code == 0 + assert app.persistence_mgr.execute_async.await_count == 5 + insert_params = compiled_params(app.persistence_mgr.execute_async.await_args_list[4].args[0]) + assert insert_params['unique_key'] == canonical_binary_key('plugin', 'test-author/test-plugin', 'test-key') + assert insert_params['value'] == b'new' + @pytest.mark.asyncio async def test_invalid_max_value_bytes_falls_back_to_default_limit(self, app): """Invalid max_value_bytes uses the 10MB default limit.""" @@ -568,6 +632,46 @@ class TestGetBinaryStorage: in statement_params.values() ) + @pytest.mark.asyncio + async def test_reads_legacy_storage_without_mutating_key(self, app): + runtime_handler = make_handler(app) + legacy_storage = SimpleNamespace( + unique_key='plugin:test-author/test-plugin:test-key', + value=b'legacy bytes', + ) + app.persistence_mgr.execute_async.side_effect = [ + make_result(), + make_result(legacy_storage), + ] + + response = await runtime_handler.actions[RuntimeToLangBotAction.GET_BINARY_STORAGE.value]( + {'key': 'test-key', 'owner_type': 'plugin', 'owner': 'ignored'} + ) + + assert response.code == 0 + assert base64.b64decode(response.data['value_base64']) == b'legacy bytes' + assert app.persistence_mgr.execute_async.await_count == 2 + + @pytest.mark.asyncio + async def test_retries_canonical_after_concurrent_legacy_adoption(self, app): + runtime_handler = make_handler(app) + canonical_storage = SimpleNamespace(value=b'adopted bytes') + app.persistence_mgr.execute_async.side_effect = [ + make_result(), + make_result(), + make_result(canonical_storage), + ] + + response = await runtime_handler.actions[RuntimeToLangBotAction.GET_BINARY_STORAGE.value]( + {'key': 'test-key', 'owner_type': 'plugin', 'owner': 'ignored'} + ) + + assert response.code == 0 + assert base64.b64decode(response.data['value_base64']) == b'adopted bytes' + assert app.persistence_mgr.execute_async.await_count == 3 + retry_params = compiled_params(app.persistence_mgr.execute_async.await_args_list[2].args[0]) + assert canonical_binary_key('plugin', 'test-author/test-plugin', 'test-key') in retry_params.values() + @pytest.mark.asyncio async def test_returns_error_when_not_found(self, app): """Missing binary storage rows return an error response.""" @@ -610,21 +714,47 @@ class TestDeleteAndListBinaryStorage: assert response.code == 0 statement_params = compiled_params(app.persistence_mgr.execute_async.await_args.args[0]) - assert 'workspace-a' in statement_params.values() + flat_values = [ + item for value in statement_params.values() for item in (value if isinstance(value, list) else [value]) + ] + assert 'workspace-a' in flat_values assert ( canonical_binary_key( 'plugin', 'test-author/test-plugin', 'test-key', ) - in statement_params.values() + in flat_values ) - assert 'forged-owner' not in statement_params.values() + assert 'forged-owner' not in flat_values + + @pytest.mark.asyncio + async def test_delete_removes_canonical_and_legacy_scoped_keys(self, app): + runtime_handler = make_handler(app) + + response = await runtime_handler.actions[RuntimeToLangBotAction.DELETE_BINARY_STORAGE.value]( + { + 'key': 'test-key', + 'owner_type': 'plugin', + 'owner': 'forged-owner', + } + ) + + assert response.code == 0 + statement_params = compiled_params(app.persistence_mgr.execute_async.await_args.args[0]) + values = [ + item for value in statement_params.values() for item in (value if isinstance(value, list) else [value]) + ] + assert 'workspace-a' in values + assert canonical_binary_key('plugin', 'test-author/test-plugin', 'test-key') in values + assert 'plugin:test-author/test-plugin:test-key' in values + assert 'test-author/test-plugin' in values + assert 'forged-owner' not in values @pytest.mark.asyncio async def test_list_keys_uses_trusted_plugin_owner(self, app): result = Mock() - result.scalars.return_value.all.return_value = ['first', 'second'] + result.scalars.return_value.all.return_value = ['first', 'second', 'first'] app.persistence_mgr.execute_async.return_value = result runtime_handler = make_handler(app) diff --git a/tests/unit_tests/test_optional_dependencies.py b/tests/unit_tests/test_optional_dependencies.py new file mode 100644 index 000000000..841344906 --- /dev/null +++ b/tests/unit_tests/test_optional_dependencies.py @@ -0,0 +1,15 @@ +from __future__ import annotations + +import tomllib +from pathlib import Path + + +def test_seekdb_is_only_declared_as_an_optional_dependency() -> None: + project_root = Path(__file__).resolve().parents[2] + with (project_root / 'pyproject.toml').open('rb') as pyproject_file: + pyproject = tomllib.load(pyproject_file) + + project = pyproject['project'] + base_dependencies = project['dependencies'] + assert not any(dependency.lower().startswith('pyseekdb') for dependency in base_dependencies) + assert project['optional-dependencies']['seekdb'] == ['pyseekdb==1.1.0.post3'] diff --git a/tests/unit_tests/vector/test_seekdb_optional.py b/tests/unit_tests/vector/test_seekdb_optional.py new file mode 100644 index 000000000..f30e6ac20 --- /dev/null +++ b/tests/unit_tests/vector/test_seekdb_optional.py @@ -0,0 +1,34 @@ +from __future__ import annotations + +import importlib +from unittest.mock import MagicMock + +import pytest + +from tests.utils.import_isolation import isolated_sys_modules + + +_INSTALL_HINT = "Install LangBot with the 'seekdb' extra" + + +def test_seekdb_vector_backend_reports_missing_optional_extra() -> None: + module_name = 'langbot.pkg.vector.vdbs.seekdb' + + with isolated_sys_modules({'pyseekdb': None}, clear=[module_name]): + seekdb_module = importlib.import_module(module_name) + + assert seekdb_module.SEEKDB_AVAILABLE is False + with pytest.raises(ImportError, match=_INSTALL_HINT): + seekdb_module.SeekDBVectorDatabase(MagicMock()) + + +@pytest.mark.asyncio +async def test_seekdb_embedding_reports_missing_optional_extra() -> None: + module_name = 'langbot.pkg.provider.modelmgr.requesters.seekdbembed' + + with isolated_sys_modules({'pyseekdb': None}, clear=[module_name]): + seekdb_embedding_module = importlib.import_module(module_name) + requester = seekdb_embedding_module.SeekDBEmbedding.__new__(seekdb_embedding_module.SeekDBEmbedding) + + with pytest.raises(ImportError, match=_INSTALL_HINT): + await requester.initialize() diff --git a/uv.lock b/uv.lock index 43b05c1f1..607df7225 100644 --- a/uv.lock +++ b/uv.lock @@ -8,10 +8,10 @@ resolution-markers = [ "python_full_version == '3.13.*' and sys_platform == 'emscripten'", "python_full_version == '3.13.*' and sys_platform != 'emscripten' and sys_platform != 'win32'", "python_full_version == '3.12.*' and sys_platform == 'win32'", - "python_full_version < '3.12' and sys_platform == 'win32'", "python_full_version == '3.12.*' and sys_platform == 'emscripten'", - "python_full_version < '3.12' and sys_platform == 'emscripten'", "python_full_version == '3.12.*' and sys_platform != 'emscripten' and sys_platform != 'win32'", + "python_full_version < '3.12' and sys_platform == 'win32'", + "python_full_version < '3.12' and sys_platform == 'emscripten'", "python_full_version < '3.12' and sys_platform != 'emscripten' and sys_platform != 'win32'", ] @@ -2054,7 +2054,6 @@ dependencies = [ { name = "pymilvus" }, { name = "pynacl" }, { name = "pypdf2" }, - { name = "pyseekdb" }, { name = "python-docx" }, { name = "python-multipart" }, { name = "python-socks" }, @@ -2079,6 +2078,11 @@ dependencies = [ { name = "websockets" }, ] +[package.optional-dependencies] +seekdb = [ + { name = "pyseekdb" }, +] + [package.dev-dependencies] dev = [ { name = "moto" }, @@ -2143,7 +2147,7 @@ requires-dist = [ { name = "pymilvus", specifier = ">=2.6.4" }, { name = "pynacl", specifier = ">=1.5.0" }, { name = "pypdf2", specifier = ">=3.0.1" }, - { name = "pyseekdb", specifier = "==1.1.0.post3" }, + { name = "pyseekdb", marker = "extra == 'seekdb'", specifier = "==1.1.0.post3" }, { name = "python-docx", specifier = ">=1.1.0" }, { name = "python-multipart", specifier = ">=0.0.27" }, { name = "python-socks", specifier = ">=2.7.1" }, @@ -2167,6 +2171,7 @@ requires-dist = [ { name = "valkey-glide", marker = "sys_platform != 'win32'", specifier = ">=2.4.1,<3.0.0" }, { name = "websockets", specifier = ">=15.0.1" }, ] +provides-extras = ["seekdb"] [package.metadata.requires-dev] dev = [ diff --git a/web/src/app/infra/http/BackendClient.ts b/web/src/app/infra/http/BackendClient.ts index 2c8f4c4b4..b369aa1b6 100644 --- a/web/src/app/infra/http/BackendClient.ts +++ b/web/src/app/infra/http/BackendClient.ts @@ -1307,6 +1307,7 @@ export class BackendClient extends BaseHttpClient { public getAccountInfo(): Promise<{ initialized: boolean; authenticated_invitation_acceptance_enabled?: boolean; + invitation_registration_enabled?: boolean; password_login_enabled?: boolean; space_login_enabled?: boolean; }> { diff --git a/web/src/app/invitations/accept/page.tsx b/web/src/app/invitations/accept/page.tsx index 0d83331b5..1d9591a54 100644 --- a/web/src/app/invitations/accept/page.tsx +++ b/web/src/app/invitations/accept/page.tsx @@ -91,7 +91,9 @@ export default function AcceptInvitationPage() { const [errorMessage, setErrorMessage] = useState(''); const [password, setPassword] = useState(''); const [confirmPassword, setConfirmPassword] = useState(''); - const [passwordRegistrationEnabled, setPasswordRegistrationEnabled] = + const [invitationRegistrationEnabled, setInvitationRegistrationEnabled] = + useState(false); + const [invitationCapabilitiesLoaded, setInvitationCapabilitiesLoaded] = useState(false); const [ authenticatedInvitationAcceptanceEnabled, @@ -116,12 +118,16 @@ export default function AcceptInvitationPage() { backendClient .getAccountInfo() .then((info) => { - setPasswordRegistrationEnabled(info.password_login_enabled !== false); + setInvitationRegistrationEnabled( + info.invitation_registration_enabled ?? + info.password_login_enabled !== false, + ); setAuthenticatedInvitationAcceptanceEnabled( info.authenticated_invitation_acceptance_enabled === true, ); }) - .catch(() => setPasswordRegistrationEnabled(false)); + .catch(() => setInvitationRegistrationEnabled(false)) + .finally(() => setInvitationCapabilitiesLoaded(true)); if (!invitationToken) { setErrorMessage(t('workspace.invitationMissing')); setStatus('error'); @@ -311,7 +317,11 @@ export default function AcceptInvitationPage() { )} - {hasLoginToken && authenticatedInvitationAcceptanceEnabled ? ( + {!invitationCapabilitiesLoaded ? ( +
+ +
+ ) : hasLoginToken && authenticatedInvitationAcceptanceEnabled ? (