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()