Merge selectable local regression tools into CI validation branch

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

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"
+      }
+    ]
   }
 }

tools/regression_contracts.json

@@ -173,7 +173,7 @@
     {"id": "auth.revocation",
      "area": "auth", "kind": "blackbox", "priority": "P0",
      "description": "Revoked credentials stop authorizing new requests",
-     "bindings": []
+     "bindings": [{"profile": "scenarios", "case": "auth_revocation", "checks": ["credential_active", "credential_revoked", "key_status", "key_cleanup"]}]
     },
     {"id": "workspace.read_isolation",
      "area": "workspace", "kind": "blackbox", "priority": "P0",
@@ -188,42 +188,42 @@
     {"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",
      "description": "A runtime restart produces a known outcome and subsequent execution works",
-     "bindings": []
+     "bindings": [{"profile": "scenarios", "case": "runtime_restart_recovery", "checks": ["process_replaced", "no_false_success", "interrupted_terminal", "response_after_restart"]}]
     },
     {"id": "storage.restart_history",
      "area": "storage", "kind": "blackbox", "priority": "P1",
      "description": "Conversation history remains usable across a Core restart",
-     "bindings": []
+     "bindings": [{"profile": "scenarios", "case": "core_restart_history", "checks": ["before_restart", "process_replaced", "history_after_restart"]}]
     },
     {"id": "model.stream_disconnect",
      "area": "model", "kind": "blackbox", "priority": "P1",
      "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",
@@ -248,7 +248,7 @@
     {"id": "agent.direct_execution",
      "area": "agent", "kind": "blackbox", "priority": "P1",
      "description": "Standalone Agent entrypoint completes through the public API",
-     "bindings": []
+     "bindings": [{"profile": "scenarios", "case": "agent_direct_execution", "checks": ["agent_response", "agent_terminal", "agent_events", "agent_cleanup"]}]
     },
     {"id": "persistence.postgres_parity",
      "area": "persistence", "kind": "blackbox", "priority": "P1",
@@ -265,6 +265,8 @@
      "description": "A real embedding model retrieves relevant fixed documents",
      "bindings": []
     },
+    {"id": "retrieval.delete_visibility", "area": "retrieval", "kind": "blackbox", "priority": "P1", "description": "Deleting one ingested document removes its retrieval results while preserving a second document", "bindings": [{"profile": "scenarios", "case": "retrieval_delete_visibility", "checks": ["retrieval_initial", "retrieval_deleted", "retrieval_control", "document_cleanup"]}]},
+    {"id": "delivery.callback_retry_identity", "area": "delivery", "kind": "blackbox", "priority": "P1", "description": "A 5xx callback retry preserves body and delivery identity, completes delivery, and permits a subsequent message", "bindings": [{"profile": "scenarios", "case": "callback_retry_identity", "checks": ["callback_retry", "callback_followup", "callback_cleanup"]}]},
     {"id": "execution.persisted_terminal",
      "area": "execution", "kind": "graybox", "priority": "P0",
      "description": "Each executed request has a unique completed persisted run and terminal event",

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)

tools/regression_lifecycle_cases.py

@@ -0,0 +1,497 @@
+"""Optional lifecycle checks using real asynchronous HTTP Bot callbacks."""
+
+from contextlib import closing
+import copy
+import hashlib
+import hmac
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+import json
+from pathlib import Path
+import secrets
+import sqlite3
+import threading
+import time
+
+from regression_coverage import checked
+from regression_local_cases import ScenarioUnavailable, api, record_run, require
+
+GROUP = 'lifecycle'
+CASE_NAMES = {'callback_retry_identity'}
+
+
+def add_cases(profile):
+    profile['cases'].append(
+        {
+            'name': 'callback_retry_identity',
+            'scenario': 'callback_retry_identity',
+            'stage': 'delivery',
+            'expect': 'LANGBOT_CALLBACK_RETRY_OK',
+        }
+    )
+
+
+def signature(secret, body, timestamp):
+    return 'sha256=' + hmac.new(secret.encode(), timestamp.encode() + b'.' + body, hashlib.sha256).hexdigest()
+
+
+def ledger(suite):
+    path = Path(suite.path(suite.profile['smoke']['ledger_db'])).resolve()
+    connection = sqlite3.connect(path.as_uri() + '?mode=ro', uri=True, timeout=min(5, suite.timeout(5)))
+    connection.row_factory = sqlite3.Row
+    return connection
+
+
+def after_id(suite):
+    with closing(ledger(suite)) as connection:
+        return connection.execute('SELECT COALESCE(MAX(id),0) FROM agent_run').fetchone()[0]
+
+
+def completed(suite, row, session_id, after, expected='completed'):
+    from regression_smoke import completed_run
+
+    with closing(ledger(suite)) as connection:
+        run = completed_run(
+            connection, suite.variables['workspace_uuid'], 'group_' + session_id, after, suite.timeout(20)
+        )
+    record_run(suite, row, run['run_id'], expected)
+    return run
+
+
+class CallbackReceiver:
+    """Reject the first callback once, then retain receipts for independent checks."""
+
+    def __init__(self, secret):
+        self.records = []
+        self.condition = threading.Condition()
+        receiver = self
+
+        class Handler(BaseHTTPRequestHandler):
+            def setup(self):
+                super().setup()
+                self.connection.settimeout(5)
+
+            def do_POST(self):
+                try:
+                    size = int(self.headers.get('Content-Length', '0'))
+                    require(0 < size <= 1024 * 1024, 'Invalid callback body size')
+                    body = self.rfile.read(size)
+                    payload = json.loads(body)
+                    timestamp = self.headers.get('X-LB-Timestamp', '')
+                    require(abs(time.time() - int(timestamp)) <= 300, 'Invalid callback timestamp')
+                    require(
+                        hmac.compare_digest(self.headers.get('X-LB-Signature', ''), signature(secret, body, timestamp)),
+                        'Invalid callback signature',
+                    )
+                    require(isinstance(payload, dict), 'Callback must be an object')
+                    record = {'body': body, 'payload': payload}
+                    with receiver.condition:
+                        status = 503 if not receiver.records else 200
+                except (ValueError, OSError) as exc:
+                    record, status = {'error': str(exc)}, 400
+                self.send_response(status)
+                self.send_header('Content-Type', 'application/json')
+                self.end_headers()
+                self.wfile.write(b'{}')
+                self.wfile.flush()
+                with receiver.condition:
+                    receiver.records.append({**record, 'status': status})
+                    receiver.condition.notify_all()
+
+            def log_message(self, *args):
+                pass
+
+        self.server = ThreadingHTTPServer(('127.0.0.1', 0), Handler)
+        self.thread = threading.Thread(target=self.server.serve_forever, kwargs={'poll_interval': 0.05})
+
+    def __enter__(self):
+        self.thread.start()
+        return self
+
+    def __exit__(self, *args):
+        self.server.shutdown()
+        self.server.server_close()
+        self.thread.join(timeout=5)
+        require(not self.thread.is_alive(), 'Callback receiver did not stop')
+
+    @property
+    def url(self):
+        return f'http://127.0.0.1:{self.server.server_port}/callback'
+
+    def wait_final(self, reply_to, timeout):
+        deadline = time.monotonic() + timeout
+        with self.condition:
+            while True:
+                errors = [item['error'] for item in self.records if 'error' in item]
+                require(not errors, f'Callback receiver rejected delivery: {errors}')
+                records = [item for item in self.records if item['payload'].get('reply_to') == reply_to]
+                if any(item['status'] == 200 and item['payload'].get('is_final') is True for item in records):
+                    return list(records)
+                remaining = deadline - time.monotonic()
+                if remaining <= 0:
+                    raise TimeoutError('No successful final callback before the observation deadline')
+                self.condition.wait(remaining)
+
+
+def verify_delivery(records, session_id, *, retried):
+    require(bool(records), 'No callback receipts')
+    statuses = [item['status'] for item in records]
+    if retried:
+        require(len(records) >= 2 and [item['status'] for item in records[:2]] == [503, 200], '5xx was not retried')
+        require(records[0]['body'] == records[1]['body'], 'Retry changed callback body or identity')
+        records = records[1:]
+    require(all(item['status'] == 200 for item in records), 'Unexpected callback failure')
+    payloads = [item['payload'] for item in records]
+    require(all(item.get('session_id') == session_id for item in payloads), 'Callback changed session identity')
+    require([item.get('sequence') for item in payloads] == list(range(1, len(payloads) + 1)), 'Callback order changed')
+    require(
+        [item.get('is_final') for item in payloads] == [False] * (len(payloads) - 1) + [True],
+        'Callback final marker is missing or duplicated',
+    )
+    parts = [part for item in payloads for part in item.get('message', [])]
+    require(all(isinstance(part, dict) for part in parts), 'Callback message has invalid components')
+    text = ''.join(str(part.get('text', '')) for part in parts if part.get('type') == 'Plain')
+    require('LANGBOT_CHAT_OK' in text, 'Callback did not deliver the expected business reply')
+    return {
+        'parts': len(payloads),
+        'finals': 1,
+        'response_statuses': statuses,
+        'body_sha256': [hashlib.sha256(item['body']).hexdigest() for item in records],
+    }
+
+
+def run_case(suite, case, row):
+    require(case['scenario'] in CASE_NAMES, 'Unknown lifecycle scenario')
+    secret = secrets.token_urlsafe(32)
+    suite.secrets.append(secret)
+    bot_uuid, primary_error = None, None
+    with CallbackReceiver(secret) as receiver:
+        try:
+            status, response = api(
+                suite,
+                'POST',
+                '/api/v1/platform/bots',
+                {
+                    'name': 'QA callback retry ' + secrets.token_hex(4),
+                    'adapter': 'http_bot',
+                    'enable': True,
+                    'adapter_config': {
+                        'signature_required': True,
+                        'inbound_secret': secret,
+                        'outbound_secret': secret,
+                        'callback_url': receiver.url,
+                        'callback_timeout': 2,
+                        'callback_max_retries': 1,
+                        'default_session_type': 'group',
+                    },
+                    'event_bindings': [
+                        {
+                            'event_pattern': 'message.received',
+                            'target_type': 'pipeline',
+                            'target_uuid': suite.variables['chat_pipeline'],
+                            'enabled': True,
+                            'filters': [],
+                        }
+                    ],
+                    'plugin_processors': [],
+                },
+            )
+            require(status == 200 and response.get('code') == 0, 'Callback Bot creation failed')
+            bot_uuid = response['data']['uuid']
+            messages = []
+            for index in range(2):
+                session_id = 'qa-callback-' + secrets.token_hex(8)
+                body = {
+                    'session_id': session_id,
+                    'session_type': 'group',
+                    'sender': {'id': 'qa-callback-sender'},
+                    'message': [{'type': 'Plain', 'text': 'ai Run the normal QA scenario'}],
+                }
+                stamp = str(int(time.time()))
+                status, accepted = api(
+                    suite,
+                    'POST',
+                    f'/bots/{bot_uuid}',
+                    body,
+                    headers={
+                        'X-LB-Timestamp': stamp,
+                        'X-LB-Signature': signature(secret, json.dumps(body).encode(), stamp),
+                        'X-LB-Idempotency-Key': secrets.token_hex(16),
+                    },
+                )
+                require(status == 202 and accepted.get('code') == 0, 'Asynchronous ingress was not accepted')
+                message_id = accepted['data']['accepted_message_id']
+                name = 'callback_retry' if index == 0 else 'callback_followup'
+                checked(
+                    row,
+                    name,
+                    lambda: verify_delivery(
+                        receiver.wait_final(message_id, suite.timeout(30)), session_id, retried=index == 0
+                    ),
+                )
+                messages.append((message_id, session_id, name))
+            with receiver.condition:
+                receipts = list(receiver.records)
+            deliveries = []
+            for index, (message_id, session_id, name) in enumerate(messages):
+                records = [item for item in receipts if item.get('payload', {}).get('reply_to') == message_id]
+                deliveries.append(checked(row, name, lambda: verify_delivery(records, session_id, retried=index == 0)))
+                completed(suite, row, session_id, row['after_id'])
+            return {'result': 'LANGBOT_CALLBACK_RETRY_OK', 'deliveries': deliveries}
+        except BaseException as exc:
+            primary_error = exc
+            raise
+        finally:
+            if bot_uuid is not None:
+
+                def cleanup():
+                    status, response = api(suite, 'DELETE', f'/api/v1/platform/bots/{bot_uuid}')
+                    require(status == 200 and response.get('code') == 0, 'Callback Bot deletion failed')
+                    status, _ = api(suite, 'GET', f'/api/v1/platform/bots/{bot_uuid}')
+                    require(status == 404, 'Deleted callback Bot remains visible')
+
+                try:
+                    checked(row, 'callback_cleanup', cleanup)
+                except Exception as exc:
+                    row.setdefault('cleanup_errors', []).append(f'{type(exc).__name__}: {exc}')
+                    if primary_error is None:
+                        raise
+
+
+def restart_profile(profile):
+    """Build the exact one-time restart fixture profile for execution and catalog checks."""
+    current = copy.deepcopy(profile)
+    for key in ('execution', 'resource_limits', 'scenario_groups'):
+        current.pop(key, None)
+    wanted = {'mock_provider', 'chat_model', 'chat_pipeline', 'chat_bot', 'slow_model', 'slow_pipeline'}
+    current['resources'] = [item for item in current['resources'] if item['id'] in wanted]
+    require(
+        {item['id'] for item in current['resources']} == wanted,
+        'Restart checks require the standard chat/slow fixtures',
+    )
+    slow_bot = copy.deepcopy(next(item for item in current['resources'] if item['id'] == 'chat_bot'))
+    slow_bot['id'] = 'restart_slow_bot'
+    slow_bot['body']['name'] = 'QA runtime restart slow'
+    slow_bot['body']['event_bindings'][0]['target_uuid'] = '${slow_pipeline}'
+    current['resources'].append(slow_bot)
+    current.update(coverage_suite='scenarios', cleanup='always', plugins=[], cases=[])
+    for name, bot, expected in (
+        ('core_restart_history', 'chat_bot', 'LANGBOT_CHAT_OK'),
+        ('runtime_restart_recovery', 'restart_slow_bot', 'LANGBOT_SLOW_STREAM_OK'),
+    ):
+        current['cases'].append(
+            {
+                'name': name,
+                'stage': 'restart',
+                'expect': expected,
+                'http_group': {
+                    'url': '${core_base_url}/bots/${' + bot + '}/sync',
+                    'secret_env': 'LANGBOT_QA_GROUP_SECRET',
+                    'message': 'ai Run the restart QA scenario',
+                },
+            }
+        )
+    return current
+
+
+def run_restart_checks(profile, profile_path, output, env, restart_owned):
+    """Run owned-service restart checks once, before fixed-PID resource sampling starts."""
+    from regression_smoke import UnresolvedRun, reconcile_scenario_runs
+    from regression_suite import Suite, resolve
+    from regression_traffic import group_request
+
+    started = time.monotonic()
+    deadline = started + 120
+
+    def timeout(maximum):
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            raise TimeoutError('Restart preflight exceeded its 120-second deadline')
+        return min(maximum, remaining)
+
+    current = restart_profile(profile)
+    output = Path(output)
+    output.mkdir(parents=True, exist_ok=False, mode=0o700)
+    suite = Suite(current, Path(profile_path), output, env)
+    suite.timeout = timeout
+    user_token = suite.env.get('LANGBOT_QA_USER_TOKEN')
+    if user_token:
+        suite.secrets.append(user_token)
+    suite.write('profile.json', current)
+    result = {'mode': 'restart_preflight', 'errors': [], 'incomplete': [], 'cases': [], 'phases': []}
+
+    def restart(row, name):
+        receipt = restart_owned(name, timeout=timeout(60))
+        require(
+            isinstance(receipt, dict)
+            and receipt.get('old_pid')
+            and receipt.get('new_pid')
+            and receipt['old_pid'] != receipt['new_pid'],
+            'Owned restart did not replace the service process',
+        )
+        row['restart'] = receipt
+        timeout(1)
+
+    try:
+        if not user_token:
+            raise ScenarioUnavailable('Restart terminal readback requires LANGBOT_QA_USER_TOKEN')
+        suite.prepare()
+        for original in current['cases']:
+            row = {'case': original['name'], 'stage': 'restart', 'status': 'failed', 'after_id': after_id(suite)}
+            result['cases'].append(row)
+            case = resolve(original, suite.variables)
+            session_id = 'qa-restart-' + secrets.token_hex(8)
+            token = secrets.token_hex(16)
+            worker = None
+            before = time.monotonic()
+            try:
+                if row['case'] == 'core_restart_history':
+                    checked(
+                        row,
+                        'before_restart',
+                        lambda: group_request(
+                            case, session_id, 'qa-member', 1, token, 'none', suite.env, timeout=timeout(20)
+                        ),
+                    )
+                    completed(suite, row, session_id, row['after_id'])
+                    checked(row, 'process_replaced', lambda: restart(row, 'core'))
+                    after = after_id(suite)
+                    checked(
+                        row,
+                        'history_after_restart',
+                        lambda: group_request(
+                            case,
+                            session_id,
+                            'qa-member',
+                            2,
+                            secrets.token_hex(16),
+                            token,
+                            suite.env,
+                            timeout=timeout(20),
+                        ),
+                    )
+                    completed(suite, row, session_id, after)
+                else:
+                    outcome = {}
+
+                    def request():
+                        try:
+                            outcome['response'] = group_request(
+                                case, session_id, 'qa-member', 1, token, 'none', suite.env, timeout=timeout(25)
+                            )
+                        except Exception as exc:
+                            outcome['error'] = type(exc).__name__ + ': ' + str(exc)
+
+                    worker = threading.Thread(target=request, daemon=True)
+                    worker.start()
+                    running_deadline = time.monotonic() + timeout(10)
+                    while True:
+                        with closing(ledger(suite)) as connection:
+                            runs = connection.execute(
+                                'SELECT run_id, status, finished_at FROM agent_run '
+                                'WHERE workspace_id=? AND conversation_id=? AND id>?',
+                                (suite.variables['workspace_uuid'], 'group_' + session_id, row['after_id']),
+                            ).fetchall()
+                        if len(runs) == 1 and runs[0]['status'] == 'running' and not runs[0]['finished_at']:
+                            break
+                        if outcome or time.monotonic() >= running_deadline:
+                            raise TimeoutError('No unique in-flight run was observed before Runtime restart')
+                        time.sleep(0.05)
+                    row['interrupted_run_id'] = runs[0]['run_id']
+                    checked(row, 'process_replaced', lambda: restart(row, 'runtime'))
+                    worker.join(timeout=timeout(25))
+                    if worker.is_alive():
+                        raise TimeoutError('Interrupted request did not finish after Runtime restart')
+                    row['interrupted_response'] = outcome
+                    checked(
+                        row,
+                        'no_false_success',
+                        lambda: require(
+                            'response' not in outcome, 'Interrupted request returned a successful business response'
+                        ),
+                    )
+                    from regression_smoke import completed_run
+
+                    with closing(ledger(suite)) as connection:
+                        terminal = completed_run(
+                            connection,
+                            suite.variables['workspace_uuid'],
+                            'group_' + session_id,
+                            row['after_id'],
+                            timeout(20),
+                        )
+
+                    def public_terminal():
+                        status, payload = api(
+                            suite,
+                            'GET',
+                            f'/api/v1/monitoring/executions/agent/{row["interrupted_run_id"]}?section=overview',
+                            headers={'Authorization': 'Bearer ' + user_token},
+                        )
+                        run = payload.get('data', {}).get('run', {})
+                        require(
+                            status == 200
+                            and payload.get('code') == 0
+                            and run.get('run_id') == row['interrupted_run_id'] == terminal['run_id']
+                            and run.get('status') == terminal['status']
+                            and run['status'] in ('failed', 'cancelled', 'timeout'),
+                            'Interrupted run did not expose the expected failure terminal state',
+                        )
+                        return run
+
+                    row['public_terminal'] = checked(row, 'interrupted_terminal', public_terminal)
+                    record_run(suite, row, terminal['run_id'], terminal['status'])
+                    recovery = resolve(current['cases'][0], suite.variables)
+                    recovery_session = 'qa-runtime-recovered-' + secrets.token_hex(8)
+                    after = after_id(suite)
+                    checked(
+                        row,
+                        'response_after_restart',
+                        lambda: group_request(
+                            recovery,
+                            recovery_session,
+                            'qa-member',
+                            1,
+                            secrets.token_hex(16),
+                            'none',
+                            suite.env,
+                            timeout=timeout(20),
+                        ),
+                    )
+                    completed(suite, row, recovery_session, after)
+                row['status'] = 'passed'
+            except Exception as exc:
+                row.update(
+                    status='unknown'
+                    if isinstance(exc, (TimeoutError, UnresolvedRun, ScenarioUnavailable))
+                    else 'failed',
+                    error=str(exc),
+                )
+                result['errors'].append(f'{row["case"]}: {row["error"]}')
+                if row['status'] == 'unknown':
+                    result['incomplete'].append(result['errors'][-1])
+            finally:
+                if worker is not None and worker.is_alive():
+                    worker.join(timeout=max(0, min(5, deadline - time.monotonic())))
+                    if worker.is_alive():
+                        error = 'Restart request worker did not stop before cleanup deadline'
+                        row.setdefault('cleanup_errors', []).append(error)
+                        result['errors'].append(f'{row["case"]}: cleanup failed: {error}')
+                row['seconds'] = time.monotonic() - before
+                suite.write('cases.json', result['cases'])
+            if row['status'] != 'passed':
+                break
+        reconcile_scenario_runs(suite, result, key='ledger_closeout')
+    except Exception as exc:
+        result['errors'].append('restart preparation: ' + str(exc))
+        if isinstance(exc, (TimeoutError, ScenarioUnavailable)):
+            result['incomplete'].append(result['errors'][-1])
+    finally:
+        suite.write('cases.json', result['cases'])
+        try:
+            suite.finish(result)
+        except Exception as exc:
+            result['errors'].append('restart finalization: ' + str(exc))
+            result.update(passed=False, outcome='FAIL')
+        result.update(elapsed_seconds=time.monotonic() - started, coverage_dir=str(output))
+        suite.write('summary.json', result)
+    return result

tools/regression_local_cases.py

@@ -0,0 +1,109 @@
+"""Optional local scenarios share the Suite's verified Workspace and credentials."""
+
+from contextlib import closing
+import importlib
+import json
+from pathlib import Path
+import sqlite3
+import time
+import urllib.error
+import urllib.request
+
+GROUPS = {
+    'faults': 'regression_execution_cases',
+    'resources': 'regression_resource_cases',
+    'lifecycle': 'regression_lifecycle_cases',
+}
+
+
+class ScenarioUnavailable(RuntimeError):
+    """An explicit prerequisite is missing; this is not a passed assertion."""
+
+
+def require(condition, message):
+    if not condition:
+        raise ValueError(message)
+
+
+def endpoint(suite, path):
+    require(path.startswith('/') and not path.startswith('//'), 'Scenario API path must be relative to Core')
+    base = suite.connection.get('endpoint')
+    require(bool(base), 'Scenario needs the endpoint verified by lbctl whoami')
+    return base.rstrip('/') + path
+
+
+def auth_headers(suite):
+    source = suite.connection.get('credential_source', '')
+    key = suite.env.get(source.removeprefix('env:')) if source.startswith('env:') else None
+    require(bool(key), 'Scenario requires environment-backed lbctl credentials')
+    if key not in suite.secrets:
+        suite.secrets.append(key)
+    return {'Authorization': 'Bearer ' + key, 'X-Workspace-Id': suite.variables['workspace_uuid']}
+
+
+headers = auth_headers
+
+
+def api(suite, method, path, body=None, headers=None):
+    request_headers = dict(auth_headers(suite) if headers is None else headers)
+    request_headers.setdefault('X-Workspace-Id', suite.variables['workspace_uuid'])
+    if body is not None:
+        request_headers['Content-Type'] = 'application/json'
+    request = urllib.request.Request(
+        endpoint(suite, path),
+        method=method,
+        headers=request_headers,
+        data=json.dumps(body).encode() if body is not None else None,
+    )
+    opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
+    try:
+        response = opener.open(request, timeout=suite.timeout(90))
+    except urllib.error.HTTPError as exc:
+        response = exc
+    with response:
+        return response.status, json.load(response)
+
+
+def record_run(suite, row, run_id, expected_status):
+    """Register exact run ownership for the independent final ledger check."""
+    require(expected_status in {'completed', 'failed', 'cancelled', 'timeout'}, 'Expected Run status must be terminal')
+    path = Path(suite.path(suite.profile['smoke']['ledger_db'])).resolve()
+    deadline = time.monotonic() + suite.timeout(30)
+    with closing(sqlite3.connect(path.as_uri() + '?mode=ro', uri=True, timeout=5)) as connection:
+        connection.row_factory = sqlite3.Row
+        while True:
+            run = connection.execute(
+                'SELECT run_id, conversation_id, status, finished_at FROM agent_run '
+                'WHERE workspace_id=? AND run_id=? AND id>?',
+                (suite.variables['workspace_uuid'], run_id, row['after_id']),
+            ).fetchone()
+            if run and run['finished_at']:
+                observed = dict(run)
+                observed.update(expected_status=expected_status, after_id=row['after_id'])
+                row.setdefault('scenario_runs', []).append(observed)
+                require(run['status'] == expected_status, f'Run ended as {run["status"]}; expected {expected_status}')
+                return observed
+            if time.monotonic() >= deadline:
+                raise TimeoutError('Scenario Run did not reach a persisted terminal state')
+            time.sleep(0.1)
+
+
+def modules():
+    return {name: importlib.import_module(module) for name, module in GROUPS.items()}
+
+
+def add_cases(profile, groups):
+    groups = list(groups)
+    require(len(groups) == len(set(groups)), 'Scenario groups must not repeat')
+    require(not (set(groups) - GROUPS.keys()), 'Unknown local scenario group')
+    for name in groups:
+        importlib.import_module(GROUPS[name]).add_cases(profile)
+    profile['scenario_groups'] = groups
+
+
+def run_case(suite, case, row):
+    for name, module in modules().items():
+        if case['scenario'] in module.CASE_NAMES:
+            require(name in suite.profile.get('scenario_groups', []), f'Scenario group is disabled: {name}')
+            return module.run_case(suite, case, row)
+    raise ValueError(f'Unknown local scenario: {case["scenario"]}')

tools/regression_resource_cases.py

@@ -0,0 +1,342 @@
+"""Optional resource lifecycle scenarios against the authenticated Workspace APIs."""
+
+import copy
+import json
+from pathlib import Path
+import sys
+import tempfile
+from uuid import uuid4
+
+from regression_coverage import checked
+from regression_local_cases import ScenarioUnavailable, api, require
+from regression_suite import resolve
+
+GROUP = 'resources'
+CASE_NAMES = {'auth_revocation', 'agent_direct_execution', 'retrieval_delete_visibility'}
+
+
+def add_cases(profile):
+    resources = {item['id']: item for item in profile['resources']}
+    base = copy.deepcopy(resources['empty_knowledge_base'])
+    base.update(id='resource_knowledge_base')
+    base['body']['name'] = 'regression document lifecycle'
+    base.pop('documents', None)
+    profile['resources'].append(base)
+    for name, stage, expected in (
+        ('auth_revocation', 'auth', 'LANGBOT_AUTH_REVOCATION_OK'),
+        ('agent_direct_execution', 'agent', 'LANGBOT_CHAT_OK'),
+        ('retrieval_delete_visibility', 'retrieval', 'LANGBOT_DOCUMENT_DELETE_OK'),
+    ):
+        profile['cases'].append({'name': name, 'scenario': name, 'stage': stage, 'expect': expected})
+
+
+def data(suite, method, path, body=None, *, headers=None):
+    status, response = api(suite, method, path, body, headers=headers)
+    require(
+        status == 200 and isinstance(response, dict) and response.get('code') == 0,
+        f'{method} {path} did not return a successful response (HTTP {status})',
+    )
+    return response.get('data', {})
+
+
+def journal(suite, row):
+    row.setdefault('resource_journal', f'resource-case-{uuid4().hex}.json')
+    suite.write(
+        row['resource_journal'],
+        {
+            'workspace_uuid': suite.variables['workspace_uuid'],
+            'resources': row.get('owned_resources', []),
+            'cleanup_errors': row.get('cleanup_errors', []),
+        },
+    )
+
+
+def cleanup(suite, row, assertion, action):
+    original = sys.exc_info()[1]
+    try:
+        checked(row, assertion, action)
+    except Exception as exc:
+        message = f'{type(exc).__name__}: {exc}'
+        for secret in suite.secrets:
+            message = message.replace(secret, '[REDACTED]')
+        row.setdefault('cleanup_errors', []).append(message)
+        if original is None:
+            raise
+    finally:
+        journal(suite, row)
+
+
+def auth_revocation(suite, case, row):
+    token = suite.env.get('LANGBOT_QA_USER_TOKEN')
+    if not token:
+        raise ScenarioUnavailable('auth.revocation requires LANGBOT_QA_USER_TOKEN with api_key.manage')
+    if token not in suite.secrets:
+        suite.secrets.append(token)
+    manager = {'Authorization': 'Bearer ' + token}
+    name = 'qa-revoke-' + uuid4().hex
+    owned = {'kind': 'api-key', 'name': name, 'state': 'creating'}
+    row.setdefault('owned_resources', []).append(owned)
+    journal(suite, row)
+
+    def revoke_owned():
+        if 'id' not in owned:
+            keys = data(suite, 'GET', '/api/v1/apikeys', headers=manager)['keys']
+            matches = [key for key in keys if key.get('name') == name]
+            require(len(matches) == 1, 'Created key identity is unresolved; inspect the resource journal')
+            owned['id'] = matches[0]['id']
+            journal(suite, row)
+        path = f'/api/v1/apikeys/{owned["id"]}'
+        key = data(suite, 'GET', path, headers=manager)['key']
+        if key['status'] != 'revoked':
+            data(suite, 'DELETE', path, headers=manager)
+            key = data(suite, 'GET', path, headers=manager)['key']
+        require(key['status'] == 'revoked', 'Owned credential remains active after cleanup')
+        owned['state'] = 'revoked'
+
+    try:
+        created = data(suite, 'POST', '/api/v1/apikeys', {'name': name, 'scopes': ['resource.view']}, headers=manager)[
+            'key'
+        ]
+        owned.update(id=created['id'], uuid=created['uuid'], state='active')
+        journal(suite, row)
+        secret = created['key']
+        require(isinstance(secret, str) and bool(secret), 'Key creation did not return its one-time secret')
+        suite.secrets.append(secret)
+        child = {'X-API-Key': secret}
+        identity = data(suite, 'GET', '/api/v1/system/context', headers=child)
+        checked(
+            row,
+            'credential_active',
+            lambda: require(
+                identity.get('workspace_uuid') == suite.variables['workspace_uuid']
+                and identity.get('api_key_id') == owned['uuid'],
+                'New credential did not authenticate as its own Workspace key',
+            ),
+        )
+        data(suite, 'GET', '/api/v1/pipelines', headers=child)
+        data(suite, 'DELETE', f'/api/v1/apikeys/{owned["id"]}', headers=manager)
+        status, response = api(suite, 'GET', '/api/v1/pipelines', headers=child)
+        checked(
+            row,
+            'credential_revoked',
+            lambda: require(
+                status == 401 and isinstance(response, dict) and response.get('code') != 0,
+                'Revoked credential still authorizes resource access',
+            ),
+        )
+        key = data(suite, 'GET', f'/api/v1/apikeys/{owned["id"]}', headers=manager)['key']
+        checked(
+            row,
+            'key_status',
+            lambda: require(
+                key.get('status') == 'revoked' and key.get('secret_available') is False and 'key' not in key,
+                'Revoked key readback has incorrect status or exposes its secret',
+            ),
+        )
+        return {'result': case['expect'], 'key_id': owned['id'], 'status': 'revoked'}
+    finally:
+        cleanup(suite, row, 'key_cleanup', revoke_owned)
+
+
+def agent_direct_execution(suite, case, row):
+    template = next(item for item in suite.profile['resources'] if item['id'] == 'chat_pipeline')
+    config = resolve(copy.deepcopy(template['body']['config']['ai']), suite.variables)
+    name = 'qa-agent-' + uuid4().hex
+    conversation = 'qa-direct-' + uuid4().hex
+    owned = {'kind': 'agent', 'name': name, 'state': 'creating'}
+    row.setdefault('owned_resources', []).append(owned)
+    journal(suite, row)
+
+    def delete_owned():
+        if 'uuid' not in owned:
+            agents = data(suite, 'GET', '/api/v1/agents')['agents']
+            matches = [agent for agent in agents if agent.get('name') == name and agent.get('kind') == 'agent']
+            require(len(matches) == 1, 'Created Agent identity is unresolved; inspect the resource journal')
+            owned['uuid'] = matches[0]['uuid']
+            journal(suite, row)
+        path = f'/api/v1/agents/{owned["uuid"]}'
+        status, _ = api(suite, 'GET', path)
+        if status != 404:
+            require(status == 200, 'Cannot read owned Agent before cleanup')
+            data(suite, 'DELETE', path)
+        status, _ = api(suite, 'GET', path)
+        require(status == 404, 'Owned Agent remains visible after cleanup')
+        owned['state'] = 'deleted'
+
+    try:
+        created = data(suite, 'POST', '/api/v1/agents', {'name': name, 'kind': 'agent', 'config': config})
+        owned.update(uuid=created['uuid'], state='created')
+        journal(suite, row)
+        path = f'/api/v1/agents/{owned["uuid"]}'
+        reply = data(
+            suite, 'POST', path + '/debug', {'text': 'Run the direct Agent scenario', 'conversation_id': conversation}
+        )
+        runs = data(suite, 'GET', path + '/runs')['items']
+        matches = [run for run in runs if run.get('conversation_id') == conversation]
+        require(len(matches) == 1, 'Direct Agent request did not produce exactly one matching Run')
+        run = matches[0]
+        row.setdefault('scenario_runs', []).append(
+            {
+                'run_id': run['run_id'],
+                'conversation_id': conversation,
+                'expected_status': 'completed',
+                'after_id': row['after_id'],
+            }
+        )
+        checked(
+            row,
+            'agent_terminal',
+            lambda: require(
+                run.get('status') == 'completed' and bool(run.get('finished_at')),
+                'Direct Agent Run did not reach completed status',
+            ),
+        )
+        checked(
+            row,
+            'agent_response',
+            lambda: require(
+                reply.get('conversation_id') == conversation and case['expect'] in reply.get('final_text', ''),
+                'Direct Agent returned the wrong conversation or business reply',
+            ),
+        )
+        events = data(suite, 'GET', path + f'/runs/{run["run_id"]}/events')
+        checked(
+            row,
+            'agent_events',
+            lambda: require(
+                events.get('run', {}).get('run_id') == run['run_id']
+                and any(
+                    event.get('type') == 'run.completed' and event.get('run_id') == run['run_id']
+                    for event in events.get('items', [])
+                ),
+                'Direct Agent completion event is missing',
+            ),
+        )
+        return {'result': reply['final_text'], 'run_id': run['run_id'], 'conversation_id': conversation}
+    finally:
+        cleanup(suite, row, 'agent_cleanup', delete_owned)
+
+
+def retrieval_delete_visibility(suite, case, row):
+    kb_uuid = suite.variables['resource_knowledge_base']
+    path = f'/api/v1/knowledge/bases/{kb_uuid}'
+    before = {item['uuid'] for item in data(suite, 'GET', path + '/files')['files']}
+    owned = {'kind': 'knowledge-files', 'knowledge_base_uuid': kb_uuid, 'files': [], 'state': 'ingesting'}
+    row.setdefault('owned_resources', []).append(owned)
+    journal(suite, row)
+
+    def files():
+        return [item for item in data(suite, 'GET', path + '/files')['files'] if item['uuid'] not in before]
+
+    def cleanup_documents():
+        remaining = files()
+        owned['files'] = [{'uuid': item['uuid'], 'status': item['status']} for item in remaining]
+        journal(suite, row)
+        unresolved = [item['uuid'] for item in remaining if item['status'] in {'pending', 'processing', 'interrupted'}]
+        for item in remaining:
+            if item['uuid'] not in unresolved:
+                data(suite, 'DELETE', path + '/files/' + item['uuid'])
+        require(
+            not unresolved, 'Document ingestion outcome is unresolved; retain the owned knowledge base for inspection'
+        )
+        require(not files(), 'Owned documents remain visible after cleanup')
+        owned['state'] = 'deleted'
+
+    def retrieve(query):
+        found = data(suite, 'POST', path + '/retrieve', {'query': query, 'retrieval_settings': {'top_k': 10}})[
+            'results'
+        ]
+        require(
+            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',
+        )
+        return found
+
+    def contains(found, 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 found
+        )
+
+    try:
+        documents, markers = [], ['obsolete' + uuid4().hex, 'retained' + uuid4().hex]
+        with tempfile.TemporaryDirectory(prefix='resource-documents-', dir=suite.output) as directory:
+            for index, marker in enumerate(markers):
+                document = Path(directory) / f'document-{index}.txt'
+                document.write_text(f'Lifecycle fixture: {marker}. This is document {index}.\n')
+                previous = {item['uuid'] for item in files()}
+                suite.call(
+                    [
+                        'knowledge-base',
+                        'ingest',
+                        kb_uuid,
+                        '--file',
+                        str(document),
+                        '--parser-plugin-id',
+                        'langbot-team/GeneralParsers',
+                        '--wait',
+                    ]
+                )
+                added = [item for item in files() if item['uuid'] not in previous]
+                require(
+                    len(added) == 1 and added[0].get('status') == 'completed',
+                    'Document ingestion did not create exactly one completed file',
+                )
+                documents.append(added[0])
+                owned['files'].append({'uuid': added[0]['uuid'], 'status': added[0]['status']})
+                journal(suite, row)
+        initial = retrieve(' '.join(markers))
+        checked(
+            row,
+            'retrieval_initial',
+            lambda: require(
+                all(contains(initial, document, marker) for document, marker in zip(documents, markers)),
+                'Both fixture documents must be independently retrievable before deletion',
+            ),
+        )
+        data(suite, 'DELETE', path + '/files/' + documents[0]['uuid'])
+        after = retrieve(' '.join(markers))
+        checked(
+            row,
+            'retrieval_deleted',
+            lambda: require(
+                all(
+                    item.get('metadata', {}).get('file_id') != documents[0]['engine_document_id']
+                    and markers[0] not in json.dumps(item, ensure_ascii=False)
+                    for item in after
+                )
+                and documents[0]['uuid'] not in {item['uuid'] for item in files()},
+                'Deleted document remains visible in retrieval or the file list',
+            ),
+        )
+        checked(
+            row,
+            'retrieval_control',
+            lambda: require(
+                contains(after, documents[1], markers[1]),
+                'Deleting one document removed the surviving control document',
+            ),
+        )
+        return {
+            'result': case['expect'],
+            'deleted_document_id': documents[0]['uuid'],
+            'control_document_id': documents[1]['uuid'],
+        }
+    finally:
+        cleanup(suite, row, 'document_cleanup', cleanup_documents)
+
+
+def run_case(suite, case, row):
+    actions = {
+        'auth_revocation': auth_revocation,
+        'agent_direct_execution': agent_direct_execution,
+        'retrieval_delete_visibility': retrieval_delete_visibility,
+    }
+    return actions[case['scenario']](suite, case, row)

tools/regression_scenarios.py

@@ -11,7 +11,7 @@ ROOT = Path(__file__).parent
 RUNNER = 'plugin:langbot-team/LocalAgent/default'
 
 
-def expanded_profile():
+def expanded_profile(groups=()):
     profile = json.loads((ROOT / 'regression_profiles/smoke.json').read_text())
     profile['coverage_suite'] = 'scenarios'
     profile['smoke']['group_turns'] = 3
@@ -102,6 +102,10 @@ def expanded_profile():
             expect='LANGBOT_' + name.upper() + '_OK',
         )
         profile['cases'].append(case)
+    if groups:
+        from regression_local_cases import add_cases
+
+        add_cases(profile, groups)
     return profile
 
 

tools/regression_smoke.py

@@ -16,6 +16,7 @@ from uuid import uuid4
 
 from regression_traffic import group_request, percentile, read_endpoint_snapshot, reconcile_runs
 from regression_coverage import checked, check_response
+from regression_local_cases import ScenarioUnavailable
 
 
 class UnresolvedRun(RuntimeError):
@@ -316,7 +317,11 @@ def run_cycle(suite, rows, health, cycle):
                     token = uuid4().hex
                     offset = mock_path.stat().st_size if case.get('mock_requests') and mock_path.exists() else 0
                     try:
-                        if rejected:
+                        if 'scenario' in case:
+                            from regression_local_cases import run_case
+
+                            data = run_case(suite, case, row)
+                        elif rejected:
                             data = checked(
                                 row,
                                 'response',
@@ -358,7 +363,7 @@ def run_cycle(suite, rows, health, cycle):
                         row['output'] = data
                         if case.get('forbid'):
                             checked(row, 'isolation', lambda: check_response(data, case['expect'], case['forbid']))
-                        row['checks'] = {}
+                        row.setdefault('checks', {})
                         if case.get('fields'):
                             row['checks']['fields'] = checked(row, 'fields', lambda: check_fields(data, case['fields']))
                         if case.get('attachment'):
@@ -392,6 +397,8 @@ def run_cycle(suite, rows, health, cycle):
                         row['status'] = 'passed'
                         previous[group] = token
                         history_turns[group] += 1
+                    except ScenarioUnavailable as exc:
+                        row.update(status='skipped', error=redact(str(exc)))
                     except urllib.error.HTTPError as exc:
                         row.update(
                             status='unknown' if exc.code >= 500 else 'failed', error=f'HTTP {exc.code}; not retried'
@@ -413,6 +420,8 @@ def run_cycle(suite, rows, health, cycle):
                         if 'timed out' in message:
                             row['timed_out'] = True
                     row['seconds'] = time.monotonic() - before
+                    for error in row.get('cleanup_errors', []):
+                        errors.append(f'{case["name"]}: cleanup failed: {redact(str(error))}')
                     if row['status'] != 'passed':
                         errors.append(f'{case["name"]}: {row["status"]}: {row["error"]}')
                         if is_group:
@@ -462,7 +471,7 @@ def run_smoke(suite):
             }
         )
         suite.write('cycles.json', cycles)
-        if not execution or not execution.admitting() or any(row['status'] == 'unknown' for row in rows):
+        if not execution or not execution.admitting() or any(row['status'] in ('unknown', 'skipped') for row in rows):
             break
     result['errors'] = errors
     result['cycles'] = cycles
@@ -470,7 +479,7 @@ def run_smoke(suite):
     result['incomplete'] = [
         f'{row["case"]}: {row["status"]}: {row["error"]}'
         for row in rows
-        if row['status'] == 'unknown' and not row.get('timed_out')
+        if row['status'] in ('unknown', 'skipped') and not row.get('timed_out')
     ]
     if not result['first_pass_complete']:
         error = 'Required first pass did not finish before the admission deadline'
@@ -486,7 +495,7 @@ def run_smoke(suite):
             for name in dict.fromkeys(row['case'] for row in rows)
         },
     }
-    executed = [row for row in rows if row.get('requires_runner') and 'after_id' in row]
+    executed = scenario_run_rows(rows)
     result['test_data'] = {
         'observed_conversations': len({row['conversation_id'] for row in executed if 'conversation_id' in row}),
         'observed_agent_runs': len({row['run']['run_id'] for row in executed if 'run' in row}),
@@ -497,9 +506,24 @@ def run_smoke(suite):
     return result
 
 
+def scenario_run_rows(rows):
+    executed = [row for row in rows if row.get('requires_runner') and 'after_id' in row]
+    for row in rows:
+        for run in row.get('scenario_runs', []):
+            executed.append(
+                {
+                    'after_id': row['after_id'],
+                    'conversation_id': run['conversation_id'],
+                    'run': {'run_id': run['run_id']},
+                    'expected_status': run['expected_status'],
+                }
+            )
+    return executed
+
+
 def reconcile_scenario_runs(suite, result, *, key='ledger'):
     """Recheck the originally observed conversations without expanding run ownership."""
-    executed = [row for row in result.get('cases', []) if row.get('requires_runner') and 'after_id' in row]
+    executed = scenario_run_rows(result.get('cases', []))
     if not executed:
         return
     try:
@@ -509,6 +533,9 @@ def reconcile_scenario_runs(suite, result, *, key='ledger'):
             min(row['after_id'] for row in executed),
             {row['conversation_id'] for row in executed if 'conversation_id' in row},
             {row['run']['run_id'] for row in executed if 'run' in row},
+            expected_statuses={
+                row['run']['run_id']: row.get('expected_status', 'completed') for row in executed if 'run' in row
+            },
         )
     except (OSError, sqlite3.Error) as exc:
         error = f'Scenario {key} evidence unavailable: {exc}'
@@ -516,5 +543,5 @@ def reconcile_scenario_runs(suite, result, *, key='ledger'):
         result.setdefault('incomplete', []).append(error)
         return
     if not result[key]['matched']:
-        error = f'Scenario {key} found extra, missing or non-completed AgentRuns'
+        error = f'Scenario {key} found extra, missing or unexpected AgentRun states'
         result['errors'].append(error)

tools/regression_suite.py

@@ -83,6 +83,7 @@ def validate(profile):
         'coverage_suite',
         'execution',
         'performance_mode',
+        'scenario_groups',
     }
     if extra := profile.keys() - allowed:
         raise ValueError(f'unknown profile fields: {sorted(extra)}')
@@ -98,6 +99,16 @@ def validate(profile):
         raise ValueError('coverage_suite must be a stable suite identifier')
     if profile.get('performance_mode', 'gate') not in {'gate', 'observe'}:
         raise ValueError('performance_mode must be gate or observe')
+    if 'scenario_groups' in profile:
+        from regression_local_cases import GROUPS
+
+        groups = profile['scenario_groups']
+        if (
+            not isinstance(groups, list)
+            or any(not isinstance(name, str) or name not in GROUPS for name in groups)
+            or len(groups) != len(set(groups))
+        ):
+            raise ValueError('scenario_groups must contain unique supported local groups')
     execution = profile.get('execution')
     if execution is not None:
         if (
@@ -192,6 +203,7 @@ def validate(profile):
             'reject_group',
             'http_probe',
             'reset_group',
+            'scenario',
         }:
             raise ValueError('unknown case field')
         if not isinstance(case.get('name'), str) or not case['name'] or case['name'] in names:
@@ -250,7 +262,21 @@ def validate(profile):
             case['reset_group'] is not True or 'http_group' not in case or 'reject_group' in case
         ):
             raise ValueError('reset_group requires an accepted group case')
-        if 'http_probe' in case:
+        if 'scenario' in case:
+            from regression_local_cases import modules
+
+            handlers = modules()
+            if (
+                'smoke' not in profile
+                or not isinstance(case['scenario'], str)
+                or not any(
+                    case['scenario'] in module.CASE_NAMES and name in profile.get('scenario_groups', [])
+                    for name, module in handlers.items()
+                )
+                or any(key in case for key in ('args', 'body', 'http_probe', 'http_group', 'box_exec', 'attachment'))
+            ):
+                raise ValueError('scenario must name a supported finite local case without another request mode')
+        elif 'http_probe' in case:
             probe = case['http_probe']
             if (
                 not isinstance(probe, dict)

tools/regression_traffic.py

@@ -131,9 +131,14 @@ def group_request(case, group_id, member_id, turn, token, previous, env=None, *,
     return payload
 
 
-def reconcile_runs(path, workspace, after_id, conversations, expected_ids):
+def reconcile_runs(path, workspace, after_id, conversations, expected_ids, *, expected_statuses=None):
     """Catch delayed extra or nonterminal runs after request-level verification."""
     actual, unfinished = set(), []
+    expected_statuses = expected_statuses or {}
+    if set(expected_statuses) - set(expected_ids) or any(
+        status not in {'completed', 'failed', 'cancelled', 'timeout'} for status in expected_statuses.values()
+    ):
+        raise ValueError('Expected statuses must name owned runs and terminal states')
     connection = sqlite3.connect(Path(path).resolve().as_uri() + '?mode=ro', uri=True)
     try:
         values = sorted(conversations)
@@ -145,7 +150,7 @@ def reconcile_runs(path, workspace, after_id, conversations, expected_ids):
             )
             for run_id, status in connection.execute(query, [workspace, after_id, *part]):
                 actual.add(run_id)
-                if status != 'completed':
+                if status != expected_statuses.get(run_id, 'completed'):
                     unfinished.append(run_id)
     finally:
         connection.close()

tools/test_regression_coverage.py

@@ -291,8 +291,13 @@ class Routes:
                     path = Path(__file__).with_name('regression_profiles') / f'{name}.json'
                     if name == 'scenarios':
                         from regression_scenarios import expanded_profile
+                        from regression_local_cases import GROUPS
 
-                        profile = expanded_profile()
+                        profile = expanded_profile(GROUPS)
+                        if 'lifecycle' in GROUPS:
+                            from regression_lifecycle_cases import restart_profile
+
+                            profile['cases'] += restart_profile(profile)['cases']
                     else:
                         profile = json.loads(path.read_text())
                     profiles[name] = {x['name']: x for x in profile['cases']}

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

tools/test_regression_lifecycle_cases.py

@@ -0,0 +1,292 @@
+"""Challenge callback evidence and cleanup with real loopback HTTP receipts."""
+
+import json
+from pathlib import Path
+import socket
+import sqlite3
+import tempfile
+import threading
+import time
+from types import SimpleNamespace
+import unittest
+from unittest.mock import patch
+import urllib.error
+import urllib.request
+
+import regression_lifecycle_cases as lifecycle
+
+
+class LifecycleTest(unittest.TestCase):
+    def exercise(self, corruption=None):
+        suite = SimpleNamespace(
+            variables={'chat_pipeline': 'fixture-pipeline'}, secrets=[], timeout=lambda limit: limit
+        )
+        case = {'scenario': 'callback_retry_identity'}
+        row, settings, workers, receivers, errors = {'after_id': 0}, {}, [], [], []
+        calls = []
+        receiver_class = lifecycle.CallbackReceiver
+
+        def receiver(secret):
+            result = receiver_class(secret)
+            receivers.append(result)
+            return result
+
+        def callback(body, message_id):
+            payload = {
+                'session_id': body['session_id'],
+                'reply_to': message_id,
+                'sequence': 1,
+                'is_final': True,
+                'message': [{'type': 'Plain', 'text': 'LANGBOT_CHAT_OK'}],
+            }
+            secret = settings['outbound_secret'] + ('wrong' if corruption == 'signature' else '')
+            try:
+                for attempt in range(2):
+                    if corruption == 'body' and attempt:
+                        payload['timestamp'] = 'changed by retry'
+                    raw = json.dumps(payload).encode()
+                    stamp = str(int(time.time()))
+                    request = urllib.request.Request(
+                        settings['callback_url'],
+                        data=raw,
+                        headers={
+                            'Content-Type': 'application/json',
+                            'X-LB-Timestamp': stamp,
+                            'X-LB-Signature': lifecycle.signature(secret, raw, stamp),
+                        },
+                    )
+                    try:
+                        with urllib.request.urlopen(request, timeout=2) as response:
+                            response.read()
+                            return
+                    except urllib.error.HTTPError as exc:
+                        if exc.code != 503:
+                            return
+                errors.append('Callback was never accepted')
+            except Exception as exc:
+                errors.append(str(exc))
+
+        def api(_suite, method, path, body=None, headers=None):
+            calls.append((method, path))
+            if method == 'POST' and path == '/api/v1/platform/bots':
+                settings.update(body['adapter_config'])
+                self.assertEqual(body['event_bindings'][0]['target_uuid'], 'fixture-pipeline')
+                return 200, {'code': 0, 'data': {'uuid': 'temporary-bot'}}
+            if method == 'POST' and path == '/bots/temporary-bot':
+                self.assertEqual(
+                    headers['X-LB-Signature'],
+                    lifecycle.signature(
+                        settings['inbound_secret'], json.dumps(body).encode(), headers['X-LB-Timestamp']
+                    ),
+                )
+                message_id = 'in-' + str(len(workers))
+                worker = threading.Thread(target=callback, args=(body, message_id))
+                workers.append(worker)
+                worker.start()
+                return 202, {'code': 0, 'data': {'accepted_message_id': message_id}}
+            if method == 'DELETE' and path == '/api/v1/platform/bots/temporary-bot':
+                return 200, {'code': 0}
+            if method == 'GET' and path == '/api/v1/platform/bots/temporary-bot':
+                return 404, {'code': -1}
+            raise AssertionError((method, path))
+
+        with (
+            patch.object(lifecycle, 'api', side_effect=api),
+            patch.object(lifecycle, 'CallbackReceiver', side_effect=receiver),
+            patch.object(lifecycle, 'completed'),
+        ):
+            if corruption is None:
+                result = lifecycle.run_case(suite, case, row)
+                self.assertEqual(result['result'], 'LANGBOT_CALLBACK_RETRY_OK')
+                self.assertEqual([item['finals'] for item in result['deliveries']], [1, 1])
+                self.assertEqual(
+                    row['assertions'],
+                    {'callback_retry': 'passed', 'callback_followup': 'passed', 'callback_cleanup': 'passed'},
+                )
+            else:
+                with self.assertRaisesRegex(ValueError, 'signature|Retry changed'):
+                    lifecycle.run_case(suite, case, row)
+                self.assertEqual(row['assertions']['callback_retry'], 'failed')
+                self.assertEqual(row['assertions']['callback_cleanup'], 'passed')
+        for worker in workers:
+            worker.join(timeout=2)
+            self.assertFalse(worker.is_alive())
+        self.assertFalse(errors, errors)
+        self.assertIn(('DELETE', '/api/v1/platform/bots/temporary-bot'), calls)
+        self.assertFalse(receivers[0].thread.is_alive())
+        with socket.socket() as client:
+            self.assertNotEqual(client.connect_ex(receivers[0].server.server_address), 0)
+        return receivers[0].records
+
+    def test_retry_followup_and_cleanup_use_actual_http_receipts(self):
+        receipts = self.exercise()
+        self.assertEqual([item['status'] for item in receipts], [503, 200, 200])
+        self.assertEqual(receipts[0]['body'], receipts[1]['body'])
+
+    def test_modified_retry_cannot_pass_and_still_cleans_up(self):
+        self.exercise('body')
+
+    def test_invalid_callback_signature_cannot_pass_and_still_cleans_up(self):
+        self.exercise('signature')
+
+    def test_duplicate_final_is_not_a_successful_delivery(self):
+        payload = {'session_id': 'session', 'sequence': 1, 'is_final': True, 'message': []}
+        records = [{'status': 200, 'payload': payload, 'body': b'{}'}] * 2
+        with self.assertRaisesRegex(ValueError, 'order changed'):
+            lifecycle.verify_delivery(records, 'session', retried=False)
+
+    def test_unknown_callback_result_survives_cleanup_failure(self):
+        class NoDelivery:
+            url = 'http://127.0.0.1:9/test-only'
+
+            def __enter__(self):
+                return self
+
+            def __exit__(self, *args):
+                pass
+
+            def wait_final(self, *args):
+                raise TimeoutError('callback outcome unknown')
+
+        suite = SimpleNamespace(variables={'chat_pipeline': 'fixture'}, secrets=[], timeout=lambda limit: limit)
+        row = {'after_id': 0}
+        responses = [
+            (200, {'code': 0, 'data': {'uuid': 'temporary-bot'}}),
+            (202, {'code': 0, 'data': {'accepted_message_id': 'accepted'}}),
+            (500, {'code': -1}),
+        ]
+        with (
+            patch.object(lifecycle, 'CallbackReceiver', return_value=NoDelivery()),
+            patch.object(lifecycle, 'api', side_effect=responses),
+        ):
+            with self.assertRaisesRegex(TimeoutError, 'callback outcome unknown'):
+                lifecycle.run_case(suite, {'scenario': 'callback_retry_identity'}, row)
+        self.assertEqual(row['assertions']['callback_cleanup'], 'failed')
+        self.assertTrue(any('deletion failed' in error for error in row['cleanup_errors']))
+        self.assertNotIn('callback_retry', row['assertions'])
+
+    def test_restart_profile_reuses_fixture_resources_without_mutating_input(self):
+        path = Path(__file__).with_name('regression_profiles') / 'smoke.json'
+        original = json.loads(path.read_text())
+        original['execution'] = {'total_seconds': 1800, 'drain_seconds': 120, 'cooldown_seconds': 60}
+        snapshot = json.dumps(original, sort_keys=True)
+        profile = lifecycle.restart_profile(original)
+        self.assertEqual(json.dumps(original, sort_keys=True), snapshot)
+        self.assertNotIn('execution', profile)
+        self.assertEqual(profile['coverage_suite'], 'scenarios')
+        self.assertEqual(
+            [case['name'] for case in profile['cases']], ['core_restart_history', 'runtime_restart_recovery']
+        )
+        self.assertEqual(len(profile['resources']), 7)
+
+    def test_restart_preflight_records_real_ledger_rows_and_preserves_failed_history(self):
+        from regression_suite import Suite
+
+        for fail_history in (False, True):
+            with self.subTest(fail_history=fail_history), tempfile.TemporaryDirectory() as directory:
+                root = Path(directory)
+                path = root / 'runs.db'
+                with sqlite3.connect(path) as connection:
+                    connection.execute(
+                        'CREATE TABLE agent_run (id INTEGER PRIMARY KEY, run_id TEXT, workspace_id TEXT, '
+                        'conversation_id TEXT, status TEXT, finished_at TEXT, status_reason TEXT)'
+                    )
+                profile_path = Path(__file__).with_name('regression_profiles') / 'smoke.json'
+                profile = json.loads(profile_path.read_text())
+                profile['smoke']['ledger_db'] = str(path)
+                release = threading.Event()
+                restarts = []
+
+                def prepare(suite):
+                    suite.variables.update(workspace_uuid='workspace', chat_bot='chat-bot', restart_slow_bot='slow-bot')
+                    suite.journal()
+
+                def group_request(case, session_id, _member, turn, _token, previous, _env, *, timeout):
+                    is_slow = 'slow-bot' in case['http_group']['url']
+                    with sqlite3.connect(path) as connection:
+                        count = connection.execute('SELECT COUNT(*) FROM agent_run').fetchone()[0]
+                        run_id = f'run-{count}'
+                        connection.execute(
+                            'INSERT INTO agent_run(run_id,workspace_id,conversation_id,status,finished_at) VALUES(?,?,?,?,?)',
+                            (
+                                run_id,
+                                'workspace',
+                                'group_' + session_id,
+                                'running' if is_slow else 'completed',
+                                None if is_slow else 'done',
+                            ),
+                        )
+                    if is_slow:
+                        if not release.wait(2):
+                            raise TimeoutError('Runtime restart was never invoked')
+                        with sqlite3.connect(path) as connection:
+                            connection.execute(
+                                "UPDATE agent_run SET status='failed',finished_at='done' WHERE run_id=?", (run_id,)
+                            )
+                        raise RuntimeError('Runtime disconnected')
+                    if turn == 2:
+                        self.assertNotEqual(previous, 'none')
+                        if fail_history:
+                            raise ValueError('history rejected after restart')
+                    return {'data': {'session_id': session_id}}
+
+                def restart(name, *, timeout):
+                    self.assertGreater(timeout, 0)
+                    self.assertLessEqual(timeout, 60)
+                    restarts.append(name)
+                    if name == 'runtime':
+                        release.set()
+                    return {'old_pid': 10, 'new_pid': 20}
+
+                def public_api(_suite, method, resource, *, headers):
+                    self.assertEqual(method, 'GET')
+                    self.assertEqual(headers, {'Authorization': 'Bearer test-only-user-token'})
+                    run_id = resource.split('/')[-1].split('?')[0]
+                    with sqlite3.connect(path) as connection:
+                        status = connection.execute(
+                            'SELECT status FROM agent_run WHERE run_id=?', (run_id,)
+                        ).fetchone()[0]
+                    return 200, {'code': 0, 'data': {'run': {'run_id': run_id, 'status': status}}}
+
+                with (
+                    patch.object(Suite, 'prepare', prepare),
+                    patch.object(Suite, 'call', return_value={'workspace_uuid': 'workspace'}),
+                    patch('regression_traffic.group_request', side_effect=group_request),
+                    patch.object(lifecycle, 'api', side_effect=public_api),
+                ):
+                    result = lifecycle.run_restart_checks(
+                        profile,
+                        profile_path,
+                        root / 'preflight',
+                        {'LANGBOT_QA_USER_TOKEN': 'test-only-user-token'},
+                        restart,
+                    )
+                self.assertEqual(result['passed'], not fail_history, result)
+                self.assertEqual(
+                    json.loads((root / 'preflight/profile.json').read_text()), lifecycle.restart_profile(profile)
+                )
+                self.assertEqual(restarts, ['core'] if fail_history else ['core', 'runtime'])
+                self.assertEqual(json.loads((root / 'preflight/owned.json').read_text())['resources'], [])
+                if fail_history:
+                    self.assertTrue(any('history rejected' in message for message in result['errors']))
+                    self.assertEqual(result['cases'][0]['assertions']['history_after_restart'], 'failed')
+                else:
+                    self.assertEqual(result['cases'][1]['scenario_runs'][0]['expected_status'], 'failed')
+                    self.assertTrue(result['ledger_closeout']['matched'])
+                    self.assertEqual(sum(len(row['scenario_runs']) for row in result['cases']), 4)
+
+    def test_restart_missing_user_token_is_incomplete_before_preparation(self):
+        from regression_suite import Suite
+
+        path = Path(__file__).with_name('regression_profiles') / 'smoke.json'
+        profile = json.loads(path.read_text())
+        with tempfile.TemporaryDirectory() as directory, patch.object(Suite, 'prepare') as prepare:
+            result = lifecycle.run_restart_checks(profile, path, Path(directory) / 'preflight', {}, None)
+        prepare.assert_not_called()
+        self.assertFalse(result['passed'])
+        self.assertEqual(result['outcome'], 'INCOMPLETE')
+        self.assertTrue(any('LANGBOT_QA_USER_TOKEN' in error for error in result['errors']))
+
+
+if __name__ == '__main__':
+    unittest.main()

tools/test_regression_local_cases.py

@@ -0,0 +1,110 @@
+"""Keep optional cases and expected failures from turning incomplete evidence green."""
+
+import json
+from pathlib import Path
+import sqlite3
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from regression_local_cases import GROUPS, ScenarioUnavailable
+from regression_scenarios import expanded_profile
+from regression_smoke import run_smoke
+from regression_suite import Suite, validate
+from regression_traffic import reconcile_runs
+
+
+class LocalScenarioTests(unittest.TestCase):
+    def test_group_switches_keep_base_cases_and_reject_unknown_handlers(self):
+        base = expanded_profile()
+        for group in GROUPS:
+            profile = expanded_profile([group])
+            validate(profile)
+            self.assertEqual(profile['scenario_groups'], [group])
+            self.assertGreater(len(profile['cases']), len(base['cases']))
+            self.assertEqual(profile['cases'][: len(base['cases'])], base['cases'])
+        for groups in (['unknown'], ['resources', 'resources']):
+            with self.assertRaises(ValueError):
+                expanded_profile(groups)
+        profile = expanded_profile(['resources'])
+        profile['scenario_groups'] = []
+        with self.assertRaises(ValueError):
+            validate(profile)
+        profile['scenario_groups'] = ['resources']
+        profile['cases'][-1]['scenario'] = 'not-a-real-handler'
+        with self.assertRaises(ValueError):
+            validate(profile)
+
+    def test_missing_prerequisite_is_skipped_and_incomplete(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            with sqlite3.connect(root / 'ledger.db') as connection:
+                connection.execute('CREATE TABLE agent_run (id INTEGER)')
+            profile = {
+                'version': 1,
+                'cli': {'config': 'context.yaml'},
+                'smoke': {'ledger_db': 'ledger.db'},
+                'cases': [{'name': 'auth_revocation', 'scenario': 'auth_revocation', 'expect': 'OK'}],
+                'observe': {'health': {}},
+            }
+            suite = Suite(profile, root / 'profile.json', root)
+            with patch('regression_local_cases.run_case', side_effect=ScenarioUnavailable('Missing user credential')):
+                result = run_smoke(suite)
+            self.assertEqual(result['status'], {'skipped': 1})
+            self.assertEqual(result['errors'], result['incomplete'])
+            self.assertIn('Missing user credential', result['incomplete'][0])
+            self.assertFalse(result['cases'][0].get('assertions'))
+            self.assertEqual(json.loads((root / 'cases.json').read_text())[0]['status'], 'skipped')
+
+    def test_cleanup_failure_remains_fail_after_unknown_request(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            with sqlite3.connect(root / 'ledger.db') as connection:
+                connection.execute('CREATE TABLE agent_run (id INTEGER)')
+            profile = {
+                'version': 1,
+                'cli': {'config': 'context.yaml'},
+                'smoke': {'ledger_db': 'ledger.db'},
+                'cases': [{'name': 'agent_direct_execution', 'scenario': 'agent_direct_execution', 'expect': 'OK'}],
+                'observe': {'health': {}},
+            }
+            suite = Suite(profile, root / 'profile.json', root)
+
+            def fail(suite, case, row):
+                row['cleanup_errors'] = ['Agent deletion returned HTTP 500']
+                raise RuntimeError('Request outcome unknown')
+
+            with patch('regression_local_cases.run_case', side_effect=fail):
+                result = run_smoke(suite)
+            suite.finish(result)
+            self.assertEqual(result['outcome'], 'FAIL')
+            self.assertEqual(result['cases'][0]['status'], 'unknown')
+            self.assertTrue(any('cleanup failed' in message for message in result['errors']))
+
+    def test_expected_failure_does_not_allow_extra_or_running_runs(self):
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / 'ledger.db'
+            with sqlite3.connect(path) as connection:
+                connection.execute(
+                    'CREATE TABLE agent_run (id INTEGER, run_id TEXT, workspace_id TEXT, conversation_id TEXT, status TEXT)'
+                )
+                connection.execute("INSERT INTO agent_run VALUES (1, 'owned', 'ws', 'session', 'failed')")
+            args = (path, 'ws', 0, {'session'}, {'owned'})
+            self.assertFalse(reconcile_runs(*args)['matched'])
+            expected = {'owned': 'failed'}
+            self.assertTrue(reconcile_runs(*args, expected_statuses=expected)['matched'])
+            with sqlite3.connect(path) as connection:
+                connection.execute("UPDATE agent_run SET status='running'")
+            self.assertFalse(reconcile_runs(*args, expected_statuses=expected)['matched'])
+            with sqlite3.connect(path) as connection:
+                connection.execute("UPDATE agent_run SET status='failed'")
+                connection.execute("INSERT INTO agent_run VALUES (2, 'extra', 'ws', 'session', 'completed')")
+            result = reconcile_runs(*args, expected_statuses=expected)
+            self.assertFalse(result['matched'])
+            self.assertEqual(result['extra'], ['extra'])
+            with self.assertRaises(ValueError):
+                reconcile_runs(*args, expected_statuses={'owned': 'running'})
+
+
+if __name__ == '__main__':
+    unittest.main()

tools/test_regression_resource_cases.py

@@ -0,0 +1,212 @@
+"""Challenge public resource lifecycle oracles and their cleanup without starting services."""
+
+import json
+from pathlib import Path
+import tempfile
+from types import SimpleNamespace
+import unittest
+from unittest.mock import patch
+
+from regression_local_cases import ScenarioUnavailable
+from regression_resource_cases import run_case
+
+
+def success(data=None):
+    return 200, {'code': 0, 'data': data or {}}
+
+
+class ResourceCasesTests(unittest.TestCase):
+    def setUp(self):
+        temporary = tempfile.TemporaryDirectory()
+        self.addCleanup(temporary.cleanup)
+        self.root = Path(temporary.name)
+        self.suite = SimpleNamespace(
+            output=self.root,
+            env={'LANGBOT_QA_USER_TOKEN': 'private-user-token'},
+            secrets=[],
+            variables={'workspace_uuid': 'workspace', 'resource_knowledge_base': 'owned-kb'},
+            profile={
+                'resources': [
+                    {
+                        'id': 'chat_pipeline',
+                        'body': {'config': {'ai': {'runner': {'id': 'runner'}, 'runner_config': {'runner': {}}}}},
+                    }
+                ]
+            },
+            write=lambda name, value: (self.root / name).write_text(json.dumps(value)),
+        )
+
+    def test_missing_management_session_skips_without_creating_credentials(self):
+        self.suite.env.clear()
+        with patch('regression_resource_cases.api') as request, self.assertRaises(ScenarioUnavailable):
+            run_case(self.suite, {'scenario': 'auth_revocation'}, {})
+        request.assert_not_called()
+
+    def test_revocation_rejects_still_authorized_key_and_never_persists_secrets(self):
+        for stale_authorization in (False, True):
+            with self.subTest(stale_authorization=stale_authorization):
+                key = {'id': 7, 'uuid': 'managed-key', 'status': 'active', 'secret_available': False}
+                row = {'after_id': 0}
+
+                def request(suite, method, path, body=None, headers=None):
+                    if method == 'POST':
+                        return success({'key': {**key, 'key': 'private-child-secret', 'secret_available': True}})
+                    if method == 'DELETE':
+                        key['status'] = 'revoked'
+                        return success()
+                    if path == '/api/v1/system/context':
+                        self.assertEqual(headers, {'X-API-Key': 'private-child-secret'})
+                        return success({'workspace_uuid': 'workspace', 'api_key_id': 'managed-key'})
+                    if path == '/api/v1/pipelines':
+                        return (
+                            success({'pipelines': []})
+                            if key['status'] == 'active' or stale_authorization
+                            else (401, {'code': -1})
+                        )
+                    return success({'key': key})
+
+                with patch('regression_resource_cases.api', side_effect=request):
+                    case = {'scenario': 'auth_revocation', 'expect': 'REVOKED'}
+                    if stale_authorization:
+                        with self.assertRaisesRegex(ValueError, 'still authorizes'):
+                            run_case(self.suite, case, row)
+                        self.assertEqual(row['assertions']['credential_revoked'], 'failed')
+                    else:
+                        self.assertEqual(run_case(self.suite, case, row)['status'], 'revoked')
+                self.assertEqual(key['status'], 'revoked')
+                self.assertEqual(row['assertions']['key_cleanup'], 'passed')
+                evidence = json.dumps(row) + (self.root / row['resource_journal']).read_text()
+                self.assertNotIn('private-child-secret', evidence)
+                self.assertNotIn('private-user-token', evidence)
+
+    def test_direct_agent_records_run_and_cleans_after_failed_public_event_oracle(self):
+        for missing_event in (False, True):
+            with self.subTest(missing_event=missing_event):
+                state, row = {}, {'after_id': 12}
+
+                def request(suite, method, path, body=None, headers=None):
+                    if method == 'POST' and path == '/api/v1/agents':
+                        state.update(uuid='own-agent', name=body['name'])
+                        return success({'uuid': 'own-agent'})
+                    if path.endswith('/debug'):
+                        state['conversation'] = body['conversation_id']
+                        return success({'conversation_id': state['conversation'], 'final_text': 'DIRECT_OK'})
+                    run = {
+                        'run_id': 'own-run',
+                        'conversation_id': state.get('conversation'),
+                        'status': 'completed',
+                        'finished_at': 1,
+                    }
+                    if path.endswith('/runs'):
+                        return success({'items': [run]})
+                    if path.endswith('/events'):
+                        return success(
+                            {
+                                'run': run,
+                                'items': [] if missing_event else [{'type': 'run.completed', 'run_id': 'own-run'}],
+                            }
+                        )
+                    if method == 'DELETE':
+                        self.assertEqual(path, '/api/v1/agents/own-agent')
+                        state['deleted'] = True
+                        return success()
+                    return (404, {'code': -1}) if state.get('deleted') else success({'agent': state})
+
+                with patch('regression_resource_cases.api', side_effect=request):
+                    case = {'scenario': 'agent_direct_execution', 'expect': 'DIRECT_OK'}
+                    if missing_event:
+                        with self.assertRaisesRegex(ValueError, 'completion event'):
+                            run_case(self.suite, case, row)
+                    else:
+                        self.assertEqual(run_case(self.suite, case, row)['run_id'], 'own-run')
+                self.assertEqual(
+                    row['scenario_runs'],
+                    [
+                        {
+                            'run_id': 'own-run',
+                            'conversation_id': state['conversation'],
+                            'after_id': 12,
+                            'expected_status': 'completed',
+                        }
+                    ],
+                )
+                self.assertTrue(state['deleted'])
+                self.assertEqual(row['assertions']['agent_cleanup'], 'passed')
+
+    def test_unknown_agent_create_recovers_owned_identity_without_retrying_creation(self):
+        calls, state, row = [], {}, {'after_id': 0}
+
+        def request(suite, method, path, body=None, headers=None):
+            calls.append((method, path))
+            if method == 'POST':
+                state.update(uuid='own-agent', name=body['name'], kind='agent')
+                raise ConnectionResetError('creation response lost')
+            if path == '/api/v1/agents':
+                return success({'agents': [state, {'uuid': 'unrelated', 'name': 'other', 'kind': 'agent'}]})
+            if method == 'DELETE':
+                self.assertEqual(path, '/api/v1/agents/own-agent')
+                state['deleted'] = True
+                return success()
+            return (404, {'code': -1}) if state.get('deleted') else success({'agent': state})
+
+        with patch('regression_resource_cases.api', side_effect=request), self.assertRaises(ConnectionResetError):
+            run_case(self.suite, {'scenario': 'agent_direct_execution', 'expect': 'OK'}, row)
+        self.assertEqual(sum(method == 'POST' for method, _ in calls), 1)
+        self.assertEqual(row['owned_resources'][0]['state'], 'deleted')
+
+    def test_delete_requires_retrieval_absence_and_preserves_control_document(self):
+        for stale_vectors in (False, True):
+            with self.subTest(stale_vectors=stale_vectors):
+                documents, row = [], {'after_id': 0}
+
+                def ingest(args):
+                    content = Path(args[args.index('--file') + 1]).read_text()
+                    index = len(documents)
+                    documents.append(
+                        {
+                            'uuid': f'file-{index}',
+                            'engine_document_id': f'engine-{index}',
+                            'status': 'completed',
+                            'content': content,
+                            'deleted': False,
+                        }
+                    )
+                    return {}
+
+                def request(suite, method, path, body=None, headers=None):
+                    if method == 'DELETE':
+                        target = next(document for document in documents if path.endswith('/' + document['uuid']))
+                        target['deleted'] = True
+                        return success()
+                    if method == 'GET':
+                        return success({'files': [document for document in documents if not document['deleted']]})
+                    return success(
+                        {
+                            'results': [
+                                {
+                                    'metadata': {'file_id': document['engine_document_id']},
+                                    'content': [{'type': 'text', 'text': document['content']}],
+                                }
+                                for document in documents
+                                if not document['deleted'] or stale_vectors
+                            ]
+                        }
+                    )
+
+                self.suite.call = ingest
+                with patch('regression_resource_cases.api', side_effect=request):
+                    case = {'scenario': 'retrieval_delete_visibility', 'expect': 'DELETED'}
+                    if stale_vectors:
+                        with self.assertRaisesRegex(ValueError, 'remains visible'):
+                            run_case(self.suite, case, row)
+                        self.assertEqual(row['assertions']['retrieval_deleted'], 'failed')
+                    else:
+                        self.assertEqual(run_case(self.suite, case, row)['result'], 'DELETED')
+                        self.assertEqual(row['assertions']['retrieval_control'], 'passed')
+                self.assertTrue(all(document['deleted'] for document in documents))
+                self.assertEqual(row['assertions']['document_cleanup'], 'passed')
+                self.assertEqual(list(self.root.glob('resource-documents-*')), [])
+
+
+if __name__ == '__main__':
+    unittest.main()