feat(training): add reproducible CSI encoder harness and evidence contracts

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

@@ -0,0 +1,36 @@
+name: CSI encoder training harness
+
+on:
+  pull_request:
+    paths:
+      - 'scripts/csi_training/**'
+      - 'scripts/train-csi-encoder.py'
+      - 'scripts/requirements-csi-training.txt'
+      - 'scripts/tests/test_csi_training*.py'
+      - '.github/workflows/csi-encoder-training.yml'
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+jobs:
+  synthetic-cpu-contract:
+    runs-on: ubuntu-latest
+    timeout-minutes: 15
+    steps:
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1
+        with:
+          persist-credentials: false
+      - uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1
+        with:
+          python-version: '3.12'
+      - name: Install pinned CPU training dependencies
+        run: |
+          python -m pip install torch==2.11.0 --index-url https://download.pytorch.org/whl/cpu
+          python -m pip install -r scripts/requirements-csi-training.txt
+      - name: Validate real training code using temporary synthetic data
+        env:
+          PYTHONPATH: scripts
+        run: python -m unittest discover -s scripts/tests -p 'test_csi_training*.py' -v
+      - name: Enforce capture data policy
+        run: bash scripts/csi-data-policy-check.sh --tracked

docs/adr/ADR-419-reproducible-csi-encoder-training.md

@@ -0,0 +1,76 @@
+# ADR-419: Reproducible CSI encoder training from authorized capture manifests
+
+Status: Proposed, implementation supplied for review
+
+Date: 2026-10-12
+
+Related: ADR-024, ADR-299, ADR-371, ADR-414, ADR-415
+
+## Problem
+
+The published `ruvnet/wifi-densepose-pretrained` v2 encoder points to a trainer
+that is absent from the current repository. The existing JavaScript trainer
+produces a separate artifact family. Neither an upload date nor a model card
+score establishes that a newer capture is available or suitable for training.
+The published metric describes temporal embeddings, not presence or pose.
+
+## Decision
+
+Add an isolated Python CLI at `scripts/train-csi-encoder.py`. It does not change
+the npm operator surface, production sensing, or a deployed model.
+
+1. Admit local feature captures using a versioned manifest with complete file
+   SHA256, declared UTC capture date, room, session, explicit partition,
+   real/synthetic evidence class, and affirmative permission to use the data.
+   Permission is an operator assertion, not independently verified consent.
+2. Require disjoint rooms by default, or explicitly choose disjoint sessions.
+   Session separation alone is not evidence of generalization across rooms.
+   Reject duplicate files, copied observations across partitions, nonfinite
+   features, malformed records, and oversized input. Raw CSI requires a
+   separately reviewed feature adapter; do not fabricate features or labels.
+3. Fit all standardization statistics on training observations only. Train an
+   8 to 64 to 128 encoder with temporal contrastive pairs. Positives are within
+   two seconds of the same session and node; negatives are more than thirty
+   seconds away. No arbitrary batch row becomes a negative.
+4. Select the checkpoint using validation only. Evaluate the final frozen
+   checkpoint and raw, standardized, and random encoder controls using the same
+   test triplets. Reusing test results for another experiment invalidates an
+   untouched acceptance claim and requires an external holdout policy.
+5. Export safetensors, preprocessing, file hashes, actual dependency versions,
+   configuration, data manifest digest, and honest metrics. Refuse output
+   directory reuse. Failed or interrupted work cannot represent completion.
+6. All runs retain `promotionAuthorized=false`. Temporal proxy improvements
+   never automatically qualify occupancy, vitals, pose, or an edge deployment.
+
+## Execution and boundaries
+
+ruOS can validate manifests and execute the CPU trainer. A separately
+authorized Vast machine can execute the same CLI with CUDA and a protected
+local dataset mount. The harness neither rents machines nor transfers data,
+reads vault credentials, publishes artifacts, or promotes candidates. A
+cooperative runtime limit is not a provider billing cap or instance teardown;
+operators must separately bound and end any GPU rental.
+
+Slack can carry private handoff metadata. Public federation messages must be
+limited to code coordination and public evidence. A failed coordination tool
+is not a successful claim, agent execution, or training result.
+
+## Alternatives
+
+Extending the existing pose gate would conflate pose and embedding metrics.
+Recovering deleted CSI from Git history conflicts with ADR-299's incident
+boundary. An unrestricted GPU sweep before resolving provenance would spend
+money without producing a defensible comparison. These options are rejected.
+
+## Validation and acceptance
+
+The focused unittest suite exercises actual CPU gradients, deterministic replay,
+artifact loading, current collector compatibility, partition leakage, malformed
+input and failure paths. Fixtures are generated temporarily and explicitly
+synthetic; they establish software behavior only. No raw captures enter Git or CI.
+
+Real model improvement remains unverified until an authorized manifest and
+adequate independent capture groups are available. A separate deployment gate
+must compare downstream task performance on untouched rooms, quantization
+parity, and measured device RAM and latency. A 4-bit file size alone proves
+none of those properties. Keep the existing deployed model for rollback.

docs/csi-encoder-training.md

@@ -0,0 +1,104 @@
+# Reproducible CSI encoder training
+
+This harness trains a small temporal embedding model from explicit eight-value
+CSI features. It is not a pose, presence, or vital-sign classifier. It produces
+local candidates only, with no automatic publishing or promotion.
+
+## Set up
+
+Use Python 3.11 or 3.12 and an isolated environment. For the tested CPU setup:
+
+```sh
+python3 -m venv .venv-csi
+. .venv-csi/bin/activate
+python -m pip install torch==2.11.0 --index-url https://download.pytorch.org/whl/cpu
+python -m pip install -r scripts/requirements-csi-training.txt
+PYTHONPATH=scripts python -m unittest discover -s scripts/tests -p 'test_csi_training*.py' -v
+```
+
+Install an appropriate CUDA build of the same PyTorch version for an authorized
+GPU host. Record the resolved environment with `python -m pip freeze` alongside
+the private run. CPU and GPU results need not be bit-identical. A requested CUDA
+run fails when CUDA is unavailable instead of silently falling back to CPU.
+
+## Prepare private data
+
+Create a private JSON manifest with `schema_version: 1` and a `files` array.
+Each entry has the following fields:
+
+| Field | Contract |
+| --- | --- |
+| `path` | Local regular file, relative to the manifest or absolute |
+| `sha256` | Exact lowercase SHA256 of the complete file |
+| `captured_at` | Actual declared capture time in UTC ISO 8601, not file modification or model upload time |
+| `room_id` | Stable pseudonymous room identifier |
+| `session_id` | Stable capture session identifier |
+| `split` | `train`, `validation`, or `test` |
+| `evidence` | `real` or `synthetic` |
+| `use_permitted` | Explicit boolean `true`, asserted by the authorized operator |
+
+Keep producer revision, sensor configuration, consent, permitted execution sites,
+label provenance and schema documentation with the private source manifest.
+The CLI validates structural evidence; it cannot prove those operator assertions.
+Never mark a capture usable solely because someone mentioned it in a message.
+
+Canonical JSONL rows contain `timestamp` in seconds, `node_id`, and `features`
+as eight finite numbers. The explicit legacy feature stream and the current
+collector feature envelope are supported. Raw subcarrier amplitudes are not
+silently repurposed as features. Missing labels are never synthesized.
+
+Room separation is the default and requires at least three independent rooms
+across training, validation and test. `--group-by session_id` allows distinct
+sessions in the same room, but cannot substantiate performance in unseen rooms.
+Each partition needs same-node/session positive pairs within two seconds and
+negative observations more than thirty seconds apart. The training set alone
+determines preprocessing statistics.
+
+## Inspect and train
+
+```sh
+python scripts/train-csi-encoder.py inspect --manifest /private/csi/manifest.json
+python scripts/train-csi-encoder.py train \
+  --manifest /private/csi/manifest.json \
+  --output /private/runs/csi-candidate-001 \
+  --device cpu --epochs 20 --seed 42 --max-seconds 600
+```
+
+For synthetic software experiments, add `--allow-synthetic`; results retain
+the `SYNTHETIC` evidence label. Do not use those results as sensor benchmarks.
+The output directory must not already exist. No input is changed or uploaded.
+
+For Vast, first provision an authorized isolated host, check its hourly price,
+bind only the approved private dataset and pin the code commit and environment.
+Then use the same command with `--device cuda`. The CLI does not access provider
+keys, create instances, copy private captures, or stop the rental. Its cooperative
+runtime limit must be backed by the operator's process and instance deadline.
+
+## Review the candidate
+
+The output includes `model.safetensors`, `preprocessing.json`, `metrics.json`,
+and `status.json`. Check completion, manifest digest, implementation source
+digests and complete file hashes.
+Load preprocessing together with the model; never feed unstandardized features
+directly into the network. Validation selects a checkpoint; the final test
+comparison uses identical triplets for the candidate and controls.
+
+The loader applies the saved training statistics and verifies completion and
+artifact hashes. Hashes detect corruption; they are not a trusted signature.
+
+```python
+# Run with PYTHONPATH=scripts. raw_features has shape [N, 8].
+from csi_training.model import load_bundle
+
+encoder = load_bundle('/private/runs/csi-candidate-001')
+embeddings = encoder.encode(raw_features)
+```
+
+`promotionAuthorized` remains false even if the temporal metric improves.
+Temporal proximity is a proxy, and shared frame structure can make it easy.
+Repeated test-driven tuning also contaminates a test set. Qualify downstream
+tasks on fresh independent rooms and sessions before deployment. Quantization,
+device RAM, latency, and compatibility with the production runtime remain
+separate measured gates.
+
+See [ADR-419](adr/ADR-419-reproducible-csi-encoder-training.md) for boundaries.

scripts/csi_training/__init__.py

@@ -0,0 +1 @@
+"""Reproducible CSI feature training; no implicit raw CSI conversion."""

scripts/csi_training/data.py

@@ -0,0 +1,257 @@
+"""Bounded, hash verified CSI feature intake and explicit grouped partitions.
+
+Manifest v1 has a ``files`` list. Each entry supplies path, sha256,
+captured_at (UTC ISO 8601), room_id, session_id, split, evidence and
+use_permitted. Capture dates are declarations, not inferred from file mtimes.
+"""
+from dataclasses import dataclass
+from datetime import datetime, timezone
+import hashlib
+import json
+import math
+from pathlib import Path
+import re
+import stat
+
+import numpy as np
+
+SPLITS = ("train", "validation", "test")
+MAX_FILE_BYTES = 256 * 1024 * 1024
+MAX_TOTAL_BYTES = 512 * 1024 * 1024
+MAX_MANIFEST_BYTES = 1024 * 1024
+MAX_LINE_BYTES = 65536
+MAX_FRAMES = 500000
+MAX_INPUT_LINES = 1000000
+
+
+class DataError(ValueError):
+    """Input does not satisfy the training evidence contract."""
+
+
+@dataclass
+class Partition:
+    features: np.ndarray
+    timestamps: np.ndarray
+    node_ids: list
+    groups: list
+    session_ids: list
+
+
+@dataclass
+class Dataset:
+    partitions: dict
+    inventory: list
+    manifest_sha256: str
+    group_by: str
+
+
+@dataclass
+class Standardizer:
+    mean: np.ndarray
+    scale: np.ndarray
+
+    def transform(self, features):
+        values = np.asarray(features, dtype=np.float64)
+        if values.ndim != 2 or values.shape[1] != 8 or not np.isfinite(values).all():
+            raise DataError("standardizer requires finite Nx8 features")
+        result = (values - self.mean) / self.scale
+        if not np.isfinite(result).all():
+            raise DataError("standardization overflow")
+        return result.astype(np.float32)
+
+
+def fit_standardizer(train_features):
+    values = np.asarray(train_features, dtype=np.float64)
+    if values.ndim != 2 or values.shape[1] != 8 or len(values) < 2 or not np.isfinite(values).all():
+        raise DataError("fit requires at least two finite training feature rows")
+    mean, scale = values.mean(axis=0), values.std(axis=0)
+    if not np.isfinite(mean).all() or not np.isfinite(scale).all():
+        raise DataError("training statistics overflow")
+    return Standardizer(mean, np.where(scale < 1e-8, 1.0, scale))
+
+
+def _json(text):
+    def pairs(items):
+        result = {}
+        for key, value in items:
+            if key in result:
+                raise DataError("duplicate JSON key")
+            result[key] = value
+        return result
+    try:
+        return json.loads(text, object_pairs_hook=pairs,
+                          parse_constant=lambda _: (_ for _ in ()).throw(DataError("nonfinite JSON number")))
+    except (ValueError, UnicodeError) as exc:
+        raise DataError("invalid JSON input") from exc
+
+
+def _regular(path, maximum):
+    if path.is_symlink():
+        raise DataError("symlink input is not permitted")
+    try:
+        info = path.stat()
+    except OSError as exc:
+        raise DataError("input file unavailable") from exc
+    if not stat.S_ISREG(info.st_mode) or info.st_size > maximum:
+        raise DataError("input must be a bounded regular file")
+    return info.st_size
+
+
+def _text(value, name):
+    if not isinstance(value, str) or not value.strip() or len(value) > 256 or any(ord(c) < 32 for c in value):
+        raise DataError(f"invalid {name}")
+    return value
+
+
+def _manifest(path):
+    path = Path(path).absolute()
+    _regular(path, MAX_MANIFEST_BYTES)
+    raw = path.read_bytes()
+    if len(raw) > MAX_MANIFEST_BYTES:
+        raise DataError("manifest exceeds limit")
+    obj = _json(raw)
+    if not isinstance(obj, dict) or type(obj.get("schema_version")) is not int or obj["schema_version"] != 1:
+        raise DataError("manifest schema_version must equal 1")
+    entries = obj.get("files")
+    if not isinstance(entries, list) or not entries or len(entries) > 1024:
+        raise DataError("manifest files must contain 1 to 1024 entries")
+    checked, digests, total = [], set(), 0
+    for index, entry in enumerate(entries):
+        if not isinstance(entry, dict):
+            raise DataError("manifest file entry must be an object")
+        for key in ("room_id", "session_id", "path"):
+            _text(entry.get(key), key)
+        digest = entry.get("sha256")
+        if not isinstance(digest, str) or not re.fullmatch(r"[0-9a-f]{64}", digest):
+            raise DataError("sha256 must be a lowercase hex digest")
+        if digest in digests:
+            raise DataError("duplicate file hash")
+        digests.add(digest)
+        if entry.get("split") not in SPLITS or entry.get("evidence") not in ("real", "synthetic"):
+            raise DataError("invalid split or evidence")
+        if entry.get("use_permitted") is not True:
+            raise DataError("explicit use_permitted=true is required")
+        captured = _text(entry.get("captured_at"), "captured_at")
+        try:
+            date = datetime.fromisoformat(captured.replace("Z", "+00:00"))
+        except ValueError as exc:
+            raise DataError("invalid captured_at") from exc
+        if date.tzinfo is None or date.utcoffset() != timezone.utc.utcoffset(date):
+            raise DataError("captured_at must specify UTC")
+        source = path.parent / entry["path"]
+        total += _regular(source, MAX_FILE_BYTES)
+        if total > MAX_TOTAL_BYTES:
+            raise DataError("total dataset exceeds byte limit")
+        checked.append((dict(entry, captured_at=date.isoformat()), source, index))
+    return checked, hashlib.sha256(raw).hexdigest()
+
+
+def discover(manifest):
+    """Inventory declared capture dates; load_manifest verifies hashes and frames."""
+    entries, _ = _manifest(manifest)
+    return sorted([dict(captured_at=e["captured_at"], sha256=e["sha256"],
+                        split=e["split"], evidence=e["evidence"],
+                        room_id=e["room_id"], session_id=e["session_id"])
+                   for e, _, _ in entries], key=lambda e: datetime.fromisoformat(e["captured_at"]), reverse=True)
+
+
+def _number(value):
+    if type(value) not in (int, float) or abs(value) > 1e20 or not math.isfinite(value):
+        raise DataError("numeric values must be finite and bounded")
+    return 0.0 if value == 0 else float(value)
+
+
+def load_manifest(manifest, group_by="room_id", allow_synthetic=False):
+    """Validate partitions without deriving labels or creating raw CSI features."""
+    if group_by not in ("room_id", "session_id"):
+        raise DataError("group_by must be room_id or session_id")
+    entries, manifest_digest = _manifest(manifest)
+    owners, session_owners, fingerprints, rows = {}, {}, {}, {s: [] for s in SPLITS}
+    seen_observations = set()
+    frame_count = 0
+    input_lines = 0
+    for entry, source, index in entries:
+        split, group = entry["split"], entry[group_by]
+        if entry["evidence"] == "synthetic" and not allow_synthetic:
+            raise DataError("synthetic data requires explicit allow_synthetic")
+        for mapping, key in ((owners, group), (session_owners, entry["session_id"])):
+            if key in mapping and mapping[key] != split:
+                raise DataError("group or session appears across splits")
+            mapping[key] = split
+        digest, last, count, size = hashlib.sha256(), {}, 0, 0
+        with source.open("rb") as stream:
+            while True:
+                line = stream.readline(MAX_LINE_BYTES + 1)
+                if not line:
+                    break
+                input_lines += 1
+                if input_lines > MAX_INPUT_LINES:
+                    raise DataError("dataset exceeds input line limit")
+                size += len(line)
+                if len(line) > MAX_LINE_BYTES or size > MAX_FILE_BYTES:
+                    raise DataError("capture line or file exceeds limit")
+                digest.update(line)
+                if not line.strip():
+                    continue
+                row = _json(line)
+                if not isinstance(row, dict):
+                    raise DataError("capture row must be an object")
+                # Mixed legacy recordings are supported only for explicit types.
+                if row.get("type") in ("vitals", "raw_csi"):
+                    continue
+                if row.get("type") not in (None, "feature", "features"):
+                    raise DataError("unsupported capture row type")
+                features = row.get("features")
+                node = row.get("node_id")
+                if isinstance(features, dict):
+                    # CsiRecorder.write_frame removes the packet type and places
+                    # explicit packet features/node_id in this nested envelope.
+                    # Outer timestamp is already seconds; timestamp_us is metadata.
+                    if row.get("type") is not None or features.get("type") not in (None, "feature", "features"):
+                        raise DataError("unsupported collector envelope")
+                    node = features.get("node_id")
+                    features = features.get("features")
+                if not isinstance(features, list) or len(features) != 8:
+                    raise DataError("feature rows require eight explicit features")
+                features = [_number(v) for v in features]
+                timestamp = _number(row.get("timestamp"))
+                if type(node) is int:
+                    if node < 0 or node > 2**32 - 1:
+                        raise DataError("invalid node_id")
+                    node = str(node)
+                else:
+                    node = _text(node, "node_id")
+                observation = (entry["session_id"], node, timestamp)
+                if observation in seen_observations:
+                    raise DataError("duplicate session/node/timestamp observation")
+                seen_observations.add(observation)
+                if node in last and timestamp < last[node]:
+                    raise DataError("timestamps must be monotonic per node")
+                last[node] = timestamp
+                # Bounded copy heuristic, not proof of absence of leakage.
+                # Idle/quantized vectors can legitimately recur at distinct times.
+                signature = json.dumps([timestamp, node, features], separators=(",", ":")).encode()
+                fingerprint = hashlib.sha256(signature).digest()
+                if fingerprint in fingerprints and fingerprints[fingerprint] != split:
+                    raise DataError("duplicate feature frame across splits")
+                fingerprints[fingerprint] = split
+                rows[split].append((features, timestamp, node, group, entry["session_id"]))
+                count += 1
+                frame_count += 1
+                if frame_count > MAX_FRAMES:
+                    raise DataError("dataset exceeds frame limit")
+        if digest.hexdigest() != entry["sha256"]:
+            raise DataError(f"file hash mismatch at manifest entry {index}")
+        if count == 0:
+            raise DataError("capture has no explicit feature rows; raw CSI requires a reviewed adapter")
+    partitions = {}
+    for split, values in rows.items():
+        if len(values) < 2:
+            raise DataError(f"partition {split} requires at least two feature rows")
+        features, timestamps, nodes, groups, sessions = zip(*values)
+        partitions[split] = Partition(np.asarray(features, dtype=np.float64),
+                                      np.asarray(timestamps, dtype=np.float64),
+                                      list(nodes), list(groups), list(sessions))
+    inventory = sorted([{k: e[k] for k in ("captured_at", "sha256", "split", "evidence", "room_id", "session_id")}
+                        for e, _, _ in entries], key=lambda e: datetime.fromisoformat(e["captured_at"]), reverse=True)
+    return Dataset(partitions, inventory, manifest_digest, group_by)

scripts/csi_training/model.py

@@ -0,0 +1,83 @@
+"""Small temporal CSI encoder; not a pose or occupancy classifier."""
+import torch
+from torch import nn
+from torch.nn import functional as F
+
+
+class CSIEncoder(nn.Module):
+    """8 -> 64 -> 128, batch normalization, GELU and unit embeddings."""
+    def __init__(self):
+        super().__init__()
+        self.w1 = nn.Linear(8, 64)
+        self.bn1 = nn.BatchNorm1d(64)
+        self.w2 = nn.Linear(64, 128)
+        self.bn2 = nn.BatchNorm1d(128)
+
+    def forward(self, x):
+        return F.normalize(self.bn2(self.w2(F.gelu(self.bn1(self.w1(x))))), dim=-1)
+
+
+class PreprocessedEncoder:
+    """Verified local candidate loader; includes train-fitted normalization.
+
+    Hashes detect accidental artifact corruption. They are not signatures and
+    do not authenticate an untrusted directory or authorize model promotion.
+    """
+    def __init__(self, directory, device='cpu'):
+        import hashlib
+        import json
+        from pathlib import Path
+        import numpy as np
+        from safetensors.torch import load_file
+
+        root = Path(directory)
+        paths = {name: root / name for name in ('status.json', 'metrics.json', 'preprocessing.json', 'model.safetensors')}
+        for name, path in paths.items():
+            if path.is_symlink() or not path.is_file() or path.stat().st_size > 2 * 1024 * 1024:
+                raise ValueError(f'Invalid bounded artifact: {name}')
+        status = json.loads(paths['status.json'].read_text())
+        if not isinstance(status, dict) or status.get('status') != 'complete' or status.get('promotionAuthorized') is not False:
+            raise ValueError('Candidate lifecycle is not complete and proposal-only')
+        report = json.loads(paths['metrics.json'].read_text())
+        if not isinstance(report, dict):
+            raise ValueError('Invalid metrics schema')
+        if report.get('architecture') != 'Linear(8,64),BN64,GELU,Linear(64,128),BN128,L2':
+            raise ValueError('Unsupported encoder architecture')
+        if report.get('status') != 'complete' or report.get('promotionAuthorized') is not False:
+            raise ValueError('Candidate is not complete')
+        for filename, key in [('model.safetensors', 'model_sha256'), ('preprocessing.json', 'preprocessing_sha256')]:
+            if hashlib.sha256(paths[filename].read_bytes()).hexdigest() != report.get(key):
+                raise ValueError(f'Artifact hash mismatch: {filename}')
+        prep = json.loads(paths['preprocessing.json'].read_text())
+        if not isinstance(prep, dict):
+            raise ValueError('Invalid preprocessing schema')
+        mean, scale = np.asarray(prep['mean'], dtype=np.float64), np.asarray(prep['scale'], dtype=np.float64)
+        if (mean.shape != (8,) or scale.shape != (8,) or not np.isfinite(mean).all()
+                or not np.isfinite(scale).all() or not (scale > 0).all()
+                or prep.get('fit_partition') != 'train' or prep.get('feature_count') != 8):
+            raise ValueError('Invalid preprocessing contract')
+        self.mean, self.scale, self.device = mean, scale, device
+        self.model = CSIEncoder().to(device).eval()
+        self.model.load_state_dict(load_file(str(paths['model.safetensors']), device=device), strict=True)
+        if any(not torch.isfinite(v).all() for v in self.model.state_dict().values()):
+            raise ValueError('Non-finite model weights')
+
+    def encode(self, features):
+        import numpy as np
+        values = np.asarray(features, dtype=np.float64)
+        if values.ndim != 2 or values.shape[1] != 8 or not 0 < len(values) <= 4096 or not np.isfinite(values).all():
+            raise ValueError('Expected 1 to 4096 finite eight-feature rows')
+        with np.errstate(over='ignore', invalid='ignore'):
+            standardized = ((values - self.mean) / self.scale).astype(np.float32)
+        if not np.isfinite(standardized).all():
+            raise ValueError('Input normalization overflow')
+        with torch.no_grad():
+            result = self.model(torch.from_numpy(standardized).to(self.device)).cpu().numpy()
+        if not np.isfinite(result).all():
+            raise ValueError('Non-finite encoder output')
+        return result
+
+
+def load_bundle(directory, device='cpu'):
+    """Load a complete local bundle with verified hashes and preprocessing."""
+    return PreprocessedEncoder(directory, device=device)

scripts/csi_training/train.py

@@ -0,0 +1,258 @@
+"""Reproducible group-held-out temporal contrastive training.
+
+No remote code, model upload, provisioning, or automatic promotion. Temporal
+triplet accuracy describes embeddings only, never occupancy or pose accuracy.
+"""
+import argparse
+import hashlib
+import json
+import math
+import os
+from pathlib import Path
+import random
+import time
+
+import numpy as np
+
+
+def temporal_triplets(partition, seed, limit=4096):
+    """One positive 0<dt<=2s and one negative >30s, within each session/node.
+
+    No in-batch negatives: each anchor has exactly its explicitly screened
+    negative. This avoids silently labeling nearby windows as negative.
+    """
+    rng = np.random.default_rng(seed)
+    groups = {}
+    for i, key in enumerate(zip(partition.session_ids, partition.node_ids)):
+        groups.setdefault(key, []).append(i)
+    candidates, eligible = [], 0
+    for indices in groups.values():
+        ids = np.asarray(indices)
+        times = np.asarray(partition.timestamps)[ids]
+        order = np.argsort(times, kind='stable')
+        ids, times = ids[order], times[order]
+        # Binary search bounds avoid allocating an NxN distance matrix.
+        for j in range(len(ids)):
+            t = times[j]
+            lo = np.searchsorted(times, t - 2, side='left')
+            hi = np.searchsorted(times, t + 2, side='right')
+            equal_lo = np.searchsorted(times, t, side='left')
+            equal_hi = np.searchsorted(times, t, side='right')
+            npos_left = equal_lo - lo
+            npos = npos_left + hi - equal_hi
+            left = np.searchsorted(times, t - 30, side='left')
+            right = np.searchsorted(times, t + 30, side='right')
+            nneg = left + len(ids) - right
+            if npos > 0 and nneg:
+                n = int(rng.integers(nneg))
+                neg = ids[n] if n < left else ids[right + n - left]
+                draw = int(rng.integers(npos))
+                p = lo + draw if draw < npos_left else equal_hi + draw - npos_left
+                candidate = (ids[j], ids[p], neg)
+                # Reservoir sampling bounds candidate storage globally while
+                # giving every eligible anchor the same inclusion probability.
+                eligible += 1
+                if len(candidates) < limit:
+                    candidates.append(candidate)
+                else:
+                    slot = int(rng.integers(eligible))
+                    if slot < limit:
+                        candidates[slot] = candidate
+    if not candidates:
+        raise ValueError('Each split needs same-session/node positive pairs 0<dt<=2s and negatives >30s')
+    rng.shuffle(candidates)
+    return np.asarray(candidates[:limit], dtype=np.int64)
+
+
+def triplet_accuracy(embeddings, triplets):
+    a, p, n = (embeddings[triplets[:, i]] for i in range(3))
+    pos, neg = np.sum(a * p, axis=1), np.sum(a * n, axis=1)
+    # Ties earn half credit; constant embeddings cannot score 100%.
+    return float(np.mean((pos > neg) + 0.5 * (pos == neg)))
+
+
+def unit_rows(x):
+    return x / np.maximum(np.linalg.norm(x, axis=1, keepdims=True), 1e-12)
+
+
+def sha256(path):
+    digest = hashlib.sha256()
+    with open(path, 'rb') as f:
+        for block in iter(lambda: f.read(1024 * 1024), b''):
+            digest.update(block)
+    return digest.hexdigest()
+
+
+def write_json(path, value):
+    path.write_text(json.dumps(value, indent=2, sort_keys=True, allow_nan=False) + '\n')
+
+
+def run_training(args):
+    # Set before CUDA context creation; deterministic ops fail rather than drift.
+    os.environ.setdefault('CUBLAS_WORKSPACE_CONFIG', ':4096:8')
+    import torch
+    import safetensors
+    from safetensors.torch import save_file
+    from .data import load_manifest, fit_standardizer
+    from .model import CSIEncoder
+
+    started = time.monotonic()
+    def check_budget():
+        if time.monotonic() - started >= args.max_seconds:
+            raise TimeoutError('Training time budget exhausted; candidate is incomplete and cannot promote')
+
+    for name, low, high in [('epochs', 1, 10000), ('batch_size', 2, 4096),
+                            ('max_triplets', 2, 100000), ('max_seconds', 1, 86400)]:
+        value = getattr(args, name)
+        if not math.isfinite(value) or not low <= value <= high:
+            raise ValueError(f'{name} must be between {low} and {high}')
+    if not 0 < args.learning_rate <= 0.1 or not 0.01 <= args.temperature <= 10:
+        raise ValueError('Invalid learning rate or temperature')
+    if not 0 <= args.seed < 2**32:
+        raise ValueError('seed must be in [0, 2**32)')
+    out = Path(args.output)
+    if out.exists():
+        raise ValueError('Output must not exist; never overwrite a training run')
+    dataset = load_manifest(args.manifest, group_by=args.group_by, allow_synthetic=args.allow_synthetic)
+    scaler = fit_standardizer(dataset.partitions['train'].features)
+    triplets = {name: temporal_triplets(part, args.seed + i, args.max_triplets)
+                for i, (name, part) in enumerate(dataset.partitions.items())}
+    values = {name: scaler.transform(part.features).astype(np.float32)
+              for name, part in dataset.partitions.items()}
+    if any(not np.isfinite(x).all() for x in values.values()):
+        raise ValueError('Non-finite standardized input')
+    if args.device == 'cuda' and not torch.cuda.is_available():
+        raise ValueError('CUDA requested but unavailable; no silent CPU fallback')
+    random.seed(args.seed)
+    np.random.seed(args.seed)
+    torch.manual_seed(args.seed)
+    torch.set_num_threads(1)
+    torch.use_deterministic_algorithms(True)
+    torch.backends.cudnn.benchmark = False
+    model = CSIEncoder().to(args.device)
+    initial = {k: v.detach().cpu().clone() for k, v in model.state_dict().items()}
+    tensors = {k: torch.from_numpy(v).to(args.device) for k, v in values.items()}
+
+    def encode_partition(network, tensor):
+        chunks = []
+        for batch in tensor.split(args.batch_size):
+            check_budget()
+            chunks.append(network(batch).cpu().numpy())
+        return np.concatenate(chunks, axis=0)
+
+    def embed(network):
+        network.eval()
+        result = {}
+        with torch.no_grad():
+            for name, tensor in tensors.items():
+                check_budget()
+                result[name] = encode_partition(network, tensor)
+        return result
+
+    # Preserve initialization; evaluate the random control and candidate test
+    # embeddings only after checkpoint selection is frozen.
+    optimizer = torch.optim.AdamW(model.parameters(), lr=args.learning_rate)
+    rng = np.random.default_rng(args.seed)
+    history, best, best_epoch, best_state = [], -1.0, 0, None
+    out.mkdir(parents=True, exist_ok=False)
+    write_json(out / 'status.json', {'status': 'running', 'promotionAuthorized': False})
+    try:
+        for epoch in range(args.epochs):
+            check_budget()
+            model.train()
+            ordered = triplets['train'][rng.permutation(len(triplets['train']))]
+            losses = []
+            for start in range(0, len(ordered), args.batch_size):
+                check_budget()
+                batch = ordered[start:start + args.batch_size]
+                # Concatenation gives BN at least three samples even for tail=1.
+                flat = torch.as_tensor(batch.reshape(-1), device=args.device)
+                z = model(tensors['train'][flat]).reshape(-1, 3, 128)
+                logits = torch.stack(((z[:, 0] * z[:, 1]).sum(-1),
+                                      (z[:, 0] * z[:, 2]).sum(-1)), dim=1) / args.temperature
+                loss = torch.nn.functional.cross_entropy(logits, torch.zeros(len(batch), dtype=torch.long, device=args.device))
+                if not torch.isfinite(loss):
+                    raise ValueError('Non-finite loss')
+                optimizer.zero_grad(set_to_none=True)
+                loss.backward()
+                torch.nn.utils.clip_grad_norm_(model.parameters(), 5, error_if_nonfinite=True)
+                optimizer.step()
+                losses.append(float(loss.detach()))
+            model.eval()
+            with torch.no_grad():
+                validation = encode_partition(model, tensors['validation'])
+            score = triplet_accuracy(validation, triplets['validation'])
+            history.append({'epoch': epoch + 1, 'loss': float(np.mean(losses)), 'validation_temporal_triplet_accuracy': score})
+            if score > best:
+                best, best_epoch = score, epoch + 1
+                best_state = {k: v.detach().cpu().clone() for k, v in model.state_dict().items()}
+        check_budget()
+        model.load_state_dict(best_state)
+        learned = embed(model)
+        random_model = CSIEncoder().to(args.device)
+        random_model.load_state_dict(initial)
+        random_embeddings = embed(random_model)
+        if any(not np.isfinite(x).all() for x in learned.values()):
+            raise ValueError('Non-finite final embeddings')
+        metrics = {name: {'triplets': len(triplets[name]),
+                    'trained': triplet_accuracy(learned[name], triplets[name]),
+                    'random_encoder': triplet_accuracy(random_embeddings[name], triplets[name]),
+                    'standardized_input': triplet_accuracy(unit_rows(values[name]), triplets[name]),
+                    'raw_input': triplet_accuracy(unit_rows(dataset.partitions[name].features), triplets[name])}
+                   for name in tensors}
+        delta = sum(float((best_state[k] - initial[k]).square().sum()) for k, _ in model.named_parameters())
+        save_file(best_state, str(out / 'model.safetensors'))
+        write_json(out / 'preprocessing.json', {'mean': np.asarray(scaler.mean).tolist(), 'scale': np.asarray(scaler.scale).tolist(), 'fit_partition': 'train', 'feature_count': 8})
+        report = {'status': 'complete', 'promotionAuthorized': False,
+                  'metric': 'temporal_triplet_accuracy_not_pose_or_presence',
+                  'architecture': 'Linear(8,64),BN64,GELU,Linear(64,128),BN128,L2',
+                  'parameter_count': sum(p.numel() for p in model.parameters()),
+                  'evidence': 'SYNTHETIC' if any(e['evidence'] == 'synthetic' for e in dataset.inventory) else 'MEASURED',
+                  'best_epoch': best_epoch, 'history': history, 'metrics': metrics,
+                  'trainable_parameter_delta_squared': delta, 'group_by': args.group_by, 'seed': args.seed, 'device': args.device,
+                  'config': {k: v for k, v in vars(args).items() if k not in ('manifest', 'output')},
+                  'manifest_sha256': dataset.manifest_sha256,
+                  'source_sha256': {name: sha256(Path(__file__).parent / name)
+                                    for name in ('train.py', 'data.py', 'model.py')},
+                  'data_files': [{'sha256': e['sha256'], 'split': e['split'], 'evidence': e['evidence'], 'captured_at': e['captured_at']} for e in dataset.inventory],
+                  'model_sha256': sha256(out / 'model.safetensors'),
+                  'preprocessing_sha256': sha256(out / 'preprocessing.json'),
+                  'versions': {'torch': torch.__version__, 'numpy': np.__version__, 'safetensors': safetensors.__version__},
+                  'elapsed_seconds': time.monotonic() - started,
+                  'limitations': ['Temporal proxy does not establish task accuracy', 'No INT4 export or hardware RAM/latency claim', 'No automatic promotion or upload', 'Time budget is cooperative between batches, use an external process timeout for a hard deadline', 'Repeated test reuse requires an external frozen holdout policy']}
+        write_json(out / 'metrics.json', report)
+        write_json(out / 'status.json', {'status': 'complete', 'promotionAuthorized': False})
+        return report
+    except BaseException:
+        write_json(out / 'status.json', {'status': 'failed', 'promotionAuthorized': False})
+        raise
+
+
+def main(argv=None):
+    parser = argparse.ArgumentParser(description=__doc__)
+    sub = parser.add_subparsers(dest='command', required=True)
+    inspect = sub.add_parser('inspect', help='Validate local manifest without training')
+    train = sub.add_parser('train', help='Train a local candidate; never publish or promote')
+    for command in (inspect, train):
+        command.add_argument('--manifest', required=True)
+        command.add_argument('--group-by', choices=['room_id', 'session_id'], default='room_id')
+        command.add_argument('--allow-synthetic', action='store_true')
+    train.add_argument('--output', required=True)
+    train.add_argument('--device', choices=['cpu', 'cuda'], default='cpu')
+    train.add_argument('--epochs', type=int, default=20)
+    train.add_argument('--batch-size', type=int, default=64)
+    train.add_argument('--max-triplets', type=int, default=4096)
+    train.add_argument('--seed', type=int, default=42)
+    train.add_argument('--max-seconds', type=float, default=600)
+    train.add_argument('--learning-rate', type=float, default=0.001)
+    train.add_argument('--temperature', type=float, default=0.1)
+    args = parser.parse_args(argv)
+    try:
+        if args.command == 'inspect':
+            from .data import load_manifest
+            result = load_manifest(args.manifest, group_by=args.group_by, allow_synthetic=args.allow_synthetic).inventory
+        else:
+            result = run_training(args)
+        print(json.dumps(result, indent=2, allow_nan=False))
+    except (ValueError, OSError, TimeoutError, ImportError) as exc:
+        parser.exit(2, f'error: {exc}\n')

scripts/requirements-csi-training.txt

@@ -0,0 +1,5 @@
+# Replayed CPU integration environment. Install the appropriate CPU/CUDA torch
+# wheel using an explicit index URL; actual versions are recorded in each run.
+numpy==2.4.4
+torch==2.11.0
+safetensors==0.8.0

scripts/tests/test_csi_training_data.py

@@ -0,0 +1,222 @@
+"""Synthetic fixture contract tests, no physical sensing claims."""
+import hashlib
+import importlib.util
+import struct
+import json
+from pathlib import Path
+import sys
+import tempfile
+import unittest
+from unittest.mock import patch
+
+import numpy as np
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+from csi_training.data import DataError, discover, fit_standardizer, load_manifest
+
+
+class DataContractTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.base = Path(self.temp.name)
+        self.manifest = self.base / "manifest.json"
+        self.entries = []
+        for i, split in enumerate(("train", "validation", "test")):
+            self.write_capture(i, [dict(type="feature", timestamp=t, node_id=1,
+                                       features=[i * 10 + t + j / 10 for j in range(8)])
+                                   for t in (0, 1, 40)])
+            self.entries.append(dict(path=f"{i}.jsonl", sha256=self.sha(i),
+                                     captured_at=f"2026-10-{10+i}T00:00:00Z",
+                                     room_id=f"room{i}", session_id=f"session{i}",
+                                     split=split, evidence="synthetic", use_permitted=True))
+        self.save()
+
+    def write_capture(self, i, rows):
+        (self.base / f"{i}.jsonl").write_text("".join(json.dumps(r) + "\n" for r in rows))
+
+    def sha(self, i):
+        return hashlib.sha256((self.base / f"{i}.jsonl").read_bytes()).hexdigest()
+
+    def save(self):
+        self.manifest.write_text(json.dumps(dict(schema_version=1, files=self.entries)))
+
+    def load(self):
+        return load_manifest(self.manifest, allow_synthetic=True)
+
+    def test_load_and_train_only_standardization(self):
+        data = self.load()
+        stats = fit_standardizer(data.partitions["train"].features)
+        np.testing.assert_allclose(stats.transform(data.partitions["train"].features).mean(0), 0, atol=1e-6)
+        self.assertGreater(stats.transform(data.partitions["test"].features).mean(), 0.5)
+        self.assertEqual(data.partitions["train"].session_ids, ["session0"] * 3)
+        self.assertEqual(discover(self.manifest)[0]["session_id"], "session2")
+        self.assertNotIn("path", data.inventory[0])
+
+    def test_synthetic_requires_opt_in(self):
+        with self.assertRaisesRegex(DataError, "synthetic"):
+            load_manifest(self.manifest)
+
+    def test_hash_tampering(self):
+        with (self.base / "0.jsonl").open("a") as f:
+            f.write("\n")
+        with self.assertRaisesRegex(DataError, "hash mismatch"):
+            self.load()
+
+    def test_group_leakage(self):
+        self.entries[1]["room_id"] = "room0"
+        self.save()
+        with self.assertRaisesRegex(DataError, "across splits"):
+            self.load()
+        load_manifest(self.manifest, group_by="session_id", allow_synthetic=True)
+
+    def test_session_leakage_even_when_room_names_differ(self):
+        self.entries[1]["session_id"] = "session0"
+        self.save()
+        with self.assertRaisesRegex(DataError, "across splits"):
+            self.load()
+
+    def test_duplicate_hash(self):
+        self.entries[1]["sha256"] = self.entries[0]["sha256"]
+        self.save()
+        with self.assertRaisesRegex(DataError, "duplicate file"):
+            self.load()
+
+    def test_copied_frames_with_different_session_ids(self):
+        rows = [dict(timestamp=t, node_id=1, features=[t+j/10 for j in range(8)]) for t in (0, 1, 40)]
+        self.write_capture(1, rows)
+        self.entries[1]["sha256"] = self.sha(1)
+        self.save()
+        with self.assertRaisesRegex(DataError, "duplicate feature"):
+            self.load()
+
+    def test_collector_plural_features_shape(self):
+        rows = [dict(type="features", node_id=1, seq=i, timestamp_us=t * 1000000,
+                     timestamp=t, features=[t+j/10 for j in range(8)],
+                     subcarriers=[t+j/10 for j in range(8)], rssi=0.0, noise_floor=0.0)
+                for i, t in enumerate((0, 1, 40))]
+        self.write_capture(0, rows)
+        self.entries[0]["sha256"] = self.sha(0)
+        self.save()
+        self.assertEqual(len(self.load().partitions["train"].features), 3)
+
+    def test_actual_collector_recording_envelope(self):
+        spec = importlib.util.spec_from_file_location("csi_collector", Path(__file__).resolve().parents[1] / "collect-training-data.py")
+        collector = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(collector)
+        recorder = collector.CsiRecorder(str(self.base), "synthetic-test")
+        recorder.open()
+        try:
+            for i, timestamp in enumerate((0, 1, 40)):
+                packet = struct.pack(collector.FEATURE_PKT_FMT, collector.MAGIC_FEATURES,
+                                     1, 0, i, timestamp * 1000000,
+                                     *[timestamp+j/10 for j in range(8)])
+                recorder.write_frame(collector._parse_feature_packet(packet))
+        finally:
+            recorder.close()
+        self.entries[0]["path"] = recorder.file_path.name
+        self.entries[0]["sha256"] = hashlib.sha256(recorder.file_path.read_bytes()).hexdigest()
+        self.save()
+        partition = self.load().partitions["train"]
+        self.assertEqual(partition.timestamps.tolist(), [0, 1, 40])
+        self.assertEqual(partition.node_ids, ["1"] * 3)
+        np.testing.assert_allclose(partition.features[1], [1+j/10 for j in range(8)], atol=1e-6)
+
+    def test_raw_collector_envelope_rejected(self):
+        self.write_capture(0, [dict(timestamp=1, subcarriers=[1]*8, rssi=0,
+                                   noise_floor=0, features=dict(node_id=1, seq=0))])
+        self.entries[0]["sha256"] = self.sha(0)
+        self.save()
+        with self.assertRaisesRegex(DataError, "eight explicit features"):
+            self.load()
+
+    def test_constant_features_at_different_times_are_not_copied_frames(self):
+        for i in range(3):
+            self.write_capture(i, [dict(timestamp=i*100+t, node_id=1, features=[0]*8)
+                                   for t in (0, 1, 40)])
+            self.entries[i]["sha256"] = self.sha(i)
+        self.save()
+        self.assertEqual(len(self.load().partitions["test"].features), 3)
+
+    def test_skipped_input_lines_are_bounded(self):
+        with (self.base / "0.jsonl").open("a") as f:
+            f.write('\n' + json.dumps(dict(type="vitals")) + '\n')
+        self.entries[0]["sha256"] = self.sha(0)
+        self.save()
+        with patch("csi_training.data.MAX_INPUT_LINES", 4):
+            with self.assertRaisesRegex(DataError, "input line limit"):
+                self.load()
+
+    def test_invalid_rows(self):
+        cases = [dict(timestamp=1, node_id=1, features=[True] * 8),
+                 dict(timestamp=1, node_id=1, features=[float("nan")] * 8),
+                 dict(timestamp=1, node_id=1, features=[1] * 7),
+                 dict(type="raw_csi", timestamp=1, node_id=1, iq_hex="0011")]
+        for row in cases:
+            with self.subTest(row=row):
+                self.write_capture(0, [row])
+                self.entries[0]["sha256"] = self.sha(0)
+                self.save()
+                with self.assertRaises(DataError):
+                    self.load()
+
+    def test_duplicate_observation_same_file(self):
+        for conflicting in (False, True):
+            with self.subTest(conflicting=conflicting):
+                self.write_capture(0, [dict(timestamp=0, node_id=1, features=[0]*8),
+                                       dict(timestamp=0, node_id=1, features=[int(conflicting)]*8)])
+                self.entries[0]["sha256"] = self.sha(0)
+                self.save()
+                with self.assertRaisesRegex(DataError, "duplicate session/node/timestamp"):
+                    self.load()
+
+    def test_duplicate_observation_across_files_same_session(self):
+        self.write_capture(3, [dict(timestamp=0, node_id=1, features=[999]*8)])
+        self.entries.append(dict(self.entries[0], path="3.jsonl", sha256=self.sha(3)))
+        self.save()
+        with self.assertRaisesRegex(DataError, "duplicate session/node/timestamp"):
+            self.load()
+
+    def test_independent_nodes_at_same_timestamp_allowed(self):
+        self.write_capture(0, [dict(timestamp=0, node_id=node, features=[node]*8)
+                               for node in (1, 2, 3)])
+        self.entries[0]["sha256"] = self.sha(0)
+        self.save()
+        self.assertEqual(len(self.load().partitions["train"].features), 3)
+
+    def test_timestamp_order(self):
+        self.write_capture(0, [dict(timestamp=t, node_id=1, features=[t]*8) for t in (2, 1)])
+        self.entries[0]["sha256"] = self.sha(0)
+        self.save()
+        with self.assertRaisesRegex(DataError, "monotonic"):
+            self.load()
+
+    def test_permissions_and_dates(self):
+        for field, value in (("use_permitted", False), ("captured_at", "2026-01-01")):
+            original = self.entries[0][field]
+            self.entries[0][field] = value
+            self.save()
+            with self.assertRaises(DataError):
+                self.load()
+            self.entries[0][field] = original
+
+    def test_symlink_and_nonregular(self):
+        (self.base / "link.jsonl").symlink_to(self.base / "0.jsonl")
+        self.entries[0]["path"] = "link.jsonl"
+        self.save()
+        with self.assertRaisesRegex(DataError, "symlink"):
+            self.load()
+        self.entries[0]["path"] = "."
+        self.save()
+        with self.assertRaisesRegex(DataError, "regular"):
+            self.load()
+
+    def test_missing_partition(self):
+        self.entries.pop()
+        self.save()
+        with self.assertRaisesRegex(DataError, "partition test"):
+            self.load()
+
+
+if __name__ == "__main__":
+    unittest.main()

scripts/tests/test_csi_training_train.py

@@ -0,0 +1,156 @@
+"""Synthetic CPU integration tests, never evidence of sensing performance."""
+import argparse
+import hashlib
+import importlib.util
+import json
+from pathlib import Path
+import tempfile
+import unittest
+
+import numpy as np
+from csi_training.data import Partition
+from csi_training.train import temporal_triplets, triplet_accuracy, run_training
+
+
+class SamplingTests(unittest.TestCase):
+    def test_triplets_respect_session_node_and_time(self):
+        t = np.tile(np.array([0, 1, 40, 41, 80, 81.]), 4)
+        nodes = ['a'] * 6 + ['b'] * 6 + ['a'] * 6 + ['b'] * 6
+        sessions = ['one'] * 12 + ['two'] * 12
+        partition = Partition(np.ones((24, 8)), t, nodes, ['room'] * 24, sessions)
+        triplets = temporal_triplets(partition, 12)
+        self.assertTrue(np.array_equal(triplets, temporal_triplets(partition, 12)))
+        for a, p, n in triplets:
+            self.assertNotEqual(a, p)
+            self.assertEqual(sessions[a], sessions[p])
+            self.assertEqual(sessions[a], sessions[n])
+            self.assertEqual(nodes[a], nodes[p])
+            self.assertEqual(nodes[a], nodes[n])
+            self.assertGreater(abs(t[a] - t[p]), 0)
+            self.assertLessEqual(abs(t[a] - t[p]), 2)
+            self.assertGreater(abs(t[a] - t[n]), 30)
+
+    def test_duplicate_timestamp_only_pairs_rejected(self):
+        times = np.array([0, 0, 40, 40, 80, 80.])
+        part = Partition(np.tile(np.eye(3, 8), (2, 1)), times,
+                         ['a'] * 6, ['r'] * 6, ['s'] * 6)
+        with self.assertRaisesRegex(ValueError, '0<dt'):
+            temporal_triplets(part, 42)
+
+    def test_equal_timestamp_group_is_excluded_from_positives(self):
+        times = np.array([0, 0, 0, 1, 40, 40, 41, 80, 80, 81.])
+        part = Partition(np.ones((10, 8)), times, ['a'] * 10, ['r'] * 10, ['s'] * 10)
+        triples = temporal_triplets(part, 42, limit=4)
+        self.assertEqual(len(triples), 4)
+        for a, p, n in triples:
+            self.assertGreater(abs(times[a] - times[p]), 0)
+            self.assertLessEqual(abs(times[a] - times[p]), 2)
+            self.assertGreater(abs(times[a] - times[n]), 30)
+
+    def test_constant_embedding_is_chance(self):
+        self.assertEqual(triplet_accuracy(np.ones((3, 128)), np.array([[0, 1, 2]])), 0.5)
+
+    def test_missing_temporal_negatives_fails(self):
+        part = Partition(np.ones((3, 8)), np.arange(3), ['a']*3, ['r']*3, ['s']*3)
+        with self.assertRaises(ValueError):
+            temporal_triplets(part, 42)
+
+
+@unittest.skipUnless(importlib.util.find_spec('torch') and importlib.util.find_spec('safetensors'), 'install requirements-csi-training.txt for real CPU training tests')
+class TrainingTests(unittest.TestCase):
+    def fixture(self, root):
+        entries = []
+        rng = np.random.default_rng(13)
+        for i, split in enumerate(['train', 'validation', 'test']):
+            rows = []
+            for state in range(8):
+                center = rng.normal(size=8)
+                for offset in range(4):
+                    rows.append({'timestamp': state * 40 + offset * 0.5, 'node_id': 'a',
+                                 'features': (center + rng.normal(scale=0.02, size=8)).tolist()})
+            path = root / f'{split}.jsonl'
+            path.write_text(''.join(json.dumps(row) + '\n' for row in rows))
+            entries.append({'path': path.name, 'sha256': hashlib.sha256(path.read_bytes()).hexdigest(),
+                            'captured_at': '2026-10-12T00:00:00Z', 'room_id': f'room{i}',
+                            'session_id': f'session{i}', 'split': split,
+                            'evidence': 'synthetic', 'use_permitted': True})
+        manifest = root / 'manifest.json'
+        manifest.write_text(json.dumps({'schema_version': 1, 'files': entries}))
+        return manifest
+
+    def args(self, manifest, output):
+        return argparse.Namespace(manifest=str(manifest), output=str(output), group_by='room_id',
+                                  allow_synthetic=True, device='cpu', epochs=2, batch_size=16,
+                                  max_triplets=128, seed=42, max_seconds=60,
+                                  learning_rate=0.001, temperature=0.1, command='train')
+
+    def test_training_updates_weights_replays_and_exports(self):
+        import torch
+        from safetensors.torch import load_file
+        from csi_training.model import CSIEncoder, PreprocessedEncoder
+        with tempfile.TemporaryDirectory() as temp:
+            root = Path(temp)
+            manifest = self.fixture(root)
+            a = run_training(self.args(manifest, root / 'a'))
+            b = run_training(self.args(manifest, root / 'b'))
+            self.assertGreater(a['trainable_parameter_delta_squared'], 0)
+            self.assertEqual(a['model_sha256'], b['model_sha256'])
+            self.assertEqual(a['history'], b['history'])
+            self.assertEqual(a['metrics'], b['metrics'])
+            self.assertFalse(a['promotionAuthorized'])
+            self.assertEqual(a['evidence'], 'SYNTHETIC')
+            model = CSIEncoder().eval()
+            model.load_state_dict(load_file(str(root / 'a' / 'model.safetensors')))
+            with torch.no_grad():
+                embedded = model(torch.ones(3, 8))
+            self.assertTrue(torch.isfinite(embedded).all())
+            self.assertTrue(torch.allclose(embedded.norm(dim=-1), torch.ones(3), atol=1e-6))
+            self.assertEqual(a['model_sha256'], hashlib.sha256((root / 'a' / 'model.safetensors').read_bytes()).hexdigest())
+            preprocessing = json.loads((root / 'a' / 'preprocessing.json').read_text())
+            self.assertEqual(preprocessing['fit_partition'], 'train')
+            self.assertEqual(len(preprocessing['mean']), 8)
+            loader = PreprocessedEncoder(root / 'a')
+            raw = np.ones((3, 8))
+            normalized = torch.from_numpy(((raw - np.array(preprocessing['mean'])) / np.array(preprocessing['scale'])).astype(np.float32))
+            with torch.no_grad():
+                expected = model(normalized).numpy()
+            np.testing.assert_allclose(loader.encode(raw), expected, atol=1e-6)
+            with self.assertRaises(ValueError):
+                loader.encode(np.full((1, 8), np.nan))
+            with self.assertRaisesRegex(ValueError, 'must not exist'):
+                run_training(self.args(manifest, root / 'a'))
+            status_path = root / 'a' / 'status.json'
+            status_path.write_text(json.dumps({'status': 'failed', 'promotionAuthorized': False}))
+            with self.assertRaisesRegex(ValueError, 'lifecycle'):
+                PreprocessedEncoder(root / 'a')
+            status_path.write_text(json.dumps({'status': 'complete', 'promotionAuthorized': False}))
+            (root / 'a' / 'preprocessing.json').write_text('{}')
+            with self.assertRaisesRegex(ValueError, 'hash mismatch'):
+                PreprocessedEncoder(root / 'a')
+
+    def test_elapsed_budget_leaves_failed_nonpromotable_status(self):
+        from unittest.mock import patch
+        with tempfile.TemporaryDirectory() as temp:
+            root = Path(temp)
+            manifest = self.fixture(root)
+            with patch('csi_training.train.time.monotonic', side_effect=[0.0, 61.0]):
+                with self.assertRaises(TimeoutError):
+                    run_training(self.args(manifest, root / 'out'))
+            status = json.loads((root / 'out' / 'status.json').read_text())
+            self.assertEqual(status['status'], 'failed')
+            self.assertFalse(status['promotionAuthorized'])
+            self.assertFalse((root / 'out' / 'model.safetensors').exists())
+
+    def test_invalid_budget_rejected_before_output(self):
+        with tempfile.TemporaryDirectory() as temp:
+            root = Path(temp)
+            manifest = self.fixture(root)
+            args = self.args(manifest, root / 'out')
+            args.max_seconds = float('nan')
+            with self.assertRaises(ValueError):
+                run_training(args)
+            self.assertFalse((root / 'out').exists())
+
+
+if __name__ == '__main__':
+    unittest.main()

scripts/train-csi-encoder.py

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+"""Local, proposal-only temporal CSI training entry point."""
+from csi_training.train import main
+
+if __name__ == '__main__':
+    main()