From a4281ab614fd73959efa035394334eff93a52b35 Mon Sep 17 00:00:00 2001 From: huanghuoguoguo <60681390+huanghuoguoguo@users.noreply.github.com> Date: Sun, 18 Oct 2026 11:32:59 +0800 Subject: [PATCH] test(tools): simulate persistent multi-group conversations --- tools/README.md | 86 ++++- tools/mock_llm.py | 37 +- tools/regression_job.py | 30 +- tools/regression_profiles/groups.json | 464 ++++++++++++++++++++++++++ tools/regression_profiles/job.json | 11 +- tools/regression_suite.py | 191 ++++++++++- tools/regression_traffic.py | 245 +++++++++++++- tools/test_mock_llm.py | 13 + tools/test_regression_groups.py | 240 +++++++++++++ tools/test_regression_job.py | 22 +- tools/test_regression_suite.py | 81 +++++ 11 files changed, 1364 insertions(+), 56 deletions(-) create mode 100644 tools/regression_profiles/groups.json create mode 100644 tools/test_regression_groups.py diff --git a/tools/README.md b/tools/README.md index dc365fd8b..d4c3a01e9 100644 --- a/tools/README.md +++ b/tools/README.md @@ -1,4 +1,4 @@ -# Mock LLM for regression tests +# LangBot regression tools Run the deterministic OpenAI Chat Completions endpoint with Python 3.11+, without API credentials or third-party dependencies: @@ -32,6 +32,9 @@ tool result must contain `LANGBOT_BOX_SENTINEL` before the next step is returned Tool-call IDs and fragmented streaming arguments use the normal OpenAI protocol. Scenario progress comes from each request's current conversation turn, so concurrent sessions and retries do not consume a shared script counter. +Tagged group traffic additionally requires the previous user turn and assistant +acknowledgement in the model input, and rejects another group's history. Final +responses carry a per-turn acknowledgement checked by the traffic driver. Add fixed response/tool sequences with `--scenarios path/to/scenarios.json`, using `mock_llm_scenarios.json` as the example. `expect_tool_output` checks a substring @@ -97,8 +100,10 @@ any failed health probe or process/container resource read. [`regression_suite.py`](regression_suite.py) turns a JSON profile into a repeatable resource preparation, traffic, observation, verdict, and cleanup sequence. Start -with [`regression_profiles/normal.json`](regression_profiles/normal.json), which -defines the chat/Box/RAG mix from the ten-minute trial. +with [`regression_profiles/groups.json`](regression_profiles/groups.json) for +continuous group conversations. The earlier +[`normal.json`](regression_profiles/normal.json) profile retains independent CLI +requests; parallel CLI workers alone do not simulate persistent groups. Use a dedicated QA Workspace. First provision the instance with `lbctl install` or your source deployment, enable Docker Box, start the mock and embedding @@ -110,9 +115,9 @@ paths resolve against the profile directory, so adjust them if you move it. Run on the Linux test host: ```sh -python3 tools/regression_suite.py validate --profile tools/regression_profiles/normal.json +python3 tools/regression_suite.py validate --profile tools/regression_profiles/groups.json python3 tools/regression_suite.py run \ - --profile tools/regression_profiles/normal.json --output .qa/regression/run-001 + --profile tools/regression_profiles/groups.json --output .qa/regression/run-001 ``` `validate` checks profile syntax and references without contacting the instance. @@ -124,12 +129,13 @@ The profile controls: - `plugins`: local archives or GitHub/marketplace request files, installed with `lbctl plugin install ... --wait`. -- `resources`: Providers, models, knowledge bases, document ingestion, and - Pipelines, created in dependency order. `${resource_id}` refers to a UUID +- `resources`: Providers, models, knowledge bases, document ingestion, Pipelines, + and Bots, created in dependency order. `${resource_id}` refers to a UUID returned by an earlier creation; `${workspace_uuid}` is the current Workspace. - Pipeline config patches merge with actual server defaults. Existing resources - can instead be supplied through `vars`, leaving `resources` empty. -- `cases`: CLI argument arrays and expected business text. Integer `weight` + Pipeline config patches merge with actual server defaults. Independent CLI + cases can use existing resources supplied through `vars`; group mode creates + its own signed HTTP Bots. +- `cases`: CLI argument arrays or `http_group` settings and expected business text. Integer `weight` values set the mix; `2/1/1/1` means `40/20/20/20`. An optional `body` object becomes the case's request file. All workers share the weighted sequence. - `phases`: each phase has a name, duration, target RPS, worker limit, and optional @@ -137,6 +143,10 @@ The profile controls: end; the concurrency limit can reduce achieved RPS. - `observe`: health URLs, host PIDs, PID files, and Docker container names/IDs. Set PIDs for disjoint process trees to avoid counting descendants twice. + `box_data_root` discovers Box containers mounted below that deployment's data + directory on every sample, including sandboxes created as new groups speak. + The complete Job sets it automatically; on an existing instance, set it to + the instance's Box data directory when sampling dynamic Docker sandboxes. - `thresholds`: minimum achieved/target RPS ratio and maximum P95 seconds per case, in addition to the existing business and health/resource checks. The sample limits are starting acceptance settings, not established product SLOs. @@ -150,7 +160,7 @@ same profile and credentials: ```sh python3 tools/regression_suite.py cleanup \ - --profile tools/regression_profiles/normal.json --output .qa/regression/run-001 + --profile tools/regression_profiles/groups.json --output .qa/regression/run-001 ``` Cleanup checks the Workspace against `owned.json` and records remaining items @@ -161,14 +171,62 @@ target/achieved RPS, phase results, threshold failures, and the final verdict. The `regression_suite.py` entry reuses an existing instance and backend processes. Use the complete lifecycle Job below to own deployment, credentials and backends. -Cases still use the existing text oracle and new-session CLI entry; continuous -conversations, IM adapter acceptance, and resource-leak verdicts need separate -scenarios. Test the suite plumbing on Linux with: +The independent `normal.json` cases use a new-session CLI entry. Group traffic +uses the signed HTTP Bot adapter instead, as described below. Real IM adapter +acceptance and resource-leak verdicts need separate scenarios. Test the suite +plumbing on Linux with: ```sh python3 -m unittest discover -s tools -p test_regression_suite.py -v ``` +### Persistent group traffic + +The group profile creates three HTTP Bots routed to chat, Box, and RAG Pipelines. +Their signed `/bots//sync` ingress produces real `GroupMessage` events and +uses Core's session manager. Four groups follow a `2/1/1` case assignment; +members take turns within each group. The same member IDs appear in different +groups to exercise group isolation. Box session selection uses the group launcher ID. + +In each group phase, `groups` selects the number of stable sessions and +`members_per_group` selects the rotating member identities. `rps` remains the +aggregate target, so a group's target speaking interval is `groups / rps` +seconds. `workers` bounds total in-flight requests independently of group count. +For example, 8 groups at 4 RPS target one message every 2 seconds per group. +Both the Suite and Job accept `--groups` and `--members-per-group` overrides. +The group count must cover the full weighted case sequence (4 in the sample). + +Only one request is in flight per group. A failure or unknown outcome stops that +group without retry while other groups continue. Every selected group must +complete at least two turns in total for the phase to pass. The fake model checks +that the preceding user message and assistant acknowledgement survived in actual +model input, including when a different member speaks next. Cross-group markers +fail the oracle. The normal Box/tool-output and RAG-context checks still apply. + +`group-state.json` carries group IDs, turn numbers and acknowledgements across +phases; existing groups keep their conversations when later phases add groups. +It refuses changed cases or a state left in progress by interruption. Do not +clear that marker to blindly resume: inspect the run and use a fresh output +directory when the last turn's outcome is unknown. Each phase summary includes +group-level turn counts and member IDs alongside case latency and health data. + +The Suite generates the HTTP Bot signing secret and passes it through +`LANGBOT_QA_GROUP_SECRET`. Resource request bodies containing credentials are +stored under `.private/`; do not share that directory. Cases specify `http_group` +with `url`, `secret_env`, and `message` instead of CLI `args`. Core needs no code +changes. The `/sync` response collapses streaming parts: this tests multi-turn +group processing, not real IM delivery or callback streaming order. +Every selected group must also receive traffic in each phase. Group latency +measures the signed HTTP request through its completed reply; independent CLI +latency includes CLI startup. Compare results within the same traffic mode. + +`lbctl` v0.3 may report Bot verification failure after a successful creation +because its safe projection omits event bindings. The Suite records the returned +Bot UUID before reading the full Core Bot API with the endpoint, Workspace and +environment credential confirmed by `lbctl whoami`. It then +checks the masked signing secret and effective route with a signed warmup turn +before load. The creation is not retried. + ## Complete lifecycle Job [`regression_job.py`](regression_job.py) owns the deployment as well as the suite. diff --git a/tools/mock_llm.py b/tools/mock_llm.py index a3e6c2da8..6a97cb3fa 100644 --- a/tools/mock_llm.py +++ b/tools/mock_llm.py @@ -17,6 +17,39 @@ import uuid MAX_BODY_BYTES = 1024 * 1024 USAGE = {'prompt_tokens': 10, 'completion_tokens': 5, 'total_tokens': 15} +GROUP_TURN = re.compile(r'LB_GROUP=(\S+) LB_TURN=(\d+) LB_TOKEN=([a-f0-9]+) LB_PREVIOUS=(none|[a-f0-9]+)') + + +def group_ack(messages, start): + """Require a real previous user turn and reply, with no other group's history.""" + turns = [] + for index, message in enumerate(messages): + if message.get('role') == 'user': + for match in GROUP_TURN.finditer(str(message.get('content', ''))): + turns.append((index, *match.groups())) + if not turns: + return '' + position, group, turn, token, previous = turns[-1] + if position != start or any(item[1] != group for item in turns): + raise ValueError('group history contains another group or lacks the current turn') + if int(turn) == 1: + if len(turns) != 1 or previous != 'none': + raise ValueError('first group turn must have an empty history') + else: + if len(turns) < 2: + raise ValueError('previous group turn is missing from model history') + prior_position, _, prior_turn, prior_token, _ = turns[-2] + replies = messages[prior_position + 1 : start] + if ( + int(prior_turn) != int(turn) - 1 + or prior_token != previous + or not any( + message.get('role') == 'assistant' and f'GROUP_ACK={previous}' in str(message.get('content', '')) + for message in replies + ) + ): + raise ValueError('previous group turn or assistant acknowledgement does not match') + return f' GROUP_ACK={token}' def mock_embeddings(body): @@ -67,6 +100,7 @@ def select_message(body, scenario): ) if start is None: raise ValueError('a user message is required') + acknowledgement = group_ack(messages, start) turn = messages[start:] calls = [i for i, m in enumerate(turn) if m.get('role') == 'assistant' and m.get('tool_calls')] for index, position in enumerate(calls): @@ -88,7 +122,8 @@ def select_message(body, scenario): result_text = '\n'.join(str(m.get('content', '')) for m in results if m.get('role') == 'tool') if expected not in result_text: raise ValueError(f'tool result must contain {expected!r}') - message = {'role': 'assistant', 'content': step.get('content')} + content = step.get('content') + message = {'role': 'assistant', 'content': content + acknowledgement if content is not None else None} if tool_calls := step.get('tool_calls'): tools = body.get('tools', []) if not isinstance(tools, list) or any( diff --git a/tools/regression_job.py b/tools/regression_job.py index b763c9563..1fc46afd0 100644 --- a/tools/regression_job.py +++ b/tools/regression_job.py @@ -17,6 +17,7 @@ import urllib.request import yaml from regression_suite import Suite, number, resolve, validate +from regression_traffic import owned_box_containers KEY_ENV = 'LANGBOT_QA_API_KEY' @@ -222,6 +223,7 @@ class Job: raise RuntimeError('Docker daemon has existing Box sandboxes; use an idle dedicated test daemon') key = secrets.token_urlsafe(32) self.env[KEY_ENV] = key + self.env['LANGBOT_API_KEY'] = key self.redactions.append(key) (self.private / 'api-key').write_text(key) (self.private / 'api-key').chmod(0o600) @@ -301,6 +303,7 @@ class Job: ) self.save() profile = self.suite_profile + profile.setdefault('vars', {})['core_base_url'] = endpoint profile['cli'] = {'executable': self.cli[0], 'config': str(self.cli_config)} profile['observe'] = {'health': health, 'pids': {}, 'containers': list(containers.values())} for service, container in containers.items(): @@ -309,15 +312,18 @@ class Job: destination = self.output / 'suite' destination.mkdir(mode=0o700) suite = Suite(profile, self.suite_path, destination, env=self.env) + self.redactions.extend(suite.secrets) suite.write('profile.json', profile) suite.prepare() # Prime every configured business case before measuring steady load. for case in profile['cases']: case = resolve(case, suite.variables) + if 'http_group' in case: + continue # Suite preparation already verifies signed group ingress and the masked secret. data = suite.call(case['args'], case.get('body')) if case['expect'] not in json.dumps(data, ensure_ascii=False): raise RuntimeError(f'warmup oracle failed: {case["name"]}') - profile['observe']['containers'] += self.owned_sandboxes() + profile['observe']['box_data_root'] = str(Path(self.state['record']['data_dir']) / 'box') suite.write('profile.json', profile) return suite @@ -325,16 +331,7 @@ class Job: record = self.state.get('record') if not record: return [] - root = (Path(record['data_dir']) / 'box').resolve() - candidates = self.checked( - ['docker', 'ps', '-a', '--filter', 'label=langbot.box=true', '--format', '{{.ID}}'] - ).splitlines() - owned = [] - for container in candidates: - mounts = json.loads(self.checked(['docker', 'inspect', '--format', '{{json .Mounts}}', container])) - if any(Path(mount['Source']).resolve().is_relative_to(root) for mount in mounts if mount.get('Source')): - owned.append(container) - return owned + return owned_box_containers(Path(record['data_dir']) / 'box', self.checked) def collect(self): self.recover_record() @@ -424,6 +421,8 @@ def main(): parser.add_argument('--duration', type=float) parser.add_argument('--rps', type=float) parser.add_argument('--workers', type=int) + parser.add_argument('--groups', type=int) + parser.add_argument('--members-per-group', type=int) args = parser.parse_args() try: if args.action == 'cleanup': @@ -435,6 +434,7 @@ def main(): if (job.private / 'api-key').exists(): key = (job.private / 'api-key').read_text() job.env[KEY_ENV] = key + job.env['LANGBOT_API_KEY'] = key job.redactions.append(key) job.cleanup() return @@ -442,7 +442,13 @@ def main(): raise ValueError('--profile is required for run/validate') profile, suite, path = load_profile(args.profile.resolve()) for phase in suite['phases']: - for flag, field in [('duration', 'duration_seconds'), ('rps', 'rps'), ('workers', 'workers')]: + for flag, field in [ + ('duration', 'duration_seconds'), + ('rps', 'rps'), + ('workers', 'workers'), + ('groups', 'groups'), + ('members_per_group', 'members_per_group'), + ]: if getattr(args, flag) is not None: phase[field] = getattr(args, flag) validate(suite) diff --git a/tools/regression_profiles/groups.json b/tools/regression_profiles/groups.json new file mode 100644 index 000000000..dd6d57e87 --- /dev/null +++ b/tools/regression_profiles/groups.json @@ -0,0 +1,464 @@ +{ + "version": 1, + "cli": { + "executable": "lbctl", + "config": "./lbctl.yaml" + }, + "vars": { + "plugin_dir": "/path/to/plugin-zips", + "mock_base_url": "http://127.0.0.1:18080/v1", + "embedding_base_url": "http://127.0.0.1:15410/v1", + "core_health": "http://127.0.0.1:5300/healthz", + "core_base_url": "http://127.0.0.1:5300" + }, + "plugins": [ + { + "source": "local", + "file": "${plugin_dir}/LocalAgent.zip" + }, + { + "source": "local", + "file": "${plugin_dir}/LangRAG.zip" + }, + { + "source": "local", + "file": "${plugin_dir}/GeneralParsers.zip" + } + ], + "resources": [ + { + "id": "mock_provider", + "kind": "provider", + "body": { + "name": "regression mock", + "requester": "openai-chat-completions", + "base_url": "${mock_base_url}", + "api_keys": [ + "test-only" + ] + } + }, + { + "id": "embedding_provider", + "kind": "provider", + "body": { + "name": "regression embedding", + "requester": "openai-chat-completions", + "base_url": "${embedding_base_url}", + "api_keys": [ + "test-only" + ] + } + }, + { + "id": "chat_model", + "kind": "model", + "type": "llm", + "body": { + "name": "mock-chat", + "provider_uuid": "${mock_provider}", + "abilities": [ + "func_call" + ], + "context_length": 8192, + "extra_args": { + "request_policy": { + "retries": 0 + } + } + } + }, + { + "id": "box_model", + "kind": "model", + "type": "llm", + "body": { + "name": "mock-box", + "provider_uuid": "${mock_provider}", + "abilities": [ + "func_call" + ], + "context_length": 8192, + "extra_args": { + "request_policy": { + "retries": 0 + } + } + } + }, + { + "id": "rag_model", + "kind": "model", + "type": "llm", + "body": { + "name": "mock-rag", + "provider_uuid": "${mock_provider}", + "abilities": [ + "func_call" + ], + "context_length": 8192, + "extra_args": { + "request_policy": { + "retries": 0 + } + } + } + }, + { + "id": "embedding_model", + "kind": "model", + "type": "embedding", + "body": { + "name": "all-MiniLM-L6-v2", + "provider_uuid": "${embedding_provider}" + } + }, + { + "id": "knowledge_base", + "kind": "knowledge-base", + "body": { + "name": "regression parser golden", + "knowledge_engine_plugin_id": "langbot-team/LangRAG", + "creation_settings": { + "embedding_model_uuid": "${embedding_model}", + "index_type": "chunk", + "chunk_size": 512, + "overlap": 50 + }, + "retrieval_settings": { + "search_type": "vector", + "top_k": 5, + "query_rewrite": "off", + "rerank": "off" + } + }, + "documents": [ + { + "file": "../../skills/skills/langbot-testing/fixtures/rag/parser-golden.html", + "parser_plugin_id": "langbot-team/GeneralParsers" + } + ] + }, + { + "id": "chat_pipeline", + "kind": "pipeline", + "body": { + "name": "regression chat", + "description": "Normal deterministic chat", + "config": { + "ai": { + "runner": { + "id": "plugin:langbot-team/LocalAgent/default", + "expire-time": 0 + }, + "runner_config": { + "plugin:langbot-team/LocalAgent/default": { + "model": { + "primary": "${chat_model}", + "fallbacks": [] + }, + "prompt": [ + { + "role": "system", + "content": "You are a QA regression assistant." + } + ], + "knowledge-bases": [], + "box-enabled": false, + "timeout": 30, + "max-round": 20, + "box-session-id-template": "{launcher_type}_{launcher_id}" + } + } + }, + "safety": { + "rate-limit": { + "limitation": 10000 + } + }, + "output": { + "force-delay": { + "min": 0, + "max": 0 + }, + "misc": { + "at-sender": false, + "quote-origin": false + } + }, + "trigger": { + "group-respond-rules": { + "prefix": [ + "ai " + ] + } + } + } + } + }, + { + "id": "box_pipeline", + "kind": "pipeline", + "body": { + "name": "regression Box", + "description": "Real Box write and read", + "config": { + "ai": { + "runner": { + "id": "plugin:langbot-team/LocalAgent/default", + "expire-time": 0 + }, + "runner_config": { + "plugin:langbot-team/LocalAgent/default": { + "model": { + "primary": "${box_model}", + "fallbacks": [] + }, + "prompt": [ + { + "role": "system", + "content": "You are a QA regression assistant." + } + ], + "knowledge-bases": [], + "box-enabled": true, + "box-session-id-template": "{launcher_type}_{launcher_id}", + "timeout": 30, + "max-tool-iterations": 5, + "max-round": 20 + } + } + }, + "safety": { + "rate-limit": { + "limitation": 10000 + } + }, + "output": { + "force-delay": { + "min": 0, + "max": 0 + }, + "misc": { + "at-sender": false, + "quote-origin": false + } + }, + "trigger": { + "group-respond-rules": { + "prefix": [ + "ai " + ] + } + } + } + } + }, + { + "id": "rag_pipeline", + "kind": "pipeline", + "body": { + "name": "regression RAG", + "description": "Retrieved model input oracle", + "config": { + "ai": { + "runner": { + "id": "plugin:langbot-team/LocalAgent/default", + "expire-time": 0 + }, + "runner_config": { + "plugin:langbot-team/LocalAgent/default": { + "model": { + "primary": "${rag_model}", + "fallbacks": [] + }, + "prompt": [ + { + "role": "system", + "content": "You are a QA regression assistant." + } + ], + "knowledge-bases": [ + "${knowledge_base}" + ], + "box-enabled": false, + "timeout": 30, + "retrieval-top-k": 5, + "max-round": 20, + "box-session-id-template": "{launcher_type}_{launcher_id}" + } + } + }, + "safety": { + "rate-limit": { + "limitation": 10000 + } + }, + "output": { + "force-delay": { + "min": 0, + "max": 0 + }, + "misc": { + "at-sender": false, + "quote-origin": false + } + }, + "trigger": { + "group-respond-rules": { + "prefix": [ + "ai " + ] + } + } + } + } + }, + { + "id": "chat_bot", + "kind": "bot", + "body": { + "name": "regression group chat", + "description": "Synthetic signed group ingress", + "adapter": "http_bot", + "enable": true, + "adapter_config": { + "inbound_secret": "${group_secret}", + "signature_required": true, + "callback_url": "http://127.0.0.1:9/unused-sync-only", + "default_session_type": "group", + "callback_timeout": 20, + "callback_max_retries": 0 + }, + "event_bindings": [ + { + "event_pattern": "message.received", + "target_type": "pipeline", + "target_uuid": "${chat_pipeline}", + "enabled": true, + "filters": [] + } + ], + "plugin_processors": [] + } + }, + { + "id": "box_bot", + "kind": "bot", + "body": { + "name": "regression group box", + "description": "Synthetic signed group ingress", + "adapter": "http_bot", + "enable": true, + "adapter_config": { + "inbound_secret": "${group_secret}", + "signature_required": true, + "callback_url": "http://127.0.0.1:9/unused-sync-only", + "default_session_type": "group", + "callback_timeout": 20, + "callback_max_retries": 0 + }, + "event_bindings": [ + { + "event_pattern": "message.received", + "target_type": "pipeline", + "target_uuid": "${box_pipeline}", + "enabled": true, + "filters": [] + } + ], + "plugin_processors": [] + } + }, + { + "id": "rag_bot", + "kind": "bot", + "body": { + "name": "regression group rag", + "description": "Synthetic signed group ingress", + "adapter": "http_bot", + "enable": true, + "adapter_config": { + "inbound_secret": "${group_secret}", + "signature_required": true, + "callback_url": "http://127.0.0.1:9/unused-sync-only", + "default_session_type": "group", + "callback_timeout": 20, + "callback_max_retries": 0 + }, + "event_bindings": [ + { + "event_pattern": "message.received", + "target_type": "pipeline", + "target_uuid": "${rag_pipeline}", + "enabled": true, + "filters": [] + } + ], + "plugin_processors": [] + } + } + ], + "cases": [ + { + "name": "chat", + "weight": 2, + "http_group": { + "url": "${core_base_url}/bots/${chat_bot}/sync", + "secret_env": "LANGBOT_QA_GROUP_SECRET", + "message": "ai What is the documented parser-rag sentinel?" + }, + "expect": "LANGBOT_CHAT_OK" + }, + { + "name": "box", + "weight": 1, + "http_group": { + "url": "${core_base_url}/bots/${box_bot}/sync", + "secret_env": "LANGBOT_QA_GROUP_SECRET", + "message": "ai What is the documented parser-rag sentinel?" + }, + "expect": "LANGBOT_BOX_OK" + }, + { + "name": "rag_chat", + "weight": 1, + "http_group": { + "url": "${core_base_url}/bots/${rag_bot}/sync", + "secret_env": "LANGBOT_QA_GROUP_SECRET", + "message": "ai What is the documented parser-rag sentinel?" + }, + "expect": "LANGBOT_RAG_OK" + } + ], + "phases": [ + { + "name": "normal", + "duration_seconds": 600, + "rps": 2, + "workers": 4, + "cooldown_seconds": 30, + "groups": 4, + "members_per_group": 3 + } + ], + "observe": { + "health": { + "core": "${core_health}", + "mock": "http://127.0.0.1:18080/healthz", + "box": "http://127.0.0.1:5410/readyz" + }, + "pids": {}, + "pid_files": {}, + "containers": [] + }, + "cleanup": "on-success", + "thresholds": { + "min_rps_ratio": 0, + "max_p95_seconds": { + "chat": 5, + "box": 10, + "rag_chat": 6 + } + } +} diff --git a/tools/regression_profiles/job.json b/tools/regression_profiles/job.json index fe31035a9..70ca15c4d 100644 --- a/tools/regression_profiles/job.json +++ b/tools/regression_profiles/job.json @@ -1,9 +1,14 @@ { "version": 1, "lbctl": "lbctl", - "suite": "normal.json", - "deployment": {"version": "v4.11.0", "port": 15351}, - "vars": {"plugin_dir": "/path/to/plugin-zips"}, + "suite": "groups.json", + "deployment": { + "version": "v4.11.0", + "port": 15351 + }, + "vars": { + "plugin_dir": "/path/to/plugin-zips" + }, "mock_image": "python:3.12-alpine", "cleanup": "on-success", "purge_data": true diff --git a/tools/regression_suite.py b/tools/regression_suite.py index 65a685107..ad7e02d96 100644 --- a/tools/regression_suite.py +++ b/tools/regression_suite.py @@ -8,11 +8,17 @@ import math import os from pathlib import Path import re +import secrets import subprocess import sys +import urllib.error +import urllib.request +from uuid import uuid4 + +from regression_traffic import group_request -KINDS = {'provider', 'model', 'pipeline', 'knowledge-base'} +KINDS = {'provider', 'model', 'pipeline', 'knowledge-base', 'bot'} REFERENCE = re.compile(r'\$\{([a-zA-Z][a-zA-Z0-9_-]*)\}') @@ -35,6 +41,21 @@ def merge(base, patch): return result +def contains_fields(actual, expected): + """Check submitted fields while allowing server-generated defaults.""" + if isinstance(expected, dict): + return isinstance(actual, dict) and all( + key in actual and contains_fields(actual[key], value) for key, value in expected.items() + ) + if isinstance(expected, list): + return ( + isinstance(actual, list) + and len(actual) == len(expected) + and all(contains_fields(left, right) for left, right in zip(actual, expected)) + ) + return actual == expected + + def number(value, label, *, minimum=0, integer=False): if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value): raise ValueError(f'{label} must be a finite number') @@ -61,7 +82,7 @@ def validate(profile): raise ValueError('cli accepts config and executable') if not isinstance(cli.get('config'), str) or not cli['config']: raise ValueError('cli.config must name a Workspace-bound lbctl config') - variables = {'profile_dir': '.', 'run_dir': '.', 'workspace_uuid': 'workspace'} + variables = {'profile_dir': '.', 'run_dir': '.', 'workspace_uuid': 'workspace', 'group_secret': 'private'} supplied = profile.get('vars', {}) if not isinstance(supplied, dict) or variables.keys() & supplied.keys(): raise ValueError('vars must be an object without reserved names') @@ -84,6 +105,16 @@ def validate(profile): raise ValueError(f'{name}: resource needs kind and body') if 'uuid' in item['body']: raise ValueError(f'{name}: resources must create new objects; body.uuid is forbidden') + if item['kind'] == 'bot' and ( + item['body'].get('adapter') != 'http_bot' + or not isinstance(item['body'].get('adapter_config'), dict) + or item['body'].get('adapter_config', {}).get('inbound_secret') != '${group_secret}' + or not any( + isinstance(case.get('http_group'), dict) and '${' + name + '}' in case['http_group'].get('url', '') + for case in profile.get('cases', []) + ) + ): + raise ValueError('Bot resources require http_bot, ${group_secret}, and a matching http_group case') if item['kind'] == 'model' and item.get('type') not in {'llm', 'embedding', 'rerank'}: raise ValueError(f'{name}: model type must be llm/embedding/rerank') resolve(item['body'], variables) # References must point backwards; never guess missing UUIDs. @@ -100,12 +131,26 @@ def validate(profile): raise ValueError('cases must be a non-empty array') names = set() for case in cases: - if case.keys() - {'name', 'weight', 'args', 'expect', 'body'}: + if case.keys() - {'name', 'weight', 'args', 'expect', 'body', 'http_group'}: raise ValueError('unknown case field') if not isinstance(case.get('name'), str) or not case['name'] or case['name'] in names: raise ValueError('case names must be non-empty and unique') names.add(case['name']) - if ( + if 'http_group' in case: + settings = case['http_group'] + if ( + 'args' in case + or 'body' in case + or not isinstance(settings, dict) + or set(settings) != {'url', 'secret_env', 'message'} + or any(not isinstance(value, str) or not value for value in settings.values()) + or not resolve(settings['url'], variables).startswith(('http://', 'https://')) + or settings['secret_env'] != 'LANGBOT_QA_GROUP_SECRET' + ): + raise ValueError( + 'http_group needs url, message and secret_env=LANGBOT_QA_GROUP_SECRET; no CLI args/body' + ) + elif ( not isinstance(case.get('args'), list) or not case['args'] or not all(isinstance(x, str) for x in case['args']) @@ -128,7 +173,15 @@ def validate(profile): if not re.fullmatch(r'[a-zA-Z][a-zA-Z0-9_-]*', name) or name in names: raise ValueError('phase names must be unique simple names') names.add(name) - if phase.keys() - {'name', 'duration_seconds', 'rps', 'workers', 'cooldown_seconds'}: + if phase.keys() - { + 'name', + 'duration_seconds', + 'rps', + 'workers', + 'cooldown_seconds', + 'groups', + 'members_per_group', + }: raise ValueError(f'{name}: unknown phase field') number(phase.get('duration_seconds'), 'duration_seconds', minimum=0.001) number(phase.get('rps'), 'rps', minimum=0.001) @@ -136,6 +189,17 @@ def validate(profile): number(phase.get('cooldown_seconds', 0), 'cooldown_seconds') if phase['workers'] > 32: raise ValueError('workers must be <= 32') + number(phase.get('groups', 0), 'groups', integer=True) + number(phase.get('members_per_group', 3), 'members_per_group', minimum=1, integer=True) + if phase.get('groups', 0) > 4096 or phase.get('members_per_group', 3) > 1000: + raise ValueError('groups must be <= 4096 and members_per_group <= 1000') + group_cases = ['http_group' in case for case in cases] + if any(group_cases) and ( + not all(group_cases) or phase.get('groups', 0) < sum(case.get('weight', 1) for case in cases) + ): + raise ValueError('group phases need only http_group cases and enough groups for all weighted cases') + if not any(group_cases) and phase.get('groups', 0): + raise ValueError('groups require http_group cases') if profile.get('cleanup', 'on-success') not in {'always', 'on-success', 'never'}: raise ValueError('cleanup must be always/on-success/never') thresholds = profile.get('thresholds', {}) @@ -151,8 +215,18 @@ def validate(profile): raise ValueError(f'unknown case in max_p95_seconds: {name}') number(value, 'max_p95_seconds', minimum=0.001) observed = resolve(profile.get('observe', {}), variables) - if not isinstance(observed, dict) or observed.keys() - {'health', 'pids', 'pid_files', 'containers'}: - raise ValueError('observe accepts health, pids, pid_files, containers') + if not isinstance(observed, dict) or observed.keys() - { + 'health', + 'pids', + 'pid_files', + 'containers', + 'box_data_root', + }: + raise ValueError('observe accepts health, pids, pid_files, containers, box_data_root') + if 'box_data_root' in observed and ( + not isinstance(observed['box_data_root'], str) or not observed['box_data_root'] + ): + raise ValueError('observe.box_data_root must name the deployment Box data directory') if not isinstance(observed.get('health'), dict) or not observed['health']: raise ValueError('observe.health must contain at least one health endpoint') for name, url in observed['health'].items(): @@ -186,8 +260,14 @@ class Suite: 'json', ] self.owned = [] - self.env = env - self.secrets = [value for key, value in (env or os.environ).items() if 'API_KEY' in key and value] + self.connection = {} + self.env = dict(os.environ if env is None else env) + if any('http_group' in case for case in profile['cases']): + self.variables['group_secret'] = secrets.token_urlsafe(32) + self.env['LANGBOT_QA_GROUP_SECRET'] = self.variables['group_secret'] + self.secrets = [ + value for key, value in self.env.items() if ('API_KEY' in key or key == 'LANGBOT_QA_GROUP_SECRET') and value + ] def path(self, value): path = Path(resolve(value, self.variables)).expanduser() @@ -195,11 +275,15 @@ class Suite: def write(self, name, value): path = self.output / name + if name == 'request.json': + private = self.output / '.private' + private.mkdir(exist_ok=True, mode=0o700) + path = private / name path.write_text(json.dumps(value, ensure_ascii=False, indent=2)) path.chmod(0o600) return str(path) - def call(self, args, body=None): + def call(self, args, body=None, *, defer_bot_readback=False): command = [*self.cli, *args] if body is not None: command += ['--file', self.write('request.json', body)] @@ -208,11 +292,50 @@ class Suite: except subprocess.TimeoutExpired as exc: raise RuntimeError('lbctl timed out; outcome unknown, no automatic retry') from exc if result.returncode: + if defer_bot_readback and args == ['bot', 'create']: + response = json.loads(result.stdout) + data = response.get('data', {}) + if ( + response.get('error', {}).get('type') == 'verification_failed' + and data.get('operation') == 'bot.create' + and data.get('uuid') + and data.get('bot', {}).get('uuid') == data['uuid'] + ): + return data error = result.stdout + result.stderr for secret in self.secrets: error = error.replace(secret, '[REDACTED]') raise RuntimeError(error[:3000]) - return json.loads(result.stdout)['data'] + response = json.loads(result.stdout) + if args == ['whoami']: + self.connection = response.get('meta') or {} + return response['data'] + + def read_bot(self, bot_uuid): + """Read full Bot fields using the endpoint and environment credential confirmed by lbctl.""" + source = self.connection.get('credential_source', '') + key = self.env.get(source.removeprefix('env:')) if source.startswith('env:') else None + if not key or not self.connection.get('endpoint'): + raise RuntimeError('Group Bot readback requires environment-backed lbctl credentials') + self.secrets.append(key) + request = urllib.request.Request( + self.connection['endpoint'].rstrip('/') + '/api/v1/platform/bots/' + bot_uuid, + headers={'Authorization': 'Bearer ' + key, 'X-Workspace-Id': self.variables['workspace_uuid']}, + ) + opener = urllib.request.build_opener(urllib.request.ProxyHandler({})) + try: + with opener.open(request, timeout=30) as response: + payload = json.load(response) + except (urllib.error.URLError, OSError) as exc: + raise RuntimeError('Bot readback request failed; the creation was not retried') from exc + if ( + not isinstance(payload, dict) + or payload.get('code') != 0 + or not isinstance(payload.get('data'), dict) + or not isinstance(payload['data'].get('bot'), dict) + ): + raise RuntimeError('Bot readback did not return a successful Bot resource') + return payload['data']['bot'] def journal(self): self.write('owned.json', {'workspace_uuid': self.variables['workspace_uuid'], 'resources': self.owned}) @@ -230,12 +353,19 @@ class Suite: args = [kind, 'apply' if kind == 'pipeline' else 'create'] if kind == 'model': args += ['--type', item['type']] - created = self.call(args, body) + created = self.call(args, body, defer_bot_readback=True) if kind == 'bot' else self.call(args, body) created = created.get('pipeline', created) if kind == 'pipeline' else created uuid = created['uuid'] self.variables[item['id']] = uuid self.owned.append({'kind': kind, 'uuid': uuid, **({'type': item['type']} if kind == 'model' else {})}) self.journal() # Record ownership before any later configuration or ingestion can fail. + if kind == 'bot': + # lbctl 0.3's Bot projection omits event bindings; verify the existing Core API instead. + actual = self.read_bot(uuid) + expected = copy.deepcopy(body) + expected['adapter_config'].pop('inbound_secret') + if not contains_fields(actual, expected): + raise RuntimeError(f'Bot readback differs from submitted non-secret fields: {uuid}') if config is not None: current = self.call(['pipeline', 'get', uuid]) current = current.get('pipeline', current) @@ -263,8 +393,20 @@ class Suite: ) self.write( 'resources.json', - {key: value for key, value in self.variables.items() if key not in self.profile.get('vars', {})}, + { + key: value + for key, value in self.variables.items() + if key not in self.profile.get('vars', {}) and key != 'group_secret' + }, ) + warmups = [] + for case in self.profile['cases']: + if 'http_group' in case: + case = resolve(case, self.variables) + group = f'qa-warmup-{uuid4().hex[:12]}' + response = group_request(case, group, 'qa-member-000', 1, uuid4().hex, 'none', self.env) + warmups.append({'case': case['name'], 'output': response}) + self.write('warmup.json', warmups) def cleanup(self): saved = json.loads((self.output / 'owned.json').read_text()) @@ -291,7 +433,10 @@ class Suite: case = resolve(original, self.variables) if 'body' in case: case['args'] += ['--file', self.write(f'case-{len(cases)}.json', case.pop('body'))] - cases.extend([{key: case[key] for key in ('name', 'args', 'expect')}] * case.get('weight', 1)) + cases.extend( + [{key: case[key] for key in ('name', 'args', 'expect', 'http_group') if key in case}] + * case.get('weight', 1) + ) cases_path = self.write('cases.json', cases) observed = resolve(self.profile.get('observe', {}), self.variables) observe_args = [] @@ -303,6 +448,8 @@ class Suite: observe_args += ['--health', f'{name}={url}'] for container in observed.get('containers', []): observe_args += ['--container', container] + if observed.get('box_data_root'): + observe_args += ['--box-data-root', self.path(observed['box_data_root'])] results = [] errors = [] for phase in self.profile['phases']: @@ -327,6 +474,12 @@ class Suite: '--cooldown', str(phase.get('cooldown_seconds', 0)), '--shared-case-sequence', + '--groups', + str(phase.get('groups', 0)), + '--members-per-group', + str(phase.get('members_per_group', 3)), + '--group-state', + str(self.output / 'group-state.json'), *observe_args, ] with (self.output / f'{phase["name"]}.log').open('w') as log: @@ -355,11 +508,19 @@ def main(): parser.add_argument('--duration', type=float, help='Override duration_seconds in every phase') parser.add_argument('--rps', type=float, help='Override RPS in every phase') parser.add_argument('--workers', type=int, help='Override workers in every phase') + parser.add_argument('--groups', type=int, help='Override group count in every group phase') + parser.add_argument('--members-per-group', type=int) args = parser.parse_args() try: profile = json.loads(args.profile.read_text()) for phase in profile.get('phases', []): - for flag, field in [('duration', 'duration_seconds'), ('rps', 'rps'), ('workers', 'workers')]: + for flag, field in [ + ('duration', 'duration_seconds'), + ('rps', 'rps'), + ('workers', 'workers'), + ('groups', 'groups'), + ('members_per_group', 'members_per_group'), + ]: if getattr(args, flag) is not None: phase[field] = getattr(args, flag) validate(profile) diff --git a/tools/regression_traffic.py b/tools/regression_traffic.py index 67a458db4..c147190cf 100644 --- a/tools/regression_traffic.py +++ b/tools/regression_traffic.py @@ -1,10 +1,12 @@ #!/usr/bin/env python3 -"""Drive normal lbctl traffic and retain business checks, health and Linux resources.""" +"""Drive independent CLI requests or persistent group conversations with Linux resource sampling.""" import argparse from collections import Counter from concurrent.futures import ThreadPoolExecutor import json +import hashlib +import hmac import math import os from pathlib import Path @@ -12,6 +14,9 @@ import subprocess import sys import threading import time +import urllib.error +import urllib.request +import uuid sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'scripts')) from cloud_runtime_soak import read_endpoint_snapshot, read_process_snapshot @@ -21,19 +26,97 @@ def percentile(values, fraction): return sorted(values)[max(0, math.ceil(len(values) * fraction) - 1)] if values else None +def owned_box_containers(data_root, checked=None): + """Discover sandboxes mounted below one run's data root, including newly created groups.""" + if checked is None: + + def checked(command): + return subprocess.check_output(command, text=True, timeout=10) + + candidates = checked(['docker', 'ps', '-a', '--filter', 'label=langbot.box=true', '--format', '{{.ID}}']).split() + if not candidates: + return [] + root = Path(data_root).resolve() + descriptions = checked(['docker', 'inspect', '--format', '{{json .}}', *candidates]) + owned = [] + for line in descriptions.splitlines(): + if not line: + continue + entry = json.loads(line) + if any( + Path(mount['Source']).resolve().is_relative_to(root) for mount in entry['Mounts'] if mount.get('Source') + ): + owned.append(entry['Id']) + return owned + + +def group_request(case, group_id, member_id, turn, token, previous, env=None): + """Send one signed GroupMessage through the existing HTTP Bot adapter.""" + settings = case['http_group'] + secret = (os.environ if env is None else env)[settings['secret_env']] + marker = f'LB_GROUP={group_id} LB_TURN={turn} LB_TOKEN={token} LB_PREVIOUS={previous}' + body = json.dumps( + { + 'session_id': group_id, + 'session_type': 'group', + 'sender': {'id': member_id, 'name': member_id, 'group_name': group_id}, + 'message': [{'type': 'Plain', 'text': settings['message'] + ' ' + marker}], + } + ).encode() + stamp = str(int(time.time())) + signature = hmac.new(secret.encode(), stamp.encode() + b'.' + body, hashlib.sha256).hexdigest() + request = urllib.request.Request( + settings['url'], + data=body, + headers={ + 'Content-Type': 'application/json', + 'X-LB-Timestamp': stamp, + 'X-LB-Signature': 'sha256=' + signature, + 'X-LB-Idempotency-Key': token, + }, + ) + opener = urllib.request.build_opener(urllib.request.ProxyHandler({})) + with opener.open(request, timeout=90) as response: + payload = json.load(response) + if response.status != 200 or not isinstance(payload, dict) or payload.get('code') != 0: + raise ValueError('group ingress did not return a completed response') + data = payload.get('data', {}) + if not isinstance(data, dict): + raise ValueError('group ingress response data must be an object') + parts = data.get('message') + if not isinstance(parts, list) or any(not isinstance(part, dict) for part in parts): + raise ValueError('group reply must contain message components') + text = ''.join(str(part.get('text', '')) for part in parts if part.get('type') == 'Plain') + if data.get('session_id') != group_id or not data.get('reply_to'): + raise ValueError('group reply session or message correlation is missing') + if case['expect'] not in text or f'GROUP_ACK={token}' not in text: + raise ValueError('group reply failed its business or conversation-history oracle') + return payload + + def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--lbctl', required=True) parser.add_argument('--config', required=True) - parser.add_argument('--cases', type=Path, required=True, help='JSON array of {name, args, expect}') + parser.add_argument( + '--cases', type=Path, required=True, help='JSON cases with name, expect, and args or http_group' + ) parser.add_argument('--output', type=Path, required=True) parser.add_argument('--duration', type=float, default=600) parser.add_argument('--rps', type=float, default=2) parser.add_argument('--workers', type=int, default=4) + parser.add_argument( + '--groups', type=int, default=0, help='Stable group sessions; zero uses independent CLI requests' + ) + parser.add_argument('--members-per-group', type=int, default=3) + parser.add_argument( + '--group-state', type=Path, help='Carry group histories across load phases; refuses interrupted runs' + ) parser.add_argument('--cooldown', type=float, default=30) parser.add_argument('--pid', action='append', default=[], metavar='NAME=PID') parser.add_argument('--health', action='append', default=[], metavar='NAME=URL') parser.add_argument('--container', action='append', default=[]) + parser.add_argument('--box-data-root', type=Path, help='Dynamically sample Box containers mounted under this root') parser.add_argument( '--shared-case-sequence', action='store_true', help='Share the weighted case sequence across workers' ) @@ -46,6 +129,8 @@ def main(): or args.cooldown < 0 ): parser.error('duration/rps must be positive, workers 1..32, cooldown nonnegative') + if not 0 <= args.groups <= 4096 or not 1 <= args.members_per_group <= 1000: + parser.error('groups must be 0..4096 (0 selects CLI mode), members-per-group 1..1000') cases = json.loads(args.cases.read_text()) if ( not isinstance(cases, list) @@ -53,26 +138,77 @@ def main(): or any( not isinstance(case, dict) or not isinstance(case.get('name'), str) - or not isinstance(case.get('args'), list) - or not all(isinstance(item, str) for item in case['args']) + or ( + not args.groups + and (not isinstance(case.get('args'), list) or not all(isinstance(item, str) for item in case['args'])) + ) or not isinstance(case.get('expect'), str) or not case['expect'] for case in cases ) ): parser.error('cases must contain name, string args, and expected response text') + if args.groups: + if args.groups < len(cases): + parser.error('groups must cover the complete weighted case sequence') + for case in cases: + settings = case.get('http_group', {}) + if ( + not isinstance(settings, dict) + or set(settings) != {'url', 'secret_env', 'message'} + or not all(isinstance(value, str) and value for value in settings.values()) + or not settings['url'].startswith(('http://', 'https://')) + or not os.environ.get(settings['secret_env']) + ): + parser.error('group cases need HTTP(S) url, message and a populated secret_env') args.output.mkdir(parents=True, exist_ok=False) pids = {name: int(pid) for name, pid in (value.split('=', 1) for value in args.pid)} endpoints = dict(value.split('=', 1) for value in args.health) rows = [] sampling_errors = [] lock = threading.Lock() + group_condition = threading.Condition(lock) stop = threading.Event() phase = ['load'] next_case = 0 started = time.monotonic() deadline = started + args.duration redactions = [value for name, value in os.environ.items() if 'API_KEY' in name and value] + group_state = None + groups = [] + if args.groups: + redactions += [os.environ[case['http_group']['secret_env']] for case in cases] + state_path = args.group_state or args.output / 'group-state.json' + fingerprint = [(case['name'], case['http_group']) for case in cases] + group_state = ( + json.loads(state_path.read_text()) + if state_path.exists() + else { + 'run_id': uuid.uuid4().hex[:12], + 'cases': fingerprint, + 'groups': [], + 'in_progress': False, + } + ) + if group_state.get('in_progress') or json.dumps(group_state['cases']) != json.dumps(fingerprint): + parser.error('group state is interrupted or belongs to different cases; inspect it before reuse') + for index in range(len(group_state['groups']), args.groups): + group_state['groups'].append( + { + 'id': f'qa-{group_state["run_id"]}-g{index:04d}', + 'case': index % len(cases), + 'turn': 1, + 'previous': 'none', + 'failed': False, + } + ) + groups = [ + dict(item, due=started + index / args.rps, busy=False) + for index, item in enumerate(group_state['groups'][: args.groups]) + ] + group_state['in_progress'] = True + state_path.write_text(json.dumps(group_state, indent=2)) + state_path.chmod(0o600) def redact(text): for secret in redactions: @@ -99,9 +235,12 @@ def main(): row['health'][name] = read_endpoint_snapshot(url, timeout_seconds=3) except RuntimeError as exc: row['health'][name] = {'error': str(exc)} - if args.container: + containers = set(args.container) + if args.box_data_root: + containers.update(owned_box_containers(args.box_data_root)) + if containers: result = subprocess.run( - ['docker', 'stats', '--no-stream', '--format', '{{json .}}', *args.container], + ['docker', 'stats', '--no-stream', '--format', '{{json .}}', *sorted(containers)], capture_output=True, text=True, timeout=10, @@ -163,6 +302,60 @@ def main(): iteration += 1 due = max(due + args.workers / args.rps, time.monotonic()) + def group_worker(_index): + while not stop.is_set(): + with group_condition: + while True: + now = time.monotonic() + if stop.is_set() or now >= deadline: + return + available = [group for group in groups if not group['busy'] and not group['failed']] + if not available: + if not any(group['busy'] for group in groups): + return + group_condition.wait(0.1) + continue + group = min(available, key=lambda item: item['due']) + if group['due'] > now: + group_condition.wait(min(1, group['due'] - now)) + continue + group['busy'] = True + break + case = cases[group['case']] + before = time.monotonic() + token = uuid.uuid4().hex + member = f'qa-member-{(group["turn"] - 1) % args.members_per_group:03d}' + row = { + 'case': case['name'], + 'group_id': group['id'], + 'member_id': member, + 'turn': group['turn'], + 'time': time.time(), + 'elapsed': before - started, + 'status': 'failed', + } + try: + row['output'] = group_request(case, group['id'], member, group['turn'], token, group['previous']) + row['status'] = 'ok' + except urllib.error.HTTPError as exc: + row.update(http_status=exc.code, error=redact(exc.read(2000).decode(errors='replace'))) + if exc.code >= 500: + row['status'] = 'unknown' + except (urllib.error.URLError, TimeoutError, OSError) as exc: + row.update(status='unknown', error=redact(str(exc)) + '; not retried') + except (ValueError, KeyError) as exc: + row['error'] = redact(str(exc)) + row['seconds'] = time.monotonic() - before + with group_condition: + rows.append(row) + with (args.output / 'requests.jsonl').open('a') as stream: + stream.write(json.dumps(row, ensure_ascii=False) + '\n') + group.update(busy=False, failed=row['status'] != 'ok') + if row['status'] == 'ok': + group.update(turn=group['turn'] + 1, previous=token) + group['due'] = max(group['due'] + args.groups / args.rps, time.monotonic()) + group_condition.notify_all() + def sampling_guard(): try: sample() @@ -174,7 +367,7 @@ def main(): sampler.start() try: with ThreadPoolExecutor(max_workers=args.workers) as pool: - list(pool.map(worker, range(args.workers))) + list(pool.map(group_worker if args.groups else worker, range(args.workers))) stop.wait(max(0, deadline - time.monotonic())) load_elapsed = time.monotonic() - started phase[0] = 'cooldown' @@ -200,6 +393,36 @@ def main(): for sample in samples ) summary['sampling_errors'] = sampling_errors + if args.groups: + for index, group in enumerate(groups): + group_state['groups'][index] = {key: value for key, value in group.items() if key not in {'due', 'busy'}} + group_state['in_progress'] = False + temporary = state_path.with_suffix('.tmp') + temporary.write_text(json.dumps(group_state, indent=2)) + temporary.chmod(0o600) + temporary.replace(state_path) + summary['groups'] = { + 'configured': args.groups, + 'members_per_group': args.members_per_group, + 'target_rps_per_group': args.rps / args.groups, + 'sessions': { + group['id']: { + 'case': cases[group['case']]['name'], + 'successful_turns_total': group['turn'] - 1, + 'failed': group['failed'], + 'requests_this_phase': sum(row.get('group_id') == group['id'] for row in rows), + 'members_this_phase': sorted( + {row['member_id'] for row in rows if row.get('group_id') == group['id']} + ), + } + for group in groups + }, + } + summary['group_failures'] = [ + group_id + for group_id, result in summary['groups']['sessions'].items() + if result['failed'] or result['successful_turns_total'] < 2 or not result['requests_this_phase'] + ] for name in dict.fromkeys(case['name'] for case in cases): selected = [row for row in rows if row['case'] == name] times = [row['seconds'] for row in selected if row['status'] == 'ok'] @@ -212,7 +435,13 @@ def main(): } (args.output / 'summary.json').write_text(json.dumps(summary, indent=2)) print(json.dumps(summary, indent=2), flush=True) - if not rows or any(row['status'] != 'ok' for row in rows) or summary['health_failures'] or sampling_errors: + if ( + not rows + or any(row['status'] != 'ok' for row in rows) + or summary['health_failures'] + or sampling_errors + or summary.get('group_failures') + ): raise SystemExit(1) diff --git a/tools/test_mock_llm.py b/tools/test_mock_llm.py index 7deb07bf5..8c8d74577 100644 --- a/tools/test_mock_llm.py +++ b/tools/test_mock_llm.py @@ -66,6 +66,19 @@ class MockLLMTest(unittest.TestCase): self.assertEqual(content, 'LANGBOT_CHAT_OK') self.assertEqual(chunks[-1]['usage']['total_tokens'], 15) + def test_group_history_requires_previous_user_reply_and_isolation(self): + first = {'role': 'user', 'content': 'LB_GROUP=group-a LB_TURN=1 LB_TOKEN=aaa LB_PREVIOUS=none'} + second = {'role': 'user', 'content': 'LB_GROUP=group-a LB_TURN=2 LB_TOKEN=bbb LB_PREVIOUS=aaa'} + reply = json.loads(self.post(messages=[first]))['choices'][0]['message'] + self.assertIn('GROUP_ACK=aaa', reply['content']) + result = json.loads(self.post(messages=[first, reply, second]))['choices'][0]['message'] + self.assertIn('GROUP_ACK=bbb', result['content']) + other = {'role': 'user', 'content': 'LB_GROUP=group-b LB_TURN=1 LB_TOKEN=ccc LB_PREVIOUS=none'} + for messages in ([second], [first, second], [other, reply, second]): + with self.subTest(messages=messages), self.assertRaises(HTTPError) as failure: + self.post(messages=messages) + self.assertEqual(failure.exception.code, 400) + def test_embedding_http_float_base64_and_validation(self): def embed(**patch): body = {'model': 'mock-embedding', 'input': ['repeatable fixture', 'another fixture'], **patch} diff --git a/tools/test_regression_groups.py b/tools/test_regression_groups.py new file mode 100644 index 000000000..3d7bd5d0c --- /dev/null +++ b/tools/test_regression_groups.py @@ -0,0 +1,240 @@ +"""Exercise signed group traffic, member rotation, phase continuity and fail-closed history.""" + +import hashlib +import hmac +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +import json +import os +from pathlib import Path +import re +import subprocess +import sys +import tempfile +import threading +import time +import unittest + +from regression_suite import Suite, validate + + +class GroupTests(unittest.TestCase): + def test_bot_readback_uses_the_cli_confirmed_endpoint_and_workspace(self): + received = [] + + class Readback(BaseHTTPRequestHandler): + def do_GET(self): + received.append((self.path, self.headers.get('Authorization'), self.headers.get('X-Workspace-Id'))) + body = json.dumps({'code': 0, 'data': {'bot': {'uuid': 'owned-bot', 'event_bindings': []}}}).encode() + self.send_response(200) + self.end_headers() + self.wfile.write(body) + + def log_message(self, *_args): + pass + + server = ThreadingHTTPServer(('127.0.0.1', 0), Readback) + thread = threading.Thread(target=server.serve_forever) + thread.start() + try: + with tempfile.TemporaryDirectory() as folder: + current = json.loads((Path(__file__).with_name('regression_profiles') / 'groups.json').read_text()) + suite = Suite(current, Path(folder) / 'profile.json', Path(folder), env={'TEST_API_KEY': 'test-key'}) + suite.variables['workspace_uuid'] = 'bound-workspace' + suite.connection = { + 'endpoint': f'http://127.0.0.1:{server.server_port}', + 'credential_source': 'env:TEST_API_KEY', + } + self.assertEqual(suite.read_bot('owned-bot')['uuid'], 'owned-bot') + self.assertEqual(received, [('/api/v1/platform/bots/owned-bot', 'Bearer test-key', 'bound-workspace')]) + finally: + server.shutdown() + server.server_close() + thread.join() + + def test_groups_profile_and_invalid_modes(self): + path = Path(__file__).with_name('regression_profiles') / 'groups.json' + profile = json.loads(path.read_text()) + validate(profile) + for field, value in [('groups', 3), ('members_per_group', 0), ('groups', 4097)]: + current = json.loads(path.read_text()) + current['phases'][0][field] = value + with self.subTest(field=field), self.assertRaises(ValueError): + validate(current) + + def test_signed_groups_continue_across_phases_and_stop_after_bad_reply(self): + secret = 'test-group-signing-key' + histories = {} + failures = [] + active = set() + lock = threading.Lock() + bad_reply = False + + class Ingress(BaseHTTPRequestHandler): + def do_GET(self): + self.respond({'code': 0}) + + def respond(self, payload, status=200): + body = json.dumps(payload).encode() + self.send_response(status) + self.send_header('Content-Length', str(len(body))) + self.end_headers() + self.wfile.write(body) + + def do_POST(self): + raw = self.rfile.read(int(self.headers['Content-Length'])) + stamp = self.headers['X-LB-Timestamp'] + expected = ( + 'sha256=' + hmac.new(secret.encode(), stamp.encode() + b'.' + raw, hashlib.sha256).hexdigest() + ) + payload = json.loads(raw) + group = payload['session_id'] + fields = dict(re.findall(r'LB_(GROUP|TURN|TOKEN|PREVIOUS)=(\S+)', payload['message'][0]['text'])) + with lock: + prior = histories.get(group, []) + if ( + self.headers['X-LB-Signature'] != expected + or group in active + or payload['session_type'] != 'group' + or fields['GROUP'] != group + or int(fields['TURN']) != len(prior) + 1 + or fields['PREVIOUS'] != (prior[-1]['token'] if prior else 'none') + ): + failures.append(payload) + active.add(group) + time.sleep(0.01) + with lock: + histories.setdefault(group, []).append( + {'token': fields['TOKEN'], 'member': payload['sender']['id']} + ) + active.remove(group) + ack = 'missing' if bad_reply and group.endswith('g0000') else fields['TOKEN'] + self.respond( + { + 'code': 0, + 'data': { + 'session_id': group, + 'reply_to': 'incoming-' + fields['TOKEN'], + 'message': [ + {'type': 'Plain', 'text': 'TEST_OK GROUP_'}, + {'type': 'Plain', 'text': 'ACK=' + ack}, + ], + }, + }, + status=500 if bad_reply == 'unknown' and group.endswith('g0000') else 200, + ) + + def log_message(self, *_args): + pass + + server = ThreadingHTTPServer(('127.0.0.1', 0), Ingress) + thread = threading.Thread(target=server.serve_forever) + thread.start() + try: + with tempfile.TemporaryDirectory() as folder: + root = Path(folder) + cases = root / 'cases.json' + cases.write_text( + json.dumps( + [ + { + 'name': 'group', + 'expect': 'TEST_OK', + 'http_group': { + 'url': f'http://127.0.0.1:{server.server_port}/bots/test/sync', + 'secret_env': 'LANGBOT_QA_GROUP_SECRET', + 'message': 'ai Test group history', + }, + } + ] + ) + ) + state = root / 'group-state.json' + command = [ + sys.executable, + str(Path(__file__).with_name('regression_traffic.py')), + '--lbctl', + 'unused', + '--config', + 'unused', + '--cases', + str(cases), + '--duration', + '0.8', + '--rps', + '40', + '--workers', + '3', + '--cooldown', + '0', + '--members-per-group', + '2', + '--health', + f'core=http://127.0.0.1:{server.server_port}/healthz', + ] + env = {**os.environ, 'LANGBOT_QA_GROUP_SECRET': secret} + summaries = [] + for groups in (4, 8): + output = root / f'phase-{groups}' + result = subprocess.run( + [*command, '--groups', str(groups), '--group-state', str(state), '--output', str(output)], + env=env, + capture_output=True, + text=True, + timeout=15, + ) + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + summaries.append(json.loads((output / 'summary.json').read_text())) + first = summaries[0]['groups']['sessions'] + second = summaries[1]['groups']['sessions'] + self.assertEqual(len(first), 4) + self.assertEqual(len(second), 8) + for group in first: + self.assertGreater(second[group]['successful_turns_total'], first[group]['successful_turns_total']) + self.assertEqual(len(second[group]['members_this_phase']), 2) + self.assertFalse(failures) + self.assertNotIn(secret, state.read_text()) + bad_reply = True + output = root / 'failed' + result = subprocess.run( + [*command, '--groups', '4', '--output', str(output)], + env=env, + capture_output=True, + text=True, + timeout=15, + ) + self.assertEqual(result.returncode, 1, result.stdout + result.stderr) + summary = json.loads((output / 'summary.json').read_text()) + failed = next(item for group, item in summary['groups']['sessions'].items() if group.endswith('g0000')) + self.assertTrue(failed['failed']) + self.assertEqual(failed['requests_this_phase'], 1) + bad_reply = 'unknown' + result = subprocess.run( + [*command, '--groups', '4', '--output', str(root / 'unknown')], + env=env, + capture_output=True, + text=True, + timeout=15, + ) + self.assertEqual(result.returncode, 1, result.stdout + result.stderr) + unknown = json.loads((root / 'unknown/summary.json').read_text()) + self.assertEqual(unknown['status'].get('unknown'), 1) + interrupted = json.loads(state.read_text()) + interrupted['in_progress'] = True + state.write_text(json.dumps(interrupted)) + result = subprocess.run( + [*command, '--groups', '4', '--group-state', str(state), '--output', str(root / 'interrupted')], + env=env, + capture_output=True, + text=True, + timeout=15, + ) + self.assertNotEqual(result.returncode, 0) + self.assertIn('interrupted', result.stderr) + finally: + server.shutdown() + server.server_close() + thread.join() + + +if __name__ == '__main__': + unittest.main() diff --git a/tools/test_regression_job.py b/tools/test_regression_job.py index 53008e325..774eb5964 100644 --- a/tools/test_regression_job.py +++ b/tools/test_regression_job.py @@ -164,15 +164,31 @@ class JobTests(unittest.TestCase): root = Path(folder) job = Job({}, profile(), root / 'profile.json', root) job.state['record'] = {'data_dir': str(job.deploy / 'data')} + containers = ['ours', 'other'] def checked(args, timeout=900): if args[1] == 'ps': - return 'ours\nother\n' - source = str(job.deploy / 'data/box/tenant') if args[-1] == 'ours' else '/unrelated/box/tenant' - return json.dumps([{'Source': source}]) + return '\n'.join(containers) + return '\n'.join( + json.dumps( + { + 'Id': container, + 'Mounts': [ + { + 'Source': str(job.deploy / 'data/box/tenant') + if container != 'other' + else '/unrelated/box/tenant' + } + ], + } + ) + for container in args[4:] + ) job.checked = checked self.assertEqual(job.owned_sandboxes(), ['ours']) + containers.append('later') + self.assertEqual(job.owned_sandboxes(), ['ours', 'later']) if __name__ == '__main__': diff --git a/tools/test_regression_suite.py b/tools/test_regression_suite.py index bd9cbfd15..ae295588a 100644 --- a/tools/test_regression_suite.py +++ b/tools/test_regression_suite.py @@ -60,6 +60,87 @@ def profile(): class ProfileTests(unittest.TestCase): + def test_created_bot_verification_failure_can_only_defer_bot_readback(self): + with tempfile.TemporaryDirectory() as folder: + root = Path(folder) + suite = Suite(profile(), root / 'profile.json', root) + response = { + 'data': {'operation': 'bot.create', 'uuid': 'owned-bot', 'bot': {'uuid': 'owned-bot'}}, + 'error': {'type': 'verification_failed'}, + } + result = subprocess.CompletedProcess([], 1, json.dumps(response), '') + with patch('regression_suite.subprocess.run', return_value=result): + with self.assertRaises(RuntimeError): + suite.call(['bot', 'create'], {'name': 'QA'}) + self.assertEqual( + suite.call(['bot', 'create'], {'name': 'QA'}, defer_bot_readback=True)['uuid'], 'owned-bot' + ) + with self.assertRaises(RuntimeError): + suite.call(['pipeline', 'apply'], {}, defer_bot_readback=True) + + def test_group_bot_is_owned_before_raw_readback_and_warmed_before_load(self): + with tempfile.TemporaryDirectory() as folder: + root = Path(folder) + current = profile() + current['resources'] = [ + { + 'id': 'chat_bot', + 'kind': 'bot', + 'body': { + 'name': 'QA', + 'adapter': 'http_bot', + 'enable': True, + 'adapter_config': {'inbound_secret': '${group_secret}'}, + 'event_bindings': [{'target_type': 'pipeline', 'target_uuid': 'pipeline-a'}], + }, + } + ] + current['cases'] = [ + { + 'name': 'chat', + 'expect': 'OK', + 'http_group': { + 'url': 'http://localhost/bots/${chat_bot}/sync', + 'message': 'ai Test', + 'secret_env': 'LANGBOT_QA_GROUP_SECRET', + }, + } + ] + current['phases'][0]['groups'] = 1 + validate(current) + for correct_route in (False, True): + suite = Suite(current, root / 'profile.json', root) + + def call(args, body=None, **kwargs): + if args == ['whoami']: + return {'workspace_uuid': 'workspace-a'} + if args == ['bot', 'create']: + self.assertTrue(kwargs['defer_bot_readback']) + return {'uuid': 'owned-bot'} + raise AssertionError(args) + + suite.call = call + actual = copy.deepcopy(current['resources'][0]['body']) + actual['adapter_config']['inbound_secret'] = '***' + if not correct_route: + actual['event_bindings'][0]['target_uuid'] = 'wrong-pipeline' + with ( + patch.object(suite, 'read_bot', return_value=actual), + patch('regression_suite.group_request', return_value={'code': 0}) as warmup, + ): + if correct_route: + suite.prepare() + warmup.assert_called_once() + self.assertNotIn(suite.variables['group_secret'], (root / 'resources.json').read_text()) + else: + with self.assertRaisesRegex(RuntimeError, 'Bot readback differs'): + suite.prepare() + warmup.assert_not_called() + self.assertEqual( + json.loads((root / 'owned.json').read_text())['resources'], + [{'kind': 'bot', 'uuid': 'owned-bot'}], + ) + def test_normal_profile_and_overrides_contract(self): validate(json.loads(Path(__file__).with_name('regression_profiles').joinpath('normal.json').read_text())) bad = profile()