diff --git a/tools/README.md b/tools/README.md new file mode 100644 index 000000000..dc365fd8b --- /dev/null +++ b/tools/README.md @@ -0,0 +1,234 @@ +# Mock LLM for regression tests + +Run the deterministic OpenAI Chat Completions endpoint with Python 3.11+, without +API credentials or third-party dependencies: + +```sh +python3 tools/mock_llm.py --port 18080 --record /tmp/langbot-mock-requests.jsonl +``` + +In LangBot, create an `openai-chat-completions` Provider whose base URL is +`http://127.0.0.1:18080/v1` and whose API key is any test-only value. Use a model +from `GET /v1/models`. The loopback address works when Core and this tool run on +the same host; containers must use an address reachable from their own network. + +| Model | Behavior | +| --- | --- | +| `mock-chat` | Returns `LANGBOT_CHAT_OK`, including SSE streaming | +| `mock-rag` | Requires the golden document sentinel in the model input, then returns `LANGBOT_RAG_OK` | +| `mock-embedding` | Returns deterministic normalized token-hash vectors from `/v1/embeddings` | +| `mock-box` | Calls `exec` twice to write/read `/workspace/mock-llm.txt`, then returns `LANGBOT_BOX_OK` | +| `mock-401`, `mock-429`, `mock-503` | Returns the selected HTTP failure | +| `mock-slow` | Delays the response by two seconds | +| `mock-slow-stream` | Delays each SSE chunk by 0.2 seconds | + +`mock-rag` requires `aurora-parser-rag-9137` in the submitted messages. The +[golden HTML fixture](../skills/skills/langbot-testing/fixtures/rag/parser-golden.html) +contains this marker; retrieve it into the model context without including it +in the user's question. + +For `mock-box`, enable Box in the Runner and expose the native `exec` tool. Each +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. + +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 +in the previous tool results; `expect_prompt` checks the submitted messages. +The server returns tool calls; it never executes +their commands. Request recordings contain prompts, tool results, responses, +status, and disconnects, but no HTTP headers. Use test data only. + +Run the focused HTTP checks on the Linux test host: + +```sh +python3 -m unittest discover -s tools -p test_mock_llm.py -v +``` + +These checks validate the mock server. A LangBot regression must additionally +verify real tool outputs and retrieved document IDs/content. Token usage is +synthetic, and this endpoint does not evaluate answer quality or embedding models. + +`mock-embedding` accepts string/batch inputs, float/base64 encoding, and optional +dimensions (default 384). It tests the embedding protocol and RAG plumbing; +token hashes do not establish semantic retrieval quality. + +## Normal traffic and resource collection + +`regression_traffic.py` executes a JSON list of normal `lbctl` cases at a selected +rate. Each case has a name, CLI arguments, and an expected response substring: + +```json +[{"name":"chat","args":["pipeline","run","PIPELINE_UUID","--message","Hello"],"expect":"LANGBOT_CHAT_OK"}] +``` + +Prepare a Workspace-bound CLI context and its API key environment variable, then +run on the Linux test host: + +```sh +python3 tools/regression_traffic.py \ + --lbctl /path/to/lbctl --config /path/to/lbctl.yaml \ + --cases /path/to/cases.json --output /path/to/new-results-directory \ + --duration 600 --rps 2 --workers 4 --cooldown 30 \ + --pid core=CORE_PID --pid mock=MOCK_PID \ + --health core=http://127.0.0.1:5300/healthz \ + --health mock=http://127.0.0.1:18080/healthz +``` + +Add `--container CONTAINER_ID` for Docker CPU/memory/PID statistics. The executing +account must be able to inspect the selected process trees and Docker daemon. +Resource collection reuses `scripts/cloud_runtime_soak.py`; it does not require +cgroup v2. PID resource totals include descendants, so select disjoint trees. + +The result directory contains `requests.jsonl`, `resources.jsonl`, `progress.json`, +and `summary.json`. The requested rate is a target: four workers bound concurrency, +and slow requests lower the achieved rate. Requests with unknown outcomes are +recorded without retry. A command fails its business check if it returns an error +or omits the expected response. Health/resource collection failures also make the +run fail. API key values are redacted from recorded CLI output. + +Latency percentiles include the entire CLI invocation and use successful +requests only. Achieved request rate is `requests / load_elapsed_seconds`. +`health_samples` counts sampling rounds; `health_failures` counts rounds with +any failed health probe or process/container resource read. + +## Configured regression suites + +[`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. + +Use a dedicated QA Workspace. First provision the instance with `lbctl install` +or your source deployment, enable Docker Box, start the mock and embedding +backends, and prepare a Workspace-bound `lbctl` config. Set the profile's CLI +config path, backend URLs, and three plugin archive paths. The sample embedding +endpoint must serve `all-MiniLM-L6-v2`; the mock serves generation only. Relative +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 run \ + --profile tools/regression_profiles/normal.json --output .qa/regression/run-001 +``` + +`validate` checks profile syntax and references without contacting the instance. +The output directory for `run` must be new. CLI options `--duration`, `--rps`, and +`--workers` override those values in every phase. No additional Python dependency +is needed. + +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 + 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` + 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 + cooldown. Multiple phases provide a step load. In-flight requests drain at the + 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. +- `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. +- `cleanup`: `on-success` (default), `always`, or `never`. Only recorded resources + created by this run are deleted, in reverse order. Plugins and the existing + instance remain installed. Failed/unknown creations are not automatically + retried; inspect the instance if a creation outcome is unknown. + +Failed runs retain resources by default. After inspection, clean them with the +same profile and credentials: + +```sh +python3 tools/regression_suite.py cleanup \ + --profile tools/regression_profiles/normal.json --output .qa/regression/run-001 +``` + +Cleanup checks the Workspace against `owned.json` and records remaining items +after partial failures. Keep `owned.json` until cleanup completes. Each run also +contains `profile.json`, resolved `resources.json`, and one directory per phase +with the original request/resource records. The root `summary.json` includes +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: + +```sh +python3 -m unittest discover -s tools -p test_regression_suite.py -v +``` + +## Complete lifecycle Job + +[`regression_job.py`](regression_job.py) owns the deployment as well as the suite. +Use [`regression_profiles/job.json`](regression_profiles/job.json) to pin the Core +Release and select the suite and plugin input directory. The job runs this chain: + +1. `lbctl install --profile all` creates a fresh deployment inside the run directory. +2. A Compose override starts our mock backend on the deployment's private network + and injects a generated test-only API key. The Job binds its own CLI context. +3. The suite installs plugins, creates resources and ingests documents. Each case + runs once as warmup before measured traffic starts. +4. Configured load phases run with health, process-tree and container sampling. +5. Service-log tails and raw request/resource records are retained. Successful + Jobs stop the deployment, remove only sandboxes mounted under their own data + root, and use `lbctl uninstall --purge-data` to reclaim the deployment. + +Both generation and embedding use our mock backend, with no external model +service or real model credentials. Plugin archives, the pinned Core image, +`lbctl`, Docker and Compose v2 are Job inputs/prerequisites. This wrapper uses +PyYAML, already included in LangBot's dependencies, to verify the CLI's deployment +record and recover ownership after an interrupted install. + +Run on a Linux host with Docker access and permission to inspect the selected +container processes. Use an idle dedicated Docker daemon: current Box runtime +startup cleanup can remove sandboxes from another runtime. The Job refuses to +start when existing Box sandboxes are present. The current `lbctl` installer +also reserves host port 5401 and 2280–2285, independently of its HTTP port. + +Set the plugin directory and the `lbctl` executable in `job.json`, then run: + +```sh +uv run --no-sync python tools/regression_job.py validate --profile tools/regression_profiles/job.json +uv run --no-sync python tools/regression_job.py run \ + --profile tools/regression_profiles/job.json --output .qa/regression/job-001 +``` + +`--duration`, `--rps` and `--workers` override every load phase. Each output +directory must be new. Job cleanup policy is `on-success`, `always`, or `never`; +it governs the whole deployment, whereas suite cleanup governs API resources +when a suite runs on an existing deployment. `purge_data: false` preserves the +deployment data after removing containers. Failed Jobs retain their environment +by default; after inspection use: + +```sh +uv run --no-sync python tools/regression_job.py cleanup \ + --output .qa/regression/job-001 +``` + +The root `summary.json` is the final Job verdict; `suite/` contains the measured +results, `mock/requests.jsonl` contains backend recordings, and `services.log` +contains up to 10,000 log lines per service. Generated credentials and deployment +inputs stay in `.private/`; share the result files, not that directory. A Job has +its own CLI locator/config and records its deployment identity before cleanup. +Changing that identity makes cleanup fail closed. Cleanup uses the saved Job +state and does not require the original profile or plugin input files. + +The Job replaces the real embedding model with `mock-embedding`. This measures +LangBot/plugin/Box/RAG overhead, not real-model inference cost or retrieval +quality. Run the Job plumbing checks on Linux with: + +```sh +uv run --no-sync python -m unittest discover -s tools -p 'test_*.py' -v +``` diff --git a/tools/mock_llm.py b/tools/mock_llm.py new file mode 100644 index 000000000..a3e6c2da8 --- /dev/null +++ b/tools/mock_llm.py @@ -0,0 +1,366 @@ +#!/usr/bin/env python3 +"""A deterministic OpenAI Chat Completions server for local regression tests.""" + +import argparse +import base64 +import hashlib +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +import json +import math +from pathlib import Path +import re +import struct +import threading +import time +import uuid + + +MAX_BODY_BYTES = 1024 * 1024 +USAGE = {'prompt_tokens': 10, 'completion_tokens': 5, 'total_tokens': 15} + + +def mock_embeddings(body): + """Return deterministic normalized token hashes, not semantic model inference.""" + inputs = body.get('input') + inputs = [inputs] if isinstance(inputs, str) else inputs + if ( + not isinstance(inputs, list) + or not 1 <= len(inputs) <= 128 + or any(not isinstance(text, str) or not text for text in inputs) + ): + raise ValueError('embedding input must contain 1..128 non-empty strings') + dimensions = body.get('dimensions', 384) + if isinstance(dimensions, bool) or not isinstance(dimensions, int) or not 1 <= dimensions <= 4096: + raise ValueError('embedding dimensions must be an integer in 1..4096') + encoding = body.get('encoding_format', 'float') + if encoding not in ('float', 'base64'): + raise ValueError('embedding encoding_format must be float or base64') + data = [] + for index, text in enumerate(inputs): + vector = [0.0] * dimensions + for token in re.findall(r'\w+', text.casefold()) or [text]: + digest = hashlib.sha256(token.encode()).digest() + vector[int.from_bytes(digest[:4], 'little') % dimensions] += 1 if digest[4] % 2 else -1 + norm = math.sqrt(sum(value * value for value in vector)) + if not norm: + vector[0], norm = 1.0, 1.0 + vector = [value / norm for value in vector] + embedding = ( + base64.b64encode(struct.pack(f'<{dimensions}f', *vector)).decode() if encoding == 'base64' else vector + ) + data.append({'object': 'embedding', 'index': index, 'embedding': embedding}) + return {'object': 'list', 'model': body['model'], 'data': data, 'usage': {'prompt_tokens': 10, 'total_tokens': 10}} + + +def select_message(body, scenario): + messages = body.get('messages') + if not isinstance(messages, list) or not messages or any(not isinstance(m, dict) for m in messages): + raise ValueError('messages must be a non-empty array of objects') + if not isinstance(body.get('stream', False), bool): + raise ValueError('stream must be a boolean') + if body.get('stream_options') is not None and not isinstance(body['stream_options'], dict): + raise ValueError('stream_options must be an object') + # Select from the current turn's history, never a shared request counter. + start = next( + (i for i in range(len(messages) - 1, -1, -1) if messages[i].get('role') == 'user'), + None, + ) + if start is None: + raise ValueError('a user message is required') + 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): + end = calls[index + 1] if index + 1 < len(calls) else len(turn) + requested = {call['id'] for call in turn[position]['tool_calls']} + returned = [m.get('tool_call_id') for m in turn[position + 1 : end] if m.get('role') == 'tool'] + if len(returned) != len(requested) or set(returned) != requested: + raise ValueError('each tool call must have exactly one matching tool result') + steps = scenario['steps'] + if len(calls) >= len(steps): + raise ValueError('tool history exceeds the configured scenario') + step = steps[len(calls)] + if expected_prompt := step.get('expect_prompt'): + if expected_prompt not in json.dumps(messages, ensure_ascii=False): + raise ValueError(f'model input must contain {expected_prompt!r}') + expected = step.get('expect_tool_output') + if expected is not None: + results = turn[calls[-1] + 1 :] if calls else [] + 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')} + if tool_calls := step.get('tool_calls'): + tools = body.get('tools', []) + if not isinstance(tools, list) or any( + not isinstance(t, dict) or not isinstance(t.get('function'), dict) for t in tools + ): + raise ValueError('tools must be an array of objects') + available = {t.get('function', {}).get('name') for t in tools if t.get('type') == 'function'} + if body.get('tool_choice') == 'none' or any(call['name'] not in available for call in tool_calls): + raise ValueError('scenario tools must be advertised and enabled by the caller') + message['tool_calls'] = [ + { + 'id': f'call_mock_{len(calls)}_{i}', + 'type': 'function', + 'function': { + 'name': call['name'], + 'arguments': json.dumps(call['arguments']), + }, + } + for i, call in enumerate(tool_calls) + ] + return message + + +class MockServer(ThreadingHTTPServer): + daemon_threads = True + + def __init__(self, address, scenarios, record_path=None): + super().__init__(address, Handler) + self.scenarios = scenarios + self.record_path = record_path + self.record_lock = threading.Lock() + + def record(self, entry): + if self.record_path: + with ( + self.record_lock, + self.record_path.open('a', encoding='utf-8') as stream, + ): + stream.write(json.dumps(entry, ensure_ascii=False) + '\n') + + +class Handler(BaseHTTPRequestHandler): + def log_message(self, *_args): + pass + + def json_response(self, status, payload): + data = json.dumps(payload, ensure_ascii=False).encode() + self.send_response(status) + self.send_header('Content-Type', 'application/json') + self.send_header('Content-Length', str(len(data))) + self.end_headers() + self.wfile.write(data) + + def do_GET(self): + if self.path == '/healthz': + self.json_response(200, {'status': 'ok'}) + elif self.path == '/v1/models': + self.json_response( + 200, + { + 'object': 'list', + 'data': [ + { + 'id': name, + 'object': 'model', + 'created': 0, + 'owned_by': 'mock', + } + for name in [*self.server.scenarios, 'mock-embedding'] + ], + }, + ) + else: + self.json_response( + 404, + { + 'error': { + 'message': 'unknown endpoint', + 'type': 'invalid_request_error', + } + }, + ) + + def do_POST(self): + entry = {'request_id': uuid.uuid4().hex, 'time': time.time(), 'path': self.path, 'status': 400} + try: + if self.path not in ('/v1/chat/completions', '/v1/embeddings'): + entry['status'] = 404 + self.json_response( + 404, + { + 'error': { + 'message': 'unknown endpoint', + 'type': 'invalid_request_error', + } + }, + ) + return + self.connection.settimeout(10) + length = int(self.headers.get('Content-Length', 0)) + if self.headers.get('Transfer-Encoding') or not 0 < length <= MAX_BODY_BYTES: + raise ValueError('a Content-Length between 1 and 1048576 is required') + body = json.loads(self.rfile.read(length)) + if not isinstance(body, dict) or not isinstance(body.get('model'), str): + raise ValueError('request must be an object with a model name') + entry['request'] = body + if self.path == '/v1/embeddings': + if body['model'] != 'mock-embedding': + raise ValueError('embedding model must be mock-embedding') + payload = mock_embeddings(body) + entry.update(status=200, response=payload) + self.json_response(200, payload) + return + scenario = self.server.scenarios.get(body['model']) + if scenario is None: + raise ValueError('unknown model; use GET /v1/models') + message = select_message(body, scenario) + time.sleep(scenario.get('delay_seconds', 0)) + status = scenario.get('status', 200) + entry['status'] = status + if status != 200: + self.json_response( + status, + { + 'error': { + 'message': 'injected mock failure', + 'type': 'api_error', + 'code': str(status), + } + }, + ) + return + entry['response'] = message + base = { + 'id': 'chatcmpl-' + entry['request_id'], + 'created': int(time.time()), + 'model': body['model'], + } + finish = 'tool_calls' if message.get('tool_calls') else 'stop' + if body.get('stream'): + self.stream_response(body, scenario, message, base, finish) + else: + self.json_response( + 200, + { + **base, + 'object': 'chat.completion', + 'choices': [{'index': 0, 'message': message, 'finish_reason': finish}], + 'usage': USAGE, + }, + ) + except (ValueError, KeyError, TypeError, TimeoutError) as exc: + entry.update(status=400, error=str(exc)) + self.json_response(400, {'error': {'message': str(exc), 'type': 'invalid_request_error'}}) + except (BrokenPipeError, ConnectionResetError): + entry['disconnected'] = True + finally: + self.server.record(entry) + + def stream_response(self, body, scenario, message, base, finish): + self.send_response(200) + self.send_header('Content-Type', 'text/event-stream') + self.send_header('Cache-Control', 'no-cache') + self.end_headers() + + def send(delta, reason=None): + payload = { + **base, + 'object': 'chat.completion.chunk', + 'choices': [{'index': 0, 'delta': delta, 'finish_reason': reason}], + } + self.wfile.write(('data: ' + json.dumps(payload, ensure_ascii=False) + '\n\n').encode()) + self.wfile.flush() + time.sleep(scenario.get('chunk_delay_seconds', 0)) + + send({'role': 'assistant'}) + content = message.get('content') or '' + for offset in range(0, len(content), 8): + send({'content': content[offset : offset + 8]}) + for index, call in enumerate(message.get('tool_calls', [])): + send( + { + 'tool_calls': [ + { + 'index': index, + 'id': call['id'], + 'type': 'function', + 'function': { + 'name': call['function']['name'], + 'arguments': '', + }, + } + ] + } + ) + arguments = call['function']['arguments'] + for offset in range(0, len(arguments), 16): + send( + { + 'tool_calls': [ + { + 'index': index, + 'function': {'arguments': arguments[offset : offset + 16]}, + } + ] + } + ) + send({}, finish) + if (body.get('stream_options') or {}).get('include_usage'): + payload = { + **base, + 'object': 'chat.completion.chunk', + 'choices': [], + 'usage': USAGE, + } + self.wfile.write(('data: ' + json.dumps(payload) + '\n\n').encode()) + self.wfile.write(b'data: [DONE]\n\n') + self.wfile.flush() + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--host', default='127.0.0.1') + parser.add_argument('--port', type=int, default=18080) + parser.add_argument( + '--scenarios', + type=Path, + default=Path(__file__).with_name('mock_llm_scenarios.json'), + ) + parser.add_argument( + '--record', + type=Path, + help='Append request/response JSONL without HTTP headers; use test data only', + ) + args = parser.parse_args() + scenarios = json.loads(args.scenarios.read_text(encoding='utf-8')) + if not isinstance(scenarios, dict) or not scenarios: + parser.error('scenarios must be a non-empty object keyed by model name') + for name, scenario in scenarios.items(): + if not isinstance(scenario, dict) or not isinstance(scenario.get('steps'), list) or not scenario['steps']: + parser.error(f'{name}: steps must be a non-empty array') + if any(not isinstance(step, dict) for step in scenario['steps']): + parser.error(f'{name}: each step must be an object') + for step in scenario['steps']: + if not isinstance(step.get('content'), str) and not step.get('tool_calls'): + parser.error(f'{name}: each step needs content or tool_calls') + if 'expect_tool_output' in step and not isinstance(step['expect_tool_output'], str): + parser.error(f'{name}: expect_tool_output must be a string') + if 'expect_prompt' in step and not isinstance(step['expect_prompt'], str): + parser.error(f'{name}: expect_prompt must be a string') + if 'tool_calls' in step: + calls = step['tool_calls'] + if not isinstance(calls, list) or any( + not isinstance(call, dict) + or not isinstance(call.get('name'), str) + or not isinstance(call.get('arguments'), dict) + for call in calls + ): + parser.error(f'{name}: tool_calls need a name and arguments object') + if scenario.get('status', 200) not in (200, 401, 429, 500, 503): + parser.error(f'{name}: unsupported injected HTTP status') + for key in ('delay_seconds', 'chunk_delay_seconds'): + value = scenario.get(key, 0) + if not isinstance(value, (int, float)) or not 0 <= value <= 60: + parser.error(f'{name}: {key} must be between 0 and 60') + with MockServer((args.host, args.port), scenarios, args.record) as server: + print(f'Mock LLM: http://{args.host}:{server.server_port}/v1', flush=True) + try: + server.serve_forever() + except KeyboardInterrupt: + pass + + +if __name__ == '__main__': + main() diff --git a/tools/mock_llm_scenarios.json b/tools/mock_llm_scenarios.json new file mode 100644 index 000000000..e8455ba2d --- /dev/null +++ b/tools/mock_llm_scenarios.json @@ -0,0 +1,20 @@ +{ + "mock-chat": { + "steps": [{"content": "LANGBOT_CHAT_OK"}] + }, + "mock-rag": { + "steps": [{"expect_prompt": "aurora-parser-rag-9137", "content": "LANGBOT_RAG_OK"}] + }, + "mock-box": { + "steps": [ + {"tool_calls": [{"name": "exec", "arguments": {"command": "printf LANGBOT_BOX_SENTINEL > /workspace/mock-llm.txt; cat /workspace/mock-llm.txt"}}]}, + {"expect_tool_output": "LANGBOT_BOX_SENTINEL", "tool_calls": [{"name": "exec", "arguments": {"command": "cat /workspace/mock-llm.txt"}}]}, + {"expect_tool_output": "LANGBOT_BOX_SENTINEL", "content": "LANGBOT_BOX_OK"} + ] + }, + "mock-401": {"status": 401, "steps": [{"content": "unused"}]}, + "mock-429": {"status": 429, "steps": [{"content": "unused"}]}, + "mock-503": {"status": 503, "steps": [{"content": "unused"}]}, + "mock-slow": {"delay_seconds": 2, "steps": [{"content": "LANGBOT_SLOW_OK"}]}, + "mock-slow-stream": {"chunk_delay_seconds": 0.2, "steps": [{"content": "LANGBOT_SLOW_STREAM_OK"}]} +} diff --git a/tools/regression_job.py b/tools/regression_job.py new file mode 100644 index 000000000..b763c9563 --- /dev/null +++ b/tools/regression_job.py @@ -0,0 +1,464 @@ +#!/usr/bin/env python3 +"""Own one complete lbctl deployment, mock backend, regression, and teardown.""" + +import argparse +import copy +import json +import os +from pathlib import Path +import re +import secrets +import shutil +import subprocess +import time +import urllib.error +import urllib.request + +import yaml + +from regression_suite import Suite, number, resolve, validate + + +KEY_ENV = 'LANGBOT_QA_API_KEY' + + +def write_json(path, data): + temporary = path.with_suffix(path.suffix + '.tmp') + temporary.write_text(json.dumps(data, ensure_ascii=False, indent=2)) + temporary.chmod(0o600) + temporary.replace(path) + + +def load_profile(path): + profile = json.loads(path.read_text()) + if not isinstance(profile, dict) or profile.get('version') != 1: + raise ValueError('job profile version must be 1') + if profile.keys() - {'version', 'lbctl', 'suite', 'deployment', 'vars', 'cleanup', 'purge_data', 'mock_image'}: + raise ValueError('unknown job profile field') + deployment = profile.get('deployment', {}) + if not isinstance(deployment, dict) or deployment.keys() - {'version', 'port'}: + raise ValueError('deployment accepts version and port') + version = deployment.get('version', '') + if not isinstance(version, str) or not re.fullmatch(r'v[0-9]+\.[0-9]+\.[0-9]+', version): + raise ValueError('deployment.version must pin a stable vX.Y.Z Release') + number(deployment.get('port', 5300), 'deployment port', minimum=1, integer=True) + if deployment.get('port', 5300) > 65535: + raise ValueError('deployment port must be <= 65535') + if profile.get('cleanup', 'on-success') not in ('always', 'on-success', 'never'): + raise ValueError('cleanup must be always/on-success/never') + if not isinstance(profile.get('purge_data', True), bool): + raise ValueError('purge_data must be boolean') + suite_path = Path(profile['suite']).expanduser() + suite_path = suite_path if suite_path.is_absolute() else path.parent / suite_path + suite = json.loads(suite_path.read_text()) + suite.setdefault('vars', {}).update(profile.get('vars', {})) + suite['vars'].update( + mock_base_url='http://regression_mock:18080/v1', embedding_base_url='http://regression_mock:18080/v1' + ) + suite['cli'] = {'config': 'job-generated.yaml'} + for item in suite.get('resources', []): + if item['kind'] == 'model' and item.get('type') == 'embedding': + item['body']['name'] = 'mock-embedding' + validate(suite) + return profile, suite, suite_path.resolve() + + +class Job: + def __init__(self, profile, suite_profile, suite_path, output): + self.profile = profile + self.suite_profile = copy.deepcopy(suite_profile) + self.suite_path = suite_path + self.output = output.resolve() + self.private = self.output / '.private' + self.deploy = self.private / 'deployment' + self.state_path = self.output / 'state.json' + self.cli_config = self.private / 'lbctl.yaml' + self.env = {**os.environ, 'XDG_CONFIG_HOME': str(self.private / 'config')} + self.cli = [profile.get('lbctl', 'lbctl'), '--config', str(self.cli_config), '-o', 'json'] + self.redactions = [] + self.state = {'job_id': secrets.token_hex(8), 'record': None, 'bound': False, 'phase': 'created'} + + def redact(self, value): + for secret in self.redactions: + value = value.replace(secret, '[REDACTED]') + return value + + def save(self): + write_json(self.state_path, self.state) + + def recover_record(self): + if (self.private / 'job-id').read_text().strip() != self.state['job_id']: + raise RuntimeError('Job ownership marker differs from state.json') + path = self.deploy / 'deployment.yaml' + if not path.exists(): + if self.state.get('record'): + raise RuntimeError('deployment record disappeared; refusing cleanup') + return + if path.is_symlink(): + raise RuntimeError('deployment record must not be a symlink') + try: + record = yaml.safe_load(path.read_text()) + except yaml.YAMLError as exc: + raise ValueError('invalid deployment YAML') from exc + if not isinstance(record, dict) or not record.get('deployment_id'): + raise RuntimeError('invalid deployment record') + if ( + Path(record['compose_file']).resolve().parent != self.deploy + or Path(record['data_dir']).resolve() != self.deploy / 'data' + or Path(record['env_file']).resolve().parent != self.deploy + ): + raise RuntimeError('deployment record is outside this Job') + if self.state.get('record') and record['deployment_id'] != self.state['record']['deployment_id']: + raise RuntimeError('deployment identity changed; refusing cleanup') + self.state['record'] = record + self.save() + + def load_redactions(self): + if self.state.get('record'): + path = Path(self.state['record']['env_file']) + if path.exists(): + for line in path.read_text().splitlines(): + name, separator, value = line.partition('=') + if separator and any(word in name for word in ('TOKEN', 'KEY')) and len(value) >= 16: + self.redactions.append(value) + + def command(self, args, timeout=900): + result = subprocess.run(args, capture_output=True, text=True, env=self.env, timeout=timeout) + return result.returncode, self.redact(result.stdout), self.redact(result.stderr) + + def checked(self, args, timeout=900): + code, stdout, stderr = self.command(args, timeout) + if code: + raise RuntimeError((stderr + stdout)[-4000:]) + return stdout + + def cli_call(self, args): + code, stdout, stderr = self.command([*self.cli, *args]) + result = json.loads(stdout) if stdout.strip() else {} + data = result.get('data', {}) + if args[0] == 'install' and isinstance(data, dict) and data.get('record'): + record = data['record'] + if ( + Path(record['compose_file']).resolve().parent != self.deploy + or Path(record['data_dir']).resolve() != self.deploy / 'data' + ): + raise RuntimeError('install returned a deployment outside this Job') + self.state['record'] = record + self.save() # Installation may have changed state even when its final health check failed. + if code: + raise RuntimeError((stderr + stdout)[-4000:]) + return data + + def compose(self, *args, override=False): + record = self.state['record'] + command = ['docker', 'compose', '--project-name', record['compose_project'], '--file', record['compose_file']] + if override: + command += ['--file', str(self.private / 'compose.override.json')] + command += ['--env-file', record['env_file'], '--profile', 'all', *args] + return self.checked(command) + + def project_containers(self): + project = self.state['record']['compose_project'] + output = self.checked( + [ + 'docker', + 'ps', + '-a', + '--filter', + f'label=com.docker.compose.project={project}', + '--format', + '{{.ID}} {{.Label "com.docker.compose.service"}}', + ] + ) + return {service: container for container, service in (line.split() for line in output.splitlines() if line)} + + def address(self, container): + networks = json.loads( + self.checked(['docker', 'inspect', '--format', '{{json .NetworkSettings.Networks}}', container]) + ) + addresses = [network['IPAddress'] for network in networks.values() if network.get('IPAddress')] + if len(addresses) != 1: + raise RuntimeError('expected one private container network') + return addresses[0] + + @staticmethod + def ready(url): + deadline = time.monotonic() + 120 + opener = urllib.request.build_opener(urllib.request.ProxyHandler({})) + while time.monotonic() < deadline: + try: + with opener.open(url, timeout=2) as response: + payload = json.load(response) + if response.status == 200 and payload.get('ready') is not False and payload.get('code', 0) == 0: + return + except (urllib.error.URLError, TimeoutError, OSError, ValueError): + pass + time.sleep(0.5) + raise RuntimeError(f'readiness deadline exceeded: {url}') + + def prepare(self): + variables = { + **self.suite_profile.get('vars', {}), + 'profile_dir': str(self.suite_path.parent), + 'run_dir': str(self.output), + } + inputs = [plugin['file'] for plugin in self.suite_profile.get('plugins', [])] + inputs += [ + document['file'] + for item in self.suite_profile.get('resources', []) + for document in item.get('documents', []) + ] + for name in inputs: + path = Path(resolve(name, variables)).expanduser() + path = path if path.is_absolute() else self.suite_path.parent / path + if not path.is_file(): + raise ValueError(f'Job input file is missing: {path}') + self.state['phase'] = 'install' + self.save() + # Current Box runtimes reap sandboxes from previous runtime IDs on startup. + # Do not start one alongside another user's live sandbox on this daemon. + boxes = self.checked(['docker', 'ps', '-a', '--filter', 'label=langbot.box=true', '--format', '{{.ID}}']) + if boxes.strip(): + 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.redactions.append(key) + (self.private / 'api-key').write_text(key) + (self.private / 'api-key').chmod(0o600) + deployment = self.profile['deployment'] + self.cli_call( + [ + 'install', + '--dir', + str(self.deploy), + '--version', + deployment['version'], + '--profile', + 'all', + '--port', + str(deployment.get('port', 5300)), + ] + ) + self.recover_record() + self.load_redactions() + mock_results = self.output / 'mock' + mock_results.mkdir(mode=0o700) + override = { + 'services': { + 'langbot': { + 'environment': {'API__GLOBAL_API_KEY': key}, + 'volumes': [f'{self.deploy / "data/box"}:{self.deploy / "data/box"}'], + }, + 'regression_mock': { + 'image': self.profile.get('mock_image', 'python:3.12-alpine'), + 'volumes': [f'{Path(__file__).parent.resolve()}:/regression:ro', f'{mock_results}:/results'], + 'command': [ + 'python', + '/regression/mock_llm.py', + '--host', + '0.0.0.0', + '--port', + '18080', + '--record', + '/results/requests.jsonl', + ], + 'networks': ['langbot_network'], + 'labels': {'langbot.regression_job': self.state['job_id']}, + }, + }, + } + write_json(self.private / 'compose.override.json', override) + self.compose('up', '-d', override=True) + containers = self.project_containers() + for service in ('langbot', 'langbot_plugin_runtime', 'langbot_box', 'regression_mock'): + if service not in containers: + raise RuntimeError(f'missing deployment service: {service}') + endpoint = self.state['record']['endpoint'] + health = { + 'core': endpoint + '/healthz', + 'mock': f'http://{self.address(containers["regression_mock"])}:18080/healthz', + 'box': f'http://{self.address(containers["langbot_box"])}:5410/readyz', + } + for url in health.values(): + self.ready(url) + identity = self.cli_call(['--endpoint', endpoint, 'whoami']) + self.cli_call( + [ + 'context', + 'add', + 'regression', + '--endpoint', + endpoint, + '--api-key-env', + KEY_ENV, + '--expect-workspace', + identity['workspace_uuid'], + ] + ) + self.cli_call(['context', 'use', 'regression']) + self.state.update( + bound=True, workspace_uuid=identity['workspace_uuid'], containers=containers, phase='resources' + ) + self.save() + profile = self.suite_profile + 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(): + pid = int(self.checked(['docker', 'inspect', '--format', '{{.State.Pid}}', container]).strip()) + profile['observe']['pids'][service] = pid + destination = self.output / 'suite' + destination.mkdir(mode=0o700) + suite = Suite(profile, self.suite_path, destination, env=self.env) + 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) + 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() + suite.write('profile.json', profile) + return suite + + def owned_sandboxes(self): + 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 + + def collect(self): + self.recover_record() + self.load_redactions() + if self.state.get('record'): + logs = self.compose('logs', '--no-color', '--tail', '10000') + (self.output / 'services.log').write_text(logs) + + def cleanup(self): + self.recover_record() + self.load_redactions() + record = self.state.get('record') + if self.state['phase'] == 'cleaned': + if record and self.project_containers(): + raise RuntimeError('containers appeared after Job cleanup; inspect deployment before acting') + return + if record: + if ( + Path(record['compose_file']).resolve().parent != self.deploy + or Path(record['data_dir']).resolve() != self.deploy / 'data' + ): + raise RuntimeError('cleanup refused: deployment is outside this Job') + if self.state['bound']: + self.cli_call(['stop']) + else: + self.compose('stop', override=(self.private / 'compose.override.json').exists()) + sandboxes = self.owned_sandboxes() + if sandboxes: + self.checked(['docker', 'rm', '-f', *sandboxes]) + if self.state['bound']: + self.cli_call( + ['uninstall', '--yes', *(['--purge-data'] if self.profile.get('purge_data', True) else [])] + ) + else: + self.compose('down', '--remove-orphans', override=(self.private / 'compose.override.json').exists()) + if self.profile.get('purge_data', True) and (self.deploy / 'data').exists(): + shutil.rmtree(self.deploy / 'data') + for path in (self.private / 'api-key', self.private / 'compose.override.json'): + path.unlink(missing_ok=True) + self.state['phase'] = 'cleaned' + self.save() + + def run(self): + self.output.mkdir(parents=True, exist_ok=False, mode=0o700) + self.private.mkdir(mode=0o700) + (self.private / 'job-id').write_text(self.state['job_id']) + write_json(self.output / 'job-profile.json', self.profile) + self.save() + result = {'errors': [], 'completed': False} + try: + suite = self.prepare() + self.state['phase'] = 'traffic' + self.save() + summary = suite.run() + summary['passed'] = not summary['errors'] + suite.write('summary.json', summary) + result.update(completed=True, suite=summary, errors=list(summary['errors'])) + self.state['phase'] = 'finished' + self.save() + except KeyboardInterrupt: + result['errors'].append('Job interrupted; in-flight outcomes may be unknown') + except (RuntimeError, ValueError, OSError, KeyError, subprocess.TimeoutExpired) as exc: + result['errors'].append(self.redact(str(exc))) + finally: + if not result['completed'] and not result['errors']: + result['errors'].append('Job aborted before completion') + try: + self.collect() + except (RuntimeError, ValueError, OSError, subprocess.TimeoutExpired) as exc: + result['errors'].append('artifact collection: ' + self.redact(str(exc))) + policy = self.profile.get('cleanup', 'on-success') + if policy == 'always' or (policy == 'on-success' and not result['errors']): + try: + self.cleanup() + except (RuntimeError, ValueError, OSError, KeyError, subprocess.TimeoutExpired) as exc: + result['errors'].append('cleanup: ' + self.redact(str(exc))) + result.update(passed=not result['errors'], phase=self.state['phase']) + write_json(self.output / 'summary.json', result) + return result + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('action', choices=('validate', 'run', 'cleanup')) + parser.add_argument('--profile', type=Path) + parser.add_argument('--output', type=Path) + parser.add_argument('--duration', type=float) + parser.add_argument('--rps', type=float) + parser.add_argument('--workers', type=int) + args = parser.parse_args() + try: + if args.action == 'cleanup': + if args.output is None: + raise ValueError('--output is required for cleanup') + profile = json.loads((args.output / 'job-profile.json').read_text()) + job = Job(profile, {}, args.output / 'unused-profile.json', args.output) + job.state = json.loads(job.state_path.read_text()) + if (job.private / 'api-key').exists(): + key = (job.private / 'api-key').read_text() + job.env[KEY_ENV] = key + job.redactions.append(key) + job.cleanup() + return + if args.profile is None: + 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')]: + if getattr(args, flag) is not None: + phase[field] = getattr(args, flag) + validate(suite) + if args.action == 'validate': + print('Job profile valid; no deployment or backend changed.') + return + if args.output is None: + raise ValueError('--output is required for run') + job = Job(profile, suite, path, args.output) + result = job.run() + print(json.dumps(result, ensure_ascii=False, indent=2)) + if not result['passed']: + raise SystemExit(1) + except (RuntimeError, ValueError, OSError, KeyError, subprocess.TimeoutExpired) as exc: + parser.exit(1, f'{type(exc).__name__}: {exc}\n') + + +if __name__ == '__main__': + main() diff --git a/tools/regression_profiles/job.json b/tools/regression_profiles/job.json new file mode 100644 index 000000000..fe31035a9 --- /dev/null +++ b/tools/regression_profiles/job.json @@ -0,0 +1,10 @@ +{ + "version": 1, + "lbctl": "lbctl", + "suite": "normal.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_profiles/normal.json b/tools/regression_profiles/normal.json new file mode 100644 index 000000000..6c5bd78e8 --- /dev/null +++ b/tools/regression_profiles/normal.json @@ -0,0 +1,111 @@ +{ + "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" + }, + "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 + }} + }, + "safety": {"rate-limit": {"limitation": 10000}}, + "output": {"force-delay": {"min": 0, "max": 0}} + } + }}, + {"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": "{global}", + "timeout": 30, "max-tool-iterations": 5 + }} + }, + "safety": {"rate-limit": {"limitation": 10000}}, + "output": {"force-delay": {"min": 0, "max": 0}} + } + }}, + {"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 + }} + }, + "safety": {"rate-limit": {"limitation": 10000}}, + "output": {"force-delay": {"min": 0, "max": 0}} + } + }} + ], + "cases": [ + {"name": "chat", "weight": 2, "args": ["pipeline", "run", "${chat_pipeline}", "--message", "Run the normal QA scenario", "--timeout", "90s"], "expect": "LANGBOT_CHAT_OK"}, + {"name": "box", "weight": 1, "args": ["pipeline", "run", "${box_pipeline}", "--message", "Run the normal QA scenario", "--timeout", "90s"], "expect": "LANGBOT_BOX_OK"}, + {"name": "rag_chat", "weight": 1, "args": ["pipeline", "run", "${rag_pipeline}", "--message", "What is the documented parser-rag sentinel?", "--timeout", "90s"], "expect": "LANGBOT_RAG_OK"}, + {"name": "rag_retrieve", "weight": 1, "args": ["knowledge-base", "retrieve", "${knowledge_base}"], "body": {"query": "What is the parser-rag golden sentinel and parser-engine pair?", "retrieval_settings": {"search_type": "vector", "top_k": 5}}, "expect": "aurora-parser-rag-9137"} + ], + "phases": [{"name": "normal", "duration_seconds": 600, "rps": 2, "workers": 4, "cooldown_seconds": 30}], + "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.95, "max_p95_seconds": {"chat": 3, "box": 6, "rag_chat": 4, "rag_retrieve": 1}} +} diff --git a/tools/regression_suite.py b/tools/regression_suite.py new file mode 100644 index 000000000..65a685107 --- /dev/null +++ b/tools/regression_suite.py @@ -0,0 +1,409 @@ +#!/usr/bin/env python3 +"""Prepare an lbctl regression from JSON, run load phases, and clean owned resources.""" + +import argparse +import copy +import json +import math +import os +from pathlib import Path +import re +import subprocess +import sys + + +KINDS = {'provider', 'model', 'pipeline', 'knowledge-base'} +REFERENCE = re.compile(r'\$\{([a-zA-Z][a-zA-Z0-9_-]*)\}') + + +def resolve(value, variables): + if isinstance(value, dict): + return {key: resolve(item, variables) for key, item in value.items()} + if isinstance(value, list): + return [resolve(item, variables) for item in value] + if isinstance(value, str): + return REFERENCE.sub(lambda match: str(variables[match[1]]), value) + return value + + +def merge(base, patch): + result = copy.deepcopy(base) + for key, value in patch.items(): + result[key] = ( + merge(result[key], value) if isinstance(value, dict) and isinstance(result.get(key), dict) else value + ) + return result + + +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') + if value < minimum or (integer and not isinstance(value, int)): + raise ValueError(f'{label} must be {"an integer " if integer else ""}>= {minimum}') + + +def validate(profile): + if not isinstance(profile, dict) or profile.get('version') != 1: + raise ValueError('profile version must be 1') + allowed = {'version', 'cli', 'vars', 'plugins', 'resources', 'cases', 'phases', 'observe', 'cleanup', 'thresholds'} + if extra := profile.keys() - allowed: + raise ValueError(f'unknown profile fields: {sorted(extra)}') + for field in ('cli', 'vars', 'observe', 'thresholds'): + if field in profile and not isinstance(profile[field], dict): + raise ValueError(f'{field} must be an object') + for field in ('plugins', 'resources', 'cases', 'phases'): + if field in profile and ( + not isinstance(profile[field], list) or any(not isinstance(item, dict) for item in profile[field]) + ): + raise ValueError(f'{field} must be an array of objects') + cli = profile.get('cli', {}) + if not isinstance(cli, dict) or cli.keys() - {'config', 'executable'}: + 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'} + supplied = profile.get('vars', {}) + if not isinstance(supplied, dict) or variables.keys() & supplied.keys(): + raise ValueError('vars must be an object without reserved names') + if any(not isinstance(value, (str, int, float)) or '${' in str(value) for value in supplied.values()): + raise ValueError('vars values must be scalar literals, without nested references') + variables.update(supplied) + for plugin in profile.get('plugins', []): + if plugin.keys() - {'source', 'file'}: + raise ValueError('plugins accept source and file') + if plugin.get('source') not in {'local', 'github', 'marketplace'} or not plugin.get('file'): + raise ValueError('plugins need source local/github/marketplace and file') + resolve(plugin, variables) + for item in profile.get('resources', []): + if item.keys() - {'id', 'kind', 'type', 'body', 'documents'}: + raise ValueError('unknown resource field') + name = item.get('id', '') + if not re.fullmatch(r'[a-zA-Z][a-zA-Z0-9_-]*', name) or name in variables: + raise ValueError(f'invalid or duplicate resource id: {name}') + if item.get('kind') not in KINDS or not isinstance(item.get('body'), dict): + 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'] == '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. + variables[name] = name + documents = item.get('documents', []) + if not isinstance(documents, list) or any(not isinstance(doc, dict) for doc in documents): + raise ValueError(f'{name}: documents must be an array of objects') + for document in documents: + if item['kind'] != 'knowledge-base' or not document.get('file') or not document.get('parser_plugin_id'): + raise ValueError(f'{name}: documents need file and parser_plugin_id on a knowledge-base') + resolve(document, variables) + cases = profile.get('cases') + if not isinstance(cases, list) or not cases: + raise ValueError('cases must be a non-empty array') + names = set() + for case in cases: + if case.keys() - {'name', 'weight', 'args', 'expect', 'body'}: + 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 ( + not isinstance(case.get('args'), list) + or not case['args'] + or not all(isinstance(x, str) for x in case['args']) + ): + raise ValueError('case args must be a non-empty string array') + if not isinstance(case.get('expect'), str) or not case['expect']: + raise ValueError('case expect must be a non-empty string') + number(case.get('weight', 1), 'case weight', minimum=1, integer=True) + resolve(case, variables) + if 'body' in case and (not isinstance(case['body'], dict) or '--file' in case['args']): + raise ValueError('case body must be an object and cannot coexist with --file') + if sum(case.get('weight', 1) for case in cases) > 1000: + raise ValueError('total case weight must be <= 1000') + phases = profile.get('phases') + if not isinstance(phases, list) or not phases: + raise ValueError('phases must be a non-empty array') + names = set() + for phase in phases: + name = phase.get('name', '') + 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'}: + 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) + number(phase.get('workers'), 'workers', minimum=1, integer=True) + number(phase.get('cooldown_seconds', 0), 'cooldown_seconds') + if phase['workers'] > 32: + raise ValueError('workers must be <= 32') + 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', {}) + if extra := thresholds.keys() - {'min_rps_ratio', 'max_p95_seconds'}: + raise ValueError(f'unknown thresholds: {sorted(extra)}') + number(thresholds.get('min_rps_ratio', 0), 'min_rps_ratio') + if thresholds.get('min_rps_ratio', 0) > 1: + raise ValueError('min_rps_ratio must be <= 1') + if not isinstance(thresholds.get('max_p95_seconds', {}), dict): + raise ValueError('max_p95_seconds must map case names to seconds') + for name, value in thresholds.get('max_p95_seconds', {}).items(): + if name not in {case['name'] for case in cases}: + 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.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(): + if not isinstance(name, str) or not isinstance(url, str) or not url.startswith(('http://', 'https://')): + raise ValueError('health endpoints must map names to HTTP(S) URLs') + for field in ('pids', 'pid_files'): + if not isinstance(observed.get(field, {}), dict): + raise ValueError(f'observe.{field} must be an object') + for pid in observed.get('pids', {}).values(): + number(pid, 'PID', minimum=1, integer=True) + if not isinstance(observed.get('containers', []), list) or any( + not isinstance(item, str) or not item for item in observed.get('containers', []) + ): + raise ValueError('observe.containers must be an array of container names/IDs') + + +class Suite: + def __init__(self, profile, profile_path, output, env=None): + self.profile = profile + self.output = output.resolve() + self.variables = { + **profile.get('vars', {}), + 'profile_dir': str(profile_path.parent.resolve()), + 'run_dir': str(self.output), + } + self.cli = [ + profile.get('cli', {}).get('executable', 'lbctl'), + '--config', + self.path(profile['cli']['config']), + '-o', + '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] + + def path(self, value): + path = Path(resolve(value, self.variables)).expanduser() + return str(path if path.is_absolute() else Path(self.variables['profile_dir']) / path) + + def write(self, name, value): + path = self.output / name + path.write_text(json.dumps(value, ensure_ascii=False, indent=2)) + path.chmod(0o600) + return str(path) + + def call(self, args, body=None): + command = [*self.cli, *args] + if body is not None: + command += ['--file', self.write('request.json', body)] + try: + result = subprocess.run(command, capture_output=True, text=True, timeout=360, env=self.env) + except subprocess.TimeoutExpired as exc: + raise RuntimeError('lbctl timed out; outcome unknown, no automatic retry') from exc + if result.returncode: + 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'] + + def journal(self): + self.write('owned.json', {'workspace_uuid': self.variables['workspace_uuid'], 'resources': self.owned}) + + def prepare(self): + identity = self.call(['whoami']) + self.variables['workspace_uuid'] = identity['workspace_uuid'] + self.journal() + for plugin in self.profile.get('plugins', []): + self.call(['plugin', 'install', plugin['source'], '--file', self.path(plugin['file']), '--wait']) + for item in self.profile.get('resources', []): + kind = item['kind'] + body = resolve(item['body'], self.variables) + config = body.pop('config', None) if kind == 'pipeline' else None + args = [kind, 'apply' if kind == 'pipeline' else 'create'] + if kind == 'model': + args += ['--type', item['type']] + created = 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 config is not None: + current = self.call(['pipeline', 'get', uuid]) + current = current.get('pipeline', current) + self.call( + ['pipeline', 'apply'], + { + 'uuid': uuid, + 'name': current['name'], + 'description': current['description'], + 'config': merge(current['config'], config), + }, + ) + for document in item.get('documents', []): + self.call( + [ + 'knowledge-base', + 'ingest', + uuid, + '--file', + self.path(document['file']), + '--parser-plugin-id', + document['parser_plugin_id'], + '--wait', + ] + ) + self.write( + 'resources.json', + {key: value for key, value in self.variables.items() if key not in self.profile.get('vars', {})}, + ) + + def cleanup(self): + saved = json.loads((self.output / 'owned.json').read_text()) + identity = self.call(['whoami']) + if saved['workspace_uuid'] != identity['workspace_uuid']: + raise RuntimeError('cleanup refused: Workspace differs from owned.json') + failures = [] + for resource in reversed(saved['resources'].copy()): + args = [resource['kind'], 'delete', resource['uuid'], '--yes'] + if resource['kind'] == 'model': + args += ['--type', resource['type']] + try: + self.call(args) + except (RuntimeError, ValueError, OSError) as exc: + failures.append(str(exc)) + else: + saved['resources'].remove(resource) + self.write('owned.json', saved) + return failures + + def run(self): + cases = [] + for original in self.profile['cases']: + 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_path = self.write('cases.json', cases) + observed = resolve(self.profile.get('observe', {}), self.variables) + observe_args = [] + for name, pid in observed.get('pids', {}).items(): + observe_args += ['--pid', f'{name}={int(pid)}'] + for name, path in observed.get('pid_files', {}).items(): + observe_args += ['--pid', f'{name}={int(Path(self.path(path)).read_text().strip())}'] + for name, url in observed.get('health', {}).items(): + observe_args += ['--health', f'{name}={url}'] + for container in observed.get('containers', []): + observe_args += ['--container', container] + results = [] + errors = [] + for phase in self.profile['phases']: + destination = self.output / phase['name'] + command = [ + sys.executable, + str(Path(__file__).with_name('regression_traffic.py')), + '--lbctl', + self.cli[0], + '--config', + self.cli[2], + '--cases', + cases_path, + '--output', + str(destination), + '--duration', + str(phase['duration_seconds']), + '--rps', + str(phase['rps']), + '--workers', + str(phase['workers']), + '--cooldown', + str(phase.get('cooldown_seconds', 0)), + '--shared-case-sequence', + *observe_args, + ] + with (self.output / f'{phase["name"]}.log').open('w') as log: + result = subprocess.run(command, stdout=log, stderr=subprocess.STDOUT, env=self.env) + summary = json.loads((destination / 'summary.json').read_text()) + summary['name'] = phase['name'] + summary['achieved_rps'] = summary['requests'] / summary['load_elapsed_seconds'] + results.append(summary) + if result.returncode: + errors.append(f'{phase["name"]}: business or health/resource checks failed') + thresholds = self.profile.get('thresholds', {}) + if summary['achieved_rps'] < phase['rps'] * thresholds.get('min_rps_ratio', 0): + errors.append(f'{phase["name"]}: achieved RPS below configured threshold') + for name, maximum in thresholds.get('max_p95_seconds', {}).items(): + p95 = summary['cases'][name]['p95_seconds'] + if p95 is None or p95 > maximum: + errors.append(f'{phase["name"]}/{name}: P95 missing or above configured threshold') + return {'phases': results, 'errors': errors} + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('action', choices=('validate', 'run', 'cleanup')) + parser.add_argument('--profile', required=True, type=Path) + parser.add_argument('--output', type=Path, help='New run directory; existing run directory for cleanup') + 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') + 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')]: + if getattr(args, flag) is not None: + phase[field] = getattr(args, flag) + validate(profile) + if args.action == 'validate': + print('Profile valid (no services or resources changed).') + return + if args.output is None: + raise ValueError('--output is required for run/cleanup') + suite = Suite(profile, args.profile.resolve(), args.output) + if args.action == 'cleanup': + errors = suite.cleanup() + if errors: + raise RuntimeError('\n'.join(errors)) + return + args.output.mkdir(parents=True, exist_ok=False, mode=0o700) + suite.write('profile.json', profile) + result = {'phases': [], 'errors': []} + completed = False + try: + suite.prepare() + result = suite.run() + completed = True + except KeyboardInterrupt: + result['errors'].append('Interrupted; in-flight outcomes may be unknown. Inspect retained resources.') + except (RuntimeError, ValueError, OSError, KeyError) as exc: + result['errors'].append(str(exc)) + finally: + if not completed and not result['errors']: + result['errors'].append('Run aborted before completion; inspect logs and retained resources.') + policy = profile.get('cleanup', 'on-success') + if policy == 'always' or (policy == 'on-success' and not result['errors']): + if (args.output / 'owned.json').exists(): + try: + result['errors'].extend(suite.cleanup()) + except (RuntimeError, ValueError, OSError, KeyError) as exc: + result['errors'].append(f'cleanup failed: {exc}') + result['passed'] = not result['errors'] + suite.write('summary.json', result) + print(json.dumps(result, ensure_ascii=False, indent=2)) + if not result['passed']: + raise SystemExit(1) + except (RuntimeError, ValueError, OSError, KeyError) as exc: + parser.exit(1, f'{type(exc).__name__}: {exc}\n') + + +if __name__ == '__main__': + main() diff --git a/tools/regression_traffic.py b/tools/regression_traffic.py new file mode 100644 index 000000000..67a458db4 --- /dev/null +++ b/tools/regression_traffic.py @@ -0,0 +1,220 @@ +#!/usr/bin/env python3 +"""Drive normal lbctl traffic and retain business checks, health and Linux resources.""" + +import argparse +from collections import Counter +from concurrent.futures import ThreadPoolExecutor +import json +import math +import os +from pathlib import Path +import subprocess +import sys +import threading +import time + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'scripts')) +from cloud_runtime_soak import read_endpoint_snapshot, read_process_snapshot + + +def percentile(values, fraction): + return sorted(values)[max(0, math.ceil(len(values) * fraction) - 1)] if values else None + + +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('--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('--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( + '--shared-case-sequence', action='store_true', help='Share the weighted case sequence across workers' + ) + args = parser.parse_args() + if ( + not all(math.isfinite(value) for value in (args.duration, args.rps, args.cooldown)) + or args.duration <= 0 + or args.rps <= 0 + or not 1 <= args.workers <= 32 + or args.cooldown < 0 + ): + parser.error('duration/rps must be positive, workers 1..32, cooldown nonnegative') + cases = json.loads(args.cases.read_text()) + if ( + not isinstance(cases, list) + or not cases + 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 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') + 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() + 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] + + def redact(text): + for secret in redactions: + text = text.replace(secret, '[REDACTED]') + return text + + def sample(): + with (args.output / 'resources.jsonl').open('w') as stream: + while not stop.is_set(): + row = { + 'time': time.time(), + 'elapsed': time.monotonic() - started, + 'phase': phase[0], + 'targets': {}, + 'health': {}, + } + for name, pid in pids.items(): + try: + row['targets'][name] = read_process_snapshot(pid) + except (OSError, RuntimeError) as exc: + row['targets'][name] = {'error': str(exc)} + for name, url in endpoints.items(): + try: + row['health'][name] = read_endpoint_snapshot(url, timeout_seconds=3) + except RuntimeError as exc: + row['health'][name] = {'error': str(exc)} + if args.container: + result = subprocess.run( + ['docker', 'stats', '--no-stream', '--format', '{{json .}}', *args.container], + capture_output=True, + text=True, + timeout=10, + ) + row['containers'] = [json.loads(line) for line in result.stdout.splitlines() if line] + if result.returncode: + row['container_error'] = result.stderr.strip() + stream.write(json.dumps(row) + '\n') + stream.flush() + with lock: + progress = { + 'elapsed': row['elapsed'], + 'phase': phase[0], + 'requests': len(rows), + 'status': dict(Counter(item['status'] for item in rows)), + } + (args.output / 'progress.json').write_text(json.dumps(progress)) + stop.wait(3) + + def worker(index): + nonlocal next_case + iteration = 0 + due = started + index / args.rps + while due < deadline: + if stop.wait(max(0, due - time.monotonic())) or time.monotonic() >= deadline: + break + if args.shared_case_sequence: + with lock: + case = cases[next_case % len(cases)] + next_case += 1 + else: + case = cases[(iteration * args.workers + index) % len(cases)] + before = time.monotonic() + row = {'case': case['name'], 'time': time.time(), 'elapsed': before - started, 'status': 'failed'} + try: + result = subprocess.run( + [args.lbctl, '--config', args.config, '-o', 'json', *case['args']], + capture_output=True, + text=True, + timeout=90, + ) + row['exit_code'] = result.returncode + row['output'] = json.loads(redact(result.stdout)) + if result.returncode == 0 and case['expect'] in json.dumps(row['output'], ensure_ascii=False): + row['status'] = 'ok' + elif 'result_unknown' in result.stdout: + row['status'] = 'unknown' + else: + row['error'] = redact(result.stderr)[:2000] or 'exit code or response oracle failed' + except subprocess.TimeoutExpired: + row.update(status='unknown', error='CLI timed out; server outcome unknown; request was not retried') + except (OSError, ValueError) as exc: + row['error'] = str(exc) + row['seconds'] = time.monotonic() - before + with lock: + rows.append(row) + with (args.output / 'requests.jsonl').open('a') as stream: + stream.write(json.dumps(row, ensure_ascii=False) + '\n') + iteration += 1 + due = max(due + args.workers / args.rps, time.monotonic()) + + def sampling_guard(): + try: + sample() + except Exception as exc: + sampling_errors.append(f'{type(exc).__name__}: {exc}') + stop.set() + + sampler = threading.Thread(target=sampling_guard) + sampler.start() + try: + with ThreadPoolExecutor(max_workers=args.workers) as pool: + list(pool.map(worker, range(args.workers))) + stop.wait(max(0, deadline - time.monotonic())) + load_elapsed = time.monotonic() - started + phase[0] = 'cooldown' + stop.wait(args.cooldown) + finally: + stop.set() + sampler.join() + summary = { + 'duration_requested_seconds': args.duration, + 'load_elapsed_seconds': load_elapsed, + 'target_rps': args.rps, + 'workers': args.workers, + 'cooldown_seconds': args.cooldown, + 'requests': len(rows), + 'status': dict(Counter(row['status'] for row in rows)), + 'cases': {}, + } + samples = [json.loads(line) for line in (args.output / 'resources.jsonl').read_text().splitlines()] + summary['health_samples'] = len(samples) + summary['health_failures'] = sum( + any('error' in target for target in [*sample['targets'].values(), *sample['health'].values()]) + or bool(sample.get('container_error')) + for sample in samples + ) + summary['sampling_errors'] = sampling_errors + 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'] + summary['cases'][name] = { + 'requests': len(selected), + 'status': dict(Counter(row['status'] for row in selected)), + 'p50_seconds': percentile(times, 0.5), + 'p95_seconds': percentile(times, 0.95), + 'p99_seconds': percentile(times, 0.99), + } + (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: + raise SystemExit(1) + + +if __name__ == '__main__': + main() diff --git a/tools/test_mock_llm.py b/tools/test_mock_llm.py new file mode 100644 index 000000000..7deb07bf5 --- /dev/null +++ b/tools/test_mock_llm.py @@ -0,0 +1,196 @@ +"""Exercise the real HTTP server without LangBot, credentials or dependencies.""" + +from concurrent.futures import ThreadPoolExecutor +import base64 +import json +from pathlib import Path +import tempfile +import struct +import threading +import unittest +from urllib.error import HTTPError +from urllib.request import Request, urlopen + +from mock_llm import MockServer + + +class MockLLMTest(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.tmp = tempfile.TemporaryDirectory() + cls.record = Path(cls.tmp.name) / 'requests.jsonl' + scenarios = json.loads(Path(__file__).with_name('mock_llm_scenarios.json').read_text()) + cls.server = MockServer(('127.0.0.1', 0), scenarios, cls.record) + cls.thread = threading.Thread(target=cls.server.serve_forever) + cls.thread.start() + cls.url = f'http://127.0.0.1:{cls.server.server_port}' + + @classmethod + def tearDownClass(cls): + cls.server.shutdown() + cls.server.server_close() + cls.thread.join() + cls.tmp.cleanup() + + def post(self, model='mock-chat', **kwargs): + body = { + 'model': model, + 'messages': [{'role': 'user', 'content': 'test'}], + **kwargs, + } + request = Request( + self.url + '/v1/chat/completions', + data=json.dumps(body).encode(), + headers={ + 'Content-Type': 'application/json', + 'Authorization': 'Bearer must-not-be-recorded', + }, + ) + with urlopen(request, timeout=5) as response: + return response.read().decode() + + def test_chat_and_model_discovery(self): + with urlopen(self.url + '/v1/models') as response: + models = json.load(response) + self.assertIn('mock-box', [m['id'] for m in models['data']]) + self.assertEqual( + json.loads(self.post())['choices'][0]['message']['content'], + 'LANGBOT_CHAT_OK', + ) + + def test_stream_and_usage(self): + events = self.post(stream=True, stream_options={'include_usage': True}).splitlines() + self.assertEqual(events[-2], 'data: [DONE]') + chunks = [json.loads(line[6:]) for line in events if line.startswith('data: {')] + content = ''.join(c['choices'][0]['delta'].get('content', '') for c in chunks if c['choices']) + self.assertEqual(content, 'LANGBOT_CHAT_OK') + self.assertEqual(chunks[-1]['usage']['total_tokens'], 15) + + def test_embedding_http_float_base64_and_validation(self): + def embed(**patch): + body = {'model': 'mock-embedding', 'input': ['repeatable fixture', 'another fixture'], **patch} + request = Request( + self.url + '/v1/embeddings', + data=json.dumps(body).encode(), + headers={'Content-Type': 'application/json'}, + ) + with urlopen(request, timeout=5) as response: + return json.load(response) + + first = embed()['data'] + self.assertEqual(first, embed()['data']) + self.assertEqual([item['index'] for item in first], [0, 1]) + self.assertEqual(len(first[0]['embedding']), 384) + self.assertAlmostEqual(sum(value * value for value in first[0]['embedding']), 1) + encoded = embed(encoding_format='base64')['data'][0]['embedding'] + decoded = struct.unpack('<384f', base64.b64decode(encoded)) + for actual, expected in zip(decoded, first[0]['embedding']): + self.assertAlmostEqual(actual, expected, places=6) + for patch in ({'input': []}, {'dimensions': 0}, {'encoding_format': 'bad'}, {'model': 'unknown'}): + with self.subTest(patch=patch), self.assertRaises(HTTPError) as caught: + embed(**patch) + self.assertEqual(caught.exception.code, 400) + + def box_turn(self, streaming=False): + messages = [{'role': 'user', 'content': 'run box checks'}] + tools = [ + { + 'type': 'function', + 'function': {'name': 'exec', 'parameters': {'type': 'object'}}, + } + ] + commands = [] + for _ in range(2): + wire = self.post('mock-box', messages=messages, tools=tools, stream=streaming) + if streaming: + deltas = [ + json.loads(line[6:])['choices'][0]['delta'] + for line in wire.splitlines() + if line.startswith('data: {') + ] + call = next(d['tool_calls'][0] for d in deltas if d.get('tool_calls') and 'id' in d['tool_calls'][0]) + arguments = ''.join(d['tool_calls'][0]['function']['arguments'] for d in deltas if d.get('tool_calls')) + call = { + **call, + 'function': {**call['function'], 'arguments': arguments}, + } + call.pop('index') + message = {'role': 'assistant', 'content': None, 'tool_calls': [call]} + else: + message = json.loads(wire)['choices'][0]['message'] + call = message['tool_calls'][0] + commands.append(json.loads(call['function']['arguments'])['command']) + messages.extend( + [ + message, + { + 'role': 'tool', + 'tool_call_id': call['id'], + 'content': 'LANGBOT_BOX_SENTINEL', + }, + ] + ) + final = json.loads(self.post('mock-box', messages=messages, tools=tools)) + self.assertEqual(final['choices'][0]['message']['content'], 'LANGBOT_BOX_OK') + self.assertIn('printf LANGBOT_BOX_SENTINEL', commands[0]) + self.assertEqual(commands[1], 'cat /workspace/mock-llm.txt') + + def test_tool_rounds_and_fragmented_stream(self): + self.box_turn() + self.box_turn(streaming=True) + + def test_concurrent_turns_do_not_share_progress(self): + with ThreadPoolExecutor(max_workers=4) as pool: + list(pool.map(lambda _: self.box_turn(), range(8))) + + def test_rag_requires_expected_model_input(self): + with self.assertRaises(HTTPError) as caught: + self.post('mock-rag') + self.assertEqual(caught.exception.code, 400) + response = json.loads( + self.post( + 'mock-rag', + messages=[ + {'role': 'system', 'content': 'Retrieved document: aurora-parser-rag-9137'}, + {'role': 'user', 'content': 'What is the documented sentinel?'}, + ], + ) + ) + self.assertEqual(response['choices'][0]['message']['content'], 'LANGBOT_RAG_OK') + + def test_errors_and_invalid_requests(self): + for status in (401, 429, 503): + with self.subTest(status=status), self.assertRaises(HTTPError) as caught: + self.post(f'mock-{status}') + self.assertEqual(caught.exception.code, status) + for kwargs in ( + {'model': 'unknown'}, + {'messages': []}, + {'stream_options': []}, + {'model': 'mock-box', 'tools': []}, + ): + with self.subTest(kwargs=kwargs), self.assertRaises(HTTPError) as caught: + self.post(**kwargs) + self.assertEqual(caught.exception.code, 400) + messages = [ + {'role': 'user', 'content': 'test'}, + {'role': 'assistant', 'tool_calls': [{'id': 'call1'}]}, + ] + with self.assertRaises(HTTPError) as caught: + self.post('mock-box', messages=messages) + self.assertEqual(caught.exception.code, 400) + messages.append({'role': 'tool', 'tool_call_id': 'call1', 'content': 'wrong output'}) + with self.assertRaises(HTTPError) as caught: + self.post('mock-box', messages=messages) + self.assertEqual(caught.exception.code, 400) + + def test_record_omits_authorization(self): + self.post() + with self.server.record_lock: + recorded = self.record.read_text() + self.assertNotIn('must-not-be-recorded', recorded) + self.assertIn('"request_id"', recorded) + + +if __name__ == '__main__': + unittest.main() diff --git a/tools/test_regression_job.py b/tools/test_regression_job.py new file mode 100644 index 000000000..53008e325 --- /dev/null +++ b/tools/test_regression_job.py @@ -0,0 +1,179 @@ +"""Exercise Job ownership and lifecycle with controlled CLI/Docker responses.""" + +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +import json +import os +from pathlib import Path +import shutil +import sys +import tempfile +import threading +import unittest +from unittest.mock import patch + +import yaml + +from regression_job import Job, main +from test_regression_suite import FAKE_CLI, profile + + +class ControlledJob(Job): + def __init__(self, manifest, suite, path, output, health, fail_install=False): + super().__init__(manifest, suite, path, output) + self.health = health + self.fail_install = fail_install + self.calls = [] + self.env['LBCTL_TEST_ROOT'] = str(output.parent) + + def command(self, args, timeout=900): + self.calls.append(args) + if args[: len(self.cli)] == self.cli: + operation = args[len(self.cli) :] + if operation[0] == 'install': + self.deploy.mkdir() + (self.deploy / 'data').mkdir() + (self.deploy / 'compose.yaml').write_text('services: {}') + (self.deploy / '.env').write_text( + 'LANGBOT_PLUGIN_RUNTIME_CONTROL_TOKEN=' + 't' * 32 + '\nLANGBOT_BOX_ENABLED=true\n' + ) + record = { + 'deployment_id': 'owned-deployment', + 'compose_project': 'test-project', + 'compose_file': str(self.deploy / 'compose.yaml'), + 'data_dir': str(self.deploy / 'data'), + 'env_file': str(self.deploy / '.env'), + 'endpoint': self.health, + } + (self.deploy / 'deployment.yaml').write_text(yaml.safe_dump(record)) + # Simulate a timeout/failed final probe with no returned record. + return ( + (1, json.dumps({'data': {}}), 'install failed') + if self.fail_install + else (0, json.dumps({'data': {'record': record}}), '') + ) + if operation[0] == '--endpoint': + return 0, json.dumps({'data': {'workspace_uuid': 'test-workspace'}}), '' + if operation[0] == 'uninstall': + shutil.rmtree(self.deploy / 'data') + return 0, json.dumps({'data': {}}), '' + if args[:2] == ['docker', 'ps']: + if 'label=langbot.box=true' in args: + return 0, '', '' + return ( + 0, + '\n'.join( + f'id-{name} {name}' + for name in ('langbot', 'langbot_plugin_runtime', 'langbot_box', 'regression_mock') + ), + '', + ) + if args[:2] == ['docker', 'inspect']: + if '{{.State.Pid}}' in args: + return 0, str(os.getpid()), '' + return 0, json.dumps({'network': {'IPAddress': '127.0.0.1'}}), '' + if args[:2] == ['docker', 'compose']: + if 'logs' in args: + return 0, self.redact(self.env.get('LANGBOT_QA_API_KEY', '') + '\n' + 't' * 32), '' + return 0, '', '' + raise AssertionError(args) + + @staticmethod + def ready(url): + pass + + def prepare(self): + suite = super().prepare() + # Real traffic driver and HTTP health probe; infrastructure calls are controlled above. + suite.profile['observe'] = {'health': {'core': self.health}} + return suite + + +class JobTests(unittest.TestCase): + def test_lifecycle_success_and_partial_install_recovery(self): + class Health(BaseHTTPRequestHandler): + def do_GET(self): + self.send_response(200) + self.end_headers() + self.wfile.write(b'{"code":0}') + + def log_message(self, *args): + pass + + server = ThreadingHTTPServer(('127.0.0.1', 0), Health) + thread = threading.Thread(target=server.serve_forever) + thread.start() + try: + with tempfile.TemporaryDirectory() as folder: + root = Path(folder) + cli = root / 'lbctl-test' + cli.write_text(f'#!{sys.executable}\n' + FAKE_CLI) + cli.chmod(0o700) + manifest = {'lbctl': str(cli), 'deployment': {'version': 'v4.11.0', 'port': 15351}} + suite = profile() + suite['resources'] = [{'id': 'chat', 'kind': 'pipeline', 'body': {'name': 'QA', 'config': {'ai': {}}}}] + suite['cases'][0].update(args=['pipeline', 'run', '${chat}'], expect='TEST_OK') + suite['phases'][0].update(duration_seconds=0.5, rps=4, workers=2) + health = f'http://127.0.0.1:{server.server_port}' + job = ControlledJob(manifest, suite, root / 'profile.json', root / 'success', health) + result = job.run() + self.assertTrue(result['passed'], result) + self.assertEqual(result['phase'], 'cleaned') + self.assertFalse((job.deploy / 'data').exists()) + self.assertFalse((job.private / 'api-key').exists()) + self.assertIn('--purge-data', job.calls[-1]) + self.assertNotIn('t' * 32, (job.output / 'services.log').read_text()) + + job = ControlledJob(manifest, suite, root / 'profile.json', root / 'failed', health, fail_install=True) + result = job.run() + self.assertFalse(result['passed']) + self.assertEqual(job.state['record']['deployment_id'], 'owned-deployment') + self.assertTrue((job.deploy / 'data').exists()) + self.assertFalse(any('uninstall' in call for call in job.calls)) + with ( + patch.object(sys, 'argv', ['regression_job.py', 'cleanup', '--output', str(job.output)]), + patch('regression_job.Job', return_value=job), + ): + main() + self.assertFalse((job.deploy / 'data').exists()) + self.assertEqual(job.state['phase'], 'cleaned') + finally: + server.shutdown() + server.server_close() + thread.join() + + def test_cleanup_refuses_changed_deployment(self): + with tempfile.TemporaryDirectory() as folder: + root = Path(folder) + job = Job({}, profile(), root / 'profile.json', root) + job.private.mkdir() + job.deploy.mkdir() + (job.private / 'job-id').write_text(job.state['job_id']) + record = { + 'deployment_id': 'old', + 'compose_file': str(job.deploy / 'compose.yaml'), + 'data_dir': str(job.deploy / 'data'), + 'env_file': str(job.deploy / '.env'), + } + job.state['record'] = record + (job.deploy / 'deployment.yaml').write_text(yaml.safe_dump({**record, 'deployment_id': 'new'})) + with self.assertRaisesRegex(RuntimeError, 'identity changed'): + job.cleanup() + + def test_sandbox_cleanup_uses_owned_mount_roots(self): + with tempfile.TemporaryDirectory() as folder: + root = Path(folder) + job = Job({}, profile(), root / 'profile.json', root) + job.state['record'] = {'data_dir': str(job.deploy / 'data')} + + 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}]) + + job.checked = checked + self.assertEqual(job.owned_sandboxes(), ['ours']) + + +if __name__ == '__main__': + unittest.main() diff --git a/tools/test_regression_suite.py b/tools/test_regression_suite.py new file mode 100644 index 000000000..bd9cbfd15 --- /dev/null +++ b/tools/test_regression_suite.py @@ -0,0 +1,228 @@ +"""Focused checks for profile validation, ownership and the CLI preparation sequence.""" + +import copy +from contextlib import redirect_stdout +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +import io +import json +import os +from pathlib import Path +import subprocess +import sys +import tempfile +import threading +import unittest +from unittest.mock import patch + +from regression_suite import Suite, main, merge, resolve, validate + + +FAKE_CLI = r""" +import json, os, sys +from pathlib import Path +args = sys.argv[5:] +root = Path(os.environ['LBCTL_TEST_ROOT']) +state_path = root / 'cli-state.json' +state = json.loads(state_path.read_text()) if state_path.exists() else {} +body = json.loads(Path(args[args.index('--file') + 1]).read_text()) if '--file' in args else {} +with (root / 'cli-calls.jsonl').open('a') as log: + log.write(json.dumps({'args': args, 'body': body}) + '\n') +if args == ['whoami']: + data = {'workspace_uuid': 'test-workspace'} +elif args[:2] == ['pipeline', 'run']: + assert args[2] in state + data = {'status': 'completed', 'reply': 'TEST_OK'} +elif args[1] == 'delete': + del state[args[2]] + state_path.write_text(json.dumps(state)) + data = {} +elif args[:2] == ['pipeline', 'get']: + data = state[args[2]] +elif args[1] in ['create', 'apply']: + uuid = body.get('uuid', f'created-{len(state)}') + data = {'uuid': uuid, 'name': body.get('name', 'test'), 'description': '', 'config': body.get('config', {'ai': {'default': True}})} + state[uuid] = data + state_path.write_text(json.dumps(state)) +else: + raise RuntimeError(args) +print(json.dumps({'ok': True, 'data': data})) +""" + + +def profile(): + return { + 'version': 1, + 'cli': {'config': 'lbctl.yaml'}, + 'cases': [{'name': 'chat', 'args': ['pipeline', 'run', 'existing'], 'expect': 'OK'}], + 'phases': [{'name': 'normal', 'duration_seconds': 1, 'rps': 2, 'workers': 4}], + 'observe': {'health': {'core': 'http://127.0.0.1:5300/healthz'}}, + } + + +class ProfileTests(unittest.TestCase): + def test_normal_profile_and_overrides_contract(self): + validate(json.loads(Path(__file__).with_name('regression_profiles').joinpath('normal.json').read_text())) + bad = profile() + for field, value in [('rps', float('nan')), ('workers', 33), ('duration_seconds', -1)]: + with self.subTest(field=field): + current = copy.deepcopy(bad) + current['phases'][0][field] = value + with self.assertRaises(ValueError): + validate(current) + + def test_missing_refs_and_existing_resource_overwrite_are_rejected(self): + current = profile() + current['cases'][0]['args'][-1] = '${missing}' + with self.assertRaises(KeyError): + validate(current) + current = profile() + current['resources'] = [{'id': 'pipeline', 'kind': 'pipeline', 'body': {'uuid': 'existing'}}] + with self.assertRaisesRegex(ValueError, 'uuid is forbidden'): + validate(current) + + def test_merge_keeps_defaults_but_replaces_lists(self): + base = {'ai': {'enabled': True, 'models': ['old']}, 'output': {'x': 1}} + actual = merge(base, {'ai': {'models': ['new']}}) + self.assertEqual(actual, {'ai': {'enabled': True, 'models': ['new']}, 'output': {'x': 1}}) + self.assertEqual(base['ai']['models'], ['old']) + self.assertEqual(resolve(['${id}', {'id': '${id}'}], {'id': 'created'}), ['created', {'id': 'created'}]) + + def test_ownership_recorded_before_configuration_failure(self): + with tempfile.TemporaryDirectory() as folder: + root = Path(folder) + current = profile() + current['resources'] = [{'id': 'chat', 'kind': 'pipeline', 'body': {'name': 'QA', 'config': {'ai': {}}}}] + suite = Suite(current, root / 'profile.json', root) + calls = [] + + def call(args, body=None): + calls.append((args, body)) + if args == ['whoami']: + return {'workspace_uuid': 'workspace-a'} + if args == ['pipeline', 'apply'] and 'uuid' not in body: + return {'uuid': 'owned-pipeline'} + if args[:2] == ['pipeline', 'get']: + return {'uuid': 'owned-pipeline', 'name': 'QA', 'description': '', 'config': {'ai': {'old': 1}}} + if args == ['pipeline', 'apply']: + raise RuntimeError('configuration failed') + return {} + + suite.call = call + with self.assertRaisesRegex(RuntimeError, 'configuration failed'): + suite.prepare() + self.assertEqual( + json.loads((root / 'owned.json').read_text())['resources'], + [{'kind': 'pipeline', 'uuid': 'owned-pipeline'}], + ) + self.assertEqual(suite.cleanup(), []) + self.assertEqual(calls[-1][0], ['pipeline', 'delete', 'owned-pipeline', '--yes']) + self.assertEqual(json.loads((root / 'owned.json').read_text())['resources'], []) + + def test_cleanup_refuses_another_workspace(self): + with tempfile.TemporaryDirectory() as folder: + root = Path(folder) + (root / 'owned.json').write_text(json.dumps({'workspace_uuid': 'other', 'resources': []})) + suite = Suite(profile(), root / 'profile.json', root) + suite.call = lambda args: {'workspace_uuid': 'current'} + with self.assertRaisesRegex(RuntimeError, 'Workspace differs'): + suite.cleanup() + + def test_aborted_run_is_not_treated_as_successful_cleanup(self): + with tempfile.TemporaryDirectory() as folder: + root = Path(folder) + config = root / 'profile.json' + config.write_text(json.dumps(profile())) + for error_type in (KeyboardInterrupt, TypeError): + output = root / error_type.__name__ + + def interrupt(suite): + suite.variables['workspace_uuid'] = 'test-workspace' + suite.journal() + raise error_type + + argv = ['regression_suite.py', 'run', '--profile', str(config), '--output', str(output)] + with ( + patch.object(sys, 'argv', argv), + patch.object(Suite, 'prepare', interrupt), + patch.object(Suite, 'cleanup') as cleanup, + redirect_stdout(io.StringIO()), + ): + with self.assertRaises(SystemExit if error_type is KeyboardInterrupt else TypeError): + main() + cleanup.assert_not_called() + self.assertFalse(json.loads((output / 'summary.json').read_text())['passed']) + + def test_subprocess_phases_and_cleanup_with_controlled_cli(self): + class Health(BaseHTTPRequestHandler): + def do_GET(self): + payload = b'{"code":0}' + self.send_response(200) + self.send_header('Content-Length', str(len(payload))) + self.end_headers() + self.wfile.write(payload) + + def log_message(self, *args): + pass + + server = ThreadingHTTPServer(('127.0.0.1', 0), Health) + thread = threading.Thread(target=server.serve_forever) + thread.start() + try: + with tempfile.TemporaryDirectory() as folder: + root = Path(folder) + cli = root / 'lbctl-test' + cli.write_text(f'#!{sys.executable}\n' + FAKE_CLI) + cli.chmod(0o700) + current = profile() + current['cli']['executable'] = str(cli) + current['resources'] = [ + {'id': 'provider', 'kind': 'provider', 'body': {'name': 'test'}}, + { + 'id': 'pipeline', + 'kind': 'pipeline', + 'body': {'name': 'test', 'config': {'ai': {'model': '${provider}'}}}, + }, + ] + current['cases'] = [ + {'name': 'chat', 'args': ['pipeline', 'run', '${pipeline}'], 'expect': 'TEST_OK', 'weight': 2}, + {'name': 'box', 'args': ['pipeline', 'run', '${pipeline}'], 'expect': 'TEST_OK', 'weight': 1}, + ] + current['phases'] = [ + {'name': name, 'duration_seconds': 0.5, 'rps': 12, 'workers': 3} for name in ('low', 'high') + ] + current['observe'] = {'health': {'core': f'http://127.0.0.1:{server.server_port}/healthz'}} + current['cleanup'] = 'always' + config = root / 'profile.json' + command = [ + sys.executable, + str(Path(__file__).with_name('regression_suite.py')), + 'run', + '--profile', + str(config), + ] + environment = {**os.environ, 'LBCTL_TEST_ROOT': folder} + for enforce_latency in (False, True): + current['thresholds'] = {'max_p95_seconds': {'chat': 0.001}} if enforce_latency else {} + config.write_text(json.dumps(current)) + output = root / ('failure' if enforce_latency else 'success') + result = subprocess.run( + [*command, '--output', str(output)], env=environment, capture_output=True, text=True, timeout=30 + ) + summary = json.loads((output / 'summary.json').read_text()) + self.assertEqual(result.returncode, int(enforce_latency), result.stdout + result.stderr) + self.assertEqual(summary['passed'], not enforce_latency) + self.assertEqual(len(summary['phases']), 2) + for phase in summary['phases']: + self.assertGreater(phase['achieved_rps'], 0) + self.assertEqual(phase['health_failures'], 0) + self.assertGreaterEqual(phase['cases']['chat']['requests'], phase['cases']['box']['requests']) + self.assertEqual(json.loads((output / 'owned.json').read_text())['resources'], []) + self.assertEqual(json.loads((root / 'cli-state.json').read_text()), {}) + finally: + server.shutdown() + server.server_close() + thread.join() + + +if __name__ == '__main__': + unittest.main()