From 34b9dbd6ea8389c61aa5cb793d2b609c27ec2d8b Mon Sep 17 00:00:00 2001 From: huanghuoguoguo <60681390+huanghuoguoguo@users.noreply.github.com> Date: Thu, 29 Oct 2026 11:42:48 +0800 Subject: [PATCH] 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()