From 77dfa17c57b234603c5f25f97a39205124fe8eda Mon Sep 17 00:00:00 2001 From: huanghuoguoguo <1051233107@qq.com> Date: Wed, 28 Oct 2026 16:11:19 +0800 Subject: [PATCH 1/3] test(regression): add selectable resource lifecycle cases (#2794) --- tools/regression_contracts.json | 5 +- tools/regression_local_cases.py | 105 ++++++++ tools/regression_resource_cases.py | 342 ++++++++++++++++++++++++ tools/regression_scenarios.py | 6 +- tools/regression_smoke.py | 41 ++- tools/regression_suite.py | 28 +- tools/regression_traffic.py | 9 +- tools/test_regression_coverage.py | 7 +- tools/test_regression_local_cases.py | 110 ++++++++ tools/test_regression_resource_cases.py | 212 +++++++++++++++ 10 files changed, 851 insertions(+), 14 deletions(-) create mode 100644 tools/regression_local_cases.py create mode 100644 tools/regression_resource_cases.py create mode 100644 tools/test_regression_local_cases.py create mode 100644 tools/test_regression_resource_cases.py diff --git a/tools/regression_contracts.json b/tools/regression_contracts.json index 048355c99..8d0729e30 100644 --- a/tools/regression_contracts.json +++ b/tools/regression_contracts.json @@ -173,7 +173,7 @@ {"id": "auth.revocation", "area": "auth", "kind": "blackbox", "priority": "P0", "description": "Revoked credentials stop authorizing new requests", - "bindings": [] + "bindings": [{"profile": "scenarios", "case": "auth_revocation", "checks": ["credential_active", "credential_revoked", "key_status", "key_cleanup"]}] }, {"id": "workspace.read_isolation", "area": "workspace", "kind": "blackbox", "priority": "P0", @@ -248,7 +248,7 @@ {"id": "agent.direct_execution", "area": "agent", "kind": "blackbox", "priority": "P1", "description": "Standalone Agent entrypoint completes through the public API", - "bindings": [] + "bindings": [{"profile": "scenarios", "case": "agent_direct_execution", "checks": ["agent_response", "agent_terminal", "agent_events", "agent_cleanup"]}] }, {"id": "persistence.postgres_parity", "area": "persistence", "kind": "blackbox", "priority": "P1", @@ -265,6 +265,7 @@ "description": "A real embedding model retrieves relevant fixed documents", "bindings": [] }, + {"id": "retrieval.delete_visibility", "area": "retrieval", "kind": "blackbox", "priority": "P1", "description": "Deleting one ingested document removes its retrieval results while preserving a second document", "bindings": [{"profile": "scenarios", "case": "retrieval_delete_visibility", "checks": ["retrieval_initial", "retrieval_deleted", "retrieval_control", "document_cleanup"]}]}, {"id": "execution.persisted_terminal", "area": "execution", "kind": "graybox", "priority": "P0", "description": "Each executed request has a unique completed persisted run and terminal event", diff --git a/tools/regression_local_cases.py b/tools/regression_local_cases.py new file mode 100644 index 000000000..951360e50 --- /dev/null +++ b/tools/regression_local_cases.py @@ -0,0 +1,105 @@ +"""Optional local scenarios share the Suite's verified Workspace and credentials.""" + +from contextlib import closing +import importlib +import json +from pathlib import Path +import sqlite3 +import time +import urllib.error +import urllib.request + +GROUPS = {'resources': 'regression_resource_cases'} + + +class ScenarioUnavailable(RuntimeError): + """An explicit prerequisite is missing; this is not a passed assertion.""" + + +def require(condition, message): + if not condition: + raise ValueError(message) + + +def endpoint(suite, path): + require(path.startswith('/') and not path.startswith('//'), 'Scenario API path must be relative to Core') + base = suite.connection.get('endpoint') + require(bool(base), 'Scenario needs the endpoint verified by lbctl whoami') + return base.rstrip('/') + path + + +def auth_headers(suite): + source = suite.connection.get('credential_source', '') + key = suite.env.get(source.removeprefix('env:')) if source.startswith('env:') else None + require(bool(key), 'Scenario requires environment-backed lbctl credentials') + if key not in suite.secrets: + suite.secrets.append(key) + return {'Authorization': 'Bearer ' + key, 'X-Workspace-Id': suite.variables['workspace_uuid']} + + +headers = auth_headers + + +def api(suite, method, path, body=None, headers=None): + request_headers = dict(auth_headers(suite) if headers is None else headers) + request_headers.setdefault('X-Workspace-Id', suite.variables['workspace_uuid']) + if body is not None: + request_headers['Content-Type'] = 'application/json' + request = urllib.request.Request( + endpoint(suite, path), + method=method, + headers=request_headers, + data=json.dumps(body).encode() if body is not None else None, + ) + opener = urllib.request.build_opener(urllib.request.ProxyHandler({})) + try: + response = opener.open(request, timeout=suite.timeout(90)) + except urllib.error.HTTPError as exc: + response = exc + with response: + return response.status, json.load(response) + + +def record_run(suite, row, run_id, expected_status): + """Register exact run ownership for the independent final ledger check.""" + require(expected_status in {'completed', 'failed', 'cancelled', 'timeout'}, 'Expected Run status must be terminal') + path = Path(suite.path(suite.profile['smoke']['ledger_db'])).resolve() + deadline = time.monotonic() + suite.timeout(30) + with closing(sqlite3.connect(path.as_uri() + '?mode=ro', uri=True, timeout=5)) as connection: + connection.row_factory = sqlite3.Row + while True: + run = connection.execute( + 'SELECT run_id, conversation_id, status, finished_at FROM agent_run ' + 'WHERE workspace_id=? AND run_id=? AND id>?', + (suite.variables['workspace_uuid'], run_id, row['after_id']), + ).fetchone() + if run and run['finished_at']: + observed = dict(run) + observed.update(expected_status=expected_status, after_id=row['after_id']) + row.setdefault('scenario_runs', []).append(observed) + require(run['status'] == expected_status, f'Run ended as {run["status"]}; expected {expected_status}') + return observed + if time.monotonic() >= deadline: + raise TimeoutError('Scenario Run did not reach a persisted terminal state') + time.sleep(0.1) + + +def modules(): + return {name: importlib.import_module(module) for name, module in GROUPS.items()} + + +def add_cases(profile, groups): + groups = list(groups) + require(len(groups) == len(set(groups)), 'Scenario groups must not repeat') + require(not (set(groups) - GROUPS.keys()), 'Unknown local scenario group') + for name in groups: + importlib.import_module(GROUPS[name]).add_cases(profile) + profile['scenario_groups'] = groups + + +def run_case(suite, case, row): + for name, module in modules().items(): + if case['scenario'] in module.CASE_NAMES: + require(name in suite.profile.get('scenario_groups', []), f'Scenario group is disabled: {name}') + return module.run_case(suite, case, row) + raise ValueError(f'Unknown local scenario: {case["scenario"]}') diff --git a/tools/regression_resource_cases.py b/tools/regression_resource_cases.py new file mode 100644 index 000000000..c1db30c48 --- /dev/null +++ b/tools/regression_resource_cases.py @@ -0,0 +1,342 @@ +"""Optional resource lifecycle scenarios against the authenticated Workspace APIs.""" + +import copy +import json +from pathlib import Path +import sys +import tempfile +from uuid import uuid4 + +from regression_coverage import checked +from regression_local_cases import ScenarioUnavailable, api, require +from regression_suite import resolve + +GROUP = 'resources' +CASE_NAMES = {'auth_revocation', 'agent_direct_execution', 'retrieval_delete_visibility'} + + +def add_cases(profile): + resources = {item['id']: item for item in profile['resources']} + base = copy.deepcopy(resources['empty_knowledge_base']) + base.update(id='resource_knowledge_base') + base['body']['name'] = 'regression document lifecycle' + base.pop('documents', None) + profile['resources'].append(base) + for name, stage, expected in ( + ('auth_revocation', 'auth', 'LANGBOT_AUTH_REVOCATION_OK'), + ('agent_direct_execution', 'agent', 'LANGBOT_CHAT_OK'), + ('retrieval_delete_visibility', 'retrieval', 'LANGBOT_DOCUMENT_DELETE_OK'), + ): + profile['cases'].append({'name': name, 'scenario': name, 'stage': stage, 'expect': expected}) + + +def data(suite, method, path, body=None, *, headers=None): + status, response = api(suite, method, path, body, headers=headers) + require( + status == 200 and isinstance(response, dict) and response.get('code') == 0, + f'{method} {path} did not return a successful response (HTTP {status})', + ) + return response.get('data', {}) + + +def journal(suite, row): + row.setdefault('resource_journal', f'resource-case-{uuid4().hex}.json') + suite.write( + row['resource_journal'], + { + 'workspace_uuid': suite.variables['workspace_uuid'], + 'resources': row.get('owned_resources', []), + 'cleanup_errors': row.get('cleanup_errors', []), + }, + ) + + +def cleanup(suite, row, assertion, action): + original = sys.exc_info()[1] + try: + checked(row, assertion, action) + except Exception as exc: + message = f'{type(exc).__name__}: {exc}' + for secret in suite.secrets: + message = message.replace(secret, '[REDACTED]') + row.setdefault('cleanup_errors', []).append(message) + if original is None: + raise + finally: + journal(suite, row) + + +def auth_revocation(suite, case, row): + token = suite.env.get('LANGBOT_QA_USER_TOKEN') + if not token: + raise ScenarioUnavailable('auth.revocation requires LANGBOT_QA_USER_TOKEN with api_key.manage') + if token not in suite.secrets: + suite.secrets.append(token) + manager = {'Authorization': 'Bearer ' + token} + name = 'qa-revoke-' + uuid4().hex + owned = {'kind': 'api-key', 'name': name, 'state': 'creating'} + row.setdefault('owned_resources', []).append(owned) + journal(suite, row) + + def revoke_owned(): + if 'id' not in owned: + keys = data(suite, 'GET', '/api/v1/apikeys', headers=manager)['keys'] + matches = [key for key in keys if key.get('name') == name] + require(len(matches) == 1, 'Created key identity is unresolved; inspect the resource journal') + owned['id'] = matches[0]['id'] + journal(suite, row) + path = f'/api/v1/apikeys/{owned["id"]}' + key = data(suite, 'GET', path, headers=manager)['key'] + if key['status'] != 'revoked': + data(suite, 'DELETE', path, headers=manager) + key = data(suite, 'GET', path, headers=manager)['key'] + require(key['status'] == 'revoked', 'Owned credential remains active after cleanup') + owned['state'] = 'revoked' + + try: + created = data(suite, 'POST', '/api/v1/apikeys', {'name': name, 'scopes': ['resource.view']}, headers=manager)[ + 'key' + ] + owned.update(id=created['id'], uuid=created['uuid'], state='active') + journal(suite, row) + secret = created['key'] + require(isinstance(secret, str) and bool(secret), 'Key creation did not return its one-time secret') + suite.secrets.append(secret) + child = {'X-API-Key': secret} + identity = data(suite, 'GET', '/api/v1/system/context', headers=child) + checked( + row, + 'credential_active', + lambda: require( + identity.get('workspace_uuid') == suite.variables['workspace_uuid'] + and identity.get('api_key_id') == owned['uuid'], + 'New credential did not authenticate as its own Workspace key', + ), + ) + data(suite, 'GET', '/api/v1/pipelines', headers=child) + data(suite, 'DELETE', f'/api/v1/apikeys/{owned["id"]}', headers=manager) + status, response = api(suite, 'GET', '/api/v1/pipelines', headers=child) + checked( + row, + 'credential_revoked', + lambda: require( + status == 401 and isinstance(response, dict) and response.get('code') != 0, + 'Revoked credential still authorizes resource access', + ), + ) + key = data(suite, 'GET', f'/api/v1/apikeys/{owned["id"]}', headers=manager)['key'] + checked( + row, + 'key_status', + lambda: require( + key.get('status') == 'revoked' and key.get('secret_available') is False and 'key' not in key, + 'Revoked key readback has incorrect status or exposes its secret', + ), + ) + return {'result': case['expect'], 'key_id': owned['id'], 'status': 'revoked'} + finally: + cleanup(suite, row, 'key_cleanup', revoke_owned) + + +def agent_direct_execution(suite, case, row): + template = next(item for item in suite.profile['resources'] if item['id'] == 'chat_pipeline') + config = resolve(copy.deepcopy(template['body']['config']['ai']), suite.variables) + name = 'qa-agent-' + uuid4().hex + conversation = 'qa-direct-' + uuid4().hex + owned = {'kind': 'agent', 'name': name, 'state': 'creating'} + row.setdefault('owned_resources', []).append(owned) + journal(suite, row) + + def delete_owned(): + if 'uuid' not in owned: + agents = data(suite, 'GET', '/api/v1/agents')['agents'] + matches = [agent for agent in agents if agent.get('name') == name and agent.get('kind') == 'agent'] + require(len(matches) == 1, 'Created Agent identity is unresolved; inspect the resource journal') + owned['uuid'] = matches[0]['uuid'] + journal(suite, row) + path = f'/api/v1/agents/{owned["uuid"]}' + status, _ = api(suite, 'GET', path) + if status != 404: + require(status == 200, 'Cannot read owned Agent before cleanup') + data(suite, 'DELETE', path) + status, _ = api(suite, 'GET', path) + require(status == 404, 'Owned Agent remains visible after cleanup') + owned['state'] = 'deleted' + + try: + created = data(suite, 'POST', '/api/v1/agents', {'name': name, 'kind': 'agent', 'config': config}) + owned.update(uuid=created['uuid'], state='created') + journal(suite, row) + path = f'/api/v1/agents/{owned["uuid"]}' + reply = data( + suite, 'POST', path + '/debug', {'text': 'Run the direct Agent scenario', 'conversation_id': conversation} + ) + runs = data(suite, 'GET', path + '/runs')['items'] + matches = [run for run in runs if run.get('conversation_id') == conversation] + require(len(matches) == 1, 'Direct Agent request did not produce exactly one matching Run') + run = matches[0] + row.setdefault('scenario_runs', []).append( + { + 'run_id': run['run_id'], + 'conversation_id': conversation, + 'expected_status': 'completed', + 'after_id': row['after_id'], + } + ) + checked( + row, + 'agent_terminal', + lambda: require( + run.get('status') == 'completed' and bool(run.get('finished_at')), + 'Direct Agent Run did not reach completed status', + ), + ) + checked( + row, + 'agent_response', + lambda: require( + reply.get('conversation_id') == conversation and case['expect'] in reply.get('final_text', ''), + 'Direct Agent returned the wrong conversation or business reply', + ), + ) + events = data(suite, 'GET', path + f'/runs/{run["run_id"]}/events') + checked( + row, + 'agent_events', + lambda: require( + events.get('run', {}).get('run_id') == run['run_id'] + and any( + event.get('type') == 'run.completed' and event.get('run_id') == run['run_id'] + for event in events.get('items', []) + ), + 'Direct Agent completion event is missing', + ), + ) + return {'result': reply['final_text'], 'run_id': run['run_id'], 'conversation_id': conversation} + finally: + cleanup(suite, row, 'agent_cleanup', delete_owned) + + +def retrieval_delete_visibility(suite, case, row): + kb_uuid = suite.variables['resource_knowledge_base'] + path = f'/api/v1/knowledge/bases/{kb_uuid}' + before = {item['uuid'] for item in data(suite, 'GET', path + '/files')['files']} + owned = {'kind': 'knowledge-files', 'knowledge_base_uuid': kb_uuid, 'files': [], 'state': 'ingesting'} + row.setdefault('owned_resources', []).append(owned) + journal(suite, row) + + def files(): + return [item for item in data(suite, 'GET', path + '/files')['files'] if item['uuid'] not in before] + + def cleanup_documents(): + remaining = files() + owned['files'] = [{'uuid': item['uuid'], 'status': item['status']} for item in remaining] + journal(suite, row) + unresolved = [item['uuid'] for item in remaining if item['status'] in {'pending', 'processing', 'interrupted'}] + for item in remaining: + if item['uuid'] not in unresolved: + data(suite, 'DELETE', path + '/files/' + item['uuid']) + require( + not unresolved, 'Document ingestion outcome is unresolved; retain the owned knowledge base for inspection' + ) + require(not files(), 'Owned documents remain visible after cleanup') + owned['state'] = 'deleted' + + def retrieve(query): + found = data(suite, 'POST', path + '/retrieve', {'query': query, 'retrieval_settings': {'top_k': 10}})[ + 'results' + ] + require( + isinstance(found, list) + and all( + isinstance(item, dict) + and isinstance(item.get('metadata'), dict) + and isinstance(item.get('content'), list) + for item in found + ), + 'Malformed retrieval response', + ) + return found + + def contains(found, document, marker): + return any( + item.get('metadata', {}).get('file_id') == document['engine_document_id'] + and marker in json.dumps(item.get('content', []), ensure_ascii=False) + for item in found + ) + + try: + documents, markers = [], ['obsolete' + uuid4().hex, 'retained' + uuid4().hex] + with tempfile.TemporaryDirectory(prefix='resource-documents-', dir=suite.output) as directory: + for index, marker in enumerate(markers): + document = Path(directory) / f'document-{index}.txt' + document.write_text(f'Lifecycle fixture: {marker}. This is document {index}.\n') + previous = {item['uuid'] for item in files()} + suite.call( + [ + 'knowledge-base', + 'ingest', + kb_uuid, + '--file', + str(document), + '--parser-plugin-id', + 'langbot-team/GeneralParsers', + '--wait', + ] + ) + added = [item for item in files() if item['uuid'] not in previous] + require( + len(added) == 1 and added[0].get('status') == 'completed', + 'Document ingestion did not create exactly one completed file', + ) + documents.append(added[0]) + owned['files'].append({'uuid': added[0]['uuid'], 'status': added[0]['status']}) + journal(suite, row) + initial = retrieve(' '.join(markers)) + checked( + row, + 'retrieval_initial', + lambda: require( + all(contains(initial, document, marker) for document, marker in zip(documents, markers)), + 'Both fixture documents must be independently retrievable before deletion', + ), + ) + data(suite, 'DELETE', path + '/files/' + documents[0]['uuid']) + after = retrieve(' '.join(markers)) + checked( + row, + 'retrieval_deleted', + lambda: require( + all( + item.get('metadata', {}).get('file_id') != documents[0]['engine_document_id'] + and markers[0] not in json.dumps(item, ensure_ascii=False) + for item in after + ) + and documents[0]['uuid'] not in {item['uuid'] for item in files()}, + 'Deleted document remains visible in retrieval or the file list', + ), + ) + checked( + row, + 'retrieval_control', + lambda: require( + contains(after, documents[1], markers[1]), + 'Deleting one document removed the surviving control document', + ), + ) + return { + 'result': case['expect'], + 'deleted_document_id': documents[0]['uuid'], + 'control_document_id': documents[1]['uuid'], + } + finally: + cleanup(suite, row, 'document_cleanup', cleanup_documents) + + +def run_case(suite, case, row): + actions = { + 'auth_revocation': auth_revocation, + 'agent_direct_execution': agent_direct_execution, + 'retrieval_delete_visibility': retrieval_delete_visibility, + } + return actions[case['scenario']](suite, case, row) diff --git a/tools/regression_scenarios.py b/tools/regression_scenarios.py index f95fa7598..a7e37deef 100644 --- a/tools/regression_scenarios.py +++ b/tools/regression_scenarios.py @@ -11,7 +11,7 @@ ROOT = Path(__file__).parent RUNNER = 'plugin:langbot-team/LocalAgent/default' -def expanded_profile(): +def expanded_profile(groups=()): profile = json.loads((ROOT / 'regression_profiles/smoke.json').read_text()) profile['coverage_suite'] = 'scenarios' profile['smoke']['group_turns'] = 3 @@ -102,6 +102,10 @@ def expanded_profile(): expect='LANGBOT_' + name.upper() + '_OK', ) profile['cases'].append(case) + if groups: + from regression_local_cases import add_cases + + add_cases(profile, groups) return profile diff --git a/tools/regression_smoke.py b/tools/regression_smoke.py index c996c0ff5..2d7a32e19 100644 --- a/tools/regression_smoke.py +++ b/tools/regression_smoke.py @@ -16,6 +16,7 @@ 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): @@ -316,7 +317,11 @@ def run_cycle(suite, rows, health, cycle): token = uuid4().hex offset = mock_path.stat().st_size if case.get('mock_requests') and mock_path.exists() else 0 try: - if rejected: + if 'scenario' in case: + from regression_local_cases import run_case + + data = run_case(suite, case, row) + elif rejected: data = checked( row, 'response', @@ -358,7 +363,7 @@ def run_cycle(suite, rows, health, cycle): row['output'] = data if case.get('forbid'): checked(row, 'isolation', lambda: check_response(data, case['expect'], case['forbid'])) - row['checks'] = {} + row.setdefault('checks', {}) if case.get('fields'): row['checks']['fields'] = checked(row, 'fields', lambda: check_fields(data, case['fields'])) if case.get('attachment'): @@ -392,6 +397,8 @@ def run_cycle(suite, rows, health, cycle): 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' @@ -413,6 +420,8 @@ def run_cycle(suite, rows, health, cycle): 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: @@ -462,7 +471,7 @@ def run_smoke(suite): } ) suite.write('cycles.json', cycles) - if not execution or not execution.admitting() or any(row['status'] == 'unknown' for row in rows): + 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 @@ -470,7 +479,7 @@ def run_smoke(suite): result['incomplete'] = [ f'{row["case"]}: {row["status"]}: {row["error"]}' for row in rows - if row['status'] == 'unknown' and not row.get('timed_out') + 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' @@ -486,7 +495,7 @@ def run_smoke(suite): for name in dict.fromkeys(row['case'] for row in rows) }, } - executed = [row for row in rows if row.get('requires_runner') and 'after_id' in row] + 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}), @@ -497,9 +506,24 @@ def run_smoke(suite): 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 = [row for row in result.get('cases', []) if row.get('requires_runner') and 'after_id' in row] + executed = scenario_run_rows(result.get('cases', [])) if not executed: return try: @@ -509,6 +533,9 @@ def reconcile_scenario_runs(suite, result, *, key='ledger'): 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}' @@ -516,5 +543,5 @@ def reconcile_scenario_runs(suite, result, *, key='ledger'): result.setdefault('incomplete', []).append(error) return if not result[key]['matched']: - error = f'Scenario {key} found extra, missing or non-completed AgentRuns' + error = f'Scenario {key} found extra, missing or unexpected AgentRun states' result['errors'].append(error) diff --git a/tools/regression_suite.py b/tools/regression_suite.py index 73c152689..ba5bd0073 100644 --- a/tools/regression_suite.py +++ b/tools/regression_suite.py @@ -83,6 +83,7 @@ def validate(profile): 'coverage_suite', 'execution', 'performance_mode', + 'scenario_groups', } if extra := profile.keys() - allowed: raise ValueError(f'unknown profile fields: {sorted(extra)}') @@ -98,6 +99,16 @@ def validate(profile): raise ValueError('coverage_suite must be a stable suite identifier') if profile.get('performance_mode', 'gate') not in {'gate', 'observe'}: raise ValueError('performance_mode must be gate or observe') + if 'scenario_groups' in profile: + from regression_local_cases import GROUPS + + groups = profile['scenario_groups'] + if ( + not isinstance(groups, list) + or any(not isinstance(name, str) or name not in GROUPS for name in groups) + or len(groups) != len(set(groups)) + ): + raise ValueError('scenario_groups must contain unique supported local groups') execution = profile.get('execution') if execution is not None: if ( @@ -192,6 +203,7 @@ def validate(profile): 'reject_group', 'http_probe', 'reset_group', + 'scenario', }: raise ValueError('unknown case field') if not isinstance(case.get('name'), str) or not case['name'] or case['name'] in names: @@ -250,7 +262,21 @@ def validate(profile): case['reset_group'] is not True or 'http_group' not in case or 'reject_group' in case ): raise ValueError('reset_group requires an accepted group case') - if 'http_probe' in case: + if 'scenario' in case: + from regression_local_cases import modules + + handlers = modules() + if ( + 'smoke' not in profile + or not isinstance(case['scenario'], str) + or not any( + case['scenario'] in module.CASE_NAMES and name in profile.get('scenario_groups', []) + for name, module in handlers.items() + ) + or any(key in case for key in ('args', 'body', 'http_probe', 'http_group', 'box_exec', 'attachment')) + ): + raise ValueError('scenario must name a supported finite local case without another request mode') + elif 'http_probe' in case: probe = case['http_probe'] if ( not isinstance(probe, dict) diff --git a/tools/regression_traffic.py b/tools/regression_traffic.py index aa1409951..05f5b5fbe 100644 --- a/tools/regression_traffic.py +++ b/tools/regression_traffic.py @@ -131,9 +131,14 @@ 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): +def reconcile_runs(path, workspace, after_id, conversations, expected_ids, *, expected_statuses=None): """Catch delayed extra or nonterminal runs after request-level verification.""" actual, unfinished = set(), [] + expected_statuses = expected_statuses or {} + if set(expected_statuses) - set(expected_ids) or any( + status not in {'completed', 'failed', 'cancelled', 'timeout'} for status in expected_statuses.values() + ): + raise ValueError('Expected statuses must name owned runs and terminal states') connection = sqlite3.connect(Path(path).resolve().as_uri() + '?mode=ro', uri=True) try: values = sorted(conversations) @@ -145,7 +150,7 @@ def reconcile_runs(path, workspace, after_id, conversations, expected_ids): ) for run_id, status in connection.execute(query, [workspace, after_id, *part]): actual.add(run_id) - if status != 'completed': + if status != expected_statuses.get(run_id, 'completed'): unfinished.append(run_id) finally: connection.close() diff --git a/tools/test_regression_coverage.py b/tools/test_regression_coverage.py index 163f8c68b..f0d7f5870 100644 --- a/tools/test_regression_coverage.py +++ b/tools/test_regression_coverage.py @@ -291,8 +291,13 @@ class Routes: path = Path(__file__).with_name('regression_profiles') / f'{name}.json' if name == 'scenarios': from regression_scenarios import expanded_profile + from regression_local_cases import GROUPS - profile = expanded_profile() + profile = expanded_profile(GROUPS) + if 'lifecycle' in GROUPS: + from regression_lifecycle_cases import restart_profile + + profile['cases'] += restart_profile(profile)['cases'] else: profile = json.loads(path.read_text()) profiles[name] = {x['name']: x for x in profile['cases']} diff --git a/tools/test_regression_local_cases.py b/tools/test_regression_local_cases.py new file mode 100644 index 000000000..e1a42e040 --- /dev/null +++ b/tools/test_regression_local_cases.py @@ -0,0 +1,110 @@ +"""Keep optional cases and expected failures from turning incomplete evidence green.""" + +import json +from pathlib import Path +import sqlite3 +import tempfile +import unittest +from unittest.mock import patch + +from regression_local_cases import GROUPS, ScenarioUnavailable +from regression_scenarios import expanded_profile +from regression_smoke import run_smoke +from regression_suite import Suite, validate +from regression_traffic import reconcile_runs + + +class LocalScenarioTests(unittest.TestCase): + def test_group_switches_keep_base_cases_and_reject_unknown_handlers(self): + base = expanded_profile() + for group in GROUPS: + profile = expanded_profile([group]) + validate(profile) + self.assertEqual(profile['scenario_groups'], [group]) + self.assertGreater(len(profile['cases']), len(base['cases'])) + self.assertEqual(profile['cases'][: len(base['cases'])], base['cases']) + for groups in (['unknown'], ['resources', 'resources']): + with self.assertRaises(ValueError): + expanded_profile(groups) + profile = expanded_profile(['resources']) + profile['scenario_groups'] = [] + with self.assertRaises(ValueError): + validate(profile) + profile['scenario_groups'] = ['resources'] + profile['cases'][-1]['scenario'] = 'not-a-real-handler' + with self.assertRaises(ValueError): + validate(profile) + + def test_missing_prerequisite_is_skipped_and_incomplete(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + with sqlite3.connect(root / 'ledger.db') as connection: + connection.execute('CREATE TABLE agent_run (id INTEGER)') + profile = { + 'version': 1, + 'cli': {'config': 'context.yaml'}, + 'smoke': {'ledger_db': 'ledger.db'}, + 'cases': [{'name': 'auth_revocation', 'scenario': 'auth_revocation', 'expect': 'OK'}], + 'observe': {'health': {}}, + } + suite = Suite(profile, root / 'profile.json', root) + with patch('regression_local_cases.run_case', side_effect=ScenarioUnavailable('Missing user credential')): + result = run_smoke(suite) + self.assertEqual(result['status'], {'skipped': 1}) + self.assertEqual(result['errors'], result['incomplete']) + self.assertIn('Missing user credential', result['incomplete'][0]) + self.assertFalse(result['cases'][0].get('assertions')) + self.assertEqual(json.loads((root / 'cases.json').read_text())[0]['status'], 'skipped') + + def test_cleanup_failure_remains_fail_after_unknown_request(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + with sqlite3.connect(root / 'ledger.db') as connection: + connection.execute('CREATE TABLE agent_run (id INTEGER)') + profile = { + 'version': 1, + 'cli': {'config': 'context.yaml'}, + 'smoke': {'ledger_db': 'ledger.db'}, + 'cases': [{'name': 'agent_direct_execution', 'scenario': 'agent_direct_execution', 'expect': 'OK'}], + 'observe': {'health': {}}, + } + suite = Suite(profile, root / 'profile.json', root) + + def fail(suite, case, row): + row['cleanup_errors'] = ['Agent deletion returned HTTP 500'] + raise RuntimeError('Request outcome unknown') + + with patch('regression_local_cases.run_case', side_effect=fail): + result = run_smoke(suite) + suite.finish(result) + self.assertEqual(result['outcome'], 'FAIL') + self.assertEqual(result['cases'][0]['status'], 'unknown') + self.assertTrue(any('cleanup failed' in message for message in result['errors'])) + + def test_expected_failure_does_not_allow_extra_or_running_runs(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / 'ledger.db' + with sqlite3.connect(path) as connection: + connection.execute( + 'CREATE TABLE agent_run (id INTEGER, run_id TEXT, workspace_id TEXT, conversation_id TEXT, status TEXT)' + ) + connection.execute("INSERT INTO agent_run VALUES (1, 'owned', 'ws', 'session', 'failed')") + args = (path, 'ws', 0, {'session'}, {'owned'}) + self.assertFalse(reconcile_runs(*args)['matched']) + expected = {'owned': 'failed'} + self.assertTrue(reconcile_runs(*args, expected_statuses=expected)['matched']) + with sqlite3.connect(path) as connection: + connection.execute("UPDATE agent_run SET status='running'") + self.assertFalse(reconcile_runs(*args, expected_statuses=expected)['matched']) + with sqlite3.connect(path) as connection: + connection.execute("UPDATE agent_run SET status='failed'") + connection.execute("INSERT INTO agent_run VALUES (2, 'extra', 'ws', 'session', 'completed')") + result = reconcile_runs(*args, expected_statuses=expected) + self.assertFalse(result['matched']) + self.assertEqual(result['extra'], ['extra']) + with self.assertRaises(ValueError): + reconcile_runs(*args, expected_statuses={'owned': 'running'}) + + +if __name__ == '__main__': + unittest.main() diff --git a/tools/test_regression_resource_cases.py b/tools/test_regression_resource_cases.py new file mode 100644 index 000000000..3e039ee6b --- /dev/null +++ b/tools/test_regression_resource_cases.py @@ -0,0 +1,212 @@ +"""Challenge public resource lifecycle oracles and their cleanup without starting services.""" + +import json +from pathlib import Path +import tempfile +from types import SimpleNamespace +import unittest +from unittest.mock import patch + +from regression_local_cases import ScenarioUnavailable +from regression_resource_cases import run_case + + +def success(data=None): + return 200, {'code': 0, 'data': data or {}} + + +class ResourceCasesTests(unittest.TestCase): + def setUp(self): + temporary = tempfile.TemporaryDirectory() + self.addCleanup(temporary.cleanup) + self.root = Path(temporary.name) + self.suite = SimpleNamespace( + output=self.root, + env={'LANGBOT_QA_USER_TOKEN': 'private-user-token'}, + secrets=[], + variables={'workspace_uuid': 'workspace', 'resource_knowledge_base': 'owned-kb'}, + profile={ + 'resources': [ + { + 'id': 'chat_pipeline', + 'body': {'config': {'ai': {'runner': {'id': 'runner'}, 'runner_config': {'runner': {}}}}}, + } + ] + }, + write=lambda name, value: (self.root / name).write_text(json.dumps(value)), + ) + + def test_missing_management_session_skips_without_creating_credentials(self): + self.suite.env.clear() + with patch('regression_resource_cases.api') as request, self.assertRaises(ScenarioUnavailable): + run_case(self.suite, {'scenario': 'auth_revocation'}, {}) + request.assert_not_called() + + def test_revocation_rejects_still_authorized_key_and_never_persists_secrets(self): + for stale_authorization in (False, True): + with self.subTest(stale_authorization=stale_authorization): + key = {'id': 7, 'uuid': 'managed-key', 'status': 'active', 'secret_available': False} + row = {'after_id': 0} + + def request(suite, method, path, body=None, headers=None): + if method == 'POST': + return success({'key': {**key, 'key': 'private-child-secret', 'secret_available': True}}) + if method == 'DELETE': + key['status'] = 'revoked' + return success() + if path == '/api/v1/system/context': + self.assertEqual(headers, {'X-API-Key': 'private-child-secret'}) + return success({'workspace_uuid': 'workspace', 'api_key_id': 'managed-key'}) + if path == '/api/v1/pipelines': + return ( + success({'pipelines': []}) + if key['status'] == 'active' or stale_authorization + else (401, {'code': -1}) + ) + return success({'key': key}) + + with patch('regression_resource_cases.api', side_effect=request): + case = {'scenario': 'auth_revocation', 'expect': 'REVOKED'} + if stale_authorization: + with self.assertRaisesRegex(ValueError, 'still authorizes'): + run_case(self.suite, case, row) + self.assertEqual(row['assertions']['credential_revoked'], 'failed') + else: + self.assertEqual(run_case(self.suite, case, row)['status'], 'revoked') + self.assertEqual(key['status'], 'revoked') + self.assertEqual(row['assertions']['key_cleanup'], 'passed') + evidence = json.dumps(row) + (self.root / row['resource_journal']).read_text() + self.assertNotIn('private-child-secret', evidence) + self.assertNotIn('private-user-token', evidence) + + def test_direct_agent_records_run_and_cleans_after_failed_public_event_oracle(self): + for missing_event in (False, True): + with self.subTest(missing_event=missing_event): + state, row = {}, {'after_id': 12} + + def request(suite, method, path, body=None, headers=None): + if method == 'POST' and path == '/api/v1/agents': + state.update(uuid='own-agent', name=body['name']) + return success({'uuid': 'own-agent'}) + if path.endswith('/debug'): + state['conversation'] = body['conversation_id'] + return success({'conversation_id': state['conversation'], 'final_text': 'DIRECT_OK'}) + run = { + 'run_id': 'own-run', + 'conversation_id': state.get('conversation'), + 'status': 'completed', + 'finished_at': 1, + } + if path.endswith('/runs'): + return success({'items': [run]}) + if path.endswith('/events'): + return success( + { + 'run': run, + 'items': [] if missing_event else [{'type': 'run.completed', 'run_id': 'own-run'}], + } + ) + if method == 'DELETE': + self.assertEqual(path, '/api/v1/agents/own-agent') + state['deleted'] = True + return success() + return (404, {'code': -1}) if state.get('deleted') else success({'agent': state}) + + with patch('regression_resource_cases.api', side_effect=request): + case = {'scenario': 'agent_direct_execution', 'expect': 'DIRECT_OK'} + if missing_event: + with self.assertRaisesRegex(ValueError, 'completion event'): + run_case(self.suite, case, row) + else: + self.assertEqual(run_case(self.suite, case, row)['run_id'], 'own-run') + self.assertEqual( + row['scenario_runs'], + [ + { + 'run_id': 'own-run', + 'conversation_id': state['conversation'], + 'after_id': 12, + 'expected_status': 'completed', + } + ], + ) + self.assertTrue(state['deleted']) + self.assertEqual(row['assertions']['agent_cleanup'], 'passed') + + def test_unknown_agent_create_recovers_owned_identity_without_retrying_creation(self): + calls, state, row = [], {}, {'after_id': 0} + + def request(suite, method, path, body=None, headers=None): + calls.append((method, path)) + if method == 'POST': + state.update(uuid='own-agent', name=body['name'], kind='agent') + raise ConnectionResetError('creation response lost') + if path == '/api/v1/agents': + return success({'agents': [state, {'uuid': 'unrelated', 'name': 'other', 'kind': 'agent'}]}) + if method == 'DELETE': + self.assertEqual(path, '/api/v1/agents/own-agent') + state['deleted'] = True + return success() + return (404, {'code': -1}) if state.get('deleted') else success({'agent': state}) + + with patch('regression_resource_cases.api', side_effect=request), self.assertRaises(ConnectionResetError): + run_case(self.suite, {'scenario': 'agent_direct_execution', 'expect': 'OK'}, row) + self.assertEqual(sum(method == 'POST' for method, _ in calls), 1) + self.assertEqual(row['owned_resources'][0]['state'], 'deleted') + + def test_delete_requires_retrieval_absence_and_preserves_control_document(self): + for stale_vectors in (False, True): + with self.subTest(stale_vectors=stale_vectors): + documents, row = [], {'after_id': 0} + + def ingest(args): + content = Path(args[args.index('--file') + 1]).read_text() + index = len(documents) + documents.append( + { + 'uuid': f'file-{index}', + 'engine_document_id': f'engine-{index}', + 'status': 'completed', + 'content': content, + 'deleted': False, + } + ) + return {} + + def request(suite, method, path, body=None, headers=None): + if method == 'DELETE': + target = next(document for document in documents if path.endswith('/' + document['uuid'])) + target['deleted'] = True + return success() + if method == 'GET': + return success({'files': [document for document in documents if not document['deleted']]}) + return success( + { + 'results': [ + { + 'metadata': {'file_id': document['engine_document_id']}, + 'content': [{'type': 'text', 'text': document['content']}], + } + for document in documents + if not document['deleted'] or stale_vectors + ] + } + ) + + self.suite.call = ingest + with patch('regression_resource_cases.api', side_effect=request): + case = {'scenario': 'retrieval_delete_visibility', 'expect': 'DELETED'} + if stale_vectors: + with self.assertRaisesRegex(ValueError, 'remains visible'): + run_case(self.suite, case, row) + self.assertEqual(row['assertions']['retrieval_deleted'], 'failed') + else: + self.assertEqual(run_case(self.suite, case, row)['result'], 'DELETED') + self.assertEqual(row['assertions']['retrieval_control'], 'passed') + self.assertTrue(all(document['deleted'] for document in documents)) + self.assertEqual(row['assertions']['document_cleanup'], 'passed') + self.assertEqual(list(self.root.glob('resource-documents-*')), []) + + +if __name__ == '__main__': + unittest.main() From a7bc48f601b07ae06b551be074b4b0bf4ae34279 Mon Sep 17 00:00:00 2001 From: huanghuoguoguo <1051233107@qq.com> Date: Wed, 28 Oct 2026 16:20:15 +0800 Subject: [PATCH 2/3] test(regression): exercise local execution faults and cleanup (#2795) * test(regression): add selectable resource lifecycle cases * test(regression): exercise local execution faults and cleanup --- tools/mock_llm.py | 9 +- tools/mock_llm_scenarios.json | 100 ++++++ tools/regression_contracts.json | 12 +- tools/regression_execution_cases.py | 427 +++++++++++++++++++++++ tools/regression_local_cases.py | 5 +- tools/test_regression_execution_cases.py | 309 ++++++++++++++++ 6 files changed, 854 insertions(+), 8 deletions(-) create mode 100644 tools/regression_execution_cases.py create mode 100644 tools/test_regression_execution_cases.py diff --git a/tools/mock_llm.py b/tools/mock_llm.py index 29eba3668..8bd037279 100644 --- a/tools/mock_llm.py +++ b/tools/mock_llm.py @@ -124,6 +124,9 @@ def select_message(body, scenario): raise ValueError('stream must be a boolean') if body.get('stream_options') is not None and not isinstance(body['stream_options'], dict): raise ValueError('stream_options must be an object') + scope = fault_scope(body, False) if scenario.get('substitute_mock_scope') else None + if scenario.get('substitute_mock_scope') and scope is None: + raise ValueError('this scenario requires mock_scope or LB_MOCK_SCOPE for tool file isolation') # Select from the current turn's history, never a shared request counter. start = next( (i for i in range(len(messages) - 1, -1, -1) if messages[i].get('role') == 'user'), @@ -172,7 +175,9 @@ def select_message(body, scenario): 'type': 'function', 'function': { 'name': call['name'], - 'arguments': json.dumps(call['arguments']), + 'arguments': json.dumps(call['arguments']).replace('{mock_scope}', scope) + if scope is not None + else json.dumps(call['arguments']), }, } for i, call in enumerate(tool_calls) @@ -471,6 +476,8 @@ def validate_scenarios(scenarios): raise ValueError(f'{name}: scenario must be an object') if scenario.get('kind', 'chat') not in ('chat', 'embedding'): raise ValueError(f'{name}: kind must be chat or embedding') + if 'substitute_mock_scope' in scenario and not isinstance(scenario['substitute_mock_scope'], bool): + raise ValueError(f'{name}: substitute_mock_scope must be a boolean') steps = scenario.get('steps', []) if not isinstance(steps, list) or (not steps and scenario.get('kind') != 'embedding'): raise ValueError(f'{name}: steps must be a non-empty array') diff --git a/tools/mock_llm_scenarios.json b/tools/mock_llm_scenarios.json index f870a43b6..0f9aaf468 100644 --- a/tools/mock_llm_scenarios.json +++ b/tools/mock_llm_scenarios.json @@ -295,5 +295,105 @@ "kind": "embedding", "status": 503, "recover_after": 1 + }, + "mock-local-interrupt": { + "steps": [ + { + "tool_calls": [ + { + "name": "exec", + "arguments": { + "command": "test ! -e /tmp/langbot-{mock_scope}-ready && test ! -e /tmp/langbot-{mock_scope}-leak && printf SETUP_READY > /tmp/langbot-{mock_scope}-ready && cat /tmp/langbot-{mock_scope}-ready" + } + } + ] + }, + { + "expect_tool_output": "SETUP_READY", + "tool_calls": [ + { + "name": "exec", + "arguments": { + "command": "sleep 8; printf LATE_SIDE_EFFECT > /tmp/langbot-{mock_scope}-leak", + "timeout_sec": 20 + } + } + ] + }, + { + "content": "UNEXPECTED_INTERRUPT_COMPLETION" + } + ], + "substitute_mock_scope": true + }, + "mock-local-interrupt-check": { + "steps": [ + { + "tool_calls": [ + { + "name": "exec", + "arguments": { + "command": "sleep 9; if test ! -f /tmp/langbot-{mock_scope}-ready; then printf INTERRUPT_SETUP_MISSING; exit 41; fi; if test -e /tmp/langbot-{mock_scope}-leak; then printf INTERRUPT_LATE_SIDE_EFFECT; exit 42; fi; rm /tmp/langbot-{mock_scope}-ready && printf INTERRUPT_CLEANUP_OK", + "timeout_sec": 15 + } + } + ] + }, + { + "expect_tool_output": "INTERRUPT_CLEANUP_OK", + "content": "INTERRUPT_CLEANUP_OK" + } + ], + "substitute_mock_scope": true + }, + "mock-local-retry-side-effect": { + "status": 503, + "recover_after": 1, + "steps": [ + { + "tool_calls": [ + { + "name": "exec", + "arguments": { + "command": "printf 'once\\n' >> /tmp/langbot-{mock_scope}-effects; printf SIDE_EFFECT_WRITTEN" + } + } + ] + }, + { + "expect_tool_output": "SIDE_EFFECT_WRITTEN", + "tool_calls": [ + { + "name": "exec", + "arguments": { + "command": "test \"$(wc -l < /tmp/langbot-{mock_scope}-effects)\" -eq 1 && rm /tmp/langbot-{mock_scope}-effects && printf SIDE_EFFECT_ONCE" + } + } + ] + }, + { + "expect_tool_output": "SIDE_EFFECT_ONCE", + "content": "MODEL_RETRY_SIDE_EFFECT_ONCE" + } + ], + "substitute_mock_scope": true + }, + "mock-local-path-escape": { + "steps": [ + { + "tool_calls": [ + { + "name": "read", + "arguments": { + "path": "/workspace/../../etc/passwd" + } + } + ] + }, + { + "expect_tool_output": "Path escapes the workspace boundary", + "content": "PATH_ESCAPE_REJECTED" + } + ] } } diff --git a/tools/regression_contracts.json b/tools/regression_contracts.json index 8d0729e30..de7495d19 100644 --- a/tools/regression_contracts.json +++ b/tools/regression_contracts.json @@ -188,17 +188,17 @@ {"id": "execution.cancel_cleanup", "area": "execution", "kind": "blackbox", "priority": "P0", "description": "Cancelling a running execution reaches cancelled state and releases its work", - "bindings": [] + "bindings": [{"profile": "scenarios", "case": "execution.cancel_cleanup", "checks": ["run_status", "exec_active", "exec_stopped", "no_late_side_effect"]}] }, {"id": "execution.timeout_cleanup", "area": "execution", "kind": "blackbox", "priority": "P0", "description": "Deadline expiry stops child work and allows a subsequent request", - "bindings": [] + "bindings": [{"profile": "scenarios", "case": "execution.timeout_cleanup", "checks": ["run_status", "exec_active", "exec_stopped", "no_late_side_effect"]}] }, {"id": "execution.retry_side_effects", "area": "execution", "kind": "blackbox", "priority": "P0", "description": "Retries do not duplicate externally visible side effects", - "bindings": [] + "bindings": [{"profile": "scenarios", "case": "execution.retry_side_effects", "checks": ["run_status", "fault_injected", "side_effect_once"]}] }, {"id": "runtime.restart_recovery", "area": "runtime", "kind": "blackbox", "priority": "P0", @@ -213,17 +213,17 @@ {"id": "model.stream_disconnect", "area": "model", "kind": "blackbox", "priority": "P1", "description": "A partial model stream disconnect is handled without a false successful completion", - "bindings": [] + "bindings": [{"profile": "scenarios", "case": "model.stream_disconnect", "checks": ["run_status", "fault_injected", "no_fallback_after_partial", "recovered"]}] }, {"id": "model.exhausted_fallback", "area": "model", "kind": "blackbox", "priority": "P1", "description": "Failure of every configured model reaches a visible failed outcome", - "bindings": [] + "bindings": [{"profile": "scenarios", "case": "model.exhausted_fallback", "checks": ["run_status", "fallback_exhausted", "recovered"]}] }, {"id": "tools.path_escape", "area": "tools", "kind": "blackbox", "priority": "P0", "description": "Box operations reject access outside their authorized filesystem scope", - "bindings": [] + "bindings": [{"profile": "scenarios", "case": "tools.path_escape", "checks": ["run_status", "path_rejected", "no_file_content"]}] }, {"id": "tools.timeout_recovery", "area": "tools", "kind": "blackbox", "priority": "P1", diff --git a/tools/regression_execution_cases.py b/tools/regression_execution_cases.py new file mode 100644 index 000000000..afe666d0b --- /dev/null +++ b/tools/regression_execution_cases.py @@ -0,0 +1,427 @@ +"""Local Agent faults with public API, mock-wire, and independent Run evidence.""" + +import copy +import hashlib +import json +from pathlib import Path +import time +import sys +import urllib.request +from uuid import uuid4 + +from regression_coverage import checked +from regression_local_cases import api, endpoint, headers, record_run, require + +GROUP = 'faults' +RUNNER = 'plugin:langbot-team/LocalAgent/default' +CASE_NAMES = { + 'execution.cancel_cleanup', + 'execution.timeout_cleanup', + 'execution.retry_side_effects', + 'model.stream_disconnect', + 'model.exhausted_fallback', + 'tools.path_escape', +} +MODELS = { + 'interrupt': 'mock-local-interrupt', + 'interrupt_check': 'mock-local-interrupt-check', + 'retry': 'mock-local-retry-side-effect', + 'disconnect': 'mock-stream-disconnect', + 'escape': 'mock-local-path-escape', +} + + +def add_cases(profile): + template = next(item for item in profile['resources'] if item['id'] == 'chat_model') + for name, model in MODELS.items(): + item = copy.deepcopy(template) + item['id'] = 'fault_' + name + '_model' + item['body']['name'] = model + item['body']['extra_args'] = {'request_policy': {'retries': 1 if name == 'retry' else 0}} + profile['resources'].append(item) + for name in sorted(CASE_NAMES): + profile['cases'].append( + {'name': name, 'scenario': name, 'stage': name.split('.')[0], 'expect': 'FAULT_CASE_OK'} + ) + + +def _data(suite, method, path, body=None): + status, payload = api(suite, method, path, body) + require( + status == 200 and isinstance(payload, dict) and payload.get('code') == 0, + f'{method} {path} returned HTTP {status}: {payload}', + ) + return payload.get('data') + + +def _poll(suite, read, message, seconds=20): + deadline = time.monotonic() + suite.timeout(seconds) + while True: + value = read() + if value: + return value + if time.monotonic() >= deadline: + raise TimeoutError(message) + time.sleep(0.1) + + +def _agent_config(suite, scope, model, *, box=False, timeout=30, fallbacks=()): + return { + 'runner': {'id': RUNNER}, + 'allowed_tools': ['exec', 'read'] if box else [], + 'runner_config': { + RUNNER: { + 'model': {'primary': suite.variables[model], 'fallbacks': [suite.variables[key] for key in fallbacks]}, + 'prompt': [{'role': 'system', 'content': 'Execute the deterministic QA scenario.'}], + 'knowledge-bases': [], + 'box-enabled': box, + 'box-session-id-template': scope, + 'timeout': timeout, + 'max-tool-iterations': 5, + 'tool-execution-mode': 'serial', + } + }, + } + + +def _sessions(suite): + result = _data(suite, 'GET', '/api/v1/box/sessions') + require(isinstance(result, list), 'Box sessions must be an array') + return {item['session_id']: item for item in result} + + +def _running_box(suite, session_id): + def read(): + return session_id if _sessions(suite).get(session_id, {}).get('status') == 'running' else None + + return _poll(suite, read, 'The long exec never became active in its Box') + + +def _stream(suite, agent, scope, row, *, interrupt=None, box_session_id=None, frame_key='frames'): + request = urllib.request.Request( + endpoint(suite, f'/api/v1/agents/{agent}/debug/stream'), + data=json.dumps({'text': f'Run the QA scenario LB_MOCK_SCOPE={scope}', 'conversation_id': scope}).encode(), + headers={**headers(suite), 'Content-Type': 'application/json'}, + ) + opener = urllib.request.build_opener(urllib.request.ProxyHandler({})) + frames = [] + row[frame_key] = frames + deadline = time.monotonic() + suite.timeout(45) + with opener.open(request, timeout=suite.timeout(45)) as response: + require(response.status == 200, 'Agent debug stream was not accepted') + while time.monotonic() < deadline: + line = response.readline(256 * 1024 + 1) + require(len(line) <= 256 * 1024, 'Agent debug frame exceeded its limit') + if not line: + break + if not line.strip(): + continue + text = line.decode('utf-8') + for secret in suite.secrets: + text = text.replace(secret, '[REDACTED]') + frame = json.loads(text) + require(isinstance(frame, dict), 'Agent debug frame must be an object') + frames.append(frame) + if interrupt and frame.get('kind') == 'result': + event = frame.get('data', {}) + if event.get('type') == 'tool.call.started' and event.get('data', {}).get('parameters', {}).get( + 'command', '' + ).startswith('sleep 8;'): + row['box_session_id'] = checked(row, 'exec_active', lambda: _running_box(suite, box_session_id)) + if interrupt == 'cancel': + break # Closing this public debug stream cancels its run. + if frame.get('kind') in {'completed', 'error'}: + return frames + else: + raise TimeoutError('Agent debug stream exceeded the observation deadline; not retried') + if interrupt == 'cancel': + require('box_session_id' in row, 'Cancellation did not reach the active long exec') + else: + raise RuntimeError('Agent debug stream ended without a terminal frame; outcome unknown') + return frames + + +def _terminal(suite, agent, conversation, row, expected, excluded=()): + def read(): + runs = _data(suite, 'GET', f'/api/v1/agents/{agent}/runs')['items'] + matches = [run for run in runs if run['conversation_id'] == conversation and run['run_id'] not in excluded] + require(len(matches) <= 1, 'One scenario submission created multiple AgentRuns') + return matches[0] if matches and matches[0].get('finished_at') else None + + run = _poll(suite, read, 'AgentRun did not terminate after the scenario') + try: + checked( + row, + 'run_status', + lambda: require(run['status'] == expected, f'Public Run status is {run["status"]}; expected {expected}'), + ) + finally: + checked(row, 'ledger_status', lambda: record_run(suite, row, run['run_id'], expected)) + detail = _data(suite, 'GET', f'/api/v1/agents/{agent}/runs/{run["run_id"]}/events') + require(not detail.get('has_more'), 'Scenario trace exceeds the bounded event page') + return run, detail['items'] + + +def _mock_rows(suite, scope, count): + path = Path(suite.path(suite.profile['smoke']['mock_records'])) + + def read(): + rows = [] + for line in path.read_text().splitlines(): + try: + item = json.loads(line) + except ValueError: + continue # A concurrent handler may still be appending its last line. + if item.get('scope') == scope: + rows.append(item) + return rows if len(rows) >= count else None + + return _poll(suite, read, 'The expected model request was not recorded', seconds=5) + + +def _tool_events(events, name): + return [ + item['data'] + for item in events + if item['type'] == 'tool.call.completed' and item['data'].get('tool_name') == name + ] + + +def _cleanup(suite, agent, scope, row, expected): + original = sys.exception() + errors = [] + try: + runs = _data(suite, 'GET', f'/api/v1/agents/{agent}/runs')['items'] + registered = {item['run_id'] for item in row.get('scenario_runs', [])} + pending = [run for run in runs if run['run_id'] not in registered] + if not registered and len(pending) == 1: + record_run(suite, row, pending[0]['run_id'], expected) + elif pending: + raise ValueError('Scenario has unexpected or unregistered AgentRuns at cleanup') + except (ValueError, OSError, RuntimeError) as exc: + errors.append(f'Run closeout: {exc}') + try: + _data(suite, 'DELETE', f'/api/v1/agents/{agent}') + status, _ = api(suite, 'GET', f'/api/v1/agents/{agent}') + require(status == 404, 'Deleted scenario Agent remains visible') + except (ValueError, OSError, RuntimeError) as exc: + errors.append(f'Agent cleanup: {exc}') + if row['case'] == 'execution.retry_side_effects': + try: + url = suite.variables['mock_base_url'].removesuffix('/v1').rstrip('/') + '/_mock/scopes/' + scope + request = urllib.request.Request(url, method='DELETE') + opener = urllib.request.build_opener(urllib.request.ProxyHandler({})) + with opener.open(request, timeout=suite.timeout(10)) as response: + payload = json.load(response) + require( + response.status == 200 and payload.get('scope') == scope and payload.get('released') in (0, 1), + 'Mock recovery scope cleanup failed', + ) + except (ValueError, OSError, RuntimeError) as exc: + errors.append(f'Mock scope cleanup: {exc}') + row['cleanup_errors'] = errors + if errors: + raise RuntimeError('; '.join(([str(original)] if original else []) + errors)) + + +def run_case(suite, case, row): + scenario = case['scenario'] + require(scenario in CASE_NAMES, 'Unknown local execution scenario') + scope = 'qa-fault-' + uuid4().hex + box_key = 'qa-fault-' + hashlib.sha256((str(suite.output.resolve()) + scenario).encode()).hexdigest()[:24] + # Core's RunnerBoxService maps the requested reuse key to this logical ID. + box_session_id = 'runner-' + hashlib.sha256(box_key.encode()).hexdigest() + interrupt = scenario in {'execution.cancel_cleanup', 'execution.timeout_cleanup'} + model = { + 'execution.cancel_cleanup': 'fault_interrupt_model', + 'execution.timeout_cleanup': 'fault_interrupt_model', + 'execution.retry_side_effects': 'fault_retry_model', + 'model.stream_disconnect': 'fault_disconnect_model', + 'model.exhausted_fallback': 'error503_model', + 'tools.path_escape': 'fault_escape_model', + }[scenario] + fallbacks = ( + ('error429_model',) + if scenario == 'model.exhausted_fallback' + else ('chat_model',) + if scenario == 'model.stream_disconnect' + else () + ) + config = _agent_config( + suite, + box_key, + model, + box=interrupt or scenario in {'execution.retry_side_effects', 'tools.path_escape'}, + timeout=5 if scenario == 'execution.timeout_cleanup' else 30, + fallbacks=fallbacks, + ) + expected = ( + 'cancelled' + if scenario == 'execution.cancel_cleanup' + else 'timeout' + if interrupt + else 'failed' + if scenario.startswith('model.') + else 'completed' + ) + created = _data(suite, 'POST', '/api/v1/agents', {'name': scope, 'config': config}) + agent = created['uuid'] + try: + row['agent_uuid'], row['conversation_id'] = agent, scope + row['box_reuse_key'] = box_key + frames = _stream( + suite, + agent, + scope, + row, + interrupt='cancel' if scenario == 'execution.cancel_cleanup' else 'timeout' if interrupt else None, + box_session_id=box_session_id, + ) + row['frames'] = frames + run, events = _terminal(suite, agent, scope, row, expected) + row['run_events'] = events + request_count = ( + 4 if scenario == 'execution.retry_side_effects' else 1 if scenario == 'model.stream_disconnect' else 2 + ) + requests = _mock_rows(suite, scope, request_count) + row['model_requests'] = [ + {'model': item['request']['model'], 'status': item['status'], 'faults': item.get('faults', [])} + for item in requests + ] + if interrupt: + checked( + row, + 'exec_active', + lambda: require('box_session_id' in row, 'No active exec was observed before interruption'), + ) + checked( + row, + 'exec_stopped', + lambda: _poll( + suite, + lambda: _sessions(suite).get(row['box_session_id'], {}).get('status') == 'idle', + 'Interrupted Box exec remains active', + ), + ) + config = _agent_config(suite, box_key, 'fault_interrupt_check_model', box=True) + _data(suite, 'PUT', f'/api/v1/agents/{agent}', {'config': config}) + recovery = _stream(suite, agent, scope, row, frame_key='recovery_frames') + calls = _tool_events([frame['data'] for frame in recovery if frame.get('kind') == 'result'], 'exec') + row['recovery_exec_results'] = calls + try: + recovery_run, recovery_events = _terminal( + suite, agent, scope, row, 'completed', excluded={run['run_id']} + ) + row['recovery_run_id'], row['recovery_events'] = recovery_run['run_id'], recovery_events + finally: + # The tool's evidence remains authoritative if its failed check also makes the model fail. + checked( + row, + 'no_late_side_effect', + lambda: require( + len(calls) == 1 + and (calls[0].get('result') or {}).get('exit_code') == 0 + and calls[0]['result'].get('stdout') == 'INTERRUPT_CLEANUP_OK', + 'Interrupted shell cleanup failed: ' + json.dumps(calls, ensure_ascii=False), + ), + ) + elif scenario == 'execution.retry_side_effects': + checked( + row, + 'fault_injected', + lambda: require( + [item['status'] for item in requests] == [503, 200, 200, 200] + and all(item['request']['model'] == 'mock-local-retry-side-effect' for item in requests), + 'Expected one 503 retry followed by tool execution and completion', + ), + ) + calls = _tool_events(events, 'exec') + checked( + row, + 'side_effect_once', + lambda: require( + len(calls) == 2 + and all(call.get('result', {}).get('exit_code') == 0 for call in calls) + and [call['result'].get('stdout') for call in calls] == ['SIDE_EFFECT_WRITTEN', 'SIDE_EFFECT_ONCE'], + 'Model retry duplicated or lost the tool side effect', + ), + ) + elif scenario == 'tools.path_escape': + calls = _tool_events(events, 'read') + checked( + row, + 'path_rejected', + lambda: require( + len(calls) == 1 and 'Path escapes the workspace boundary' in str(calls[0].get('error')), + 'The real Host tool did not reject the path escape', + ), + ) + checked( + row, + 'no_file_content', + lambda: require('root:' not in json.dumps(frames), 'Rejected file content appeared in Agent output'), + ) + else: + if scenario == 'model.stream_disconnect': + checked( + row, + 'fault_injected', + lambda: require( + len(requests) == 1 + and requests[0].get('injected_disconnect') is True + and not requests[0]['response_complete'], + 'The stream disconnect was not observed', + ), + ) + checked( + row, + 'no_fallback_after_partial', + lambda: require( + requests[0]['request']['model'] == 'mock-stream-disconnect' + and any( + item.get('data', {}).get('type') == 'message.delta' + and item['data'].get('data', {}).get('chunk', {}).get('content') + for item in frames + ), + 'The model did not commit partial output before failing', + ), + ) + else: + checked( + row, + 'fallback_exhausted', + lambda: require( + [(item['request']['model'], item['status']) for item in requests] + == [('mock-503', 503), ('mock-429', 429)], + 'The configured fallback sequence was not exhausted', + ), + ) + config = _agent_config(suite, scope, 'chat_model') + _data(suite, 'PUT', f'/api/v1/agents/{agent}', {'config': config}) + recovery = _stream(suite, agent, scope, row, frame_key='recovery_frames') + _terminal(suite, agent, scope, row, 'completed', excluded={run['run_id']}) + checked( + row, + 'recovered', + lambda: require( + any( + item.get('kind') == 'completed' and item.get('data', {}).get('final_text') == 'LANGBOT_CHAT_OK' + for item in recovery + ), + 'A healthy model did not recover after the failed run', + ), + ) + if config['runner_config'][RUNNER]['box-enabled']: + checked( + row, + 'box_idle', + lambda: _poll( + suite, + lambda: _sessions(suite).get(box_session_id, {}).get('status') == 'idle', + 'The scenario Box did not return to idle', + ), + ) + row['box_session_id'] = box_session_id + return {'result': 'FAULT_CASE_OK', 'scenario': scenario, 'run_id': run['run_id']} + finally: + _cleanup(suite, agent, scope, row, expected) diff --git a/tools/regression_local_cases.py b/tools/regression_local_cases.py index 951360e50..0436d68f2 100644 --- a/tools/regression_local_cases.py +++ b/tools/regression_local_cases.py @@ -9,7 +9,10 @@ import time import urllib.error import urllib.request -GROUPS = {'resources': 'regression_resource_cases'} +GROUPS = { + 'faults': 'regression_execution_cases', + 'resources': 'regression_resource_cases', +} class ScenarioUnavailable(RuntimeError): diff --git a/tools/test_regression_execution_cases.py b/tools/test_regression_execution_cases.py new file mode 100644 index 000000000..c42dda48c --- /dev/null +++ b/tools/test_regression_execution_cases.py @@ -0,0 +1,309 @@ +"""Challenge fault oracles and their cleanup; exercise NDJSON over real HTTP.""" + +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +import hashlib +import json +from pathlib import Path +import sqlite3 +import subprocess +import tempfile +import threading +from types import SimpleNamespace +import unittest +from unittest.mock import patch +from urllib.error import HTTPError +from urllib.request import Request, urlopen + +import regression_execution_cases as cases +from mock_llm import MockServer + + +class ExecutionCasesTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.root = Path(self.tmp.name) + self.ledger = self.root / 'ledger.db' + with sqlite3.connect(self.ledger) as db: + db.execute( + 'CREATE TABLE agent_run(id INTEGER, run_id TEXT, conversation_id TEXT, status TEXT, finished_at TEXT, workspace_id TEXT)' + ) + self.suite = SimpleNamespace( + output=self.root, + variables={ + **{f'fault_{name}_model': name for name in cases.MODELS}, + 'chat_model': 'chat', + 'error503_model': '503', + 'error429_model': '429', + 'workspace_uuid': 'workspace', + 'mock_base_url': 'http://unused.invalid/v1', + }, + profile={'smoke': {'ledger_db': str(self.ledger), 'mock_records': str(self.root / 'mock.jsonl')}}, + path=lambda value: value, + timeout=lambda maximum: maximum, + secrets=[], + connection={'endpoint': 'http://unused.invalid', 'credential_source': 'env:TEST_KEY'}, + env={'TEST_KEY': 'test-only'}, + ) + + def run_fixture( + self, scenario, *, tool_ok=True, fault=True, primary_status=None, recovery_status='completed', delete_ok=True + ): + expected = { + 'execution.cancel_cleanup': 'cancelled', + 'execution.timeout_cleanup': 'timeout', + 'execution.retry_side_effects': 'completed', + 'tools.path_escape': 'completed', + 'model.stream_disconnect': 'failed', + 'model.exhausted_fallback': 'failed', + }[scenario] + row = {'case': scenario, 'after_id': 0} + self.row = row + self.deleted = False + runs, events = [], {} + statuses = { + 'execution.retry_side_effects': [503, 200, 200, 200], + 'model.stream_disconnect': [200], + 'model.exhausted_fallback': [503, 429], + }.get(scenario, [200, 200]) + models = { + 'execution.retry_side_effects': ['mock-local-retry-side-effect'] * 4, + 'model.stream_disconnect': ['mock-stream-disconnect'], + 'model.exhausted_fallback': ['mock-503', 'mock-429'], + }.get(scenario, ['fixture'] * 2) + requests = [ + { + 'scope': 'qa-fault-fixture', + 'status': status if fault else 200, + 'request': {'model': model}, + 'injected_disconnect': fault, + 'response_complete': not fault, + } + for status, model in zip(statuses, models) + ] + (self.root / 'mock.jsonl').write_text(''.join(json.dumps(item) + '\n' for item in requests)) + box_key = 'qa-fault-' + hashlib.sha256((str(self.root.resolve()) + scenario).encode()).hexdigest()[:24] + box_session_id = 'runner-' + hashlib.sha256(box_key.encode()).hexdigest() + + def api(_suite, method, path, body=None): + if method == 'DELETE': + self.deleted = True + return (200, {'code': 0}) if delete_ok else (500, {'code': 'delete_failed'}) + if method == 'GET' and path == '/api/v1/agents/agent': + return 404, {'code': -1} + if path == '/api/v1/box/sessions': + return 200, {'code': 0, 'data': [{'session_id': box_session_id, 'status': 'idle'}]} + if path.endswith('/events'): + run_id = path.split('/')[-2] + return 200, {'code': 0, 'data': {'items': events[run_id], 'has_more': False}} + if path.endswith('/runs'): + return 200, {'code': 0, 'data': {'items': list(runs)}} + return 200, {'code': 0, 'data': {'uuid': 'agent'}} + + def stream(_suite, _agent, scope, _row, **kwargs): + index = len(runs) + 1 + run_id = f'run-{index}' + status = (primary_status or expected) if index == 1 else recovery_status + run = {'run_id': run_id, 'conversation_id': scope, 'status': status, 'finished_at': 'now'} + runs.append(run) + with sqlite3.connect(self.ledger) as db: + db.execute( + 'INSERT INTO agent_run VALUES(?,?,?,?,?,?)', (index, run_id, scope, status, 'now', 'workspace') + ) + tool = 'read' if scenario == 'tools.path_escape' else 'exec' + stdout = 'INTERRUPT_CLEANUP_OK' if index > 1 else 'SIDE_EFFECT_ONCE' + result = {'exit_code': 0 if tool_ok else 1, 'stdout': stdout if tool_ok else 'DUPLICATE_OR_LEAK'} + if not tool_ok and index > 1: + result = {'exit_code': 42, 'stdout': 'INTERRUPT_LATE_SIDE_EFFECT', 'stderr': ''} + events[run_id] = [ + { + 'type': 'tool.call.completed', + 'data': { + 'tool_name': tool, + 'result': result, + 'error': 'Path escapes the workspace boundary.' if tool == 'read' else None, + }, + } + ] + if scenario == 'execution.retry_side_effects': + events[run_id].insert( + 0, + { + 'type': 'tool.call.completed', + 'data': { + 'tool_name': 'exec', + 'result': {'exit_code': 0, 'stdout': 'SIDE_EFFECT_WRITTEN'}, + }, + }, + ) + if kwargs.get('interrupt'): + row['box_session_id'] = box_session_id + return [ + *[{'kind': 'result', 'data': event} for event in events[run_id]], + {'kind': 'result', 'data': {'type': 'message.delta', 'data': {'chunk': {'content': 'LANGBOT_'}}}}, + {'kind': 'completed', 'data': {'final_text': 'LANGBOT_CHAT_OK'}}, + ] + + class Released: + status = 200 + + def __enter__(self): + from io import BytesIO + + self.read = BytesIO(b'{"scope":"qa-fault-fixture","released":1}').read + return self + + def __exit__(self, *_): + pass + + with ( + patch.object(cases, 'api', side_effect=api), + patch.object(cases, '_stream', side_effect=stream), + patch.object(cases, 'uuid4', return_value=SimpleNamespace(hex='fixture')), + patch.object(cases.urllib.request.OpenerDirector, 'open', return_value=Released()), + ): + return cases.run_case(self.suite, {'scenario': scenario}, row) + + def test_fault_cases_preserve_expected_terminal_runs_and_delete_agent(self): + for scenario in sorted(cases.CASE_NAMES): + with self.subTest(scenario=scenario): + with sqlite3.connect(self.ledger) as db: + db.execute('DELETE FROM agent_run') + result = self.run_fixture(scenario) + self.assertEqual(result['result'], 'FAULT_CASE_OK') + self.assertTrue(self.deleted) + self.assertTrue(self.row['scenario_runs']) + self.assertEqual(self.row['assertions']['run_status'], 'passed') + self.assertEqual(self.row['cleanup_errors'], []) + + def test_duplicate_side_effect_and_untriggered_fault_fail_without_replay(self): + for tool_ok, fault, assertion in ((False, True, 'side_effect_once'), (True, False, 'fault_injected')): + with self.subTest(assertion=assertion): + with sqlite3.connect(self.ledger) as db: + db.execute('DELETE FROM agent_run') + with self.assertRaises(ValueError): + self.run_fixture('execution.retry_side_effects', tool_ok=tool_ok, fault=fault) + self.assertEqual(self.row['assertions'][assertion], 'failed') + self.assertTrue(self.deleted) + self.assertEqual(len(self.row['scenario_runs']), 1) + + def test_late_side_effect_is_a_failure_even_when_run_is_cancelled(self): + with self.assertRaisesRegex(ValueError, 'INTERRUPT_LATE_SIDE_EFFECT'): + self.run_fixture('execution.cancel_cleanup', tool_ok=False, recovery_status='failed') + self.assertEqual(self.row['assertions']['no_late_side_effect'], 'failed') + self.assertEqual([run['status'] for run in self.row['scenario_runs']], ['cancelled', 'failed']) + self.assertEqual(self.row['recovery_exec_results'][0]['result']['exit_code'], 42) + self.assertTrue(self.deleted) + + def test_recovery_command_distinguishes_missing_setup_and_late_side_effect(self): + scenarios = json.loads(Path(cases.__file__).with_name('mock_llm_scenarios.json').read_text()) + command = scenarios['mock-local-interrupt-check']['steps'][0]['tool_calls'][0]['arguments']['command'] + command = command.removeprefix('sleep 9; ').replace('/tmp/langbot-{mock_scope}', str(self.root / 'fixture')) + ready, leaked = self.root / 'fixture-ready', self.root / 'fixture-leak' + for setup, leak, status, output in ( + (False, False, 41, 'INTERRUPT_SETUP_MISSING'), + (True, True, 42, 'INTERRUPT_LATE_SIDE_EFFECT'), + (True, False, 0, 'INTERRUPT_CLEANUP_OK'), + ): + with self.subTest(setup=setup, leak=leak): + if setup: + ready.write_text('SETUP_READY') + if leak: + leaked.write_text('LATE_SIDE_EFFECT') + else: + leaked.unlink(missing_ok=True) + result = subprocess.run(['sh', '-c', command], capture_output=True, text=True, timeout=2) + self.assertEqual((result.returncode, result.stdout), (status, output)) + self.assertEqual(ready.exists(), setup and status != 0) + + def test_unexpected_run_status_is_recorded_and_cleanup_failure_keeps_original(self): + with self.assertRaisesRegex(RuntimeError, 'Run ended as completed.*Agent cleanup'): + self.run_fixture('model.exhausted_fallback', primary_status='completed', delete_ok=False) + self.assertEqual(self.row['scenario_runs'][0]['status'], 'completed') + self.assertEqual(self.row['scenario_runs'][0]['expected_status'], 'failed') + self.assertTrue(self.deleted) + + def test_real_ndjson_transport_and_cancel_close(self): + requests = [] + frames = [ + { + 'kind': 'result', + 'data': {'type': 'tool.call.started', 'data': {'parameters': {'command': 'sleep 8; printf late'}}}, + }, + {'kind': 'completed', 'data': {'final_text': 'done'}}, + ] + + class Handler(BaseHTTPRequestHandler): + def do_POST(self): + requests.append(json.loads(self.rfile.read(int(self.headers['Content-Length'])))) + self.send_response(200) + self.send_header('Content-Type', 'application/x-ndjson') + self.end_headers() + for frame in frames: + self.wfile.write(json.dumps(frame).encode() + b'\n') + self.wfile.flush() + + def log_message(self, *_): + pass + + server = ThreadingHTTPServer(('127.0.0.1', 0), Handler) + thread = threading.Thread(target=server.serve_forever) + thread.start() + self.suite.connection['endpoint'] = f'http://127.0.0.1:{server.server_port}' + try: + self.assertEqual(cases._stream(self.suite, 'agent', 'scope', {})[-1]['kind'], 'completed') + row = {} + with patch.object(cases, '_running_box', return_value='box'): + cancelled = cases._stream(self.suite, 'agent', 'scope', row, interrupt='cancel') + self.assertEqual(len(cancelled), 1) + self.assertEqual(row['box_session_id'], 'box') + frames.pop() + with self.assertRaisesRegex(RuntimeError, 'outcome unknown'): + cases._stream(self.suite, 'agent', 'scope', {}) + self.assertEqual(len(requests), 3) + finally: + server.shutdown() + server.server_close() + thread.join() + + def test_real_mock_http_scopes_isolate_tool_paths_and_reject_unsafe_substitution(self): + scenarios = json.loads(Path(cases.__file__).with_name('mock_llm_scenarios.json').read_text()) + with MockServer(('127.0.0.1', 0), scenarios) as server: + thread = threading.Thread(target=server.serve_forever) + thread.start() + + def request(scope): + body = { + 'model': 'mock-local-interrupt', + 'mock_scope': scope, + 'messages': [{'role': 'user', 'content': 'test'}], + 'tools': [{'type': 'function', 'function': {'name': 'exec'}}], + } + with urlopen( + Request( + f'http://127.0.0.1:{server.server_port}/v1/chat/completions', data=json.dumps(body).encode() + ), + timeout=5, + ) as response: + return json.load(response) + + try: + commands = [] + for scope in ('run-a', 'run-b'): + call = request(scope)['choices'][0]['message']['tool_calls'][0] + command = json.loads(call['function']['arguments'])['command'] + self.assertIn(f'/tmp/langbot-{scope}-ready', command) + self.assertNotIn('{mock_scope}', command) + commands.append(command) + self.assertNotEqual(commands[0], commands[1]) + for invalid in (None, 'unsafe;echo', '../../escape'): + with self.subTest(scope=invalid), self.assertRaises(HTTPError) as caught: + request(invalid) + self.assertEqual(caught.exception.code, 400) + finally: + server.shutdown() + thread.join() + + +if __name__ == '__main__': + unittest.main() From d099033d79ee5d09e69bf6a0f870b639eac1806f Mon Sep 17 00:00:00 2001 From: huanghuoguoguo <1051233107@qq.com> Date: Wed, 28 Oct 2026 16:23:46 +0800 Subject: [PATCH 3/3] test(regression): verify callback retries and owned service restarts (#2796) * test(regression): add selectable resource lifecycle cases * test(regression): exercise local execution faults and cleanup * test(regression): verify callback retries and owned service restarts --- tools/regression_contracts.json | 5 +- tools/regression_lifecycle_cases.py | 497 +++++++++++++++++++++++ tools/regression_local_cases.py | 1 + tools/test_regression_lifecycle_cases.py | 292 +++++++++++++ 4 files changed, 793 insertions(+), 2 deletions(-) create mode 100644 tools/regression_lifecycle_cases.py create mode 100644 tools/test_regression_lifecycle_cases.py diff --git a/tools/regression_contracts.json b/tools/regression_contracts.json index de7495d19..a12488748 100644 --- a/tools/regression_contracts.json +++ b/tools/regression_contracts.json @@ -203,12 +203,12 @@ {"id": "runtime.restart_recovery", "area": "runtime", "kind": "blackbox", "priority": "P0", "description": "A runtime restart produces a known outcome and subsequent execution works", - "bindings": [] + "bindings": [{"profile": "scenarios", "case": "runtime_restart_recovery", "checks": ["process_replaced", "no_false_success", "interrupted_terminal", "response_after_restart"]}] }, {"id": "storage.restart_history", "area": "storage", "kind": "blackbox", "priority": "P1", "description": "Conversation history remains usable across a Core restart", - "bindings": [] + "bindings": [{"profile": "scenarios", "case": "core_restart_history", "checks": ["before_restart", "process_replaced", "history_after_restart"]}] }, {"id": "model.stream_disconnect", "area": "model", "kind": "blackbox", "priority": "P1", @@ -266,6 +266,7 @@ "bindings": [] }, {"id": "retrieval.delete_visibility", "area": "retrieval", "kind": "blackbox", "priority": "P1", "description": "Deleting one ingested document removes its retrieval results while preserving a second document", "bindings": [{"profile": "scenarios", "case": "retrieval_delete_visibility", "checks": ["retrieval_initial", "retrieval_deleted", "retrieval_control", "document_cleanup"]}]}, + {"id": "delivery.callback_retry_identity", "area": "delivery", "kind": "blackbox", "priority": "P1", "description": "A 5xx callback retry preserves body and delivery identity, completes delivery, and permits a subsequent message", "bindings": [{"profile": "scenarios", "case": "callback_retry_identity", "checks": ["callback_retry", "callback_followup", "callback_cleanup"]}]}, {"id": "execution.persisted_terminal", "area": "execution", "kind": "graybox", "priority": "P0", "description": "Each executed request has a unique completed persisted run and terminal event", diff --git a/tools/regression_lifecycle_cases.py b/tools/regression_lifecycle_cases.py new file mode 100644 index 000000000..e08c3e7fd --- /dev/null +++ b/tools/regression_lifecycle_cases.py @@ -0,0 +1,497 @@ +"""Optional lifecycle checks using real asynchronous HTTP Bot callbacks.""" + +from contextlib import closing +import copy +import hashlib +import hmac +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +import json +from pathlib import Path +import secrets +import sqlite3 +import threading +import time + +from regression_coverage import checked +from regression_local_cases import ScenarioUnavailable, api, record_run, require + +GROUP = 'lifecycle' +CASE_NAMES = {'callback_retry_identity'} + + +def add_cases(profile): + profile['cases'].append( + { + 'name': 'callback_retry_identity', + 'scenario': 'callback_retry_identity', + 'stage': 'delivery', + 'expect': 'LANGBOT_CALLBACK_RETRY_OK', + } + ) + + +def signature(secret, body, timestamp): + return 'sha256=' + hmac.new(secret.encode(), timestamp.encode() + b'.' + body, hashlib.sha256).hexdigest() + + +def ledger(suite): + path = Path(suite.path(suite.profile['smoke']['ledger_db'])).resolve() + connection = sqlite3.connect(path.as_uri() + '?mode=ro', uri=True, timeout=min(5, suite.timeout(5))) + connection.row_factory = sqlite3.Row + return connection + + +def after_id(suite): + with closing(ledger(suite)) as connection: + return connection.execute('SELECT COALESCE(MAX(id),0) FROM agent_run').fetchone()[0] + + +def completed(suite, row, session_id, after, expected='completed'): + from regression_smoke import completed_run + + with closing(ledger(suite)) as connection: + run = completed_run( + connection, suite.variables['workspace_uuid'], 'group_' + session_id, after, suite.timeout(20) + ) + record_run(suite, row, run['run_id'], expected) + return run + + +class CallbackReceiver: + """Reject the first callback once, then retain receipts for independent checks.""" + + def __init__(self, secret): + self.records = [] + self.condition = threading.Condition() + receiver = self + + class Handler(BaseHTTPRequestHandler): + def setup(self): + super().setup() + self.connection.settimeout(5) + + def do_POST(self): + try: + size = int(self.headers.get('Content-Length', '0')) + require(0 < size <= 1024 * 1024, 'Invalid callback body size') + body = self.rfile.read(size) + payload = json.loads(body) + timestamp = self.headers.get('X-LB-Timestamp', '') + require(abs(time.time() - int(timestamp)) <= 300, 'Invalid callback timestamp') + require( + hmac.compare_digest(self.headers.get('X-LB-Signature', ''), signature(secret, body, timestamp)), + 'Invalid callback signature', + ) + require(isinstance(payload, dict), 'Callback must be an object') + record = {'body': body, 'payload': payload} + with receiver.condition: + status = 503 if not receiver.records else 200 + except (ValueError, OSError) as exc: + record, status = {'error': str(exc)}, 400 + self.send_response(status) + self.send_header('Content-Type', 'application/json') + self.end_headers() + self.wfile.write(b'{}') + self.wfile.flush() + with receiver.condition: + receiver.records.append({**record, 'status': status}) + receiver.condition.notify_all() + + def log_message(self, *args): + pass + + self.server = ThreadingHTTPServer(('127.0.0.1', 0), Handler) + self.thread = threading.Thread(target=self.server.serve_forever, kwargs={'poll_interval': 0.05}) + + def __enter__(self): + self.thread.start() + return self + + def __exit__(self, *args): + self.server.shutdown() + self.server.server_close() + self.thread.join(timeout=5) + require(not self.thread.is_alive(), 'Callback receiver did not stop') + + @property + def url(self): + return f'http://127.0.0.1:{self.server.server_port}/callback' + + def wait_final(self, reply_to, timeout): + deadline = time.monotonic() + timeout + with self.condition: + while True: + errors = [item['error'] for item in self.records if 'error' in item] + require(not errors, f'Callback receiver rejected delivery: {errors}') + records = [item for item in self.records if item['payload'].get('reply_to') == reply_to] + if any(item['status'] == 200 and item['payload'].get('is_final') is True for item in records): + return list(records) + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError('No successful final callback before the observation deadline') + self.condition.wait(remaining) + + +def verify_delivery(records, session_id, *, retried): + require(bool(records), 'No callback receipts') + statuses = [item['status'] for item in records] + if retried: + require(len(records) >= 2 and [item['status'] for item in records[:2]] == [503, 200], '5xx was not retried') + require(records[0]['body'] == records[1]['body'], 'Retry changed callback body or identity') + records = records[1:] + require(all(item['status'] == 200 for item in records), 'Unexpected callback failure') + payloads = [item['payload'] for item in records] + require(all(item.get('session_id') == session_id for item in payloads), 'Callback changed session identity') + require([item.get('sequence') for item in payloads] == list(range(1, len(payloads) + 1)), 'Callback order changed') + require( + [item.get('is_final') for item in payloads] == [False] * (len(payloads) - 1) + [True], + 'Callback final marker is missing or duplicated', + ) + parts = [part for item in payloads for part in item.get('message', [])] + require(all(isinstance(part, dict) for part in parts), 'Callback message has invalid components') + text = ''.join(str(part.get('text', '')) for part in parts if part.get('type') == 'Plain') + require('LANGBOT_CHAT_OK' in text, 'Callback did not deliver the expected business reply') + return { + 'parts': len(payloads), + 'finals': 1, + 'response_statuses': statuses, + 'body_sha256': [hashlib.sha256(item['body']).hexdigest() for item in records], + } + + +def run_case(suite, case, row): + require(case['scenario'] in CASE_NAMES, 'Unknown lifecycle scenario') + secret = secrets.token_urlsafe(32) + suite.secrets.append(secret) + bot_uuid, primary_error = None, None + with CallbackReceiver(secret) as receiver: + try: + status, response = api( + suite, + 'POST', + '/api/v1/platform/bots', + { + 'name': 'QA callback retry ' + secrets.token_hex(4), + 'adapter': 'http_bot', + 'enable': True, + 'adapter_config': { + 'signature_required': True, + 'inbound_secret': secret, + 'outbound_secret': secret, + 'callback_url': receiver.url, + 'callback_timeout': 2, + 'callback_max_retries': 1, + 'default_session_type': 'group', + }, + 'event_bindings': [ + { + 'event_pattern': 'message.received', + 'target_type': 'pipeline', + 'target_uuid': suite.variables['chat_pipeline'], + 'enabled': True, + 'filters': [], + } + ], + 'plugin_processors': [], + }, + ) + require(status == 200 and response.get('code') == 0, 'Callback Bot creation failed') + bot_uuid = response['data']['uuid'] + messages = [] + for index in range(2): + session_id = 'qa-callback-' + secrets.token_hex(8) + body = { + 'session_id': session_id, + 'session_type': 'group', + 'sender': {'id': 'qa-callback-sender'}, + 'message': [{'type': 'Plain', 'text': 'ai Run the normal QA scenario'}], + } + stamp = str(int(time.time())) + status, accepted = api( + suite, + 'POST', + f'/bots/{bot_uuid}', + body, + headers={ + 'X-LB-Timestamp': stamp, + 'X-LB-Signature': signature(secret, json.dumps(body).encode(), stamp), + 'X-LB-Idempotency-Key': secrets.token_hex(16), + }, + ) + require(status == 202 and accepted.get('code') == 0, 'Asynchronous ingress was not accepted') + message_id = accepted['data']['accepted_message_id'] + name = 'callback_retry' if index == 0 else 'callback_followup' + checked( + row, + name, + lambda: verify_delivery( + receiver.wait_final(message_id, suite.timeout(30)), session_id, retried=index == 0 + ), + ) + messages.append((message_id, session_id, name)) + with receiver.condition: + receipts = list(receiver.records) + deliveries = [] + for index, (message_id, session_id, name) in enumerate(messages): + records = [item for item in receipts if item.get('payload', {}).get('reply_to') == message_id] + deliveries.append(checked(row, name, lambda: verify_delivery(records, session_id, retried=index == 0))) + completed(suite, row, session_id, row['after_id']) + return {'result': 'LANGBOT_CALLBACK_RETRY_OK', 'deliveries': deliveries} + except BaseException as exc: + primary_error = exc + raise + finally: + if bot_uuid is not None: + + def cleanup(): + status, response = api(suite, 'DELETE', f'/api/v1/platform/bots/{bot_uuid}') + require(status == 200 and response.get('code') == 0, 'Callback Bot deletion failed') + status, _ = api(suite, 'GET', f'/api/v1/platform/bots/{bot_uuid}') + require(status == 404, 'Deleted callback Bot remains visible') + + try: + checked(row, 'callback_cleanup', cleanup) + except Exception as exc: + row.setdefault('cleanup_errors', []).append(f'{type(exc).__name__}: {exc}') + if primary_error is None: + raise + + +def restart_profile(profile): + """Build the exact one-time restart fixture profile for execution and catalog checks.""" + current = copy.deepcopy(profile) + for key in ('execution', 'resource_limits', 'scenario_groups'): + current.pop(key, None) + wanted = {'mock_provider', 'chat_model', 'chat_pipeline', 'chat_bot', 'slow_model', 'slow_pipeline'} + current['resources'] = [item for item in current['resources'] if item['id'] in wanted] + require( + {item['id'] for item in current['resources']} == wanted, + 'Restart checks require the standard chat/slow fixtures', + ) + slow_bot = copy.deepcopy(next(item for item in current['resources'] if item['id'] == 'chat_bot')) + slow_bot['id'] = 'restart_slow_bot' + slow_bot['body']['name'] = 'QA runtime restart slow' + slow_bot['body']['event_bindings'][0]['target_uuid'] = '${slow_pipeline}' + current['resources'].append(slow_bot) + current.update(coverage_suite='scenarios', cleanup='always', plugins=[], cases=[]) + for name, bot, expected in ( + ('core_restart_history', 'chat_bot', 'LANGBOT_CHAT_OK'), + ('runtime_restart_recovery', 'restart_slow_bot', 'LANGBOT_SLOW_STREAM_OK'), + ): + current['cases'].append( + { + 'name': name, + 'stage': 'restart', + 'expect': expected, + 'http_group': { + 'url': '${core_base_url}/bots/${' + bot + '}/sync', + 'secret_env': 'LANGBOT_QA_GROUP_SECRET', + 'message': 'ai Run the restart QA scenario', + }, + } + ) + return current + + +def run_restart_checks(profile, profile_path, output, env, restart_owned): + """Run owned-service restart checks once, before fixed-PID resource sampling starts.""" + from regression_smoke import UnresolvedRun, reconcile_scenario_runs + from regression_suite import Suite, resolve + from regression_traffic import group_request + + started = time.monotonic() + deadline = started + 120 + + def timeout(maximum): + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError('Restart preflight exceeded its 120-second deadline') + return min(maximum, remaining) + + current = restart_profile(profile) + output = Path(output) + output.mkdir(parents=True, exist_ok=False, mode=0o700) + suite = Suite(current, Path(profile_path), output, env) + suite.timeout = timeout + user_token = suite.env.get('LANGBOT_QA_USER_TOKEN') + if user_token: + suite.secrets.append(user_token) + suite.write('profile.json', current) + result = {'mode': 'restart_preflight', 'errors': [], 'incomplete': [], 'cases': [], 'phases': []} + + def restart(row, name): + receipt = restart_owned(name, timeout=timeout(60)) + require( + isinstance(receipt, dict) + and receipt.get('old_pid') + and receipt.get('new_pid') + and receipt['old_pid'] != receipt['new_pid'], + 'Owned restart did not replace the service process', + ) + row['restart'] = receipt + timeout(1) + + try: + if not user_token: + raise ScenarioUnavailable('Restart terminal readback requires LANGBOT_QA_USER_TOKEN') + suite.prepare() + for original in current['cases']: + row = {'case': original['name'], 'stage': 'restart', 'status': 'failed', 'after_id': after_id(suite)} + result['cases'].append(row) + case = resolve(original, suite.variables) + session_id = 'qa-restart-' + secrets.token_hex(8) + token = secrets.token_hex(16) + worker = None + before = time.monotonic() + try: + if row['case'] == 'core_restart_history': + checked( + row, + 'before_restart', + lambda: group_request( + case, session_id, 'qa-member', 1, token, 'none', suite.env, timeout=timeout(20) + ), + ) + completed(suite, row, session_id, row['after_id']) + checked(row, 'process_replaced', lambda: restart(row, 'core')) + after = after_id(suite) + checked( + row, + 'history_after_restart', + lambda: group_request( + case, + session_id, + 'qa-member', + 2, + secrets.token_hex(16), + token, + suite.env, + timeout=timeout(20), + ), + ) + completed(suite, row, session_id, after) + else: + outcome = {} + + def request(): + try: + outcome['response'] = group_request( + case, session_id, 'qa-member', 1, token, 'none', suite.env, timeout=timeout(25) + ) + except Exception as exc: + outcome['error'] = type(exc).__name__ + ': ' + str(exc) + + worker = threading.Thread(target=request, daemon=True) + worker.start() + running_deadline = time.monotonic() + timeout(10) + while True: + with closing(ledger(suite)) as connection: + runs = connection.execute( + 'SELECT run_id, status, finished_at FROM agent_run ' + 'WHERE workspace_id=? AND conversation_id=? AND id>?', + (suite.variables['workspace_uuid'], 'group_' + session_id, row['after_id']), + ).fetchall() + if len(runs) == 1 and runs[0]['status'] == 'running' and not runs[0]['finished_at']: + break + if outcome or time.monotonic() >= running_deadline: + raise TimeoutError('No unique in-flight run was observed before Runtime restart') + time.sleep(0.05) + row['interrupted_run_id'] = runs[0]['run_id'] + checked(row, 'process_replaced', lambda: restart(row, 'runtime')) + worker.join(timeout=timeout(25)) + if worker.is_alive(): + raise TimeoutError('Interrupted request did not finish after Runtime restart') + row['interrupted_response'] = outcome + checked( + row, + 'no_false_success', + lambda: require( + 'response' not in outcome, 'Interrupted request returned a successful business response' + ), + ) + from regression_smoke import completed_run + + with closing(ledger(suite)) as connection: + terminal = completed_run( + connection, + suite.variables['workspace_uuid'], + 'group_' + session_id, + row['after_id'], + timeout(20), + ) + + def public_terminal(): + status, payload = api( + suite, + 'GET', + f'/api/v1/monitoring/executions/agent/{row["interrupted_run_id"]}?section=overview', + headers={'Authorization': 'Bearer ' + user_token}, + ) + run = payload.get('data', {}).get('run', {}) + require( + status == 200 + and payload.get('code') == 0 + and run.get('run_id') == row['interrupted_run_id'] == terminal['run_id'] + and run.get('status') == terminal['status'] + and run['status'] in ('failed', 'cancelled', 'timeout'), + 'Interrupted run did not expose the expected failure terminal state', + ) + return run + + row['public_terminal'] = checked(row, 'interrupted_terminal', public_terminal) + record_run(suite, row, terminal['run_id'], terminal['status']) + recovery = resolve(current['cases'][0], suite.variables) + recovery_session = 'qa-runtime-recovered-' + secrets.token_hex(8) + after = after_id(suite) + checked( + row, + 'response_after_restart', + lambda: group_request( + recovery, + recovery_session, + 'qa-member', + 1, + secrets.token_hex(16), + 'none', + suite.env, + timeout=timeout(20), + ), + ) + completed(suite, row, recovery_session, after) + row['status'] = 'passed' + except Exception as exc: + row.update( + status='unknown' + if isinstance(exc, (TimeoutError, UnresolvedRun, ScenarioUnavailable)) + else 'failed', + error=str(exc), + ) + result['errors'].append(f'{row["case"]}: {row["error"]}') + if row['status'] == 'unknown': + result['incomplete'].append(result['errors'][-1]) + finally: + if worker is not None and worker.is_alive(): + worker.join(timeout=max(0, min(5, deadline - time.monotonic()))) + if worker.is_alive(): + error = 'Restart request worker did not stop before cleanup deadline' + row.setdefault('cleanup_errors', []).append(error) + result['errors'].append(f'{row["case"]}: cleanup failed: {error}') + row['seconds'] = time.monotonic() - before + suite.write('cases.json', result['cases']) + if row['status'] != 'passed': + break + reconcile_scenario_runs(suite, result, key='ledger_closeout') + except Exception as exc: + result['errors'].append('restart preparation: ' + str(exc)) + if isinstance(exc, (TimeoutError, ScenarioUnavailable)): + result['incomplete'].append(result['errors'][-1]) + finally: + suite.write('cases.json', result['cases']) + try: + suite.finish(result) + except Exception as exc: + result['errors'].append('restart finalization: ' + str(exc)) + result.update(passed=False, outcome='FAIL') + result.update(elapsed_seconds=time.monotonic() - started, coverage_dir=str(output)) + suite.write('summary.json', result) + return result diff --git a/tools/regression_local_cases.py b/tools/regression_local_cases.py index 0436d68f2..84f3501a2 100644 --- a/tools/regression_local_cases.py +++ b/tools/regression_local_cases.py @@ -12,6 +12,7 @@ import urllib.request GROUPS = { 'faults': 'regression_execution_cases', 'resources': 'regression_resource_cases', + 'lifecycle': 'regression_lifecycle_cases', } diff --git a/tools/test_regression_lifecycle_cases.py b/tools/test_regression_lifecycle_cases.py new file mode 100644 index 000000000..ff88d8fe0 --- /dev/null +++ b/tools/test_regression_lifecycle_cases.py @@ -0,0 +1,292 @@ +"""Challenge callback evidence and cleanup with real loopback HTTP receipts.""" + +import json +from pathlib import Path +import socket +import sqlite3 +import tempfile +import threading +import time +from types import SimpleNamespace +import unittest +from unittest.mock import patch +import urllib.error +import urllib.request + +import regression_lifecycle_cases as lifecycle + + +class LifecycleTest(unittest.TestCase): + def exercise(self, corruption=None): + suite = SimpleNamespace( + variables={'chat_pipeline': 'fixture-pipeline'}, secrets=[], timeout=lambda limit: limit + ) + case = {'scenario': 'callback_retry_identity'} + row, settings, workers, receivers, errors = {'after_id': 0}, {}, [], [], [] + calls = [] + receiver_class = lifecycle.CallbackReceiver + + def receiver(secret): + result = receiver_class(secret) + receivers.append(result) + return result + + def callback(body, message_id): + payload = { + 'session_id': body['session_id'], + 'reply_to': message_id, + 'sequence': 1, + 'is_final': True, + 'message': [{'type': 'Plain', 'text': 'LANGBOT_CHAT_OK'}], + } + secret = settings['outbound_secret'] + ('wrong' if corruption == 'signature' else '') + try: + for attempt in range(2): + if corruption == 'body' and attempt: + payload['timestamp'] = 'changed by retry' + raw = json.dumps(payload).encode() + stamp = str(int(time.time())) + request = urllib.request.Request( + settings['callback_url'], + data=raw, + headers={ + 'Content-Type': 'application/json', + 'X-LB-Timestamp': stamp, + 'X-LB-Signature': lifecycle.signature(secret, raw, stamp), + }, + ) + try: + with urllib.request.urlopen(request, timeout=2) as response: + response.read() + return + except urllib.error.HTTPError as exc: + if exc.code != 503: + return + errors.append('Callback was never accepted') + except Exception as exc: + errors.append(str(exc)) + + def api(_suite, method, path, body=None, headers=None): + calls.append((method, path)) + if method == 'POST' and path == '/api/v1/platform/bots': + settings.update(body['adapter_config']) + self.assertEqual(body['event_bindings'][0]['target_uuid'], 'fixture-pipeline') + return 200, {'code': 0, 'data': {'uuid': 'temporary-bot'}} + if method == 'POST' and path == '/bots/temporary-bot': + self.assertEqual( + headers['X-LB-Signature'], + lifecycle.signature( + settings['inbound_secret'], json.dumps(body).encode(), headers['X-LB-Timestamp'] + ), + ) + message_id = 'in-' + str(len(workers)) + worker = threading.Thread(target=callback, args=(body, message_id)) + workers.append(worker) + worker.start() + return 202, {'code': 0, 'data': {'accepted_message_id': message_id}} + if method == 'DELETE' and path == '/api/v1/platform/bots/temporary-bot': + return 200, {'code': 0} + if method == 'GET' and path == '/api/v1/platform/bots/temporary-bot': + return 404, {'code': -1} + raise AssertionError((method, path)) + + with ( + patch.object(lifecycle, 'api', side_effect=api), + patch.object(lifecycle, 'CallbackReceiver', side_effect=receiver), + patch.object(lifecycle, 'completed'), + ): + if corruption is None: + result = lifecycle.run_case(suite, case, row) + self.assertEqual(result['result'], 'LANGBOT_CALLBACK_RETRY_OK') + self.assertEqual([item['finals'] for item in result['deliveries']], [1, 1]) + self.assertEqual( + row['assertions'], + {'callback_retry': 'passed', 'callback_followup': 'passed', 'callback_cleanup': 'passed'}, + ) + else: + with self.assertRaisesRegex(ValueError, 'signature|Retry changed'): + lifecycle.run_case(suite, case, row) + self.assertEqual(row['assertions']['callback_retry'], 'failed') + self.assertEqual(row['assertions']['callback_cleanup'], 'passed') + for worker in workers: + worker.join(timeout=2) + self.assertFalse(worker.is_alive()) + self.assertFalse(errors, errors) + self.assertIn(('DELETE', '/api/v1/platform/bots/temporary-bot'), calls) + self.assertFalse(receivers[0].thread.is_alive()) + with socket.socket() as client: + self.assertNotEqual(client.connect_ex(receivers[0].server.server_address), 0) + return receivers[0].records + + def test_retry_followup_and_cleanup_use_actual_http_receipts(self): + receipts = self.exercise() + self.assertEqual([item['status'] for item in receipts], [503, 200, 200]) + self.assertEqual(receipts[0]['body'], receipts[1]['body']) + + def test_modified_retry_cannot_pass_and_still_cleans_up(self): + self.exercise('body') + + def test_invalid_callback_signature_cannot_pass_and_still_cleans_up(self): + self.exercise('signature') + + def test_duplicate_final_is_not_a_successful_delivery(self): + payload = {'session_id': 'session', 'sequence': 1, 'is_final': True, 'message': []} + records = [{'status': 200, 'payload': payload, 'body': b'{}'}] * 2 + with self.assertRaisesRegex(ValueError, 'order changed'): + lifecycle.verify_delivery(records, 'session', retried=False) + + def test_unknown_callback_result_survives_cleanup_failure(self): + class NoDelivery: + url = 'http://127.0.0.1:9/test-only' + + def __enter__(self): + return self + + def __exit__(self, *args): + pass + + def wait_final(self, *args): + raise TimeoutError('callback outcome unknown') + + suite = SimpleNamespace(variables={'chat_pipeline': 'fixture'}, secrets=[], timeout=lambda limit: limit) + row = {'after_id': 0} + responses = [ + (200, {'code': 0, 'data': {'uuid': 'temporary-bot'}}), + (202, {'code': 0, 'data': {'accepted_message_id': 'accepted'}}), + (500, {'code': -1}), + ] + with ( + patch.object(lifecycle, 'CallbackReceiver', return_value=NoDelivery()), + patch.object(lifecycle, 'api', side_effect=responses), + ): + with self.assertRaisesRegex(TimeoutError, 'callback outcome unknown'): + lifecycle.run_case(suite, {'scenario': 'callback_retry_identity'}, row) + self.assertEqual(row['assertions']['callback_cleanup'], 'failed') + self.assertTrue(any('deletion failed' in error for error in row['cleanup_errors'])) + self.assertNotIn('callback_retry', row['assertions']) + + def test_restart_profile_reuses_fixture_resources_without_mutating_input(self): + path = Path(__file__).with_name('regression_profiles') / 'smoke.json' + original = json.loads(path.read_text()) + original['execution'] = {'total_seconds': 1800, 'drain_seconds': 120, 'cooldown_seconds': 60} + snapshot = json.dumps(original, sort_keys=True) + profile = lifecycle.restart_profile(original) + self.assertEqual(json.dumps(original, sort_keys=True), snapshot) + self.assertNotIn('execution', profile) + self.assertEqual(profile['coverage_suite'], 'scenarios') + self.assertEqual( + [case['name'] for case in profile['cases']], ['core_restart_history', 'runtime_restart_recovery'] + ) + self.assertEqual(len(profile['resources']), 7) + + def test_restart_preflight_records_real_ledger_rows_and_preserves_failed_history(self): + from regression_suite import Suite + + for fail_history in (False, True): + with self.subTest(fail_history=fail_history), tempfile.TemporaryDirectory() as directory: + root = Path(directory) + path = root / 'runs.db' + with sqlite3.connect(path) as connection: + connection.execute( + 'CREATE TABLE agent_run (id INTEGER PRIMARY KEY, run_id TEXT, workspace_id TEXT, ' + 'conversation_id TEXT, status TEXT, finished_at TEXT, status_reason TEXT)' + ) + profile_path = Path(__file__).with_name('regression_profiles') / 'smoke.json' + profile = json.loads(profile_path.read_text()) + profile['smoke']['ledger_db'] = str(path) + release = threading.Event() + restarts = [] + + def prepare(suite): + suite.variables.update(workspace_uuid='workspace', chat_bot='chat-bot', restart_slow_bot='slow-bot') + suite.journal() + + def group_request(case, session_id, _member, turn, _token, previous, _env, *, timeout): + is_slow = 'slow-bot' in case['http_group']['url'] + with sqlite3.connect(path) as connection: + count = connection.execute('SELECT COUNT(*) FROM agent_run').fetchone()[0] + run_id = f'run-{count}' + connection.execute( + 'INSERT INTO agent_run(run_id,workspace_id,conversation_id,status,finished_at) VALUES(?,?,?,?,?)', + ( + run_id, + 'workspace', + 'group_' + session_id, + 'running' if is_slow else 'completed', + None if is_slow else 'done', + ), + ) + if is_slow: + if not release.wait(2): + raise TimeoutError('Runtime restart was never invoked') + with sqlite3.connect(path) as connection: + connection.execute( + "UPDATE agent_run SET status='failed',finished_at='done' WHERE run_id=?", (run_id,) + ) + raise RuntimeError('Runtime disconnected') + if turn == 2: + self.assertNotEqual(previous, 'none') + if fail_history: + raise ValueError('history rejected after restart') + return {'data': {'session_id': session_id}} + + def restart(name, *, timeout): + self.assertGreater(timeout, 0) + self.assertLessEqual(timeout, 60) + restarts.append(name) + if name == 'runtime': + release.set() + return {'old_pid': 10, 'new_pid': 20} + + def public_api(_suite, method, resource, *, headers): + self.assertEqual(method, 'GET') + self.assertEqual(headers, {'Authorization': 'Bearer test-only-user-token'}) + run_id = resource.split('/')[-1].split('?')[0] + with sqlite3.connect(path) as connection: + status = connection.execute( + 'SELECT status FROM agent_run WHERE run_id=?', (run_id,) + ).fetchone()[0] + return 200, {'code': 0, 'data': {'run': {'run_id': run_id, 'status': status}}} + + with ( + patch.object(Suite, 'prepare', prepare), + patch.object(Suite, 'call', return_value={'workspace_uuid': 'workspace'}), + patch('regression_traffic.group_request', side_effect=group_request), + patch.object(lifecycle, 'api', side_effect=public_api), + ): + result = lifecycle.run_restart_checks( + profile, + profile_path, + root / 'preflight', + {'LANGBOT_QA_USER_TOKEN': 'test-only-user-token'}, + restart, + ) + self.assertEqual(result['passed'], not fail_history, result) + self.assertEqual( + json.loads((root / 'preflight/profile.json').read_text()), lifecycle.restart_profile(profile) + ) + self.assertEqual(restarts, ['core'] if fail_history else ['core', 'runtime']) + self.assertEqual(json.loads((root / 'preflight/owned.json').read_text())['resources'], []) + if fail_history: + self.assertTrue(any('history rejected' in message for message in result['errors'])) + self.assertEqual(result['cases'][0]['assertions']['history_after_restart'], 'failed') + else: + self.assertEqual(result['cases'][1]['scenario_runs'][0]['expected_status'], 'failed') + self.assertTrue(result['ledger_closeout']['matched']) + self.assertEqual(sum(len(row['scenario_runs']) for row in result['cases']), 4) + + def test_restart_missing_user_token_is_incomplete_before_preparation(self): + from regression_suite import Suite + + path = Path(__file__).with_name('regression_profiles') / 'smoke.json' + profile = json.loads(path.read_text()) + with tempfile.TemporaryDirectory() as directory, patch.object(Suite, 'prepare') as prepare: + result = lifecycle.run_restart_checks(profile, path, Path(directory) / 'preflight', {}, None) + prepare.assert_not_called() + self.assertFalse(result['passed']) + self.assertEqual(result['outcome'], 'INCOMPLETE') + self.assertTrue(any('LANGBOT_QA_USER_TOKEN' in error for error in result['errors'])) + + +if __name__ == '__main__': + unittest.main()