test(tools): expand smoke coverage across execution stages

tools/README.md

@@ -293,22 +293,38 @@ uv run --no-sync python -m unittest discover -s tools -p 'test_*.py' -v
 
 ## Finite core smoke
 
-Use [`regression_profiles/smoke.json`](regression_profiles/smoke.json) for a short
-functional check on an existing Core. It provisions fresh providers, models,
-Pipelines, a knowledge base and a signed HTTP Bot through `lbctl`, ingests the
-HTML fixture through GeneralParsers, then sends nine requests across six cases:
-chat, two Docker Box exec calls, explicit file export, RAG chat, direct retrieval,
-and two interleaved groups with two turns each. Group members change between
-turns; the mock checks prior user/reply history and rejects cross-group history.
+Use [`regression_profiles/smoke.json`](regression_profiles/smoke.json) for a finite
+functional check on an existing Core. It prepares 35 resources through `lbctl`,
+ingests HTML and plain-text fixtures through GeneralParsers, then executes 28
+cases / 43 requests. Reports group the verdicts by stage:
 
-Start `mock_llm.py` with the bundled scenarios as described above. Copy the smoke
-profile and configure the CLI executable/config, service URLs, plugin archives,
-and `vars.core_db` to the target Core's existing SQLite `langbot.db`. Keep fixture
-paths valid relative to the copied profile. Set `plugins: []` to reuse installed
-plugins; LocalAgent, LangRAG and GeneralParsers must still be enabled and initialized.
-The embedding endpoint must serve the configured model (`all-MiniLM-L6-v2` by
-default). Generation is deterministic mock traffic; Core, plugins, Box, document
-parsing and retrieval execute normally.
+| Stage | Cases and independent checks |
+| --- | --- |
+| Configuration | Identity and plugins; provider, model, Pipeline and knowledge-base readback; document completed status and Workspace/KB ownership |
+| Ingress | Missing and incorrect signatures return HTTP 401 / code 40101 without creating a group AgentRun; valid signatures are exercised by group conversations |
+| Conversation | Ordinary and Unicode chat; chat, Box and RAG groups each use two interleaved groups with three turns and rotating members; model input must retain prior user/reply history without cross-group history |
+| Model | Slow SSE reply; HTTP 401, 429 and 503 primary failures followed by successful fallback; recorded model/status/stream sequence must match |
+| Tools | Docker write/read across rounds, two tool calls in one model response, and expected exit code 7 with separate stdout/stderr returned to the model |
+| Delivery | ASCII and Unicode filenames/content delivered as one attachment with exact bytes |
+| Retrieval | HTML and text parsing/ingestion, RAG model input, retrieval from two knowledge bases with distinct sentinels and KB ownership, and an empty knowledge base returning an empty list |
+
+Cleanup is checked separately: delete journaled API resources and confirm their
+UUIDs are absent from CLI list readback. Evidence goes to `cleanup-checks.json`;
+cleanup failures affect the overall verdict.
+
+Start a dedicated `mock_llm.py` with the bundled scenarios and `--record` as
+shown above. Copy the smoke profile and configure the CLI executable/config,
+service URLs, plugin archives, `vars.mock_records` to that mock recording file,
+and `vars.core_db` to the target Core's existing SQLite `langbot.db`. The suite
+must read both local files. Keep fixture paths valid relative to the copied
+profile. Do not share the mock with other traffic during this run: fallback
+checks compare requests appended during each case.
+
+Set `plugins: []` to reuse installed plugins; LocalAgent, LangRAG and
+GeneralParsers must still be enabled and initialized. The embedding endpoint
+must serve the configured model (`all-MiniLM-L6-v2` by default). Generation is
+deterministic mock traffic; Core, plugins, Box, document parsing and retrieval
+execute normally.
 
 ```sh
 uv run --no-sync python tools/regression_suite.py validate --profile /path/to/smoke.json
@@ -318,22 +334,31 @@ uv run --no-sync python tools/regression_suite.py run \
 
 A correct reply alone does not pass. Each Pipeline/group request must match one
 new AgentRun in the authenticated workspace with `completed` status and a persisted
-`run.completed` event. Box cases additionally check actual Docker execution,
-exit codes and sentinel output; export checks the delivered filename and exact
-attachment bytes. The ledger is opened read-only. This first version requires
+`run.completed` event. The ledger is opened read-only. This version requires
 local SQLite access; remote-only Core and PostgreSQL ledgers are not supported.
 The observation timeout does not extend the Runner deadline or retry requests.
 Unknown outcomes fail the suite; later turns of that group are skipped while
-independent cases continue.
+independent cases continue. An expected tool exit code of 7 passes only when
+its output is correct and the agent handles the result and completes normally.
 
-`summary.json` contains the verdict; exit code 0 means all checks and required
-cleanup passed. `cases.json` records per-request evidence, `environment.json`
-records identity/capabilities/plugins, and `owned.json` tracks created resources.
-Successful runs delete those API resources. Failures retain them by default; use
-`uv run --no-sync python tools/regression_suite.py cleanup --profile /path/to/smoke.json --output .qa/smoke-001`
-after inspection. Box sessions and AgentRun records follow the reused service's
-normal retention policy. Load override flags are rejected for smoke.
+`summary.json` contains the overall and per-stage verdicts; exit code 0 means all
+checks and required cleanup passed. `cases.json` records per-request evidence,
+`environment.json` records identity/capabilities/plugins, and `owned.json` tracks
+created resources. Successful runs delete those API resources. Failures retain
+them by default; after inspection use:
+
+```sh
+uv run --no-sync python tools/regression_suite.py cleanup \
+  --profile /path/to/smoke.json --output .qa/smoke-001
+```
+
+Box sessions and AgentRun records follow the reused service's normal retention
+policy. Load override flags are rejected for smoke. `stage` labels cases;
+`fields` compares exact response values by dotted path (including array indices),
+`forbid` rejects response substrings, and `mock_requests` checks the actual
+model/status/stream sequence. `box_exec` defaults to exit code 0; error-handling
+cases can specify `exit_code` and `stderr` explicitly.
 
 To use a fresh deployment, select this suite in `job.json`; the Job supplies its
-SQLite path and uses mock embedding. Smoke skips load warmup and timed phases.
-The dedicated Docker daemon requirement above still applies.
+SQLite/recording paths and uses mock embedding. Smoke skips load warmup and timed
+phases. The dedicated Docker daemon requirement above still applies.

tools/fixtures/smoke-isolation.txt

@@ -0,0 +1,2 @@
+The separate knowledge base sentinel is cedar-isolation-4826.
+This document belongs only to the second smoke knowledge base.

tools/mock_llm_scenarios.json

@@ -76,5 +76,94 @@
         "content": "LANGBOT_EXPORT_OK"
       }
     ]
+  },
+  "mock-smoke-unicode": {
+    "steps": [
+      {
+        "expect_prompt": "请原样处理：你好，世界 🌍",
+        "content": "你好，世界 🌍 LANGBOT_UNICODE_OK"
+      }
+    ]
+  },
+  "mock-smoke-exec-error": {
+    "steps": [
+      {
+        "tool_calls": [
+          {
+            "name": "exec",
+            "arguments": {
+              "command": "printf EXPECTED_STDOUT; printf EXPECTED_STDERR >&2; exit 7"
+            }
+          }
+        ]
+      },
+      {
+        "expect_tool_output": "EXPECTED_STDERR",
+        "content": "LANGBOT_EXEC_ERROR_HANDLED"
+      }
+    ]
+  },
+  "mock-smoke-parallel-tools": {
+    "steps": [
+      {
+        "tool_calls": [
+          {
+            "name": "exec",
+            "arguments": {
+              "command": "printf PARALLEL_OK_A"
+            }
+          },
+          {
+            "name": "exec",
+            "arguments": {
+              "command": "printf PARALLEL_OK_B"
+            }
+          }
+        ]
+      },
+      {
+        "expect_tool_output": "PARALLEL_OK_B",
+        "content": "LANGBOT_PARALLEL_TOOLS_OK"
+      }
+    ]
+  },
+  "mock-smoke-export-unicode": {
+    "steps": [
+      {
+        "tool_calls": [
+          {
+            "name": "exec",
+            "arguments": {
+              "command": "printf '你好，文件 🌍' > /tmp/输出.txt; cat /tmp/输出.txt"
+            }
+          }
+        ]
+      },
+      {
+        "expect_tool_output": "你好，文件 🌍",
+        "tool_calls": [
+          {
+            "name": "export_box_files",
+            "arguments": {
+              "paths": [
+                "/tmp/输出.txt"
+              ]
+            }
+          }
+        ]
+      },
+      {
+        "expect_tool_output": "accepted",
+        "content": "LANGBOT_UNICODE_EXPORT_OK"
+      }
+    ]
+  },
+  "mock-smoke-rag-other": {
+    "steps": [
+      {
+        "expect_prompt": "cedar-isolation-4826",
+        "content": "LANGBOT_OTHER_RAG_OK"
+      }
+    ]
   }
 }

tools/regression_job.py

@@ -305,6 +305,8 @@ class Job:
         profile = self.suite_profile
         profile.setdefault('vars', {})['core_base_url'] = endpoint
         if 'smoke' in profile:
+            if 'mock_records' in profile['smoke']:
+                profile['smoke']['mock_records'] = str(self.output / 'mock' / 'requests.jsonl')
             profile['smoke']['ledger_db'] = str(Path(self.state['record']['data_dir']) / 'langbot.db')
         profile['cli'] = {'executable': self.cli[0], 'config': str(self.cli_config)}
         profile['observe'] = {'health': health, 'pids': {}, 'containers': list(containers.values())}

tools/regression_smoke.py

@@ -8,6 +8,7 @@ from pathlib import Path
 import sqlite3
 import time
 import urllib.error
+import urllib.request
 from uuid import uuid4
 
 from regression_traffic import group_request, read_endpoint_snapshot
@@ -57,9 +58,10 @@ def check_run(connection, run, case):
         if len(calls) != expected['count'] or any(
             not isinstance(item, dict)
             or item.get('backend') != expected['backend']
-            or item.get('exit_code') != 0
+            or item.get('exit_code') != expected.get('exit_code', 0)
             or not isinstance(item.get('stdout'), str)
             or expected['stdout'] not in item['stdout']
+            or ('stderr' in expected and expected['stderr'] not in str(item.get('stderr', '')))
             for item in calls
         ):
             raise ValueError('Box execution count/backend/exit code/output did not match the oracle')
@@ -88,6 +90,92 @@ def check_attachment(data, expected):
     return {'name': expected['name'], 'bytes': len(content), 'content_matched': True}
 
 
+def check_fields(data, expected):
+    for path, value in expected.items():
+        actual = data
+        try:
+            for key in path.split('.'):
+                actual = actual[int(key)] if isinstance(actual, list) else actual[key]
+        except (KeyError, IndexError, TypeError, ValueError) as exc:
+            raise ValueError(f'Response field {path} is missing or malformed') from exc
+        if actual != value:
+            raise ValueError(f'Response field {path} differs from the expected value')
+    return expected
+
+
+def check_mock_requests(path, offset, expected):
+    deadline = time.monotonic() + 3
+    while True:
+        with path.open('rb') as stream:
+            stream.seek(offset)
+            lines = stream.read().splitlines()
+        records = [json.loads(line) for line in lines]
+        actual = [
+            {'model': row['request']['model'], 'status': row['status'], 'stream': row['request'].get('stream', False)}
+            for row in records
+            if row['path'] == '/v1/chat/completions'
+        ]
+        if len(actual) >= len(expected) or time.monotonic() >= deadline:
+            if actual != expected:
+                raise ValueError(f'Model request sequence differs: expected {expected}, got {actual}')
+            return actual
+        time.sleep(0.05)
+
+
+def rejected_group(case, group):
+    settings = case['http_group']
+    headers = {'Content-Type': 'application/json'}
+    if case['reject_group'] == 'bad_signature':
+        headers.update({'X-LB-Timestamp': str(int(time.time())), 'X-LB-Signature': 'sha256=' + '0' * 64})
+    request = urllib.request.Request(
+        settings['url'],
+        headers=headers,
+        data=json.dumps(
+            {
+                'session_id': group,
+                'session_type': 'group',
+                'sender': {'id': 'qa-rejected'},
+                'message': [{'type': 'Plain', 'text': settings['message']}],
+            }
+        ).encode(),
+    )
+    opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
+    try:
+        with opener.open(request, timeout=10):
+            raise ValueError('Unsigned/tampered group message was accepted')
+    except urllib.error.HTTPError as exc:
+        payload = json.load(exc)
+        if exc.code != 401 or not isinstance(payload, dict) or payload.get('code') != 40101:
+            raise ValueError('Expected HTTP 401 with bad_signature code 40101') from exc
+        return {'http_status': exc.code, 'code': payload['code']}
+
+
+def verify_cleanup(suite, deleted):
+    checks = []
+    for kind, model_type in dict.fromkeys((item['kind'], item.get('type')) for item in deleted):
+        args = [kind, 'list'] + (['--type', model_type] if model_type else [])
+        data = suite.call(args)
+        key = {
+            'provider': 'providers',
+            'model': 'models',
+            'pipeline': 'pipelines',
+            'bot': 'bots',
+            'knowledge-base': 'bases',
+        }[kind]
+        items = data.get(key) if isinstance(data, dict) else None
+        if not isinstance(items, list) or any(
+            not isinstance(item, dict) or not isinstance(item.get('uuid'), str) for item in items
+        ):
+            raise ValueError(f'Malformed {kind} list response during cleanup verification')
+        visible = {item['uuid'] for item in items}
+        expected_absent = {item['uuid'] for item in deleted if item['kind'] == kind and item.get('type') == model_type}
+        remaining = sorted(expected_absent & visible)
+        checks.append({'kind': kind, 'type': model_type, 'deleted_count': len(expected_absent), 'remaining': remaining})
+    suite.write('cleanup-checks.json', checks)
+    if any(row['remaining'] for row in checks):
+        raise ValueError('Deleted resources remain visible in CLI list readback')
+
+
 def run_smoke(suite):
     # Imported here because Suite dispatches to this module after profile validation.
     from regression_suite import resolve
@@ -96,6 +184,7 @@ def run_smoke(suite):
     path = Path(suite.path(settings['ledger_db'])).resolve()
     if not path.is_file():
         raise ValueError('smoke.ledger_db must be an existing target Core SQLite database')
+    mock_path = Path(suite.path(settings['mock_records'])) if settings.get('mock_records') else None
     rows, errors, health = [], [], []
     started = time.monotonic()
 
@@ -123,15 +212,23 @@ def run_smoke(suite):
         for original in suite.profile['cases']:
             case = resolve(original, suite.variables)
             is_group = 'http_group' in case
-            runner = is_group or case.get('args', [])[:2] == ['pipeline', 'run']
+            rejected = 'reject_group' in case
+            runner = not rejected and (is_group or case.get('args', [])[:2] == ['pipeline', 'run'])
             groups = (
-                [f'qa-smoke-{uuid4().hex}' for _ in range(settings.get('groups_per_case', 2))] if is_group else [None]
+                [f'qa-smoke-{uuid4().hex}' for _ in range(settings.get('groups_per_case', 2))]
+                if is_group and not rejected
+                else [f'qa-smoke-{uuid4().hex}' if rejected else None]
             )
             previous = {group: 'none' for group in groups}
             retired = set()
-            for turn in range(1, (settings.get('group_turns', 2) if is_group else 1) + 1):
+            for turn in range(1, (settings.get('group_turns', 2) if is_group and not rejected else 1) + 1):
                 for group in groups:
-                    row = {'case': case['name'], 'status': 'failed', 'requires_runner': runner}
+                    row = {
+                        'case': case['name'],
+                        'stage': case.get('stage', 'execution'),
+                        'status': 'failed',
+                        'requires_runner': runner,
+                    }
                     if is_group:
                         row.update(group_id=group, turn=turn, member_id=f'qa-member-{turn % 2:03d}')
                     if group in retired:
@@ -141,8 +238,17 @@ def run_smoke(suite):
                     before = time.monotonic()
                     after_id = connection.execute('SELECT COALESCE(MAX(id),0) FROM agent_run').fetchone()[0]
                     token = uuid4().hex
+                    offset = mock_path.stat().st_size if case.get('mock_requests') and mock_path.exists() else 0
                     try:
-                        if is_group:
+                        if rejected:
+                            data = rejected_group(case, group)
+                            count = connection.execute(
+                                'SELECT COUNT(*) FROM agent_run WHERE workspace_id=? AND conversation_id=? AND id>?',
+                                (suite.variables['workspace_uuid'], 'group_' + group, after_id),
+                            ).fetchone()[0]
+                            if count:
+                                raise ValueError('Rejected ingress created an AgentRun')
+                        elif is_group:
                             data = group_request(
                                 case, group, row['member_id'], turn, token, previous[group], suite.env
                             )['data']
@@ -151,6 +257,11 @@ def run_smoke(suite):
                             if case['expect'] not in json.dumps(data, ensure_ascii=False):
                                 raise ValueError('Expected business response was not returned')
                         row['output'] = data
+                        if any(value in json.dumps(data, ensure_ascii=False) for value in case.get('forbid', [])):
+                            raise ValueError('Response contains forbidden content from another knowledge base')
+                        row['checks'] = {}
+                        if case.get('fields'):
+                            row['checks']['fields'] = check_fields(data, case['fields'])
                         if runner:
                             if not isinstance(data, dict):
                                 raise ValueError('Runner response must be an object')
@@ -165,9 +276,13 @@ def run_smoke(suite):
                                 settings.get('timeout_seconds', 30),
                             )
                             row['run'] = run
-                            row['checks'] = check_run(connection, run, case)
+                            row['checks'].update(check_run(connection, run, case))
                             if case.get('attachment'):
                                 row['checks']['attachment'] = check_attachment(data, case['attachment'])
+                        if case.get('mock_requests'):
+                            row['checks']['mock_requests'] = check_mock_requests(
+                                mock_path, offset, case['mock_requests']
+                            )
                         row['status'] = 'passed'
                         previous[group] = token
                     except urllib.error.HTTPError as exc:
@@ -176,7 +291,7 @@ def run_smoke(suite):
                         )
                     except (UnresolvedRun, urllib.error.URLError, TimeoutError) as exc:
                         row.update(status='unknown', error=redact(str(exc)))
-                    except (RuntimeError, ValueError, OSError, KeyError, sqlite3.Error) as exc:
+                    except (RuntimeError, ValueError, OSError, KeyError, IndexError, sqlite3.Error) as exc:
                         message = redact(str(exc))
                         row.update(
                             status='unknown'
@@ -204,6 +319,10 @@ def run_smoke(suite):
         'errors': errors,
         'elapsed_seconds': time.monotonic() - started,
         'status': dict(Counter(row['status'] for row in rows)),
+        'stages': {
+            stage: dict(Counter(row['status'] for row in rows if row['stage'] == stage))
+            for stage in dict.fromkeys(row['stage'] for row in rows)
+        },
         'case_count': len(suite.profile['cases']),
         'requests': len(rows),
     }

tools/regression_suite.py

@@ -143,11 +143,59 @@ def validate(profile):
         raise ValueError('cases must be a non-empty array')
     names = set()
     for case in cases:
-        if case.keys() - {'name', 'weight', 'args', 'expect', 'body', 'http_group', 'box_exec', 'attachment'}:
+        if case.keys() - {
+            'name',
+            'weight',
+            'args',
+            'expect',
+            'body',
+            'http_group',
+            'box_exec',
+            'attachment',
+            'stage',
+            'mock_requests',
+            'fields',
+            'forbid',
+            'reject_group',
+        }:
             raise ValueError('unknown case field')
         if not isinstance(case.get('name'), str) or not case['name'] or case['name'] in names:
             raise ValueError('case names must be non-empty and unique')
         names.add(case['name'])
+        if 'stage' in case and (not isinstance(case['stage'], str) or not case['stage']):
+            raise ValueError('case stage must be a non-empty string')
+        if any(key in case for key in ('mock_requests', 'fields', 'forbid', 'reject_group')) and 'smoke' not in profile:
+            raise ValueError('extended case checks require smoke mode')
+        if 'fields' in case and (not isinstance(case['fields'], dict) or not case['fields']):
+            raise ValueError('fields must be a non-empty mapping of dotted paths to exact values')
+        if 'fields' in case and any(not isinstance(key, str) or not key for key in case['fields']):
+            raise ValueError('field paths must be non-empty strings')
+        if 'forbid' in case and (
+            not isinstance(case['forbid'], list)
+            or not case['forbid']
+            or any(not isinstance(item, str) or not item for item in case['forbid'])
+        ):
+            raise ValueError('forbid must contain non-empty response substrings')
+        if 'mock_requests' in case:
+            requests = case['mock_requests']
+            if not isinstance(requests, list) or not requests or not profile.get('smoke', {}).get('mock_records'):
+                raise ValueError('mock_requests requires expected requests and smoke.mock_records')
+            for request in requests:
+                if not isinstance(request, dict) or set(request) != {'model', 'status', 'stream'}:
+                    raise ValueError('mock_requests entries require model, status and stream')
+                if (
+                    not isinstance(request['model'], str)
+                    or not request['model']
+                    or not isinstance(request['stream'], bool)
+                ):
+                    raise ValueError('invalid mock request model/stream')
+                number(request['status'], 'mock request status', minimum=100, integer=True)
+        if 'reject_group' in case and (
+            'http_group' not in case
+            or case['reject_group'] not in ('missing_signature', 'bad_signature')
+            or any(key in case for key in ('box_exec', 'attachment', 'mock_requests'))
+        ):
+            raise ValueError('reject_group must name a signature rejection on an HTTP group case')
         if 'http_group' in case:
             settings = case['http_group']
             if (
@@ -176,8 +224,16 @@ def validate(profile):
             raise ValueError('case body must be an object and cannot coexist with --file')
         if 'box_exec' in case:
             check = case['box_exec']
-            if not isinstance(check, dict) or set(check) != {'count', 'backend', 'stdout'}:
+            if (
+                not isinstance(check, dict)
+                or not {'count', 'backend', 'stdout'} <= set(check)
+                or check.keys() - {'count', 'backend', 'stdout', 'exit_code', 'stderr'}
+            ):
                 raise ValueError('box_exec requires count, backend and stdout')
+            if 'exit_code' in check:
+                number(check['exit_code'], 'box_exec.exit_code', minimum=0, integer=True)
+            if 'stderr' in check and not isinstance(check['stderr'], str):
+                raise ValueError('box_exec.stderr must be a string')
             number(check['count'], 'box_exec.count', minimum=1, integer=True)
             if any(not isinstance(check[k], str) or not check[k] for k in ('backend', 'stdout')):
                 raise ValueError('box_exec backend/stdout must be non-empty strings')
@@ -201,6 +257,7 @@ def validate(profile):
     if 'smoke' in profile:
         if not isinstance(smoke, dict) or smoke.keys() - {
             'ledger_db',
+            'mock_records',
             'timeout_seconds',
             'group_turns',
             'groups_per_case',
@@ -209,6 +266,8 @@ def validate(profile):
             raise ValueError('invalid smoke settings')
         if not isinstance(smoke.get('ledger_db'), str) or not smoke['ledger_db']:
             raise ValueError('smoke.ledger_db must name the target Core SQLite database')
+        if 'mock_records' in smoke and (not isinstance(smoke['mock_records'], str) or not smoke['mock_records']):
+            raise ValueError('smoke.mock_records must name the dedicated mock recording file')
         resolve(smoke, variables)
         number(smoke.get('timeout_seconds', 30), 'smoke timeout_seconds', minimum=0.001)
         number(smoke.get('group_turns', 2), 'smoke group_turns', minimum=2, integer=True)
@@ -486,7 +545,8 @@ class Suite:
         if saved['workspace_uuid'] != identity['workspace_uuid']:
             raise RuntimeError('cleanup refused: Workspace differs from owned.json')
         failures = []
-        for resource in reversed(saved['resources'].copy()):
+        original = saved['resources'].copy()
+        for resource in reversed(original):
             args = [resource['kind'], 'delete', resource['uuid'], '--yes']
             if resource['kind'] == 'model':
                 args += ['--type', resource['type']]
@@ -497,6 +557,13 @@ class Suite:
             else:
                 saved['resources'].remove(resource)
                 self.write('owned.json', saved)
+        if 'smoke' in self.profile:
+            from regression_smoke import verify_cleanup
+
+            try:
+                verify_cleanup(self, [item for item in original if item not in saved['resources']])
+            except (RuntimeError, ValueError, OSError, KeyError) as exc:
+                failures.append(f'cleanup readback failed: {exc}')
         return failures
 
     def run(self):

tools/test_regression_smoke.py

@@ -5,9 +5,19 @@ import copy
 import json
 from pathlib import Path
 import sqlite3
+import tempfile
+from types import SimpleNamespace
 import unittest
 
-from regression_smoke import UnresolvedRun, check_attachment, check_run, completed_run
+from regression_smoke import (
+    UnresolvedRun,
+    check_attachment,
+    check_fields,
+    check_mock_requests,
+    check_run,
+    completed_run,
+    verify_cleanup,
+)
 from regression_suite import validate
 
 
@@ -88,6 +98,93 @@ class SmokeOracleTest(unittest.TestCase):
             with self.subTest(parts=parts), self.assertRaises(ValueError):
                 check_attachment({'replies': [{'message_chain': parts}]}, expected)
 
+    def test_fields_check_empty_results_and_missing_paths(self):
+        check_fields({'results': []}, {'results': []})
+        check_fields({'results': [{'metadata': {'kb': 'owned'}}]}, {'results.0.metadata.kb': 'owned'})
+        for data in [{'results': [1]}, {}, {'results': None}]:
+            with self.subTest(data=data), self.assertRaises(ValueError):
+                check_fields(data, {'results': []})
+        with self.assertRaises(ValueError):
+            check_fields({'results': None}, {'results.0.metadata.kb': 'owned'})
+
+    def test_fallback_must_call_failed_primary_then_streaming_backup(self):
+        expected = [
+            {'model': 'primary', 'status': 429, 'stream': True},
+            {'model': 'backup', 'status': 200, 'stream': True},
+        ]
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / 'requests.jsonl'
+            prefix = b'earlier request ignored\n'
+            for rows in [
+                expected,
+                list(reversed(expected)),
+                [expected[0], {**expected[1], 'stream': False}],
+                [expected[0], {**expected[1], 'status': 503}],
+            ]:
+                path.write_bytes(
+                    prefix
+                    + b''.join(
+                        (
+                            json.dumps(
+                                {
+                                    'path': '/v1/chat/completions',
+                                    'status': row['status'],
+                                    'request': {'model': row['model'], 'stream': row['stream']},
+                                }
+                            )
+                            + '\n'
+                        ).encode()
+                        for row in rows
+                    )
+                )
+                if rows == expected:
+                    self.assertEqual(check_mock_requests(path, len(prefix), expected), expected)
+                else:
+                    with self.assertRaises(ValueError):
+                        check_mock_requests(path, len(prefix), expected)
+
+    def test_expected_tool_failure_requires_exact_exit_and_stderr(self):
+        case = {'box_exec': {'count': 1, 'backend': 'docker', 'stdout': 'out', 'exit_code': 7, 'stderr': 'err'}}
+        for exit_code, stderr in [(7, 'err'), (0, 'err'), (7, '')]:
+            self.db.execute('DELETE FROM agent_run_event WHERE sequence=2')
+            self.db.execute(
+                'INSERT INTO agent_run_event VALUES (?, ?, ?, ?)',
+                (
+                    'run',
+                    2,
+                    'tool.call.completed',
+                    json.dumps(
+                        {
+                            'tool_name': 'exec',
+                            'result': {'backend': 'docker', 'stdout': 'out', 'exit_code': exit_code, 'stderr': stderr},
+                        }
+                    ),
+                ),
+            )
+            if exit_code == 7 and stderr == 'err':
+                check_run(self.db, self.run_row(), case)
+            else:
+                with self.assertRaises(ValueError):
+                    check_run(self.db, self.run_row(), case)
+
+    def test_cleanup_readback_rejects_resources_still_visible(self):
+        deleted = [{'kind': 'pipeline', 'uuid': 'owned'}]
+        evidence = {}
+        for visible in ['unrelated', 'owned']:
+            suite = SimpleNamespace(
+                call=lambda args: {'pipelines': [{'uuid': visible}]},
+                write=lambda name, value: evidence.update({name: value}),
+            )
+            if visible == 'owned':
+                with self.assertRaises(ValueError):
+                    verify_cleanup(suite, deleted)
+            else:
+                verify_cleanup(suite, deleted)
+                self.assertEqual(evidence['cleanup-checks.json'][0]['remaining'], [])
+        suite.call = lambda args: {'unexpected': []}
+        with self.assertRaises(ValueError):
+            verify_cleanup(suite, deleted)
+
     def test_profile_supports_mixed_cases_and_rejects_ignored_oracles(self):
         profile = json.loads(Path(__file__).with_name('regression_profiles').joinpath('smoke.json').read_text())
         validate(profile)
@@ -95,7 +192,11 @@ class SmokeOracleTest(unittest.TestCase):
             with self.subTest(changes=changes), self.assertRaises(ValueError):
                 validate({**profile, **changes})
         wrong = copy.deepcopy(profile)
-        wrong['cases'][4]['box_exec'] = {'count': 1, 'backend': 'docker', 'stdout': 'x'}
+        next(case for case in wrong['cases'] if case['name'] == 'rag_retrieve')['box_exec'] = {
+            'count': 1,
+            'backend': 'docker',
+            'stdout': 'x',
+        }
         with self.assertRaises(ValueError):
             validate(wrong)