test(regression): add concurrent conversation isolation job (#2811)

Squash merge after conflict resolution and all required checks passed.

tools/README.md

@@ -561,6 +561,34 @@ An HTTP route hit includes rejected requests and is not proof of all behavior
 behind that route. Invalid MCP modes must return HTTP 400 and leave the original
 configuration unchanged. Any failed contract or cleanup makes the job fail.
 
+## Concurrent isolation job
+
+Run this finite job on a disposable Linux/Docker host:
+
+```sh
+uv run --no-sync python tools/regression_ci.py --suite isolation \
+  --lbctl /path/to/lbctl --plugins plugins --output .qa/isolation-001
+```
+
+It uses two concurrent requests in one Workspace. Three scenarios cover two
+sessions on one Pipeline, two sessions on different Pipelines/models, and
+cancelling an active Box tool while a neighboring conversation completes.
+The first two scenarios each make two turns per session, checking request tokens,
+previous-turn history, Bot identity, model routing and exact terminal results.
+The cancellation scenario reuses the delayed-side-effect check and requires the
+Box to become idle. Read-only ledger intervals must overlap; serialized execution
+cannot pass as a concurrency test. Every accepted request must match exactly one
+terminal run, and mock records must contain exactly the owned model calls.
+
+HTTP Bot `/sync` collects cumulative stream snapshots, so each text component
+must be a monotonic prefix of that request's expected reply and the last must be
+complete. The job cannot count final delivery frames on this lossy endpoint.
+The 300-second guard bounds execution after preparation; there are no throughput
+or latency thresholds. `gate/suite/cases.json` retains per-request evidence and
+`gate/gate.json` the overall verdict. The job is explicit and does not change
+default daily or PR smoke cases. It does not test cross-Workspace isolation or
+high concurrency.
+
 ## Sustained regression and baseline comparison
 
 [`regression_profiles/endurance.json`](regression_profiles/endurance.json) runs

tools/regression_ci.py

@@ -21,7 +21,15 @@ from regression_suite import Suite
 from regression_traffic import owned_box_containers, read_endpoint_snapshot
 
 ROOT = Path(__file__).resolve().parents[1]
-SUITE_BUDGETS = {'smoke': 60, 'scenarios': 300, 'daily': 1860, 'endurance': 900, 'api': 600, 'state': 300}
+SUITE_BUDGETS = {
+    'smoke': 60,
+    'scenarios': 300,
+    'daily': 1860,
+    'endurance': 900,
+    'api': 600,
+    'state': 300,
+    'isolation': 300,
+}
 sys.path.insert(0, str(ROOT))
 from tests.e2e.utils.config_factory import create_minimal_config, create_test_directories  # noqa: E402
 
@@ -44,6 +52,10 @@ def suite_profile(name, run_id, groups=(), api_seed=1):
         profile.update(resources=[], cases=[], coverage_suite='api')
         profile['vars']['api_seed'] = api_seed
         add_cases(profile)
+    if name == 'isolation':
+        from regression_isolation_cases import isolation_profile
+
+        profile = isolation_profile(profile)
     for resource in profile['resources']:
         if resource['kind'] == 'pipeline':
             resource['body'].setdefault(
@@ -183,8 +195,8 @@ def run(args):
         groups = args.scenario_groups
         if groups is None:
             groups = list(DEFAULT_GROUPS) if args.suite in ('scenarios', 'daily') else []
-            if args.suite == 'api':
-                groups = ['api']
+            if args.suite in ('api', 'isolation'):
+                groups = [args.suite]
             if 'plugins' in groups and not {'misc/EssentialCommands', 'misc/FAQManager'} <= set(args.extra_plugins):
                 groups.remove('plugins')
         profile, profile_path = suite_profile(args.suite, secrets.token_hex(8), groups, args.api_seed)

tools/regression_contracts.json

@@ -449,6 +449,9 @@
      "area": "state", "kind": "blackbox", "priority": "P1",
      "description": "Knowledge documents survive restarts; replacement preserves control content and deletion removes stale results",
      "bindings": [{"profile": "state", "case": "durable_state_lifecycle", "checks": ["created_retrievable", "after_runtime_retrievable", "after_core_retrievable", "updated_retrievable", "updated_obsolete_absent", "persisted_update_retrievable", "persisted_update_obsolete_absent", "documents_deleted", "deleted_retrieval_empty"]}]
-    }
+    },
+    {"id": "isolation.same_pipeline", "area": "isolation", "kind": "graybox", "priority": "P1", "description": "Concurrent sessions in one Pipeline retain their own request identity and prior history", "bindings": [{"profile": "isolation", "case": "isolation.same_pipeline", "checks": ["conversation_isolation", "overlap_turn_1", "overlap_turn_2", "wire_identity_history_once"]}]},
+    {"id": "isolation.cross_pipeline", "area": "isolation", "kind": "graybox", "priority": "P1", "description": "Concurrent Pipelines retain their own model routing, Bot identity, and conversation history", "bindings": [{"profile": "isolation", "case": "isolation.cross_pipeline", "checks": ["conversation_isolation", "overlap_turn_1", "overlap_turn_2", "wire_identity_history_once"]}]},
+    {"id": "isolation.cancel_neighbor", "area": "isolation", "kind": "graybox", "priority": "P1", "description": "Cancelling an active tool run leaves a concurrent neighbor completed and no delayed tool side effect", "bindings": [{"profile": "isolation", "case": "isolation.cancel_neighbor", "checks": ["cancel_isolation", "neighbor_active_at_cancel", "cancellation_overlapped_neighbor", "neighbor_identity_once", "no_late_side_effect", "box_idle"]}]}
   ]
 }

tools/regression_execution_cases.py

@@ -97,7 +97,7 @@ def _running_box(suite, session_id):
     return _poll(suite, read, 'The long exec never became active in its Box')
 
 
-def _stream(suite, agent, scope, row, *, interrupt=None, box_session_id=None, frame_key='frames'):
+def _stream(suite, agent, scope, row, *, interrupt=None, box_session_id=None, frame_key='frames', before_cancel=None):
     request = urllib.request.Request(
         endpoint(suite, f'/api/v1/agents/{agent}/debug/stream'),
         data=json.dumps({'text': f'Run the QA scenario LB_MOCK_SCOPE={scope}', 'conversation_id': scope}).encode(),
@@ -130,6 +130,8 @@ def _stream(suite, agent, scope, row, *, interrupt=None, box_session_id=None, fr
                     row['box_session_id'] = checked(row, 'exec_active', lambda: _running_box(suite, box_session_id))
                     row['box_before_interrupt'] = _sessions(suite).get(row['box_session_id'])
                     if interrupt == 'cancel':
+                        if before_cancel is not None:
+                            before_cancel()
                         break  # Closing this public debug stream cancels its run.
             if frame.get('kind') in {'completed', 'error'}:
                 return frames
@@ -225,7 +227,7 @@ def _cleanup(suite, agent, scope, row, expected):
         raise RuntimeError('; '.join(([str(original)] if original else []) + errors))
 
 
-def run_case(suite, case, row):
+def run_case(suite, case, row, *, before_cancel=None):
     scenario = case['scenario']
     require(scenario in CASE_NAMES, 'Unknown local execution scenario')
     scope = 'qa-fault-' + uuid4().hex
@@ -277,6 +279,7 @@ def run_case(suite, case, row):
             row,
             interrupt='cancel' if scenario == 'execution.cancel_cleanup' else 'timeout' if interrupt else None,
             box_session_id=box_session_id,
+            before_cancel=before_cancel,
         )
         row['frames'] = frames
         run, events = _terminal(suite, agent, scope, row, expected)

tools/regression_isolation_cases.py

@@ -0,0 +1,328 @@
+"""Bounded concurrent conversations and cancellation isolation in one Workspace."""
+
+from concurrent.futures import ThreadPoolExecutor
+from contextlib import ExitStack, closing
+import copy
+import json
+from pathlib import Path
+import threading
+import secrets
+from uuid import uuid4
+
+import regression_execution_cases as execution
+from regression_api_checks import owned_resource
+from regression_coverage import checked
+from regression_lifecycle_cases import after_id, ledger
+from regression_local_cases import record_run, require
+from regression_smoke import check_run, completed_run
+from regression_traffic import group_request
+
+CASE_NAMES = ('isolation.same_pipeline', 'isolation.cross_pipeline', 'isolation.cancel_neighbor')
+
+
+def add_cases(profile):
+    require(profile.get('coverage_suite') == 'isolation', 'Isolation requires its dedicated suite')
+    profile['cases'] = [
+        {'name': name, 'scenario': name, 'stage': 'isolation', 'expect': 'ISOLATION_OK'} for name in CASE_NAMES
+    ]
+    profile['scenario_groups'] = ['isolation']
+
+
+def isolation_profile(base):
+    profile = copy.deepcopy(base)
+    execution.add_cases(profile)
+    wanted = {
+        'mock_provider',
+        'chat_model',
+        'slow_model',
+        'chat_pipeline',
+        'fault_interrupt_model',
+        'fault_interrupt_check_model',
+    }
+    profile['resources'] = [item for item in profile['resources'] if item['id'] in wanted]
+    resources = {item['id']: item for item in profile['resources']}
+    resources['slow_model']['body']['name'] = 'mock-slow'
+    resources['chat_model']['body']['name'] = 'mock-slow-stream'
+    resources['chat_pipeline']['body']['config']['ai']['runner_config'][execution.RUNNER]['model']['primary'] = (
+        '${slow_model}'
+    )
+    other = copy.deepcopy(resources['chat_pipeline'])
+    other['id'], other['body']['name'] = 'isolation_pipeline', 'regression isolation peer'
+    other['body']['config']['ai']['runner_config'][execution.RUNNER]['model']['primary'] = '${chat_model}'
+    profile['resources'].append(other)
+    profile['coverage_suite'] = 'isolation'
+    add_cases(profile)
+    return profile
+
+
+def lane(suite, bot, expected, model):
+    return {
+        'group': 'qa-isolation-' + uuid4().hex,
+        'previous': 'none',
+        'model': model,
+        'bot': bot,
+        'case': {
+            'expect': expected,
+            'http_group': {
+                'url': suite.variables['core_base_url'] + '/bots/' + bot + '/sync',
+                'secret_env': 'LANGBOT_QA_GROUP_SECRET',
+                'message': 'ai Run the isolation QA scenario',
+            },
+        },
+    }
+
+
+def check_reply(parts, expected):
+    # /sync retains cumulative stream snapshots and drops their final/sequence metadata.
+    texts = [part.get('text') for part in parts if part.get('type') == 'Plain']
+    require(len(texts) == len(parts) and bool(texts), 'Reply must contain only text snapshots')
+    require(all(isinstance(text, str) and expected.startswith(text) for text in texts), 'Reply contains foreign output')
+    require(
+        texts[-1] == expected and all(b.startswith(a) for a, b in zip(texts, texts[1:])),
+        'Reply lost its final output or reversed its stream',
+    )
+
+
+def turn(suite, item, number, barrier, receipt):
+    receipt.update(
+        group=item['group'],
+        turn=number,
+        token=uuid4().hex,
+        after_id=after_id(suite),
+        model=item['model'],
+        expected=item['case']['expect'],
+    )
+    barrier.wait(timeout=suite.timeout(10))
+    response_error = None
+    try:
+        receipt['response'] = checked(
+            receipt,
+            'response_identity',
+            lambda: group_request(
+                item['case'],
+                item['group'],
+                'qa-isolation-member',
+                number,
+                receipt['token'],
+                item['previous'],
+                suite.env,
+                timeout=suite.timeout(45),
+            ),
+        )
+        checked(
+            receipt,
+            'exact_reply',
+            lambda: check_reply(
+                receipt['response']['data']['message'], item['case']['expect'] + ' GROUP_ACK=' + receipt['token']
+            ),
+        )
+    except Exception as exc:
+        response_error = exc
+    try:
+        with closing(ledger(suite)) as db:
+            run = completed_run(
+                db, suite.variables['workspace_uuid'], 'group_' + item['group'], receipt['after_id'], suite.timeout(20)
+            )
+            checked(receipt, 'run_events', lambda: check_run(db, run, item['case']))
+            receipt['interval'] = dict(
+                db.execute(
+                    'SELECT run_id, started_at, finished_at, bot_id FROM agent_run WHERE run_id=?', (run['run_id'],)
+                ).fetchone()
+            )
+        checked(
+            receipt,
+            'bot_identity',
+            lambda: require(receipt['interval']['bot_id'] == item['bot'], 'Run belongs to a different Bot'),
+        )
+        record_run(suite, receipt, run['run_id'], 'completed')
+    except Exception as exc:
+        if response_error is not None:
+            receipt['ledger_error'] = str(exc)
+            raise response_error from exc
+        raise
+    if response_error is not None:
+        raise response_error
+    item['previous'] = receipt['token']
+    return receipt
+
+
+def overlapping(intervals):
+    require(
+        len(intervals) == 2 and all(x.get('started_at') and x.get('finished_at') for x in intervals),
+        'Two persisted execution intervals are required',
+    )
+    require(
+        max(x['started_at'] for x in intervals) < min(x['finished_at'] for x in intervals),
+        'The requests did not execute concurrently',
+    )
+    return intervals
+
+
+def merge_receipts(row, receipts):
+    row.setdefault('requests', []).extend(receipts)
+    for receipt in receipts:
+        row.setdefault('scenario_runs', []).extend(receipt.get('scenario_runs', []))
+
+
+def verify_wire(suite, row, receipts):
+    path = Path(suite.path(suite.profile['smoke']['mock_records']))
+    records = [json.loads(line) for line in path.read_text().splitlines()]
+    owned = {receipt['group'] for receipt in receipts}
+    from mock_llm import GROUP_TURN, group_ack
+
+    matched = []
+    for record in records:
+        messages = record.get('request', {}).get('messages', [])
+        starts = [i for i, message in enumerate(messages) if message.get('role') == 'user']
+        if not starts:
+            continue
+        markers = list(GROUP_TURN.finditer(str(messages[starts[-1]].get('content', ''))))
+        if not markers or markers[-1][1] not in owned:
+            continue
+        require(
+            record['status'] == 200 and record.get('response_complete'), 'Owned model request failed or did not finish'
+        )
+        expected = next((receipt for receipt in receipts if receipt['token'] == markers[-1][3]), None)
+        require(expected is not None, 'Model request has an unowned token')
+        require(record['request']['model'] == expected['model'], 'Conversation used an unexpected model')
+        acknowledgement = group_ack(messages, starts[-1])
+        require(
+            record.get('response', {}).get('content') == expected['expected'] + acknowledgement,
+            'Mock response does not identify the current request',
+        )
+        matched.append(markers[-1][3])
+    require(
+        sorted(matched) == sorted(receipt['token'] for receipt in receipts),
+        'Model requests were duplicated, lost, or assigned to a different conversation',
+    )
+    return {'requests': len(matched), 'history_isolated': True}
+
+
+def parallel_history(suite, row, bots):
+    lanes = [lane(suite, *bot) for bot in bots]
+    with ThreadPoolExecutor(max_workers=2) as pool:
+        for number in (1, 2):
+            barrier, receipts = threading.Barrier(2), [{}, {}]
+            futures = [
+                pool.submit(turn, suite, item, number, barrier, receipt) for item, receipt in zip(lanes, receipts)
+            ]
+            errors = []
+            try:
+                for future in futures:
+                    try:
+                        future.result(timeout=suite.timeout(70))
+                    except Exception as exc:
+                        errors.append(exc)
+            finally:
+                merge_receipts(row, receipts)
+            if errors:
+                row['request_errors'] = [str(error) for error in errors]
+                raise errors[0]
+            checked(row, f'overlap_turn_{number}', lambda: overlapping([receipt['interval'] for receipt in receipts]))
+    row['wire'] = checked(row, 'wire_identity_history_once', lambda: verify_wire(suite, row, row['requests']))
+
+
+def cancel_neighbor(suite, row, bot):
+    peer, receipt, future = lane(suite, *bot), {}, None
+    with ThreadPoolExecutor(max_workers=1) as pool:
+
+        def before_cancel():
+            nonlocal future
+            future = pool.submit(turn, suite, peer, 1, threading.Barrier(1), receipt)
+
+            def active():
+                with closing(ledger(suite)) as db:
+                    matches = db.execute(
+                        'SELECT run_id, started_at FROM agent_run WHERE workspace_id=? AND conversation_id=? AND id>? '
+                        'AND started_at IS NOT NULL AND finished_at IS NULL',
+                        (suite.variables['workspace_uuid'], 'group_' + peer['group'], row['after_id']),
+                    ).fetchall()
+                require(len(matches) <= 1, 'Neighbor submission created multiple active runs')
+                return dict(matches[0]) if matches else None
+
+            row['neighbor_at_cancel'] = checked(
+                row,
+                'neighbor_active_at_cancel',
+                lambda: execution._poll(
+                    suite,
+                    active,
+                    'Neighbor never became active before cancellation',
+                    seconds=10,
+                ),
+            )
+            require(
+                execution._sessions(suite).get(row['box_session_id'], {}).get('status') == 'running',
+                'Target tool stopped before the neighbor became active',
+            )
+
+        try:
+            execution.run_case(suite, {'scenario': 'execution.cancel_cleanup'}, row, before_cancel=before_cancel)
+        finally:
+            if future is not None:
+                try:
+                    future.result(timeout=suite.timeout(70))
+                finally:
+                    merge_receipts(row, [receipt])
+        require(future is not None, 'Cancellation never reached its active execution checkpoint')
+    cancelled = next(run for run in row['scenario_runs'] if run['expected_status'] == 'cancelled')
+    with closing(ledger(suite)) as db:
+        interval = dict(
+            db.execute(
+                'SELECT run_id, started_at, finished_at FROM agent_run WHERE run_id=?', (cancelled['run_id'],)
+            ).fetchone()
+        )
+    checked(row, 'cancellation_overlapped_neighbor', lambda: overlapping([interval, receipt['interval']]))
+    row['wire'] = checked(row, 'neighbor_identity_once', lambda: verify_wire(suite, row, [receipt]))
+
+
+def run_case(suite, case, row):
+    require(case['scenario'] in CASE_NAMES, 'Unknown isolation scenario')
+    secret = secrets.token_urlsafe(32)
+    suite.secrets.append(secret)
+    suite.env['LANGBOT_QA_GROUP_SECRET'] = secret
+    with ExitStack() as stack:
+        bots = []
+        pipelines = [('chat_pipeline', 'LANGBOT_SLOW_OK', 'mock-slow')]
+        if case['scenario'] == 'isolation.cross_pipeline':
+            pipelines.append(('isolation_pipeline', 'LANGBOT_SLOW_STREAM_OK', 'mock-slow-stream'))
+        for pipeline, expected, model in pipelines:
+            body = {
+                'name': 'qa-isolation-' + uuid4().hex,
+                'adapter': 'http_bot',
+                'enable': True,
+                'adapter_config': {
+                    'inbound_secret': secret,
+                    'signature_required': True,
+                    'default_session_type': 'group',
+                    'callback_url': 'http://127.0.0.1:9/unused-sync-only',
+                    'callback_max_retries': 0,
+                },
+                'event_bindings': [
+                    {
+                        'event_pattern': 'message.received',
+                        'target_type': 'pipeline',
+                        'target_uuid': suite.variables[pipeline],
+                        'enabled': True,
+                        'filters': [],
+                    }
+                ],
+                'plugin_processors': [],
+            }
+            _, bot = stack.enter_context(
+                owned_resource(
+                    suite,
+                    row,
+                    '/api/v1/platform/bots',
+                    body,
+                    item_key='bot',
+                    list_key='bots',
+                )
+            )
+            bots.append((bot, expected, model))
+        if case['scenario'] == 'isolation.cancel_neighbor':
+            checked(row, 'cancel_isolation', lambda: cancel_neighbor(suite, row, bots[0]))
+        else:
+            if case['scenario'] == 'isolation.same_pipeline':
+                bots = [bots[0], bots[0]]
+            checked(row, 'conversation_isolation', lambda: parallel_history(suite, row, bots))
+    return {'result': 'ISOLATION_OK', 'owned_runs': len(row.get('scenario_runs', []))}

tools/regression_local_cases.py

@@ -15,6 +15,7 @@ GROUPS = {
     'lifecycle': 'regression_lifecycle_cases',
     'plugins': 'regression_plugin_cases',
     'api': 'regression_api_cases',
+    'isolation': 'regression_isolation_cases',
 }
 
 

tools/test_regression_coverage.py

@@ -289,7 +289,7 @@ class Routes:
                 name = binding['profile']
                 if name not in profiles:
                     path = Path(__file__).with_name('regression_profiles') / f'{name}.json'
-                    if name in ('api', 'state'):
+                    if name in ('api', 'state', 'isolation'):
                         from regression_ci import suite_profile
 
                         profile, _ = suite_profile(name, 'catalog-test')

tools/test_regression_execution_cases.py

@@ -280,12 +280,21 @@ class ExecutionCasesTests(unittest.TestCase):
         try:
             self.assertEqual(cases._stream(self.suite, 'agent', 'scope', {})[-1]['kind'], 'completed')
             row = {}
+            checkpoints = []
             with (
                 patch.object(cases, '_running_box', return_value='box'),
                 patch.object(cases, '_sessions', return_value={'box': {'status': 'running'}}),
             ):
-                cancelled = cases._stream(self.suite, 'agent', 'scope', row, interrupt='cancel')
+                cancelled = cases._stream(
+                    self.suite,
+                    'agent',
+                    'scope',
+                    row,
+                    interrupt='cancel',
+                    before_cancel=lambda: checkpoints.append(row['box_session_id']),
+                )
             self.assertEqual(len(cancelled), 1)
+            self.assertEqual(checkpoints, ['box'])
             self.assertEqual(row['box_session_id'], 'box')
             frames.pop()
             with self.assertRaisesRegex(RuntimeError, 'outcome unknown'):

tools/test_regression_isolation.py

@@ -0,0 +1,86 @@
+"""Challenge isolation oracles with serialized, duplicated, and crossed evidence."""
+
+import copy
+import json
+from pathlib import Path
+import tempfile
+from types import SimpleNamespace
+import unittest
+
+from regression_ci import suite_profile
+from regression_isolation_cases import CASE_NAMES, check_reply, overlapping, verify_wire
+from regression_local_cases import DEFAULT_GROUPS
+from regression_scenarios import expanded_profile
+from regression_suite import validate
+
+
+class IsolationTests(unittest.TestCase):
+    def test_sync_stream_snapshots_cannot_hide_foreign_or_incomplete_output(self):
+        expected = 'LANGBOT_SLOW_OK GROUP_ACK=aaaa'
+
+        def parts(texts):
+            return [{'type': 'Plain', 'text': text} for text in texts]
+
+        check_reply(parts(['LANGBOT_', expected, expected]), expected)
+        for texts in ([], ['LANGBOT_'], [expected, 'LANGBOT_'], [expected, expected + ' GROUP_ACK=bbbb']):
+            with self.subTest(texts=texts), self.assertRaises(ValueError):
+                check_reply(parts(texts), expected)
+
+    def test_dedicated_finite_profile_does_not_expand_daily(self):
+        profile, _ = suite_profile('isolation', 'test')
+        validate(profile)
+        self.assertEqual([case['scenario'] for case in profile['cases']], list(CASE_NAMES))
+        self.assertNotIn('execution', profile)
+        self.assertNotIn('isolation', DEFAULT_GROUPS)
+        self.assertFalse(set(CASE_NAMES) & {case.get('scenario') for case in expanded_profile(DEFAULT_GROUPS)['cases']})
+        with self.assertRaisesRegex(ValueError, 'dedicated suite'):
+            expanded_profile(['isolation'])
+
+    def test_serial_execution_or_incomplete_timing_cannot_pass_overlap(self):
+        first = {'started_at': '2026-10-29 01:00:00', 'finished_at': '2026-10-29 01:00:02'}
+        second = {'started_at': '2026-10-29 01:00:01', 'finished_at': '2026-10-29 01:00:03'}
+        self.assertEqual(overlapping([first, second]), [first, second])
+        for invalid in ({**second, 'started_at': first['finished_at']}, {**second, 'finished_at': None}):
+            with self.subTest(invalid=invalid), self.assertRaises(ValueError):
+                overlapping([first, invalid])
+
+    def test_wire_rejects_crossed_history_duplicates_wrong_model_and_wrong_reply(self):
+        receipt = {'group': 'test-a', 'token': 'bbbb', 'model': 'mock-slow', 'expected': 'LANGBOT_SLOW_OK'}
+        record = {
+            'status': 200,
+            'response_complete': True,
+            'request': {
+                'model': 'mock-slow',
+                'messages': [
+                    {'role': 'user', 'content': 'LB_GROUP=test-a LB_TURN=1 LB_TOKEN=aaaa LB_PREVIOUS=none'},
+                    {'role': 'assistant', 'content': 'LANGBOT_SLOW_OK GROUP_ACK=aaaa'},
+                    {'role': 'user', 'content': 'LB_GROUP=test-a LB_TURN=2 LB_TOKEN=bbbb LB_PREVIOUS=aaaa'},
+                ],
+            },
+            'response': {'content': 'LANGBOT_SLOW_OK GROUP_ACK=bbbb'},
+        }
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / 'wire.jsonl'
+            suite = SimpleNamespace(profile={'smoke': {'mock_records': str(path)}}, path=lambda value: value)
+
+            def check(rows):
+                path.write_text(''.join(json.dumps(row) + '\n' for row in rows))
+                return verify_wire(suite, {}, [receipt])
+
+            self.assertEqual(check([record])['requests'], 1)
+            crossed = copy.deepcopy(record)
+            crossed['request']['messages'][0]['content'] = crossed['request']['messages'][0]['content'].replace(
+                'test-a', 'test-b'
+            )
+            wrong_model = copy.deepcopy(record)
+            wrong_model['request']['model'] = 'mock-chat'
+            wrong_reply = copy.deepcopy(record)
+            wrong_reply['response']['content'] = 'LANGBOT_SLOW_OK GROUP_ACK=aaaa'
+            incomplete = {**record, 'response_complete': False}
+            for rows in ([], [record, record], [crossed], [wrong_model], [wrong_reply], [incomplete]):
+                with self.subTest(rows=rows), self.assertRaises(ValueError):
+                    check(rows)
+
+
+if __name__ == '__main__':
+    unittest.main()