Reject changed manifests before dataset decode

docs/adr/ADR-427-enforced-confirmation-training-mode.md

@@ -27,7 +27,9 @@ manifest and match; missing provenance blocks the gate. Trial number is not
 treated as a physical session. This binds supplied declarations to the exact
 records the supervised loader will consume without assigning activity semantics. Invalid
 input exits 1; valid but blocked evidence exits 2.
-The dataset loader's returned manifest digest is checked again before training.
+The dataset loader parses and hashes the manifest through one checked descriptor,
+then compares that digest before resolving or decoding the referenced dataset
+bundle. Its returned digest is checked again before training.
 Confirmation mode cannot optimize an existing bundle and every receipt remains
 `promotionAuthorized: false`.
 
@@ -42,9 +44,10 @@ This closes accidental gate bypass within the selected mode and binds the
 declared review to exact manifest bytes and retained record identities. It does not authenticate captures,
 consent, licenses, aliases, stable physical identities, or exposure-ledger
 completeness. The inputs are unsigned. The manifest hash binds declarations and
-referenced payload hashes, not provenance truth. A concurrent manifest change
-may be decoded before the loader's digest mismatch aborts, although it cannot
-reach training. Passing software tests makes no sensing-accuracy claim.
+referenced payload hashes, not provenance truth. Parent directories remain a
+trusted deployment boundary, and the separately referenced NPZ still has its
+own digest/open race boundary. Passing software tests makes no sensing-accuracy
+claim.
 
 The current ESP-Fi adapter pins the canonical dataset name and source revision,
 but its upstream records do not establish physical session or device identity.

scripts/csi_training/espfi.py

@@ -143,9 +143,10 @@ def prepare(archive, output, pool=5):
     return path
 
 
-def load_dataset(manifest):
+def load_dataset(manifest, expected_manifest_sha256=None):
     """Load all partitions for a final frozen evaluation, including test signals."""
-    return _load_partitions(manifest, ('train', 'validation', 'test'))
+    return _load_partitions(manifest, ('train', 'validation', 'test'),
+                            expected_manifest_sha256)
 
 
 def load_development(manifest):
@@ -158,11 +159,25 @@ def load_development(manifest):
     return _load_partitions(manifest, ('train', 'validation'))
 
 
-def _load_partitions(manifest, splits):
-    path = Path(manifest).resolve()
-    if path.stat().st_size > 3_000_000:
-        raise ValueError('manifest too large')
-    meta = json.loads(path.read_text())
+def _load_partitions(manifest, splits, expected_manifest_sha256=None):
+    supplied_path = Path(manifest)
+    if expected_manifest_sha256 is not None:
+        from .confirmation import (MAX_DATASET_MANIFEST_BYTES, digest as checked_digest,
+                                   read_json_and_sha256)
+        checked_digest(expected_manifest_sha256)
+        meta, manifest_sha256 = read_json_and_sha256(
+            supplied_path, MAX_DATASET_MANIFEST_BYTES)
+        if manifest_sha256 != expected_manifest_sha256:
+            raise ValueError('Dataset manifest changed after confirmation preflight')
+        # The final manifest component is not reopened. Parent directories remain
+        # a trusted deployment boundary for resolving the referenced NPZ.
+        path = supplied_path.parent.resolve() / supplied_path.name
+    else:
+        path = supplied_path.resolve()
+        if path.stat().st_size > 3_000_000:
+            raise ValueError('manifest too large')
+        meta = json.loads(path.read_text())
+        manifest_sha256 = digest(path)
     if meta.get('schema_version') != 1 or meta.get('classes') != CLASSES:
         raise ValueError('unsupported dataset contract')
     if (meta.get('dataset') != DATASET or meta.get('source_revision') != REVISION or meta.get('archive_sha256') != ARCHIVE_SHA256
@@ -248,7 +263,7 @@ def _load_partitions(manifest, splits):
                        for s in ('train', 'validation', 'test')}
     if meta.get('counts') != expected_counts:
         raise ValueError('split counts mismatch')
-    meta['manifest_sha256'] = digest(path)
+    meta['manifest_sha256'] = manifest_sha256
     meta['loaded_partitions'] = list(splits)
     return parts, meta
 

scripts/csi_training/espfi_train.py

@@ -391,7 +391,10 @@ def main(argv=None):
             print(json.dumps(preflight, sort_keys=True))
             raise SystemExit(2)
     from .espfi import load_dataset
-    partitions, provenance = load_dataset(args.manifest)
+    partitions, provenance = load_dataset(
+        args.manifest,
+        expected_manifest_sha256=(preflight['dataset_manifest_sha256']
+                                  if preflight else None))
     if preflight:
         if provenance.get('manifest_sha256') != preflight['dataset_manifest_sha256']:
             raise ValueError('Dataset manifest changed after confirmation preflight')

scripts/tests/test_espfi.py

@@ -105,6 +105,31 @@ class EspfiTests(unittest.TestCase):
             with self.assertRaisesRegex(ValueError, 'provenance'):
                 espfi.load_dataset(manifest)
 
+    def test_expected_manifest_hash_checked_before_bundle_decode(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            manifest, _, _, _ = self.fixture(tmp)
+            expected = espfi.digest(manifest)
+            _, provenance = espfi.load_dataset(
+                manifest, expected_manifest_sha256=expected)
+            self.assertEqual(provenance['manifest_sha256'], expected)
+            manifest.write_bytes(manifest.read_bytes() + b' ')
+            with patch('csi_training.espfi.zipfile.ZipFile') as archive, \
+                 patch('csi_training.espfi.np.load') as load:
+                with self.assertRaisesRegex(ValueError, 'changed'):
+                    espfi.load_dataset(manifest, expected_manifest_sha256=expected)
+                archive.assert_not_called();load.assert_not_called()
+
+    def test_expected_manifest_hash_rejects_symlink_before_bundle_decode(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            manifest, _, _, _ = self.fixture(tmp)
+            expected = espfi.digest(manifest)
+            link = Path(tmp) / 'manifest-link.json'
+            link.symlink_to(manifest)
+            with patch('csi_training.espfi.np.load') as load:
+                with self.assertRaises(OSError):
+                    espfi.load_dataset(link, expected_manifest_sha256=expected)
+                load.assert_not_called()
+
     def test_development_never_deserializes_test_arrays(self):
         with tempfile.TemporaryDirectory() as tmp:
             manifest, _, _, _ = self.fixture(tmp)

scripts/tests/test_espfi_train.py

@@ -242,7 +242,9 @@ class ConfirmationGateIntegrationTests(unittest.TestCase):
                  patch('csi_training.espfi.load_dataset',return_value=(parts,provenance)) as load, \
                  patch('csi_training.espfi_train.optimize',return_value=report) as train:
                 main(self.argv(root,dataset,plan,ledger))
-                load.assert_called_once();train.assert_called_once()
+                load.assert_called_once_with(
+                    str(dataset), expected_manifest_sha256=regular_sha256(dataset))
+                train.assert_called_once()
                 attached=train.call_args.args[1]['confirmation_preflight']
                 self.assertTrue(attached['preflight_passed'])
                 self.assertFalse(attached['promotionAuthorized'])