From 687fc52b362d2cb66bad781bdae6c9b934ea89a8 Mon Sep 17 00:00:00 2001 From: huanghuoguoguo <1051233107@qq.com> Date: Fri, 30 Oct 2026 08:35:09 +0800 Subject: [PATCH 1/7] Merge pull request #2810 from langbot-app/test/state-lifecycle-job test(regression): add durable state lifecycle job --- tools/README.md | 23 +++ tools/regression_ci.py | 41 +++-- tools/regression_contracts.json | 15 ++ tools/regression_state.py | 278 ++++++++++++++++++++++++++++++ tools/test_regression_coverage.py | 4 +- tools/test_regression_state.py | 43 +++++ 6 files changed, 386 insertions(+), 18 deletions(-) create mode 100644 tools/regression_state.py create mode 100644 tools/test_regression_state.py diff --git a/tools/README.md b/tools/README.md index 0c3f8d5cb..9afa44dce 100644 --- a/tools/README.md +++ b/tools/README.md @@ -498,6 +498,29 @@ bind their own extensions to avoid changing unrelated prompts/tools. The suite a even after failures. The CI owner then stops its service processes and removes its Box containers; a process guard timeout can interrupt API cleanup. +## Durable state lifecycle job + +Run the finite state job on the same disposable Linux stack: + +```sh +uv run --no-sync python tools/regression_ci.py --suite state \ + --lbctl /path/to/lbctl --plugins plugins --output .qa/state-001 +``` + +The dependent chain creates a Pipeline, model bindings, HTTP Bot and isolated +knowledge base. It verifies conversation history and exact document content +through Runtime and Core restarts, changes the Pipeline model and replaces one +document while retaining a control document, then repeats the readback after +another Core restart. Deleting all documents must leave both file listings and +retrieval empty; normal Suite cleanup checks the owned configuration resources. + +The business chain has a 270-second budget and a 30-second cleanup reserve; +plugin installation and initial service startup are outside this budget. +`gate/suite/cases.json` retains assertions, process replacement receipts, exact +HTTP statuses and retrieved content. A missing phase or failed assertion fails +the job. The job is opt-in and does not change Smoke or the daily schedule. It +checks graceful process restart durability, not abrupt power loss or migrations. + ## Daily API contract job `daily-endurance.yml` runs API contracts alongside the 30-minute endurance job diff --git a/tools/regression_ci.py b/tools/regression_ci.py index 95d4de210..186c0883f 100644 --- a/tools/regression_ci.py +++ b/tools/regression_ci.py @@ -21,7 +21,7 @@ from regression_suite import Suite from regression_traffic import owned_box_containers, read_endpoint_snapshot ROOT = Path(__file__).resolve().parents[1] -SUITE_BUDGETS = {'smoke': 60, 'scenarios': 300, 'daily': 1860, 'endurance': 900, 'api': 600} +SUITE_BUDGETS = {'smoke': 60, 'scenarios': 300, 'daily': 1860, 'endurance': 900, 'api': 600, 'state': 300} sys.path.insert(0, str(ROOT)) from tests.e2e.utils.config_factory import create_minimal_config, create_test_directories # noqa: E402 @@ -34,6 +34,10 @@ def suite_profile(name, run_id, groups=(), api_seed=1): profile = expanded_profile(groups) else: profile = json.loads(profile_path.read_text()) + if name == 'state': + from regression_state import state_profile + + profile = state_profile(profile) if name == 'api': from regression_api_cases import add_cases @@ -309,21 +313,26 @@ def run(args): ) generated = private / f'{args.suite}.json' generated.write_text(json.dumps(profile, indent=2)) - result = supervise( - [ - sys.executable, - str(ROOT / 'tools/regression_suite.py'), - 'run', - '--profile', - str(generated), - '--output', - str(root / 'gate/suite'), - ], - root / 'gate', - SUITE_BUDGETS[args.suite] - restart_elapsed, - pids, - env, - ) + if args.suite == 'state': + from regression_state import run_state_checks + + result = run_state_checks(profile, profile_path, root / 'gate/suite', env, restart_owned) + else: + result = supervise( + [ + sys.executable, + str(ROOT / 'tools/regression_suite.py'), + 'run', + '--profile', + str(generated), + '--output', + str(root / 'gate/suite'), + ], + root / 'gate', + SUITE_BUDGETS[args.suite] - restart_elapsed, + pids, + env, + ) evidence = [root / 'gate/suite'] if restart_result is not None: evidence.append(root / 'restart-checks') diff --git a/tools/regression_contracts.json b/tools/regression_contracts.json index 2597653b9..87910bdb8 100644 --- a/tools/regression_contracts.json +++ b/tools/regression_contracts.json @@ -434,6 +434,21 @@ "priority": "P1", "description": "Invalid MCP transport is rejected without changing stored configuration", "bindings": [{"profile": "api", "case": "fixed_api_mcp", "checks": ["api_chain", "mcp_invalid_transport_preserved"]}, {"profile": "api", "case": "shuffled_api_mcp", "checks": ["api_chain", "mcp_invalid_transport_preserved"]}] + }, + {"id": "state.configuration_persistence", + "area": "state", "kind": "blackbox", "priority": "P1", + "description": "Created and updated pipeline configuration survives owned Core and Runtime restarts", + "bindings": [{"profile": "state", "case": "durable_state_lifecycle", "checks": ["created_configuration", "after_runtime_configuration", "after_core_configuration", "updated_configuration", "persisted_update_configuration", "initial_runtime_process_replaced", "initial_core_process_replaced", "updated_core_process_replaced"]}] + }, + {"id": "state.history_and_model_update", + "area": "state", "kind": "blackbox", "priority": "P1", + "description": "Conversation history survives restarts and the same session uses the updated model", + "bindings": [{"profile": "state", "case": "durable_state_lifecycle", "checks": ["created_chat_history", "after_runtime_chat_history", "after_core_chat_history", "updated_chat_history", "persisted_update_chat_history"]}] + }, + {"id": "state.knowledge_replacement", + "area": "state", "kind": "blackbox", "priority": "P1", + "description": "Knowledge documents survive restarts; replacement preserves control content and deletion removes stale results", + "bindings": [{"profile": "state", "case": "durable_state_lifecycle", "checks": ["created_retrievable", "after_runtime_retrievable", "after_core_retrievable", "updated_retrievable", "updated_obsolete_absent", "persisted_update_retrievable", "persisted_update_obsolete_absent", "documents_deleted", "deleted_retrieval_empty"]}] } ] } diff --git a/tools/regression_state.py b/tools/regression_state.py new file mode 100644 index 000000000..3b4deba1d --- /dev/null +++ b/tools/regression_state.py @@ -0,0 +1,278 @@ +"""Finite durable-state lifecycle against the runner's disposable Core and Runtime.""" + +import copy +import json +from pathlib import Path +import secrets +import time + +from regression_api_checks import check, request +from regression_coverage import checked +from regression_lifecycle_cases import after_id, completed +from regression_local_cases import require +from regression_smoke import reconcile_scenario_runs +from regression_suite import Suite, contains_fields, resolve, validate +from regression_traffic import group_request + + +def state_profile(profile): + """Keep only the fixtures used by the dependent lifecycle chain.""" + profile = copy.deepcopy(profile) + wanted = { + 'mock_provider', + 'embedding_provider', + 'chat_model', + 'unicode_model', + 'embedding_model', + 'empty_knowledge_base', + 'chat_pipeline', + 'chat_bot', + } + profile['resources'] = [item for item in profile['resources'] if item['id'] in wanted] + require({item['id'] for item in profile['resources']} == wanted, 'State lifecycle fixtures are incomplete') + profile.update(coverage_suite='state', cleanup='always') + profile['cases'] = [ + { + 'name': 'durable_state_lifecycle', + 'stage': 'state', + 'expect': 'LANGBOT_CHAT_OK', + 'http_group': { + 'url': '${core_base_url}/bots/${chat_bot}/sync', + 'secret_env': 'LANGBOT_QA_GROUP_SECRET', + 'message': 'ai Run the durable state scenario', + }, + } + ] + return profile + + +def document_visible(results, 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 results + ) + + +def document_absent(results, document, marker): + return all( + item.get('metadata', {}).get('file_id') != document['engine_document_id'] + and marker not in json.dumps(item, ensure_ascii=False) + for item in results + ) + + +def run_state_checks(profile, profile_path, output, env, restart_owned, budget=300): + """Exercise create/restart/update/restart/delete while retaining exact business evidence.""" + validate(profile) + output = Path(output) + output.mkdir(parents=True, exist_ok=False, mode=0o700) + started = time.monotonic() + deadline = started + budget - 30 + suite = Suite(profile, profile_path, output, env) + + def timeout(maximum): + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError('State lifecycle exhausted its execution deadline') + return min(maximum, remaining) + + suite.timeout = timeout + suite.write('profile.json', profile) + row = {'case': 'durable_state_lifecycle', 'stage': 'state', 'status': 'unknown', 'restarts': []} + result = {'mode': 'state', 'cases': [row], 'errors': [], 'incomplete': [], 'phases': []} + + def restart(service, phase): + receipt = restart_owned(service, timeout=timeout(60)) + check( + row, + phase + '_process_replaced', + bool(receipt.get('old_pid')) and bool(receipt.get('new_pid')) and receipt['old_pid'] != receipt['new_pid'], + 'Owned restart did not replace its process', + ) + row['restarts'].append({'phase': phase, **receipt}) + + try: + suite.prepare() + row['after_id'] = after_id(suite) + case = resolve(profile['cases'][0], suite.variables) + pipeline_path = '/api/v1/pipelines/' + suite.variables['chat_pipeline'] + kb_path = '/api/v1/knowledge/bases/' + suite.variables['empty_knowledge_base'] + pipeline = request(suite, row, 'GET', pipeline_path)['pipeline'] + submitted = resolve( + next(r for r in profile['resources'] if r['id'] == 'chat_pipeline')['body'], suite.variables + ) + check( + row, + 'created_configuration', + contains_fields(pipeline, submitted), + 'Pipeline differs from created configuration', + ) + expected_pipeline = { + key: pipeline[key] for key in ('uuid', 'name', 'description', 'config', 'extensions_preferences') + } + + def configuration(phase): + current = request(suite, row, 'GET', pipeline_path)['pipeline'] + check( + row, + phase + '_configuration', + contains_fields(current, expected_pipeline), + 'Persisted pipeline configuration changed', + ) + + def files(): + return request(suite, row, 'GET', kb_path + '/files')['files'] + + def ingest(label, marker): + previous = {item['uuid'] for item in files()} + document = output / (label + '.txt') + document.write_text(f'Durable lifecycle document {label}: {marker}.\n', encoding='utf-8') + suite.call( + [ + 'knowledge-base', + 'ingest', + suite.variables['empty_knowledge_base'], + '--file', + str(document), + '--parser-plugin-id', + 'langbot-team/GeneralParsers', + '--wait', + ] + ) + added = [item for item in files() if item['uuid'] not in previous] + check( + row, + label + '_ingested', + len(added) == 1 and added[0].get('status') == 'completed' and bool(added[0].get('engine_document_id')), + 'Expected exactly one completed document', + ) + row.setdefault('documents', {})[label] = { + 'uuid': added[0]['uuid'], + 'engine_document_id': added[0]['engine_document_id'], + 'marker': marker, + } + return added[0] + + markers = {name: name + secrets.token_hex(16) for name in ('original', 'control', 'replacement')} + documents = {name: ingest(name, markers[name]) for name in ('original', 'control')} + session, previous, turn = 'qa-state-' + secrets.token_hex(16), 'none', 0 + + def chat(phase): + nonlocal previous, turn + token = secrets.token_hex(16) + after = after_id(suite) + turn += 1 + checked( + row, + phase + '_chat_history', + lambda: group_request( + case, + session, + 'qa-state-member', + turn, + token, + previous, + suite.env, + timeout=timeout(30), + ), + ) + completed(suite, row, session, after) + previous = token + + def retrieval(phase, *, updated=False): + found = request( + suite, + row, + 'POST', + kb_path + '/retrieve', + { + 'query': ' '.join(markers.values()), + 'retrieval_settings': {'top_k': 10}, + }, + )['results'] + check( + row, + phase + '_retrieval_shape', + 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', + ) + selected = ('replacement', 'control') if updated else ('original', 'control') + check( + row, + phase + '_retrievable', + all(document_visible(found, documents[name], markers[name]) for name in selected), + 'Persisted documents are missing or return incorrect content', + ) + if updated: + check( + row, + phase + '_obsolete_absent', + document_absent(found, documents['original'], markers['original']) + and documents['original']['uuid'] not in {item['uuid'] for item in files()}, + 'Replaced document is still visible through retrieval or file listing', + ) + row.setdefault('retrievals', []).append({'phase': phase, 'results': found}) + + chat('created') + retrieval('created') + for service in ('runtime', 'core'): + restart(service, 'initial_' + service) + configuration('after_' + service) + chat('after_' + service) + retrieval('after_' + service) + + updated = copy.deepcopy(expected_pipeline['config']) + updated['ai']['runner_config']['plugin:langbot-team/LocalAgent/default']['model']['primary'] = suite.variables[ + 'unicode_model' + ] + request(suite, row, 'PUT', pipeline_path, {'description': 'Updated durable state fixture', 'config': updated}) + expected_pipeline.update(description='Updated durable state fixture', config=updated) + configuration('updated') + case['expect'] = 'LANGBOT_UNICODE_OK' + case['http_group']['message'] = 'ai 请原样处理：你好，世界 🌍' + chat('updated') + documents['replacement'] = ingest('replacement', markers['replacement']) + request(suite, row, 'DELETE', kb_path + '/files/' + documents['original']['uuid']) + retrieval('updated', updated=True) + restart('core', 'updated_core') + configuration('persisted_update') + chat('persisted_update') + retrieval('persisted_update', updated=True) + for name in ('replacement', 'control'): + request(suite, row, 'DELETE', kb_path + '/files/' + documents[name]['uuid']) + check(row, 'documents_deleted', not files(), 'Deleted document metadata remains visible') + found = request( + suite, + row, + 'POST', + kb_path + '/retrieve', + { + 'query': ' '.join(markers.values()), + 'retrieval_settings': {'top_k': 10}, + }, + )['results'] + check(row, 'deleted_retrieval_empty', found == [], 'Deleted knowledge content remains retrievable') + reconcile_scenario_runs(suite, result, key='ledger_closeout') + require(not result['errors'], 'Lifecycle run ledger did not reconcile') + row['status'] = 'passed' + except Exception as exc: + row.update(status='unknown' if isinstance(exc, (TimeoutError, OSError)) else 'failed', error=str(exc)) + result['errors'].append('state lifecycle: ' + str(exc)) + if row['status'] == 'unknown': + result['incomplete'].append(result['errors'][-1]) + finally: + row['seconds'] = time.monotonic() - started + suite.write('cases.json', result['cases']) + # Keep a bounded cleanup window even when the business chain reaches its deadline. + deadline = time.monotonic() + 30 + suite.finish(result) + result.update(elapsed_seconds=time.monotonic() - started, case_count=1, requests=1) + suite.write('summary.json', result) + return result diff --git a/tools/test_regression_coverage.py b/tools/test_regression_coverage.py index a0d87e941..8ccfe9daa 100644 --- a/tools/test_regression_coverage.py +++ b/tools/test_regression_coverage.py @@ -289,10 +289,10 @@ class Routes: name = binding['profile'] if name not in profiles: path = Path(__file__).with_name('regression_profiles') / f'{name}.json' - if name == 'api': + if name in ('api', 'state'): from regression_ci import suite_profile - profile, _ = suite_profile('api', 'catalog-test') + profile, _ = suite_profile(name, 'catalog-test') elif name == 'scenarios': from regression_scenarios import expanded_profile from regression_local_cases import DEFAULT_GROUPS diff --git a/tools/test_regression_state.py b/tools/test_regression_state.py new file mode 100644 index 000000000..49d523e0a --- /dev/null +++ b/tools/test_regression_state.py @@ -0,0 +1,43 @@ +"""Challenge lifecycle retrieval oracles and keep the optional suite isolated.""" + +import copy +import unittest + +from regression_ci import SUITE_BUDGETS, suite_profile +from regression_state import document_absent, document_visible +from regression_suite import validate + + +class StateLifecycleTest(unittest.TestCase): + def test_document_identity_and_content_are_both_required(self): + document = {'engine_document_id': 'original-file'} + original = {'metadata': {'file_id': 'original-file'}, 'content': [{'text': 'original-marker'}]} + self.assertTrue(document_visible([original], document, 'original-marker')) + self.assertFalse(document_visible([], document, 'original-marker')) + for changed in ( + {'metadata': {'file_id': 'other-file'}, 'content': original['content']}, + {'metadata': original['metadata'], 'content': [{'text': 'other-marker'}]}, + ): + with self.subTest(changed=changed): + self.assertFalse(document_visible([changed], document, 'original-marker')) + self.assertFalse(document_absent([changed], document, 'original-marker')) + self.assertFalse(document_absent([original], document, 'original-marker')) + self.assertTrue(document_absent([], document, 'original-marker')) + + def test_state_profile_does_not_change_smoke_or_daily_cases(self): + smoke, _ = suite_profile('smoke', 'before') + daily, _ = suite_profile('daily', 'before') + expected = copy.deepcopy((smoke, daily)) + state, _ = suite_profile('state', 'state') + validate(state) + self.assertEqual((smoke, daily), expected) + self.assertEqual(state['coverage_suite'], 'state') + self.assertNotIn('execution', state) + self.assertNotIn('scenario_groups', state) + self.assertEqual(SUITE_BUDGETS['state'], 300) + self.assertTrue(all(not resource.get('documents') for resource in state['resources'])) + self.assertTrue(all('state' not in case['name'] for case in smoke['cases'] + daily['cases'])) + + +if __name__ == '__main__': + unittest.main() From 52b07cb01f07508638e03d611bdbcab18ea2d746 Mon Sep 17 00:00:00 2001 From: huanghuoguoguo <1051233107@qq.com> Date: Fri, 30 Oct 2026 08:53:23 +0800 Subject: [PATCH 2/7] fix(regression): tolerate sandbox removal during resource sampling Squash merge after all required checks passed. --- tools/regression_traffic.py | 51 ++++++-- tools/test_regression_containers.py | 179 +++++++++++++++++++++++++++ tools/test_regression_load_ledger.py | 2 +- 3 files changed, 223 insertions(+), 9 deletions(-) create mode 100644 tools/test_regression_containers.py diff --git a/tools/regression_traffic.py b/tools/regression_traffic.py index 05f5b5fbe..3dd4c6b2f 100644 --- a/tools/regression_traffic.py +++ b/tools/regression_traffic.py @@ -28,23 +28,54 @@ def percentile(values, fraction): return sorted(values)[max(0, math.ceil(len(values) * fraction) - 1)] if values else None +def missing_containers(error, candidates): + messages = { + prefix + container: container + for container in candidates + for prefix in ('Error: No such object: ', 'Error response from daemon: No such container: ') + } + lines = [line.strip() for line in error.splitlines() if line.strip()] + return {messages[line] for line in lines} if lines and all(line in messages for line in lines) else set() + + def owned_box_containers(data_root, checked=None): """Discover sandboxes mounted below one run's data root, including newly created groups.""" if checked is None: def checked(command): - return subprocess.check_output(command, text=True, timeout=10) + return subprocess.check_output(command, text=True, stderr=subprocess.PIPE, timeout=10) candidates = checked(['docker', 'ps', '-a', '--filter', 'label=langbot.box=true', '--format', '{{.ID}}']).split() if not candidates: return [] root = Path(data_root).resolve() - descriptions = checked(['docker', 'inspect', '--format', '{{json .}}', *candidates]) owned = [] - for line in descriptions.splitlines(): - if not line: + for container in candidates: + try: + description = checked(['docker', 'inspect', '--format', '{{json .}}', container]) + except subprocess.CalledProcessError as exc: + if ( + exc.returncode != 1 + or (exc.output or '').strip() + or not missing_containers(exc.stderr or '', {container}) + ): + raise + continue # A sandbox may expire between discovery and inspection. + except RuntimeError as exc: + if not missing_containers(str(exc), {container}): + raise continue - entry = json.loads(line) + entry = json.loads(description) + if ( + not isinstance(entry, dict) + or not isinstance(entry.get('Id'), str) + or not entry['Id'].startswith(container) + or not isinstance(entry.get('Mounts'), list) + or any( + not isinstance(mount, dict) or not isinstance(mount.get('Source', ''), str) for mount in entry['Mounts'] + ) + ): + raise ValueError(f'Malformed container inspection: {container}') if any( Path(mount['Source']).resolve().is_relative_to(root) for mount in entry['Mounts'] if mount.get('Source') ): @@ -70,11 +101,15 @@ def sample_containers(configured, data_root, run=subprocess.run, discover=owned_ result = stats(containers) observation = {} - if result.returncode and data_root: + missing = missing_containers(result.stderr, containers) if result.returncode == 1 else set() + # Docker 26 can close the stats stream with EOF when a sampled sandbox is deleted. + eof = result.returncode == 1 and result.stderr.strip() == 'EOF' and not result.stdout.strip() + if (missing or eof) and data_root: current = selected() - if containers - current: + removed = containers - current + if removed and (eof or missing <= removed): observation['container_churn'] = { - 'removed': sorted(containers - current), + 'removed': sorted(removed), 'added': sorted(current - containers), 'first_error': result.stderr.strip(), } diff --git a/tools/test_regression_containers.py b/tools/test_regression_containers.py new file mode 100644 index 000000000..ff1e35718 --- /dev/null +++ b/tools/test_regression_containers.py @@ -0,0 +1,179 @@ +"""Container expiry must not hide Docker failures or expand cleanup ownership.""" + +import json +import subprocess +from types import SimpleNamespace +import unittest +from unittest.mock import Mock + +from regression_traffic import owned_box_containers, sample_containers + + +def description(container, mount): + return json.dumps({'Id': container, 'Mounts': [{'Source': mount}]}) + + +class ContainerSamplingTests(unittest.TestCase): + def test_eof_resamples_survivors_and_new_containers_after_confirmed_deletion(self): + run = Mock( + side_effect=[ + SimpleNamespace(returncode=1, stdout='', stderr='EOF\n'), + SimpleNamespace( + returncode=0, + stdout='{"Container":"configured","PIDs":"1"}\n' + '{"Container":"live","PIDs":"1"}\n{"Container":"new","PIDs":"1"}\n', + stderr='', + ), + ] + ) + result = sample_containers(['configured'], '/owned', run, Mock(side_effect=[['old', 'live'], ['live', 'new']])) + self.assertNotIn('container_error', result) + self.assertEqual(result['container_churn'], {'removed': ['old'], 'added': ['new'], 'first_error': 'EOF'}) + self.assertEqual([entry['Container'] for entry in result['containers']], ['configured', 'live', 'new']) + self.assertEqual(run.call_count, 2) + self.assertEqual(run.call_args.args[0][-3:], ['configured', 'live', 'new']) + + def test_eof_without_confirmed_dynamic_deletion_remains_failure(self): + for configured, discoveries in ( + ([], [['live'], ['live']]), + ([], [['live'], ['live', 'new']]), + (['configured'], [['configured', 'live'], ['live']]), + ): + with self.subTest(configured=configured, discoveries=discoveries): + run = Mock(return_value=SimpleNamespace(returncode=1, stdout='', stderr='EOF')) + result = sample_containers(configured, '/owned', run, Mock(side_effect=discoveries)) + self.assertEqual(result['container_error'], 'EOF') + self.assertNotIn('container_churn', result) + self.assertEqual(run.call_count, 1) + + def test_eof_discovery_failure_cannot_confirm_deletion(self): + error = subprocess.CalledProcessError(1, ['docker', 'ps'], '', 'permission denied') + run = Mock(return_value=SimpleNamespace(returncode=1, stdout='', stderr='EOF')) + with self.assertRaises(subprocess.CalledProcessError) as caught: + sample_containers([], '/owned', run, Mock(side_effect=[['old'], error])) + self.assertIs(caught.exception, error) + self.assertEqual(run.call_count, 1) + + def test_eof_retry_failure_is_not_hidden_or_retried_again(self): + for error in ('EOF', 'permission denied', 'Error response from daemon: No such container: configured'): + with self.subTest(error=error): + run = Mock( + side_effect=[ + SimpleNamespace(returncode=1, stdout='', stderr='EOF'), + SimpleNamespace(returncode=1, stdout='', stderr=error), + ] + ) + result = sample_containers(['configured'], '/owned', run, Mock(side_effect=[['old', 'live'], ['live']])) + self.assertEqual(result['container_error'], error) + self.assertEqual(result['container_churn']['first_error'], 'EOF') + self.assertEqual(run.call_count, 2) + run = Mock( + side_effect=[ + SimpleNamespace(returncode=1, stdout='', stderr='EOF'), + SimpleNamespace(returncode=0, stdout='{truncated', stderr=''), + ] + ) + with self.assertRaises(ValueError): + sample_containers([], '/owned', run, Mock(side_effect=[['old', 'live'], ['live']])) + + def test_only_empty_output_and_pure_exit_one_eof_can_recover(self): + for code, error, stdout in ( + (2, 'EOF', ''), + (1, 'EOF\npermission denied', ''), + (1, 'EOF', '{"Container":"partial"}'), + ): + with self.subTest(code=code, error=error, stdout=stdout): + run = Mock(return_value=SimpleNamespace(returncode=code, stdout=stdout, stderr=error)) + discover = Mock(side_effect=[['old', 'live'], ['live']]) + result = sample_containers([], '/owned', run, discover) + self.assertEqual(result['container_error'], error) + self.assertEqual(run.call_count, 1) + self.assertEqual(discover.call_count, 1) + + def test_expiry_between_list_and_inspect_preserves_owned_survivors(self): + for error in ( + subprocess.CalledProcessError(1, ['docker', 'inspect'], '\n', 'Error: No such object: expired'), + RuntimeError('Error: No such object: expired\n'), + ): + with self.subTest(error=type(error).__name__): + checked = Mock( + side_effect=[ + 'expired live foreign', + error, + description('live', '/owned/box/session'), + description('foreign', '/owned/box-other/session'), + ] + ) + self.assertEqual(owned_box_containers('/owned/box', checked), ['live']) + + def test_inspect_errors_are_not_reported_as_expiry(self): + for error in ( + subprocess.CalledProcessError(1, [], '', 'permission denied'), + subprocess.CalledProcessError(1, [], '', 'Cannot connect to the Docker daemon'), + subprocess.CalledProcessError(1, [], '', 'Error: No such object: different-id'), + subprocess.CalledProcessError(1, [], '', 'Error: No such object: live\npermission denied'), + subprocess.CalledProcessError(2, [], '', 'Error: No such object: live'), + subprocess.CalledProcessError(1, [], '{"partial":true}', 'Error: No such object: live'), + subprocess.TimeoutExpired(['docker', 'inspect'], 10), + RuntimeError('Cannot connect to the Docker daemon'), + ): + with self.subTest(error=repr(error)): + with self.assertRaises(type(error)) as caught: + owned_box_containers('/owned', Mock(side_effect=['live', error])) + self.assertIs(caught.exception, error) + + def test_malformed_inspection_fails_instead_of_dropping_a_sample(self): + for output in ( + '', + '{truncated', + '{}', + '{"Mounts":null}', + '[]', + '{"Id":"live","Mounts":[null]}', + '{"Id":"other","Mounts":[]}', + '{"Id":"live","Mounts":[{"Source":42}]}', + ): + with self.subTest(output=output): + with self.assertRaises((ValueError, KeyError, TypeError)): + owned_box_containers('/owned', Mock(side_effect=['live', output])) + + def test_discovery_failure_is_not_an_empty_inventory(self): + error = subprocess.CalledProcessError(1, [], '', 'permission denied') + with self.assertRaises(subprocess.CalledProcessError) as caught: + owned_box_containers('/owned', Mock(side_effect=error)) + self.assertIs(caught.exception, error) + + def test_unrelated_stats_failure_is_not_hidden_by_concurrent_removal(self): + for error in ( + 'permission denied', + 'Cannot connect to the Docker daemon', + 'Error response from daemon: No such container: old\npermission denied', + ): + with self.subTest(error=error): + run = Mock(return_value=SimpleNamespace(returncode=1, stdout='', stderr=error)) + discover = Mock(side_effect=[['old', 'live'], ['live']]) + result = sample_containers([], '/owned', run, discover) + self.assertEqual(result['container_error'], error) + self.assertEqual(run.call_count, 1) + self.assertEqual(discover.call_count, 1) + + def test_configured_container_loss_remains_a_failure(self): + error = 'Error response from daemon: No such container: configured' + run = Mock(return_value=SimpleNamespace(returncode=1, stdout='', stderr=error)) + result = sample_containers(['configured'], '/owned', run, Mock(return_value=[])) + self.assertEqual(result['container_error'], error) + self.assertEqual(run.call_count, 1) + + def test_all_dynamic_containers_removed_returns_observed_churn(self): + for error in ('Error response from daemon: No such container: old', 'EOF'): + with self.subTest(error=error): + run = Mock(return_value=SimpleNamespace(returncode=1, stdout='', stderr=error)) + result = sample_containers([], '/owned', run, Mock(side_effect=[['old'], []])) + self.assertEqual(result['containers'], []) + self.assertEqual(result['container_churn']['removed'], ['old']) + self.assertEqual(result['container_churn']['first_error'], error) + self.assertNotIn('container_error', result) + + +if __name__ == '__main__': + unittest.main() diff --git a/tools/test_regression_load_ledger.py b/tools/test_regression_load_ledger.py index 19cedbdac..9fb021984 100644 --- a/tools/test_regression_load_ledger.py +++ b/tools/test_regression_load_ledger.py @@ -42,7 +42,7 @@ class LoadLedgerTests(unittest.TestCase): self.assertEqual(result['not_completed'], ['late']) def test_container_churn_recovers_once_but_other_collection_errors_fail(self): - failed = SimpleNamespace(returncode=1, stdout='', stderr='No such container: old') + failed = SimpleNamespace(returncode=1, stdout='', stderr='Error response from daemon: No such container: old') success = SimpleNamespace(returncode=0, stdout='{"MemUsage":"1MiB / 1GiB","PIDs":"1"}\n', stderr='') run = Mock(side_effect=[failed, success]) result = sample_containers([], '/owned', run, Mock(side_effect=[['old', 'live'], ['live']])) From 21ff044df1fd887dfc9f6dd96461f69f5bb3e36c Mon Sep 17 00:00:00 2001 From: huanghuoguoguo <1051233107@qq.com> Date: Fri, 30 Oct 2026 09:06:07 +0800 Subject: [PATCH 3/7] test(regression): add concurrent conversation isolation job (#2811) Squash merge after conflict resolution and all required checks passed. --- tools/README.md | 28 ++ tools/regression_ci.py | 18 +- tools/regression_contracts.json | 5 +- tools/regression_execution_cases.py | 7 +- tools/regression_isolation_cases.py | 328 +++++++++++++++++++++++ tools/regression_local_cases.py | 1 + tools/test_regression_coverage.py | 2 +- tools/test_regression_execution_cases.py | 11 +- tools/test_regression_isolation.py | 86 ++++++ 9 files changed, 478 insertions(+), 8 deletions(-) create mode 100644 tools/regression_isolation_cases.py create mode 100644 tools/test_regression_isolation.py diff --git a/tools/README.md b/tools/README.md index 9afa44dce..d3c22e111 100644 --- a/tools/README.md +++ b/tools/README.md @@ -561,6 +561,34 @@ An HTTP route hit includes rejected requests and is not proof of all behavior behind that route. Invalid MCP modes must return HTTP 400 and leave the original configuration unchanged. Any failed contract or cleanup makes the job fail. +## Concurrent isolation job + +Run this finite job on a disposable Linux/Docker host: + +```sh +uv run --no-sync python tools/regression_ci.py --suite isolation \ + --lbctl /path/to/lbctl --plugins plugins --output .qa/isolation-001 +``` + +It uses two concurrent requests in one Workspace. Three scenarios cover two +sessions on one Pipeline, two sessions on different Pipelines/models, and +cancelling an active Box tool while a neighboring conversation completes. +The first two scenarios each make two turns per session, checking request tokens, +previous-turn history, Bot identity, model routing and exact terminal results. +The cancellation scenario reuses the delayed-side-effect check and requires the +Box to become idle. Read-only ledger intervals must overlap; serialized execution +cannot pass as a concurrency test. Every accepted request must match exactly one +terminal run, and mock records must contain exactly the owned model calls. + +HTTP Bot `/sync` collects cumulative stream snapshots, so each text component +must be a monotonic prefix of that request's expected reply and the last must be +complete. The job cannot count final delivery frames on this lossy endpoint. +The 300-second guard bounds execution after preparation; there are no throughput +or latency thresholds. `gate/suite/cases.json` retains per-request evidence and +`gate/gate.json` the overall verdict. The job is explicit and does not change +default daily or PR smoke cases. It does not test cross-Workspace isolation or +high concurrency. + ## Sustained regression and baseline comparison [`regression_profiles/endurance.json`](regression_profiles/endurance.json) runs diff --git a/tools/regression_ci.py b/tools/regression_ci.py index 186c0883f..69b0cd9e1 100644 --- a/tools/regression_ci.py +++ b/tools/regression_ci.py @@ -21,7 +21,15 @@ from regression_suite import Suite from regression_traffic import owned_box_containers, read_endpoint_snapshot ROOT = Path(__file__).resolve().parents[1] -SUITE_BUDGETS = {'smoke': 60, 'scenarios': 300, 'daily': 1860, 'endurance': 900, 'api': 600, 'state': 300} +SUITE_BUDGETS = { + 'smoke': 60, + 'scenarios': 300, + 'daily': 1860, + 'endurance': 900, + 'api': 600, + 'state': 300, + 'isolation': 300, +} sys.path.insert(0, str(ROOT)) from tests.e2e.utils.config_factory import create_minimal_config, create_test_directories # noqa: E402 @@ -44,6 +52,10 @@ def suite_profile(name, run_id, groups=(), api_seed=1): profile.update(resources=[], cases=[], coverage_suite='api') profile['vars']['api_seed'] = api_seed add_cases(profile) + if name == 'isolation': + from regression_isolation_cases import isolation_profile + + profile = isolation_profile(profile) for resource in profile['resources']: if resource['kind'] == 'pipeline': resource['body'].setdefault( @@ -183,8 +195,8 @@ def run(args): groups = args.scenario_groups if groups is None: groups = list(DEFAULT_GROUPS) if args.suite in ('scenarios', 'daily') else [] - if args.suite == 'api': - groups = ['api'] + if args.suite in ('api', 'isolation'): + groups = [args.suite] if 'plugins' in groups and not {'misc/EssentialCommands', 'misc/FAQManager'} <= set(args.extra_plugins): groups.remove('plugins') profile, profile_path = suite_profile(args.suite, secrets.token_hex(8), groups, args.api_seed) diff --git a/tools/regression_contracts.json b/tools/regression_contracts.json index 87910bdb8..8e9299889 100644 --- a/tools/regression_contracts.json +++ b/tools/regression_contracts.json @@ -449,6 +449,9 @@ "area": "state", "kind": "blackbox", "priority": "P1", "description": "Knowledge documents survive restarts; replacement preserves control content and deletion removes stale results", "bindings": [{"profile": "state", "case": "durable_state_lifecycle", "checks": ["created_retrievable", "after_runtime_retrievable", "after_core_retrievable", "updated_retrievable", "updated_obsolete_absent", "persisted_update_retrievable", "persisted_update_obsolete_absent", "documents_deleted", "deleted_retrieval_empty"]}] - } + }, + {"id": "isolation.same_pipeline", "area": "isolation", "kind": "graybox", "priority": "P1", "description": "Concurrent sessions in one Pipeline retain their own request identity and prior history", "bindings": [{"profile": "isolation", "case": "isolation.same_pipeline", "checks": ["conversation_isolation", "overlap_turn_1", "overlap_turn_2", "wire_identity_history_once"]}]}, + {"id": "isolation.cross_pipeline", "area": "isolation", "kind": "graybox", "priority": "P1", "description": "Concurrent Pipelines retain their own model routing, Bot identity, and conversation history", "bindings": [{"profile": "isolation", "case": "isolation.cross_pipeline", "checks": ["conversation_isolation", "overlap_turn_1", "overlap_turn_2", "wire_identity_history_once"]}]}, + {"id": "isolation.cancel_neighbor", "area": "isolation", "kind": "graybox", "priority": "P1", "description": "Cancelling an active tool run leaves a concurrent neighbor completed and no delayed tool side effect", "bindings": [{"profile": "isolation", "case": "isolation.cancel_neighbor", "checks": ["cancel_isolation", "neighbor_active_at_cancel", "cancellation_overlapped_neighbor", "neighbor_identity_once", "no_late_side_effect", "box_idle"]}]} ] } diff --git a/tools/regression_execution_cases.py b/tools/regression_execution_cases.py index cfdbb372d..d130a250e 100644 --- a/tools/regression_execution_cases.py +++ b/tools/regression_execution_cases.py @@ -97,7 +97,7 @@ def _running_box(suite, session_id): 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'): +def _stream(suite, agent, scope, row, *, interrupt=None, box_session_id=None, frame_key='frames', before_cancel=None): 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(), @@ -130,6 +130,8 @@ def _stream(suite, agent, scope, row, *, interrupt=None, box_session_id=None, fr row['box_session_id'] = checked(row, 'exec_active', lambda: _running_box(suite, box_session_id)) row['box_before_interrupt'] = _sessions(suite).get(row['box_session_id']) if interrupt == 'cancel': + if before_cancel is not None: + before_cancel() break # Closing this public debug stream cancels its run. if frame.get('kind') in {'completed', 'error'}: return frames @@ -225,7 +227,7 @@ def _cleanup(suite, agent, scope, row, expected): raise RuntimeError('; '.join(([str(original)] if original else []) + errors)) -def run_case(suite, case, row): +def run_case(suite, case, row, *, before_cancel=None): scenario = case['scenario'] require(scenario in CASE_NAMES, 'Unknown local execution scenario') scope = 'qa-fault-' + uuid4().hex @@ -277,6 +279,7 @@ def run_case(suite, case, row): row, interrupt='cancel' if scenario == 'execution.cancel_cleanup' else 'timeout' if interrupt else None, box_session_id=box_session_id, + before_cancel=before_cancel, ) row['frames'] = frames run, events = _terminal(suite, agent, scope, row, expected) diff --git a/tools/regression_isolation_cases.py b/tools/regression_isolation_cases.py new file mode 100644 index 000000000..d1a62f9dd --- /dev/null +++ b/tools/regression_isolation_cases.py @@ -0,0 +1,328 @@ +"""Bounded concurrent conversations and cancellation isolation in one Workspace.""" + +from concurrent.futures import ThreadPoolExecutor +from contextlib import ExitStack, closing +import copy +import json +from pathlib import Path +import threading +import secrets +from uuid import uuid4 + +import regression_execution_cases as execution +from regression_api_checks import owned_resource +from regression_coverage import checked +from regression_lifecycle_cases import after_id, ledger +from regression_local_cases import record_run, require +from regression_smoke import check_run, completed_run +from regression_traffic import group_request + +CASE_NAMES = ('isolation.same_pipeline', 'isolation.cross_pipeline', 'isolation.cancel_neighbor') + + +def add_cases(profile): + require(profile.get('coverage_suite') == 'isolation', 'Isolation requires its dedicated suite') + profile['cases'] = [ + {'name': name, 'scenario': name, 'stage': 'isolation', 'expect': 'ISOLATION_OK'} for name in CASE_NAMES + ] + profile['scenario_groups'] = ['isolation'] + + +def isolation_profile(base): + profile = copy.deepcopy(base) + execution.add_cases(profile) + wanted = { + 'mock_provider', + 'chat_model', + 'slow_model', + 'chat_pipeline', + 'fault_interrupt_model', + 'fault_interrupt_check_model', + } + profile['resources'] = [item for item in profile['resources'] if item['id'] in wanted] + resources = {item['id']: item for item in profile['resources']} + resources['slow_model']['body']['name'] = 'mock-slow' + resources['chat_model']['body']['name'] = 'mock-slow-stream' + resources['chat_pipeline']['body']['config']['ai']['runner_config'][execution.RUNNER]['model']['primary'] = ( + '${slow_model}' + ) + other = copy.deepcopy(resources['chat_pipeline']) + other['id'], other['body']['name'] = 'isolation_pipeline', 'regression isolation peer' + other['body']['config']['ai']['runner_config'][execution.RUNNER]['model']['primary'] = '${chat_model}' + profile['resources'].append(other) + profile['coverage_suite'] = 'isolation' + add_cases(profile) + return profile + + +def lane(suite, bot, expected, model): + return { + 'group': 'qa-isolation-' + uuid4().hex, + 'previous': 'none', + 'model': model, + 'bot': bot, + 'case': { + 'expect': expected, + 'http_group': { + 'url': suite.variables['core_base_url'] + '/bots/' + bot + '/sync', + 'secret_env': 'LANGBOT_QA_GROUP_SECRET', + 'message': 'ai Run the isolation QA scenario', + }, + }, + } + + +def check_reply(parts, expected): + # /sync retains cumulative stream snapshots and drops their final/sequence metadata. + texts = [part.get('text') for part in parts if part.get('type') == 'Plain'] + require(len(texts) == len(parts) and bool(texts), 'Reply must contain only text snapshots') + require(all(isinstance(text, str) and expected.startswith(text) for text in texts), 'Reply contains foreign output') + require( + texts[-1] == expected and all(b.startswith(a) for a, b in zip(texts, texts[1:])), + 'Reply lost its final output or reversed its stream', + ) + + +def turn(suite, item, number, barrier, receipt): + receipt.update( + group=item['group'], + turn=number, + token=uuid4().hex, + after_id=after_id(suite), + model=item['model'], + expected=item['case']['expect'], + ) + barrier.wait(timeout=suite.timeout(10)) + response_error = None + try: + receipt['response'] = checked( + receipt, + 'response_identity', + lambda: group_request( + item['case'], + item['group'], + 'qa-isolation-member', + number, + receipt['token'], + item['previous'], + suite.env, + timeout=suite.timeout(45), + ), + ) + checked( + receipt, + 'exact_reply', + lambda: check_reply( + receipt['response']['data']['message'], item['case']['expect'] + ' GROUP_ACK=' + receipt['token'] + ), + ) + except Exception as exc: + response_error = exc + try: + with closing(ledger(suite)) as db: + run = completed_run( + db, suite.variables['workspace_uuid'], 'group_' + item['group'], receipt['after_id'], suite.timeout(20) + ) + checked(receipt, 'run_events', lambda: check_run(db, run, item['case'])) + receipt['interval'] = dict( + db.execute( + 'SELECT run_id, started_at, finished_at, bot_id FROM agent_run WHERE run_id=?', (run['run_id'],) + ).fetchone() + ) + checked( + receipt, + 'bot_identity', + lambda: require(receipt['interval']['bot_id'] == item['bot'], 'Run belongs to a different Bot'), + ) + record_run(suite, receipt, run['run_id'], 'completed') + except Exception as exc: + if response_error is not None: + receipt['ledger_error'] = str(exc) + raise response_error from exc + raise + if response_error is not None: + raise response_error + item['previous'] = receipt['token'] + return receipt + + +def overlapping(intervals): + require( + len(intervals) == 2 and all(x.get('started_at') and x.get('finished_at') for x in intervals), + 'Two persisted execution intervals are required', + ) + require( + max(x['started_at'] for x in intervals) < min(x['finished_at'] for x in intervals), + 'The requests did not execute concurrently', + ) + return intervals + + +def merge_receipts(row, receipts): + row.setdefault('requests', []).extend(receipts) + for receipt in receipts: + row.setdefault('scenario_runs', []).extend(receipt.get('scenario_runs', [])) + + +def verify_wire(suite, row, receipts): + path = Path(suite.path(suite.profile['smoke']['mock_records'])) + records = [json.loads(line) for line in path.read_text().splitlines()] + owned = {receipt['group'] for receipt in receipts} + from mock_llm import GROUP_TURN, group_ack + + matched = [] + for record in records: + messages = record.get('request', {}).get('messages', []) + starts = [i for i, message in enumerate(messages) if message.get('role') == 'user'] + if not starts: + continue + markers = list(GROUP_TURN.finditer(str(messages[starts[-1]].get('content', '')))) + if not markers or markers[-1][1] not in owned: + continue + require( + record['status'] == 200 and record.get('response_complete'), 'Owned model request failed or did not finish' + ) + expected = next((receipt for receipt in receipts if receipt['token'] == markers[-1][3]), None) + require(expected is not None, 'Model request has an unowned token') + require(record['request']['model'] == expected['model'], 'Conversation used an unexpected model') + acknowledgement = group_ack(messages, starts[-1]) + require( + record.get('response', {}).get('content') == expected['expected'] + acknowledgement, + 'Mock response does not identify the current request', + ) + matched.append(markers[-1][3]) + require( + sorted(matched) == sorted(receipt['token'] for receipt in receipts), + 'Model requests were duplicated, lost, or assigned to a different conversation', + ) + return {'requests': len(matched), 'history_isolated': True} + + +def parallel_history(suite, row, bots): + lanes = [lane(suite, *bot) for bot in bots] + with ThreadPoolExecutor(max_workers=2) as pool: + for number in (1, 2): + barrier, receipts = threading.Barrier(2), [{}, {}] + futures = [ + pool.submit(turn, suite, item, number, barrier, receipt) for item, receipt in zip(lanes, receipts) + ] + errors = [] + try: + for future in futures: + try: + future.result(timeout=suite.timeout(70)) + except Exception as exc: + errors.append(exc) + finally: + merge_receipts(row, receipts) + if errors: + row['request_errors'] = [str(error) for error in errors] + raise errors[0] + checked(row, f'overlap_turn_{number}', lambda: overlapping([receipt['interval'] for receipt in receipts])) + row['wire'] = checked(row, 'wire_identity_history_once', lambda: verify_wire(suite, row, row['requests'])) + + +def cancel_neighbor(suite, row, bot): + peer, receipt, future = lane(suite, *bot), {}, None + with ThreadPoolExecutor(max_workers=1) as pool: + + def before_cancel(): + nonlocal future + future = pool.submit(turn, suite, peer, 1, threading.Barrier(1), receipt) + + def active(): + with closing(ledger(suite)) as db: + matches = db.execute( + 'SELECT run_id, started_at FROM agent_run WHERE workspace_id=? AND conversation_id=? AND id>? ' + 'AND started_at IS NOT NULL AND finished_at IS NULL', + (suite.variables['workspace_uuid'], 'group_' + peer['group'], row['after_id']), + ).fetchall() + require(len(matches) <= 1, 'Neighbor submission created multiple active runs') + return dict(matches[0]) if matches else None + + row['neighbor_at_cancel'] = checked( + row, + 'neighbor_active_at_cancel', + lambda: execution._poll( + suite, + active, + 'Neighbor never became active before cancellation', + seconds=10, + ), + ) + require( + execution._sessions(suite).get(row['box_session_id'], {}).get('status') == 'running', + 'Target tool stopped before the neighbor became active', + ) + + try: + execution.run_case(suite, {'scenario': 'execution.cancel_cleanup'}, row, before_cancel=before_cancel) + finally: + if future is not None: + try: + future.result(timeout=suite.timeout(70)) + finally: + merge_receipts(row, [receipt]) + require(future is not None, 'Cancellation never reached its active execution checkpoint') + cancelled = next(run for run in row['scenario_runs'] if run['expected_status'] == 'cancelled') + with closing(ledger(suite)) as db: + interval = dict( + db.execute( + 'SELECT run_id, started_at, finished_at FROM agent_run WHERE run_id=?', (cancelled['run_id'],) + ).fetchone() + ) + checked(row, 'cancellation_overlapped_neighbor', lambda: overlapping([interval, receipt['interval']])) + row['wire'] = checked(row, 'neighbor_identity_once', lambda: verify_wire(suite, row, [receipt])) + + +def run_case(suite, case, row): + require(case['scenario'] in CASE_NAMES, 'Unknown isolation scenario') + secret = secrets.token_urlsafe(32) + suite.secrets.append(secret) + suite.env['LANGBOT_QA_GROUP_SECRET'] = secret + with ExitStack() as stack: + bots = [] + pipelines = [('chat_pipeline', 'LANGBOT_SLOW_OK', 'mock-slow')] + if case['scenario'] == 'isolation.cross_pipeline': + pipelines.append(('isolation_pipeline', 'LANGBOT_SLOW_STREAM_OK', 'mock-slow-stream')) + for pipeline, expected, model in pipelines: + body = { + 'name': 'qa-isolation-' + uuid4().hex, + 'adapter': 'http_bot', + 'enable': True, + 'adapter_config': { + 'inbound_secret': secret, + 'signature_required': True, + 'default_session_type': 'group', + 'callback_url': 'http://127.0.0.1:9/unused-sync-only', + 'callback_max_retries': 0, + }, + 'event_bindings': [ + { + 'event_pattern': 'message.received', + 'target_type': 'pipeline', + 'target_uuid': suite.variables[pipeline], + 'enabled': True, + 'filters': [], + } + ], + 'plugin_processors': [], + } + _, bot = stack.enter_context( + owned_resource( + suite, + row, + '/api/v1/platform/bots', + body, + item_key='bot', + list_key='bots', + ) + ) + bots.append((bot, expected, model)) + if case['scenario'] == 'isolation.cancel_neighbor': + checked(row, 'cancel_isolation', lambda: cancel_neighbor(suite, row, bots[0])) + else: + if case['scenario'] == 'isolation.same_pipeline': + bots = [bots[0], bots[0]] + checked(row, 'conversation_isolation', lambda: parallel_history(suite, row, bots)) + return {'result': 'ISOLATION_OK', 'owned_runs': len(row.get('scenario_runs', []))} diff --git a/tools/regression_local_cases.py b/tools/regression_local_cases.py index b935f4527..3f395c011 100644 --- a/tools/regression_local_cases.py +++ b/tools/regression_local_cases.py @@ -15,6 +15,7 @@ GROUPS = { 'lifecycle': 'regression_lifecycle_cases', 'plugins': 'regression_plugin_cases', 'api': 'regression_api_cases', + 'isolation': 'regression_isolation_cases', } diff --git a/tools/test_regression_coverage.py b/tools/test_regression_coverage.py index 8ccfe9daa..42d05187f 100644 --- a/tools/test_regression_coverage.py +++ b/tools/test_regression_coverage.py @@ -289,7 +289,7 @@ class Routes: name = binding['profile'] if name not in profiles: path = Path(__file__).with_name('regression_profiles') / f'{name}.json' - if name in ('api', 'state'): + if name in ('api', 'state', 'isolation'): from regression_ci import suite_profile profile, _ = suite_profile(name, 'catalog-test') diff --git a/tools/test_regression_execution_cases.py b/tools/test_regression_execution_cases.py index c6d28284b..8baf83cd5 100644 --- a/tools/test_regression_execution_cases.py +++ b/tools/test_regression_execution_cases.py @@ -280,12 +280,21 @@ class ExecutionCasesTests(unittest.TestCase): try: self.assertEqual(cases._stream(self.suite, 'agent', 'scope', {})[-1]['kind'], 'completed') row = {} + checkpoints = [] with ( patch.object(cases, '_running_box', return_value='box'), patch.object(cases, '_sessions', return_value={'box': {'status': 'running'}}), ): - cancelled = cases._stream(self.suite, 'agent', 'scope', row, interrupt='cancel') + cancelled = cases._stream( + self.suite, + 'agent', + 'scope', + row, + interrupt='cancel', + before_cancel=lambda: checkpoints.append(row['box_session_id']), + ) self.assertEqual(len(cancelled), 1) + self.assertEqual(checkpoints, ['box']) self.assertEqual(row['box_session_id'], 'box') frames.pop() with self.assertRaisesRegex(RuntimeError, 'outcome unknown'): diff --git a/tools/test_regression_isolation.py b/tools/test_regression_isolation.py new file mode 100644 index 000000000..c39449423 --- /dev/null +++ b/tools/test_regression_isolation.py @@ -0,0 +1,86 @@ +"""Challenge isolation oracles with serialized, duplicated, and crossed evidence.""" + +import copy +import json +from pathlib import Path +import tempfile +from types import SimpleNamespace +import unittest + +from regression_ci import suite_profile +from regression_isolation_cases import CASE_NAMES, check_reply, overlapping, verify_wire +from regression_local_cases import DEFAULT_GROUPS +from regression_scenarios import expanded_profile +from regression_suite import validate + + +class IsolationTests(unittest.TestCase): + def test_sync_stream_snapshots_cannot_hide_foreign_or_incomplete_output(self): + expected = 'LANGBOT_SLOW_OK GROUP_ACK=aaaa' + + def parts(texts): + return [{'type': 'Plain', 'text': text} for text in texts] + + check_reply(parts(['LANGBOT_', expected, expected]), expected) + for texts in ([], ['LANGBOT_'], [expected, 'LANGBOT_'], [expected, expected + ' GROUP_ACK=bbbb']): + with self.subTest(texts=texts), self.assertRaises(ValueError): + check_reply(parts(texts), expected) + + def test_dedicated_finite_profile_does_not_expand_daily(self): + profile, _ = suite_profile('isolation', 'test') + validate(profile) + self.assertEqual([case['scenario'] for case in profile['cases']], list(CASE_NAMES)) + self.assertNotIn('execution', profile) + self.assertNotIn('isolation', DEFAULT_GROUPS) + self.assertFalse(set(CASE_NAMES) & {case.get('scenario') for case in expanded_profile(DEFAULT_GROUPS)['cases']}) + with self.assertRaisesRegex(ValueError, 'dedicated suite'): + expanded_profile(['isolation']) + + def test_serial_execution_or_incomplete_timing_cannot_pass_overlap(self): + first = {'started_at': '2026-10-29 01:00:00', 'finished_at': '2026-10-29 01:00:02'} + second = {'started_at': '2026-10-29 01:00:01', 'finished_at': '2026-10-29 01:00:03'} + self.assertEqual(overlapping([first, second]), [first, second]) + for invalid in ({**second, 'started_at': first['finished_at']}, {**second, 'finished_at': None}): + with self.subTest(invalid=invalid), self.assertRaises(ValueError): + overlapping([first, invalid]) + + def test_wire_rejects_crossed_history_duplicates_wrong_model_and_wrong_reply(self): + receipt = {'group': 'test-a', 'token': 'bbbb', 'model': 'mock-slow', 'expected': 'LANGBOT_SLOW_OK'} + record = { + 'status': 200, + 'response_complete': True, + 'request': { + 'model': 'mock-slow', + 'messages': [ + {'role': 'user', 'content': 'LB_GROUP=test-a LB_TURN=1 LB_TOKEN=aaaa LB_PREVIOUS=none'}, + {'role': 'assistant', 'content': 'LANGBOT_SLOW_OK GROUP_ACK=aaaa'}, + {'role': 'user', 'content': 'LB_GROUP=test-a LB_TURN=2 LB_TOKEN=bbbb LB_PREVIOUS=aaaa'}, + ], + }, + 'response': {'content': 'LANGBOT_SLOW_OK GROUP_ACK=bbbb'}, + } + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / 'wire.jsonl' + suite = SimpleNamespace(profile={'smoke': {'mock_records': str(path)}}, path=lambda value: value) + + def check(rows): + path.write_text(''.join(json.dumps(row) + '\n' for row in rows)) + return verify_wire(suite, {}, [receipt]) + + self.assertEqual(check([record])['requests'], 1) + crossed = copy.deepcopy(record) + crossed['request']['messages'][0]['content'] = crossed['request']['messages'][0]['content'].replace( + 'test-a', 'test-b' + ) + wrong_model = copy.deepcopy(record) + wrong_model['request']['model'] = 'mock-chat' + wrong_reply = copy.deepcopy(record) + wrong_reply['response']['content'] = 'LANGBOT_SLOW_OK GROUP_ACK=aaaa' + incomplete = {**record, 'response_complete': False} + for rows in ([], [record, record], [crossed], [wrong_model], [wrong_reply], [incomplete]): + with self.subTest(rows=rows), self.assertRaises(ValueError): + check(rows) + + +if __name__ == '__main__': + unittest.main() From 7119cc94ba195702f9c4c927c17f43e2d4cc2928 Mon Sep 17 00:00:00 2001 From: huanghuoguoguo <1051233107@qq.com> Date: Fri, 30 Oct 2026 09:20:37 +0800 Subject: [PATCH 4/7] test(regression): add plugin lifecycle recovery job (#2812) Squash merge after resolving companion-suite conflicts and passing all required checks. --- tools/README.md | 29 ++ tools/regression_ci.py | 22 +- tools/regression_contracts.json | 22 ++ .../components/pages/probe/index.html | 1 + .../components/pages/probe/probe.py | 20 ++ .../components/pages/probe/probe.yaml | 12 + .../plugin_lifecycle/main.py | 12 + tools/regression_local_cases.py | 1 + tools/regression_plugin_lifecycle.py | 285 ++++++++++++++++++ tools/test_regression_coverage.py | 2 +- tools/test_regression_plugin_lifecycle.py | 118 ++++++++ 11 files changed, 517 insertions(+), 7 deletions(-) create mode 100644 tools/regression_fixtures/plugin_lifecycle/components/pages/probe/index.html create mode 100644 tools/regression_fixtures/plugin_lifecycle/components/pages/probe/probe.py create mode 100644 tools/regression_fixtures/plugin_lifecycle/components/pages/probe/probe.yaml create mode 100644 tools/regression_fixtures/plugin_lifecycle/main.py create mode 100644 tools/regression_plugin_lifecycle.py create mode 100644 tools/test_regression_plugin_lifecycle.py diff --git a/tools/README.md b/tools/README.md index d3c22e111..0c94b0223 100644 --- a/tools/README.md +++ b/tools/README.md @@ -291,6 +291,35 @@ quality. Run the Job plumbing checks on Linux with: uv run --no-sync python -m unittest discover -s tools -p 'test_*.py' -v ``` +## Local plugin lifecycle + +Run the separate plugin lifecycle Job on a disposable Linux stack: + +```sh +uv run --no-sync python tools/regression_ci.py --suite plugin-lifecycle \ + --lbctl /path/to/lbctl --output .qa/plugin-lifecycle-001 +``` + +Its three chains install a dependency-free local Page fixture, store a Unicode +marker and configure it, then exercise an update, a repeated installation and +invalid archive, or an initialization failure followed by explicitly reinstalling +a known good package and updating again. Assertions check completed task status, +the executing version, retained configuration and plugin storage, and unique +installation visibility. Cleanup verifies API removal and termination of observed +worker PIDs, including their process start times. The 600-second budget is a cap; +the Job stops when its finite cases finish. + +The current HTTP installation path does not promise automatic rollback after +initialization failure. This Job verifies explicit recovery; the separate +`plugins.upgrade_recovery` coverage obligation remains unclaimed. A failed +generation must be absent or reject Page calls. If Core instead restores the old +version automatically, that version must pass its full configuration, storage and +Page oracle. It uses the +existing runner's stack preparation and local HTTP uploads, installs only the +bundled fixtures without Marketplace access, and is not included in smoke or daily +by default. Cases, task outcomes, worker identities and cleanup journals are saved +under `gate/suite/`. + ## Finite core smoke Use [`regression_profiles/smoke.json`](regression_profiles/smoke.json) for a finite diff --git a/tools/regression_ci.py b/tools/regression_ci.py index 69b0cd9e1..11561cb35 100644 --- a/tools/regression_ci.py +++ b/tools/regression_ci.py @@ -29,6 +29,7 @@ SUITE_BUDGETS = { 'api': 600, 'state': 300, 'isolation': 300, + 'plugin-lifecycle': 600, } sys.path.insert(0, str(ROOT)) from tests.e2e.utils.config_factory import create_minimal_config, create_test_directories # noqa: E402 @@ -56,6 +57,11 @@ def suite_profile(name, run_id, groups=(), api_seed=1): from regression_isolation_cases import isolation_profile profile = isolation_profile(profile) + if name == 'plugin-lifecycle': + from regression_plugin_lifecycle import add_cases + + profile.update(resources=[], cases=[], coverage_suite='plugin-lifecycle') + add_cases(profile) for resource in profile['resources']: if resource['kind'] == 'pipeline': resource['body'].setdefault( @@ -177,11 +183,15 @@ def run(args): return {'service': name, 'old_pid': old_pid, 'new_pid': pids[name], 'old_exit_code': process.returncode} try: - packages = ( - prepare_source_plugins(root / 'plugins', args.plugins, args.extra_plugins) - if args.plugins is not None - else prepare_published_plugins(root / 'plugins', args.plugin_manifest) - ) + if args.suite == 'plugin-lifecycle': + (root / 'plugins').mkdir() + packages = {'version': 1, 'selection': 'local-fixtures', 'plugins': []} + else: + packages = ( + prepare_source_plugins(root / 'plugins', args.plugins, args.extra_plugins) + if args.plugins is not None + else prepare_published_plugins(root / 'plugins', args.plugin_manifest) + ) config_path = create_minimal_config(private, port=15301) create_test_directories(private) config = yaml.safe_load(config_path.read_text()) @@ -195,7 +205,7 @@ def run(args): groups = args.scenario_groups if groups is None: groups = list(DEFAULT_GROUPS) if args.suite in ('scenarios', 'daily') else [] - if args.suite in ('api', 'isolation'): + if args.suite in ('api', 'isolation', 'plugin-lifecycle'): groups = [args.suite] if 'plugins' in groups and not {'misc/EssentialCommands', 'misc/FAQManager'} <= set(args.extra_plugins): groups.remove('plugins') diff --git a/tools/regression_contracts.json b/tools/regression_contracts.json index 8e9299889..4bd65c3c3 100644 --- a/tools/regression_contracts.json +++ b/tools/regression_contracts.json @@ -256,6 +256,28 @@ "description": "Plugin upgrade failure preserves a usable installation", "bindings": [] }, + {"id": "plugins.upgrade_persistence", + "area": "plugins", "kind": "blackbox", "priority": "P1", + "description": "A local plugin update runs the new version with its existing config and stored data", + "bindings": [{"profile": "plugin-lifecycle", "case": "plugin_upgrade", "checks": ["upgrade_v2_task", "v2_identity", "v2_unique", "v2_state", "plugin_cleanup"]}], + "routes": ["POST /api/v1/plugins/install/local", "GET /api/v1/plugins//", "PUT /api/v1/plugins///config", "POST /api/v1/plugins///page-api"] + }, + {"id": "plugins.package_rejection", + "area": "plugins", "kind": "blackbox", "priority": "P1", + "description": "Reinstalling a local package preserves state and an invalid archive fails without disrupting that installation", + "bindings": [{"profile": "plugin-lifecycle", "case": "plugin_rejected_package", "checks": ["reinstall_v1_task", "after_reinstall_state", "invalid_archive_task", "after_invalid_archive_identity", "after_invalid_archive_state", "plugin_cleanup"]}] + }, + {"id": "plugins.explicit_recovery", + "area": "plugins", "kind": "blackbox", "priority": "P1", + "description": "A failed plugin initialization reports failure and reinstalling a known good package restores persisted state and later upgrades", + "bindings": [{"profile": "plugin-lifecycle", "case": "plugin_failed_activation", "checks": ["broken_v2_task", "broken_v2_error", "failed_generation_safe", "restore_v1_task", "restored_v1_state", "recovered_v3_task", "v3_state", "plugin_cleanup"]}] + }, + {"id": "plugins.observed_worker_cleanup", + "area": "plugins", "kind": "graybox", "priority": "P1", + "description": "Uninstall removes each fixture from the API and stops its observed worker process identities", + "bindings": [{"profile": "plugin-lifecycle", "case": "plugin_upgrade", "checks": ["uninstall_task", "uninstalled_from_list", "uninstalled_workers_stopped"]}, {"profile": "plugin-lifecycle", "case": "plugin_rejected_package", "checks": ["uninstall_task", "uninstalled_from_list", "uninstalled_workers_stopped"]}, {"profile": "plugin-lifecycle", "case": "plugin_failed_activation", "checks": ["uninstall_task", "uninstalled_from_list", "uninstalled_workers_stopped"]}], + "routes": ["DELETE /api/v1/plugins//"] + }, {"id": "agent.direct_execution", "area": "agent", "kind": "blackbox", "priority": "P1", "description": "Standalone Agent entrypoint completes through the public API", diff --git a/tools/regression_fixtures/plugin_lifecycle/components/pages/probe/index.html b/tools/regression_fixtures/plugin_lifecycle/components/pages/probe/index.html new file mode 100644 index 000000000..12e53e287 --- /dev/null +++ b/tools/regression_fixtures/plugin_lifecycle/components/pages/probe/index.html @@ -0,0 +1 @@ +Lifecycle probe diff --git a/tools/regression_fixtures/plugin_lifecycle/components/pages/probe/probe.py b/tools/regression_fixtures/plugin_lifecycle/components/pages/probe/probe.py new file mode 100644 index 000000000..46d86ac1a --- /dev/null +++ b/tools/regression_fixtures/plugin_lifecycle/components/pages/probe/probe.py @@ -0,0 +1,20 @@ +import os + +from langbot_plugin.api.definition.components.page import Page, PageRequest, PageResponse + + +class ProbePage(Page): + async def handle_api(self, request: PageRequest) -> PageResponse: + if request.endpoint != '/state' or request.method not in ('GET', 'POST'): + return PageResponse.fail('Unsupported lifecycle probe request') + if request.method == 'POST': + await self.plugin.set_plugin_storage('probe', request.body['token'].encode()) + token = (await self.plugin.get_plugin_storage('probe')).decode() + return PageResponse.ok( + { + 'version': self.plugin.version, + 'token': token, + 'config': self.plugin.get_config(), + 'pid': os.getpid(), + } + ) diff --git a/tools/regression_fixtures/plugin_lifecycle/components/pages/probe/probe.yaml b/tools/regression_fixtures/plugin_lifecycle/components/pages/probe/probe.yaml new file mode 100644 index 000000000..14e13e9d1 --- /dev/null +++ b/tools/regression_fixtures/plugin_lifecycle/components/pages/probe/probe.yaml @@ -0,0 +1,12 @@ +apiVersion: v1 +kind: Page +metadata: + name: probe + label: + en_US: Lifecycle probe +spec: + path: index.html +execution: + python: + path: probe.py + attr: ProbePage diff --git a/tools/regression_fixtures/plugin_lifecycle/main.py b/tools/regression_fixtures/plugin_lifecycle/main.py new file mode 100644 index 000000000..941769a28 --- /dev/null +++ b/tools/regression_fixtures/plugin_lifecycle/main.py @@ -0,0 +1,12 @@ +from langbot_plugin.api.definition.plugin import BasePlugin + +VERSION = '0.0.0' +FAIL_INITIALIZE = False + + +class LifecyclePlugin(BasePlugin): + version = VERSION + + async def initialize(self): + if FAIL_INITIALIZE: + raise RuntimeError('QA deliberate initialization failure') diff --git a/tools/regression_local_cases.py b/tools/regression_local_cases.py index 3f395c011..e48369d04 100644 --- a/tools/regression_local_cases.py +++ b/tools/regression_local_cases.py @@ -16,6 +16,7 @@ GROUPS = { 'plugins': 'regression_plugin_cases', 'api': 'regression_api_cases', 'isolation': 'regression_isolation_cases', + 'plugin-lifecycle': 'regression_plugin_lifecycle', } diff --git a/tools/regression_plugin_lifecycle.py b/tools/regression_plugin_lifecycle.py new file mode 100644 index 000000000..3ddaa1a4e --- /dev/null +++ b/tools/regression_plugin_lifecycle.py @@ -0,0 +1,285 @@ +"""Exercise local plugin generations through HTTP tasks and real Page invocations.""" + +import io +import json +from pathlib import Path +import time +import urllib.request +from uuid import uuid4 +import zipfile + +import yaml + +from regression_api_checks import check, request +from regression_coverage import checked +from regression_local_cases import api, auth_headers, endpoint, require +from regression_resource_cases import cleanup, journal + +CASE_NAMES = ('plugin_upgrade', 'plugin_rejected_package', 'plugin_failed_activation') +FIXTURE = Path(__file__).with_name('regression_fixtures') / 'plugin_lifecycle' +AUTHOR = 'regression-qa' + + +def add_cases(profile): + require(profile.get('coverage_suite') == 'plugin-lifecycle', 'Plugin lifecycle requires its dedicated suite') + profile['scenario_groups'] = ['plugin-lifecycle'] + profile['cases'].extend( + {'name': name, 'scenario': name, 'stage': 'plugins', 'expect': 'PLUGIN_LIFECYCLE_OK'} for name in CASE_NAMES + ) + + +def package(name, version, *, broken=False): + manifest = { + 'apiVersion': 'v1', + 'kind': 'Plugin', + 'metadata': { + 'author': AUTHOR, + 'name': name, + 'version': version, + 'label': {'en_US': 'Lifecycle probe'}, + 'description': {'en_US': 'Local regression fixture'}, + 'icon': 'icon.svg', + }, + 'spec': { + 'config': [{'name': 'marker', 'type': 'string', 'default': '', 'label': {'en_US': 'Marker'}}], + 'components': {'Page': {'fromDirs': [{'path': 'components/pages/', 'maxDepth': 2}]}}, + }, + 'execution': {'python': {'path': 'main.py', 'attr': 'LifecyclePlugin'}}, + } + buffer = io.BytesIO() + with zipfile.ZipFile(buffer, 'w', zipfile.ZIP_DEFLATED) as archive: + archive.writestr('manifest.yaml', yaml.safe_dump(manifest)) + archive.writestr('icon.svg', '') + for path in sorted(FIXTURE.rglob('*')): + if not path.is_file() or '__pycache__' in path.parts: + continue + content = path.read_text() + if path.name == 'main.py': + content = content.replace("VERSION = '0.0.0'", f'VERSION = {version!r}') + if broken: + content = content.replace('FAIL_INITIALIZE = False', 'FAIL_INITIALIZE = True') + archive.writestr(str(path.relative_to(FIXTURE)), content) + return buffer.getvalue() + + +def wait_task(suite, row, task_id, expected, label): + deadline = time.monotonic() + suite.timeout(90) + require(isinstance(task_id, int), 'Plugin operation did not return a task ID') + while True: + task = request(suite, row, 'GET', f'/api/v1/system/tasks/{task_id}') + require(task.get('id') == task_id, 'Plugin task identity changed') + status = task.get('status') + require(status in {'running', 'succeeded', 'failed', 'cancelled'}, 'Malformed plugin task status') + if status != 'running': + row.setdefault('plugin_tasks', []).append({'label': label, **task}) + check(row, label + '_task', status == expected, f'{label}: expected {expected}, got {status}') + if expected == 'failed': + check(row, label + '_error', bool(task.get('error')), 'Failed task omitted its error') + return + if time.monotonic() >= deadline: + raise TimeoutError(f'{label}: plugin task did not finish') + time.sleep(0.2) + + +def upload(suite, row, content, name, *, expected='succeeded'): + boundary = 'qa' + uuid4().hex + body = ( + ( + f'--{boundary}\r\nContent-Disposition: form-data; name="file"; filename="{name}.lbpkg"\r\n' + 'Content-Type: application/octet-stream\r\n\r\n' + ).encode() + + content + + f'\r\n--{boundary}--\r\n'.encode() + ) + headers = auth_headers(suite) + headers['Content-Type'] = 'multipart/form-data; boundary=' + boundary + path = '/api/v1/plugins/install/local' + call = urllib.request.Request(endpoint(suite, path), data=body, headers=headers, method='POST') + opener = urllib.request.build_opener(urllib.request.ProxyHandler({})) + with opener.open(call, timeout=suite.timeout(90)) as response: + result = json.load(response) + check( + row, + name + '_upload', + response.status == 200 and result.get('code') == 0, + 'Local plugin upload was not accepted', + ) + wait_task(suite, row, result['data']['task_id'], expected, name) + + +def process_identity(pid): + try: + # Start time distinguishes an old worker from a reused PID. + return Path(f'/proc/{pid}/stat').read_text().rsplit(')', 1)[1].split()[19] + except FileNotFoundError: + return None + + +def metadata(plugin): + return plugin['manifest']['manifest']['metadata'] + + +def failed_generation_safe(suite, row, path, verify_good): + def verify(): + status, response = api(suite, 'GET', path) + row['failed_generation'] = {'http_status': status, 'plugin': response.get('data', {}).get('plugin')} + if status == 404: + require(response.get('code') not in (None, 0), 'Missing plugin returned a successful business status') + return + require(status == 200 and response.get('code') == 0, 'Failed generation cannot be observed') + info = response.get('data', {}).get('plugin') + require(isinstance(info, dict), 'Failed generation metadata is missing') + identity = metadata(info) + require( + identity.get('author') == AUTHOR and identity.get('name') == path.rsplit('/', 1)[1], + 'Failed generation identity changed', + ) + if identity.get('version') == '0.0.1': + verify_good() # A future automatic restore must pass the complete old-version state oracle. + return + require( + identity.get('version') == '0.0.2' and info.get('status') in {'unmounted', 'mounted'}, + 'Failed plugin generation is initialized or has an unexpected version/state', + ) + page_status, page = api( + suite, + 'POST', + path + '/page-api', + { + 'page_id': 'probe', + 'endpoint': '/state', + 'method': 'GET', + 'body': None, + }, + ) + row['failed_generation']['page_status'] = page_status + require( + page_status in {400, 404, 409, 500, 503} and page.get('code') not in (None, 0), + 'Failed plugin generation remains callable', + ) + + checked(row, 'failed_generation_safe', verify) + + +def workers_stopped(suite, workers): + deadline = time.monotonic() + suite.timeout(10) + while any(process_identity(int(pid)) == started for pid, started in workers.items()): + if time.monotonic() >= deadline: + raise ValueError('An observed plugin worker remains after uninstall or upgrade') + time.sleep(0.1) + + +def run_case(suite, case, row): + name = 'lifecycle-' + uuid4().hex + path = f'/api/v1/plugins/{AUTHOR}/{name}' + owned = {'kind': 'plugin', 'author': AUTHOR, 'name': name, 'state': 'creating'} + row.setdefault('owned_resources', []).append(owned) + journal(suite, row) + workers = row['plugin_workers'] = {} + marker = 'persistent-中文-' + uuid4().hex + + def installed(version, label, *, write=False): + deadline = time.monotonic() + suite.timeout(45) + while True: + status, response = api(suite, 'GET', path) + info = response.get('data', {}).get('plugin', {}) + if status == 200 and info.get('status') == 'initialized': + break + require(status in {200, 404}, 'Plugin readiness query failed') + if time.monotonic() >= deadline: + raise TimeoutError('Plugin did not become ready after operation') + time.sleep(0.2) + identity = metadata(info) + check( + row, + label + '_identity', + identity.get('author') == AUTHOR + and identity.get('name') == name + and identity.get('version') == version + and info.get('enabled') is True, + 'Plugin generation differs', + ) + listed = request(suite, row, 'GET', '/api/v1/plugins')['plugins'] + check( + row, + label + '_unique', + sum(metadata(p)['author'] == AUTHOR and metadata(p)['name'] == name for p in listed) == 1, + 'Plugin generation is missing or duplicated', + ) + state = request( + suite, + row, + 'POST', + path + '/page-api', + { + 'page_id': 'probe', + 'endpoint': '/state', + 'method': 'POST' if write else 'GET', + 'body': {'token': marker} if write else None, + }, + ) + check( + row, + label + '_state', + state.get('version') == version + and state.get('token') == marker + and state.get('config') == {'marker': marker}, + 'Plugin call lost version, stored data, or config', + ) + pid = state.get('pid') + require(isinstance(pid, int) and pid > 0, 'Plugin did not return its worker PID') + started = process_identity(pid) + require(started is not None, 'Observed plugin worker already disappeared') + old = {key: value for key, value in workers.items() if key != str(pid)} + workers[str(pid)] = started + checked(row, label + '_old_workers_stopped', lambda: workers_stopped(suite, old)) + return state + + def remove(): + status, response = api(suite, 'GET', path) + if status != 404: + require(status == 200 and response.get('code') == 0, 'Cannot confirm owned plugin before cleanup') + info = metadata(response['data']['plugin']) + require(info.get('author') == AUTHOR and info.get('name') == name, 'Plugin cleanup identity changed') + # A failed generation can have a durable installation without a registered worker. + task = request(suite, row, 'DELETE', path + '?delete_data=true') + wait_task(suite, row, task['task_id'], 'succeeded', 'uninstall') + request(suite, row, 'GET', path, expected=404) + listed = request(suite, row, 'GET', '/api/v1/plugins')['plugins'] + check( + row, + 'uninstalled_from_list', + not any(metadata(p)['author'] == AUTHOR and metadata(p)['name'] == name for p in listed), + 'Uninstalled plugin remains listed', + ) + checked(row, 'uninstalled_workers_stopped', lambda: workers_stopped(suite, workers)) + owned['state'] = 'deleted' + + request(suite, row, 'GET', path, expected=404) + try: + first = package(name, '0.0.1') + upload(suite, row, first, 'install_v1') + owned['state'] = 'installed' + journal(suite, row) + request(suite, row, 'PUT', path + '/config', {'marker': marker}) + installed('0.0.1', 'v1', write=True) + if case['scenario'] == 'plugin_upgrade': + upload(suite, row, package(name, '0.0.2'), 'upgrade_v2') + installed('0.0.2', 'v2') + elif case['scenario'] == 'plugin_rejected_package': + upload(suite, row, first, 'reinstall_v1') + installed('0.0.1', 'after_reinstall') + upload(suite, row, b'not a plugin zip', 'invalid_archive', expected='failed') + installed('0.0.1', 'after_invalid_archive') + elif case['scenario'] == 'plugin_failed_activation': + upload(suite, row, package(name, '0.0.2', broken=True), 'broken_v2', expected='failed') + failed_generation_safe(suite, row, path, lambda: installed('0.0.1', 'automatic_restore_v1')) + upload(suite, row, first, 'restore_v1') + installed('0.0.1', 'restored_v1') + upload(suite, row, package(name, '0.0.3'), 'recovered_v3') + installed('0.0.3', 'v3') + else: + raise ValueError('Unknown plugin lifecycle case') + return {'result': case['expect']} + finally: + cleanup(suite, row, 'plugin_cleanup', remove) diff --git a/tools/test_regression_coverage.py b/tools/test_regression_coverage.py index 42d05187f..f701cebce 100644 --- a/tools/test_regression_coverage.py +++ b/tools/test_regression_coverage.py @@ -289,7 +289,7 @@ class Routes: name = binding['profile'] if name not in profiles: path = Path(__file__).with_name('regression_profiles') / f'{name}.json' - if name in ('api', 'state', 'isolation'): + if name in ('api', 'state', 'isolation', 'plugin-lifecycle'): from regression_ci import suite_profile profile, _ = suite_profile(name, 'catalog-test') diff --git a/tools/test_regression_plugin_lifecycle.py b/tools/test_regression_plugin_lifecycle.py new file mode 100644 index 000000000..467ae1a20 --- /dev/null +++ b/tools/test_regression_plugin_lifecycle.py @@ -0,0 +1,118 @@ +"""Challenge plugin lifecycle task verdicts and worker cleanup evidence.""" + +from types import SimpleNamespace +import unittest +from unittest.mock import Mock, patch + +from regression_ci import suite_profile +from regression_local_cases import DEFAULT_GROUPS +from regression_plugin_lifecycle import failed_generation_safe, wait_task, workers_stopped +from regression_suite import validate + + +class PluginLifecycleTests(unittest.TestCase): + def test_failed_generation_cannot_pass_by_http_status_alone(self): + path = '/api/v1/plugins/regression-qa/probe' + + def plugin(version, status): + return ( + 200, + { + 'code': 0, + 'data': { + 'plugin': { + 'status': status, + 'manifest': { + 'manifest': { + 'metadata': {'author': 'regression-qa', 'name': 'probe', 'version': version}, + } + }, + } + }, + }, + ) + + for responses in ( + [(404, {'code': 0})], + [plugin('0.0.2', 'initialized')], + [plugin('0.0.9', 'mounted')], + [plugin('0.0.2', 'mounted'), (200, {'code': 0, 'data': {}})], + [plugin('0.0.2', 'mounted'), (500, {'code': 0})], + ): + row = {} + with self.subTest(responses=responses), patch('regression_plugin_lifecycle.api', side_effect=responses): + with self.assertRaises(ValueError): + failed_generation_safe(None, row, path, Mock()) + self.assertEqual(row['assertions']['failed_generation_safe'], 'failed') + + for responses in ( + [(404, {'code': -1})], + [plugin('0.0.2', 'mounted'), (400, {'code': -1})], + ): + row, verify_good = {}, Mock() + with self.subTest(responses=responses), patch('regression_plugin_lifecycle.api', side_effect=responses): + failed_generation_safe(None, row, path, verify_good) + self.assertEqual(row['assertions']['failed_generation_safe'], 'passed') + verify_good.assert_not_called() + + for state_error in (None, ValueError('Restored plugin lost configuration or stored data')): + row, verify_good = {}, Mock(side_effect=state_error) + with ( + self.subTest(state_error=state_error), + patch('regression_plugin_lifecycle.api', return_value=plugin('0.0.1', 'initialized')), + ): + if state_error: + with self.assertRaisesRegex(ValueError, 'lost configuration'): + failed_generation_safe(None, row, path, verify_good) + self.assertEqual(row['assertions']['failed_generation_safe'], 'failed') + else: + failed_generation_safe(None, row, path, verify_good) + self.assertEqual(row['assertions']['failed_generation_safe'], 'passed') + verify_good.assert_called_once_with() + + def test_job_is_opt_in_and_does_not_expand_daily(self): + profile, _ = suite_profile('plugin-lifecycle', 'run') + validate(profile) + self.assertEqual(len(profile['cases']), 3) + self.assertEqual(profile['resources'], []) + self.assertNotIn('plugin-lifecycle', DEFAULT_GROUPS) + for name in ('smoke', 'daily'): + existing, _ = suite_profile(name, 'run') + self.assertFalse(any(c.get('scenario', '').startswith('plugin_') for c in existing['cases'])) + + def test_expected_failure_requires_a_failed_task_not_http_success(self): + suite = SimpleNamespace(timeout=lambda seconds: seconds) + for task in ( + {'id': 7, 'status': 'succeeded'}, + {'id': 7, 'status': 'cancelled'}, + {'id': 7, 'status': 'failed', 'error': None}, + {'id': 8, 'status': 'failed', 'error': {'type': 'task_failed'}}, + ): + with self.subTest(task=task), patch('regression_plugin_lifecycle.request', return_value=task): + with self.assertRaises(ValueError): + wait_task(suite, {}, 7, 'failed', 'broken_upgrade') + row = {} + task = {'id': 7, 'status': 'failed', 'error': {'type': 'task_failed'}} + with patch('regression_plugin_lifecycle.request', return_value=task): + wait_task(suite, row, 7, 'failed', 'broken_upgrade') + self.assertEqual(row['assertions']['broken_upgrade_task'], 'passed') + self.assertEqual(row['plugin_tasks'][0]['status'], 'failed') + + def test_live_worker_is_a_cleanup_failure_but_reused_pid_is_not(self): + suite = SimpleNamespace(timeout=lambda seconds: 0) + with patch('regression_plugin_lifecycle.process_identity', return_value='same-start'): + with self.assertRaisesRegex(ValueError, 'worker remains'): + workers_stopped(suite, {'100': 'same-start'}) + for identity in (None, 'new-start'): + with patch('regression_plugin_lifecycle.process_identity', return_value=identity): + workers_stopped(suite, {'100': 'same-start'}) + + def test_unfinished_task_cannot_be_accepted(self): + suite = SimpleNamespace(timeout=lambda seconds: 0) + with patch('regression_plugin_lifecycle.request', Mock(return_value={'id': 7, 'status': 'running'})): + with self.assertRaisesRegex(TimeoutError, 'did not finish'): + wait_task(suite, {}, 7, 'succeeded', 'install') + + +if __name__ == '__main__': + unittest.main() From 98a242320b42992bbd9ac04cc568279aa5acf945 Mon Sep 17 00:00:00 2001 From: huanghuoguoguo <1051233107@qq.com> Date: Fri, 30 Oct 2026 09:33:00 +0800 Subject: [PATCH 5/7] test(cli): add compatibility job with semantic assertions (#2813) Squash merge after rebasing onto the merged companion suites and passing all required checks. --- tools/README.md | 20 +++++ tools/regression_ci.py | 7 +- tools/regression_cli_cases.py | 115 +++++++++++++++++++++++++++++ tools/regression_contracts.json | 5 +- tools/regression_local_cases.py | 1 + tools/regression_suite.py | 5 ++ tools/test_regression_cli_cases.py | 88 ++++++++++++++++++++++ tools/test_regression_coverage.py | 2 +- 8 files changed, 240 insertions(+), 3 deletions(-) create mode 100644 tools/regression_cli_cases.py create mode 100644 tools/test_regression_cli_cases.py diff --git a/tools/README.md b/tools/README.md index 0c94b0223..2e3f42a94 100644 --- a/tools/README.md +++ b/tools/README.md @@ -618,6 +618,26 @@ or latency thresholds. `gate/suite/cases.json` retains per-request evidence and default daily or PR smoke cases. It does not test cross-Workspace isolation or high concurrency. +## CLI compatibility check + +Run on the same disposable Linux host: + +```sh +uv run --no-sync python tools/regression_ci.py --suite cli \ + --lbctl /path/to/lbctl --plugins plugins --output .qa/cli-001 +``` + +The finite job checks +Workspace identity, capability discovery, Pipeline create/read/update/delete, +and invalid credentials. It uses the supplied CLI binary and records its version; +select the latest stable release when validating the evolving CLI. + +Assertions allow additional output fields and changes to human-readable wording. +Workspace identity, created resource identity, actual backend state, cleanup and +nonzero exit on authentication failure remain required. HTTP readback provides +an independent oracle, so an exit code of zero alone cannot pass a mutation. +The suite is opt-in and adds no daily schedule or PR gate. + ## Sustained regression and baseline comparison [`regression_profiles/endurance.json`](regression_profiles/endurance.json) runs diff --git a/tools/regression_ci.py b/tools/regression_ci.py index 11561cb35..a9ad01412 100644 --- a/tools/regression_ci.py +++ b/tools/regression_ci.py @@ -30,6 +30,7 @@ SUITE_BUDGETS = { 'state': 300, 'isolation': 300, 'plugin-lifecycle': 600, + 'cli': 180, } sys.path.insert(0, str(ROOT)) from tests.e2e.utils.config_factory import create_minimal_config, create_test_directories # noqa: E402 @@ -62,6 +63,10 @@ def suite_profile(name, run_id, groups=(), api_seed=1): profile.update(resources=[], cases=[], coverage_suite='plugin-lifecycle') add_cases(profile) + if name == 'cli': + from regression_cli_cases import configure + + configure(profile) for resource in profile['resources']: if resource['kind'] == 'pipeline': resource['body'].setdefault( @@ -205,7 +210,7 @@ def run(args): groups = args.scenario_groups if groups is None: groups = list(DEFAULT_GROUPS) if args.suite in ('scenarios', 'daily') else [] - if args.suite in ('api', 'isolation', 'plugin-lifecycle'): + if args.suite in ('api', 'isolation', 'plugin-lifecycle', 'cli'): groups = [args.suite] if 'plugins' in groups and not {'misc/EssentialCommands', 'misc/FAQManager'} <= set(args.extra_plugins): groups.remove('plugins') diff --git a/tools/regression_cli_cases.py b/tools/regression_cli_cases.py new file mode 100644 index 000000000..175109a15 --- /dev/null +++ b/tools/regression_cli_cases.py @@ -0,0 +1,115 @@ +"""Published CLI compatibility checks with strict backend state and flexible presentation.""" + +import subprocess + +from regression_api_checks import check, request, tracked_resource, unique_name +from regression_coverage import checked +from regression_local_cases import require +from regression_resource_cases import journal + +CASE_NAMES = ('cli.identity', 'cli.pipeline_roundtrip', 'cli.invalid_credential') + + +def configure(profile): + profile.update(resources=[], cases=[], coverage_suite='cli', scenario_groups=['cli']) + profile['cases'] = [ + {'name': name, 'scenario': name, 'stage': 'cli', 'expect': 'CLI_COMPAT_OK'} for name in CASE_NAMES + ] + + +def identity(suite, row): + row['cli_version'] = suite.call(['version']) + result = suite.call(['whoami']) + check( + row, + 'workspace_identity', + result.get('workspace_uuid') == suite.variables['workspace_uuid'], + 'CLI resolved a different Workspace', + ) + capabilities = suite.call(['capabilities']) + check(row, 'capability_discovery', bool(capabilities), 'CLI capability discovery returned no data') + + +def pipeline_roundtrip(suite, row): + name, collection = unique_name(), '/api/v1/pipelines' + with tracked_resource(suite, row, collection, name, item_key='pipeline', list_key='pipelines') as owned: + created = suite.call(['pipeline', 'apply'], {'name': name, 'description': 'CLI fixture 中文'}) + matches = [p for p in request(suite, row, 'GET', collection)['pipelines'] if p['name'] == name] + require(len(matches) == 1, 'CLI create did not persist exactly one owned Pipeline') + uid = matches[0]['uuid'] + owned.update(uuid=uid, state='created') + journal(suite, row) + path = collection + '/' + uid + check( + row, + 'create_identity', + created.get('pipeline', created).get('uuid') == uid, + 'CLI create omitted the created resource identity', + ) + current = suite.call(['pipeline', 'get', uid]) + check( + row, + 'cli_get', + current.get('pipeline', current).get('uuid') == uid, + 'CLI get omitted the owned entity identity', + ) + listed = suite.call(['pipeline', 'list']) + items = listed.get('pipelines', []) if isinstance(listed, dict) else listed + check( + row, + 'cli_list', + isinstance(items, list) and any(item.get('uuid') == uid for item in items), + 'CLI list omitted the owned entity identity', + ) + before = request(suite, row, 'GET', path)['pipeline'] + check(row, 'create_state', before['description'] == 'CLI fixture 中文', 'CLI create changed submitted content') + suite.call( + ['pipeline', 'apply'], + { + 'uuid': uid, + 'name': name + '-updated', + 'description': 'CLI updated 🌍', + 'config': before['config'], + }, + ) + after = request(suite, row, 'GET', path)['pipeline'] + check( + row, + 'update_state', + after['name'] == name + '-updated' + and after['description'] == 'CLI updated 🌍' + and after['config'] == before['config'], + 'CLI update lost submitted or existing configuration', + ) + suite.call(['pipeline', 'delete', uid, '--yes']) + request(suite, row, 'GET', path, expected=404) + + +def invalid_credential(suite, row): + source = suite.connection.get('credential_source', '') + require(source.startswith('env:'), 'CLI credential probe requires the disposable environment credential') + result = subprocess.run( + [*suite.cli, 'pipeline', 'list'], + capture_output=True, + text=True, + timeout=suite.timeout(30), + env={**suite.env, source.removeprefix('env:'): 'qa-invalid-credential'}, + ) + row['rejected_exit_code'] = result.returncode + # Exit code is stable; error wording and JSON error details may evolve. + check(row, 'credential_rejected', result.returncode != 0, 'CLI reported success with an invalid credential') + check(row, 'error_signal', bool(result.stdout.strip() or result.stderr.strip()), 'CLI failure had no diagnostic') + suite.call(['pipeline', 'list']) + + +def run_case(suite, case, row): + checked( + row, + 'cli_chain', + lambda: { + 'cli.identity': identity, + 'cli.pipeline_roundtrip': pipeline_roundtrip, + 'cli.invalid_credential': invalid_credential, + }[case['scenario']](suite, row), + ) + return {'result': case['expect']} diff --git a/tools/regression_contracts.json b/tools/regression_contracts.json index 4bd65c3c3..11738735d 100644 --- a/tools/regression_contracts.json +++ b/tools/regression_contracts.json @@ -474,6 +474,9 @@ }, {"id": "isolation.same_pipeline", "area": "isolation", "kind": "graybox", "priority": "P1", "description": "Concurrent sessions in one Pipeline retain their own request identity and prior history", "bindings": [{"profile": "isolation", "case": "isolation.same_pipeline", "checks": ["conversation_isolation", "overlap_turn_1", "overlap_turn_2", "wire_identity_history_once"]}]}, {"id": "isolation.cross_pipeline", "area": "isolation", "kind": "graybox", "priority": "P1", "description": "Concurrent Pipelines retain their own model routing, Bot identity, and conversation history", "bindings": [{"profile": "isolation", "case": "isolation.cross_pipeline", "checks": ["conversation_isolation", "overlap_turn_1", "overlap_turn_2", "wire_identity_history_once"]}]}, - {"id": "isolation.cancel_neighbor", "area": "isolation", "kind": "graybox", "priority": "P1", "description": "Cancelling an active tool run leaves a concurrent neighbor completed and no delayed tool side effect", "bindings": [{"profile": "isolation", "case": "isolation.cancel_neighbor", "checks": ["cancel_isolation", "neighbor_active_at_cancel", "cancellation_overlapped_neighbor", "neighbor_identity_once", "no_late_side_effect", "box_idle"]}]} + {"id": "isolation.cancel_neighbor", "area": "isolation", "kind": "graybox", "priority": "P1", "description": "Cancelling an active tool run leaves a concurrent neighbor completed and no delayed tool side effect", "bindings": [{"profile": "isolation", "case": "isolation.cancel_neighbor", "checks": ["cancel_isolation", "neighbor_active_at_cancel", "cancellation_overlapped_neighbor", "neighbor_identity_once", "no_late_side_effect", "box_idle"]}]}, + {"id": "cli.identity", "area": "cli", "kind": "blackbox", "priority": "P2", "description": "CLI resolves the intended Workspace and discovers capabilities without freezing presentation", "bindings": [{"profile": "cli", "case": "cli.identity", "checks": ["cli_chain", "workspace_identity", "capability_discovery"]}]}, + {"id": "cli.pipeline_roundtrip", "area": "cli", "kind": "blackbox", "priority": "P2", "description": "CLI mutations preserve submitted and existing Pipeline state, with independent HTTP readback", "bindings": [{"profile": "cli", "case": "cli.pipeline_roundtrip", "checks": ["cli_chain", "create_identity", "cli_get", "cli_list", "create_state", "update_state"]}]}, + {"id": "cli.invalid_credential", "area": "cli", "kind": "blackbox", "priority": "P2", "description": "CLI rejects invalid credentials with a nonzero exit and a diagnostic signal", "bindings": [{"profile": "cli", "case": "cli.invalid_credential", "checks": ["cli_chain", "credential_rejected", "error_signal"]}]} ] } diff --git a/tools/regression_local_cases.py b/tools/regression_local_cases.py index e48369d04..0860bd9fc 100644 --- a/tools/regression_local_cases.py +++ b/tools/regression_local_cases.py @@ -17,6 +17,7 @@ GROUPS = { 'api': 'regression_api_cases', 'isolation': 'regression_isolation_cases', 'plugin-lifecycle': 'regression_plugin_lifecycle', + 'cli': 'regression_cli_cases', } diff --git a/tools/regression_suite.py b/tools/regression_suite.py index ba5bd0073..98e614631 100644 --- a/tools/regression_suite.py +++ b/tools/regression_suite.py @@ -568,6 +568,11 @@ class Suite: error = error.replace(secret, '[REDACTED]') raise RuntimeError(error[:3000]) response = json.loads(result.stdout) + if not isinstance(response, dict) or response.get('ok') is False or response.get('error'): + error = result.stdout + for secret in self.secrets: + error = error.replace(secret, '[REDACTED]') + raise RuntimeError('CLI returned a failure response: ' + error[:3000]) if args == ['whoami']: self.connection = response.get('meta') or {} return response['data'] diff --git a/tools/test_regression_cli_cases.py b/tools/test_regression_cli_cases.py new file mode 100644 index 000000000..d22d0a226 --- /dev/null +++ b/tools/test_regression_cli_cases.py @@ -0,0 +1,88 @@ +"""Keep CLI presentation flexible while rejecting wrong identity and false success.""" + +from pathlib import Path +from types import SimpleNamespace +import tempfile +import json +import unittest +from unittest.mock import Mock, patch + +from regression_cli_cases import identity, invalid_credential, pipeline_roundtrip +from regression_ci import suite_profile +from regression_suite import Suite, validate + + +class CLICompatibilityTests(unittest.TestCase): + def test_wrong_workspace_fails_even_when_cli_exits_successfully(self): + suite = SimpleNamespace( + call=Mock(side_effect=[{'version': 'future', 'new_field': True}, {'workspace_uuid': 'wrong'}]), + variables={'workspace_uuid': 'expected'}, + ) + row = {} + with self.assertRaisesRegex(ValueError, 'different Workspace'): + identity(suite, row) + self.assertEqual(row['assertions']['workspace_identity'], 'failed') + + def test_invalid_credential_must_have_nonzero_exit(self): + suite = SimpleNamespace( + connection={'credential_source': 'env:QA_KEY'}, + cli=['lbctl', '-o', 'json'], + env={'QA_KEY': 'secret'}, + timeout=lambda value: value, + ) + with patch( + 'regression_cli_cases.subprocess.run', return_value=SimpleNamespace(returncode=0, stdout='{}', stderr='') + ): + with self.assertRaisesRegex(ValueError, 'success with an invalid credential'): + invalid_credential(suite, {}) + + def test_successful_cli_create_without_backend_state_fails(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) + suite = Suite( + {'cli': {'config': 'unused'}, 'cases': [], 'vars': {'workspace_uuid': 'test'}}, path / 'p.json', path + ) + suite.call = Mock(return_value={'uuid': 'reported-but-not-created'}) + row = {} + with ( + patch('regression_cli_cases.request', return_value={'pipelines': []}), + patch('regression_api_checks.api', return_value=(200, {'code': 0, 'data': {'pipelines': []}})), + ): + with self.assertRaisesRegex(ValueError, 'exactly one owned Pipeline'): + pipeline_roundtrip(suite, row) + self.assertEqual(row['owned_resources'][0]['state'], 'absent') + + def test_zero_exit_does_not_hide_an_explicit_failure_envelope(self): + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) + suite = Suite({'cli': {'config': 'unused'}, 'cases': []}, path / 'p.json', path, env={}) + for response in ( + {'ok': False, 'data': {'requested_uuid': 'owned'}}, + {'error': {'type': 'failed'}, 'data': {'uuid': 'owned'}}, + ): + process = SimpleNamespace(returncode=0, stdout=json.dumps(response), stderr='') + with self.subTest(response=response), patch('regression_suite.subprocess.run', return_value=process): + with self.assertRaisesRegex(RuntimeError, 'failure response'): + suite.call(['pipeline', 'get', 'owned']) + process = SimpleNamespace( + returncode=0, + stdout=json.dumps( + {'ok': True, 'new_field': 'allowed', 'data': {'pipeline': {'uuid': 'owned', 'extra': 1}}} + ), + stderr='', + ) + with patch('regression_suite.subprocess.run', return_value=process): + self.assertEqual(suite.call(['pipeline', 'get', 'owned'])['pipeline']['uuid'], 'owned') + + def test_profile_is_finite_and_uses_no_shared_resources(self): + profile, _ = suite_profile('cli', 'unit-test') + validate(profile) + self.assertEqual(profile['resources'], []) + self.assertEqual(len(profile['cases']), 3) + self.assertNotIn('execution', profile) + smoke, _ = suite_profile('smoke', 'unit-test') + self.assertFalse(any(case.get('scenario', '').startswith('cli.') for case in smoke['cases'])) + + +if __name__ == '__main__': + unittest.main() diff --git a/tools/test_regression_coverage.py b/tools/test_regression_coverage.py index f701cebce..86c7efa30 100644 --- a/tools/test_regression_coverage.py +++ b/tools/test_regression_coverage.py @@ -289,7 +289,7 @@ class Routes: name = binding['profile'] if name not in profiles: path = Path(__file__).with_name('regression_profiles') / f'{name}.json' - if name in ('api', 'state', 'isolation', 'plugin-lifecycle'): + if name in ('api', 'state', 'isolation', 'plugin-lifecycle', 'cli'): from regression_ci import suite_profile profile, _ = suite_profile(name, 'catalog-test') From 70fb56e27475a105a18d5d5ab0754314fad11ce4 Mon Sep 17 00:00:00 2001 From: huanghuoguoguo <1051233107@qq.com> Date: Fri, 30 Oct 2026 10:21:33 +0800 Subject: [PATCH 6/7] fix(agent): keep reset generation out of external conversation ID (#2818) Fixes #2814 After a session reset, keep the reset generation marker out of provider-facing external conversation state so Runner plugins start a new remote conversation. --- src/langbot/pkg/agent/runner/persistent_state_store.py | 5 +++-- tests/unit_tests/api/service/test_session_context_reset.py | 2 +- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/src/langbot/pkg/agent/runner/persistent_state_store.py b/src/langbot/pkg/agent/runner/persistent_state_store.py index 4665f127a..14c9a3403 100644 --- a/src/langbot/pkg/agent/runner/persistent_state_store.py +++ b/src/langbot/pkg/agent/runner/persistent_state_store.py @@ -188,7 +188,7 @@ class PersistentStateStore: except json.JSONDecodeError: pass # Skip invalid JSON - # Seed external.conversation_id from event.conversation_id if not set + # Seed the host conversation ID only before the conversation is reset. if self._check_scope_enabled('conversation', binding) and event.conversation_id: if 'external.conversation_id' not in snapshot['conversation']: from .context_reset import get_reset_generation @@ -196,7 +196,8 @@ class PersistentStateStore: generation = await get_reset_generation( self._db_engine, event.workspace_id, event.bot_id, event.conversation_id ) - snapshot['conversation']['external.conversation_id'] = generation or event.conversation_id + if generation is None: + snapshot['conversation']['external.conversation_id'] = event.conversation_id return snapshot diff --git a/tests/unit_tests/api/service/test_session_context_reset.py b/tests/unit_tests/api/service/test_session_context_reset.py index 70989432f..6a5892a11 100644 --- a/tests/unit_tests/api/service/test_session_context_reset.py +++ b/tests/unit_tests/api/service/test_session_context_reset.py @@ -153,7 +153,7 @@ async def test_reset_starts_new_external_runner_conversation(context_service): snapshot = await store.build_snapshot_from_event(event, binding, descriptor) generation = await get_reset_generation(engine, 'w', 'b', 'person_u') assert isinstance(generation, str) and generation and generation not in {'remote-old', 'person_u'} - assert snapshot['conversation']['external.conversation_id'] == generation + assert 'external.conversation_id' not in snapshot['conversation'] @pytest.mark.asyncio From 292fe43731728db58dc66fbbc9596dee2c16855d Mon Sep 17 00:00:00 2001 From: huanghuoguoguo <1051233107@qq.com> Date: Fri, 30 Oct 2026 11:11:31 +0800 Subject: [PATCH 7/7] test(plugins): use in-tree sources without hash pins Use the in-tree plugins directly for consumer contracts and avoid per-edit hash refreshes. --- plugins/MIGRATION.md | 10 ++++++---- plugins/scripts/test_repository.py | 5 +++++ tests/utils/legacy_plugin_contract.py | 14 ++++++++------ 3 files changed, 19 insertions(+), 10 deletions(-) diff --git a/plugins/MIGRATION.md b/plugins/MIGRATION.md index 62d15b09d..3988dc3b3 100644 --- a/plugins/MIGRATION.md +++ b/plugins/MIGRATION.md @@ -20,10 +20,12 @@ remain historical evidence, not certification of newly built packages. installation, configuration and Marketplace publication remain independent. - Each plugin retains its dependency declarations and test environment. Core's pytest configuration and pre-commit formatting do not apply to plugin suites. -- Core consumer tests use this checkout's `plugins/` and integrity-checked files - in `tests/fixtures/legacy_plugin_identity_sources.json`. Refresh the relevant - hashes deliberately when changing those consumer contracts. Published SDK - compatibility continues to be tested by the independent plugin workflows. +- Core consumer tests use this checkout's `plugins/` directly when the source lock + is marked `in-tree`; the listed files are required to exist, but their contents + are not pinned so ordinary plugin changes do not require fixture updates. + Hashes remain available for externally provisioned, revision-pinned sources. + Published SDK compatibility continues to be tested by the independent plugin + workflows. - Core wheels, sdists, Docker contexts, release ZIPs and fnOS staging exclude this tree and `.lbpkg` files. Git export archives exclude this tree as well; a normal Git checkout still contains all plugin sources. diff --git a/plugins/scripts/test_repository.py b/plugins/scripts/test_repository.py index 2f6fe01c9..2af5bb192 100644 --- a/plugins/scripts/test_repository.py +++ b/plugins/scripts/test_repository.py @@ -137,6 +137,11 @@ class RepositoryTests(unittest.TestCase): lock = json.loads((CORE / 'tests/fixtures/legacy_plugin_identity_sources.json').read_text()) self.assertEqual(lock['root'], 'plugins') self.assertEqual(lock['repository'], 'langbot-app/LangBot') + if lock.get('revision') == 'in-tree': + for key in ('files', 'local_agent_files'): + for relative in lock[key]: + self.assertTrue((ROOT / relative).is_file(), relative) + return for key in ('files', 'local_agent_files'): for relative, expected in lock[key].items(): self.assertEqual(hashlib.sha256((ROOT / relative).read_bytes()).hexdigest(), expected, relative) diff --git a/tests/utils/legacy_plugin_contract.py b/tests/utils/legacy_plugin_contract.py index e2bbf5a9c..3bbc84892 100644 --- a/tests/utils/legacy_plugin_contract.py +++ b/tests/utils/legacy_plugin_contract.py @@ -25,16 +25,18 @@ def verified_plugin_root(*, files_key: str = 'files') -> Path: pytest.fail('LANGBOT_LEGACY_PLUGIN_ROOT is set but empty', pytrace=False) root = Path(configured).resolve() lock = json.loads(SOURCE_LOCK.read_text()) + verify_hashes = lock.get('revision') != 'in-tree' for relative, expected in lock[files_key].items(): path = root / relative if not path.resolve().is_relative_to(root) or not path.is_file(): pytest.fail(f'Official plugin fixture is missing or outside its root: {relative}', pytrace=False) - actual = hashlib.sha256(path.read_bytes()).hexdigest() - if actual != expected: - pytest.fail( - f'Official plugin fixture hash mismatch: {relative}; expected {lock["repository"]}@{lock["revision"]}', - pytrace=False, - ) + if verify_hashes: + actual = hashlib.sha256(path.read_bytes()).hexdigest() + if actual != expected: + pytest.fail( + f'Official plugin fixture hash mismatch: {relative}; expected {lock["repository"]}@{lock["revision"]}', + pytrace=False, + ) return root