From a7bc48f601b07ae06b551be074b4b0bf4ae34279 Mon Sep 17 00:00:00 2001 From: huanghuoguoguo <1051233107@qq.com> Date: Wed, 28 Oct 2026 16:20:15 +0800 Subject: [PATCH] test(regression): exercise local execution faults and cleanup (#2795) * test(regression): add selectable resource lifecycle cases * test(regression): exercise local execution faults and cleanup --- tools/mock_llm.py | 9 +- tools/mock_llm_scenarios.json | 100 ++++++ tools/regression_contracts.json | 12 +- tools/regression_execution_cases.py | 427 +++++++++++++++++++++++ tools/regression_local_cases.py | 5 +- tools/test_regression_execution_cases.py | 309 ++++++++++++++++ 6 files changed, 854 insertions(+), 8 deletions(-) create mode 100644 tools/regression_execution_cases.py create mode 100644 tools/test_regression_execution_cases.py diff --git a/tools/mock_llm.py b/tools/mock_llm.py index 29eba3668..8bd037279 100644 --- a/tools/mock_llm.py +++ b/tools/mock_llm.py @@ -124,6 +124,9 @@ def select_message(body, scenario): raise ValueError('stream must be a boolean') if body.get('stream_options') is not None and not isinstance(body['stream_options'], dict): raise ValueError('stream_options must be an object') + scope = fault_scope(body, False) if scenario.get('substitute_mock_scope') else None + if scenario.get('substitute_mock_scope') and scope is None: + raise ValueError('this scenario requires mock_scope or LB_MOCK_SCOPE for tool file isolation') # Select from the current turn's history, never a shared request counter. start = next( (i for i in range(len(messages) - 1, -1, -1) if messages[i].get('role') == 'user'), @@ -172,7 +175,9 @@ def select_message(body, scenario): 'type': 'function', 'function': { 'name': call['name'], - 'arguments': json.dumps(call['arguments']), + 'arguments': json.dumps(call['arguments']).replace('{mock_scope}', scope) + if scope is not None + else json.dumps(call['arguments']), }, } for i, call in enumerate(tool_calls) @@ -471,6 +476,8 @@ def validate_scenarios(scenarios): raise ValueError(f'{name}: scenario must be an object') if scenario.get('kind', 'chat') not in ('chat', 'embedding'): raise ValueError(f'{name}: kind must be chat or embedding') + if 'substitute_mock_scope' in scenario and not isinstance(scenario['substitute_mock_scope'], bool): + raise ValueError(f'{name}: substitute_mock_scope must be a boolean') steps = scenario.get('steps', []) if not isinstance(steps, list) or (not steps and scenario.get('kind') != 'embedding'): raise ValueError(f'{name}: steps must be a non-empty array') diff --git a/tools/mock_llm_scenarios.json b/tools/mock_llm_scenarios.json index f870a43b6..0f9aaf468 100644 --- a/tools/mock_llm_scenarios.json +++ b/tools/mock_llm_scenarios.json @@ -295,5 +295,105 @@ "kind": "embedding", "status": 503, "recover_after": 1 + }, + "mock-local-interrupt": { + "steps": [ + { + "tool_calls": [ + { + "name": "exec", + "arguments": { + "command": "test ! -e /tmp/langbot-{mock_scope}-ready && test ! -e /tmp/langbot-{mock_scope}-leak && printf SETUP_READY > /tmp/langbot-{mock_scope}-ready && cat /tmp/langbot-{mock_scope}-ready" + } + } + ] + }, + { + "expect_tool_output": "SETUP_READY", + "tool_calls": [ + { + "name": "exec", + "arguments": { + "command": "sleep 8; printf LATE_SIDE_EFFECT > /tmp/langbot-{mock_scope}-leak", + "timeout_sec": 20 + } + } + ] + }, + { + "content": "UNEXPECTED_INTERRUPT_COMPLETION" + } + ], + "substitute_mock_scope": true + }, + "mock-local-interrupt-check": { + "steps": [ + { + "tool_calls": [ + { + "name": "exec", + "arguments": { + "command": "sleep 9; if test ! -f /tmp/langbot-{mock_scope}-ready; then printf INTERRUPT_SETUP_MISSING; exit 41; fi; if test -e /tmp/langbot-{mock_scope}-leak; then printf INTERRUPT_LATE_SIDE_EFFECT; exit 42; fi; rm /tmp/langbot-{mock_scope}-ready && printf INTERRUPT_CLEANUP_OK", + "timeout_sec": 15 + } + } + ] + }, + { + "expect_tool_output": "INTERRUPT_CLEANUP_OK", + "content": "INTERRUPT_CLEANUP_OK" + } + ], + "substitute_mock_scope": true + }, + "mock-local-retry-side-effect": { + "status": 503, + "recover_after": 1, + "steps": [ + { + "tool_calls": [ + { + "name": "exec", + "arguments": { + "command": "printf 'once\\n' >> /tmp/langbot-{mock_scope}-effects; printf SIDE_EFFECT_WRITTEN" + } + } + ] + }, + { + "expect_tool_output": "SIDE_EFFECT_WRITTEN", + "tool_calls": [ + { + "name": "exec", + "arguments": { + "command": "test \"$(wc -l < /tmp/langbot-{mock_scope}-effects)\" -eq 1 && rm /tmp/langbot-{mock_scope}-effects && printf SIDE_EFFECT_ONCE" + } + } + ] + }, + { + "expect_tool_output": "SIDE_EFFECT_ONCE", + "content": "MODEL_RETRY_SIDE_EFFECT_ONCE" + } + ], + "substitute_mock_scope": true + }, + "mock-local-path-escape": { + "steps": [ + { + "tool_calls": [ + { + "name": "read", + "arguments": { + "path": "/workspace/../../etc/passwd" + } + } + ] + }, + { + "expect_tool_output": "Path escapes the workspace boundary", + "content": "PATH_ESCAPE_REJECTED" + } + ] } } diff --git a/tools/regression_contracts.json b/tools/regression_contracts.json index 8d0729e30..de7495d19 100644 --- a/tools/regression_contracts.json +++ b/tools/regression_contracts.json @@ -188,17 +188,17 @@ {"id": "execution.cancel_cleanup", "area": "execution", "kind": "blackbox", "priority": "P0", "description": "Cancelling a running execution reaches cancelled state and releases its work", - "bindings": [] + "bindings": [{"profile": "scenarios", "case": "execution.cancel_cleanup", "checks": ["run_status", "exec_active", "exec_stopped", "no_late_side_effect"]}] }, {"id": "execution.timeout_cleanup", "area": "execution", "kind": "blackbox", "priority": "P0", "description": "Deadline expiry stops child work and allows a subsequent request", - "bindings": [] + "bindings": [{"profile": "scenarios", "case": "execution.timeout_cleanup", "checks": ["run_status", "exec_active", "exec_stopped", "no_late_side_effect"]}] }, {"id": "execution.retry_side_effects", "area": "execution", "kind": "blackbox", "priority": "P0", "description": "Retries do not duplicate externally visible side effects", - "bindings": [] + "bindings": [{"profile": "scenarios", "case": "execution.retry_side_effects", "checks": ["run_status", "fault_injected", "side_effect_once"]}] }, {"id": "runtime.restart_recovery", "area": "runtime", "kind": "blackbox", "priority": "P0", @@ -213,17 +213,17 @@ {"id": "model.stream_disconnect", "area": "model", "kind": "blackbox", "priority": "P1", "description": "A partial model stream disconnect is handled without a false successful completion", - "bindings": [] + "bindings": [{"profile": "scenarios", "case": "model.stream_disconnect", "checks": ["run_status", "fault_injected", "no_fallback_after_partial", "recovered"]}] }, {"id": "model.exhausted_fallback", "area": "model", "kind": "blackbox", "priority": "P1", "description": "Failure of every configured model reaches a visible failed outcome", - "bindings": [] + "bindings": [{"profile": "scenarios", "case": "model.exhausted_fallback", "checks": ["run_status", "fallback_exhausted", "recovered"]}] }, {"id": "tools.path_escape", "area": "tools", "kind": "blackbox", "priority": "P0", "description": "Box operations reject access outside their authorized filesystem scope", - "bindings": [] + "bindings": [{"profile": "scenarios", "case": "tools.path_escape", "checks": ["run_status", "path_rejected", "no_file_content"]}] }, {"id": "tools.timeout_recovery", "area": "tools", "kind": "blackbox", "priority": "P1", diff --git a/tools/regression_execution_cases.py b/tools/regression_execution_cases.py new file mode 100644 index 000000000..afe666d0b --- /dev/null +++ b/tools/regression_execution_cases.py @@ -0,0 +1,427 @@ +"""Local Agent faults with public API, mock-wire, and independent Run evidence.""" + +import copy +import hashlib +import json +from pathlib import Path +import time +import sys +import urllib.request +from uuid import uuid4 + +from regression_coverage import checked +from regression_local_cases import api, endpoint, headers, record_run, require + +GROUP = 'faults' +RUNNER = 'plugin:langbot-team/LocalAgent/default' +CASE_NAMES = { + 'execution.cancel_cleanup', + 'execution.timeout_cleanup', + 'execution.retry_side_effects', + 'model.stream_disconnect', + 'model.exhausted_fallback', + 'tools.path_escape', +} +MODELS = { + 'interrupt': 'mock-local-interrupt', + 'interrupt_check': 'mock-local-interrupt-check', + 'retry': 'mock-local-retry-side-effect', + 'disconnect': 'mock-stream-disconnect', + 'escape': 'mock-local-path-escape', +} + + +def add_cases(profile): + template = next(item for item in profile['resources'] if item['id'] == 'chat_model') + for name, model in MODELS.items(): + item = copy.deepcopy(template) + item['id'] = 'fault_' + name + '_model' + item['body']['name'] = model + item['body']['extra_args'] = {'request_policy': {'retries': 1 if name == 'retry' else 0}} + profile['resources'].append(item) + for name in sorted(CASE_NAMES): + profile['cases'].append( + {'name': name, 'scenario': name, 'stage': name.split('.')[0], 'expect': 'FAULT_CASE_OK'} + ) + + +def _data(suite, method, path, body=None): + status, payload = api(suite, method, path, body) + require( + status == 200 and isinstance(payload, dict) and payload.get('code') == 0, + f'{method} {path} returned HTTP {status}: {payload}', + ) + return payload.get('data') + + +def _poll(suite, read, message, seconds=20): + deadline = time.monotonic() + suite.timeout(seconds) + while True: + value = read() + if value: + return value + if time.monotonic() >= deadline: + raise TimeoutError(message) + time.sleep(0.1) + + +def _agent_config(suite, scope, model, *, box=False, timeout=30, fallbacks=()): + return { + 'runner': {'id': RUNNER}, + 'allowed_tools': ['exec', 'read'] if box else [], + 'runner_config': { + RUNNER: { + 'model': {'primary': suite.variables[model], 'fallbacks': [suite.variables[key] for key in fallbacks]}, + 'prompt': [{'role': 'system', 'content': 'Execute the deterministic QA scenario.'}], + 'knowledge-bases': [], + 'box-enabled': box, + 'box-session-id-template': scope, + 'timeout': timeout, + 'max-tool-iterations': 5, + 'tool-execution-mode': 'serial', + } + }, + } + + +def _sessions(suite): + result = _data(suite, 'GET', '/api/v1/box/sessions') + require(isinstance(result, list), 'Box sessions must be an array') + return {item['session_id']: item for item in result} + + +def _running_box(suite, session_id): + def read(): + return session_id if _sessions(suite).get(session_id, {}).get('status') == 'running' else None + + return _poll(suite, read, 'The long exec never became active in its Box') + + +def _stream(suite, agent, scope, row, *, interrupt=None, box_session_id=None, frame_key='frames'): + request = urllib.request.Request( + endpoint(suite, f'/api/v1/agents/{agent}/debug/stream'), + data=json.dumps({'text': f'Run the QA scenario LB_MOCK_SCOPE={scope}', 'conversation_id': scope}).encode(), + headers={**headers(suite), 'Content-Type': 'application/json'}, + ) + opener = urllib.request.build_opener(urllib.request.ProxyHandler({})) + frames = [] + row[frame_key] = frames + deadline = time.monotonic() + suite.timeout(45) + with opener.open(request, timeout=suite.timeout(45)) as response: + require(response.status == 200, 'Agent debug stream was not accepted') + while time.monotonic() < deadline: + line = response.readline(256 * 1024 + 1) + require(len(line) <= 256 * 1024, 'Agent debug frame exceeded its limit') + if not line: + break + if not line.strip(): + continue + text = line.decode('utf-8') + for secret in suite.secrets: + text = text.replace(secret, '[REDACTED]') + frame = json.loads(text) + require(isinstance(frame, dict), 'Agent debug frame must be an object') + frames.append(frame) + if interrupt and frame.get('kind') == 'result': + event = frame.get('data', {}) + if event.get('type') == 'tool.call.started' and event.get('data', {}).get('parameters', {}).get( + 'command', '' + ).startswith('sleep 8;'): + row['box_session_id'] = checked(row, 'exec_active', lambda: _running_box(suite, box_session_id)) + if interrupt == 'cancel': + break # Closing this public debug stream cancels its run. + if frame.get('kind') in {'completed', 'error'}: + return frames + else: + raise TimeoutError('Agent debug stream exceeded the observation deadline; not retried') + if interrupt == 'cancel': + require('box_session_id' in row, 'Cancellation did not reach the active long exec') + else: + raise RuntimeError('Agent debug stream ended without a terminal frame; outcome unknown') + return frames + + +def _terminal(suite, agent, conversation, row, expected, excluded=()): + def read(): + runs = _data(suite, 'GET', f'/api/v1/agents/{agent}/runs')['items'] + matches = [run for run in runs if run['conversation_id'] == conversation and run['run_id'] not in excluded] + require(len(matches) <= 1, 'One scenario submission created multiple AgentRuns') + return matches[0] if matches and matches[0].get('finished_at') else None + + run = _poll(suite, read, 'AgentRun did not terminate after the scenario') + try: + checked( + row, + 'run_status', + lambda: require(run['status'] == expected, f'Public Run status is {run["status"]}; expected {expected}'), + ) + finally: + checked(row, 'ledger_status', lambda: record_run(suite, row, run['run_id'], expected)) + detail = _data(suite, 'GET', f'/api/v1/agents/{agent}/runs/{run["run_id"]}/events') + require(not detail.get('has_more'), 'Scenario trace exceeds the bounded event page') + return run, detail['items'] + + +def _mock_rows(suite, scope, count): + path = Path(suite.path(suite.profile['smoke']['mock_records'])) + + def read(): + rows = [] + for line in path.read_text().splitlines(): + try: + item = json.loads(line) + except ValueError: + continue # A concurrent handler may still be appending its last line. + if item.get('scope') == scope: + rows.append(item) + return rows if len(rows) >= count else None + + return _poll(suite, read, 'The expected model request was not recorded', seconds=5) + + +def _tool_events(events, name): + return [ + item['data'] + for item in events + if item['type'] == 'tool.call.completed' and item['data'].get('tool_name') == name + ] + + +def _cleanup(suite, agent, scope, row, expected): + original = sys.exception() + errors = [] + try: + runs = _data(suite, 'GET', f'/api/v1/agents/{agent}/runs')['items'] + registered = {item['run_id'] for item in row.get('scenario_runs', [])} + pending = [run for run in runs if run['run_id'] not in registered] + if not registered and len(pending) == 1: + record_run(suite, row, pending[0]['run_id'], expected) + elif pending: + raise ValueError('Scenario has unexpected or unregistered AgentRuns at cleanup') + except (ValueError, OSError, RuntimeError) as exc: + errors.append(f'Run closeout: {exc}') + try: + _data(suite, 'DELETE', f'/api/v1/agents/{agent}') + status, _ = api(suite, 'GET', f'/api/v1/agents/{agent}') + require(status == 404, 'Deleted scenario Agent remains visible') + except (ValueError, OSError, RuntimeError) as exc: + errors.append(f'Agent cleanup: {exc}') + if row['case'] == 'execution.retry_side_effects': + try: + url = suite.variables['mock_base_url'].removesuffix('/v1').rstrip('/') + '/_mock/scopes/' + scope + request = urllib.request.Request(url, method='DELETE') + opener = urllib.request.build_opener(urllib.request.ProxyHandler({})) + with opener.open(request, timeout=suite.timeout(10)) as response: + payload = json.load(response) + require( + response.status == 200 and payload.get('scope') == scope and payload.get('released') in (0, 1), + 'Mock recovery scope cleanup failed', + ) + except (ValueError, OSError, RuntimeError) as exc: + errors.append(f'Mock scope cleanup: {exc}') + row['cleanup_errors'] = errors + if errors: + raise RuntimeError('; '.join(([str(original)] if original else []) + errors)) + + +def run_case(suite, case, row): + scenario = case['scenario'] + require(scenario in CASE_NAMES, 'Unknown local execution scenario') + scope = 'qa-fault-' + uuid4().hex + box_key = 'qa-fault-' + hashlib.sha256((str(suite.output.resolve()) + scenario).encode()).hexdigest()[:24] + # Core's RunnerBoxService maps the requested reuse key to this logical ID. + box_session_id = 'runner-' + hashlib.sha256(box_key.encode()).hexdigest() + interrupt = scenario in {'execution.cancel_cleanup', 'execution.timeout_cleanup'} + model = { + 'execution.cancel_cleanup': 'fault_interrupt_model', + 'execution.timeout_cleanup': 'fault_interrupt_model', + 'execution.retry_side_effects': 'fault_retry_model', + 'model.stream_disconnect': 'fault_disconnect_model', + 'model.exhausted_fallback': 'error503_model', + 'tools.path_escape': 'fault_escape_model', + }[scenario] + fallbacks = ( + ('error429_model',) + if scenario == 'model.exhausted_fallback' + else ('chat_model',) + if scenario == 'model.stream_disconnect' + else () + ) + config = _agent_config( + suite, + box_key, + model, + box=interrupt or scenario in {'execution.retry_side_effects', 'tools.path_escape'}, + timeout=5 if scenario == 'execution.timeout_cleanup' else 30, + fallbacks=fallbacks, + ) + expected = ( + 'cancelled' + if scenario == 'execution.cancel_cleanup' + else 'timeout' + if interrupt + else 'failed' + if scenario.startswith('model.') + else 'completed' + ) + created = _data(suite, 'POST', '/api/v1/agents', {'name': scope, 'config': config}) + agent = created['uuid'] + try: + row['agent_uuid'], row['conversation_id'] = agent, scope + row['box_reuse_key'] = box_key + frames = _stream( + suite, + agent, + scope, + row, + interrupt='cancel' if scenario == 'execution.cancel_cleanup' else 'timeout' if interrupt else None, + box_session_id=box_session_id, + ) + row['frames'] = frames + run, events = _terminal(suite, agent, scope, row, expected) + row['run_events'] = events + request_count = ( + 4 if scenario == 'execution.retry_side_effects' else 1 if scenario == 'model.stream_disconnect' else 2 + ) + requests = _mock_rows(suite, scope, request_count) + row['model_requests'] = [ + {'model': item['request']['model'], 'status': item['status'], 'faults': item.get('faults', [])} + for item in requests + ] + if interrupt: + checked( + row, + 'exec_active', + lambda: require('box_session_id' in row, 'No active exec was observed before interruption'), + ) + checked( + row, + 'exec_stopped', + lambda: _poll( + suite, + lambda: _sessions(suite).get(row['box_session_id'], {}).get('status') == 'idle', + 'Interrupted Box exec remains active', + ), + ) + config = _agent_config(suite, box_key, 'fault_interrupt_check_model', box=True) + _data(suite, 'PUT', f'/api/v1/agents/{agent}', {'config': config}) + recovery = _stream(suite, agent, scope, row, frame_key='recovery_frames') + calls = _tool_events([frame['data'] for frame in recovery if frame.get('kind') == 'result'], 'exec') + row['recovery_exec_results'] = calls + try: + recovery_run, recovery_events = _terminal( + suite, agent, scope, row, 'completed', excluded={run['run_id']} + ) + row['recovery_run_id'], row['recovery_events'] = recovery_run['run_id'], recovery_events + finally: + # The tool's evidence remains authoritative if its failed check also makes the model fail. + checked( + row, + 'no_late_side_effect', + lambda: require( + len(calls) == 1 + and (calls[0].get('result') or {}).get('exit_code') == 0 + and calls[0]['result'].get('stdout') == 'INTERRUPT_CLEANUP_OK', + 'Interrupted shell cleanup failed: ' + json.dumps(calls, ensure_ascii=False), + ), + ) + elif scenario == 'execution.retry_side_effects': + checked( + row, + 'fault_injected', + lambda: require( + [item['status'] for item in requests] == [503, 200, 200, 200] + and all(item['request']['model'] == 'mock-local-retry-side-effect' for item in requests), + 'Expected one 503 retry followed by tool execution and completion', + ), + ) + calls = _tool_events(events, 'exec') + checked( + row, + 'side_effect_once', + lambda: require( + len(calls) == 2 + and all(call.get('result', {}).get('exit_code') == 0 for call in calls) + and [call['result'].get('stdout') for call in calls] == ['SIDE_EFFECT_WRITTEN', 'SIDE_EFFECT_ONCE'], + 'Model retry duplicated or lost the tool side effect', + ), + ) + elif scenario == 'tools.path_escape': + calls = _tool_events(events, 'read') + checked( + row, + 'path_rejected', + lambda: require( + len(calls) == 1 and 'Path escapes the workspace boundary' in str(calls[0].get('error')), + 'The real Host tool did not reject the path escape', + ), + ) + checked( + row, + 'no_file_content', + lambda: require('root:' not in json.dumps(frames), 'Rejected file content appeared in Agent output'), + ) + else: + if scenario == 'model.stream_disconnect': + checked( + row, + 'fault_injected', + lambda: require( + len(requests) == 1 + and requests[0].get('injected_disconnect') is True + and not requests[0]['response_complete'], + 'The stream disconnect was not observed', + ), + ) + checked( + row, + 'no_fallback_after_partial', + lambda: require( + requests[0]['request']['model'] == 'mock-stream-disconnect' + and any( + item.get('data', {}).get('type') == 'message.delta' + and item['data'].get('data', {}).get('chunk', {}).get('content') + for item in frames + ), + 'The model did not commit partial output before failing', + ), + ) + else: + checked( + row, + 'fallback_exhausted', + lambda: require( + [(item['request']['model'], item['status']) for item in requests] + == [('mock-503', 503), ('mock-429', 429)], + 'The configured fallback sequence was not exhausted', + ), + ) + config = _agent_config(suite, scope, 'chat_model') + _data(suite, 'PUT', f'/api/v1/agents/{agent}', {'config': config}) + recovery = _stream(suite, agent, scope, row, frame_key='recovery_frames') + _terminal(suite, agent, scope, row, 'completed', excluded={run['run_id']}) + checked( + row, + 'recovered', + lambda: require( + any( + item.get('kind') == 'completed' and item.get('data', {}).get('final_text') == 'LANGBOT_CHAT_OK' + for item in recovery + ), + 'A healthy model did not recover after the failed run', + ), + ) + if config['runner_config'][RUNNER]['box-enabled']: + checked( + row, + 'box_idle', + lambda: _poll( + suite, + lambda: _sessions(suite).get(box_session_id, {}).get('status') == 'idle', + 'The scenario Box did not return to idle', + ), + ) + row['box_session_id'] = box_session_id + return {'result': 'FAULT_CASE_OK', 'scenario': scenario, 'run_id': run['run_id']} + finally: + _cleanup(suite, agent, scope, row, expected) diff --git a/tools/regression_local_cases.py b/tools/regression_local_cases.py index 951360e50..0436d68f2 100644 --- a/tools/regression_local_cases.py +++ b/tools/regression_local_cases.py @@ -9,7 +9,10 @@ import time import urllib.error import urllib.request -GROUPS = {'resources': 'regression_resource_cases'} +GROUPS = { + 'faults': 'regression_execution_cases', + 'resources': 'regression_resource_cases', +} class ScenarioUnavailable(RuntimeError): diff --git a/tools/test_regression_execution_cases.py b/tools/test_regression_execution_cases.py new file mode 100644 index 000000000..c42dda48c --- /dev/null +++ b/tools/test_regression_execution_cases.py @@ -0,0 +1,309 @@ +"""Challenge fault oracles and their cleanup; exercise NDJSON over real HTTP.""" + +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +import hashlib +import json +from pathlib import Path +import sqlite3 +import subprocess +import tempfile +import threading +from types import SimpleNamespace +import unittest +from unittest.mock import patch +from urllib.error import HTTPError +from urllib.request import Request, urlopen + +import regression_execution_cases as cases +from mock_llm import MockServer + + +class ExecutionCasesTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.root = Path(self.tmp.name) + self.ledger = self.root / 'ledger.db' + with sqlite3.connect(self.ledger) as db: + db.execute( + 'CREATE TABLE agent_run(id INTEGER, run_id TEXT, conversation_id TEXT, status TEXT, finished_at TEXT, workspace_id TEXT)' + ) + self.suite = SimpleNamespace( + output=self.root, + variables={ + **{f'fault_{name}_model': name for name in cases.MODELS}, + 'chat_model': 'chat', + 'error503_model': '503', + 'error429_model': '429', + 'workspace_uuid': 'workspace', + 'mock_base_url': 'http://unused.invalid/v1', + }, + profile={'smoke': {'ledger_db': str(self.ledger), 'mock_records': str(self.root / 'mock.jsonl')}}, + path=lambda value: value, + timeout=lambda maximum: maximum, + secrets=[], + connection={'endpoint': 'http://unused.invalid', 'credential_source': 'env:TEST_KEY'}, + env={'TEST_KEY': 'test-only'}, + ) + + def run_fixture( + self, scenario, *, tool_ok=True, fault=True, primary_status=None, recovery_status='completed', delete_ok=True + ): + expected = { + 'execution.cancel_cleanup': 'cancelled', + 'execution.timeout_cleanup': 'timeout', + 'execution.retry_side_effects': 'completed', + 'tools.path_escape': 'completed', + 'model.stream_disconnect': 'failed', + 'model.exhausted_fallback': 'failed', + }[scenario] + row = {'case': scenario, 'after_id': 0} + self.row = row + self.deleted = False + runs, events = [], {} + statuses = { + 'execution.retry_side_effects': [503, 200, 200, 200], + 'model.stream_disconnect': [200], + 'model.exhausted_fallback': [503, 429], + }.get(scenario, [200, 200]) + models = { + 'execution.retry_side_effects': ['mock-local-retry-side-effect'] * 4, + 'model.stream_disconnect': ['mock-stream-disconnect'], + 'model.exhausted_fallback': ['mock-503', 'mock-429'], + }.get(scenario, ['fixture'] * 2) + requests = [ + { + 'scope': 'qa-fault-fixture', + 'status': status if fault else 200, + 'request': {'model': model}, + 'injected_disconnect': fault, + 'response_complete': not fault, + } + for status, model in zip(statuses, models) + ] + (self.root / 'mock.jsonl').write_text(''.join(json.dumps(item) + '\n' for item in requests)) + box_key = 'qa-fault-' + hashlib.sha256((str(self.root.resolve()) + scenario).encode()).hexdigest()[:24] + box_session_id = 'runner-' + hashlib.sha256(box_key.encode()).hexdigest() + + def api(_suite, method, path, body=None): + if method == 'DELETE': + self.deleted = True + return (200, {'code': 0}) if delete_ok else (500, {'code': 'delete_failed'}) + if method == 'GET' and path == '/api/v1/agents/agent': + return 404, {'code': -1} + if path == '/api/v1/box/sessions': + return 200, {'code': 0, 'data': [{'session_id': box_session_id, 'status': 'idle'}]} + if path.endswith('/events'): + run_id = path.split('/')[-2] + return 200, {'code': 0, 'data': {'items': events[run_id], 'has_more': False}} + if path.endswith('/runs'): + return 200, {'code': 0, 'data': {'items': list(runs)}} + return 200, {'code': 0, 'data': {'uuid': 'agent'}} + + def stream(_suite, _agent, scope, _row, **kwargs): + index = len(runs) + 1 + run_id = f'run-{index}' + status = (primary_status or expected) if index == 1 else recovery_status + run = {'run_id': run_id, 'conversation_id': scope, 'status': status, 'finished_at': 'now'} + runs.append(run) + with sqlite3.connect(self.ledger) as db: + db.execute( + 'INSERT INTO agent_run VALUES(?,?,?,?,?,?)', (index, run_id, scope, status, 'now', 'workspace') + ) + tool = 'read' if scenario == 'tools.path_escape' else 'exec' + stdout = 'INTERRUPT_CLEANUP_OK' if index > 1 else 'SIDE_EFFECT_ONCE' + result = {'exit_code': 0 if tool_ok else 1, 'stdout': stdout if tool_ok else 'DUPLICATE_OR_LEAK'} + if not tool_ok and index > 1: + result = {'exit_code': 42, 'stdout': 'INTERRUPT_LATE_SIDE_EFFECT', 'stderr': ''} + events[run_id] = [ + { + 'type': 'tool.call.completed', + 'data': { + 'tool_name': tool, + 'result': result, + 'error': 'Path escapes the workspace boundary.' if tool == 'read' else None, + }, + } + ] + if scenario == 'execution.retry_side_effects': + events[run_id].insert( + 0, + { + 'type': 'tool.call.completed', + 'data': { + 'tool_name': 'exec', + 'result': {'exit_code': 0, 'stdout': 'SIDE_EFFECT_WRITTEN'}, + }, + }, + ) + if kwargs.get('interrupt'): + row['box_session_id'] = box_session_id + return [ + *[{'kind': 'result', 'data': event} for event in events[run_id]], + {'kind': 'result', 'data': {'type': 'message.delta', 'data': {'chunk': {'content': 'LANGBOT_'}}}}, + {'kind': 'completed', 'data': {'final_text': 'LANGBOT_CHAT_OK'}}, + ] + + class Released: + status = 200 + + def __enter__(self): + from io import BytesIO + + self.read = BytesIO(b'{"scope":"qa-fault-fixture","released":1}').read + return self + + def __exit__(self, *_): + pass + + with ( + patch.object(cases, 'api', side_effect=api), + patch.object(cases, '_stream', side_effect=stream), + patch.object(cases, 'uuid4', return_value=SimpleNamespace(hex='fixture')), + patch.object(cases.urllib.request.OpenerDirector, 'open', return_value=Released()), + ): + return cases.run_case(self.suite, {'scenario': scenario}, row) + + def test_fault_cases_preserve_expected_terminal_runs_and_delete_agent(self): + for scenario in sorted(cases.CASE_NAMES): + with self.subTest(scenario=scenario): + with sqlite3.connect(self.ledger) as db: + db.execute('DELETE FROM agent_run') + result = self.run_fixture(scenario) + self.assertEqual(result['result'], 'FAULT_CASE_OK') + self.assertTrue(self.deleted) + self.assertTrue(self.row['scenario_runs']) + self.assertEqual(self.row['assertions']['run_status'], 'passed') + self.assertEqual(self.row['cleanup_errors'], []) + + def test_duplicate_side_effect_and_untriggered_fault_fail_without_replay(self): + for tool_ok, fault, assertion in ((False, True, 'side_effect_once'), (True, False, 'fault_injected')): + with self.subTest(assertion=assertion): + with sqlite3.connect(self.ledger) as db: + db.execute('DELETE FROM agent_run') + with self.assertRaises(ValueError): + self.run_fixture('execution.retry_side_effects', tool_ok=tool_ok, fault=fault) + self.assertEqual(self.row['assertions'][assertion], 'failed') + self.assertTrue(self.deleted) + self.assertEqual(len(self.row['scenario_runs']), 1) + + def test_late_side_effect_is_a_failure_even_when_run_is_cancelled(self): + with self.assertRaisesRegex(ValueError, 'INTERRUPT_LATE_SIDE_EFFECT'): + self.run_fixture('execution.cancel_cleanup', tool_ok=False, recovery_status='failed') + self.assertEqual(self.row['assertions']['no_late_side_effect'], 'failed') + self.assertEqual([run['status'] for run in self.row['scenario_runs']], ['cancelled', 'failed']) + self.assertEqual(self.row['recovery_exec_results'][0]['result']['exit_code'], 42) + self.assertTrue(self.deleted) + + def test_recovery_command_distinguishes_missing_setup_and_late_side_effect(self): + scenarios = json.loads(Path(cases.__file__).with_name('mock_llm_scenarios.json').read_text()) + command = scenarios['mock-local-interrupt-check']['steps'][0]['tool_calls'][0]['arguments']['command'] + command = command.removeprefix('sleep 9; ').replace('/tmp/langbot-{mock_scope}', str(self.root / 'fixture')) + ready, leaked = self.root / 'fixture-ready', self.root / 'fixture-leak' + for setup, leak, status, output in ( + (False, False, 41, 'INTERRUPT_SETUP_MISSING'), + (True, True, 42, 'INTERRUPT_LATE_SIDE_EFFECT'), + (True, False, 0, 'INTERRUPT_CLEANUP_OK'), + ): + with self.subTest(setup=setup, leak=leak): + if setup: + ready.write_text('SETUP_READY') + if leak: + leaked.write_text('LATE_SIDE_EFFECT') + else: + leaked.unlink(missing_ok=True) + result = subprocess.run(['sh', '-c', command], capture_output=True, text=True, timeout=2) + self.assertEqual((result.returncode, result.stdout), (status, output)) + self.assertEqual(ready.exists(), setup and status != 0) + + def test_unexpected_run_status_is_recorded_and_cleanup_failure_keeps_original(self): + with self.assertRaisesRegex(RuntimeError, 'Run ended as completed.*Agent cleanup'): + self.run_fixture('model.exhausted_fallback', primary_status='completed', delete_ok=False) + self.assertEqual(self.row['scenario_runs'][0]['status'], 'completed') + self.assertEqual(self.row['scenario_runs'][0]['expected_status'], 'failed') + self.assertTrue(self.deleted) + + def test_real_ndjson_transport_and_cancel_close(self): + requests = [] + frames = [ + { + 'kind': 'result', + 'data': {'type': 'tool.call.started', 'data': {'parameters': {'command': 'sleep 8; printf late'}}}, + }, + {'kind': 'completed', 'data': {'final_text': 'done'}}, + ] + + class Handler(BaseHTTPRequestHandler): + def do_POST(self): + requests.append(json.loads(self.rfile.read(int(self.headers['Content-Length'])))) + self.send_response(200) + self.send_header('Content-Type', 'application/x-ndjson') + self.end_headers() + for frame in frames: + self.wfile.write(json.dumps(frame).encode() + b'\n') + self.wfile.flush() + + def log_message(self, *_): + pass + + server = ThreadingHTTPServer(('127.0.0.1', 0), Handler) + thread = threading.Thread(target=server.serve_forever) + thread.start() + self.suite.connection['endpoint'] = f'http://127.0.0.1:{server.server_port}' + try: + self.assertEqual(cases._stream(self.suite, 'agent', 'scope', {})[-1]['kind'], 'completed') + row = {} + with patch.object(cases, '_running_box', return_value='box'): + cancelled = cases._stream(self.suite, 'agent', 'scope', row, interrupt='cancel') + self.assertEqual(len(cancelled), 1) + self.assertEqual(row['box_session_id'], 'box') + frames.pop() + with self.assertRaisesRegex(RuntimeError, 'outcome unknown'): + cases._stream(self.suite, 'agent', 'scope', {}) + self.assertEqual(len(requests), 3) + finally: + server.shutdown() + server.server_close() + thread.join() + + def test_real_mock_http_scopes_isolate_tool_paths_and_reject_unsafe_substitution(self): + scenarios = json.loads(Path(cases.__file__).with_name('mock_llm_scenarios.json').read_text()) + with MockServer(('127.0.0.1', 0), scenarios) as server: + thread = threading.Thread(target=server.serve_forever) + thread.start() + + def request(scope): + body = { + 'model': 'mock-local-interrupt', + 'mock_scope': scope, + 'messages': [{'role': 'user', 'content': 'test'}], + 'tools': [{'type': 'function', 'function': {'name': 'exec'}}], + } + with urlopen( + Request( + f'http://127.0.0.1:{server.server_port}/v1/chat/completions', data=json.dumps(body).encode() + ), + timeout=5, + ) as response: + return json.load(response) + + try: + commands = [] + for scope in ('run-a', 'run-b'): + call = request(scope)['choices'][0]['message']['tool_calls'][0] + command = json.loads(call['function']['arguments'])['command'] + self.assertIn(f'/tmp/langbot-{scope}-ready', command) + self.assertNotIn('{mock_scope}', command) + commands.append(command) + self.assertNotEqual(commands[0], commands[1]) + for invalid in (None, 'unsafe;echo', '../../escape'): + with self.subTest(scope=invalid), self.assertRaises(HTTPError) as caught: + request(invalid) + self.assertEqual(caught.exception.code, 400) + finally: + server.shutdown() + thread.join() + + +if __name__ == '__main__': + unittest.main()