diff --git a/skills/skills/langbot-dev/SKILL.md b/skills/skills/langbot-dev/SKILL.md index dbcdb0625..0ae0dcd1b 100644 --- a/skills/skills/langbot-dev/SKILL.md +++ b/skills/skills/langbot-dev/SKILL.md @@ -130,3 +130,10 @@ uv run python tests/manual/mcp_smoke.py # MCP server e2e smoke - `langbot-deploy` — Docker/compose deployment + config. - `langbot-mcp-ops` — operating the LangBot MCP server. - [`langbot-cli`](https://github.com/langbot-app/langbot-cli) — `lbctl`, a standalone Service API client for managing running Workspaces. + +### Migration installation diagnostics + +Migration remains an HTTP-only operation. Poll its system task handle and inspect +`runtime.result.installations[].diagnostic`: `kind` is an allowlisted network +category and `http_status` is present for HTTP failures. Never expose raw exception +messages, request URLs, headers or response bodies in this diagnostic object. diff --git a/src/langbot/pkg/api/http/service/pipeline_migration.py b/src/langbot/pkg/api/http/service/pipeline_migration.py index 5a3dc296a..2173ff7d4 100644 --- a/src/langbot/pkg/api/http/service/pipeline_migration.py +++ b/src/langbot/pkg/api/http/service/pipeline_migration.py @@ -98,6 +98,23 @@ class MigrationInstallContext(TaskContext): self.progress[key] = min(value, 100) if key == 'progress_percent' else value self.progress['updated_at'] = time.time() + def record_network_failure(self, exc: httpx.HTTPError): + # Deliberate allowlist: never serialize exception messages, URLs, + # headers, proxy credentials or response bodies into task metadata. + kinds = ( + (httpx.ConnectTimeout, 'connect_timeout'), + (httpx.ReadTimeout, 'read_timeout'), + (httpx.WriteTimeout, 'write_timeout'), + (httpx.PoolTimeout, 'pool_timeout'), + (httpx.ConnectError, 'connect_error'), + (httpx.RemoteProtocolError, 'incomplete_response'), + (httpx.HTTPStatusError, 'http_status'), + ) + diagnostic = {'kind': next((name for cls, name in kinds if isinstance(exc, cls)), 'network_error')} + if isinstance(exc, httpx.HTTPStatusError): + diagnostic['http_status'] = exc.response.status_code + self.progress['diagnostic'] = diagnostic + def finish(self, code=None): self.publish() self.progress.update(status='failed' if code else 'completed', code=code, finished_at=time.time()) @@ -562,15 +579,18 @@ class PipelineMigrationService: install_code = 'plugin_runtime_unavailable' except MarketplacePluginVersionNotFoundError: install_code = 'plugin_version_unavailable' - except httpx.TimeoutException: + except httpx.TimeoutException as exc: + install_context.record_network_failure(exc) install_code = 'plugin_download_timeout' except httpx.HTTPStatusError as exc: + install_context.record_network_failure(exc) install_code = ( 'plugin_version_unavailable' if exc.response.status_code == 404 else 'plugin_marketplace_unavailable' ) - except httpx.RequestError: + except httpx.RequestError as exc: + install_context.record_network_failure(exc) install_code = 'plugin_download_failed' except PluginInstallationFailedError as exc: install_code = { diff --git a/src/langbot/pkg/api/mcp/server.py b/src/langbot/pkg/api/mcp/server.py index 527bd00d8..dd51a752e 100644 --- a/src/langbot/pkg/api/mcp/server.py +++ b/src/langbot/pkg/api/mcp/server.py @@ -12,7 +12,9 @@ Scope (first version): core read operations plus the most common writes for bots, pipelines, LLM/embedding models, knowledge bases, MCP servers, skills, and read-only system info. This intentionally does NOT expose every one of the ~25 HTTP route groups — that keeps the agent surface small, safe, and -maintainable. Extend deliberately. +maintainable. Extend deliberately. Pipeline migration remains HTTP-only; its +installation task metadata includes credential-free diagnostic.kind and optional +diagnostic.http_status fields, never raw network exception text. """ from __future__ import annotations diff --git a/src/langbot/pkg/persistence/alembic/versions/0039_execution_lease_indexes.py b/src/langbot/pkg/persistence/alembic/versions/0039_execution_lease_indexes.py index eba466381..e1984874d 100644 --- a/src/langbot/pkg/persistence/alembic/versions/0039_execution_lease_indexes.py +++ b/src/langbot/pkg/persistence/alembic/versions/0039_execution_lease_indexes.py @@ -33,6 +33,10 @@ def upgrade(): for table, index, columns in INDEXES: if not inspector.has_table(table): continue + # Partial historical schemas may predate individual payload/lease columns. + # Index only available columns; do not invent missing application data. + if not set(columns) <= {item['name'] for item in inspector.get_columns(table)}: + continue # Fresh installations create these indexes from ORM metadata before migrations. if index not in {item['name'] for item in inspector.get_indexes(table)}: op.create_index(index, table, columns) diff --git a/src/langbot/pkg/pipeline/controller.py b/src/langbot/pkg/pipeline/controller.py index 11ebeae09..161aecfeb 100644 --- a/src/langbot/pkg/pipeline/controller.py +++ b/src/langbot/pkg/pipeline/controller.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import copy import traceback from ..core import app @@ -59,6 +60,25 @@ class Controller: query.pipeline_config = pipeline.pipeline_entity.config query.variables['_pipeline_bound_plugins'] = pipeline.bound_plugins query.variables['_pipeline_bound_mcp_servers'] = pipeline.bound_mcp_servers + if getattr(getattr(query, 'launcher_type', None), 'value', None) == 'group': + from .resprule.resprule import GroupRespondRuleCheckStage + from .entities import ResultType + + # Busy sessions reach steering before the normal stage chain. + # Check on a copy: matchers strip mentions/prefixes, and a + # failed claim must leave the queued message untouched. + candidate = copy.copy(query) + candidate.message_chain = query.message_chain.model_copy(deep=True) + candidate.variables = dict(query.variables) + gate = GroupRespondRuleCheckStage(self.ap) + await gate.initialize(query.pipeline_config) + result = await gate.process(candidate, 'GroupRespondRuleCheckStage') + random_key = '_group_response_random_draw' + if random_key in candidate.variables: + query.variables[random_key] = candidate.variables[random_key] + if result.result_type == ResultType.INTERRUPT: + return False + query = candidate return await self.ap.agent_run_orchestrator.try_claim_steering_from_query(query) except Exception as exc: self.ap.logger.warning( diff --git a/src/langbot/pkg/pipeline/monitoring_helper.py b/src/langbot/pkg/pipeline/monitoring_helper.py index 25b58ec8b..63a28ed44 100644 --- a/src/langbot/pkg/pipeline/monitoring_helper.py +++ b/src/langbot/pkg/pipeline/monitoring_helper.py @@ -308,6 +308,16 @@ class MonitoringHelper: ) -> str: """Record query processing error, returns message_id""" try: + # Finish the original inbound row before writing optional error + # details. Otherwise the queue can be empty while WebUI says pending. + original_message_id = query.variables.get('_monitoring_message_id') + if original_message_id: + await ap.monitoring_service.update_message_status( + get_query_execution_context(query), + message_id=original_message_id, + status='error', + level='error', + ) session_id = f'{query.launcher_type.value if hasattr(query.launcher_type, "value") else query.launcher_type}_{query.launcher_id}' # Get sender name from message event diff --git a/src/langbot/pkg/pipeline/resprule/rules/random.py b/src/langbot/pkg/pipeline/resprule/rules/random.py index 2bfe8b71c..4eae7cc19 100644 --- a/src/langbot/pkg/pipeline/resprule/rules/random.py +++ b/src/langbot/pkg/pipeline/resprule/rules/random.py @@ -18,4 +18,10 @@ class RandomRespRule(rule_model.GroupRespondRule): ) -> entities.RuleJudgeResult: random_rate = rule_dict['random'] - return entities.RuleJudgeResult(matching=random.random() < random_rate, replacement=message_chain) + # A busy query may be checked for steering several times before its + # normal pipeline turn. Draw once so queue wakeups do not increase + # its chance of being admitted. + key = '_group_response_random_draw' + if key not in query.variables: + query.variables[key] = random.random() + return entities.RuleJudgeResult(matching=query.variables[key] < random_rate, replacement=message_chain) diff --git a/tests/integration/persistence/test_partial_history_postgres.py b/tests/integration/persistence/test_partial_history_postgres.py new file mode 100644 index 000000000..cd9fd241e --- /dev/null +++ b/tests/integration/persistence/test_partial_history_postgres.py @@ -0,0 +1,53 @@ +"""Verify partial history index migration against PostgreSQL, without shared tables.""" + +import importlib +import os +import uuid + +import pytest +import sqlalchemy as sa +from alembic.migration import MigrationContext +from alembic.operations import Operations +from sqlalchemy.ext.asyncio import create_async_engine + +MIGRATION = importlib.import_module('langbot.pkg.persistence.alembic.versions.0039_execution_lease_indexes') + + +@pytest.mark.asyncio +@pytest.mark.parametrize('table,index,columns', MIGRATION.INDEXES) +async def test_partial_history_indexes(table, index, columns): + url = os.environ.get('TEST_POSTGRES_URL') + if not url: + pytest.skip('Disposable TEST_POSTGRES_URL required') + schema = 'partial_' + uuid.uuid4().hex + admin = create_async_engine(url) + engine = create_async_engine(url, connect_args={'server_settings': {'search_path': schema}}) + + def exercise(connection): + metadata = sa.MetaData() + present = set(columns) - {columns[-1]} + model = sa.Table(table, metadata, *(sa.Column(name, sa.String) for name in sorted(present))) + metadata.create_all(connection) + connection.execute(model.insert().values(**{name: 'history' for name in present})) + with Operations.context(MigrationContext.configure(connection)): + MIGRATION.upgrade() + MIGRATION.upgrade() + indexes = {item['name'] for item in sa.inspect(connection).get_indexes(table)} + assert index not in indexes + assert indexes == { + name for target, name, required in MIGRATION.INDEXES if target == table and set(required) <= present + } + assert connection.scalar(sa.select(sa.func.count()).select_from(model)) == 1 + MIGRATION.downgrade() + assert not sa.inspect(connection).get_indexes(table) + + try: + async with admin.begin() as connection: + await connection.execute(sa.text(f'CREATE SCHEMA {schema}')) + async with engine.begin() as connection: + await connection.run_sync(exercise) + finally: + await engine.dispose() + async with admin.begin() as connection: + await connection.execute(sa.text(f'DROP SCHEMA IF EXISTS {schema} CASCADE')) + await admin.dispose() diff --git a/tests/manual/lang17_fixture.py b/tests/manual/lang17_fixture.py new file mode 100644 index 000000000..a7864168a --- /dev/null +++ b/tests/manual/lang17_fixture.py @@ -0,0 +1,116 @@ +"""Real dedicated-worker faults and bounded OneBot load with synthetic QA runners.""" + +import io +import os +import zipfile + +import yaml + +RUNNER = """import asyncio, os, time +from langbot_plugin.api.definition.components.runner.runner import Runner +from langbot_plugin.api.entities.builtin.provider.message import Message, ContentElement +from langbot_plugin.api.entities.builtin.runner import RunnerResult +class Probe(Runner): + async def run(self, ctx): + text = ctx.input.to_text() + if 'RAISE' in text: raise RuntimeError('LANG17 handler fault') + if 'EXIT' in text: os._exit(17) + if 'BLOCK' in text: + until = time.monotonic() + 3 + while time.monotonic() < until: pass + if 'TIMEOUT' in text: await asyncio.wait_for(asyncio.sleep(3), 0.2) + if 'SLOW' in text: await asyncio.sleep(0.1) + if 'WAIT' in text: await asyncio.sleep(60) + if 'BUSY' in text: + await asyncio.sleep(6) + pulled = await ctx.steering_pull() + yield RunnerResult.message_completed(ctx.run_id, Message(role='assistant', content='STEERING:' + str(pulled.model_dump()))) + yield RunnerResult.run_completed(ctx.run_id) + return + content = [ContentElement.from_image_base64('data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+jRZkAAAAASUVORK5CYII=')] if 'IMAGE' in text else 'LANG17_HEALTHY' + yield RunnerResult.message_completed(ctx.run_id, Message(role='assistant', content=content)) + yield RunnerResult.run_completed(ctx.run_id) +""" + + +LISTENER = """import asyncio, os, time +from langbot_plugin.api.definition.components.common.event_listener import EventListener +from langbot_plugin.api.entities import events +from langbot_plugin.api.entities.builtin.platform.message import MessageChain, Plain +class ProbeListener(EventListener): + async def initialize(self): + @self.handler(events.GroupNormalMessageReceived) + async def handle(ctx): + text = ctx.event.text_message + name = 'PLUGIN_NAME' + if name == 'FaultProbe': + if 'LRAISE' in text: raise RuntimeError('LANG17 listener fault') + if 'LTIMEOUT' in text: await asyncio.wait_for(asyncio.sleep(3), 0.2) + if 'LEXIT' in text: os._exit(19) + if 'LBLOCK' in text: + until = time.monotonic() + 3 + while time.monotonic() < until: pass + if 'CONTRACT' in text: + previous = str(ctx.event.reply_message_chain or '') + ctx.event.reply_message_chain = MessageChain([Plain(text=previous + name + '>')]) + ctx.prevent_default() + if 'STOP' in text: ctx.prevent_postorder() +""" + + +def package(name, init_failure=False, dependency_failure=False): + manifest = { + 'apiVersion': 'v1', + 'kind': 'Plugin', + 'metadata': { + 'author': 'lang17-qa', + 'name': name, + 'version': '0.0.1', + 'label': {'en_US': name}, + 'description': {'en_US': 'Isolated synthetic QA fixture'}, + 'icon': 'assets/icon.svg', + }, + 'spec': {'config': [], 'components': {'Runner': {'fromDirs': [{'path': 'components/runner/'}]}}}, + 'execution': {'python': {'path': 'main.py', 'attr': 'ProbePlugin'}}, + } + component = { + 'apiVersion': 'langbot/v1', + 'kind': 'Runner', + 'metadata': {'name': 'default', 'label': {'en_US': 'Probe'}, 'description': {'en_US': 'No model calls'}}, + 'spec': { + 'usages': ['agent'], + 'config': [], + 'capabilities': {'streaming': True, 'tool_calling': False, 'steering': True}, + 'permissions': {}, + }, + 'execution': {'python': {'path': 'default.py', 'attr': 'Probe'}}, + } + main = 'from langbot_plugin.api.definition.plugin import BasePlugin\nclass ProbePlugin(BasePlugin):\n async def initialize(self):\n ' + main += "raise RuntimeError('LANG17 initialization fault')\n" if init_failure else 'pass\n' + listener_files = {} + if os.environ.get('LANG17_LISTENERS') == '1': + manifest['spec']['components']['EventListener'] = {'fromDirs': [{'path': 'components/event_listener/'}]} + listener_manifest = { + 'apiVersion': 'v1', + 'kind': 'EventListener', + 'metadata': {'name': 'events', 'label': {'en_US': 'QA events'}}, + 'spec': {}, + 'execution': {'python': {'path': 'events.py', 'attr': 'ProbeListener'}}, + } + listener_files = { + 'components/event_listener/events.yaml': yaml.safe_dump(listener_manifest), + 'components/event_listener/events.py': LISTENER.replace('PLUGIN_NAME', name), + } + stream = io.BytesIO() + with zipfile.ZipFile(stream, 'w') as z: + for path, data in { + **listener_files, + 'manifest.yaml': yaml.safe_dump(manifest), + 'main.py': main, + 'components/runner/default.yaml': yaml.safe_dump(component), + 'components/runner/default.py': RUNNER, + 'requirements.txt': 'not a valid requirement === !!!' if dependency_failure else '', + 'assets/icon.svg': '', + }.items(): + z.writestr(path, data) + return stream.getvalue() diff --git a/tests/manual/lang17_install.py b/tests/manual/lang17_install.py new file mode 100644 index 000000000..b4d83e1e8 --- /dev/null +++ b/tests/manual/lang17_install.py @@ -0,0 +1,78 @@ +"""Real marketplace install through isolated Core APIs; no model calls.""" + +import asyncio +import json +from pathlib import Path +import secrets +import sqlite3 +import subprocess +import os +import time + +import httpx + + +async def main(): + mode = os.environ['LANG17_INSTALL'] + result = {'mode': mode, 'requested_plugin': 'langbot-team/LocalAgent@0.2.1'} + async with httpx.AsyncClient(base_url='http://127.0.0.1:16399', timeout=20, trust_env=False) as client: + + async def api(method, path, data=None): + response = await client.request(method, '/api/v1' + path, json=data) + body = response.json() + if response.status_code >= 400 or body.get('code') != 0: + raise RuntimeError(f'{path}: HTTP {response.status_code}: {body.get("msg")}') + return body.get('data', {}) + + try: + account = {'user': 'lang17-install@example.invalid', 'password': secrets.token_urlsafe(24)} + await api('POST', '/user/init', account) + auth = await api('POST', '/user/auth', account) + client.headers['Authorization'] = 'Bearer ' + auth['token'] + if mode == 'migration': + pipeline = (await api('POST', '/pipelines', {'name': '4.10.10 synthetic', 'description': 'QA'}))['uuid'] + repo = Path(__file__).resolve().parents[2] + old_config = subprocess.check_output( + ['git', 'show', 'v4.10.10:src/langbot/templates/default-pipeline-config.json'], cwd=repo + ).decode() + with sqlite3.connect('data/langbot.db') as db: + db.execute( + 'update legacy_pipelines set config=?, for_version=? where uuid=?', + (old_config, '4.10.10', pipeline), + ) + preview = await api('GET', '/pipelines/_/migration/preview') + result['preview'] = [ + {k: v for k, v in item.items() if k != 'preview_token'} for item in preview['items'] + ] + task = await api( + 'POST', '/pipelines/_/migration/execute', {'confirmed': True, 'all': True, 'install_plugins': True} + ) + else: + task = await api( + 'POST', + '/plugins/install/marketplace', + {'plugin_author': 'langbot-team', 'plugin_name': 'LocalAgent', 'plugin_version': '0.2.1'}, + ) + start = time.monotonic() + while time.monotonic() - start < 120: + current = await api('GET', '/system/tasks/' + str(task['task_id'])) + result['task'] = { + 'runtime': {k: v for k, v in current.get('runtime', {}).items() if k != 'exception_traceback'}, + 'context': {k: v for k, v in current.get('context', {}).items() if k != 'log'}, + } + if current.get('runtime', {}).get('done'): + break + await asyncio.sleep(1) + result['seconds'] = round(time.monotonic() - start, 2) + result['plugins'] = await api('GET', '/plugins') + assert result['task']['runtime']['done'], 'Install deadline exceeded' + assert not result['task']['runtime'].get('exception'), 'Install task failed' + assert 'LocalAgent' in str(result['plugins']), 'Plugin missing from actual inventory' + except Exception as exc: + result['probe_error'] = str(exc) + raise + finally: + Path('install-result.json').write_text(json.dumps(result, ensure_ascii=False, indent=2)) + + +asyncio.run(main()) diff --git a/tests/manual/lang17_live.py b/tests/manual/lang17_live.py new file mode 100644 index 000000000..0deba89b9 --- /dev/null +++ b/tests/manual/lang17_live.py @@ -0,0 +1,227 @@ +"""Bounded real Core/RT/OneBot probe; no shared Docker access or external messages.""" + +import json +import os +from pathlib import Path +import secrets +import socket +import subprocess +import sys +import time +import urllib.request + +import psutil +import yaml + +repo = Path(__file__).resolve().parents[2] +source = Path(os.environ.get('LANG17_SOURCE', str(repo))).resolve() +root = repo.parent / 'evidence' / ('live-' + time.strftime('%Y%m%d-%H%M%S')) +root.mkdir(mode=0o700) +ports = [16399, 16400, 16401, 14000, 13280, 13281, 16420, 16410] +for port in ports: + with socket.socket() as sock: + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + sock.bind(('127.0.0.1', port)) +env = {key: os.environ[key] for key in ('PATH', 'HOME', 'LANG') if key in os.environ} +for key in ('XDG_CACHE_HOME', 'UV_CACHE_DIR', 'TMPDIR'): + path = root / key.lower() + path.mkdir() + env[key] = str(path) +env.update( + LANGBOT_DATA_ROOT=str(root / 'data'), + PYTHONPATH=str(source / 'src'), + PYTHONUNBUFFERED='1', + PLAYWRIGHT_BROWSERS_PATH=str(repo.parent / 'evidence/playwright-browsers'), + NO_PROXY='127.0.0.1,localhost', + LANGBOT_PLUGIN_RUNTIME_CONTROL_TOKEN=secrets.token_hex(32), + VITE_API_PROXY_TARGET='http://127.0.0.1:16399', + LANG17_BASE='1' if source != repo else '', + DOCKER_HOST='unix://' + str(root / 'nonexistent-docker.sock'), +) +config = yaml.safe_load((source / 'src/langbot/templates/config.yaml').read_text()) +config['api'].update(host='127.0.0.1', port=16399, webhook_prefix='http://127.0.0.1:16399') +config['plugin'].update(runtime_ws_url='ws://127.0.0.1:16400/control/ws') +config['box']['enabled'] = bool(os.environ.get('LANG17_BOX_ROOT')) +if os.environ.get('LANG17_BOX_ROOT'): + env['PATH'] = str(Path(os.environ['LANG17_DOCKER_BIN']).resolve()) + os.pathsep + env['PATH'] + env['LANG17_TOOL'] = '1' + env['LANGBOT_BOX_CONTROL_TOKEN'] = secrets.token_hex(32) + config['box']['backend'] = 'docker' + config['box']['runtime']['endpoint'] = 'ws://127.0.0.1:16410' + config['box']['local']['host_root'] = str(Path(os.environ['LANG17_BOX_ROOT']).resolve()) + config['box']['local']['allowed_mount_roots'] = [config['box']['local']['host_root']] + config['box']['local']['image'] = 'alpine:3.21' + config['box']['limits']['max_sessions'] = 1 + config['box']['default_memory_mb'] = 256 +config['space']['disable_telemetry'] = True +if os.environ.get('LANG17_NETWORK'): + config['space']['url'] = 'http://127.0.0.1:16420' + config['space']['disable_models_service'] = True +config['database']['sqlite']['path'] = str(root / 'data/langbot.db') +(root / 'data').mkdir() +if os.environ.get('LANG17_SEED_DB'): + import shutil + + seed = Path(os.environ['LANG17_SEED_DB']) + shutil.copy2(seed, root / 'data/langbot.db') + shutil.copytree(seed.parent / 'labels', root / 'data/labels') + env['LANG17_UPGRADE_ACCOUNT'] = str(Path(os.environ['LANG17_UPGRADE_ACCOUNT']).resolve()) +(root / 'data/config.yaml').write_text(yaml.safe_dump(config)) +for key in ('LANG17_LISTENERS', 'LANG17_SHORT', 'LANG17_SOAK_ROUNDS', 'LANG17_BOUND', 'LANG17_INSTALL'): + if key in os.environ: + env[key] = os.environ[key] +duration = int(os.environ.get('LANG17_SECONDS', '150')) +owned, logs, children = {}, [], [] +result = { + 'sha': subprocess.check_output(['git', 'rev-parse', 'HEAD'], cwd=source, text=True).strip(), + 'limits': {'rss_bytes': 2 * 1024**3, 'cpu_percent': 200, 'hot_samples': 5, 'seconds': duration}, + 'box': 'isolated daemon via task Docker wrapper' if os.environ.get('LANG17_BOX_ROOT') else 'disabled', + 'samples': [], + 'health': {}, +} + + +def launch(role, command, cwd): + log = (root / (role + '.log')).open('w') + logs.append(log) + child = subprocess.Popen(command, cwd=cwd, env=env, stdout=log, stderr=subprocess.STDOUT) + children.append(child) + owned[child.pid] = (role, psutil.Process(child.pid)) + return child + + +try: + launch( + 'rt', + [sys.executable, '-m', 'langbot_plugin.cli', 'rt', '--ws-control-port', '16400', '--ws-debug-port', '16401'], + root, + ) + if os.environ.get('LANG17_BOX_ROOT'): + launch( + 'box', + [sys.executable, '-m', 'langbot_plugin.cli', 'box', '--host', '127.0.0.1', '--ws-control-port', '16410'], + root, + ) + core = launch( + 'core', + [sys.executable, str(source / 'main.py'), '--standalone-runtime'] + + (['--standalone-box'] if os.environ.get('LANG17_BOX_ROOT') else []), + root, + ) + launch( + 'web', + [str(repo / 'web/node_modules/.bin/vite'), 'preview', '--host', '127.0.0.1', '--port', '14000', '--strictPort'], + repo / 'web', + ) + started, hot, probe = time.monotonic(), 0, None + while time.monotonic() - started < duration: + if core.poll() is not None: + raise RuntimeError('Core exited') + for role, parent in list(owned.values()): + try: + for child in parent.children(recursive=True): + owned.setdefault(child.pid, (role.split('-child')[0] + '-child', child)) + except psutil.NoSuchProcess: + pass + rows = [] + for pid, (role, proc) in list(owned.items()): + try: + rows.append( + { + 'pid': pid, + 'role': role, + 'rss': proc.memory_info().rss, + 'cpu': proc.cpu_percent(), + 'fds': proc.num_fds(), + 'threads': proc.num_threads(), + } + ) + except psutil.NoSuchProcess: + pass + sample = {'seconds': round(time.monotonic() - started, 2), 'processes': rows} + for role, port in [('core', 16399), ('rt', 16400), ('web', 14000)]: + try: + with urllib.request.urlopen( + f'http://127.0.0.1:{port}' + ('/' if role == 'web' else '/healthz'), timeout=0.5 + ) as response: + body = response.read() + result['health'][role] = response.status + if role != 'web': + sample[role] = json.loads(body) + except Exception: + pass + result['samples'].append(sample) + # Include a short idle baseline before synthetic OneBot traffic. + if probe is None and len(result['health']) == 3 and sample['seconds'] >= 25: + probe = launch( + 'probe', + [ + sys.executable, + str( + repo + / ( + 'tests/manual/lang17_upgrade.py' + if os.environ.get('LANG17_SEED_DB') + else ( + 'tests/manual/lang17_localagent.py' + if os.environ.get('LANG17_LOCALAGENT') + else ( + 'tests/manual/lang17_matrix.py' + if os.environ.get('LANG17_MATRIX') + else ( + 'tests/manual/lang17_network.py' + if os.environ.get('LANG17_NETWORK') + else ( + 'tests/manual/lang17_install.py' + if os.environ.get('LANG17_INSTALL') + else 'tests/manual/lang17_onebot.py' + ) + ) + ) + ) + ) + ), + ] + + (['--fault'] if os.environ.get('LANG17_FAULT') else []), + root, + ) + if probe is not None and probe.poll() is not None: + result['probe_exit_code'] = probe.returncode + if time.monotonic() - started >= int(os.environ.get('LANG17_MIN_SECONDS', '70')): + break + hot = hot + 1 if sum(row['cpu'] for row in rows) > 200 else 0 + if sum(row['rss'] for row in rows) > 2 * 1024**3 or hot >= 5: + raise RuntimeError('Resource threshold') + time.sleep(2) +finally: + daemon = json.loads(subprocess.check_output(['multica', 'daemon', 'status', '--output', 'json']))['pid'] + for pid, (_, proc) in reversed(list(owned.items())): + assert pid != daemon + try: + proc.terminate() + except psutil.NoSuchProcess: + pass + _, alive = psutil.wait_procs([proc for _, proc in owned.values()], timeout=8) + for proc in alive: + assert proc.pid != daemon + proc.kill() + psutil.wait_procs(alive, timeout=3) + for child in children: + child.wait(timeout=3) + for log in logs: + log.close() + result['survivors'] = [pid for pid, (_, proc) in owned.items() if proc.is_running()] + (root / 'metrics.json').write_text(json.dumps(result, indent=2)) + print( + json.dumps( + { + 'directory': root.name, + 'health': result['health'], + 'probe_exit_code': result.get('probe_exit_code'), + 'survivors': result['survivors'], + } + ) + ) + +if result.get('probe_exit_code') != 0: + raise SystemExit('Probe failed or exceeded its deadline') diff --git a/tests/manual/lang17_localagent.py b/tests/manual/lang17_localagent.py new file mode 100644 index 000000000..b0c00b1a0 --- /dev/null +++ b/tests/manual/lang17_localagent.py @@ -0,0 +1,235 @@ +"""Real LocalAgent worker with a loopback model protocol fixture and Box disabled.""" + +import asyncio +import json +import os +from pathlib import Path +import secrets +import time + +from aiohttp import web +import httpx +import websockets + + +async def main(): + calls = [] + tool_mode = bool(os.environ.get('LANG17_TOOL')) + tool_results = [] + box_snapshots = [] + + async def completion(request): + body = await request.json() + calls.append({'message_count': len(body['messages']), 'stream': body.get('stream')}) + base = {'id': 'qa-completion', 'object': 'chat.completion', 'created': int(time.time()), 'model': 'qa-model'} + tool_messages = [message for message in body['messages'] if message.get('role') == 'tool'] + tool_results.extend(message.get('content') for message in tool_messages) + wants_tool = tool_mode and 'tool request' in str(body['messages']) and not tool_messages + delta = {'role': 'assistant', 'content': 'LOCALAGENT_NO_BOX_OK'} + finish = 'stop' + if wants_tool: + name = next(tool['function']['name'] for tool in body['tools'] if 'exec' in tool['function']['name']) + delta = { + 'role': 'assistant', + 'tool_calls': [ + { + 'index': 0, + 'id': 'qa-exec', + 'type': 'function', + 'function': {'name': name, 'arguments': json.dumps({'command': 'printf LANG17_TOOL_OK'})}, + } + ], + } + finish = 'tool_calls' + if body.get('stream'): + response = web.StreamResponse(headers={'Content-Type': 'text/event-stream'}) + await response.prepare(request) + for delta, finish in [(delta, None), ({}, finish)]: + chunk = { + **base, + 'object': 'chat.completion.chunk', + 'choices': [{'index': 0, 'delta': delta, 'finish_reason': finish}], + } + await response.write(('data: ' + json.dumps(chunk) + '\n\n').encode()) + await response.write(b'data: [DONE]\n\n') + return response + return web.json_response( + { + **base, + 'choices': [ + { + 'index': 0, + 'message': {'role': 'assistant', 'content': 'LOCALAGENT_NO_BOX_OK'}, + 'finish_reason': 'stop', + } + ], + 'usage': {'prompt_tokens': 1, 'completion_tokens': 1, 'total_tokens': 2}, + } + ) + + app = web.Application() + app.router.add_post('/v1/chat/completions', completion) + server = web.AppRunner(app) + await server.setup() + site = web.TCPSite(server, '127.0.0.1', 16420) + await site.start() + try: + async with httpx.AsyncClient(base_url='http://127.0.0.1:16399', timeout=30, trust_env=False) as client: + + async def api(method, path, data=None): + response = await client.request(method, '/api/v1' + path, json=data) + response.raise_for_status() + body = response.json() + assert body['code'] == 0, (path, body.get('msg')) + return body.get('data', {}) + + account = {'user': 'qa-localagent@example.invalid', 'password': secrets.token_urlsafe(24)} + await api('POST', '/user/init', account) + client.headers['Authorization'] = 'Bearer ' + (await api('POST', '/user/auth', account))['token'] + package = Path(__file__).resolve().parents[3] / 'evidence/LocalAgent-integration.lbpkg' + response = await client.post( + '/api/v1/plugins/install/local', files={'file': (package.name, package.read_bytes())} + ) + response.raise_for_status() + task = response.json()['data']['task_id'] + for _ in range(60): + state = (await api('GET', '/system/tasks/' + str(task)))['runtime'] + if state['done']: + assert not state.get('exception'), 'Plugin installation failed' + break + await asyncio.sleep(1) + else: + raise AssertionError('Plugin install deadline') + model = await api( + 'POST', + '/provider/models/llm', + { + 'name': 'qa-model', + 'abilities': ['func_call'], + 'provider': { + 'requester': 'openai-chat-completions', + 'base_url': 'http://127.0.0.1:16420/v1', + 'api_keys': ['synthetic-key'], + }, + }, + ) + pipeline = ( + await api('POST', '/pipelines', {'name': 'LocalAgent no Box', 'description': 'protocol fixture'}) + )['uuid'] + config = (await api('GET', '/pipelines/' + pipeline))['pipeline']['config'] + runner = 'plugin:langbot-team/LocalAgent/default' + config['trigger']['group-respond-rules'] = {'at': True, 'prefix': ['!qa'], 'regexp': [], 'random': 0} + config['ai'] = { + 'runner': {'id': runner, 'expire-time': 0}, + 'runner_config': { + runner: { + 'box-enabled': True, + 'model': {'primary': model['uuid'], 'fallbacks': [], 'reasoning': {}}, + 'enable-all-tools': False, + 'tools': ['exec'], + 'max-tool-iterations': 2, + } + }, + } + await api('PUT', '/pipelines/' + pipeline, {'config': config}) + bot = await api( + 'POST', + '/platform/bots', + { + 'name': 'LocalAgent QA', + 'description': 'isolated', + 'adapter': 'aiocqhttp', + 'enable': True, + 'adapter_config': {'host': '127.0.0.1', 'port': 13280, 'access-token': ''}, + 'event_bindings': [ + { + 'id': 'qa', + 'event_pattern': 'message.received', + 'target_type': 'pipeline', + 'target_uuid': pipeline, + 'filters': [], + 'enabled': True, + } + ], + 'plugin_processors': [], + }, + ) + await api('PUT', '/platform/bots/' + bot['uuid'], {'enable': True}) + replies = [] + async with websockets.connect( + 'ws://127.0.0.1:13280/ws/', additional_headers={'X-Self-ID': '900001', 'X-Client-Role': 'Universal'} + ) as ws: + + async def respond(): + async for raw in ws: + request = json.loads(raw) + if request.get('action', '').startswith('send_'): + replies.append(request['params']) + data = ( + {'user_id': 900001, 'nickname': 'QA'} + if request.get('action') == 'get_login_info' + else {'message_id': 1} + ) + await ws.send( + json.dumps({'status': 'ok', 'retcode': 0, 'data': data, 'echo': request.get('echo')}) + ) + + responder = asyncio.create_task(respond()) + try: + for i, text in enumerate( + ['ordinary ignored', '!qa answer plainly'] + (['!qa tool request'] if tool_mode else []) + ): + await ws.send( + json.dumps( + { + 'time': int(time.time()), + 'self_id': 900001, + 'post_type': 'message', + 'message_type': 'group', + 'sub_type': 'normal', + 'message_id': 100 + i, + 'group_id': 900100, + 'user_id': 900002, + 'message': text, + 'raw_message': text, + 'sender': {'user_id': 900002, 'nickname': 'QA', 'role': 'member'}, + } + ) + ) + await asyncio.sleep(3 if i == 0 else 12) + if i == 0: + assert not calls and not replies + if tool_mode: + sessions = await api('GET', '/box/sessions') + box_snapshots.append({'step': i, 'sessions': sessions}) + assert len(sessions) == (1 if i == 2 else 0), box_snapshots + finally: + responder.cancel() + await asyncio.gather(responder, return_exceptions=True) + assert calls and 'LOCALAGENT_NO_BOX_OK' in str(replies), replies + if tool_mode: + assert 'LANG17_TOOL_OK' in str(tool_results), tool_results + Path('localagent-result.json').write_text( + json.dumps( + { + 'model': 'synthetic loopback protocol fixture', + 'calls': calls, + 'replies': replies, + 'box': 'isolated Docker daemon' + if tool_mode + else 'Core disabled; plugin enabled with exec allowed; answer succeeded without Box', + 'box_health': (await client.get('http://127.0.0.1:16410/healthz')).json() + if tool_mode + else None, + 'box_snapshots': box_snapshots, + 'tool_results': tool_results, + 'health': (await client.get('/healthz')).json(), + }, + indent=2, + ) + ) + finally: + await server.cleanup() + + +asyncio.run(main()) diff --git a/tests/manual/lang17_matrix.py b/tests/manual/lang17_matrix.py new file mode 100644 index 000000000..b204c797e --- /dev/null +++ b/tests/manual/lang17_matrix.py @@ -0,0 +1,365 @@ +"""Real dedicated-worker faults and bounded OneBot load with synthetic QA runners.""" + +import asyncio +import io +import json +import os +from pathlib import Path +import secrets +import sqlite3 +import time +import zipfile + +import httpx +import websockets +import yaml + +RUNNER = """import asyncio, os, time +from langbot_plugin.api.definition.components.runner.runner import Runner +from langbot_plugin.api.entities.builtin.provider.message import Message +from langbot_plugin.api.entities.builtin.runner import RunnerResult +class Probe(Runner): + async def run(self, ctx): + text = ctx.input.to_text() + if 'RAISE' in text.split(): raise RuntimeError('LANG17 handler fault') + if 'EXIT' in text.split(): os._exit(17) + if 'BLOCK' in text.split(): + until = time.monotonic() + 3 + while time.monotonic() < until: pass + if 'TIMEOUT' in text.split(): await asyncio.wait_for(asyncio.sleep(3), 0.2) + if 'SLOW' in text.split(): await asyncio.sleep(0.1) + if 'WAIT' in text.split(): await asyncio.sleep(60) + yield RunnerResult.message_completed(ctx.run_id, Message(role='assistant', content='LANG17_HEALTHY')) + yield RunnerResult.run_completed(ctx.run_id) +""" + + +LISTENER = """import asyncio, os, time +from langbot_plugin.api.definition.components.common.event_listener import EventListener +from langbot_plugin.api.entities import events +from langbot_plugin.api.entities.builtin.platform.message import MessageChain, Plain +class ProbeListener(EventListener): + async def initialize(self): + @self.handler(events.GroupNormalMessageReceived) + async def handle(ctx): + text = ctx.event.text_message + name = 'PLUGIN_NAME' + if name == 'FaultProbe': + if 'LRAISE' in text: raise RuntimeError('LANG17 listener fault') + if 'LTIMEOUT' in text: await asyncio.wait_for(asyncio.sleep(3), 0.2) + if 'LEXIT' in text: os._exit(19) + if 'LBLOCK' in text: + until = time.monotonic() + 3 + while time.monotonic() < until: pass + if 'CONTRACT' in text: + previous = str(ctx.event.reply_message_chain or '') + ctx.event.reply_message_chain = MessageChain([Plain(text=previous + name + '>')]) + ctx.prevent_default() + if 'STOP' in text: ctx.prevent_postorder() +""" + + +def package(name, init_failure=False, dependency_failure=False): + manifest = { + 'apiVersion': 'v1', + 'kind': 'Plugin', + 'metadata': { + 'author': 'lang17-qa', + 'name': name, + 'version': '0.0.1', + 'label': {'en_US': name}, + 'description': {'en_US': 'Isolated synthetic QA fixture'}, + 'icon': 'assets/icon.svg', + }, + 'spec': {'config': [], 'components': {'Runner': {'fromDirs': [{'path': 'components/runner/'}]}}}, + 'execution': {'python': {'path': 'main.py', 'attr': 'ProbePlugin'}}, + } + component = { + 'apiVersion': 'langbot/v1', + 'kind': 'Runner', + 'metadata': {'name': 'default', 'label': {'en_US': 'Probe'}, 'description': {'en_US': 'No model calls'}}, + 'spec': { + 'usages': ['agent'], + 'config': [], + 'capabilities': {'streaming': True, 'tool_calling': False}, + 'permissions': {}, + }, + 'execution': {'python': {'path': 'default.py', 'attr': 'Probe'}}, + } + main = 'from langbot_plugin.api.definition.plugin import BasePlugin\nclass ProbePlugin(BasePlugin):\n async def initialize(self):\n ' + main += "raise RuntimeError('LANG17 initialization fault')\n" if init_failure else 'pass\n' + listener_files = {} + if os.environ.get('LANG17_LISTENERS') == '1': + manifest['spec']['components']['EventListener'] = {'fromDirs': [{'path': 'components/event_listener/'}]} + listener_manifest = { + 'apiVersion': 'v1', + 'kind': 'EventListener', + 'metadata': {'name': 'events', 'label': {'en_US': 'QA events'}}, + 'spec': {}, + 'execution': {'python': {'path': 'events.py', 'attr': 'ProbeListener'}}, + } + listener_files = { + 'components/event_listener/events.yaml': yaml.safe_dump(listener_manifest), + 'components/event_listener/events.py': LISTENER.replace('PLUGIN_NAME', name), + } + stream = io.BytesIO() + with zipfile.ZipFile(stream, 'w') as z: + for path, data in { + **listener_files, + 'manifest.yaml': yaml.safe_dump(manifest), + 'main.py': main, + 'components/runner/default.yaml': yaml.safe_dump(component), + 'components/runner/default.py': RUNNER, + 'requirements.txt': 'not a valid requirement === !!!' if dependency_failure else '', + 'assets/icon.svg': '', + }.items(): + z.writestr(path, data) + return stream.getvalue() + + +async def main(): + results = { + 'synthetic_plugins': True, + 'stages': [], + 'sent': 0, + 'listeners': os.environ.get('LANG17_LISTENERS') == '1', + 'explicit_plugin_bindings': os.environ.get('LANG17_BOUND') == '1', + } + async with httpx.AsyncClient(base_url='http://127.0.0.1:16399', timeout=15, trust_env=False) as client: + + async def api(method, path, data=None): + r = await client.request(method, '/api/v1' + path, json=data) + b = r.json() + if r.status_code >= 400 or b.get('code') != 0: + raise RuntimeError(f'{path}: HTTP {r.status_code}: {b.get("msg")}') + return b.get('data', {}) + + async def snapshot(label): + health = (await client.get('/healthz')).json() + list_started = time.monotonic() + try: + plugin_list = await api('GET', '/plugins') + listing = {'ok': True, 'count': len(plugin_list.get('plugins', []))} + except Exception as exc: + listing = {'ok': False, 'error_type': type(exc).__name__} + listing['seconds'] = round(time.monotonic() - list_started, 3) + with sqlite3.connect('data/langbot.db') as db: + states = list(db.execute('select role,status,count(*) from monitoring_messages group by role,status')) + active = await api('GET', '/monitoring/executions?status=running&source=pipeline') + results['stages'].append( + { + 'label': label, + 'monotonic': time.monotonic(), + 'health': health, + 'message_states': states, + 'plugin_list': listing, + 'api_running_queries': active['total'], + } + ) + Path('matrix-result.json').write_text(json.dumps(results, indent=2)) + + async def install(name, **kwargs): + response = await client.post( + '/api/v1/plugins/install/local', files={'file': (name + '.lbpkg', package(name, **kwargs))} + ) + task = response.json()['data']['task_id'] + for _ in range(45): + state = (await api('GET', '/system/tasks/' + str(task)))['runtime'] + if state['done']: + return {'name': name, 'done': True, 'failed': bool(state.get('exception'))} + await asyncio.sleep(1) + return {'name': name, 'done': False} + + account = {'user': 'qa-matrix@example.invalid', 'password': secrets.token_urlsafe(24)} + await api('POST', '/user/init', account) + client.headers['Authorization'] = 'Bearer ' + (await api('POST', '/user/auth', account))['token'] + results['installations'] = [await install('HealthyProbe'), await install('FaultProbe')] + + async def bot(name, port): + pipeline = (await api('POST', '/pipelines', {'name': name, 'description': 'QA fixture'}))['uuid'] + config = (await api('GET', '/pipelines/' + pipeline))['pipeline']['config'] + config['trigger']['group-respond-rules'] = {'at': True, 'prefix': ['qa:'], 'regexp': [], 'random': 0} + runner = 'plugin:lang17-qa/' + name + '/default' + config['ai'] = {'runner': {'id': runner, 'expire-time': 0}, 'runner_config': {runner: {}}} + payload = {'config': config} + if os.environ.get('LANG17_BOUND') == '1': + payload['extensions_preferences'] = { + 'enable_all_plugins': False, + 'enable_all_mcp_servers': True, + 'plugins': [{'author': 'lang17-qa', 'name': name}], + 'mcp_servers': [], + } + await api('PUT', '/pipelines/' + pipeline, payload) + b = await api( + 'POST', + '/platform/bots', + { + 'name': name, + 'description': 'QA', + 'adapter': 'aiocqhttp', + 'enable': True, + 'adapter_config': {'host': '127.0.0.1', 'port': port, 'access-token': ''}, + 'event_bindings': [ + { + 'id': 'default', + 'event_pattern': 'message.received', + 'target_type': 'pipeline', + 'target_uuid': pipeline, + 'filters': [], + 'enabled': True, + } + ], + 'plugin_processors': [], + }, + ) + await api('PUT', '/platform/bots/' + b['uuid'], {'enable': True}) + + await bot('HealthyProbe', 13280) + await bot('FaultProbe', 13281) + await asyncio.sleep(1) + async with ( + websockets.connect( + 'ws://127.0.0.1:13280/ws/', additional_headers={'X-Self-ID': '910001', 'X-Client-Role': 'Universal'} + ) as healthy, + websockets.connect( + 'ws://127.0.0.1:13281/ws/', additional_headers={'X-Self-ID': '910001', 'X-Client-Role': 'Universal'} + ) as fault, + ): + replies = [] + + async def responder(ws, owner): + async for raw in ws: + r = json.loads(raw) + if r.get('action') == 'send_msg': + replies.append( + {'owner': owner, 'time': time.monotonic(), 'message': r.get('params', {}).get('message')} + ) + await ws.send( + json.dumps({'status': 'ok', 'retcode': 0, 'data': {'message_id': 1}, 'echo': r.get('echo')}) + ) + + responders = [ + asyncio.create_task(responder(healthy, 'healthy')), + asyncio.create_task(responder(fault, 'fault')), + ] + + async def send(ws, text, group=910100): + results['sent'] += 1 + await ws.send( + json.dumps( + { + 'time': int(time.time()), + 'self_id': 910001, + 'post_type': 'message', + 'message_type': 'group', + 'sub_type': 'normal', + 'message_id': results['sent'], + 'group_id': group, + 'user_id': 910002, + 'message': text, + 'raw_message': text, + 'sender': {'user_id': 910002, 'nickname': 'QA', 'role': 'member'}, + } + ) + ) + + try: + await send(healthy, 'qa: HEALTHY') + await asyncio.sleep(2) + await snapshot('healthy_baseline') + assert any('LANG17_HEALTHY' in str(r['message']) for r in replies), 'Healthy runner did not execute' + if os.environ.get('LANG17_LISTENERS') == '1': + for text in ('CONTRACT', 'CONTRACT STOP'): + first = len(replies) + await send(healthy, 'qa: ' + text, 910600) + await asyncio.sleep(2) + results.setdefault('contracts', {})[text] = replies[first:] + expected = 'HealthyProbe>' if 'STOP' in text else 'HealthyProbe>FaultProbe>' + actual = [ + element['data']['text'] + for reply in replies[first:] + for element in reply['message'] + if element['type'] == 'text' + ] + assert actual == [expected], (text, actual) + + for kind in ('LRAISE', 'LTIMEOUT', 'LBLOCK', 'LEXIT'): + results.setdefault('fault_started', {})[kind] = time.monotonic() + await send(fault, 'qa: ' + kind, 910601) + await asyncio.sleep(0.3) + await send(healthy, 'qa: HEALTHY control', 910101) + await asyncio.sleep(5) + await snapshot(kind) + await asyncio.sleep(3) + if os.environ.get('LANG17_CANCEL') == '1': + await send(healthy, 'qa: WAIT', 910500) + await asyncio.sleep(1) + await snapshot('before_cancel') + Path('cancel-ready').touch() + return + for kind in ('RAISE', 'TIMEOUT', 'BLOCK'): + results.setdefault('fault_started', {})[kind] = time.monotonic() + await send(fault, 'qa: ' + kind) + await asyncio.sleep(0.3) + await send(healthy, 'qa: HEALTHY control', 910101) + await asyncio.sleep(5) + await snapshot(kind) + await send(fault, 'qa: SLOW', 910400) + await asyncio.sleep(0.05) + await fault.close() + await asyncio.sleep(6) + await snapshot('onebot_disconnect') + async with websockets.connect( + 'ws://127.0.0.1:13281/ws/', + additional_headers={'X-Self-ID': '910001', 'X-Client-Role': 'Universal'}, + ) as reconnected: + responders.append(asyncio.create_task(responder(reconnected, 'fault_reconnected'))) + await send(reconnected, 'qa: HEALTHY reconnect', 910401) + await asyncio.sleep(2) + await snapshot('onebot_reconnected') + results.setdefault('fault_started', {})['EXIT'] = time.monotonic() + await send(reconnected, 'qa: EXIT', 910402) + await asyncio.sleep(0.3) + await send(healthy, 'qa: HEALTHY control', 910101) + await asyncio.sleep(5) + await snapshot('EXIT') + if os.environ.get('LANG17_BOUND') == '1' or os.environ.get('LANG17_SHORT') == '1': + results['complete'] = True + results['replies'] = replies + Path('matrix-result.json').write_text(json.dumps(results, indent=2)) + return + results['installations'] += [ + await install('InitFailure', init_failure=True), + await install('DependencyFailure', dependency_failure=True), + ] + await send(healthy, 'qa: HEALTHY after installs') + await asyncio.sleep(3) + await snapshot('installation_faults') + for _ in range(160): + await send(healthy, 'qa: SLOW', 910200) + await asyncio.sleep(1) + await snapshot('burst_peak') + await asyncio.sleep(25) + await snapshot('burst_drained') + for i in range(int(os.environ.get('LANG17_SOAK_ROUNDS', '6'))): + for _ in range(50): + await send(healthy, 'ordinary unrelated chat', 910300) + await send(healthy, 'qa: HEALTHY soak', 910301) + await asyncio.sleep(10) + await snapshot('soak_' + str(i)) + await asyncio.sleep(20) + await snapshot('idle_recovery') + results['complete'] = True + results['replies'] = replies + Path('matrix-result.json').write_text(json.dumps(results, indent=2)) + finally: + for task in responders: + task.cancel() + await asyncio.gather(*responders, return_exceptions=True) + Path('matrix-finished').touch() + + +try: + asyncio.run(main()) +finally: + Path('matrix-finished').touch() diff --git a/tests/manual/lang17_network.py b/tests/manual/lang17_network.py new file mode 100644 index 000000000..9ffdbc9d8 --- /dev/null +++ b/tests/manual/lang17_network.py @@ -0,0 +1,111 @@ +"""Real migration/extension API failures against a synthetic local marketplace.""" + +import asyncio +import json +from pathlib import Path +import secrets +import sqlite3 +import subprocess +import threading +import time +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +import httpx + +mode = 'http503' + + +class Handler(BaseHTTPRequestHandler): + def log_message(self, *args): + pass + + def do_GET(self): + if mode == 'timeout': + time.sleep(17) + if mode == 'disconnect': + self.connection.close() + return + self.send_response(503 if mode == 'http503' else 200) + self.send_header('Content-Length', '9999' if mode == 'truncated' else '0') + self.end_headers() + if mode == 'truncated': + self.wfile.write(b'incomplete package') + self.wfile.flush() + self.close_connection = True + + +async def main(): + global mode + server = ThreadingHTTPServer(('127.0.0.1', 16420), Handler) + threading.Thread(target=server.serve_forever, daemon=True).start() + results = [] + try: + async with httpx.AsyncClient(base_url='http://127.0.0.1:16399', timeout=20, trust_env=False) as client: + + async def api(method, path, data=None): + r = await client.request(method, '/api/v1' + path, json=data) + b = r.json() + if r.status_code >= 400 or b.get('code') != 0: + raise RuntimeError(f'{path}: HTTP {r.status_code}') + return b.get('data', {}) + + account = {'user': 'qa-network@example.invalid', 'password': secrets.token_urlsafe(24)} + await api('POST', '/user/init', account) + token = (await api('POST', '/user/auth', account))['token'] + client.headers['Authorization'] = 'Bearer ' + token + pipeline = ( + await api('POST', '/pipelines', {'name': 'Network failure probe', 'description': 'isolated QA'}) + )['uuid'] + old = subprocess.check_output( + ['git', 'show', 'v4.10.10:src/langbot/templates/default-pipeline-config.json'], + cwd=Path(__file__).resolve().parents[2], + ).decode() + with sqlite3.connect('data/langbot.db') as db: + db.execute( + 'update legacy_pipelines set config=?,for_version=? where uuid=?', (old, '4.10.10', pipeline) + ) + for scenario in ('http503', 'disconnect', 'truncated', 'timeout', 'offline'): + mode = scenario + if scenario == 'offline': + server.shutdown() + server.server_close() + for route in ('migration', 'extension'): + start = time.monotonic() + if route == 'migration': + task = await api( + 'POST', + '/pipelines/_/migration/execute', + {'confirmed': True, 'all': True, 'install_plugins': True}, + ) + else: + task = await api( + 'POST', + '/plugins/install/marketplace', + {'plugin_author': 'langbot-team', 'plugin_name': 'LocalAgent', 'plugin_version': '0.2.1'}, + ) + for _ in range(45): + state = await api('GET', '/system/tasks/' + str(task['task_id'])) + if state['runtime']['done']: + break + await asyncio.sleep(1) + runtime = state['runtime'] + assert runtime['done'], (scenario, route, 'task failed to terminate') + results.append( + { + 'scenario': scenario, + 'route': route, + 'seconds': round(time.monotonic() - start, 2), + 'done': runtime['done'], + 'state': runtime['state'], + 'exception_type': str(runtime.get('exception', '')).split(':')[0][:150], + 'result': runtime.get('result'), + } + ) + Path('network-result.json').write_text(json.dumps(results, indent=2)) + finally: + server.shutdown() + server.server_close() + Path('network-finished').touch() + + +asyncio.run(main()) diff --git a/tests/manual/lang17_onebot.py b/tests/manual/lang17_onebot.py new file mode 100644 index 000000000..db17607f4 --- /dev/null +++ b/tests/manual/lang17_onebot.py @@ -0,0 +1,288 @@ +"""Real isolated Core/OneBot ingress probe with a synthetic protocol peer, no QQ traffic.""" + +import asyncio +import os +import json +from pathlib import Path +import secrets +import sqlite3 +import time +import sys +import subprocess +import datetime + +import httpx +import websockets +from lang17_fixture import package + + +async def main(): + async with httpx.AsyncClient(base_url='http://127.0.0.1:16399', timeout=15, trust_env=False) as client: + + async def api(method, path, data=None): + response = await client.request(method, '/api/v1' + path, json=data) + body = response.json() + if response.status_code >= 400 or body.get('code') != 0: + raise RuntimeError(f'{method} {path}: {response.status_code} {body.get("msg")}') + return body.get('data', {}) + + account = {'user': 'lang17@example.invalid', 'password': secrets.token_urlsafe(24)} + await api('POST', '/user/init', account) + auth = await api('POST', '/user/auth', account) + client.headers['Authorization'] = 'Bearer ' + auth['token'] + await api('POST', '/system/wizard/completed', {'status': 'skipped'}) + response = await client.post( + '/api/v1/plugins/install/local', files={'file': ('HealthyProbe.lbpkg', package('HealthyProbe'))} + ) + response.raise_for_status() + task = response.json()['data']['task_id'] + for _ in range(60): + state = (await api('GET', '/system/tasks/' + str(task)))['runtime'] + if state['done']: + assert not state.get('exception'), state.get('exception') + break + await asyncio.sleep(1) + else: + raise RuntimeError('Plugin installation timed out') + pipeline = (await api('POST', '/pipelines', {'name': 'LANG17 response gate', 'description': 'isolated probe'}))[ + 'uuid' + ] + config = (await api('GET', '/pipelines/' + pipeline))['pipeline']['config'] + config['trigger']['group-respond-rules'] = {'at': True, 'prefix': ['!qa'], 'regexp': []} + if '--fault' not in sys.argv: + config['trigger']['group-respond-rules']['random'] = 0 + runner = 'plugin:lang17-qa/HealthyProbe/default' + config['ai'] = {'runner': {'id': runner, 'expire-time': 0}, 'runner_config': {runner: {}}} + await api('PUT', '/pipelines/' + pipeline, {'config': config}) + bot = await api( + 'POST', + '/platform/bots', + { + 'name': 'LANG17 OneBot', + 'description': 'synthetic protocol peer only', + 'adapter': 'aiocqhttp', + 'enable': True, + 'adapter_config': {'host': '127.0.0.1', 'port': 13280, 'access-token': ''}, + 'event_bindings': [ + { + 'id': 'default-migrated', + 'event_pattern': 'message.received', + 'target_type': 'pipeline', + 'target_uuid': pipeline, + 'filters': [], + 'enabled': True, + 'description': 'Migrated from default pipeline binding', + } + ], + 'plugin_processors': [], + }, + ) + # The release create API loads but does not run the adapter; applying + # enable through the update API starts it without restarting Core. + await api('PUT', '/platform/bots/' + bot['uuid'], {'enable': True}) + await asyncio.sleep(1) + async with websockets.connect( + 'ws://127.0.0.1:13280/ws/', + additional_headers={ + 'X-Self-ID': '900001', + 'X-Client-Role': 'Universal', + }, + ) as ws: + actions = [] + sent_payloads = [] + box_snapshots = ( + [{'case': 'before', 'sessions': await api('GET', '/box/sessions')}] if '--box' in sys.argv else [] + ) + + async def respond(): + async for raw in ws: + request = json.loads(raw) + actions.append(request.get('action')) + if str(request.get('action')).startswith('send_'): + sent_payloads.append(request.get('params', {})) + data = {'message_id': 1} + if request.get('action') == 'get_login_info': + data = {'user_id': 900001, 'nickname': 'LANG17'} + await ws.send(json.dumps({'status': 'ok', 'retcode': 0, 'data': data, 'echo': request.get('echo')})) + + responder = asyncio.create_task(respond()) + cases = [('unrelated', 'ordinary chat'), ('keyword', '!qa hello'), ('mention', '[CQ:at,qq=900001]hello')] + if '--box' in sys.argv: + cases = [('unrelated', 'ordinary chat'), ('keyword', '!qa Reply exactly QA_OK without using tools.')] + if os.environ.get('LANG17_LLM_KEY'): + cases.append(('tool', '!qa Use exec to run printf LANG17_TOOL_OK and report its output.')) + if '--fault' not in sys.argv and '--box' not in sys.argv: + cases += [ + ('other_mention', '[CQ:at,qq=900003]hello'), + ('private', 'hello'), + ('failure', '!qa RAISE'), + ('timeout', '!qa TIMEOUT'), + ('image', '!qa IMAGE'), + ] + try: + for i, (name, content) in enumerate(cases): + await ws.send( + json.dumps( + { + 'time': int(time.time()), + 'self_id': 900001, + 'post_type': 'message', + 'message_type': 'private' if name == 'private' else 'group', + 'sub_type': 'friend' if name == 'private' else 'normal', + 'message_id': 100 + i, + 'group_id': 900100 if '--box' in sys.argv else 900100 + i, + 'user_id': 900002, + 'message': content, + 'raw_message': content, + 'sender': {'user_id': 900002, 'nickname': 'QA', 'card': '', 'role': 'member'}, + } + ) + ) + await asyncio.sleep(20 if os.environ.get('LANG17_LLM_KEY') else (8 if '--box' in sys.argv else 2)) + if '--box' in sys.argv: + box_snapshots.append({'case': name, 'sessions': await api('GET', '/box/sessions')}) + if '--fault' not in sys.argv and '--box' not in sys.argv: + for i, text in enumerate(['!qa BUSY', 'ordinary busy ignored', '!qa accepted followup']): + await ws.send( + json.dumps( + { + 'time': int(time.time()), + 'self_id': 900001, + 'post_type': 'message', + 'message_type': 'group', + 'sub_type': 'normal', + 'message_id': 180 + i, + 'group_id': 900180, + 'user_id': 900002, + 'message': text, + 'raw_message': text, + 'sender': {'user_id': 900002, 'nickname': 'QA', 'role': 'member'}, + } + ) + ) + await asyncio.sleep(1) + await asyncio.sleep(6) + steering = [payload for payload in sent_payloads if 'STEERING:' in str(payload)] + assert len(steering) == 1, steering + assert 'accepted followup' in str(steering), steering + assert 'ordinary busy ignored' not in str(steering), steering + Path('busy-steering.json').write_text(json.dumps(steering, indent=2)) + for i in range(30): + await ws.send( + json.dumps( + { + 'time': int(time.time()), + 'self_id': 900001, + 'post_type': 'message', + 'message_type': 'group', + 'sub_type': 'normal', + 'message_id': 200 + i, + 'group_id': 900200, + 'user_id': 900002, + 'message': 'unrelated burst', + 'raw_message': 'unrelated burst', + 'sender': {'user_id': 900002, 'nickname': 'QA', 'role': 'member'}, + } + ) + ) + await asyncio.sleep(4) + finally: + responder.cancel() + try: + await responder + except asyncio.CancelledError: + pass + health = (await client.get('/healthz')).json() + executions = await api('GET', '/monitoring/executions?source=pipeline') + details = [] + for row in executions['items'][:3]: + detail = await api( + 'GET', + '/monitoring/executions/pipeline/' + + row['id'] + + ('' if os.environ.get('LANG17_BASE') else '?section=overview'), + ) + details.append({'id': row['id'], 'pages': list(detail['pages'])}) + Path('monitoring-api.json').write_text(json.dumps({'executions': executions, 'details': details}, indent=2)) + with sqlite3.connect('data/langbot.db') as db: + tables = [row[0] for row in db.execute("select name from sqlite_master where type='table'")] + monitoring = {} + for table in tables: + if table in ('monitoring_messages', 'monitoring_sessions'): + db.row_factory = sqlite3.Row + monitoring[table] = [dict(row) for row in db.execute('select * from ' + table)] + # Historical input is seeded only into this task-owned database. + with sqlite3.connect('data/langbot.db') as db: + db.row_factory = sqlite3.Row + template = dict(db.execute("SELECT * FROM monitoring_messages WHERE role='user' LIMIT 1").fetchone()) + template.update( + status='pending', + role='user', + execution_owner_id=None, + execution_lease_expires_at=None, + event_id=None, + run_id=None, + session_id='group_history', + timestamp=(datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(hours=1)) + .replace(tzinfo=None) + .isoformat(' '), + message_content=json.dumps([{'type': 'Image', 'base64': 'eA==' * 65536}]), + ) + for i in range(64): + template['id'] = 'history-' + str(i) + db.execute( + 'INSERT INTO monitoring_messages (' + + ','.join(template) + + ') VALUES (' + + ','.join('?' for _ in template) + + ')', + list(template.values()), + ) + history = {'seeded_rows': 64, 'payload_characters_per_row': len(template['message_content']), 'samples': []} + for _ in range(3): + started = time.monotonic() + sessions = await api('GET', '/monitoring/sessions?botId=' + bot['uuid']) + session_seconds = time.monotonic() - started + started = time.monotonic() + async with client.stream('GET', '/api/v1/monitoring/in-flight/stream') as stream: + stream.raise_for_status() + async for line in stream.aiter_lines(): + if line.startswith('data: '): + snapshot = json.loads(line[6:]) + break + history['samples'].append( + { + 'sessions_seconds': session_seconds, + 'snapshot_seconds': time.monotonic() - started, + 'snapshot': snapshot, + 'session_count': sessions['total'], + } + ) + Path('history-api.json').write_text(json.dumps(history, indent=2)) + if not os.environ.get('LANG17_BASE'): + auth_path = Path('browser-auth.json') + auth_path.write_text(json.dumps({'token': auth['token']})) + auth_path.chmod(0o600) + try: + subprocess.run(['node', str(Path(__file__).with_name('lang17_ui.mjs'))], check=True, timeout=35) + finally: + auth_path.unlink() + Path('onebot-result.json').write_text( + json.dumps( + { + 'cases': cases, + 'box_snapshots': box_snapshots, + 'live_model': os.environ.get('LANG17_LLM_MODEL'), + 'synthetic_runner_real_runtime': True, + 'actions': actions, + 'health': health, + 'monitoring': monitoring, + 'pipeline': pipeline, + 'bot': bot, + }, + indent=2, + ) + ) + + +asyncio.run(main()) diff --git a/tests/manual/lang17_postgres.py b/tests/manual/lang17_postgres.py new file mode 100644 index 000000000..2c7eb9da0 --- /dev/null +++ b/tests/manual/lang17_postgres.py @@ -0,0 +1,78 @@ +"""Run media contracts in a recorded, bounded, disposable PostgreSQL container.""" + +import json +import os +from pathlib import Path +import secrets +import socket +import subprocess +import sys +import time + +repo = Path(__file__).resolve().parents[2] +evidence = repo.parent / 'evidence' +with socket.socket() as sock: + sock.bind(('127.0.0.1', 0)) + port = sock.getsockname()[1] +password = secrets.token_hex(20) +env = dict(os.environ, POSTGRES_PASSWORD=password) +cid = subprocess.check_output( + [ + 'docker', + 'run', + '-d', + '--name', + 'lang17-pg-' + secrets.token_hex(4), + '--label', + 'qa.task=LANG-17', + '--memory', + '512m', + '--cpus', + '1', + '--pids-limit', + '128', + '-e', + 'POSTGRES_PASSWORD', + '-p', + f'127.0.0.1:{port}:5432', + 'pgvector/pgvector:pg16', + ], + env=env, + text=True, +).strip() +(evidence / 'owned-postgres.json').write_text(json.dumps({'id': cid, 'port': port})) +try: + for _ in range(30): + ready = subprocess.run(['docker', 'exec', cid, 'pg_isready', '-U', 'postgres'], capture_output=True) + if ready.returncode == 0: + break + time.sleep(1) + else: + raise RuntimeError('PostgreSQL startup timeout') + env['TEST_POSTGRES_URL'] = f'postgresql+asyncpg://postgres:{password}@127.0.0.1:{port}/postgres' + with (evidence / 'integration-postgres.log').open('w') as log: + run = subprocess.run( + [ + sys.executable, + '-m', + 'pytest', + 'tests/integration/persistence/test_media_reference_postgres.py', + 'tests/integration/persistence/test_migrations_postgres.py', + 'tests/integration/persistence/test_partial_history_postgres.py', + '-q', + '--tb=short', + ], + cwd=repo, + env=env, + stdout=log, + stderr=subprocess.STDOUT, + timeout=240, + ) + print('PostgreSQL media tests exit:', run.returncode) + if run.returncode: + raise RuntimeError('PostgreSQL media tests failed') +finally: + labels = json.loads(subprocess.check_output(['docker', 'inspect', '--format', '{{json .Config.Labels}}', cid])) + assert labels.get('qa.task') == 'LANG-17' + subprocess.run(['docker', 'rm', '-f', cid], check=True, stdout=subprocess.DEVNULL) + print('Removed recorded task container') diff --git a/tests/manual/lang17_summarize.py b/tests/manual/lang17_summarize.py new file mode 100644 index 000000000..f8b776020 --- /dev/null +++ b/tests/manual/lang17_summarize.py @@ -0,0 +1,52 @@ +"""Summarize explicit task run directories without exporting configs or credentials.""" + +import json +from pathlib import Path +import statistics +import sys + + +def summarize(root): + result = {'run': root.name} + if (root / 'metrics.json').exists(): + metrics = json.loads((root / 'metrics.json').read_text()) + result.update( + sha=metrics['sha'], + health=metrics['health'], + survivors=metrics['survivors'], + probe_exit=metrics.get('probe_exit_code'), + ) + samples = metrics['samples'] + result['seconds'] = samples[-1]['seconds'] if samples else 0 + roles = {process['role'] for sample in samples for process in sample['processes']} + result['resources'] = {} + for role in sorted(roles): + rows = [[p for p in sample['processes'] if p['role'] == role] for sample in samples] + rss = [sum(p['rss'] for p in row) / 1024**2 for row in rows if row] + cpu = [sum(p['cpu'] for p in row) for row in rows if row] + result['resources'][role] = { + 'peak_rss_mib': round(max(rss), 2), + 'mean_sampled_cpu_percent': round(statistics.mean(cpu), 2), + } + if (root / 'matrix-result.json').exists(): + matrix = json.loads((root / 'matrix-result.json').read_text()) + result['matrix'] = { + 'complete': matrix.get('complete', False), + 'sent': matrix['sent'], + 'last_stage': matrix['stages'][-1]['label'], + } + result['matrix']['healthy_latency_from_fault_seconds'] = {} + for kind, start in matrix.get('fault_started', {}).items(): + replies = [r for r in matrix.get('replies', []) if r['owner'] == 'healthy' and r['time'] >= start] + result['matrix']['healthy_latency_from_fault_seconds'][kind] = ( + round(replies[0]['time'] - start, 3) if replies else None + ) + last = matrix['stages'][-1] + result['matrix']['final_query_pool'] = last['health']['resources']['query_pool'] + result['matrix']['final_message_states'] = last['message_states'] + result['matrix']['plugin_api_max_seconds'] = max(s['plugin_list']['seconds'] for s in matrix['stages']) + return result + + +if __name__ == '__main__': + print(json.dumps([summarize(Path(path)) for path in sys.argv[1:]], indent=2)) diff --git a/tests/manual/lang17_ui.mjs b/tests/manual/lang17_ui.mjs new file mode 100644 index 000000000..ec7421047 --- /dev/null +++ b/tests/manual/lang17_ui.mjs @@ -0,0 +1,31 @@ +import { createRequire } from 'node:module'; +import { readFile, writeFile } from 'node:fs/promises'; +const require = createRequire(new URL('../../web/package.json', import.meta.url)); +const { chromium } = require('@playwright/test'); +const { token } = JSON.parse(await readFile('browser-auth.json', 'utf8')); +const browser = await chromium.launch({ headless: true }); +try { + const page = await browser.newPage({ viewport: { width: 1280, height: 900 } }); + const errors = []; + const responses = []; + page.on('pageerror', e => errors.push(e.message)); + page.on('response', r => { + if (r.url().includes('/monitoring/')) responses.push({ path: new URL(r.url()).pathname, status: r.status() }); + }); + await page.addInitScript(token => { localStorage.setItem('token', token); localStorage.setItem('i18nextLng', 'en-US'); }, token); + const dataReady = page.waitForResponse(r => r.url().includes('/api/v1/monitoring/executions?') && r.status() === 200, { timeout: 20000 }); + await page.goto('http://127.0.0.1:14000/home/monitoring', { waitUntil: 'domcontentloaded', timeout: 20000 }); + try { + await dataReady; + await page.locator('#monitoring-filter-bot').waitFor({ timeout: 20000 }); + const skipGuide = page.getByRole('button', { name: 'Skip', exact: true }); + if (await skipGuide.isVisible()) await skipGuide.click(); + await page.getByText('LANG17 response gate', { exact: true }).first().waitFor({ timeout: 10000 }); + } finally { + await page.screenshot({ path: 'monitoring.png', fullPage: true }); + await writeFile('browser-result.json', JSON.stringify({ url: page.url(), body: await page.locator('body').innerText(), errors, responses }, null, 2)); + } + if (errors.length) throw new Error('Browser errors; inspect browser-result.json'); +} finally { + await browser.close(); +} diff --git a/tests/manual/lang17_upgrade.py b/tests/manual/lang17_upgrade.py new file mode 100644 index 000000000..a4fe53c99 --- /dev/null +++ b/tests/manual/lang17_upgrade.py @@ -0,0 +1,46 @@ +"""Verify a copy of this task's synthetic 4.10.10 database after real boot.""" + +import asyncio +import json +import os +from pathlib import Path +import sqlite3 + +import httpx + + +async def main(): + async with httpx.AsyncClient(base_url='http://127.0.0.1:16399', trust_env=False) as client: + account_path = Path(os.environ['LANG17_UPGRADE_ACCOUNT']) + if not account_path.is_absolute(): + account_path = Path(__file__).resolve().parents[2] / account_path + account = json.loads(account_path.read_text()) + response = await client.post('/api/v1/user/auth', json=account) + response.raise_for_status() + client.headers['Authorization'] = 'Bearer ' + response.json()['data']['token'] + preview = await client.get('/api/v1/pipelines/_/migration/preview') + preview.raise_for_status() + body = preview.json() + assert body['code'] == 0 + assert body['data']['items'], 'Legacy pipeline must remain available for migration' + health = (await client.get('/healthz')).json() + with sqlite3.connect('data/langbot.db') as db: + revisions = list(db.execute('SELECT version_num FROM alembic_version')) + assert revisions == [('0039_execution_lease_indexes',)], revisions + assert db.execute('SELECT count(*) FROM legacy_pipelines').fetchone()[0] >= 1 + assert db.execute('SELECT count(*) FROM bots').fetchone()[0] >= 1 + Path('upgrade-result.json').write_text( + json.dumps( + { + 'revision': revisions, + 'legacy_pipeline_count': len(body['data']['items']), + 'health': health, + 'login': 'passed', + 'scope': 'schema upgrade and preserved legacy migration preview; runner conversion tested separately', + }, + indent=2, + ) + ) + + +asyncio.run(main()) diff --git a/tests/unit_tests/api/service/test_pipeline_migration.py b/tests/unit_tests/api/service/test_pipeline_migration.py index 362af7209..240fa7cc1 100644 --- a/tests/unit_tests/api/service/test_pipeline_migration.py +++ b/tests/unit_tests/api/service/test_pipeline_migration.py @@ -962,3 +962,31 @@ async def test_missing_marketplace_release_has_specific_migration_result(env): assert all(r['code'] == 'plugin_version_unavailable' for r in metadata['results']) assert metadata['installations'][0]['code'] == 'plugin_version_unavailable' assert 'private-token' not in str(metadata) + + +@pytest.mark.parametrize('kind', ['connect', 'read_timeout', 'truncated', 'http']) +def test_network_diagnostic_is_allowlisted_and_credential_free(kind): + import httpx + from langbot.pkg.api.http.service.pipeline_migration import MigrationInstallContext + + request = httpx.Request('GET', 'https://secret-user:secret-password@example.invalid/package?token=private-token') + errors = { + 'connect': httpx.ConnectError('private-token', request=request), + 'read_timeout': httpx.ReadTimeout('private-token', request=request), + 'truncated': httpx.RemoteProtocolError('private-token', request=request), + 'http': httpx.HTTPStatusError( + 'private-token', request=request, response=httpx.Response(503, request=request, text='private-token') + ), + } + ctx = MigrationInstallContext({'author': 'team', 'name': 'Runner', 'version': '1'}) + ctx.record_network_failure(errors[kind]) + expected = { + 'connect': 'connect_error', + 'read_timeout': 'read_timeout', + 'truncated': 'incomplete_response', + 'http': 'http_status', + } + assert ctx.progress['diagnostic']['kind'] == expected[kind] + assert ctx.progress['diagnostic'].get('http_status') == (503 if kind == 'http' else None) + assert 'private-token' not in str(ctx.progress) + assert 'secret-' not in str(ctx.progress) diff --git a/tests/unit_tests/persistence/test_partial_history_indexes.py b/tests/unit_tests/persistence/test_partial_history_indexes.py new file mode 100644 index 000000000..23e82d95e --- /dev/null +++ b/tests/unit_tests/persistence/test_partial_history_indexes.py @@ -0,0 +1,36 @@ +"""Partial historical tables must not block unrelated maintenance indexes.""" + +import importlib + +import pytest +import sqlalchemy as sa +from alembic.migration import MigrationContext +from alembic.operations import Operations + +MIGRATION = importlib.import_module('langbot.pkg.persistence.alembic.versions.0039_execution_lease_indexes') + + +@pytest.mark.parametrize('table,index,columns', MIGRATION.INDEXES) +def test_missing_column_preserves_rows_and_defers_only_unavailable_indexes(table, index, columns): + engine = sa.create_engine('sqlite:///:memory:') + metadata = sa.MetaData() + present = set(columns) - {columns[-1]} + model = sa.Table(table, metadata, *(sa.Column(name, sa.String) for name in sorted(present))) + try: + with engine.begin() as connection: + metadata.create_all(connection) + connection.execute(model.insert().values(**{name: 'history' for name in present})) + with Operations.context(MigrationContext.configure(connection)): + MIGRATION.upgrade() + MIGRATION.upgrade() + indexes = {item['name'] for item in sa.inspect(connection).get_indexes(table)} + assert index not in indexes + expected = { + name for target, name, required in MIGRATION.INDEXES if target == table and set(required) <= present + } + assert indexes == expected + assert connection.scalar(sa.select(sa.func.count()).select_from(model)) == 1 + MIGRATION.downgrade() + assert not sa.inspect(connection).get_indexes(table) + finally: + engine.dispose() diff --git a/tests/unit_tests/pipeline/test_controller.py b/tests/unit_tests/pipeline/test_controller.py index b0b10a4d6..dce3b47b5 100644 --- a/tests/unit_tests/pipeline/test_controller.py +++ b/tests/unit_tests/pipeline/test_controller.py @@ -81,6 +81,38 @@ async def test_try_claim_steering_sets_pipeline_context_before_claiming(): app.agent_run_orchestrator.try_claim_steering_from_query.assert_awaited_once_with(query) +@pytest.mark.asyncio +@pytest.mark.parametrize('text,expected', [('ordinary chat', False), ('qa: hello', True)]) +async def test_busy_group_steering_obeys_response_rules(sample_query, text, expected): + from langbot_plugin.api.entities.builtin.provider.session import LauncherTypes, Session + from langbot_plugin.api.entities.builtin.platform.message import MessageChain, Plain + + app = make_app() + app.sess_mgr.get_session.return_value = Session( + instance_uuid='test-instance', + workspace_uuid='test-workspace', + placement_generation=1, + launcher_type=LauncherTypes.GROUP, + launcher_id=sample_query.launcher_id, + ) + pipeline = make_pipeline() + pipeline.pipeline_entity.config['trigger'] = { + 'group-respond-rules': {'at': True, 'prefix': ['qa:'], 'regexp': [], 'random': 0} + } + app.pipeline_mgr.get_pipeline_by_uuid.return_value = pipeline + app.agent_run_orchestrator.try_claim_steering_from_query.return_value = True + sample_query.launcher_type = LauncherTypes.GROUP + sample_query.message_chain = MessageChain([Plain(text=text)]) + claimed = await Controller(app)._try_claim_steering_before_session_slot(sample_query) + assert claimed is expected, app.logger.warning.call_args + assert str(sample_query.message_chain) == text + if expected: + forwarded = app.agent_run_orchestrator.try_claim_steering_from_query.await_args.args[0] + assert str(forwarded.message_chain) == ' hello' + else: + app.agent_run_orchestrator.try_claim_steering_from_query.assert_not_awaited() + + @pytest.mark.asyncio async def test_consumer_transfers_query_from_queue_to_running_task(): app = make_app() diff --git a/tests/unit_tests/pipeline/test_monitoring_failure.py b/tests/unit_tests/pipeline/test_monitoring_failure.py new file mode 100644 index 000000000..869c327f1 --- /dev/null +++ b/tests/unit_tests/pipeline/test_monitoring_failure.py @@ -0,0 +1,24 @@ +"""A stage exception must finish the original monitoring row.""" + +from types import SimpleNamespace +from unittest.mock import AsyncMock, Mock + +import pytest + +from langbot.pkg.pipeline.monitoring_helper import MonitoringHelper +from langbot.pkg.pipeline.pool import get_query_execution_context + + +@pytest.mark.asyncio +async def test_exception_finishes_original_message_even_if_error_detail_write_fails(sample_query): + sample_query.variables['_monitoring_message_id'] = 'original-message' + service = SimpleNamespace( + update_message_status=AsyncMock(), record_message=AsyncMock(side_effect=RuntimeError('detail failure')) + ) + app = SimpleNamespace(monitoring_service=service, logger=Mock()) + await MonitoringHelper.record_query_error( + app, sample_query, 'bot', 'bot', 'pipeline', 'pipeline', KeyError('random') + ) + service.update_message_status.assert_awaited_once_with( + get_query_execution_context(sample_query), message_id='original-message', status='error', level='error' + ) diff --git a/tests/unit_tests/pipeline/test_resprule.py b/tests/unit_tests/pipeline/test_resprule.py index 4032ad6ff..e3211f599 100644 --- a/tests/unit_tests/pipeline/test_resprule.py +++ b/tests/unit_tests/pipeline/test_resprule.py @@ -282,3 +282,17 @@ async def test_incomplete_form_callback_does_not_bypass_response_rules(mock_app, result = await stage.process(sample_query, 'GroupRespondRuleCheckStage') assert result.result_type == entities.ResultType.INTERRUPT + + +@pytest.mark.asyncio +async def test_random_response_draw_is_reused_across_queue_rechecks(sample_query, mock_app, monkeypatch): + from langbot.pkg.pipeline.resprule.rules.random import RandomRespRule + from unittest.mock import Mock + + draw = Mock(side_effect=[0.9, 0.1]) + monkeypatch.setattr('langbot.pkg.pipeline.resprule.rules.random.random.random', draw) + matcher = RandomRespRule(mock_app) + for _ in range(3): + result = await matcher.match('hello', sample_query.message_chain, {'random': 0.5}, sample_query) + assert result.matching is False + draw.assert_called_once() diff --git a/tests/unit_tests/storage/test_inline_media_cleanup.py b/tests/unit_tests/storage/test_inline_media_cleanup.py index e3aa1bafe..820352721 100644 --- a/tests/unit_tests/storage/test_inline_media_cleanup.py +++ b/tests/unit_tests/storage/test_inline_media_cleanup.py @@ -214,8 +214,11 @@ async def test_event_scan_budget_advances_past_foreign_events_without_reading_pa plan = ' '.join( row[3] for row in await connection.exec_driver_sql('EXPLAIN QUERY PLAN ' + statement, parameters) ) - if 'CORRELATED' in plan: + # SQLite 3.53 reports a flattened EXISTS lookup instead of a + # CORRELATED subquery. Classify the query by SQL, not planner wording. + if 'EXISTS (' in statement: assert 'SEARCH agent_run_event USING INTEGER PRIMARY KEY' in plan, plan + assert 'USING INDEX ix_agent_run_run_id' in plan, plan else: assert 'COVERING INDEX ix_agent_run_event_media_ttl' in plan, plan assert 'TEMP B-TREE' not in plan, plan diff --git a/web/src/app/home/pipelines/MigrationInstallProgress.tsx b/web/src/app/home/pipelines/MigrationInstallProgress.tsx index e36bf765d..aa376e9ac 100644 --- a/web/src/app/home/pipelines/MigrationInstallProgress.tsx +++ b/web/src/app/home/pipelines/MigrationInstallProgress.tsx @@ -166,6 +166,16 @@ export default function MigrationInstallProgress({

{item.author}/{item.name}

+ {item.diagnostic && ( +

+ {t('pipelineMigration.networkDiagnostic', { + kind: item.diagnostic.kind, + })} + {item.diagnostic.http_status !== undefined + ? ` · HTTP ${item.diagnostic.http_status}` + : ''} +

+ )} {item.steps.map((step, index) => (