Add frozen nonlinear CSI search and audited SOTA evidence

.github/workflows/csi-encoder-training.yml

@@ -7,6 +7,7 @@ on:
       - 'scripts/train-csi-encoder.py'
       - 'scripts/train-espfi.py'
       - 'scripts/improve-espfi.py'
+      - 'scripts/improve-espfi-kernel.py'
       - 'scripts/requirements-espfi-training.txt'
       - 'scripts/requirements-csi-training.txt'
       - 'scripts/tests/test_csi_training*.py'

docs/espfi-model-optimization.md

@@ -222,3 +222,65 @@ old encoder's eight engineered features.
 
 Acceptance remains a new untouched room and participant, resolved semantic
 labels, and target-device measurement before any production promotion.
+
+## Nonlinear development search
+
+The next frozen grid contains 56 candidates on the same two training-only
+folds. Relative spectral and combined features each use random Fourier maps
+of 128 or 512 dimensions, gamma multipliers 0.1, 1 or 10 divided by the input
+feature count, and ridge alpha 0.01, 0.1, 1 or 10. Projection seed is 42.
+Six linear controls use alpha 1, 10 or 100; two nearest-centroid controls
+complete the grid. Fold scalers see only their training records. Mean fold
+macro F1 selects the candidate before development evaluation.
+
+```sh
+python scripts/improve-espfi-kernel.py \
+  --manifest /private/espfi/prepared/manifest.json \
+  --baseline /private/espfi/generalization-001-float32 \
+  --output /private/espfi/kernel-001 --max-seconds 600
+```
+
+The CLI fixes BLAS and OpenMP to one thread before imports. The output records
+the frozen search contract before fitting and the complete candidate selection
+before external development inference. Load completed candidates using
+`csi_training.espfi_kernel.KernelBundle` and the same pooled amplitude contract.
+
+Random Fourier maps permit nonlinear decision boundaries without including
+training support vectors in the exported model. Account for their stored
+projection matrices separately from fitted classifier parameters. Compare
+the selected candidate with the frozen float32 spectral ridge using paired
+full-API latency and artifact bytes as well as development macro F1.
+
+The same development partition has now informed multiple research rounds.
+It cannot establish independent generalization, and the previously opened
+test cannot recover that role by being renamed. See the primary-source
+[benchmark evidence](espfi-sota-evidence.md) for the distinct source-only,
+calibration and deployment protocols and the gate for a SOTA claim.
+
+### Measured nonlinear result
+
+Training-only selection chose combined features, 128 Fourier dimensions,
+gamma multiplier 0.1 and ridge alpha 0.01. Mean inner-fold macro F1 was
+0.306028, versus 0.297443 for the spectral ridge control. Development macro
+F1 reached 0.383999, versus 0.366065 for the incumbent. Accuracy rose from
+29/70 to 31/70. Activities 3 and 7 regressed to zero F1, so this candidate
+does not improve every class.
+
+The float64 export contains 903 classifier scalars, 60,032 fixed random-map
+scalars and 1,064 preprocessing scalars. Its 496,488-byte artifact is 28.97
+times the incumbent's 17,140 bytes. In the verified 500-pair full-API timing,
+candidate p50/p95 were 0.168208/0.312287 ms, versus incumbent
+0.134792/0.238876 ms. Median latency was 24.79% higher on this shared CPU.
+No hardware or stable tail-latency claim follows.
+
+The first real run was launched before a failed synthetic reload check was
+inspected. That check exposed floating-point evaluation differences caused
+by transposed, noncontiguous centroid weights. Canonical contiguous fitted
+state fixed the software issue. After tests passed, the unchanged frozen
+grid was replayed. Selection, model bytes and all 70 development logits
+were identical to the original real run. Both source snapshots and receipts
+are retained; the replay is verification, not independent accuracy evidence.
+
+Decision: retain the small spectral model and archive the nonlinear model
+as an accuracy experiment. No deployed model is replaced, no final-test
+arrays were evaluated, and neither candidate establishes SOTA.

docs/espfi-sota-evidence.md

@@ -0,0 +1,39 @@
+# ESP-Fi HAR benchmark evidence
+
+Review date: 2026-10-12. External results below are **CLAIMED**, not reproduced by RuView. A larger score on a different split is not evidence of superiority. No SOTA claim is authorized by this review.
+
+## Current reference targets
+
+| Reference | Reported result | Protocol and deployment | Comparability |
+| --- | --- | --- | --- |
+| TriGFormer, SSRN preprint posted 2026-09-04 | Macro F1 0.8014 for unseen environments; 0.7234 for simultaneous environment and participant shift | Source data only; no target examples or target statistics during training. Reported 333,000 parameters and 10.577 GFLOP inference assigned to a gateway | Closest research reference, but exact folds and preprocessing remain unverified |
+| CSI SelfAdapt ESP32, public commit 2026-06-05, manuscript marked submitted | Accuracy 0.2512 across four environment folds; 0.3335 across eight participant folds; 0.6715 within domain | CompactCSINet, 384 frame windows, stride 96, 40 epochs, host MPS evaluation | Different metric, windows, split and averaging from RuView |
+| CSI SelfAdapt personalization | Accuracy 0.5828 with 10 labeled windows per class; 0.7386 with 50 | Last block adaptation using labeled data from the target participant, averaged over eight participants and three adaptation seeds | Not source only generalization; compare in a separate calibration track |
+
+TriGFormer separates temporal, subcarrier and local dynamics, combines amplitude with temporal differences, and uses supervised contrastive learning across source environments. Its abstract identifies local processing and temporal differences as useful contributors. These are hypotheses for controlled ablations, not a verified reproduction recipe. The SSRN page lists 20 pages. Direct access to the page and PDF returned access errors during this review; primary indexed abstract text was available. No author code repository was located. Fold membership, seeds, confidence intervals, duplicate handling and exact joint shift protocol could not be checked. [Primary abstract](https://papers.ssrn.com/sol3/papers.cfm?abstract_id=7411437), [DOI](https://doi.org/10.2139/ssrn.7411437).
+
+## Verified source details
+
+The official ESP-Fi publication is Wen et al., *Ad Hoc Networks* 186, article 104192, issue date 2026-05-01. The publisher abstract establishes four environments and seven activities but does not expose a comparable numerical result. Full paper access failed; do not infer a benchmark score from its abstract. [Publisher](https://www.sciencedirect.com/science/article/abs/pii/S1570870526000582), [DOI](https://doi.org/10.1016/j.adhoc.2026.104192).
+
+The official repository at revision `c8bf6aa0d680fa02695c85a82b05e79b3ebb10c4` provides seven model families and sample normalization. Its README describes checkpoint selection using training accuracy and supplies no validation split. This differs from RuView's development procedure. The public archive contains 2,240 recordings; RuView verified the pinned LFS digest and quarantined two duplicate signal groups involving four recordings. The README activity number mapping conflicts with benchmark directory names, so the RuView adapter retains numeric activities. [Pinned official README](https://github.com/AutoSmartGroup/ESP-Fi-HAR/blob/c8bf6aa0d680fa02695c85a82b05e79b3ebb10c4/README.md).
+
+The SelfAdapt repository was inspected at revision `f9b9ac0faf7d0b033ce5b5041d6e40453fb56054`. Its source evaluation computes training normalization before carving trial 10 into validation. The held environment or participant remains excluded from this normalization, but validation statistics enter preprocessing. Its parser does not perform RuView's duplicate quarantine. Consequently even reproducing its scripts is not identical to RuView's audited intake.
+
+SelfAdapt personalization uses target participant trials 1 through 8 for calibration and target normalization, then trials 9 and 10 for evaluation. Its prominent 0.806 int8 accuracy and 151 ms device inference result concern the three class Wallhack task, not seven class ESP-Fi. The ESP-Fi result JSON reports host MPS evaluation. Hardware claims from that paper cannot be transferred to RuView.
+
+Pinned primary evidence:
+
+1. [ESP-Fi source evaluation](https://github.com/Rehan000/CSI-SelfAdapt-ESP32/blob/f9b9ac0faf7d0b033ce5b5041d6e40453fb56054/scripts/eval_espfi_shift.py).
+2. [Source result JSON](https://github.com/Rehan000/CSI-SelfAdapt-ESP32/blob/f9b9ac0faf7d0b033ce5b5041d6e40453fb56054/results/e3b_espfi_shift.json).
+3. [Personalization code](https://github.com/Rehan000/CSI-SelfAdapt-ESP32/blob/f9b9ac0faf7d0b033ce5b5041d6e40453fb56054/scripts/adapt_espfi_person.py).
+4. [Personalization result JSON](https://github.com/Rehan000/CSI-SelfAdapt-ESP32/blob/f9b9ac0faf7d0b033ce5b5041d6e40453fb56054/results/e5b_espfi_personalize.json).
+5. [Full manuscript source](https://github.com/Rehan000/CSI-SelfAdapt-ESP32/blob/f9b9ac0faf7d0b033ce5b5041d6e40453fb56054/paper/main.tex).
+
+## Decision and acceptance
+
+Use 0.8014 and 0.7234 macro F1 as provisional external research references only. The current fixed RuView split, with one room and participant combination in the test partition, is not their benchmark. Once test feedback influences later development, that test is no longer untouched confirmation evidence.
+
+Next action: obtain the TriGFormer protocol and freeze a comparable evaluation contract before attempting a superiority claim. Preserve a separate source only track and a labeled calibration track. Record signal digests, complete fold memberships, quarantine policy, training statistics, seed list, parameter count, latency device, memory and metrics.
+
+Acceptance test: an independent runner reproduces both baseline and candidate under identical frozen folds and compute budgets, with test arrays inaccessible during selection, then reports paired per fold macro F1 and uncertainty. Until that passes, describe improvements only relative to the locally reproduced baseline and its stated protocol.

scripts/csi_training/espfi_kernel.py

@@ -0,0 +1,213 @@
+"""Frozen nonlinear development search with no support-vector retention."""
+import argparse
+import json
+from pathlib import Path
+import time
+
+import numpy as np
+from safetensors.numpy import load_file, save_file
+from .espfi_generalize import features, folds, GeneralizationBundle
+from .espfi_train import digest, scores, write_json
+
+FAMILIES = ('relative_spectral', 'relative_combined')
+GRID = ([{'family': family, 'kind': 'linear', 'alpha': alpha} for family in FAMILIES for alpha in (1., 10., 100.)]
+        + [{'family': family, 'kind': 'centroid'} for family in FAMILIES]
+        + [{'family': family, 'kind': 'rff', 'dimensions': dimensions, 'gamma_multiplier': gamma, 'alpha': alpha}
+           for family in FAMILIES for dimensions in (128, 512) for gamma in (.1, 1., 10.) for alpha in (.01, .1, 1., 10.)])
+CONTRACT = {'seed': 42, 'grid': GRID, 'input': 'nonnegative_amplitude_N52T190',
+            'selection': 'mean training-only grouped CV macro F1, then fewer stored scalars, then declaration order',
+            'folds': [{'train_room': 1, 'validation_room': 2}, {'train_room': 2, 'validation_room': 1}],
+            'train_participants': [1, 2, 3, 4], 'validation_participants': [5, 6],
+            'gamma': 'multiplier / 468 after train-only standardization',
+            'rff': 'sqrt(2/M) cos(Z @ Gaussian(0,2gamma) + Uniform(0,2pi))',
+            'support_vectors': False}
+
+
+def transformed(state, values):
+    values = np.asarray(values, dtype=state['feature_mean'].dtype)
+    z = (values - state['feature_mean']) / state['feature_scale']
+    if 'projection' in state:
+        return np.sqrt(2. / state['phase'].size) * np.cos(z @ state['projection'] + state['phase'])
+    return z
+
+
+def predict_features(state, values):
+    z = transformed(state, values) - state['output_mean']
+    result = z @ state['weights'] + state['bias']
+    if not np.isfinite(result).all():
+        raise ValueError('Nonfinite classifier output')
+    return result
+
+
+def fit(values, labels, candidate):
+    mean, scale = values.mean(0), np.maximum(values.std(0), 1e-6)
+    state = {'feature_mean': mean, 'feature_scale': scale}
+    if candidate['kind'] == 'rff':
+        rng = np.random.default_rng(42)
+        dimensions = candidate['dimensions']
+        gamma = candidate['gamma_multiplier'] / values.shape[1]
+        state['projection'] = rng.normal(0, np.sqrt(2 * gamma), (values.shape[1], dimensions))
+        state['phase'] = rng.uniform(0, 2 * np.pi, dimensions)
+    z = transformed(state, values)
+    if candidate['kind'] == 'centroid':
+        centroids = np.stack([z[labels == label].mean(0) for label in range(7)])
+        state.update({'output_mean': np.zeros(z.shape[1]), 'weights': 2 * centroids.T,
+                      'bias': -np.square(centroids).sum(1)})
+    else:
+        output_mean = z.mean(0)
+        z = z - output_mean
+        target = np.eye(7)[labels]
+        bias = target.mean(0)
+        target -= bias
+        if len(z) < z.shape[1]:
+            gram = z @ z.T
+            gram.flat[::len(gram) + 1] += candidate['alpha']
+            weights = z.T @ np.linalg.solve(gram, target)
+        else:
+            gram = z.T @ z
+            gram.flat[::len(gram) + 1] += candidate['alpha']
+            weights = np.linalg.solve(gram, z.T @ target)
+        state.update({'output_mean': output_mean, 'weights': weights, 'bias': bias})
+    if any(not np.isfinite(v).all() for v in state.values()):
+        raise ValueError('Nonfinite model state')
+    # Canonical layout avoids BLAS reduction-order differences after export.
+    return {k: np.ascontiguousarray(v) for k, v in state.items()}
+
+
+class KernelBundle:
+    def __init__(self, directory):
+        root = Path(directory)
+        for name in ('metrics.json', 'status.json', 'model.safetensors'):
+            path = root / name
+            if path.is_symlink() or not path.is_file() or path.stat().st_size > 4 * 1024**2:
+                raise ValueError('Invalid bounded artifact')
+        report = json.loads((root / 'metrics.json').read_text())
+        status = json.loads((root / 'status.json').read_text())
+        if (status.get('status') != 'complete' or status.get('promotionAuthorized') is not False
+                or report.get('status') != 'complete' or report.get('promotionAuthorized') is not False
+                or report.get('task') != 'public_amplitude_nonlinear_development' or report.get('contract') != CONTRACT):
+            raise ValueError('Invalid completed kernel candidate')
+        if report.get('model_sha256') != digest(root / 'model.safetensors'):
+            raise ValueError('Model hash mismatch')
+        self.candidate = report['selected']['candidate']
+        if self.candidate not in GRID:
+            raise ValueError('Unknown candidate')
+        self.state = load_file(str(root / 'model.safetensors'))
+        width = self.candidate.get('dimensions', 468)
+        expected = {'feature_mean': (468,), 'feature_scale': (468,), 'output_mean': (width,),
+                    'weights': (width, 7), 'bias': (7,)}
+        if self.candidate['kind'] == 'rff':
+            expected.update({'projection': (468, width), 'phase': (width,)})
+        if set(self.state) != set(expected) or any(self.state[k].shape != shape or self.state[k].dtype != np.float64 or not np.isfinite(self.state[k]).all() for k, shape in expected.items()) or not (self.state['feature_scale'] > 0).all():
+            raise ValueError('Invalid bounded classifier state')
+
+    def predict(self, amplitudes):
+        return predict_features(self.state, features(amplitudes, self.candidate['family']))
+
+
+def search(parts, provenance, baseline_directory, output, max_seconds=600):
+    if set(parts) != {'train', 'validation'} or not 1 <= max_seconds <= 3600:
+        raise ValueError('Requires bounded development-only partitions')
+    start = time.monotonic()
+    baseline = GeneralizationBundle(baseline_directory)
+    baseline_report = json.loads((Path(baseline_directory) / 'metrics.json').read_text())
+    if baseline_report['provenance']['manifest_sha256'] != provenance['manifest_sha256']:
+        raise ValueError('Data provenance differs from incumbent')
+    train, validation = parts['train'], parts['validation']
+    y = np.asarray(train['y'])
+    if y.dtype.kind not in 'iu' or y.shape != (len(train['x']),) or set(y) != set(range(7)):
+        raise ValueError('Invalid training labels')
+    fold_ids = folds(train['identity'], y)
+    ids = np.asarray(validation['identity'])
+    if ids.shape != (len(validation['y']), 4) or not np.all(ids[:, 0] == 3) or not np.all(ids[:, 1] == 7) or not np.array_equal(ids[:, 2] - 1, validation['y']):
+        raise ValueError('Invalid development identities')
+    root = Path(output)
+    root.mkdir(exist_ok=False, parents=True)
+    write_json(root / 'status.json', {'status': 'running', 'promotionAuthorized': False})
+    write_json(root / 'search_contract.json', CONTRACT)
+    try:
+        cache = {family: features(train['x'], family) for family in FAMILIES}
+        results, selected, best = [], None, None
+        for candidate in GRID:
+            x = cache[candidate['family']]
+            evaluations = []
+            for fit_ids, eval_ids in fold_ids:
+                if time.monotonic() - start > max_seconds:
+                    raise TimeoutError('Nonlinear search budget exceeded')
+                state = fit(x[fit_ids], y[fit_ids], candidate)
+                evaluations.append(scores(y[eval_ids], predict_features(state, x[eval_ids]).argmax(1), 7))
+            result = {'candidate': candidate, 'folds': evaluations,
+                      'mean_cv_macro_f1': float(np.mean([e['macro_f1'] for e in evaluations])),
+                      'stored_scalar_count': sum(v.size for v in state.values())}
+            results.append(result)
+            rank = (result['mean_cv_macro_f1'], -result['stored_scalar_count'])
+            if best is None or rank > best:
+                selected, best = result, rank
+        # Freeze selection before touching external development values or labels.
+        write_json(root / 'selection.json', {'selected': selected, 'candidates': results, 'contract': CONTRACT})
+        chosen = selected['candidate']
+        state = fit(cache[chosen['family']], y, chosen)
+        save_file({k: np.ascontiguousarray(v) for k, v in state.items()}, str(root / 'model.safetensors'))
+        dev_features = features(validation['x'], chosen['family'])
+        prediction = predict_features(state, dev_features)
+        reloaded = load_file(str(root / 'model.safetensors'))
+        if not np.array_equal(prediction, predict_features(reloaded, dev_features)):
+            raise ValueError('Export changed logits')
+        incumbent = scores(validation['y'], baseline.predict(validation['x']).argmax(1), 7)
+        measured = scores(validation['y'], prediction.argmax(1), 7)
+        # Full same-method inference including feature extraction and input/output checks.
+        sample = validation['x'][:1]
+        def inference():
+            return predict_features(state, features(sample, chosen['family']))
+        for _ in range(20):
+            inference()
+            baseline.predict(sample)
+        timings = {'candidate': [], 'incumbent': []}
+        for i in range(500):
+            pairs = [('candidate', inference), ('incumbent', lambda: baseline.predict(sample))]
+            if i % 2:
+                pairs.reverse()
+            for name, fn in pairs:
+                before = time.perf_counter_ns()
+                fn()
+                timings[name].append((time.perf_counter_ns() - before) / 1e6)
+        report = {'schema_version': 1, 'task': 'public_amplitude_nonlinear_development', 'status': 'complete',
+                  'promotionAuthorized': False, 'contract': CONTRACT, 'selected': selected, 'candidates': results,
+                  'evidence': 'SYNTHETIC' if provenance.get('evidence') == 'SYNTHETIC' else 'MEASURED',
+                  'development': measured, 'incumbent_development': incumbent,
+                  'development_macro_f1_delta': measured['macro_f1'] - incumbent['macro_f1'],
+                  'development_improvement': measured['macro_f1'] > incumbent['macro_f1'],
+                  'provenance': provenance, 'implementation_sha256': digest(__file__),
+                  'feature_implementation_sha256': digest(Path(__file__).with_name('espfi_generalize.py')),
+                  'baseline_metrics_sha256': digest(Path(baseline_directory) / 'metrics.json'),
+                  'baseline_model_sha256': digest(Path(baseline_directory) / 'model.safetensors'),
+                  'model_sha256': digest(root / 'model.safetensors'), 'model_bytes': (root / 'model.safetensors').stat().st_size,
+                  'stored_scalar_count': sum(v.size for v in state.values()),
+                  'classifier_scalar_count': state['weights'].size + state['bias'].size,
+                  'random_projection_scalar_count': sum(state[k].size for k in ('projection', 'phase') if k in state),
+                  'reload_max_abs_logit_error': 0.0,
+                  'latency': {name: {'p50_ms': float(np.percentile(t, 50)), 'p95_ms': float(np.percentile(t, 95))} for name, t in timings.items()},
+                  'timing_scope': '500 paired CPU batch1 calls, input/output checks plus features/classifier; excludes IO and temporal pooling',
+                  'test_evaluation': 'Final-test arrays not deserialized or predictions computed',
+                  'limitations': ['Adaptive development result, not independent generalization evidence', 'No comparable SOTA claim; new untouched room/participant required'],
+                  'elapsed_seconds': time.monotonic() - start, 'max_seconds': max_seconds,
+                  'gpu_used': False, 'incremental_gpu_spend_usd': 0}
+        write_json(root / 'metrics.json', report)
+        write_json(root / 'status.json', {'status': 'complete', 'promotionAuthorized': False})
+        return report
+    except Exception:
+        write_json(root / 'status.json', {'status': 'failed', 'promotionAuthorized': False})
+        raise
+
+
+def main(argv=None):
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--manifest', required=True)
+    parser.add_argument('--baseline', required=True)
+    parser.add_argument('--output', required=True)
+    parser.add_argument('--max-seconds', type=int, default=600)
+    args = parser.parse_args(argv)
+    from .espfi import load_development
+    parts, metadata = load_development(args.manifest)
+    report = search(parts, metadata, args.baseline, args.output, args.max_seconds)
+    print(json.dumps({k: report[k] for k in ('selected', 'development', 'incumbent_development', 'development_macro_f1_delta', 'latency', 'model_bytes')}, indent=2))

scripts/improve-espfi-kernel.py

@@ -0,0 +1,10 @@
+#!/usr/bin/env python3
+"""Bounded nonlinear development experiment; never reads final-test arrays."""
+import os
+os.environ['OPENBLAS_NUM_THREADS'] = '1'
+os.environ['OMP_NUM_THREADS'] = '1'
+os.environ['MKL_NUM_THREADS'] = '1'
+from csi_training.espfi_kernel import main
+
+if __name__ == '__main__':
+    main()

scripts/tests/test_espfi_kernel.py

@@ -0,0 +1,66 @@
+"""Synthetic nonlinear-search integrity; not sensing accuracy evidence."""
+import copy
+from pathlib import Path
+import sys
+import tempfile
+import unittest
+import numpy as np
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+from csi_training.espfi_kernel import GRID, KernelBundle, fit, predict_features, search
+from csi_training.espfi_generalize import compare_development
+from csi_training.espfi_train import optimize
+from test_espfi_generalize import fixture
+
+
+class KernelTests(unittest.TestCase):
+    def test_frozen_grid_and_deterministic_random_projection(self):
+        self.assertEqual(len(GRID), 56)
+        x = np.random.default_rng(1).normal(size=(28, 468))
+        y = np.arange(28) % 7
+        candidate = next(c for c in GRID if c['kind'] == 'rff')
+        first, second = fit(x, y, candidate), fit(x, y, candidate)
+        for key in first:
+            np.testing.assert_array_equal(first[key], second[key])
+        np.testing.assert_allclose(first['feature_mean'], x.mean(0))
+        self.assertEqual(first['projection'].shape, (468, 128))
+        self.assertEqual(predict_features(first, x).shape, (28, 7))
+        self.assertNotIn('support_vectors', first)
+
+    def test_centroid_agrees_with_explicit_distances(self):
+        x = np.random.default_rng(2).normal(size=(28, 468))
+        y = np.arange(28) % 7
+        state = fit(x, y, {'kind': 'centroid'})
+        z = (x - state['feature_mean']) / state['feature_scale']
+        means = np.stack([z[y == i].mean(0) for i in range(7)])
+        explicit = np.square(z[:, None, :] - means[None, :, :]).sum(-1).argmin(1)
+        np.testing.assert_array_equal(explicit, predict_features(state, x).argmax(1))
+
+    def test_external_development_cannot_select_and_export_is_verified(self):
+        parts, metadata = fixture()
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            initial = copy.deepcopy(parts)
+            initial['test'] = copy.deepcopy(parts['validation'])
+            optimize(initial, metadata, root / 'initial', epochs=1, variants=('statistics',))
+            compare_development(parts, metadata, root / 'initial', root / 'incumbent')
+            first = search(parts, metadata, root / 'incumbent', root / 'first')
+            mutated = copy.deepcopy(parts)
+            mutated['validation']['x'] *= 2
+            mutated['validation']['y'] = (mutated['validation']['y'] + 1) % 7
+            mutated['validation']['identity'][:, 2] = mutated['validation']['y'] + 1
+            second = search(mutated, metadata, root / 'incumbent', root / 'second')
+            self.assertEqual(first['selected'], second['selected'])
+            self.assertEqual(first['model_sha256'], second['model_sha256'])
+            self.assertEqual(len(first['candidates']), 56)
+            bundle = KernelBundle(root / 'first')
+            self.assertEqual(bundle.predict(parts['validation']['x']).shape, (7, 7))
+            self.assertFalse(first['promotionAuthorized'])
+            self.assertNotIn('test', first)
+            with open(root / 'first' / 'model.safetensors', 'ab') as stream:
+                stream.write(b'bad')
+            with self.assertRaisesRegex(ValueError, 'hash mismatch'):
+                KernelBundle(root / 'first')
+
+
+if __name__ == '__main__':
+    unittest.main()