mirror of
https://github.com/langbot-app/LangBot.git
synced 2026-10-24 09:25:21 +08:00
huanghuoguoguo
5ec41f15a1
* test(tools): add deterministic lbctl regression job * test(tools): simulate persistent multi-group conversations * test(tools): add finite core smoke with terminal run and file oracles * test(tools): expand smoke coverage across execution stages * ci: add a timed real core smoke and lightweight tool checks * fix(ci): keep infrastructure failures visible in smoke verdicts
329 lines
16 KiB
Python
329 lines
16 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
|
|
from pathlib import Path
|
|
import sqlite3
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
from uuid import uuid4
|
|
|
|
from regression_traffic import group_request, read_endpoint_snapshot
|
|
|
|
|
|
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):
|
|
deadline = time.monotonic() + 3
|
|
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 rejected_group(case, group):
|
|
settings = case['http_group']
|
|
headers = {'Content-Type': 'application/json'}
|
|
if case['reject_group'] == 'bad_signature':
|
|
headers.update({'X-LB-Timestamp': str(int(time.time())), 'X-LB-Signature': 'sha256=' + '0' * 64})
|
|
request = urllib.request.Request(
|
|
settings['url'],
|
|
headers=headers,
|
|
data=json.dumps(
|
|
{
|
|
'session_id': group,
|
|
'session_type': 'group',
|
|
'sender': {'id': 'qa-rejected'},
|
|
'message': [{'type': 'Plain', 'text': settings['message']}],
|
|
}
|
|
).encode(),
|
|
)
|
|
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
|
|
try:
|
|
with opener.open(request, timeout=10):
|
|
raise ValueError('Unsigned/tampered group message was accepted')
|
|
except urllib.error.HTTPError as exc:
|
|
payload = json.load(exc)
|
|
if exc.code != 401 or not isinstance(payload, dict) or payload.get('code') != 40101:
|
|
raise ValueError('Expected HTTP 401 with bad_signature code 40101') from exc
|
|
return {'http_status': exc.code, 'code': payload['code']}
|
|
|
|
|
|
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_smoke(suite):
|
|
# 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
|
|
rows, errors, health = [], [], []
|
|
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, 'endpoint': name}
|
|
try:
|
|
row['result'] = read_endpoint_snapshot(url, timeout_seconds=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']:
|
|
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()
|
|
for turn in range(1, (settings.get('group_turns', 2) if is_group and not rejected else 1) + 1):
|
|
for group in groups:
|
|
row = {
|
|
'case': case['name'],
|
|
'stage': case.get('stage', 'execution'),
|
|
'status': 'failed',
|
|
'requires_runner': runner,
|
|
}
|
|
if is_group:
|
|
row.update(group_id=group, turn=turn, member_id=f'qa-member-{turn % 2:03d}')
|
|
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]
|
|
token = uuid4().hex
|
|
offset = mock_path.stat().st_size if case.get('mock_requests') and mock_path.exists() else 0
|
|
try:
|
|
if rejected:
|
|
data = rejected_group(case, group)
|
|
count = connection.execute(
|
|
'SELECT COUNT(*) FROM agent_run WHERE workspace_id=? AND conversation_id=? AND id>?',
|
|
(suite.variables['workspace_uuid'], 'group_' + group, after_id),
|
|
).fetchone()[0]
|
|
if count:
|
|
raise ValueError('Rejected ingress created an AgentRun')
|
|
elif is_group:
|
|
data = group_request(
|
|
case, group, row['member_id'], turn, token, previous[group], suite.env
|
|
)['data']
|
|
else:
|
|
data = suite.call(case['args'], case.get('body'))
|
|
if case['expect'] not in json.dumps(data, ensure_ascii=False):
|
|
raise ValueError('Expected business response was not returned')
|
|
row['output'] = data
|
|
if any(value in json.dumps(data, ensure_ascii=False) for value in case.get('forbid', [])):
|
|
raise ValueError('Response contains forbidden content from another knowledge base')
|
|
row['checks'] = {}
|
|
if case.get('fields'):
|
|
row['checks']['fields'] = check_fields(data, case['fields'])
|
|
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')
|
|
run = completed_run(
|
|
connection,
|
|
suite.variables['workspace_uuid'],
|
|
'group_' + session if is_group else session,
|
|
after_id,
|
|
settings.get('timeout_seconds', 30),
|
|
)
|
|
row['run'] = run
|
|
row['checks'].update(check_run(connection, run, case))
|
|
if case.get('attachment'):
|
|
row['checks']['attachment'] = check_attachment(data, case['attachment'])
|
|
if case.get('mock_requests'):
|
|
row['checks']['mock_requests'] = check_mock_requests(
|
|
mock_path, offset, case['mock_requests']
|
|
)
|
|
row['status'] = 'passed'
|
|
previous[group] = token
|
|
except urllib.error.HTTPError as exc:
|
|
row.update(
|
|
status='unknown' if exc.code >= 500 else 'failed', error=f'HTTP {exc.code}; not retried'
|
|
)
|
|
except (UnresolvedRun, urllib.error.URLError, TimeoutError) as exc:
|
|
row.update(status='unknown', error=redact(str(exc)))
|
|
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,
|
|
)
|
|
row['seconds'] = time.monotonic() - before
|
|
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',
|
|
'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),
|
|
}
|