Merge remote-tracking branch 'origin/master' into feat/cross-platform-ai-cards

This commit is contained in:
fdc310
2026-10-30 12:13:31 +08:00
28 changed files with 1860 additions and 51 deletions
+6 -4
View File
@@ -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.
+5
View File
@@ -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)
@@ -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
@@ -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
+8 -6
View File
@@ -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
+100
View File
@@ -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
@@ -498,6 +527,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
@@ -538,6 +590,54 @@ 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.
## 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
+59 -23
View File
@@ -21,7 +21,17 @@ 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,
'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
@@ -34,12 +44,29 @@ 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
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)
if name == 'plugin-lifecycle':
from regression_plugin_lifecycle import add_cases
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(
@@ -161,11 +188,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())
@@ -179,8 +210,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', 'plugin-lifecycle', 'cli'):
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)
@@ -309,21 +340,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')
+115
View File
@@ -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']}
+44 -1
View File
@@ -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/<author>/<plugin_name>", "PUT /api/v1/plugins/<author>/<plugin_name>/config", "POST /api/v1/plugins/<author>/<plugin_name>/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/<author>/<plugin_name>"]
},
{"id": "agent.direct_execution",
"area": "agent", "kind": "blackbox", "priority": "P1",
"description": "Standalone Agent entrypoint completes through the public API",
@@ -434,6 +456,27 @@
"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"]}]
},
{"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": "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"]}]}
]
}
+5 -2
View File
@@ -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)
@@ -0,0 +1 @@
<!doctype html><title>Lifecycle probe</title>
@@ -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(),
}
)
@@ -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
@@ -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')
+328
View File
@@ -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', []))}
+3
View File
@@ -15,6 +15,9 @@ GROUPS = {
'lifecycle': 'regression_lifecycle_cases',
'plugins': 'regression_plugin_cases',
'api': 'regression_api_cases',
'isolation': 'regression_isolation_cases',
'plugin-lifecycle': 'regression_plugin_lifecycle',
'cli': 'regression_cli_cases',
}
+285
View File
@@ -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', '<svg xmlns="http://www.w3.org/2000/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)
+278
View File
@@ -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
+5
View File
@@ -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']
+43 -8
View File
@@ -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(),
}
+88
View File
@@ -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()
+179
View File
@@ -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()
+2 -2
View File
@@ -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', 'isolation', 'plugin-lifecycle', 'cli'):
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
+10 -1
View File
@@ -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'):
+86
View File
@@ -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()
+1 -1
View File
@@ -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']]))
+118
View File
@@ -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()
+43
View File
@@ -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()