Files
LangBot/tools/regression_smoke.py

548 lines
26 KiB
Python

"""Run finite CLI/group smoke cases with an independent read-only AgentRun oracle."""
from collections import Counter
from contextlib import closing
import base64
import json
import hashlib
import hmac
import os
from pathlib import Path
import sqlite3
import time
import urllib.error
import urllib.request
from uuid import uuid4
from regression_traffic import group_request, percentile, read_endpoint_snapshot, reconcile_runs
from regression_coverage import checked, check_response
from regression_local_cases import ScenarioUnavailable
class UnresolvedRun(RuntimeError):
"""The request cannot yet be matched to one authoritative terminal run."""
def completed_run(connection, workspace, conversation, after_id, timeout):
deadline = time.monotonic() + timeout
while True:
rows = connection.execute(
'SELECT run_id, status, status_reason, finished_at FROM agent_run '
'WHERE workspace_id=? AND conversation_id=? AND id>? ORDER BY id',
(workspace, conversation, after_id),
).fetchall()
if len(rows) > 1:
raise UnresolvedRun('More than one AgentRun matches this request')
if rows and (rows[0]['finished_at'] or rows[0]['status'] in {'completed', 'failed', 'timeout', 'cancelled'}):
return dict(rows[0])
if time.monotonic() >= deadline:
raise UnresolvedRun('No unique terminal AgentRun before the observation deadline; not retried')
time.sleep(0.1)
def check_run(connection, run, case):
if run['status'] != 'completed':
raise ValueError(f'AgentRun {run["run_id"]} ended as {run["status"]}: {run["status_reason"]}')
events = [
(kind, json.loads(data or '{}'))
for kind, data in connection.execute(
'SELECT type, data_json FROM agent_run_event WHERE run_id=? ORDER BY sequence', (run['run_id'],)
)
]
if any(not isinstance(data, dict) for _kind, data in events):
raise ValueError('AgentRun event payload must be an object')
if not any(kind == 'run.completed' for kind, _data in events):
raise ValueError('Completed AgentRun is missing its persisted run.completed event')
checks = {'terminal_event': True}
if expected := case.get('box_exec'):
calls = [
data.get('result', {})
for kind, data in events
if kind == 'tool.call.completed' and data.get('tool_name') == 'exec'
]
if len(calls) != expected['count'] or any(
not isinstance(item, dict)
or item.get('backend') != expected['backend']
or item.get('exit_code') != expected.get('exit_code', 0)
or not isinstance(item.get('stdout'), str)
or expected['stdout'] not in item['stdout']
or ('stderr' in expected and expected['stderr'] not in str(item.get('stderr', '')))
for item in calls
):
raise ValueError('Box execution count/backend/exit code/output did not match the oracle')
checks['box_exec'] = {'calls': len(calls), 'backend': expected['backend'], 'output_matched': True}
return checks
def check_attachment(data, expected):
replies = data.get('replies')
if not isinstance(replies, list) or any(
not isinstance(reply, dict)
or not isinstance(reply.get('message_chain'), list)
or any(not isinstance(part, dict) for part in reply['message_chain'])
for reply in replies
):
raise ValueError('Malformed file delivery response')
attachments = [part for reply in replies for part in reply.get('message_chain', []) if part.get('type') == 'File']
if len(attachments) != 1 or attachments[0].get('name') != expected['name']:
raise ValueError('Expected exactly one file attachment with the configured name')
encoded = attachments[0].get('base64')
if not isinstance(encoded, str):
raise ValueError('Attachment must contain base64 bytes')
content = base64.b64decode(encoded, validate=True)
if content != expected['text'].encode('utf-8'):
raise ValueError('Exported attachment bytes differ from the expected fixture')
return {'name': expected['name'], 'bytes': len(content), 'content_matched': True}
def check_fields(data, expected):
for path, value in expected.items():
actual = data
try:
for key in path.split('.'):
actual = actual[int(key)] if isinstance(actual, list) else actual[key]
except (KeyError, IndexError, TypeError, ValueError) as exc:
raise ValueError(f'Response field {path} is missing or malformed') from exc
if actual != value:
raise ValueError(f'Response field {path} differs from the expected value')
return expected
def check_mock_requests(path, offset, expected, *, timeout=3):
deadline = time.monotonic() + timeout
while True:
with path.open('rb') as stream:
stream.seek(offset)
lines = stream.read().splitlines()
records = [json.loads(line) for line in lines]
actual = [
{'model': row['request']['model'], 'status': row['status'], 'stream': row['request'].get('stream', False)}
for row in records
if row['path'] == '/v1/chat/completions'
]
if len(actual) >= len(expected) or time.monotonic() >= deadline:
if actual != expected:
raise ValueError(f'Model request sequence differs: expected {expected}, got {actual}')
return actual
time.sleep(0.05)
def signed_headers(body, secret, timestamp=None):
timestamp = str(int(time.time()) if timestamp is None else timestamp)
signature = hmac.new(secret.encode(), timestamp.encode() + b'.' + body, hashlib.sha256).hexdigest()
return {'Content-Type': 'application/json', 'X-LB-Timestamp': timestamp, 'X-LB-Signature': 'sha256=' + signature}
def rejected_group(case, group, env=None, *, timeout=10):
settings = case['http_group']
env = os.environ if env is None else env
kind = case['reject_group']
payload = {
'session_id': group,
'session_type': 'group',
'sender': {'id': 'qa-rejected'},
'message': [{'type': 'Plain', 'text': settings['message']}],
}
if kind == 'missing_session':
payload.pop('session_id')
if kind == 'invalid_message':
payload['message'] = 'not a message chain'
body = json.dumps(payload).encode()
if kind == 'invalid_json':
body = b'{'
if kind == 'invalid_object':
body = b'[]'
timestamp = {
'expired_signature': int(time.time()) - 600,
'future_signature': int(time.time()) + 600,
'bad_timestamp': 'invalid',
}.get(kind)
headers = {'Content-Type': 'application/json'}
if kind == 'bad_signature':
headers.update({'X-LB-Timestamp': str(int(time.time())), 'X-LB-Signature': 'sha256=' + '0' * 64})
elif kind != 'missing_signature':
headers = signed_headers(body, env[settings['secret_env']], timestamp)
if kind == 'tampered_body':
body += b' '
request = urllib.request.Request(settings['url'], headers=headers, data=body)
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
try:
with opener.open(request, timeout=timeout):
raise ValueError('Invalid bot ingress was accepted')
except urllib.error.HTTPError as exc:
payload = json.load(exc)
return {'http_status': exc.code, 'code': payload.get('code')}
def api_probe(case, *, timeout=10):
probe = case['http_probe']
headers = {} if probe['auth'] == 'missing' else {'X-API-Key': 'qa-intentionally-invalid'}
request = urllib.request.Request(probe['url'], headers=headers)
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
try:
response = opener.open(request, timeout=timeout)
except urllib.error.HTTPError as exc:
response = exc
with response:
return {'http_status': response.status, 'body': json.load(response)}
def reset_group(case, group, env, *, timeout=10):
body = json.dumps({'session_id': group, 'session_type': 'group'}).encode()
settings = case['http_group']
request = urllib.request.Request(
settings['url'].removesuffix('/sync') + '/reset',
data=body,
headers=signed_headers(body, env[settings['secret_env']]),
)
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
try:
with opener.open(request, timeout=timeout) as response:
result = json.load(response)
except urllib.error.HTTPError as exc:
# A received error response violates reset's success contract; do not retry it.
raise ValueError(f'Reset returned HTTP {exc.code}; expected 200') from exc
check_fields(result, {'code': 0, 'data.session_id': group, 'data.removed': True})
return result
def verify_cleanup(suite, deleted):
checks = []
for kind, model_type in dict.fromkeys((item['kind'], item.get('type')) for item in deleted):
args = [kind, 'list'] + (['--type', model_type] if model_type else [])
data = suite.call(args)
key = {
'provider': 'providers',
'model': 'models',
'pipeline': 'pipelines',
'bot': 'bots',
'knowledge-base': 'bases',
}[kind]
items = data.get(key) if isinstance(data, dict) else None
if not isinstance(items, list) or any(
not isinstance(item, dict) or not isinstance(item.get('uuid'), str) for item in items
):
raise ValueError(f'Malformed {kind} list response during cleanup verification')
visible = {item['uuid'] for item in items}
expected_absent = {item['uuid'] for item in deleted if item['kind'] == kind and item.get('type') == model_type}
remaining = sorted(expected_absent & visible)
checks.append({'kind': kind, 'type': model_type, 'deleted_count': len(expected_absent), 'remaining': remaining})
suite.write('cleanup-checks.json', checks)
if any(row['remaining'] for row in checks):
raise ValueError('Deleted resources remain visible in CLI list readback')
def run_cycle(suite, rows, health, cycle):
# Imported here because Suite dispatches to this module after profile validation.
from regression_suite import resolve
settings = suite.profile['smoke']
path = Path(suite.path(settings['ledger_db'])).resolve()
if not path.is_file():
raise ValueError('smoke.ledger_db must be an existing target Core SQLite database')
mock_path = Path(suite.path(settings['mock_records'])) if settings.get('mock_records') else None
errors = []
initial_count = len(rows)
complete = True
execution = getattr(suite, 'execution', None)
started = time.monotonic()
def redact(value):
for secret in suite.secrets:
value = value.replace(secret, '[REDACTED]')
return value
def sample_health(phase):
for name, url in resolve(suite.profile['observe']['health'], suite.variables).items():
row = {'phase': phase, 'cycle': cycle, 'endpoint': name}
try:
row['result'] = read_endpoint_snapshot(url, timeout_seconds=suite.timeout(3))
except (RuntimeError, OSError, ValueError) as exc:
row['error'] = redact(str(exc))
errors.append(f'{phase}/{name}: health check failed')
health.append(row)
suite.write('health.json', health)
sample_health('before')
# shortcut: local SQLite only; use a terminal-run API when lbctl exposes one.
with closing(sqlite3.connect(path.as_uri() + '?mode=ro', uri=True, timeout=5)) as connection:
connection.row_factory = sqlite3.Row
connection.execute('PRAGMA query_only=ON')
for original in suite.profile['cases']:
if execution and not execution.admitting():
complete = False
break
case = resolve(original, suite.variables)
is_group = 'http_group' in case
rejected = 'reject_group' in case
runner = not rejected and (is_group or case.get('args', [])[:2] == ['pipeline', 'run'])
groups = (
[f'qa-smoke-{uuid4().hex}' for _ in range(settings.get('groups_per_case', 2))]
if is_group and not rejected
else [f'qa-smoke-{uuid4().hex}' if rejected else None]
)
previous = {group: 'none' for group in groups}
retired = set()
reset_done = False
history_turns = {group: 0 for group in groups}
for turn in range(1, (settings.get('group_turns', 2) if is_group and not rejected else 1) + 1):
for group_index, group in enumerate(groups):
if execution and not execution.admitting():
complete = False
break
row = {
'case': case['name'],
'cycle': cycle,
'stage': case.get('stage', 'execution'),
'status': 'failed',
'requires_runner': runner,
}
if is_group:
row.update(
group_id=group, turn=turn, group_index=group_index, member_id=f'qa-member-{turn % 2:03d}'
)
if case.get('reset_group') and group_index == 1 and turn >= 2 and not reset_done:
row.update(status='skipped', error='Peer reset did not complete; isolation cannot be checked')
rows.append(row)
continue
if group in retired:
row.update(status='skipped', error='Earlier turn failed or is unknown; group not retried')
rows.append(row)
continue
before = time.monotonic()
after_id = connection.execute('SELECT COALESCE(MAX(id),0) FROM agent_run').fetchone()[0]
row['after_id'] = after_id
if runner and is_group:
row['conversation_id'] = 'group_' + group
token = uuid4().hex
offset = mock_path.stat().st_size if case.get('mock_requests') and mock_path.exists() else 0
try:
if 'scenario' in case:
from regression_local_cases import run_case
data = run_case(suite, case, row)
elif rejected:
data = checked(
row,
'response',
lambda: rejected_group(case, group, suite.env, timeout=suite.timeout(10)),
)
count = connection.execute(
'SELECT COUNT(*) FROM agent_run WHERE workspace_id=? AND id>?',
(suite.variables['workspace_uuid'], after_id),
).fetchone()[0]
checked(row, 'no_run', lambda: check_fields({'runs': count}, {'runs': 0}))
elif is_group:
if case.get('reset_group') and turn == 2 and group_index == 0:
checked(
row, 'reset', lambda: reset_group(case, group, suite.env, timeout=suite.timeout(10))
)
reset_done = True
history_turns[group] = 0
previous[group] = 'none'
data = checked(
row,
'response',
lambda: group_request(
case,
group,
row['member_id'],
history_turns[group] + 1,
token,
previous[group],
suite.env,
timeout=suite.timeout(90),
),
)['data']
elif 'http_probe' in case:
data = api_probe(case, timeout=suite.timeout(10))
checked(row, 'response', lambda: check_response(data, case['expect']))
else:
data = suite.call(case['args'], case.get('body'))
checked(row, 'response', lambda: check_response(data, case['expect']))
row['output'] = data
if case.get('forbid'):
checked(row, 'isolation', lambda: check_response(data, case['expect'], case['forbid']))
row.setdefault('checks', {})
if case.get('fields'):
row['checks']['fields'] = checked(row, 'fields', lambda: check_fields(data, case['fields']))
if case.get('attachment'):
row['checks']['attachment'] = checked(
row, 'attachment', lambda: check_attachment(data, case['attachment'])
)
if case.get('mock_requests'):
row['checks']['mock_requests'] = checked(
row,
'model_sequence',
lambda: check_mock_requests(
mock_path, offset, case['mock_requests'], timeout=suite.timeout(3)
),
)
if runner:
if not isinstance(data, dict):
raise ValueError('Runner response must be an object')
session = data.get('session_id')
if not isinstance(session, str) or not session:
raise UnresolvedRun('Reply has no session ID for AgentRun reconciliation')
row['conversation_id'] = 'group_' + session if is_group else session
run = completed_run(
connection,
suite.variables['workspace_uuid'],
'group_' + session if is_group else session,
after_id,
suite.timeout(settings.get('timeout_seconds', 30)),
)
row['run'] = run
row['checks'].update(checked(row, 'ledger', lambda: check_run(connection, run, case)))
row['status'] = 'passed'
previous[group] = token
history_turns[group] += 1
except ScenarioUnavailable as exc:
row.update(status='skipped', error=redact(str(exc)))
except urllib.error.HTTPError as exc:
row.update(
status='unknown' if exc.code >= 500 else 'failed', error=f'HTTP {exc.code}; not retried'
)
except TimeoutError as exc:
row.update(status='unknown', timed_out=True, error=redact(str(exc)))
except (UnresolvedRun, urllib.error.URLError, ConnectionError) as exc:
row.update(status='unknown', error=redact(str(exc)))
if isinstance(getattr(exc, 'reason', None), TimeoutError):
row['timed_out'] = True
except (RuntimeError, ValueError, OSError, KeyError, IndexError, sqlite3.Error) as exc:
message = redact(str(exc))
row.update(
status='unknown'
if 'outcome unknown' in message or 'result_unknown' in message
else 'failed',
error=message,
)
if 'timed out' in message:
row['timed_out'] = True
row['seconds'] = time.monotonic() - before
for error in row.get('cleanup_errors', []):
errors.append(f'{case["name"]}: cleanup failed: {redact(str(error))}')
if row['status'] != 'passed':
errors.append(f'{case["name"]}: {row["status"]}: {row["error"]}')
if is_group:
retired.add(group)
rows.append(row)
suite.write('cases.json', rows)
suite.write(
'progress.json',
{'completed_requests': len(rows), 'status': dict(Counter(x['status'] for x in rows))},
)
suite.write('cases.json', rows)
sample_health('after')
return {
'mode': 'smoke',
'cycle_complete': complete,
'cycle_requests': len(rows) - initial_count,
'phases': [],
'cases': rows,
'errors': errors,
'elapsed_seconds': time.monotonic() - started,
'status': dict(Counter(row['status'] for row in rows)),
'stages': {
stage: dict(Counter(row['status'] for row in rows if row['stage'] == stage))
for stage in dict.fromkeys(row['stage'] for row in rows)
},
'case_count': len(suite.profile['cases']),
'requests': len(rows),
}
def run_smoke(suite):
rows, health, errors, cycles = [], [], [], []
execution = getattr(suite, 'execution', None)
while True:
cycle = len(cycles) + 1
if execution:
execution.mark(f'cycle-{cycle}')
result = run_cycle(suite, rows, health, cycle)
errors.extend(result['errors'])
cycles.append(
{
'cycle': cycle,
'requests': result['cycle_requests'],
'complete': result['cycle_complete'],
'elapsed_seconds': result['elapsed_seconds'],
'status': dict(Counter(row['status'] for row in rows if row['cycle'] == cycle)),
}
)
suite.write('cycles.json', cycles)
if not execution or not execution.admitting() or any(row['status'] in ('unknown', 'skipped') for row in rows):
break
result['errors'] = errors
result['cycles'] = cycles
result['first_pass_complete'] = cycles[0]['complete']
result['incomplete'] = [
f'{row["case"]}: {row["status"]}: {row["error"]}'
for row in rows
if row['status'] in ('unknown', 'skipped') and not row.get('timed_out')
]
if not result['first_pass_complete']:
error = 'Required first pass did not finish before the admission deadline'
result['errors'].append(error)
result['incomplete'].append(error)
result['elapsed_seconds'] = sum(cycle['elapsed_seconds'] for cycle in cycles)
result['performance_mode'] = 'observe'
result['observations'] = {
'successful_requests_per_second': sum(row['status'] == 'passed' for row in rows)
/ max(result['elapsed_seconds'], 0.000001),
'step_p95_seconds': {
name: percentile([row['seconds'] for row in rows if row['case'] == name and 'seconds' in row], 0.95)
for name in dict.fromkeys(row['case'] for row in rows)
},
}
executed = scenario_run_rows(rows)
result['test_data'] = {
'observed_conversations': len({row['conversation_id'] for row in executed if 'conversation_id' in row}),
'observed_agent_runs': len({row['run']['run_id'] for row in executed if 'run' in row}),
'group_sessions': len({row['group_id'] for row in executed if 'group_id' in row}),
'owned_api_resources': len(suite.owned),
}
reconcile_scenario_runs(suite, result)
return result
def scenario_run_rows(rows):
executed = [row for row in rows if row.get('requires_runner') and 'after_id' in row]
for row in rows:
for run in row.get('scenario_runs', []):
executed.append(
{
'after_id': row['after_id'],
'conversation_id': run['conversation_id'],
'run': {'run_id': run['run_id']},
'expected_status': run['expected_status'],
}
)
return executed
def reconcile_scenario_runs(suite, result, *, key='ledger'):
"""Recheck the originally observed conversations without expanding run ownership."""
executed = scenario_run_rows(result.get('cases', []))
if not executed:
return
try:
result[key] = reconcile_runs(
suite.path(suite.profile['smoke']['ledger_db']),
suite.variables['workspace_uuid'],
min(row['after_id'] for row in executed),
{row['conversation_id'] for row in executed if 'conversation_id' in row},
{row['run']['run_id'] for row in executed if 'run' in row},
expected_statuses={
row['run']['run_id']: row.get('expected_status', 'completed') for row in executed if 'run' in row
},
)
except (OSError, sqlite3.Error) as exc:
error = f'Scenario {key} evidence unavailable: {exc}'
result['errors'].append(error)
result.setdefault('incomplete', []).append(error)
return
if not result[key]['matched']:
error = f'Scenario {key} found extra, missing or unexpected AgentRun states'
result['errors'].append(error)