diff --git a/tools/README.md b/tools/README.md index fb86e551f..1b3c00e82 100644 --- a/tools/README.md +++ b/tools/README.md @@ -401,3 +401,76 @@ as required status checks in the repository ruleset. A green workflow alone does not enforce merge protection. Longer load and soak tests use the existing Suite and Job entry points with separately calibrated latency, throughput and resource budgets; the one-minute smoke does not replace them. + +## Sustained regression and baseline comparison + +[`regression_profiles/endurance.json`](regression_profiles/endurance.json) runs +nine minutes of mixed chat/Box/RAG traffic at 1 request/second across four stable +groups, followed by one minute without traffic to check recovery. Configure its +CLI context, plugin inputs, model endpoints, Core SQLite path and Core/Runtime/Box +PID files as for smoke. Set `plugins: []` to reuse initialized plugins. Configure +`observe.box_data_root` to the target deployment's Box mount root; the caller must +have Docker inspection access. Do not start another Box runtime on a shared daemon. + +```sh +uv run --no-sync python tools/regression_suite.py validate \ + --profile /path/to/endurance.json +uv run --no-sync python tools/regression_suite.py run \ + --profile /path/to/endurance.json --output .qa/endurance-baseline +# For an hour including the one-minute cooldown, use --duration 3540. +``` + +This profile checks more than reply text: + +- Each Runner request is reconciled to one completed AgentRun with a persisted + completion event; Box additionally verifies two real Docker exec results. + After drain and cooldown, the complete set of runs for the phase's conversations + must match the verified set. Delayed extra, missing and nonterminal runs fail. +- At least 90% of target successful throughput must be delivered, with per-case + P95 limits. Reports retain planned/actual counts, scheduling delay and request + drain time. The load is bounded by workers and serial group sessions; it is + not an open arrival-rate capacity benchmark. +- Process RSS, FD, thread and process-count medians from the final 30 seconds of cooldown are compared + with a stable window after the first minute. Docker memory, PIDs and container + counts are checked separately because sandbox processes are not children of + the Box service. If a discovered container disappears before stats collection, + the sampler refreshes the set and retries once, retaining the first error and + membership change. Unchanged-set errors, failed recovery, missing samples and + budget excesses fail. + +The supplied limits are initial QA budgets, not product SLOs or proof of a leak. +Fixed groups prevent deliberate session growth from being mistaken for leakage; +Core and plugin caches can still retain memory legitimately. Inspect failures +before changing a budget. Process-tree CPU remains a raw diagnostic: child exits +can reduce the cumulative sum, so it is not used as a CPU regression verdict. + +`requests.jsonl` separates dispatch/response latency from ledger setup and result +verification. Resource samples go to `resources.jsonl`; `summary.json` contains +throughput, latency, recovery and ledger closeout results. Failed runs retain +resources and evidence. The normal Suite cleanup command removes only owned API +resources after inspection. + +To compare two successful runs from the same reserved test host: + +```sh +uv run --no-sync python tools/regression_analysis.py \ + --baseline .qa/endurance-baseline --candidate .qa/endurance-candidate \ + --output .qa/comparison.json +``` + +The comparison rejects failed runs, missing dependency evidence, different enabled +plugin versions, fixtures, expanded case/resource inputs, phase geometry, budgets, +or observer-host identity. Both runs must use identical load/analysis/smoke oracle +files and mock scenarios; their hashes are recorded in the contract. Target URL +hosts, ports and credentials may differ; URL schemes and +paths, model names and actual stimulus variables remain part of the contract. +Both runs must use the same hardware allocation, model backend and background +load; the contract does not establish those external conditions automatically. +The default comparison allows at most 20% P95 growth and 10% successful-throughput +loss, requires 30 requests per case in both runs, and rejects invalid metrics or +incomplete ledger closeout. Set tolerances from repeated controls on your host. + +Keep this longer test outside the one-minute merge gate while its budgets and +repeatability are being evaluated. The same profile can be selected by the +release-deployment Job, which supplies its ledger path and process IDs; that +fresh-deployment path still needs separate end-to-end validation. diff --git a/tools/regression_analysis.py b/tools/regression_analysis.py new file mode 100644 index 000000000..f643fa760 --- /dev/null +++ b/tools/regression_analysis.py @@ -0,0 +1,286 @@ +#!/usr/bin/env python3 +"""Compare compatible load runs and check resource recovery without treating missing data as zero.""" + +import argparse +import hashlib +import json +import math +import os +import platform +from pathlib import Path +import statistics +import re +from urllib.parse import urlsplit + + +METRICS = {'rss_bytes', 'open_fds', 'threads', 'processes'} + + +def validate_resource_limits(limits): + if ( + not isinstance(limits, dict) + or not {'warmup_seconds', 'window_seconds', 'min_samples', 'targets'} <= set(limits) + or limits.keys() - {'warmup_seconds', 'window_seconds', 'min_samples', 'targets', 'containers'} + ): + raise ValueError('resource_limits requires warmup_seconds, window_seconds, min_samples and targets') + for key in ('warmup_seconds', 'window_seconds', 'min_samples'): + value = limits[key] + if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value) or value <= 0: + raise ValueError(f'resource_limits.{key} must be positive and finite') + if not isinstance(limits['min_samples'], int) or limits['min_samples'] < 3: + raise ValueError('resource_limits.min_samples must be an integer >= 3') + if not isinstance(limits['targets'], dict) or not limits['targets']: + raise ValueError('resource_limits.targets must contain process targets') + for name, metrics in limits['targets'].items(): + if not name or not isinstance(metrics, dict) or not metrics or metrics.keys() - METRICS: + raise ValueError('resource limits support rss_bytes, open_fds, threads and processes') + if any( + isinstance(v, bool) or not isinstance(v, (float, int)) or not math.isfinite(v) or v < 0 + for v in metrics.values() + ): + raise ValueError('resource growth allowances must be finite and nonnegative') + + if 'containers' in limits: + values = limits['containers'] + if not isinstance(values, dict) or not values or values.keys() - {'memory_bytes', 'pids', 'count'}: + raise ValueError('container limits support memory_bytes, pids and count') + if any( + isinstance(v, bool) or not isinstance(v, (int, float)) or not math.isfinite(v) or v < 0 + for v in values.values() + ): + raise ValueError('container growth allowances must be finite and nonnegative') + + +def container_metrics(row): + entries = row.get('containers') + if not isinstance(entries, list) or row.get('container_error'): + return {} + memory = pids = 0 + units = {'B': 1, 'kB': 1000, 'KiB': 1024, 'MB': 1000**2, 'MiB': 1024**2, 'GB': 1000**3, 'GiB': 1024**3} + for item in entries: + match = re.fullmatch(r'([0-9.]+)\s*([A-Za-z]+)', item.get('MemUsage', '').split('/')[0].strip()) + try: + if match is None: + return {} + memory += float(match[1]) * units[match[2]] + pids += int(item['PIDs']) + except (ValueError, KeyError, TypeError): + return {} + return {'memory_bytes': memory, 'pids': pids, 'count': len(entries)} + + +def resource_verdict(directory, limits): + validate_resource_limits(limits) + rows = [json.loads(line) for line in (Path(directory) / 'resources.jsonl').read_text().splitlines()] + result = {'errors': [], 'targets': {}, 'limits': limits} + cooldown = [row for row in rows if row['phase'] == 'cooldown'] + if not cooldown: + result['errors'].append('No cooldown resource samples; recovery is unknown') + return result + end = max(row['elapsed'] for row in cooldown) + baseline = [ + row + for row in rows + if row['phase'] == 'load' + and limits['warmup_seconds'] <= row['elapsed'] < limits['warmup_seconds'] + limits['window_seconds'] + ] + tail = [row for row in cooldown if row['elapsed'] >= end - limits['window_seconds']] + if min(len(baseline), len(tail)) < limits['min_samples']: + result['errors'].append('Insufficient stable/cooldown samples; resource recovery is unknown') + return result + targets = dict(limits['targets']) + if limits.get('containers'): + targets['docker_containers'] = limits['containers'] + for name, metrics in targets.items(): + target = result['targets'][name] = {} + for metric, allowance in metrics.items(): + + def value(row): + source = container_metrics(row) if name == 'docker_containers' else row['targets'].get(name, {}) + return source.get(metric) + + before = [value(row) for row in baseline] + after = [value(row) for row in tail] + if any( + not isinstance(v, (int, float)) or not math.isfinite(v) or v < 0 or (metric == 'rss_bytes' and v == 0) + for v in before + after + ): + result['errors'].append(f'{name}/{metric}: missing or invalid samples') + continue + initial, final = statistics.median(before), statistics.median(after) + growth = final - initial + target[metric] = { + 'baseline_median': initial, + 'cooldown_median': final, + 'growth': growth, + 'allowance': allowance, + } + if growth > allowance: + result['errors'].append(f'{name}/{metric}: retained growth {growth:g} exceeds {allowance:g}') + return result + + +def workload_contract(profile, profile_path): + """Fingerprint stimuli and load geometry, excluding target addresses and generated UUIDs.""" + from regression_suite import resolve + + profile_path = Path(profile_path).resolve() + variables = { + **profile.get('vars', {}), + 'profile_dir': str(profile_path.parent), + **{item['id']: '${' + item['id'] + '}' for item in profile.get('resources', []) if 'id' in item}, + 'workspace_uuid': '${workspace_uuid}', + 'group_secret': '${group_secret}', + 'run_dir': '${run_dir}', + } + resources = resolve(profile.get('resources', []), variables) + cases = resolve(profile.get('cases', []), variables) + for resource in resources: + for document in resource.get('documents', []): + path = Path(document['file']) + path = path if path.is_absolute() else profile_path.parent / path + document['file'] = {'sha256': hashlib.sha256(path.read_bytes()).hexdigest()} + if resource.get('kind') == 'provider': + body = resource['body'] + if body.get('base_url'): + url = urlsplit(body['base_url']) + body['base_url'] = {'scheme': url.scheme, 'path': url.path, 'query': url.query} + if 'api_keys' in body: + body['api_keys'] = '' + for case in cases: + if 'http_group' in case: + url = urlsplit(case['http_group']['url']) + case['http_group']['url'] = {'scheme': url.scheme, 'path': url.path, 'query': url.query} + contract = {key: profile.get(key) for key in ('phases', 'thresholds', 'resource_limits')} + contract.update(resources=resources, cases=cases) + contract['observer_host'] = {'name': platform.node(), 'machine': platform.machine(), 'cpus': os.cpu_count()} + contract['oracle_sha256'] = { + name: hashlib.sha256(Path(__file__).with_name(name).read_bytes()).hexdigest() + for name in ('regression_traffic.py', 'regression_smoke.py', 'regression_analysis.py') + } + contract['timing_contract'] = 'dispatch-after-ledger-snapshot-v1' + contract['verified_runs'] = bool(profile.get('verification')) + contract['mock_scenarios_sha256'] = hashlib.sha256( + Path(__file__).with_name('mock_llm_scenarios.json').read_bytes() + ).hexdigest() + return contract + + +def compare(baseline, candidate, *, max_p95_ratio=1.2, min_rps_ratio=0.9, min_requests=30): + roots = [Path(baseline), Path(candidate)] + summaries = [json.loads((root / 'summary.json').read_text()) for root in roots] + contracts = [json.loads((root / 'workload.json').read_text()) for root in roots] + errors, comparisons = [], [] + plugins = [] + for root in roots: + environment = root / 'environment.json' + if not environment.exists(): + errors.append('Plugin dependency evidence is missing') + continue + records = json.loads(environment.read_text())['plugins']['plugins'] + plugins.append( + sorted( + (item['author'], item['name'], item['version']) + for item in records + if item.get('enabled') and item.get('status') == 'initialized' + ) + ) + if any(not value for value in plugins): + errors.append('Enabled plugin evidence is empty') + if len(plugins) == 2 and plugins[0] != plugins[1]: + errors.append('Enabled plugin versions differ; runs are not comparable') + if contracts[0] != contracts[1]: + errors.append('Workload/fixtures/oracles differ; runs are not comparable') + if not all(summary.get('passed') is True for summary in summaries): + errors.append('Both runs must pass functional, delivery and resource checks') + if not all(contract.get('verified_runs') for contract in contracts): + errors.append('Both runs require ledger verification') + if errors: + return {'passed': False, 'errors': errors, 'comparisons': comparisons} + old_phases, new_phases = summaries[0]['phases'], summaries[1]['phases'] + if not old_phases or not new_phases: + return {'passed': False, 'errors': ['Measured phases are missing'], 'comparisons': []} + if [phase['name'] for phase in old_phases] != [phase['name'] for phase in new_phases]: + return {'passed': False, 'errors': ['Phase sets differ'], 'comparisons': []} + for old, new in zip(old_phases, new_phases): + if any( + phase.get('verification_enabled') is not True + or phase.get('verified_runs') != phase.get('runner_requests') + or phase.get('ledger_closeout', {}).get('matched') is not True + for phase in (old, new) + ): + errors.append(f'{new["name"]}: runner reconciliation is incomplete') + continue + if any( + not isinstance(phase.get('achieved_rps'), (int, float)) + or not math.isfinite(phase['achieved_rps']) + or phase['achieved_rps'] <= 0 + for phase in (old, new) + ): + errors.append(f'{new["name"]}: successful throughput is invalid') + continue + if not old['cases'] or not new['cases'] or old['cases'].keys() != new['cases'].keys(): + errors.append(f'{new["name"]}: case sets differ') + continue + if new['achieved_rps'] < old['achieved_rps'] * min_rps_ratio: + errors.append(f'{new["name"]}: successful throughput regressed') + for name, current in new['cases'].items(): + previous = old['cases'][name] + values = [previous.get('p95_seconds'), current.get('p95_seconds')] + if min(previous['requests'], current['requests']) < min_requests or any( + not isinstance(v, (int, float)) or not math.isfinite(v) or v <= 0 for v in values + ): + errors.append(f'{new["name"]}/{name}: insufficient samples or invalid P95') + continue + ratio = current['p95_seconds'] / previous['p95_seconds'] + comparisons.append( + { + 'phase': new['name'], + 'case': name, + 'p95_ratio': ratio, + 'baseline_requests': previous['requests'], + 'candidate_requests': current['requests'], + } + ) + if ratio > max_p95_ratio: + errors.append(f'{new["name"]}/{name}: P95 regressed by {(ratio - 1) * 100:.1f}%') + return { + 'passed': not errors, + 'errors': errors, + 'comparisons': comparisons, + 'max_p95_ratio': max_p95_ratio, + 'min_rps_ratio': min_rps_ratio, + 'min_requests': min_requests, + } + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--baseline', type=Path, required=True) + parser.add_argument('--candidate', type=Path, required=True) + parser.add_argument('--output', type=Path, required=True) + parser.add_argument('--max-p95-ratio', type=float, default=1.2) + parser.add_argument('--min-rps-ratio', type=float, default=0.9) + parser.add_argument('--min-requests', type=int, default=30) + args = parser.parse_args() + if ( + not math.isfinite(args.max_p95_ratio) + or args.max_p95_ratio < 1 + or not 0 < args.min_rps_ratio <= 1 + or args.min_requests < 1 + ): + parser.error('require finite max-p95-ratio >= 1, min-rps-ratio in (0,1], min-requests >= 1') + result = compare( + args.baseline, + args.candidate, + max_p95_ratio=args.max_p95_ratio, + min_rps_ratio=args.min_rps_ratio, + min_requests=args.min_requests, + ) + args.output.write_text(json.dumps(result, indent=2) + '\n') + print(json.dumps(result, indent=2)) + raise SystemExit(0 if result['passed'] else 1) + + +if __name__ == '__main__': + main() diff --git a/tools/regression_job.py b/tools/regression_job.py index 099c85ab7..cd43bcb9e 100644 --- a/tools/regression_job.py +++ b/tools/regression_job.py @@ -308,16 +308,27 @@ class Job: if 'mock_records' in profile['smoke']: profile['smoke']['mock_records'] = str(self.output / 'mock' / 'requests.jsonl') profile['smoke']['ledger_db'] = str(Path(self.state['record']['data_dir']) / 'langbot.db') + if 'verification' in profile: + profile['verification']['ledger_db'] = str(Path(self.state['record']['data_dir']) / 'langbot.db') 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 + name = ( + {'langbot': 'core', 'langbot_plugin_runtime': 'runtime', 'langbot_box': 'box'}.get(service, service) + if profile.get('verification') + else service + ) + profile['observe']['pids'][name] = pid destination = self.output / 'suite' destination.mkdir(mode=0o700) suite = Suite(profile, self.suite_path, destination, env=self.env) self.redactions.extend(suite.secrets) suite.write('profile.json', profile) + if 'smoke' not in profile: + from regression_analysis import workload_contract + + suite.write('workload.json', workload_contract(profile, self.suite_path)) suite.prepare() if 'smoke' in profile: return suite diff --git a/tools/regression_profiles/endurance.json b/tools/regression_profiles/endurance.json new file mode 100644 index 000000000..25480525a --- /dev/null +++ b/tools/regression_profiles/endurance.json @@ -0,0 +1,513 @@ +{ + "version": 1, + "cli": { + "executable": "lbctl", + "config": "./lbctl.yaml" + }, + "vars": { + "plugin_dir": "/path/to/plugin-zips", + "mock_base_url": "http://127.0.0.1:18080/v1", + "embedding_base_url": "http://127.0.0.1:15410/v1", + "core_health": "http://127.0.0.1:5300/healthz", + "core_base_url": "http://127.0.0.1:5300", + "core_db": "/path/to/core/data/langbot.db", + "core_pid_file": "/path/to/core.pid", + "runtime_pid_file": "/path/to/runtime.pid", + "box_pid_file": "/path/to/box.pid", + "box_data_root": "/path/to/core/box" + }, + "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-smoke-box", + "provider_uuid": "${mock_provider}", + "abilities": [ + "func_call" + ], + "context_length": 8192, + "extra_args": { + "request_policy": { + "retries": 0 + } + } + } + }, + { + "id": "rag_model", + "kind": "model", + "type": "llm", + "body": { + "name": "mock-rag", + "provider_uuid": "${mock_provider}", + "abilities": [ + "func_call" + ], + "context_length": 8192, + "extra_args": { + "request_policy": { + "retries": 0 + } + } + } + }, + { + "id": "embedding_model", + "kind": "model", + "type": "embedding", + "body": { + "name": "all-MiniLM-L6-v2", + "provider_uuid": "${embedding_provider}" + } + }, + { + "id": "knowledge_base", + "kind": "knowledge-base", + "body": { + "name": "regression parser golden", + "knowledge_engine_plugin_id": "langbot-team/LangRAG", + "creation_settings": { + "embedding_model_uuid": "${embedding_model}", + "index_type": "chunk", + "chunk_size": 512, + "overlap": 50 + }, + "retrieval_settings": { + "search_type": "vector", + "top_k": 5, + "query_rewrite": "off", + "rerank": "off" + } + }, + "documents": [ + { + "file": "../../skills/skills/langbot-testing/fixtures/rag/parser-golden.html", + "parser_plugin_id": "langbot-team/GeneralParsers" + } + ] + }, + { + "id": "chat_pipeline", + "kind": "pipeline", + "body": { + "name": "regression chat", + "description": "Normal deterministic chat", + "config": { + "ai": { + "runner": { + "id": "plugin:langbot-team/LocalAgent/default", + "expire-time": 0 + }, + "runner_config": { + "plugin:langbot-team/LocalAgent/default": { + "model": { + "primary": "${chat_model}", + "fallbacks": [] + }, + "prompt": [ + { + "role": "system", + "content": "You are a QA regression assistant." + } + ], + "knowledge-bases": [], + "box-enabled": false, + "timeout": 30, + "max-round": 20, + "box-session-id-template": "{launcher_type}_{launcher_id}" + } + } + }, + "safety": { + "rate-limit": { + "limitation": 10000 + } + }, + "output": { + "force-delay": { + "min": 0, + "max": 0 + }, + "misc": { + "at-sender": false, + "quote-origin": false + } + }, + "trigger": { + "group-respond-rules": { + "prefix": [ + "ai " + ] + } + } + } + } + }, + { + "id": "box_pipeline", + "kind": "pipeline", + "body": { + "name": "regression Box", + "description": "Real Box write and read", + "config": { + "ai": { + "runner": { + "id": "plugin:langbot-team/LocalAgent/default", + "expire-time": 0 + }, + "runner_config": { + "plugin:langbot-team/LocalAgent/default": { + "model": { + "primary": "${box_model}", + "fallbacks": [] + }, + "prompt": [ + { + "role": "system", + "content": "You are a QA regression assistant." + } + ], + "knowledge-bases": [], + "box-enabled": true, + "box-session-id-template": "{launcher_type}_{launcher_id}", + "timeout": 30, + "max-tool-iterations": 5, + "max-round": 20 + } + } + }, + "safety": { + "rate-limit": { + "limitation": 10000 + } + }, + "output": { + "force-delay": { + "min": 0, + "max": 0 + }, + "misc": { + "at-sender": false, + "quote-origin": false + } + }, + "trigger": { + "group-respond-rules": { + "prefix": [ + "ai " + ] + } + } + } + } + }, + { + "id": "rag_pipeline", + "kind": "pipeline", + "body": { + "name": "regression RAG", + "description": "Retrieved model input oracle", + "config": { + "ai": { + "runner": { + "id": "plugin:langbot-team/LocalAgent/default", + "expire-time": 0 + }, + "runner_config": { + "plugin:langbot-team/LocalAgent/default": { + "model": { + "primary": "${rag_model}", + "fallbacks": [] + }, + "prompt": [ + { + "role": "system", + "content": "You are a QA regression assistant." + } + ], + "knowledge-bases": [ + "${knowledge_base}" + ], + "box-enabled": false, + "timeout": 30, + "retrieval-top-k": 5, + "max-round": 20, + "box-session-id-template": "{launcher_type}_{launcher_id}" + } + } + }, + "safety": { + "rate-limit": { + "limitation": 10000 + } + }, + "output": { + "force-delay": { + "min": 0, + "max": 0 + }, + "misc": { + "at-sender": false, + "quote-origin": false + } + }, + "trigger": { + "group-respond-rules": { + "prefix": [ + "ai " + ] + } + } + } + } + }, + { + "id": "chat_bot", + "kind": "bot", + "body": { + "name": "regression group chat", + "description": "Synthetic signed group ingress", + "adapter": "http_bot", + "enable": true, + "adapter_config": { + "inbound_secret": "${group_secret}", + "signature_required": true, + "callback_url": "http://127.0.0.1:9/unused-sync-only", + "default_session_type": "group", + "callback_timeout": 20, + "callback_max_retries": 0 + }, + "event_bindings": [ + { + "event_pattern": "message.received", + "target_type": "pipeline", + "target_uuid": "${chat_pipeline}", + "enabled": true, + "filters": [] + } + ], + "plugin_processors": [] + } + }, + { + "id": "box_bot", + "kind": "bot", + "body": { + "name": "regression group box", + "description": "Synthetic signed group ingress", + "adapter": "http_bot", + "enable": true, + "adapter_config": { + "inbound_secret": "${group_secret}", + "signature_required": true, + "callback_url": "http://127.0.0.1:9/unused-sync-only", + "default_session_type": "group", + "callback_timeout": 20, + "callback_max_retries": 0 + }, + "event_bindings": [ + { + "event_pattern": "message.received", + "target_type": "pipeline", + "target_uuid": "${box_pipeline}", + "enabled": true, + "filters": [] + } + ], + "plugin_processors": [] + } + }, + { + "id": "rag_bot", + "kind": "bot", + "body": { + "name": "regression group rag", + "description": "Synthetic signed group ingress", + "adapter": "http_bot", + "enable": true, + "adapter_config": { + "inbound_secret": "${group_secret}", + "signature_required": true, + "callback_url": "http://127.0.0.1:9/unused-sync-only", + "default_session_type": "group", + "callback_timeout": 20, + "callback_max_retries": 0 + }, + "event_bindings": [ + { + "event_pattern": "message.received", + "target_type": "pipeline", + "target_uuid": "${rag_pipeline}", + "enabled": true, + "filters": [] + } + ], + "plugin_processors": [] + } + } + ], + "cases": [ + { + "name": "chat", + "weight": 2, + "http_group": { + "url": "${core_base_url}/bots/${chat_bot}/sync", + "secret_env": "LANGBOT_QA_GROUP_SECRET", + "message": "ai What is the documented parser-rag sentinel?" + }, + "expect": "LANGBOT_CHAT_OK" + }, + { + "name": "box", + "weight": 1, + "http_group": { + "url": "${core_base_url}/bots/${box_bot}/sync", + "secret_env": "LANGBOT_QA_GROUP_SECRET", + "message": "ai What is the documented parser-rag sentinel?" + }, + "expect": "LANGBOT_BOX_OK", + "box_exec": { + "count": 2, + "backend": "docker", + "stdout": "LANGBOT_BOX_SENTINEL" + } + }, + { + "name": "rag_chat", + "weight": 1, + "http_group": { + "url": "${core_base_url}/bots/${rag_bot}/sync", + "secret_env": "LANGBOT_QA_GROUP_SECRET", + "message": "ai What is the documented parser-rag sentinel?" + }, + "expect": "LANGBOT_RAG_OK" + } + ], + "phases": [ + { + "name": "sustained", + "duration_seconds": 540, + "rps": 1, + "workers": 4, + "cooldown_seconds": 60, + "groups": 4, + "members_per_group": 3 + } + ], + "observe": { + "health": { + "core": "${core_health}", + "mock": "http://127.0.0.1:18080/healthz", + "box": "http://127.0.0.1:5410/readyz" + }, + "pids": {}, + "pid_files": { + "core": "${core_pid_file}", + "runtime": "${runtime_pid_file}", + "box": "${box_pid_file}" + }, + "containers": [], + "box_data_root": "${box_data_root}" + }, + "cleanup": "on-success", + "thresholds": { + "min_rps_ratio": 0.9, + "max_p95_seconds": { + "chat": 5, + "box": 10, + "rag_chat": 6 + } + }, + "verification": { + "ledger_db": "${core_db}", + "timeout_seconds": 5 + }, + "resource_limits": { + "warmup_seconds": 60, + "window_seconds": 30, + "min_samples": 5, + "targets": { + "core": { + "rss_bytes": 67108864, + "open_fds": 8, + "threads": 8, + "processes": 0 + }, + "runtime": { + "rss_bytes": 67108864, + "open_fds": 8, + "threads": 8, + "processes": 0 + }, + "box": { + "rss_bytes": 33554432, + "open_fds": 8, + "threads": 8, + "processes": 0 + } + }, + "containers": { + "memory_bytes": 67108864, + "pids": 8, + "count": 0 + } + } +} diff --git a/tools/regression_suite.py b/tools/regression_suite.py index 4397d3905..ee6ea0da0 100644 --- a/tools/regression_suite.py +++ b/tools/regression_suite.py @@ -78,6 +78,8 @@ def validate(profile): 'cleanup', 'thresholds', 'smoke', + 'verification', + 'resource_limits', } if extra := profile.keys() - allowed: raise ValueError(f'unknown profile fields: {sorted(extra)}') @@ -138,6 +140,18 @@ def validate(profile): 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) + verification = profile.get('verification') + if verification is not None: + if ( + 'smoke' in profile + or not isinstance(verification, dict) + or verification.keys() - {'ledger_db', 'timeout_seconds'} + ): + raise ValueError('verification is a load-only ledger configuration') + if not isinstance(verification.get('ledger_db'), str) or not verification['ledger_db']: + raise ValueError('verification.ledger_db is required') + resolve(verification, variables) + number(verification.get('timeout_seconds', 5), 'verification timeout', minimum=0.001) cases = profile.get('cases') if not isinstance(cases, list) or not cases: raise ValueError('cases must be a non-empty array') @@ -164,7 +178,7 @@ def validate(profile): names.add(case['name']) if 'stage' in case and (not isinstance(case['stage'], str) or not case['stage']): raise ValueError('case stage must be a non-empty string') - if any(key in case for key in ('mock_requests', 'fields', 'forbid', 'reject_group')) and 'smoke' not in profile: + if any(key in case for key in ('mock_requests', 'reject_group')) and 'smoke' not in profile: raise ValueError('extended case checks require smoke mode') if 'fields' in case and (not isinstance(case['fields'], dict) or not case['fields']): raise ValueError('fields must be a non-empty mapping of dotted paths to exact values') @@ -237,14 +251,14 @@ def validate(profile): number(check['count'], 'box_exec.count', minimum=1, integer=True) if any(not isinstance(check[k], str) or not check[k] for k in ('backend', 'stdout')): raise ValueError('box_exec backend/stdout must be non-empty strings') - if 'smoke' not in profile: - raise ValueError('box_exec checks require smoke mode') + if 'smoke' not in profile and verification is None: + raise ValueError('box_exec requires smoke or load verification') if 'http_group' not in case and case.get('args', [])[:2] != ['pipeline', 'run']: raise ValueError('box_exec requires a Pipeline or group case') if 'attachment' in case: check = case['attachment'] if ( - 'smoke' not in profile + ('smoke' not in profile and verification is None) or case.get('args', [])[:2] != ['pipeline', 'run'] or not isinstance(check, dict) or set(check) != {'name', 'text'} @@ -329,6 +343,12 @@ def validate(profile): 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) + if 'resource_limits' in profile: + if 'smoke' in profile: + raise ValueError('resource_limits requires measured load and cooldown phases') + from regression_analysis import validate_resource_limits + + validate_resource_limits(profile['resource_limits']) observed = resolve(profile.get('observe', {}), variables) if not isinstance(observed, dict) or observed.keys() - { 'health', @@ -352,6 +372,21 @@ def validate(profile): raise ValueError(f'observe.{field} must be an object') for pid in observed.get('pids', {}).values(): number(pid, 'PID', minimum=1, integer=True) + if profile.get('resource_limits'): + if profile['resource_limits'].get('containers') and not ( + observed.get('containers') or observed.get('box_data_root') + ): + raise ValueError('Container budgets require container names or a Box data root') + configured = set(observed.get('pids', {})) | set(observed.get('pid_files', {})) + if set(profile['resource_limits']['targets']) - configured: + raise ValueError('Resource limit targets must have configured PIDs or PID files') + for phase in phases: + limits = profile['resource_limits'] + if ( + phase['duration_seconds'] < limits['warmup_seconds'] + limits['window_seconds'] + or phase.get('cooldown_seconds', 0) < limits['window_seconds'] + ): + raise ValueError('Phase must cover the configured warmup, resource window and cooldown') if not isinstance(observed.get('containers', []), list) or any( not isinstance(item, str) or not item for item in observed.get('containers', []) ): @@ -456,14 +491,15 @@ class Suite: self.write('owned.json', {'workspace_uuid': self.variables['workspace_uuid'], 'resources': self.owned}) def prepare(self): - if 'smoke' in self.profile and not Path(self.path(self.profile['smoke']['ledger_db'])).is_file(): - raise ValueError('smoke.ledger_db must be an existing target Core SQLite database') + verification = self.profile.get('smoke') or self.profile.get('verification') + if verification and not Path(self.path(verification['ledger_db'])).is_file(): + raise ValueError('ledger_db must be an existing target Core SQLite database') 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']) - if 'smoke' in self.profile: + if 'smoke' in self.profile or 'verification' in self.profile: capabilities = self.call(['capabilities']) plugins = self.call(['plugin', 'list']) ready = { @@ -471,7 +507,7 @@ class Suite: for item in plugins['plugins'] if item.get('enabled') and item.get('status') == 'initialized' } - missing = set(self.profile['smoke'].get('required_plugins', [])) - ready + missing = set(self.profile.get('smoke', {}).get('required_plugins', [])) - ready self.write('environment.json', {'identity': identity, 'capabilities': capabilities, 'plugins': plugins}) if missing: raise RuntimeError(f'Required plugins are not initialized: {sorted(missing)}') @@ -557,7 +593,7 @@ class Suite: else: saved['resources'].remove(resource) self.write('owned.json', saved) - if 'smoke' in self.profile: + if 'smoke' in self.profile or 'verification' in self.profile: from regression_smoke import verify_cleanup try: @@ -577,7 +613,22 @@ class Suite: 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', 'http_group') if key in case}] + [ + { + key: case[key] + for key in ( + 'name', + 'args', + 'expect', + 'http_group', + 'box_exec', + 'attachment', + 'fields', + 'forbid', + ) + if key in case + } + ] * case.get('weight', 1) ) cases_path = self.write('cases.json', cases) @@ -593,6 +644,15 @@ class Suite: observe_args += ['--container', container] if observed.get('box_data_root'): observe_args += ['--box-data-root', self.path(observed['box_data_root'])] + if verification := self.profile.get('verification'): + observe_args += [ + '--ledger-db', + self.path(verification['ledger_db']), + '--workspace', + self.variables['workspace_uuid'], + '--verification-timeout', + str(verification.get('timeout_seconds', 5)), + ] results = [] errors = [] for phase in self.profile['phases']: @@ -629,8 +689,18 @@ class Suite: 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'] + summary['achieved_rps'] = ( + summary.get( + 'successful_requests', sum(row.get('status', {}).get('ok', 0) for row in summary['cases'].values()) + ) + / summary['load_elapsed_seconds'] + ) results.append(summary) + if self.profile.get('resource_limits'): + from regression_analysis import resource_verdict + + summary['resources'] = resource_verdict(destination, self.profile['resource_limits']) + errors.extend(f'{phase["name"]}: {error}' for error in summary['resources']['errors']) if result.returncode: errors.append(f'{phase["name"]}: business or health/resource checks failed') thresholds = self.profile.get('thresholds', {}) @@ -684,6 +754,10 @@ def main(): return args.output.mkdir(parents=True, exist_ok=False, mode=0o700) suite.write('profile.json', profile) + if 'smoke' not in profile: + from regression_analysis import workload_contract + + suite.write('workload.json', workload_contract(profile, args.profile)) result = {'phases': [], 'errors': []} completed = False try: diff --git a/tools/regression_traffic.py b/tools/regression_traffic.py index c147190cf..cbd0c5bc9 100644 --- a/tools/regression_traffic.py +++ b/tools/regression_traffic.py @@ -11,6 +11,7 @@ import math import os from pathlib import Path import subprocess +import sqlite3 import sys import threading import time @@ -50,6 +51,41 @@ def owned_box_containers(data_root, checked=None): return owned +def sample_containers(configured, data_root, run=subprocess.run, discover=owned_box_containers): + def selected(): + return set(configured) | (set(discover(data_root)) if data_root else set()) + + containers = selected() + if not containers: + return {'containers': []} + + def stats(names): + return run( + ['docker', 'stats', '--no-stream', '--format', '{{json .}}', *sorted(names)], + capture_output=True, + text=True, + timeout=10, + ) + + result = stats(containers) + observation = {} + if result.returncode and data_root: + current = selected() + if containers - current: + observation['container_churn'] = { + 'removed': sorted(containers - current), + 'added': sorted(current - containers), + 'first_error': result.stderr.strip(), + } + if not current: + return {**observation, 'containers': []} + result = stats(current) + observation['containers'] = [json.loads(line) for line in result.stdout.splitlines() if line] + if result.returncode: + observation['container_error'] = result.stderr.strip() + return observation + + def group_request(case, group_id, member_id, turn, token, previous, env=None): """Send one signed GroupMessage through the existing HTTP Bot adapter.""" settings = case['http_group'] @@ -94,6 +130,34 @@ def group_request(case, group_id, member_id, turn, token, previous, env=None): return payload +def reconcile_runs(path, workspace, after_id, conversations, expected_ids): + """Catch delayed extra or nonterminal runs after request-level verification.""" + actual, unfinished = set(), [] + connection = sqlite3.connect(Path(path).resolve().as_uri() + '?mode=ro', uri=True) + try: + values = sorted(conversations) + for start in range(0, len(values), 500): + part = values[start : start + 500] + query = ( + 'SELECT run_id, status FROM agent_run WHERE workspace_id=? AND id>? ' + 'AND conversation_id IN (' + ','.join('?' for _ in part) + ')' + ) + for run_id, status in connection.execute(query, [workspace, after_id, *part]): + actual.add(run_id) + if status != 'completed': + unfinished.append(run_id) + finally: + connection.close() + return { + 'matched': actual == set(expected_ids) and not unfinished, + 'expected_count': len(expected_ids), + 'actual_count': len(actual), + 'extra': sorted(actual - set(expected_ids)), + 'missing': sorted(set(expected_ids) - actual), + 'not_completed': unfinished, + } + + def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--lbctl', required=True) @@ -112,6 +176,9 @@ def main(): parser.add_argument( '--group-state', type=Path, help='Carry group histories across load phases; refuses interrupted runs' ) + parser.add_argument('--ledger-db', type=Path) + parser.add_argument('--workspace') + parser.add_argument('--verification-timeout', type=float, default=5) 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') @@ -161,16 +228,28 @@ def main(): or not os.environ.get(settings['secret_env']) ): parser.error('group cases need HTTP(S) url, message and a populated secret_env') + if args.ledger_db and (not args.ledger_db.is_file() or not args.workspace): + parser.error('ledger verification requires an existing SQLite file and workspace') + if not math.isfinite(args.verification_timeout) or args.verification_timeout <= 0: + parser.error('verification timeout must be positive') + from regression_smoke import UnresolvedRun, check_run, check_attachment, check_fields, completed_run + 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 = [] + verified_ids = set() + conversations = set() sampling_errors = [] lock = threading.Lock() group_condition = threading.Condition(lock) stop = threading.Event() phase = ['load'] next_case = 0 + phase_start_id = 0 + if args.ledger_db: + with sqlite3.connect(args.ledger_db.resolve().as_uri() + '?mode=ro', uri=True) as db: + phase_start_id = db.execute('SELECT COALESCE(MAX(id),0) FROM agent_run').fetchone()[0] started = time.monotonic() deadline = started + args.duration redactions = [value for name, value in os.environ.items() if 'API_KEY' in name and value] @@ -215,6 +294,52 @@ def main(): text = text.replace(secret, '[REDACTED]') return text + def open_ledger(): + if not args.ledger_db: + return None, None + connection = sqlite3.connect(args.ledger_db.resolve().as_uri() + '?mode=ro', uri=True, timeout=5) + connection.row_factory = sqlite3.Row + connection.execute('PRAGMA query_only=ON') + after_id = connection.execute('SELECT COALESCE(MAX(id),0) FROM agent_run').fetchone()[0] + return connection, after_id + + def verify(row, case, connection, after_id, is_group=False): + began = time.monotonic() + data = row['output']['data'] + checks = {} + if case.get('fields'): + checks['fields'] = check_fields(data, case['fields']) + if any(value in json.dumps(data, ensure_ascii=False) for value in case.get('forbid', [])): + raise ValueError('Response contains forbidden content') + runner = is_group or case.get('args', [])[:2] == ['pipeline', 'run'] + row['requires_runner'] = runner + if connection and runner: + if not isinstance(data, dict): + raise ValueError('Runner response data must be an object') + session = data.get('session_id') + if not isinstance(session, str) or not session: + raise UnresolvedRun('Response has no conversation identity') + conversation = 'group_' + session if is_group else session + with lock: + conversations.add(conversation) + run = completed_run( + connection, + args.workspace, + conversation, + after_id, + args.verification_timeout, + ) + row['run'] = run + checks.update(check_run(connection, run, case)) + if case.get('attachment'): + checks['attachment'] = check_attachment(data, case['attachment']) + with lock: + if run['run_id'] in verified_ids: + raise ValueError('Two requests matched the same AgentRun') + verified_ids.add(run['run_id']) + row['checks'] = checks + row['verification_seconds'] = time.monotonic() - began + def sample(): with (args.output / 'resources.jsonl').open('w') as stream: while not stop.is_set(): @@ -235,19 +360,8 @@ def main(): row['health'][name] = read_endpoint_snapshot(url, timeout_seconds=3) except RuntimeError as exc: row['health'][name] = {'error': str(exc)} - containers = set(args.container) - if args.box_data_root: - containers.update(owned_box_containers(args.box_data_root)) - if containers: - result = subprocess.run( - ['docker', 'stats', '--no-stream', '--format', '{{json .}}', *sorted(containers)], - 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() + if args.container or args.box_data_root: + row.update(sample_containers(args.container, args.box_data_root)) stream.write(json.dumps(row) + '\n') stream.flush() with lock: @@ -274,17 +388,37 @@ def main(): else: case = cases[(iteration * args.workers + index) % len(cases)] before = time.monotonic() - row = {'case': case['name'], 'time': time.time(), 'elapsed': before - started, 'status': 'failed'} + row = { + 'case': case['name'], + 'time': time.time(), + 'elapsed': before - started, + 'status': 'failed', + 'schedule_lag_seconds': max(0, before - due), + } + connection = None try: + connection, after_id = open_ledger() + dispatch = time.monotonic() + row.update( + ledger_setup_seconds=dispatch - before, + elapsed=dispatch - started, + schedule_lag_seconds=max(0, dispatch - due), + ) result = subprocess.run( [args.lbctl, '--config', args.config, '-o', 'json', *case['args']], capture_output=True, text=True, timeout=90, ) + row['seconds'] = time.monotonic() - dispatch 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): + if not isinstance(row['output'], dict): + raise ValueError('CLI response must be an object') + if result.returncode == 0 and case['expect'] in json.dumps( + row['output'].get('data'), ensure_ascii=False + ): + verify(row, case, connection, after_id) row['status'] = 'ok' elif 'result_unknown' in result.stdout: row['status'] = 'unknown' @@ -292,9 +426,14 @@ def main(): 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 + except UnresolvedRun as exc: + row.update(status='unknown', error=str(exc)) + except (OSError, ValueError, KeyError, sqlite3.Error) as exc: + row['error'] = redact(str(exc)) + finally: + if connection: + connection.close() + row.setdefault('seconds', time.monotonic() - before) with lock: rows.append(row) with (args.output / 'requests.jsonl').open('a') as stream: @@ -333,9 +472,20 @@ def main(): 'time': time.time(), 'elapsed': before - started, 'status': 'failed', + 'schedule_lag_seconds': max(0, before - group['due']), } + connection = None try: + connection, after_id = open_ledger() + dispatch = time.monotonic() + row.update( + ledger_setup_seconds=dispatch - before, + elapsed=dispatch - started, + schedule_lag_seconds=max(0, dispatch - group['due']), + ) row['output'] = group_request(case, group['id'], member, group['turn'], token, group['previous']) + row['seconds'] = time.monotonic() - dispatch + verify(row, case, connection, after_id, is_group=True) row['status'] = 'ok' except urllib.error.HTTPError as exc: row.update(http_status=exc.code, error=redact(exc.read(2000).decode(errors='replace'))) @@ -343,9 +493,14 @@ def main(): row['status'] = 'unknown' except (urllib.error.URLError, TimeoutError, OSError) as exc: row.update(status='unknown', error=redact(str(exc)) + '; not retried') - except (ValueError, KeyError) as exc: + except UnresolvedRun as exc: + row.update(status='unknown', error=str(exc)) + except (ValueError, KeyError, sqlite3.Error) as exc: row['error'] = redact(str(exc)) - row['seconds'] = time.monotonic() - before + finally: + if connection: + connection.close() + row.setdefault('seconds', time.monotonic() - before) with group_condition: rows.append(row) with (args.output / 'requests.jsonl').open('a') as stream: @@ -382,9 +537,23 @@ def main(): 'workers': args.workers, 'cooldown_seconds': args.cooldown, 'requests': len(rows), + 'successful_requests': sum(row['status'] == 'ok' for row in rows), + 'verified_runs': len(verified_ids), + 'runner_requests': sum(row.get('requires_runner', False) for row in rows), + 'verification_enabled': bool(args.ledger_db), + 'planned_requests': math.ceil(args.duration * args.rps), + 'delivery_ratio': len(rows) / math.ceil(args.duration * args.rps), + 'drain_seconds': max(0, load_elapsed - args.duration), + 'schedule_lag_p95_seconds': percentile([row['schedule_lag_seconds'] for row in rows], 0.95), + 'load_model': 'bounded workers; serialized groups; dispatch latency excludes verification', 'status': dict(Counter(row['status'] for row in rows)), 'cases': {}, } + if args.ledger_db: + conversations.update('group_' + group['id'] for group in groups) + summary['ledger_closeout'] = reconcile_runs( + args.ledger_db, args.workspace, phase_start_id, conversations, verified_ids + ) samples = [json.loads(line) for line in (args.output / 'resources.jsonl').read_text().splitlines()] summary['health_samples'] = len(samples) summary['health_failures'] = sum( @@ -441,6 +610,7 @@ def main(): or summary['health_failures'] or sampling_errors or summary.get('group_failures') + or (args.ledger_db and not summary['ledger_closeout']['matched']) ): raise SystemExit(1) diff --git a/tools/test_regression_analysis.py b/tools/test_regression_analysis.py new file mode 100644 index 000000000..f74708753 --- /dev/null +++ b/tools/test_regression_analysis.py @@ -0,0 +1,175 @@ +"""Counterexamples for load resource recovery and compatible-run comparisons.""" + +import copy +import json +from pathlib import Path +import tempfile +import unittest + +from regression_analysis import compare, container_metrics, resource_verdict, workload_contract + + +class AnalysisTests(unittest.TestCase): + def setUp(self): + temporary = tempfile.TemporaryDirectory() + self.addCleanup(temporary.cleanup) + self.root = Path(temporary.name) + self.limits = { + 'warmup_seconds': 10, + 'window_seconds': 10, + 'min_samples': 3, + 'targets': {'core': {'rss_bytes': 100, 'open_fds': 2}}, + } + + def resources(self, tail_fds=10, omit=False): + samples = [ + {'phase': phase, 'elapsed': second, 'targets': {'core': {'rss_bytes': 1000, 'open_fds': fds}}} + for phase, seconds, fds in [('load', range(10, 20), 10), ('cooldown', range(30, 40), tail_fds)] + for second in seconds + ] + if omit: + samples = [row for row in samples if row['phase'] == 'load'] + (self.root / 'resources.jsonl').write_text(''.join(json.dumps(row) + '\n' for row in samples)) + + def test_healthy_responses_cannot_hide_retained_fds(self): + self.resources() + self.assertEqual(resource_verdict(self.root, self.limits)['errors'], []) + self.resources(tail_fds=40) + self.assertIn('retained growth', resource_verdict(self.root, self.limits)['errors'][0]) + + def test_missing_cooldown_is_unknown_not_zero_growth(self): + self.resources(omit=True) + self.assertIn('unknown', resource_verdict(self.root, self.limits)['errors'][0]) + self.resources() + data = (self.root / 'resources.jsonl').read_text().replace('"core"', '"other"') + (self.root / 'resources.jsonl').write_text(data) + self.assertTrue(resource_verdict(self.root, self.limits)['errors']) + + def test_docker_memory_is_independent_of_service_process_rss(self): + row = {'containers': [{'MemUsage': '10MiB / 1GiB', 'PIDs': '2'}, {'MemUsage': '5MiB / 1GiB', 'PIDs': '3'}]} + self.assertEqual(container_metrics(row), {'memory_bytes': 15 * 1024**2, 'pids': 5, 'count': 2}) + self.assertEqual(container_metrics({'container_error': 'EOF', **row}), {}) + self.resources() + rows = [json.loads(line) for line in (self.root / 'resources.jsonl').read_text().splitlines()] + for sample in rows: + sample['containers'] = [ + {'MemUsage': ('10MiB' if sample['phase'] == 'load' else '100MiB') + ' / 1GiB', 'PIDs': '1'} + ] + (self.root / 'resources.jsonl').write_text(''.join(json.dumps(row) + '\n' for row in rows)) + verdict = resource_verdict(self.root, {**self.limits, 'containers': {'memory_bytes': 1024**2}}) + self.assertIn('docker_containers/memory_bytes', verdict['errors'][0]) + + def test_empty_observed_container_set_is_successful_recovery(self): + self.resources() + rows = [json.loads(line) for line in (self.root / 'resources.jsonl').read_text().splitlines()] + for row in rows: + row['containers'] = [{'MemUsage': '10MiB / 1GiB', 'PIDs': '2'}] if row['phase'] == 'load' else [] + limits = {**self.limits, 'containers': {'memory_bytes': 0, 'pids': 0, 'count': 0}} + + def verdict(): + (self.root / 'resources.jsonl').write_text(''.join(json.dumps(row) + '\n' for row in rows)) + return resource_verdict(self.root, limits) + + result = verdict() + self.assertEqual(result['errors'], []) + for metric in limits['containers']: + self.assertEqual(result['targets']['docker_containers'][metric]['cooldown_median'], 0) + rows[-1]['container_error'] = 'collection failed' + self.assertTrue(verdict()['errors']) + del rows[-1]['container_error'] + del rows[-1]['containers'] + self.assertTrue(verdict()['errors']) + + def test_stimulus_variables_are_compared_but_target_origin_is_not(self): + profile = { + 'vars': {'prompt': 'first', 'endpoint': 'http://127.0.0.1:1'}, + 'cases': [ + {'args': ['pipeline', 'run', '--message', '${prompt}'], 'http_group': {'url': '${endpoint}/sync'}} + ], + } + before = workload_contract(profile, self.root / 'profile.json') + profile['vars']['endpoint'] = 'http://127.0.0.1:2' + self.assertEqual(before, workload_contract(profile, self.root / 'profile.json')) + profile['vars']['prompt'] = 'changed' + self.assertNotEqual(before, workload_contract(profile, self.root / 'profile.json')) + + def fixture_runs(self): + summary = { + 'passed': True, + 'phases': [ + { + 'name': 'sustained', + 'achieved_rps': 1, + 'verification_enabled': True, + 'verified_runs': 100, + 'ledger_closeout': {'matched': True}, + 'runner_requests': 100, + 'cases': {'chat': {'requests': 100, 'p95_seconds': 1}}, + } + ], + } + for name in ['baseline', 'candidate']: + root = self.root / name + root.mkdir() + (root / 'summary.json').write_text(json.dumps(summary)) + (root / 'workload.json').write_text(json.dumps({'verified_runs': True, 'fixture': 'golden'})) + (root / 'environment.json').write_text( + json.dumps( + { + 'plugins': { + 'plugins': [ + { + 'author': 'team', + 'name': 'Runner', + 'version': '1', + 'enabled': True, + 'status': 'initialized', + } + ] + } + } + ) + ) + return self.root / 'baseline', self.root / 'candidate', summary + + def test_comparison_rejects_slow_underfed_missing_or_unverified_samples(self): + baseline, candidate, original = self.fixture_runs() + self.assertTrue(compare(baseline, candidate)['passed']) + for field, value in [('p95_seconds', 1.5), ('requests', 2), ('p95_seconds', float('nan'))]: + summary = copy.deepcopy(original) + summary['phases'][0]['cases']['chat'][field] = value + (candidate / 'summary.json').write_text(json.dumps(summary)) + with self.subTest(field=field, value=value): + self.assertFalse(compare(baseline, candidate)['passed']) + for field, value in [ + ('achieved_rps', 0.5), + ('achieved_rps', float('nan')), + ('verified_runs', 99), + ('ledger_closeout', {'matched': False}), + ]: + summary = copy.deepcopy(original) + summary['phases'][0][field] = value + (candidate / 'summary.json').write_text(json.dumps(summary)) + with self.subTest(field=field): + self.assertFalse(compare(baseline, candidate)['passed']) + + def test_comparison_refuses_different_workload_or_failed_baseline(self): + baseline, candidate, summary = self.fixture_runs() + (candidate / 'workload.json').write_text(json.dumps({'verified_runs': True, 'fixture': 'changed'})) + self.assertIn('not comparable', compare(baseline, candidate)['errors'][0]) + (candidate / 'workload.json').write_text((baseline / 'workload.json').read_text()) + summary['passed'] = False + (baseline / 'summary.json').write_text(json.dumps(summary)) + self.assertFalse(compare(baseline, candidate)['passed']) + + def test_fixture_content_changes_contract_even_at_same_path(self): + fixture = self.root / 'fixture.txt' + fixture.write_text('before') + profile = {'resources': [{'documents': [{'file': 'fixture.txt'}]}]} + before = workload_contract(profile, self.root / 'profile.json') + fixture.write_text('after') + self.assertNotEqual(before, workload_contract(profile, self.root / 'profile.json')) + + +if __name__ == '__main__': + unittest.main() diff --git a/tools/test_regression_load_ledger.py b/tools/test_regression_load_ledger.py new file mode 100644 index 000000000..19cedbdac --- /dev/null +++ b/tools/test_regression_load_ledger.py @@ -0,0 +1,137 @@ +"""Check delayed duplicate runs and separate ledger setup from request latency.""" + +from contextlib import redirect_stdout +import io +import json +from pathlib import Path +import sqlite3 +import sys +import tempfile +from types import SimpleNamespace +import time +import unittest +from unittest.mock import patch, Mock +from uuid import uuid4 + +from regression_traffic import main, reconcile_runs, sample_containers + + +class LoadLedgerTests(unittest.TestCase): + def setUp(self): + directory = tempfile.TemporaryDirectory() + self.addCleanup(directory.cleanup) + self.root = Path(directory.name) + self.db = self.root / 'ledger.db' + with sqlite3.connect(self.db) as db: + db.executescript(""" + CREATE TABLE agent_run (id INTEGER PRIMARY KEY, run_id TEXT, workspace_id TEXT, + conversation_id TEXT, status TEXT, status_reason TEXT, finished_at TEXT); + CREATE TABLE agent_run_event (run_id TEXT, sequence INTEGER, type TEXT, data_json TEXT); + """) + + def test_phase_closeout_catches_delayed_extra_and_unfinished_runs(self): + with sqlite3.connect(self.db) as db: + db.execute("INSERT INTO agent_run VALUES (1,'expected','ws','group','completed','','now')") + db.execute("INSERT INTO agent_run VALUES (2,'unrelated','other','group','completed','','now')") + self.assertTrue(reconcile_runs(self.db, 'ws', 0, {'group'}, {'expected'})['matched']) + with sqlite3.connect(self.db) as db: + db.execute("INSERT INTO agent_run VALUES (3,'late','ws','group','running','',NULL)") + result = reconcile_runs(self.db, 'ws', 0, {'group'}, {'expected'}) + self.assertFalse(result['matched']) + self.assertEqual(result['extra'], ['late']) + self.assertEqual(result['not_completed'], ['late']) + + def test_container_churn_recovers_once_but_other_collection_errors_fail(self): + failed = SimpleNamespace(returncode=1, stdout='', stderr='No such container: old') + success = SimpleNamespace(returncode=0, stdout='{"MemUsage":"1MiB / 1GiB","PIDs":"1"}\n', stderr='') + run = Mock(side_effect=[failed, success]) + result = sample_containers([], '/owned', run, Mock(side_effect=[['old', 'live'], ['live']])) + self.assertNotIn('container_error', result) + self.assertEqual(result['container_churn']['removed'], ['old']) + self.assertEqual(run.call_count, 2) + self.assertEqual(run.call_args.args[0][-1], 'live') + run = Mock(return_value=failed) + result = sample_containers([], '/owned', run, Mock(return_value=['live'])) + self.assertIn('container_error', result) + self.assertEqual(run.call_count, 1) + run = Mock(return_value=failed) + result = sample_containers([], '/owned', run, Mock(side_effect=[['old', 'live'], ['live']])) + self.assertIn('container_error', result) + self.assertEqual(run.call_count, 2) + + def test_slow_ledger_snapshot_does_not_inflate_response_latency(self): + connect = sqlite3.connect + cases = self.root / 'cases.json' + cases.write_text( + json.dumps( + [ + { + 'name': 'chat', + 'expect': 'OK', + 'http_group': { + 'url': 'http://127.0.0.1:1/unused', + 'message': 'hello', + 'secret_env': 'TEST_GROUP_SECRET', + }, + } + ] + ) + ) + + def slow_connect(*args, **kwargs): + time.sleep(0.12) + return connect(*args, **kwargs) + + def response(case, group, *_args): + run_id = uuid4().hex + with connect(self.db) as db: + db.execute( + 'INSERT INTO agent_run VALUES (NULL,?,?,?,?,?,?)', + (run_id, 'ws', 'group_' + group, 'completed', '', 'now'), + ) + db.execute('INSERT INTO agent_run_event VALUES (?,1,?,?)', (run_id, 'run.completed', '{}')) + return {'data': {'session_id': group}} + + output = self.root / 'run' + argv = [ + 'traffic', + '--lbctl', + 'unused', + '--config', + 'unused', + '--cases', + str(cases), + '--output', + str(output), + '--duration', + '0.4', + '--cooldown', + '0', + '--rps', + '8', + '--workers', + '1', + '--groups', + '1', + '--ledger-db', + str(self.db), + '--workspace', + 'ws', + ] + with ( + patch.object(sys, 'argv', argv), + patch.dict('os.environ', {'TEST_GROUP_SECRET': 'test'}), + patch('regression_traffic.sqlite3.connect', side_effect=slow_connect), + patch('regression_traffic.group_request', side_effect=response), + redirect_stdout(io.StringIO()), + ): + main() + rows = [json.loads(line) for line in (output / 'requests.jsonl').read_text().splitlines()] + self.assertGreaterEqual(len(rows), 2) + self.assertTrue(all(row['ledger_setup_seconds'] >= 0.12 for row in rows)) + self.assertTrue(all(row['seconds'] < row['ledger_setup_seconds'] for row in rows)) + self.assertTrue(json.loads((output / 'summary.json').read_text())['ledger_closeout']['matched']) + + +if __name__ == '__main__': + unittest.main()