test(tools): add verified sustained runs and regression analysis

tools/README.md

@@ -401,3 +401,74 @@ as required status checks in the repository ruleset. A green workflow alone does
 not enforce merge protection. Longer load and soak tests use the existing Suite
 and Job entry points with separately calibrated latency, throughput and resource
 budgets; the one-minute smoke does not replace them.
+
+## Sustained regression and baseline comparison
+
+[`regression_profiles/endurance.json`](regression_profiles/endurance.json) runs
+nine minutes of mixed chat/Box/RAG traffic at 1 request/second across four stable
+groups, followed by one minute without traffic to check recovery. Configure its
+CLI context, plugin inputs, model endpoints, Core SQLite path and Core/Runtime/Box
+PID files as for smoke. Set `plugins: []` to reuse initialized plugins. Configure
+`observe.box_data_root` to the target deployment's Box mount root; the caller must
+have Docker inspection access. Do not start another Box runtime on a shared daemon.
+
+```sh
+uv run --no-sync python tools/regression_suite.py validate \
+  --profile /path/to/endurance.json
+uv run --no-sync python tools/regression_suite.py run \
+  --profile /path/to/endurance.json --output .qa/endurance-baseline
+# For an hour including the one-minute cooldown, use --duration 3540.
+```
+
+This profile checks more than reply text:
+
+- Each Runner request is reconciled to one completed AgentRun with a persisted
+  completion event; Box additionally verifies two real Docker exec results.
+  After drain and cooldown, the complete set of runs for the phase's conversations
+  must match the verified set. Delayed extra, missing and nonterminal runs fail.
+- At least 90% of target successful throughput must be delivered, with per-case
+  P95 limits. Reports retain planned/actual counts, scheduling delay and request
+  drain time. The load is bounded by workers and serial group sessions; it is
+  not an open arrival-rate capacity benchmark.
+- Process RSS, FD, thread and process-count medians after cooldown are compared
+  with a stable window after the first minute. Docker memory, PIDs and container
+  counts are checked separately because sandbox processes are not children of
+  the Box service. If a discovered container disappears before stats collection,
+  the sampler refreshes the set and retries once, retaining the first error and
+  membership change. Unchanged-set errors, failed recovery, missing samples and
+  budget excesses fail.
+
+The supplied limits are initial QA budgets, not product SLOs or proof of a leak.
+Fixed groups prevent deliberate session growth from being mistaken for leakage;
+Core and plugin caches can still retain memory legitimately. Inspect failures
+before changing a budget. Process-tree CPU remains a raw diagnostic: child exits
+can reduce the cumulative sum, so it is not used as a CPU regression verdict.
+
+`requests.jsonl` separates dispatch/response latency from ledger setup and result
+verification. Resource samples go to `resources.jsonl`; `summary.json` contains
+throughput, latency, recovery and ledger closeout results. Failed runs retain
+resources and evidence. The normal Suite cleanup command removes only owned API
+resources after inspection.
+
+To compare two successful runs from the same reserved test host:
+
+```sh
+uv run --no-sync python tools/regression_analysis.py \
+  --baseline .qa/endurance-baseline --candidate .qa/endurance-candidate \
+  --output .qa/comparison.json
+```
+
+The comparison rejects failed runs, missing dependency evidence, different enabled
+plugin versions, fixtures, expanded case/resource inputs, phase geometry, budgets,
+or observer-host identity. Target URL origins and credentials may differ; URL
+paths, model names and actual stimulus variables remain part of the contract.
+Both runs must use the same hardware allocation, model backend and background
+load; the contract does not establish those external conditions automatically.
+The default comparison allows at most 20% P95 growth and 10% successful-throughput
+loss, requires 30 requests per case in both runs, and rejects invalid metrics or
+incomplete ledger closeout. Set tolerances from repeated controls on your host.
+
+Keep this longer test outside the one-minute merge gate while its budgets and
+repeatability are being evaluated. The same profile can be selected by the
+release-deployment Job, which supplies its ledger path and process IDs; that
+fresh-deployment path still needs separate end-to-end validation.

tools/regression_analysis.py

@@ -0,0 +1,289 @@
+#!/usr/bin/env python3
+"""Compare compatible load runs and check resource recovery without treating missing data as zero."""
+
+import argparse
+import hashlib
+import json
+import math
+import os
+import platform
+from pathlib import Path
+import statistics
+import re
+from urllib.parse import urlsplit
+
+
+METRICS = {'rss_bytes', 'open_fds', 'threads', 'processes'}
+
+
+def validate_resource_limits(limits):
+    if (
+        not isinstance(limits, dict)
+        or not {'warmup_seconds', 'window_seconds', 'min_samples', 'targets'} <= set(limits)
+        or limits.keys() - {'warmup_seconds', 'window_seconds', 'min_samples', 'targets', 'containers'}
+    ):
+        raise ValueError('resource_limits requires warmup_seconds, window_seconds, min_samples and targets')
+    for key in ('warmup_seconds', 'window_seconds', 'min_samples'):
+        value = limits[key]
+        if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value) or value <= 0:
+            raise ValueError(f'resource_limits.{key} must be positive and finite')
+    if not isinstance(limits['min_samples'], int) or limits['min_samples'] < 3:
+        raise ValueError('resource_limits.min_samples must be an integer >= 3')
+    if not isinstance(limits['targets'], dict) or not limits['targets']:
+        raise ValueError('resource_limits.targets must contain process targets')
+    for name, metrics in limits['targets'].items():
+        if not name or not isinstance(metrics, dict) or not metrics or metrics.keys() - METRICS:
+            raise ValueError('resource limits support rss_bytes, open_fds, threads and processes')
+        if any(
+            isinstance(v, bool) or not isinstance(v, (float, int)) or not math.isfinite(v) or v < 0
+            for v in metrics.values()
+        ):
+            raise ValueError('resource growth allowances must be finite and nonnegative')
+
+    if 'containers' in limits:
+        values = limits['containers']
+        if not isinstance(values, dict) or not values or values.keys() - {'memory_bytes', 'pids', 'count'}:
+            raise ValueError('container limits support memory_bytes, pids and count')
+        if any(
+            isinstance(v, bool) or not isinstance(v, (int, float)) or not math.isfinite(v) or v < 0
+            for v in values.values()
+        ):
+            raise ValueError('container growth allowances must be finite and nonnegative')
+
+
+def container_metrics(row):
+    entries = row.get('containers')
+    if not isinstance(entries, list) or not entries or row.get('container_error'):
+        return {}
+    memory = pids = 0
+    units = {'B': 1, 'kB': 1000, 'KiB': 1024, 'MB': 1000**2, 'MiB': 1024**2, 'GB': 1000**3, 'GiB': 1024**3}
+    for item in entries:
+        match = re.fullmatch(r'([0-9.]+)\s*([A-Za-z]+)', item.get('MemUsage', '').split('/')[0].strip())
+        try:
+            if match is None:
+                return {}
+            memory += float(match[1]) * units[match[2]]
+            pids += int(item['PIDs'])
+        except (ValueError, KeyError, TypeError):
+            return {}
+    return {'memory_bytes': memory, 'pids': pids, 'count': len(entries)}
+
+
+def resource_verdict(directory, limits):
+    validate_resource_limits(limits)
+    rows = [json.loads(line) for line in (Path(directory) / 'resources.jsonl').read_text().splitlines()]
+    result = {'errors': [], 'targets': {}, 'limits': limits}
+    cooldown = [row for row in rows if row['phase'] == 'cooldown']
+    if not cooldown:
+        result['errors'].append('No cooldown resource samples; recovery is unknown')
+        return result
+    end = max(row['elapsed'] for row in cooldown)
+    baseline = [
+        row
+        for row in rows
+        if row['phase'] == 'load'
+        and limits['warmup_seconds'] <= row['elapsed'] < limits['warmup_seconds'] + limits['window_seconds']
+    ]
+    tail = [row for row in cooldown if row['elapsed'] >= end - limits['window_seconds']]
+    if min(len(baseline), len(tail)) < limits['min_samples']:
+        result['errors'].append('Insufficient stable/cooldown samples; resource recovery is unknown')
+        return result
+    targets = dict(limits['targets'])
+    if limits.get('containers'):
+        targets['docker_containers'] = limits['containers']
+    for name, metrics in targets.items():
+        target = result['targets'][name] = {}
+        for metric, allowance in metrics.items():
+
+            def value(row):
+                source = container_metrics(row) if name == 'docker_containers' else row['targets'].get(name, {})
+                return source.get(metric)
+
+            before = [value(row) for row in baseline]
+            after = [value(row) for row in tail]
+            if any(
+                not isinstance(v, (int, float))
+                or not math.isfinite(v)
+                or v < 0
+                or (metric in {'rss_bytes', 'memory_bytes'} and v == 0)
+                for v in before + after
+            ):
+                result['errors'].append(f'{name}/{metric}: missing or invalid samples')
+                continue
+            initial, final = statistics.median(before), statistics.median(after)
+            growth = final - initial
+            target[metric] = {
+                'baseline_median': initial,
+                'cooldown_median': final,
+                'growth': growth,
+                'allowance': allowance,
+            }
+            if growth > allowance:
+                result['errors'].append(f'{name}/{metric}: retained growth {growth:g} exceeds {allowance:g}')
+    return result
+
+
+def workload_contract(profile, profile_path):
+    """Fingerprint stimuli and load geometry, excluding target addresses and generated UUIDs."""
+    from regression_suite import resolve
+
+    profile_path = Path(profile_path).resolve()
+    variables = {
+        **profile.get('vars', {}),
+        'profile_dir': str(profile_path.parent),
+        **{item['id']: '${' + item['id'] + '}' for item in profile.get('resources', []) if 'id' in item},
+        'workspace_uuid': '${workspace_uuid}',
+        'group_secret': '${group_secret}',
+        'run_dir': '${run_dir}',
+    }
+    resources = resolve(profile.get('resources', []), variables)
+    cases = resolve(profile.get('cases', []), variables)
+    for resource in resources:
+        for document in resource.get('documents', []):
+            path = Path(document['file'])
+            path = path if path.is_absolute() else profile_path.parent / path
+            document['file'] = {'sha256': hashlib.sha256(path.read_bytes()).hexdigest()}
+        if resource.get('kind') == 'provider':
+            body = resource['body']
+            if body.get('base_url'):
+                url = urlsplit(body['base_url'])
+                body['base_url'] = {'scheme': url.scheme, 'path': url.path, 'query': url.query}
+            if 'api_keys' in body:
+                body['api_keys'] = '<credentials>'
+    for case in cases:
+        if 'http_group' in case:
+            url = urlsplit(case['http_group']['url'])
+            case['http_group']['url'] = {'scheme': url.scheme, 'path': url.path, 'query': url.query}
+    contract = {key: profile.get(key) for key in ('phases', 'thresholds', 'resource_limits')}
+    contract.update(resources=resources, cases=cases)
+    contract['observer_host'] = {'name': platform.node(), 'machine': platform.machine(), 'cpus': os.cpu_count()}
+    contract['oracle_sha256'] = {
+        name: hashlib.sha256(Path(__file__).with_name(name).read_bytes()).hexdigest()
+        for name in ('regression_traffic.py', 'regression_smoke.py', 'regression_analysis.py')
+    }
+    contract['timing_contract'] = 'dispatch-after-ledger-snapshot-v1'
+    contract['verified_runs'] = bool(profile.get('verification'))
+    contract['mock_scenarios_sha256'] = hashlib.sha256(
+        Path(__file__).with_name('mock_llm_scenarios.json').read_bytes()
+    ).hexdigest()
+    return contract
+
+
+def compare(baseline, candidate, *, max_p95_ratio=1.2, min_rps_ratio=0.9, min_requests=30):
+    roots = [Path(baseline), Path(candidate)]
+    summaries = [json.loads((root / 'summary.json').read_text()) for root in roots]
+    contracts = [json.loads((root / 'workload.json').read_text()) for root in roots]
+    errors, comparisons = [], []
+    plugins = []
+    for root in roots:
+        environment = root / 'environment.json'
+        if not environment.exists():
+            errors.append('Plugin dependency evidence is missing')
+            continue
+        records = json.loads(environment.read_text())['plugins']['plugins']
+        plugins.append(
+            sorted(
+                (item['author'], item['name'], item['version'])
+                for item in records
+                if item.get('enabled') and item.get('status') == 'initialized'
+            )
+        )
+    if any(not value for value in plugins):
+        errors.append('Enabled plugin evidence is empty')
+    if len(plugins) == 2 and plugins[0] != plugins[1]:
+        errors.append('Enabled plugin versions differ; runs are not comparable')
+    if contracts[0] != contracts[1]:
+        errors.append('Workload/fixtures/oracles differ; runs are not comparable')
+    if not all(summary.get('passed') is True for summary in summaries):
+        errors.append('Both runs must pass functional, delivery and resource checks')
+    if not all(contract.get('verified_runs') for contract in contracts):
+        errors.append('Both runs require ledger verification')
+    if errors:
+        return {'passed': False, 'errors': errors, 'comparisons': comparisons}
+    old_phases, new_phases = summaries[0]['phases'], summaries[1]['phases']
+    if not old_phases or not new_phases:
+        return {'passed': False, 'errors': ['Measured phases are missing'], 'comparisons': []}
+    if [phase['name'] for phase in old_phases] != [phase['name'] for phase in new_phases]:
+        return {'passed': False, 'errors': ['Phase sets differ'], 'comparisons': []}
+    for old, new in zip(old_phases, new_phases):
+        if any(
+            phase.get('verification_enabled') is not True
+            or phase.get('verified_runs') != phase.get('runner_requests')
+            or phase.get('ledger_closeout', {}).get('matched') is not True
+            for phase in (old, new)
+        ):
+            errors.append(f'{new["name"]}: runner reconciliation is incomplete')
+            continue
+        if any(
+            not isinstance(phase.get('achieved_rps'), (int, float))
+            or not math.isfinite(phase['achieved_rps'])
+            or phase['achieved_rps'] <= 0
+            for phase in (old, new)
+        ):
+            errors.append(f'{new["name"]}: successful throughput is invalid')
+            continue
+        if not old['cases'] or not new['cases'] or old['cases'].keys() != new['cases'].keys():
+            errors.append(f'{new["name"]}: case sets differ')
+            continue
+        if new['achieved_rps'] < old['achieved_rps'] * min_rps_ratio:
+            errors.append(f'{new["name"]}: successful throughput regressed')
+        for name, current in new['cases'].items():
+            previous = old['cases'][name]
+            values = [previous.get('p95_seconds'), current.get('p95_seconds')]
+            if min(previous['requests'], current['requests']) < min_requests or any(
+                not isinstance(v, (int, float)) or not math.isfinite(v) or v <= 0 for v in values
+            ):
+                errors.append(f'{new["name"]}/{name}: insufficient samples or invalid P95')
+                continue
+            ratio = current['p95_seconds'] / previous['p95_seconds']
+            comparisons.append(
+                {
+                    'phase': new['name'],
+                    'case': name,
+                    'p95_ratio': ratio,
+                    'baseline_requests': previous['requests'],
+                    'candidate_requests': current['requests'],
+                }
+            )
+            if ratio > max_p95_ratio:
+                errors.append(f'{new["name"]}/{name}: P95 regressed by {(ratio - 1) * 100:.1f}%')
+    return {
+        'passed': not errors,
+        'errors': errors,
+        'comparisons': comparisons,
+        'max_p95_ratio': max_p95_ratio,
+        'min_rps_ratio': min_rps_ratio,
+        'min_requests': min_requests,
+    }
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--baseline', type=Path, required=True)
+    parser.add_argument('--candidate', type=Path, required=True)
+    parser.add_argument('--output', type=Path, required=True)
+    parser.add_argument('--max-p95-ratio', type=float, default=1.2)
+    parser.add_argument('--min-rps-ratio', type=float, default=0.9)
+    parser.add_argument('--min-requests', type=int, default=30)
+    args = parser.parse_args()
+    if (
+        not math.isfinite(args.max_p95_ratio)
+        or args.max_p95_ratio < 1
+        or not 0 < args.min_rps_ratio <= 1
+        or args.min_requests < 1
+    ):
+        parser.error('require finite max-p95-ratio >= 1, min-rps-ratio in (0,1], min-requests >= 1')
+    result = compare(
+        args.baseline,
+        args.candidate,
+        max_p95_ratio=args.max_p95_ratio,
+        min_rps_ratio=args.min_rps_ratio,
+        min_requests=args.min_requests,
+    )
+    args.output.write_text(json.dumps(result, indent=2) + '\n')
+    print(json.dumps(result, indent=2))
+    raise SystemExit(0 if result['passed'] else 1)
+
+
+if __name__ == '__main__':
+    main()

tools/regression_job.py

@@ -308,16 +308,27 @@ class Job:
             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')
+        if 'verification' in profile:
+            profile['verification']['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())}
         for service, container in containers.items():
             pid = int(self.checked(['docker', 'inspect', '--format', '{{.State.Pid}}', container]).strip())
-            profile['observe']['pids'][service] = pid
+            name = (
+                {'langbot': 'core', 'langbot_plugin_runtime': 'runtime', 'langbot_box': 'box'}.get(service, service)
+                if profile.get('verification')
+                else service
+            )
+            profile['observe']['pids'][name] = pid
         destination = self.output / 'suite'
         destination.mkdir(mode=0o700)
         suite = Suite(profile, self.suite_path, destination, env=self.env)
         self.redactions.extend(suite.secrets)
         suite.write('profile.json', profile)
+        if 'smoke' not in profile:
+            from regression_analysis import workload_contract
+
+            suite.write('workload.json', workload_contract(profile, self.suite_path))
         suite.prepare()
         if 'smoke' in profile:
             return suite

tools/regression_profiles/endurance.json

@@ -0,0 +1,513 @@
+{
+  "version": 1,
+  "cli": {
+    "executable": "lbctl",
+    "config": "./lbctl.yaml"
+  },
+  "vars": {
+    "plugin_dir": "/path/to/plugin-zips",
+    "mock_base_url": "http://127.0.0.1:18080/v1",
+    "embedding_base_url": "http://127.0.0.1:15410/v1",
+    "core_health": "http://127.0.0.1:5300/healthz",
+    "core_base_url": "http://127.0.0.1:5300",
+    "core_db": "/path/to/core/data/langbot.db",
+    "core_pid_file": "/path/to/core.pid",
+    "runtime_pid_file": "/path/to/runtime.pid",
+    "box_pid_file": "/path/to/box.pid",
+    "box_data_root": "/path/to/core/box"
+  },
+  "plugins": [
+    {
+      "source": "local",
+      "file": "${plugin_dir}/LocalAgent.zip"
+    },
+    {
+      "source": "local",
+      "file": "${plugin_dir}/LangRAG.zip"
+    },
+    {
+      "source": "local",
+      "file": "${plugin_dir}/GeneralParsers.zip"
+    }
+  ],
+  "resources": [
+    {
+      "id": "mock_provider",
+      "kind": "provider",
+      "body": {
+        "name": "regression mock",
+        "requester": "openai-chat-completions",
+        "base_url": "${mock_base_url}",
+        "api_keys": [
+          "test-only"
+        ]
+      }
+    },
+    {
+      "id": "embedding_provider",
+      "kind": "provider",
+      "body": {
+        "name": "regression embedding",
+        "requester": "openai-chat-completions",
+        "base_url": "${embedding_base_url}",
+        "api_keys": [
+          "test-only"
+        ]
+      }
+    },
+    {
+      "id": "chat_model",
+      "kind": "model",
+      "type": "llm",
+      "body": {
+        "name": "mock-chat",
+        "provider_uuid": "${mock_provider}",
+        "abilities": [
+          "func_call"
+        ],
+        "context_length": 8192,
+        "extra_args": {
+          "request_policy": {
+            "retries": 0
+          }
+        }
+      }
+    },
+    {
+      "id": "box_model",
+      "kind": "model",
+      "type": "llm",
+      "body": {
+        "name": "mock-smoke-box",
+        "provider_uuid": "${mock_provider}",
+        "abilities": [
+          "func_call"
+        ],
+        "context_length": 8192,
+        "extra_args": {
+          "request_policy": {
+            "retries": 0
+          }
+        }
+      }
+    },
+    {
+      "id": "rag_model",
+      "kind": "model",
+      "type": "llm",
+      "body": {
+        "name": "mock-rag",
+        "provider_uuid": "${mock_provider}",
+        "abilities": [
+          "func_call"
+        ],
+        "context_length": 8192,
+        "extra_args": {
+          "request_policy": {
+            "retries": 0
+          }
+        }
+      }
+    },
+    {
+      "id": "embedding_model",
+      "kind": "model",
+      "type": "embedding",
+      "body": {
+        "name": "all-MiniLM-L6-v2",
+        "provider_uuid": "${embedding_provider}"
+      }
+    },
+    {
+      "id": "knowledge_base",
+      "kind": "knowledge-base",
+      "body": {
+        "name": "regression parser golden",
+        "knowledge_engine_plugin_id": "langbot-team/LangRAG",
+        "creation_settings": {
+          "embedding_model_uuid": "${embedding_model}",
+          "index_type": "chunk",
+          "chunk_size": 512,
+          "overlap": 50
+        },
+        "retrieval_settings": {
+          "search_type": "vector",
+          "top_k": 5,
+          "query_rewrite": "off",
+          "rerank": "off"
+        }
+      },
+      "documents": [
+        {
+          "file": "../../skills/skills/langbot-testing/fixtures/rag/parser-golden.html",
+          "parser_plugin_id": "langbot-team/GeneralParsers"
+        }
+      ]
+    },
+    {
+      "id": "chat_pipeline",
+      "kind": "pipeline",
+      "body": {
+        "name": "regression chat",
+        "description": "Normal deterministic chat",
+        "config": {
+          "ai": {
+            "runner": {
+              "id": "plugin:langbot-team/LocalAgent/default",
+              "expire-time": 0
+            },
+            "runner_config": {
+              "plugin:langbot-team/LocalAgent/default": {
+                "model": {
+                  "primary": "${chat_model}",
+                  "fallbacks": []
+                },
+                "prompt": [
+                  {
+                    "role": "system",
+                    "content": "You are a QA regression assistant."
+                  }
+                ],
+                "knowledge-bases": [],
+                "box-enabled": false,
+                "timeout": 30,
+                "max-round": 20,
+                "box-session-id-template": "{launcher_type}_{launcher_id}"
+              }
+            }
+          },
+          "safety": {
+            "rate-limit": {
+              "limitation": 10000
+            }
+          },
+          "output": {
+            "force-delay": {
+              "min": 0,
+              "max": 0
+            },
+            "misc": {
+              "at-sender": false,
+              "quote-origin": false
+            }
+          },
+          "trigger": {
+            "group-respond-rules": {
+              "prefix": [
+                "ai "
+              ]
+            }
+          }
+        }
+      }
+    },
+    {
+      "id": "box_pipeline",
+      "kind": "pipeline",
+      "body": {
+        "name": "regression Box",
+        "description": "Real Box write and read",
+        "config": {
+          "ai": {
+            "runner": {
+              "id": "plugin:langbot-team/LocalAgent/default",
+              "expire-time": 0
+            },
+            "runner_config": {
+              "plugin:langbot-team/LocalAgent/default": {
+                "model": {
+                  "primary": "${box_model}",
+                  "fallbacks": []
+                },
+                "prompt": [
+                  {
+                    "role": "system",
+                    "content": "You are a QA regression assistant."
+                  }
+                ],
+                "knowledge-bases": [],
+                "box-enabled": true,
+                "box-session-id-template": "{launcher_type}_{launcher_id}",
+                "timeout": 30,
+                "max-tool-iterations": 5,
+                "max-round": 20
+              }
+            }
+          },
+          "safety": {
+            "rate-limit": {
+              "limitation": 10000
+            }
+          },
+          "output": {
+            "force-delay": {
+              "min": 0,
+              "max": 0
+            },
+            "misc": {
+              "at-sender": false,
+              "quote-origin": false
+            }
+          },
+          "trigger": {
+            "group-respond-rules": {
+              "prefix": [
+                "ai "
+              ]
+            }
+          }
+        }
+      }
+    },
+    {
+      "id": "rag_pipeline",
+      "kind": "pipeline",
+      "body": {
+        "name": "regression RAG",
+        "description": "Retrieved model input oracle",
+        "config": {
+          "ai": {
+            "runner": {
+              "id": "plugin:langbot-team/LocalAgent/default",
+              "expire-time": 0
+            },
+            "runner_config": {
+              "plugin:langbot-team/LocalAgent/default": {
+                "model": {
+                  "primary": "${rag_model}",
+                  "fallbacks": []
+                },
+                "prompt": [
+                  {
+                    "role": "system",
+                    "content": "You are a QA regression assistant."
+                  }
+                ],
+                "knowledge-bases": [
+                  "${knowledge_base}"
+                ],
+                "box-enabled": false,
+                "timeout": 30,
+                "retrieval-top-k": 5,
+                "max-round": 20,
+                "box-session-id-template": "{launcher_type}_{launcher_id}"
+              }
+            }
+          },
+          "safety": {
+            "rate-limit": {
+              "limitation": 10000
+            }
+          },
+          "output": {
+            "force-delay": {
+              "min": 0,
+              "max": 0
+            },
+            "misc": {
+              "at-sender": false,
+              "quote-origin": false
+            }
+          },
+          "trigger": {
+            "group-respond-rules": {
+              "prefix": [
+                "ai "
+              ]
+            }
+          }
+        }
+      }
+    },
+    {
+      "id": "chat_bot",
+      "kind": "bot",
+      "body": {
+        "name": "regression group chat",
+        "description": "Synthetic signed group ingress",
+        "adapter": "http_bot",
+        "enable": true,
+        "adapter_config": {
+          "inbound_secret": "${group_secret}",
+          "signature_required": true,
+          "callback_url": "http://127.0.0.1:9/unused-sync-only",
+          "default_session_type": "group",
+          "callback_timeout": 20,
+          "callback_max_retries": 0
+        },
+        "event_bindings": [
+          {
+            "event_pattern": "message.received",
+            "target_type": "pipeline",
+            "target_uuid": "${chat_pipeline}",
+            "enabled": true,
+            "filters": []
+          }
+        ],
+        "plugin_processors": []
+      }
+    },
+    {
+      "id": "box_bot",
+      "kind": "bot",
+      "body": {
+        "name": "regression group box",
+        "description": "Synthetic signed group ingress",
+        "adapter": "http_bot",
+        "enable": true,
+        "adapter_config": {
+          "inbound_secret": "${group_secret}",
+          "signature_required": true,
+          "callback_url": "http://127.0.0.1:9/unused-sync-only",
+          "default_session_type": "group",
+          "callback_timeout": 20,
+          "callback_max_retries": 0
+        },
+        "event_bindings": [
+          {
+            "event_pattern": "message.received",
+            "target_type": "pipeline",
+            "target_uuid": "${box_pipeline}",
+            "enabled": true,
+            "filters": []
+          }
+        ],
+        "plugin_processors": []
+      }
+    },
+    {
+      "id": "rag_bot",
+      "kind": "bot",
+      "body": {
+        "name": "regression group rag",
+        "description": "Synthetic signed group ingress",
+        "adapter": "http_bot",
+        "enable": true,
+        "adapter_config": {
+          "inbound_secret": "${group_secret}",
+          "signature_required": true,
+          "callback_url": "http://127.0.0.1:9/unused-sync-only",
+          "default_session_type": "group",
+          "callback_timeout": 20,
+          "callback_max_retries": 0
+        },
+        "event_bindings": [
+          {
+            "event_pattern": "message.received",
+            "target_type": "pipeline",
+            "target_uuid": "${rag_pipeline}",
+            "enabled": true,
+            "filters": []
+          }
+        ],
+        "plugin_processors": []
+      }
+    }
+  ],
+  "cases": [
+    {
+      "name": "chat",
+      "weight": 2,
+      "http_group": {
+        "url": "${core_base_url}/bots/${chat_bot}/sync",
+        "secret_env": "LANGBOT_QA_GROUP_SECRET",
+        "message": "ai What is the documented parser-rag sentinel?"
+      },
+      "expect": "LANGBOT_CHAT_OK"
+    },
+    {
+      "name": "box",
+      "weight": 1,
+      "http_group": {
+        "url": "${core_base_url}/bots/${box_bot}/sync",
+        "secret_env": "LANGBOT_QA_GROUP_SECRET",
+        "message": "ai What is the documented parser-rag sentinel?"
+      },
+      "expect": "LANGBOT_BOX_OK",
+      "box_exec": {
+        "count": 2,
+        "backend": "docker",
+        "stdout": "LANGBOT_BOX_SENTINEL"
+      }
+    },
+    {
+      "name": "rag_chat",
+      "weight": 1,
+      "http_group": {
+        "url": "${core_base_url}/bots/${rag_bot}/sync",
+        "secret_env": "LANGBOT_QA_GROUP_SECRET",
+        "message": "ai What is the documented parser-rag sentinel?"
+      },
+      "expect": "LANGBOT_RAG_OK"
+    }
+  ],
+  "phases": [
+    {
+      "name": "sustained",
+      "duration_seconds": 540,
+      "rps": 1,
+      "workers": 4,
+      "cooldown_seconds": 60,
+      "groups": 4,
+      "members_per_group": 3
+    }
+  ],
+  "observe": {
+    "health": {
+      "core": "${core_health}",
+      "mock": "http://127.0.0.1:18080/healthz",
+      "box": "http://127.0.0.1:5410/readyz"
+    },
+    "pids": {},
+    "pid_files": {
+      "core": "${core_pid_file}",
+      "runtime": "${runtime_pid_file}",
+      "box": "${box_pid_file}"
+    },
+    "containers": [],
+    "box_data_root": "${box_data_root}"
+  },
+  "cleanup": "on-success",
+  "thresholds": {
+    "min_rps_ratio": 0.9,
+    "max_p95_seconds": {
+      "chat": 5,
+      "box": 10,
+      "rag_chat": 6
+    }
+  },
+  "verification": {
+    "ledger_db": "${core_db}",
+    "timeout_seconds": 5
+  },
+  "resource_limits": {
+    "warmup_seconds": 60,
+    "window_seconds": 30,
+    "min_samples": 5,
+    "targets": {
+      "core": {
+        "rss_bytes": 67108864,
+        "open_fds": 8,
+        "threads": 8,
+        "processes": 0
+      },
+      "runtime": {
+        "rss_bytes": 67108864,
+        "open_fds": 8,
+        "threads": 8,
+        "processes": 0
+      },
+      "box": {
+        "rss_bytes": 33554432,
+        "open_fds": 8,
+        "threads": 8,
+        "processes": 0
+      }
+    },
+    "containers": {
+      "memory_bytes": 67108864,
+      "pids": 8,
+      "count": 0
+    }
+  }
+}

tools/regression_suite.py

@@ -78,6 +78,8 @@ def validate(profile):
         'cleanup',
         'thresholds',
         'smoke',
+        'verification',
+        'resource_limits',
     }
     if extra := profile.keys() - allowed:
         raise ValueError(f'unknown profile fields: {sorted(extra)}')
@@ -138,6 +140,18 @@ def validate(profile):
             if item['kind'] != 'knowledge-base' or not document.get('file') or not document.get('parser_plugin_id'):
                 raise ValueError(f'{name}: documents need file and parser_plugin_id on a knowledge-base')
             resolve(document, variables)
+    verification = profile.get('verification')
+    if verification is not None:
+        if (
+            'smoke' in profile
+            or not isinstance(verification, dict)
+            or verification.keys() - {'ledger_db', 'timeout_seconds'}
+        ):
+            raise ValueError('verification is a load-only ledger configuration')
+        if not isinstance(verification.get('ledger_db'), str) or not verification['ledger_db']:
+            raise ValueError('verification.ledger_db is required')
+        resolve(verification, variables)
+        number(verification.get('timeout_seconds', 5), 'verification timeout', minimum=0.001)
     cases = profile.get('cases')
     if not isinstance(cases, list) or not cases:
         raise ValueError('cases must be a non-empty array')
@@ -164,7 +178,7 @@ def validate(profile):
         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:
+        if any(key in case for key in ('mock_requests', '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')
@@ -237,14 +251,14 @@ def validate(profile):
             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')
-            if 'smoke' not in profile:
-                raise ValueError('box_exec checks require smoke mode')
+            if 'smoke' not in profile and verification is None:
+                raise ValueError('box_exec requires smoke or load verification')
             if 'http_group' not in case and case.get('args', [])[:2] != ['pipeline', 'run']:
                 raise ValueError('box_exec requires a Pipeline or group case')
         if 'attachment' in case:
             check = case['attachment']
             if (
-                'smoke' not in profile
+                ('smoke' not in profile and verification is None)
                 or case.get('args', [])[:2] != ['pipeline', 'run']
                 or not isinstance(check, dict)
                 or set(check) != {'name', 'text'}
@@ -329,6 +343,12 @@ def validate(profile):
         if name not in {case['name'] for case in cases}:
             raise ValueError(f'unknown case in max_p95_seconds: {name}')
         number(value, 'max_p95_seconds', minimum=0.001)
+    if 'resource_limits' in profile:
+        if 'smoke' in profile:
+            raise ValueError('resource_limits requires measured load and cooldown phases')
+        from regression_analysis import validate_resource_limits
+
+        validate_resource_limits(profile['resource_limits'])
     observed = resolve(profile.get('observe', {}), variables)
     if not isinstance(observed, dict) or observed.keys() - {
         'health',
@@ -352,6 +372,21 @@ def validate(profile):
             raise ValueError(f'observe.{field} must be an object')
     for pid in observed.get('pids', {}).values():
         number(pid, 'PID', minimum=1, integer=True)
+    if profile.get('resource_limits'):
+        if profile['resource_limits'].get('containers') and not (
+            observed.get('containers') or observed.get('box_data_root')
+        ):
+            raise ValueError('Container budgets require container names or a Box data root')
+        configured = set(observed.get('pids', {})) | set(observed.get('pid_files', {}))
+        if set(profile['resource_limits']['targets']) - configured:
+            raise ValueError('Resource limit targets must have configured PIDs or PID files')
+        for phase in phases:
+            limits = profile['resource_limits']
+            if (
+                phase['duration_seconds'] < limits['warmup_seconds'] + limits['window_seconds']
+                or phase.get('cooldown_seconds', 0) < limits['window_seconds']
+            ):
+                raise ValueError('Phase must cover the configured warmup, resource window and cooldown')
     if not isinstance(observed.get('containers', []), list) or any(
         not isinstance(item, str) or not item for item in observed.get('containers', [])
     ):
@@ -456,14 +491,15 @@ class Suite:
         self.write('owned.json', {'workspace_uuid': self.variables['workspace_uuid'], 'resources': self.owned})
 
     def prepare(self):
-        if 'smoke' in self.profile and not Path(self.path(self.profile['smoke']['ledger_db'])).is_file():
-            raise ValueError('smoke.ledger_db must be an existing target Core SQLite database')
+        verification = self.profile.get('smoke') or self.profile.get('verification')
+        if verification and not Path(self.path(verification['ledger_db'])).is_file():
+            raise ValueError('ledger_db must be an existing target Core SQLite database')
         identity = self.call(['whoami'])
         self.variables['workspace_uuid'] = identity['workspace_uuid']
         self.journal()
         for plugin in self.profile.get('plugins', []):
             self.call(['plugin', 'install', plugin['source'], '--file', self.path(plugin['file']), '--wait'])
-        if 'smoke' in self.profile:
+        if 'smoke' in self.profile or 'verification' in self.profile:
             capabilities = self.call(['capabilities'])
             plugins = self.call(['plugin', 'list'])
             ready = {
@@ -471,7 +507,7 @@ class Suite:
                 for item in plugins['plugins']
                 if item.get('enabled') and item.get('status') == 'initialized'
             }
-            missing = set(self.profile['smoke'].get('required_plugins', [])) - ready
+            missing = set(self.profile.get('smoke', {}).get('required_plugins', [])) - ready
             self.write('environment.json', {'identity': identity, 'capabilities': capabilities, 'plugins': plugins})
             if missing:
                 raise RuntimeError(f'Required plugins are not initialized: {sorted(missing)}')
@@ -557,7 +593,7 @@ class Suite:
             else:
                 saved['resources'].remove(resource)
                 self.write('owned.json', saved)
-        if 'smoke' in self.profile:
+        if 'smoke' in self.profile or 'verification' in self.profile:
             from regression_smoke import verify_cleanup
 
             try:
@@ -577,7 +613,22 @@ class Suite:
             if 'body' in case:
                 case['args'] += ['--file', self.write(f'case-{len(cases)}.json', case.pop('body'))]
             cases.extend(
-                [{key: case[key] for key in ('name', 'args', 'expect', 'http_group') if key in case}]
+                [
+                    {
+                        key: case[key]
+                        for key in (
+                            'name',
+                            'args',
+                            'expect',
+                            'http_group',
+                            'box_exec',
+                            'attachment',
+                            'fields',
+                            'forbid',
+                        )
+                        if key in case
+                    }
+                ]
                 * case.get('weight', 1)
             )
         cases_path = self.write('cases.json', cases)
@@ -593,6 +644,15 @@ class Suite:
             observe_args += ['--container', container]
         if observed.get('box_data_root'):
             observe_args += ['--box-data-root', self.path(observed['box_data_root'])]
+        if verification := self.profile.get('verification'):
+            observe_args += [
+                '--ledger-db',
+                self.path(verification['ledger_db']),
+                '--workspace',
+                self.variables['workspace_uuid'],
+                '--verification-timeout',
+                str(verification.get('timeout_seconds', 5)),
+            ]
         results = []
         errors = []
         for phase in self.profile['phases']:
@@ -629,8 +689,18 @@ class Suite:
                 result = subprocess.run(command, stdout=log, stderr=subprocess.STDOUT, env=self.env)
             summary = json.loads((destination / 'summary.json').read_text())
             summary['name'] = phase['name']
-            summary['achieved_rps'] = summary['requests'] / summary['load_elapsed_seconds']
+            summary['achieved_rps'] = (
+                summary.get(
+                    'successful_requests', sum(row.get('status', {}).get('ok', 0) for row in summary['cases'].values())
+                )
+                / summary['load_elapsed_seconds']
+            )
             results.append(summary)
+            if self.profile.get('resource_limits'):
+                from regression_analysis import resource_verdict
+
+                summary['resources'] = resource_verdict(destination, self.profile['resource_limits'])
+                errors.extend(f'{phase["name"]}: {error}' for error in summary['resources']['errors'])
             if result.returncode:
                 errors.append(f'{phase["name"]}: business or health/resource checks failed')
             thresholds = self.profile.get('thresholds', {})
@@ -684,6 +754,10 @@ def main():
             return
         args.output.mkdir(parents=True, exist_ok=False, mode=0o700)
         suite.write('profile.json', profile)
+        if 'smoke' not in profile:
+            from regression_analysis import workload_contract
+
+            suite.write('workload.json', workload_contract(profile, args.profile))
         result = {'phases': [], 'errors': []}
         completed = False
         try:

tools/regression_traffic.py

@@ -11,6 +11,7 @@ import math
 import os
 from pathlib import Path
 import subprocess
+import sqlite3
 import sys
 import threading
 import time
@@ -50,6 +51,41 @@ def owned_box_containers(data_root, checked=None):
     return owned
 
 
+def sample_containers(configured, data_root, run=subprocess.run, discover=owned_box_containers):
+    def selected():
+        return set(configured) | (set(discover(data_root)) if data_root else set())
+
+    containers = selected()
+    if not containers:
+        return {'containers': []}
+
+    def stats(names):
+        return run(
+            ['docker', 'stats', '--no-stream', '--format', '{{json .}}', *sorted(names)],
+            capture_output=True,
+            text=True,
+            timeout=10,
+        )
+
+    result = stats(containers)
+    observation = {}
+    if result.returncode and data_root:
+        current = selected()
+        if containers - current:
+            observation['container_churn'] = {
+                'removed': sorted(containers - current),
+                'added': sorted(current - containers),
+                'first_error': result.stderr.strip(),
+            }
+            if not current:
+                return {**observation, 'containers': []}
+            result = stats(current)
+    observation['containers'] = [json.loads(line) for line in result.stdout.splitlines() if line]
+    if result.returncode:
+        observation['container_error'] = result.stderr.strip()
+    return observation
+
+
 def group_request(case, group_id, member_id, turn, token, previous, env=None):
     """Send one signed GroupMessage through the existing HTTP Bot adapter."""
     settings = case['http_group']
@@ -94,6 +130,34 @@ 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):
+    """Catch delayed extra or nonterminal runs after request-level verification."""
+    actual, unfinished = set(), []
+    connection = sqlite3.connect(Path(path).resolve().as_uri() + '?mode=ro', uri=True)
+    try:
+        values = sorted(conversations)
+        for start in range(0, len(values), 500):
+            part = values[start : start + 500]
+            query = (
+                'SELECT run_id, status FROM agent_run WHERE workspace_id=? AND id>? '
+                'AND conversation_id IN (' + ','.join('?' for _ in part) + ')'
+            )
+            for run_id, status in connection.execute(query, [workspace, after_id, *part]):
+                actual.add(run_id)
+                if status != 'completed':
+                    unfinished.append(run_id)
+    finally:
+        connection.close()
+    return {
+        'matched': actual == set(expected_ids) and not unfinished,
+        'expected_count': len(expected_ids),
+        'actual_count': len(actual),
+        'extra': sorted(actual - set(expected_ids)),
+        'missing': sorted(set(expected_ids) - actual),
+        'not_completed': unfinished,
+    }
+
+
 def main():
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument('--lbctl', required=True)
@@ -112,6 +176,9 @@ def main():
     parser.add_argument(
         '--group-state', type=Path, help='Carry group histories across load phases; refuses interrupted runs'
     )
+    parser.add_argument('--ledger-db', type=Path)
+    parser.add_argument('--workspace')
+    parser.add_argument('--verification-timeout', type=float, default=5)
     parser.add_argument('--cooldown', type=float, default=30)
     parser.add_argument('--pid', action='append', default=[], metavar='NAME=PID')
     parser.add_argument('--health', action='append', default=[], metavar='NAME=URL')
@@ -161,16 +228,28 @@ def main():
                 or not os.environ.get(settings['secret_env'])
             ):
                 parser.error('group cases need HTTP(S) url, message and a populated secret_env')
+    if args.ledger_db and (not args.ledger_db.is_file() or not args.workspace):
+        parser.error('ledger verification requires an existing SQLite file and workspace')
+    if not math.isfinite(args.verification_timeout) or args.verification_timeout <= 0:
+        parser.error('verification timeout must be positive')
+    from regression_smoke import UnresolvedRun, check_run, check_attachment, check_fields, completed_run
+
     args.output.mkdir(parents=True, exist_ok=False)
     pids = {name: int(pid) for name, pid in (value.split('=', 1) for value in args.pid)}
     endpoints = dict(value.split('=', 1) for value in args.health)
     rows = []
+    verified_ids = set()
+    conversations = set()
     sampling_errors = []
     lock = threading.Lock()
     group_condition = threading.Condition(lock)
     stop = threading.Event()
     phase = ['load']
     next_case = 0
+    phase_start_id = 0
+    if args.ledger_db:
+        with sqlite3.connect(args.ledger_db.resolve().as_uri() + '?mode=ro', uri=True) as db:
+            phase_start_id = db.execute('SELECT COALESCE(MAX(id),0) FROM agent_run').fetchone()[0]
     started = time.monotonic()
     deadline = started + args.duration
     redactions = [value for name, value in os.environ.items() if 'API_KEY' in name and value]
@@ -215,6 +294,52 @@ def main():
             text = text.replace(secret, '[REDACTED]')
         return text
 
+    def open_ledger():
+        if not args.ledger_db:
+            return None, None
+        connection = sqlite3.connect(args.ledger_db.resolve().as_uri() + '?mode=ro', uri=True, timeout=5)
+        connection.row_factory = sqlite3.Row
+        connection.execute('PRAGMA query_only=ON')
+        after_id = connection.execute('SELECT COALESCE(MAX(id),0) FROM agent_run').fetchone()[0]
+        return connection, after_id
+
+    def verify(row, case, connection, after_id, is_group=False):
+        began = time.monotonic()
+        data = row['output']['data']
+        checks = {}
+        if case.get('fields'):
+            checks['fields'] = check_fields(data, case['fields'])
+        if any(value in json.dumps(data, ensure_ascii=False) for value in case.get('forbid', [])):
+            raise ValueError('Response contains forbidden content')
+        runner = is_group or case.get('args', [])[:2] == ['pipeline', 'run']
+        row['requires_runner'] = runner
+        if connection and runner:
+            if not isinstance(data, dict):
+                raise ValueError('Runner response data must be an object')
+            session = data.get('session_id')
+            if not isinstance(session, str) or not session:
+                raise UnresolvedRun('Response has no conversation identity')
+            conversation = 'group_' + session if is_group else session
+            with lock:
+                conversations.add(conversation)
+            run = completed_run(
+                connection,
+                args.workspace,
+                conversation,
+                after_id,
+                args.verification_timeout,
+            )
+            row['run'] = run
+            checks.update(check_run(connection, run, case))
+            if case.get('attachment'):
+                checks['attachment'] = check_attachment(data, case['attachment'])
+            with lock:
+                if run['run_id'] in verified_ids:
+                    raise ValueError('Two requests matched the same AgentRun')
+                verified_ids.add(run['run_id'])
+        row['checks'] = checks
+        row['verification_seconds'] = time.monotonic() - began
+
     def sample():
         with (args.output / 'resources.jsonl').open('w') as stream:
             while not stop.is_set():
@@ -235,19 +360,8 @@ def main():
                         row['health'][name] = read_endpoint_snapshot(url, timeout_seconds=3)
                     except RuntimeError as exc:
                         row['health'][name] = {'error': str(exc)}
-                containers = set(args.container)
-                if args.box_data_root:
-                    containers.update(owned_box_containers(args.box_data_root))
-                if containers:
-                    result = subprocess.run(
-                        ['docker', 'stats', '--no-stream', '--format', '{{json .}}', *sorted(containers)],
-                        capture_output=True,
-                        text=True,
-                        timeout=10,
-                    )
-                    row['containers'] = [json.loads(line) for line in result.stdout.splitlines() if line]
-                    if result.returncode:
-                        row['container_error'] = result.stderr.strip()
+                if args.container or args.box_data_root:
+                    row.update(sample_containers(args.container, args.box_data_root))
                 stream.write(json.dumps(row) + '\n')
                 stream.flush()
                 with lock:
@@ -274,17 +388,37 @@ def main():
             else:
                 case = cases[(iteration * args.workers + index) % len(cases)]
             before = time.monotonic()
-            row = {'case': case['name'], 'time': time.time(), 'elapsed': before - started, 'status': 'failed'}
+            row = {
+                'case': case['name'],
+                'time': time.time(),
+                'elapsed': before - started,
+                'status': 'failed',
+                'schedule_lag_seconds': max(0, before - due),
+            }
+            connection = None
             try:
+                connection, after_id = open_ledger()
+                dispatch = time.monotonic()
+                row.update(
+                    ledger_setup_seconds=dispatch - before,
+                    elapsed=dispatch - started,
+                    schedule_lag_seconds=max(0, dispatch - due),
+                )
                 result = subprocess.run(
                     [args.lbctl, '--config', args.config, '-o', 'json', *case['args']],
                     capture_output=True,
                     text=True,
                     timeout=90,
                 )
+                row['seconds'] = time.monotonic() - dispatch
                 row['exit_code'] = result.returncode
                 row['output'] = json.loads(redact(result.stdout))
-                if result.returncode == 0 and case['expect'] in json.dumps(row['output'], ensure_ascii=False):
+                if not isinstance(row['output'], dict):
+                    raise ValueError('CLI response must be an object')
+                if result.returncode == 0 and case['expect'] in json.dumps(
+                    row['output'].get('data'), ensure_ascii=False
+                ):
+                    verify(row, case, connection, after_id)
                     row['status'] = 'ok'
                 elif 'result_unknown' in result.stdout:
                     row['status'] = 'unknown'
@@ -292,9 +426,14 @@ def main():
                     row['error'] = redact(result.stderr)[:2000] or 'exit code or response oracle failed'
             except subprocess.TimeoutExpired:
                 row.update(status='unknown', error='CLI timed out; server outcome unknown; request was not retried')
-            except (OSError, ValueError) as exc:
-                row['error'] = str(exc)
-            row['seconds'] = time.monotonic() - before
+            except UnresolvedRun as exc:
+                row.update(status='unknown', error=str(exc))
+            except (OSError, ValueError, KeyError, sqlite3.Error) as exc:
+                row['error'] = redact(str(exc))
+            finally:
+                if connection:
+                    connection.close()
+            row.setdefault('seconds', time.monotonic() - before)
             with lock:
                 rows.append(row)
                 with (args.output / 'requests.jsonl').open('a') as stream:
@@ -333,9 +472,20 @@ def main():
                 'time': time.time(),
                 'elapsed': before - started,
                 'status': 'failed',
+                'schedule_lag_seconds': max(0, before - group['due']),
             }
+            connection = None
             try:
+                connection, after_id = open_ledger()
+                dispatch = time.monotonic()
+                row.update(
+                    ledger_setup_seconds=dispatch - before,
+                    elapsed=dispatch - started,
+                    schedule_lag_seconds=max(0, dispatch - group['due']),
+                )
                 row['output'] = group_request(case, group['id'], member, group['turn'], token, group['previous'])
+                row['seconds'] = time.monotonic() - dispatch
+                verify(row, case, connection, after_id, is_group=True)
                 row['status'] = 'ok'
             except urllib.error.HTTPError as exc:
                 row.update(http_status=exc.code, error=redact(exc.read(2000).decode(errors='replace')))
@@ -343,9 +493,14 @@ def main():
                     row['status'] = 'unknown'
             except (urllib.error.URLError, TimeoutError, OSError) as exc:
                 row.update(status='unknown', error=redact(str(exc)) + '; not retried')
-            except (ValueError, KeyError) as exc:
+            except UnresolvedRun as exc:
+                row.update(status='unknown', error=str(exc))
+            except (ValueError, KeyError, sqlite3.Error) as exc:
                 row['error'] = redact(str(exc))
-            row['seconds'] = time.monotonic() - before
+            finally:
+                if connection:
+                    connection.close()
+            row.setdefault('seconds', time.monotonic() - before)
             with group_condition:
                 rows.append(row)
                 with (args.output / 'requests.jsonl').open('a') as stream:
@@ -382,9 +537,23 @@ def main():
         'workers': args.workers,
         'cooldown_seconds': args.cooldown,
         'requests': len(rows),
+        'successful_requests': sum(row['status'] == 'ok' for row in rows),
+        'verified_runs': len(verified_ids),
+        'runner_requests': sum(row.get('requires_runner', False) for row in rows),
+        'verification_enabled': bool(args.ledger_db),
+        'planned_requests': math.ceil(args.duration * args.rps),
+        'delivery_ratio': len(rows) / math.ceil(args.duration * args.rps),
+        'drain_seconds': max(0, load_elapsed - args.duration),
+        'schedule_lag_p95_seconds': percentile([row['schedule_lag_seconds'] for row in rows], 0.95),
+        'load_model': 'bounded workers; serialized groups; dispatch latency excludes verification',
         'status': dict(Counter(row['status'] for row in rows)),
         'cases': {},
     }
+    if args.ledger_db:
+        conversations.update('group_' + group['id'] for group in groups)
+        summary['ledger_closeout'] = reconcile_runs(
+            args.ledger_db, args.workspace, phase_start_id, conversations, verified_ids
+        )
     samples = [json.loads(line) for line in (args.output / 'resources.jsonl').read_text().splitlines()]
     summary['health_samples'] = len(samples)
     summary['health_failures'] = sum(
@@ -441,6 +610,7 @@ def main():
         or summary['health_failures']
         or sampling_errors
         or summary.get('group_failures')
+        or (args.ledger_db and not summary['ledger_closeout']['matched'])
     ):
         raise SystemExit(1)
 

tools/test_regression_analysis.py

@@ -0,0 +1,154 @@
+"""Counterexamples for load resource recovery and compatible-run comparisons."""
+
+import copy
+import json
+from pathlib import Path
+import tempfile
+import unittest
+
+from regression_analysis import compare, container_metrics, resource_verdict, workload_contract
+
+
+class AnalysisTests(unittest.TestCase):
+    def setUp(self):
+        temporary = tempfile.TemporaryDirectory()
+        self.addCleanup(temporary.cleanup)
+        self.root = Path(temporary.name)
+        self.limits = {
+            'warmup_seconds': 10,
+            'window_seconds': 10,
+            'min_samples': 3,
+            'targets': {'core': {'rss_bytes': 100, 'open_fds': 2}},
+        }
+
+    def resources(self, tail_fds=10, omit=False):
+        samples = [
+            {'phase': phase, 'elapsed': second, 'targets': {'core': {'rss_bytes': 1000, 'open_fds': fds}}}
+            for phase, seconds, fds in [('load', range(10, 20), 10), ('cooldown', range(30, 40), tail_fds)]
+            for second in seconds
+        ]
+        if omit:
+            samples = [row for row in samples if row['phase'] == 'load']
+        (self.root / 'resources.jsonl').write_text(''.join(json.dumps(row) + '\n' for row in samples))
+
+    def test_healthy_responses_cannot_hide_retained_fds(self):
+        self.resources()
+        self.assertEqual(resource_verdict(self.root, self.limits)['errors'], [])
+        self.resources(tail_fds=40)
+        self.assertIn('retained growth', resource_verdict(self.root, self.limits)['errors'][0])
+
+    def test_missing_cooldown_is_unknown_not_zero_growth(self):
+        self.resources(omit=True)
+        self.assertIn('unknown', resource_verdict(self.root, self.limits)['errors'][0])
+        self.resources()
+        data = (self.root / 'resources.jsonl').read_text().replace('"core"', '"other"')
+        (self.root / 'resources.jsonl').write_text(data)
+        self.assertTrue(resource_verdict(self.root, self.limits)['errors'])
+
+    def test_docker_memory_is_independent_of_service_process_rss(self):
+        row = {'containers': [{'MemUsage': '10MiB / 1GiB', 'PIDs': '2'}, {'MemUsage': '5MiB / 1GiB', 'PIDs': '3'}]}
+        self.assertEqual(container_metrics(row), {'memory_bytes': 15 * 1024**2, 'pids': 5, 'count': 2})
+        self.assertEqual(container_metrics({'container_error': 'EOF', **row}), {})
+        self.resources()
+        rows = [json.loads(line) for line in (self.root / 'resources.jsonl').read_text().splitlines()]
+        for sample in rows:
+            sample['containers'] = [
+                {'MemUsage': ('10MiB' if sample['phase'] == 'load' else '100MiB') + ' / 1GiB', 'PIDs': '1'}
+            ]
+        (self.root / 'resources.jsonl').write_text(''.join(json.dumps(row) + '\n' for row in rows))
+        verdict = resource_verdict(self.root, {**self.limits, 'containers': {'memory_bytes': 1024**2}})
+        self.assertIn('docker_containers/memory_bytes', verdict['errors'][0])
+
+    def test_stimulus_variables_are_compared_but_target_origin_is_not(self):
+        profile = {
+            'vars': {'prompt': 'first', 'endpoint': 'http://127.0.0.1:1'},
+            'cases': [
+                {'args': ['pipeline', 'run', '--message', '${prompt}'], 'http_group': {'url': '${endpoint}/sync'}}
+            ],
+        }
+        before = workload_contract(profile, self.root / 'profile.json')
+        profile['vars']['endpoint'] = 'http://127.0.0.1:2'
+        self.assertEqual(before, workload_contract(profile, self.root / 'profile.json'))
+        profile['vars']['prompt'] = 'changed'
+        self.assertNotEqual(before, workload_contract(profile, self.root / 'profile.json'))
+
+    def fixture_runs(self):
+        summary = {
+            'passed': True,
+            'phases': [
+                {
+                    'name': 'sustained',
+                    'achieved_rps': 1,
+                    'verification_enabled': True,
+                    'verified_runs': 100,
+                    'ledger_closeout': {'matched': True},
+                    'runner_requests': 100,
+                    'cases': {'chat': {'requests': 100, 'p95_seconds': 1}},
+                }
+            ],
+        }
+        for name in ['baseline', 'candidate']:
+            root = self.root / name
+            root.mkdir()
+            (root / 'summary.json').write_text(json.dumps(summary))
+            (root / 'workload.json').write_text(json.dumps({'verified_runs': True, 'fixture': 'golden'}))
+            (root / 'environment.json').write_text(
+                json.dumps(
+                    {
+                        'plugins': {
+                            'plugins': [
+                                {
+                                    'author': 'team',
+                                    'name': 'Runner',
+                                    'version': '1',
+                                    'enabled': True,
+                                    'status': 'initialized',
+                                }
+                            ]
+                        }
+                    }
+                )
+            )
+        return self.root / 'baseline', self.root / 'candidate', summary
+
+    def test_comparison_rejects_slow_underfed_missing_or_unverified_samples(self):
+        baseline, candidate, original = self.fixture_runs()
+        self.assertTrue(compare(baseline, candidate)['passed'])
+        for field, value in [('p95_seconds', 1.5), ('requests', 2), ('p95_seconds', float('nan'))]:
+            summary = copy.deepcopy(original)
+            summary['phases'][0]['cases']['chat'][field] = value
+            (candidate / 'summary.json').write_text(json.dumps(summary))
+            with self.subTest(field=field, value=value):
+                self.assertFalse(compare(baseline, candidate)['passed'])
+        for field, value in [
+            ('achieved_rps', 0.5),
+            ('achieved_rps', float('nan')),
+            ('verified_runs', 99),
+            ('ledger_closeout', {'matched': False}),
+        ]:
+            summary = copy.deepcopy(original)
+            summary['phases'][0][field] = value
+            (candidate / 'summary.json').write_text(json.dumps(summary))
+            with self.subTest(field=field):
+                self.assertFalse(compare(baseline, candidate)['passed'])
+
+    def test_comparison_refuses_different_workload_or_failed_baseline(self):
+        baseline, candidate, summary = self.fixture_runs()
+        (candidate / 'workload.json').write_text(json.dumps({'verified_runs': True, 'fixture': 'changed'}))
+        self.assertIn('not comparable', compare(baseline, candidate)['errors'][0])
+        (candidate / 'workload.json').write_text((baseline / 'workload.json').read_text())
+        summary['passed'] = False
+        (baseline / 'summary.json').write_text(json.dumps(summary))
+        self.assertFalse(compare(baseline, candidate)['passed'])
+
+    def test_fixture_content_changes_contract_even_at_same_path(self):
+        fixture = self.root / 'fixture.txt'
+        fixture.write_text('before')
+        profile = {'resources': [{'documents': [{'file': 'fixture.txt'}]}]}
+        before = workload_contract(profile, self.root / 'profile.json')
+        fixture.write_text('after')
+        self.assertNotEqual(before, workload_contract(profile, self.root / 'profile.json'))
+
+
+if __name__ == '__main__':
+    unittest.main()

tools/test_regression_load_ledger.py

@@ -0,0 +1,137 @@
+"""Check delayed duplicate runs and separate ledger setup from request latency."""
+
+from contextlib import redirect_stdout
+import io
+import json
+from pathlib import Path
+import sqlite3
+import sys
+import tempfile
+from types import SimpleNamespace
+import time
+import unittest
+from unittest.mock import patch, Mock
+from uuid import uuid4
+
+from regression_traffic import main, reconcile_runs, sample_containers
+
+
+class LoadLedgerTests(unittest.TestCase):
+    def setUp(self):
+        directory = tempfile.TemporaryDirectory()
+        self.addCleanup(directory.cleanup)
+        self.root = Path(directory.name)
+        self.db = self.root / 'ledger.db'
+        with sqlite3.connect(self.db) as db:
+            db.executescript("""
+                CREATE TABLE agent_run (id INTEGER PRIMARY KEY, run_id TEXT, workspace_id TEXT,
+                    conversation_id TEXT, status TEXT, status_reason TEXT, finished_at TEXT);
+                CREATE TABLE agent_run_event (run_id TEXT, sequence INTEGER, type TEXT, data_json TEXT);
+            """)
+
+    def test_phase_closeout_catches_delayed_extra_and_unfinished_runs(self):
+        with sqlite3.connect(self.db) as db:
+            db.execute("INSERT INTO agent_run VALUES (1,'expected','ws','group','completed','','now')")
+            db.execute("INSERT INTO agent_run VALUES (2,'unrelated','other','group','completed','','now')")
+        self.assertTrue(reconcile_runs(self.db, 'ws', 0, {'group'}, {'expected'})['matched'])
+        with sqlite3.connect(self.db) as db:
+            db.execute("INSERT INTO agent_run VALUES (3,'late','ws','group','running','',NULL)")
+        result = reconcile_runs(self.db, 'ws', 0, {'group'}, {'expected'})
+        self.assertFalse(result['matched'])
+        self.assertEqual(result['extra'], ['late'])
+        self.assertEqual(result['not_completed'], ['late'])
+
+    def test_container_churn_recovers_once_but_other_collection_errors_fail(self):
+        failed = SimpleNamespace(returncode=1, stdout='', stderr='No such container: old')
+        success = SimpleNamespace(returncode=0, stdout='{"MemUsage":"1MiB / 1GiB","PIDs":"1"}\n', stderr='')
+        run = Mock(side_effect=[failed, success])
+        result = sample_containers([], '/owned', run, Mock(side_effect=[['old', 'live'], ['live']]))
+        self.assertNotIn('container_error', result)
+        self.assertEqual(result['container_churn']['removed'], ['old'])
+        self.assertEqual(run.call_count, 2)
+        self.assertEqual(run.call_args.args[0][-1], 'live')
+        run = Mock(return_value=failed)
+        result = sample_containers([], '/owned', run, Mock(return_value=['live']))
+        self.assertIn('container_error', result)
+        self.assertEqual(run.call_count, 1)
+        run = Mock(return_value=failed)
+        result = sample_containers([], '/owned', run, Mock(side_effect=[['old', 'live'], ['live']]))
+        self.assertIn('container_error', result)
+        self.assertEqual(run.call_count, 2)
+
+    def test_slow_ledger_snapshot_does_not_inflate_response_latency(self):
+        connect = sqlite3.connect
+        cases = self.root / 'cases.json'
+        cases.write_text(
+            json.dumps(
+                [
+                    {
+                        'name': 'chat',
+                        'expect': 'OK',
+                        'http_group': {
+                            'url': 'http://127.0.0.1:1/unused',
+                            'message': 'hello',
+                            'secret_env': 'TEST_GROUP_SECRET',
+                        },
+                    }
+                ]
+            )
+        )
+
+        def slow_connect(*args, **kwargs):
+            time.sleep(0.12)
+            return connect(*args, **kwargs)
+
+        def response(case, group, *_args):
+            run_id = uuid4().hex
+            with connect(self.db) as db:
+                db.execute(
+                    'INSERT INTO agent_run VALUES (NULL,?,?,?,?,?,?)',
+                    (run_id, 'ws', 'group_' + group, 'completed', '', 'now'),
+                )
+                db.execute('INSERT INTO agent_run_event VALUES (?,1,?,?)', (run_id, 'run.completed', '{}'))
+            return {'data': {'session_id': group}}
+
+        output = self.root / 'run'
+        argv = [
+            'traffic',
+            '--lbctl',
+            'unused',
+            '--config',
+            'unused',
+            '--cases',
+            str(cases),
+            '--output',
+            str(output),
+            '--duration',
+            '0.4',
+            '--cooldown',
+            '0',
+            '--rps',
+            '8',
+            '--workers',
+            '1',
+            '--groups',
+            '1',
+            '--ledger-db',
+            str(self.db),
+            '--workspace',
+            'ws',
+        ]
+        with (
+            patch.object(sys, 'argv', argv),
+            patch.dict('os.environ', {'TEST_GROUP_SECRET': 'test'}),
+            patch('regression_traffic.sqlite3.connect', side_effect=slow_connect),
+            patch('regression_traffic.group_request', side_effect=response),
+            redirect_stdout(io.StringIO()),
+        ):
+            main()
+        rows = [json.loads(line) for line in (output / 'requests.jsonl').read_text().splitlines()]
+        self.assertGreaterEqual(len(rows), 2)
+        self.assertTrue(all(row['ledger_setup_seconds'] >= 0.12 for row in rows))
+        self.assertTrue(all(row['seconds'] < row['ledger_setup_seconds'] for row in rows))
+        self.assertTrue(json.loads((output / 'summary.json').read_text())['ledger_closeout']['matched'])
+
+
+if __name__ == '__main__':
+    unittest.main()