diff --git a/Recordings/README.md b/Recordings/README.md index 6b73ab4..58ba0b6 100644 --- a/Recordings/README.md +++ b/Recordings/README.md @@ -11,3 +11,21 @@ Playback the most recent recording with: ```bash ./Scripts/run-synthetic.sh --profile playback --recording Recordings/.csv ``` + +## Session events + +A recording can be summarised into structured, replayable observations: + +```bash +python3 Scripts/extract_session_events.py Recordings/ +``` + +This writes `/session-events.jsonl` — one `nc-eeg-session-event-v0` +record per observation. Events carry the **SHA-256 of `eeg.csv` plus a sample +range**, never signal, so the log is meaningless without the recording it points +into and the privacy posture above is unchanged: the log stays here, and here is +gitignored. + +They are **signal observations only**. No sleep stage, no intervention outcome — +`Scripts/session_event_contract.py` rejects a record that asserts either. See +[`docs/architecture/session-events.md`](../docs/architecture/session-events.md). diff --git a/Scripts/extract_session_events.py b/Scripts/extract_session_events.py new file mode 100644 index 0000000..03cd921 --- /dev/null +++ b/Scripts/extract_session_events.py @@ -0,0 +1,435 @@ +#!/usr/bin/env python3 +"""Extract structured session events from a recording, offline and deterministically. + + python3 Scripts/extract_session_events.py [--out events.jsonl] + python3 Scripts/extract_session_events.py --stdout + +Reads `/eeg.csv` and emits one `nc-eeg-session-event-v0` record per +observation. **No signal is copied.** Each event carries the SHA-256 of the +`eeg.csv` it points into plus a sample range, so the window is re-derivable by +anyone holding the recording and meaningless to anyone who is not. + +Determinism is a contract, not an aspiration: the same file yields byte-identical +output. Nothing consults the clock, the filesystem order, or a random seed, and +every threshold is pinned in `PARAMS` and echoed into each record so a later +reader can tell which rules produced it. + +**Observations only.** No stage is inferred, no intervention is implied. See +`Scripts/session_event_contract.py` for the boundary the validator enforces. + +Standard library only — the system interpreter on a clean runner has no numpy. +""" + +from __future__ import annotations + +import argparse +import csv +import hashlib +import json +import math +import sys +from pathlib import Path +from typing import Any, Iterator + +sys.path.insert(0, str(Path(__file__).resolve().parent)) +from session_event_contract import ( # noqa: E402 + CHANNELS, EVENT_KINDS, INTERPRETATION, SAMPLE_RATE_HZ, + SESSION_EVENT_SCHEMA, SESSION_MANIFEST_SCHEMA, + validate_session_event_file, validate_session_manifest, +) + +# Pinned analysis parameters. Changing any of these changes the events, so the +# set is hashed into every record's `params_sha256`: two logs are comparable +# only if that digest matches. +PARAMS: dict[str, Any] = { + "window_samples": 1024, # 4 s at 256 Hz — the canonical window + "stride_samples": 256, # 1 s hop + "baseline_windows": 30, # first 30 windows form the in-session baseline + "gap_factor": 1.5, # sample-clock gap > 1.5 × nominal ⇒ throughput event + "clip_microvolts": 800.0, # |x| at or above this counts as clipped + "clip_fraction_trigger": 0.10, # ≥10 % clipped samples in a window + "rms_ratio_trigger": 3.0, # per-channel RMS ≥ 3 × its baseline median + "band_ratio_trigger": 2.5, # band power ≥ 2.5 × its baseline median + "burst_sigma": 6.0, # |x| ≥ 6 × baseline RMS ⇒ burst sample + "burst_min_samples": 8, # ≥8 such samples in a window ⇒ artifact_burst + "bands": {"theta": [4.0, 8.0], "alpha": [8.0, 13.0], "beta": [13.0, 30.0]}, +} + + +def params_digest() -> str: + return hashlib.sha256( + json.dumps(PARAMS, sort_keys=True, separators=(",", ":")).encode("utf-8") + ).hexdigest() + + +def file_sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1 << 20), b""): + digest.update(chunk) + return digest.hexdigest() + + +def read_eeg(path: Path) -> tuple[list[float], list[list[float]]]: + """Return (timestamps, per-channel sample lists) in canonical channel order.""" + times: list[float] = [] + channels: list[list[float]] = [[] for _ in CHANNELS] + with path.open(newline="") as handle: + reader = csv.DictReader(handle) + missing = [c for c in CHANNELS if c not in (reader.fieldnames or [])] + if missing: + raise SystemExit(f"{path} is missing channel column(s): {missing}") + for line_no, row in enumerate(reader, start=2): + # Fail closed. A row with a missing cell, a non-numeric value, or a + # NaN/inf would otherwise propagate into a baseline median and + # silently move every threshold derived from it. + try: + t = float(row["t_seconds"]) + values = [float(row[name]) for name in CHANNELS] + except (TypeError, ValueError) as exc: + raise SystemExit(f"{path}:{line_no}: unparseable row ({exc})") from exc + if not math.isfinite(t) or any(not math.isfinite(v) for v in values): + raise SystemExit(f"{path}:{line_no}: non-finite value") + if len(row) < len(CHANNELS) + 1: + raise SystemExit(f"{path}:{line_no}: expected {len(CHANNELS)} channels") + times.append(t) + for index, value in enumerate(values): + channels[index].append(value) + if len(times) >= 2: + # The recording must agree with the montage contract it will be + # referenced under; a 128 Hz file indexed as 256 Hz points at the wrong + # seconds for every event in the log. + span = times[-1] - times[0] + if span > 0: + observed_rate = (len(times) - 1) / span + if abs(observed_rate - SAMPLE_RATE_HZ) / SAMPLE_RATE_HZ > 0.05: + raise SystemExit( + f"{path}: sample rate ~{observed_rate:.1f} Hz disagrees with the " + f"{SAMPLE_RATE_HZ} Hz contract by more than 5%") + return times, channels + + +def goertzel_power(samples: list[float], low_hz: float, high_hz: float) -> float: + """Summed power in [low, high) via Goertzel over the integer bins. + + Goertzel rather than a full transform because this is stdlib-only: it costs + O(n) per bin and the bands here need a few dozen bins, where a hand-rolled + DFT would be O(n²) per window and far too slow to run over a night. + + Demeaned *and* Hann-windowed before transforming. Both matter for the same + reason the burst detector is per-channel: a DC offset or a rectangular + window's spectral leakage lets a large-amplitude channel bleed energy across + every bin, which would reintroduce the electrode-baseline problem in the + frequency domain. The Hann coherent-gain factor is divided back out so the + power stays comparable to an unwindowed estimate. + """ + n = len(samples) + if n == 0: + return 0.0 + mean = sum(samples) / n + window = [0.5 - 0.5 * math.cos(2.0 * math.pi * i / (n - 1)) for i in range(n)] + centred = [(s - mean) * w for s, w in zip(samples, window)] + coherent_gain = sum(window) / n # 0.5 for Hann + resolution = SAMPLE_RATE_HZ / n + first = max(1, int(math.ceil(low_hz / resolution))) + last = int(math.floor((high_hz - 1e-9) / resolution)) + total = 0.0 + for k in range(first, last + 1): + omega = 2.0 * math.pi * k / n + coeff = 2.0 * math.cos(omega) + s1 = s2 = 0.0 + for value in centred: + s0 = value + coeff * s1 - s2 + s2, s1 = s1, s0 + total += s1 * s1 + s2 * s2 - coeff * s1 * s2 + return total / (n * n * coherent_gain * coherent_gain) + + +def rms(samples: list[float]) -> float: + if not samples: + return 0.0 + return math.sqrt(sum(s * s for s in samples) / len(samples)) + + +def median(values: list[float]) -> float: + if not values: + return 0.0 + ordered = sorted(values) + mid = len(ordered) // 2 + if len(ordered) % 2: + return ordered[mid] + return (ordered[mid - 1] + ordered[mid]) / 2.0 + + +def windows(total: int) -> Iterator[tuple[int, int]]: + size, stride = PARAMS["window_samples"], PARAMS["stride_samples"] + start = 0 + while start + size <= total: + yield start, size + start += stride + + +def make_event( + kind: str, digest: str, start: int, count: int, + observed: dict[str, Any], baseline: tuple[int, int] | None, +) -> dict[str, Any]: + # The id is derived from content, so re-running produces the same ids and a + # diff shows only real changes. + # Everything that changes the meaning of the record, and nothing that does + # not: schema, recording, kind, range, parameters, and the observation + # itself. Deliberately no filesystem path — the same bytes under a different + # name must produce the same reference, or the artifact is not + # content-addressed. + seed = ":".join(( + SESSION_EVENT_SCHEMA, digest, kind, str(start), str(count), + params_digest(), json.dumps(observed, sort_keys=True, separators=(",", ":")), + )) + return { + "schema_version": SESSION_EVENT_SCHEMA, + "event_id": hashlib.sha256(seed.encode("utf-8")).hexdigest()[:32], + "kind": kind, + "source": { + "recording_sha256": digest, + "start_sample": start, + "sample_count": count, + "channel_order": list(CHANNELS), + "sample_rate_hz": SAMPLE_RATE_HZ, + }, + "observed": observed, + "baseline_ref": ( + None if baseline is None + else {"start_sample": baseline[0], "sample_count": baseline[1]} + ), + "params_sha256": params_digest(), + "interpretation": INTERPRETATION, + "stage_claim": None, + "intervention_claim": None, + } + + +def extract(recording_dir: Path) -> list[dict[str, Any]]: + """Events only. Use `extract_with_manifest` when eligibility matters.""" + eeg_path = recording_dir / "eeg.csv" + if not eeg_path.is_file(): + raise SystemExit(f"no eeg.csv in {recording_dir}") + digest = file_sha256(eeg_path) + times, channels = read_eeg(eeg_path) + total = len(times) + events: list[dict[str, Any]] = [] + + # 1 · throughput gaps in the sample clock. This is the watchdog's own + # condition (HealthSnapshot's `eeg-zero-throughput`) recorded as a + # referenceable span rather than only raised as a live flag. + nominal = 1.0 / SAMPLE_RATE_HZ + for index in range(1, total): + delta = times[index] - times[index - 1] + if delta > nominal * PARAMS["gap_factor"]: + events.append(make_event( + "zero_throughput", digest, index - 1, 2, + {"gap_seconds": round(delta, 6), + "expected_seconds": round(nominal, 6), + "missing_samples_estimate": int(round(delta / nominal)) - 1}, + None)) + + if total < PARAMS["window_samples"]: + return events + + window_list = list(windows(total)) + baseline_count = min(PARAMS["baseline_windows"], len(window_list)) + baseline_span = (window_list[0][0], + window_list[baseline_count - 1][0] + PARAMS["window_samples"] + - window_list[0][0]) + + # Per-channel and per-band baselines: the median over the opening windows of + # this same session. In-session by construction — a cross-session baseline + # would silently import another night's headset fit. + per_channel_rms: list[list[float]] = [[] for _ in CHANNELS] + per_band: dict[str, list[float]] = {b: [] for b in PARAMS["bands"]} + for start, size in window_list[:baseline_count]: + for ch_index, samples in enumerate(channels): + per_channel_rms[ch_index].append(rms(samples[start:start + size])) + merged = [sum(c[start + i] for c in channels) / len(channels) for i in range(size)] + for band, (low, high) in PARAMS["bands"].items(): + per_band[band].append(goertzel_power(merged, low, high)) + + rms_baseline = [median(v) for v in per_channel_rms] + band_baseline = {b: median(v) for b, v in per_band.items()} + + # Detector eligibility, per channel. + # + # The burst detector compares a channel against its *own* baseline. When + # that baseline is itself pathological — a saturated electrode — the + # comparison is vacuous: nothing can exceed 6x a rail. On a real recording + # this produced zero artifact_burst records for a visibly broken AF7, and + # an absent record is ambiguous. It could mean clean, or no transient, or + # a detector that was never eligible to fire. Those must not be the same + # observation, so eligibility is stated rather than inferred from silence. + eligibility: dict[str, dict[str, Any]] = {} + for ch_index, name in enumerate(CHANNELS): + base = rms_baseline[ch_index] + baseline_samples = channels[ch_index][baseline_span[0]: + baseline_span[0] + baseline_span[1]] + clipped = sum(1 for s in baseline_samples if abs(s) >= PARAMS["clip_microvolts"]) + clipped_fraction = clipped / len(baseline_samples) if baseline_samples else 0.0 + if base <= 0: + status, reason = "suppressed", "channel_silent" + elif base >= PARAMS["clip_microvolts"] or clipped_fraction >= PARAMS["clip_fraction_trigger"]: + status, reason = "suppressed", "channel_saturated" + else: + status, reason = "eligible", None + eligibility[name] = { + "detector_status": status, + "suppressed_reason": reason, + "baseline_rms_microvolts": round(base, 6), + "baseline_clipped_fraction": round(clipped_fraction, 6), + } + + for start, size in window_list: + window_channels = [c[start:start + size] for c in channels] + + # 2 · per-channel health: clipping fraction, or RMS far above baseline. + for ch_index, name in enumerate(CHANNELS): + samples = window_channels[ch_index] + clipped = sum(1 for s in samples if abs(s) >= PARAMS["clip_microvolts"]) + fraction = clipped / len(samples) + channel_rms = rms(samples) + base = rms_baseline[ch_index] + ratio = (channel_rms / base) if base > 0 else 0.0 + if fraction >= PARAMS["clip_fraction_trigger"] or ratio >= PARAMS["rms_ratio_trigger"]: + events.append(make_event( + "channel_health_change", digest, start, size, + {"channel": name, + "clipped_fraction": round(fraction, 6), + "rms_microvolts": round(channel_rms, 6), + "baseline_rms_microvolts": round(base, 6), + "rms_ratio": round(ratio, 6)}, + baseline_span)) + + # 3 · band excursion against the in-session baseline. + merged = [sum(c[i] for c in window_channels) / len(window_channels) + for i in range(size)] + for band, (low, high) in PARAMS["bands"].items(): + power = goertzel_power(merged, low, high) + base = band_baseline[band] + ratio = (power / base) if base > 0 else 0.0 + if ratio >= PARAMS["band_ratio_trigger"]: + events.append(make_event( + "band_excursion", digest, start, size, + {"band": band, "low_hz": low, "high_hz": high, + "power": round(power, 9), + "baseline_power": round(base, 9), + "power_ratio": round(ratio, 6)}, + baseline_span)) + + # 4 · artifact burst: samples far above *that channel's own* baseline. + # + # Deliberately per-channel. A cross-channel envelope makes one bad + # electrode trip the burst detector on every window — on the first run + # against a real recording, a saturated AF7 (~900 µV against ~20 µV + # elsewhere) produced an artifact_burst almost every second, restating + # a fault `channel_health_change` already reports. A burst is a + # departure from a channel's own norm, so the norm has to be its own. + for ch_index, name in enumerate(CHANNELS): + if eligibility[name]["detector_status"] != "eligible": + continue # recorded in the manifest, not inferable from silence + base = rms_baseline[ch_index] + threshold = PARAMS["burst_sigma"] * base + hits = sum(1 for s in window_channels[ch_index] if abs(s) >= threshold) + if hits >= PARAMS["burst_min_samples"]: + events.append(make_event( + "artifact_burst", digest, start, size, + {"channel": name, + "samples_over_threshold": hits, + "threshold_microvolts": round(threshold, 6), + "baseline_rms_microvolts": round(base, 6)}, + baseline_span)) + + # Stable order: by position, then kind, then id. Independent of dict order. + events.sort(key=lambda e: (e["source"]["start_sample"], e["kind"], e["event_id"])) + _LAST_MANIFEST.clear() + _LAST_MANIFEST.update(build_manifest(digest, total, eligibility, events)) + return events + + +# Populated by `extract`; read by `extract_with_manifest`. A module-level box +# rather than a changed return type, so existing callers keep working. +_LAST_MANIFEST: dict[str, Any] = {} + + +def build_manifest( + digest: str, total: int, eligibility: dict[str, dict[str, Any]], + events: list[dict[str, Any]], +) -> dict[str, Any]: + """The session-level record that makes an *absent* event interpretable. + + Every disposition here is deliberately negative. This artifact indexes a + recording; it is not a dataset, not a label set, and not evidence. It + credits nothing toward the §21 five-clean-session gate, because whether a + session was clean is a judgement about the recording, not about whether an + extractor ran over it. + """ + counts: dict[str, int] = {} + for event in events: + counts[event["kind"]] = counts.get(event["kind"], 0) + 1 + return { + "schema_version": SESSION_MANIFEST_SCHEMA, + "recording_sha256": digest, + "recording_sample_count": total, + "params_sha256": params_digest(), + "sample_interval_convention": "half_open_[start_sample,_start_sample+sample_count)", + "channels": {name: eligibility[name] for name in CHANNELS}, + "event_counts": {kind: counts.get(kind, 0) for kind in sorted(EVENT_KINDS)}, + "contains_signal": False, + "science_status": "pipeline_only", + "label_status": "heuristic_observation", + "live_control": False, + "promotion_status": "not_eligible", + "clean_session_gate_credited": False, + } + + +def extract_with_manifest( + recording_dir: Path, +) -> tuple[list[dict[str, Any]], dict[str, Any]]: + events = extract(recording_dir) + return events, dict(_LAST_MANIFEST) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("recording", type=Path, help="a Recordings/ directory") + parser.add_argument("--out", type=Path, default=None, + help="output JSONL (default: /session-events.jsonl)") + parser.add_argument("--stdout", action="store_true", help="write to stdout instead") + args = parser.parse_args(argv) + + events, manifest = extract_with_manifest(args.recording) + validate_session_event_file(events) + validate_session_manifest(manifest) + + lines = "".join( + json.dumps(e, sort_keys=True, separators=(",", ":")) + "\n" for e in events) + if args.stdout: + sys.stdout.write(lines) + else: + destination = args.out or (args.recording / "session-events.jsonl") + destination.write_text(lines, encoding="utf-8") + manifest_path = destination.with_name("session-events-manifest.json") + manifest_path.write_text( + json.dumps(manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8") + print(f"{manifest_path}") + for name in CHANNELS: + entry = manifest["channels"][name] + note = f" ({entry['suppressed_reason']})" if entry["suppressed_reason"] else "" + print(f" {name:5s} burst detector: {entry['detector_status']}{note}") + kinds: dict[str, int] = {} + for event in events: + kinds[event["kind"]] = kinds.get(event["kind"], 0) + 1 + print(f"{destination}: {len(events)} events") + for kind in sorted(kinds): + print(f" {kind:24s} {kinds[kind]}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/Scripts/session_event_contract.py b/Scripts/session_event_contract.py new file mode 100644 index 0000000..167a5ad --- /dev/null +++ b/Scripts/session_event_contract.py @@ -0,0 +1,250 @@ +#!/usr/bin/env python3 +"""Contract for `nc-eeg-session-event-v0` — structured references into a recording. + +A session event does not carry signal. It carries a *reference*: the SHA-256 of +the `eeg.csv` it points into, plus a sample range. Anyone holding the recording +can re-derive the window and recompute what was observed; anyone without it +learns nothing. That is what makes the artifact model-agnostic — EEGNet, EEGPT, +a future encoder, or a plain spectral routine all consume the same reference. + +**These are signal observations, not stage claims.** The Muse montage has no +chin EMG (`SLEEP_CYCLE_DESIGN.md` §291, risk R1), so nothing here may imply REM, +and no sleep stage is inferable from any field. Detection is gated behind the +§21 toolkit gate; intervention efficacy is gated behind the D8 pre-registration +(`Evaluation/reports/decision_registry.md` entry 8). This artifact is upstream of +both and asserts neither — the validator enforces that rather than documenting it. + +Standard library only. The checker runs as a direct `python3` step, and the +system interpreter on a clean runner has no numpy. +""" + +from __future__ import annotations + +import re +from typing import Any + +SESSION_EVENT_SCHEMA = "nc-eeg-session-event-v0" +SESSION_MANIFEST_SCHEMA = "nc-eeg-session-manifest-v0" + +# The canonical Muse montage and rate, matching +# NeuralComposeEEG/configs/muse-four-channel-v0.json. +CHANNELS: tuple[str, ...] = ("TP9", "AF7", "AF8", "TP10") +SAMPLE_RATE_HZ = 256 + +EVENT_KINDS: frozenset[str] = frozenset({ + "zero_throughput", # a gap in the sample clock + "channel_health_change", # per-channel RMS / clipping transition + "band_excursion", # band power departing an in-session baseline + "artifact_burst", # high-amplitude transient run +}) + +INTERPRETATION = "signal_observation_only" + +# Field names that would smuggle an interpretation into an observation record. +# `stage`/`n1`/`rem` are stage claims; `intervention`/`efficacy` are outcome +# claims; `embedding`/`latent` would make the artifact model-specific and defeat +# the reason it stores a reference instead of a vector. +FORBIDDEN_SUBSTRINGS: tuple[str, ...] = ( + "stage", "n1", "rem", "sleep", "intervention", "efficacy", + "embedding", "latent", "vector", +) +# `stage_claim` / `intervention_claim` are the two deliberate exceptions: they +# exist precisely to be pinned at null. +NULL_CLAIM_FIELDS: tuple[str, ...] = ("stage_claim", "intervention_claim") + +_SHA256 = re.compile(r"^[0-9a-f]{64}$") + + +class ContractError(ValueError): + """A session-event record that must not be written or consumed.""" + + +def _require(condition: bool, message: str) -> None: + if not condition: + raise ContractError(message) + + +def _walk_keys(value: Any, path: str = "") -> list[str]: + """Every key path in a nested record, so a forbidden name cannot hide.""" + found: list[str] = [] + if isinstance(value, dict): + for key, sub in value.items(): + here = f"{path}.{key}" if path else str(key) + found.append(here) + found.extend(_walk_keys(sub, here)) + elif isinstance(value, list): + for index, sub in enumerate(value): + found.extend(_walk_keys(sub, f"{path}[{index}]")) + return found + + +def validate_session_event( + record: dict[str, Any], + *, + recording_sample_count: int | None = None, +) -> None: + """Validate one event. Pass `recording_sample_count` to bound the range. + + Without it the range is checked for internal consistency only; a reference + past the end of a recording is caught when the recording is available. + """ + _require(isinstance(record, dict), "event must be a JSON object") + _require(record.get("schema_version") == SESSION_EVENT_SCHEMA, + f"schema_version must be {SESSION_EVENT_SCHEMA}") + + event_id = record.get("event_id") + _require(isinstance(event_id, str) and event_id != "", "event_id must be a non-empty string") + + kind = record.get("kind") + _require(kind in EVENT_KINDS, + f"kind must be one of {sorted(EVENT_KINDS)}; got {kind!r}") + + _require(record.get("interpretation") == INTERPRETATION, + f"interpretation must be {INTERPRETATION!r} — this artifact makes no " + "stage or efficacy claim") + + for field in NULL_CLAIM_FIELDS: + _require(field in record, f"{field} must be present and null") + _require(record[field] is None, + f"{field} must be null; a session event may not assert one") + + # No forbidden name anywhere in the record, at any depth. The two + # `*_claim` fields above are the sanctioned exceptions. + for key_path in _walk_keys(record): + leaf = key_path.split(".")[-1].split("[")[0].lower() + if leaf in NULL_CLAIM_FIELDS: + continue + for banned in FORBIDDEN_SUBSTRINGS: + _require(banned not in leaf, + f"field {key_path!r} contains {banned!r}: a session event records " + "what was measured, never what it might mean") + + source = record.get("source") + _require(isinstance(source, dict), "source must be an object") + + digest = source.get("recording_sha256") + _require(isinstance(digest, str) and _SHA256.match(digest) is not None, + "source.recording_sha256 must be a lowercase SHA-256 hex digest") + + start = source.get("start_sample") + count = source.get("sample_count") + _require(isinstance(start, int) and not isinstance(start, bool) and start >= 0, + "source.start_sample must be a non-negative integer") + _require(isinstance(count, int) and not isinstance(count, bool) and count > 0, + "source.sample_count must be a positive integer") + + if recording_sample_count is not None: + _require(start + count <= recording_sample_count, + f"reference [{start}, {start + count}) exceeds the recording's " + f"{recording_sample_count} samples") + + _require(list(source.get("channel_order") or []) == list(CHANNELS), + f"source.channel_order must be exactly {list(CHANNELS)}") + _require(source.get("sample_rate_hz") == SAMPLE_RATE_HZ, + f"source.sample_rate_hz must be {SAMPLE_RATE_HZ}") + + observed = record.get("observed") + _require(isinstance(observed, dict) and observed != {}, + "observed must be a non-empty object") + for key, value in observed.items(): + _require(isinstance(value, (int, float, str, bool)) or value is None, + f"observed.{key} must be a scalar — an event carries measurements, " + "not signal") + + baseline = record.get("baseline_ref") + if baseline is not None: + _require(isinstance(baseline, dict), "baseline_ref must be an object or null") + b_start, b_count = baseline.get("start_sample"), baseline.get("sample_count") + _require(isinstance(b_start, int) and b_start >= 0, + "baseline_ref.start_sample must be a non-negative integer") + _require(isinstance(b_count, int) and b_count > 0, + "baseline_ref.sample_count must be a positive integer") + + +def validate_session_event_file( + records: list[dict[str, Any]], + *, + recording_sample_count: int | None = None, +) -> None: + """Validate a whole event log, including cross-record invariants.""" + for index, record in enumerate(records): + try: + validate_session_event(record, recording_sample_count=recording_sample_count) + except ContractError as exc: + raise ContractError(f"event {index}: {exc}") from exc + + digests = {r["source"]["recording_sha256"] for r in records} + _require(len(digests) <= 1, + f"an event log references {len(digests)} recordings; one log describes one recording") + + ids = [r["event_id"] for r in records] + _require(len(ids) == len(set(ids)), "event_id values must be unique within a log") + + +# ── session manifest ──────────────────────────────────────────────────────── + +DETECTOR_STATUSES: frozenset[str] = frozenset({"eligible", "suppressed"}) +SUPPRESSED_REASONS: frozenset[str] = frozenset({"channel_saturated", "channel_silent"}) + +# Dispositions that must be present and must be exactly these values. They are +# all negative on purpose: this artifact indexes a recording, and an index is +# not a dataset, not a label set, and not evidence. +REQUIRED_DISPOSITIONS: dict[str, Any] = { + "contains_signal": False, + "science_status": "pipeline_only", + "label_status": "heuristic_observation", + "live_control": False, + "promotion_status": "not_eligible", + "clean_session_gate_credited": False, +} + + +def validate_session_manifest(manifest: dict[str, Any]) -> None: + """Validate the session manifest. + + The manifest exists so an *absent* event is interpretable. A channel whose + baseline is already saturated cannot produce an `artifact_burst`, and on a + real recording one did not — silence there means the detector was never + eligible, not that the channel was clean. Encoding that distinction is the + manifest's entire job, so a manifest that omits a channel is rejected. + """ + _require(isinstance(manifest, dict), "manifest must be a JSON object") + _require(manifest.get("schema_version") == SESSION_MANIFEST_SCHEMA, + f"schema_version must be {SESSION_MANIFEST_SCHEMA}") + + digest = manifest.get("recording_sha256") + _require(isinstance(digest, str) and _SHA256.match(digest) is not None, + "recording_sha256 must be a lowercase SHA-256 hex digest") + + total = manifest.get("recording_sample_count") + _require(isinstance(total, int) and not isinstance(total, bool) and total >= 0, + "recording_sample_count must be a non-negative integer") + + _require(manifest.get("sample_interval_convention", "").startswith("half_open"), + "sample_interval_convention must declare the half-open convention") + + channels = manifest.get("channels") + _require(isinstance(channels, dict), "channels must be an object") + for name in CHANNELS: + _require(name in channels, + f"channels is missing {name}: every channel must state detector " + "eligibility, or an absent event is ambiguous") + entry = channels[name] + _require(isinstance(entry, dict), f"channels.{name} must be an object") + status = entry.get("detector_status") + _require(status in DETECTOR_STATUSES, + f"channels.{name}.detector_status must be one of {sorted(DETECTOR_STATUSES)}") + reason = entry.get("suppressed_reason") + if status == "suppressed": + _require(reason in SUPPRESSED_REASONS, + f"channels.{name} is suppressed and must name a reason from " + f"{sorted(SUPPRESSED_REASONS)}") + else: + _require(reason is None, + f"channels.{name} is eligible; suppressed_reason must be null") + + for key, expected in REQUIRED_DISPOSITIONS.items(): + _require(key in manifest, f"manifest must state {key}") + _require(manifest[key] == expected, + f"{key} must be {expected!r} — this artifact indexes a recording; " + "it is not a dataset, a label set, or evidence") diff --git a/Tests/eval/test_session_events.py b/Tests/eval/test_session_events.py new file mode 100644 index 0000000..51e66e2 --- /dev/null +++ b/Tests/eval/test_session_events.py @@ -0,0 +1,463 @@ +"""Tests for the session-event reference artifact (nc-eeg-session-event-v0). + +Three properties matter, and each has its own group below. + + * **Contract** — an event may reference signal and report measurements, but may + never carry a stage claim, an efficacy claim, or an embedding. The validator + enforces the boundary rather than the documentation describing it. + * **Determinism** — the same recording yields byte-identical output, and a + one-sample edit changes the recording digest so every reference into it is + invalidated rather than silently re-pointed. + * **Replay** — given only `recording_sha256` and a sample range, the window is + re-derivable and `observed` recomputes to equality. This is the property + that makes the artifact model-agnostic; without it the reference is just a + number and the log is unfalsifiable. + +Standard library only, and the fixtures are synthesised here rather than read +from `Recordings/`: no `eeg.csv` is committed (only the golden report), so a +test depending on one would skip vacuously on a clean checkout. + +pytest-style + __main__ runner, matching the rest of Tests/eval. +""" +import copy +import csv +import hashlib +import json +import math +import sys +import tempfile +from pathlib import Path + +REPO = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(REPO / "Scripts")) + +import extract_session_events as extractor # noqa: E402 +from session_event_contract import ( # noqa: E402 + CHANNELS, ContractError, REQUIRED_DISPOSITIONS, SAMPLE_RATE_HZ, + SESSION_EVENT_SCHEMA, validate_session_event, validate_session_event_file, + validate_session_manifest, +) + +WINDOW = extractor.PARAMS["window_samples"] + + +# ── fixtures ──────────────────────────────────────────────────────────────── + +def _write_recording(directory: Path, *, samples: int = 12000, + gap_at: int | None = None, clip_channel: int | None = None, + clip_from: int | None = None) -> Path: + """A deterministic synthetic recording. No randomness, no clock.""" + directory.mkdir(parents=True, exist_ok=True) + path = directory / "eeg.csv" + period = 1.0 / SAMPLE_RATE_HZ + with path.open("w", newline="") as handle: + writer = csv.writer(handle) + writer.writerow(["t_seconds", *CHANNELS]) + t = 1_000_000.0 + for i in range(samples): + if gap_at is not None and i == gap_at: + t += period * 40 # a visible hole in the sample clock + row = [] + for ch in range(len(CHANNELS)): + # 10 Hz carrier, per-channel phase offset, ~20 µV amplitude. + value = 20.0 * math.sin(2 * math.pi * 10.0 * i / SAMPLE_RATE_HZ + ch) + if (clip_channel == ch and clip_from is not None and i >= clip_from): + value = 900.0 if (i % 2 == 0) else -900.0 + row.append(round(value, 6)) + writer.writerow([f"{t:.9f}", *row]) + t += period + return path + + +def _valid_event() -> dict: + return { + "schema_version": SESSION_EVENT_SCHEMA, + "event_id": "a" * 32, + "kind": "channel_health_change", + "source": { + "recording_sha256": "b" * 64, + "start_sample": 0, + "sample_count": WINDOW, + "channel_order": list(CHANNELS), + "sample_rate_hz": SAMPLE_RATE_HZ, + }, + "observed": {"channel": "TP9", "rms_microvolts": 21.5}, + "baseline_ref": {"start_sample": 0, "sample_count": WINDOW}, + "interpretation": "signal_observation_only", + "stage_claim": None, + "intervention_claim": None, + } + + +def _rejects(record: dict, fragment: str) -> None: + try: + validate_session_event(record) + except ContractError as exc: + assert fragment in str(exc), f"expected {fragment!r} in: {exc}" + return + raise AssertionError(f"expected rejection mentioning {fragment!r}") + + +# ── contract ──────────────────────────────────────────────────────────────── + +def test_valid_event_is_accepted(): + validate_session_event(_valid_event()) + + +def test_rejects_a_stage_claim(): + """The whole point. Detection is gated behind the §21 toolkit gate; an + observation record may not pre-empt it.""" + e = _valid_event(); e["stage_claim"] = "N1" + _rejects(e, "stage_claim must be null") + + +def test_rejects_an_intervention_claim(): + """Efficacy is gated behind the D8 pre-registration.""" + e = _valid_event(); e["intervention_claim"] = "cue_delivered" + _rejects(e, "intervention_claim must be null") + + +def test_rejects_a_smuggled_stage_field_at_any_depth(): + e = _valid_event(); e["observed"]["sleep_stage_hint"] = "n2" + _rejects(e, "never what it might mean") + e = _valid_event(); e["source"]["rem_probability"] = 0.4 + _rejects(e, "never what it might mean") + + +def test_rejects_an_embedding(): + """A reference exists so the artifact survives a change of encoder. Storing + a vector would bind it to one.""" + e = _valid_event(); e["observed"] = {"latent_vector": 1.0} + _rejects(e, "never what it might mean") + + +def test_rejects_non_scalar_observed_value(): + """`observed` carries measurements, not signal.""" + e = _valid_event(); e["observed"]["samples"] = [1.0, 2.0, 3.0] + _rejects(e, "must be a scalar") + + +def test_rejects_a_bad_recording_digest(): + for bad in ("", "not-a-digest", "B" * 64, "b" * 63): + e = _valid_event(); e["source"]["recording_sha256"] = bad + _rejects(e, "SHA-256 hex digest") + + +def test_rejects_a_range_past_the_end_of_the_recording(): + e = _valid_event() + e["source"]["start_sample"] = 900 + e["source"]["sample_count"] = 200 + try: + validate_session_event(e, recording_sample_count=1000) + except ContractError as exc: + assert "exceeds the recording" in str(exc) + return + raise AssertionError("a reference past the end must be rejected") + + +def test_rejects_a_foreign_montage(): + e = _valid_event(); e["source"]["channel_order"] = ["Fz", "Cz", "Pz", "Oz"] + _rejects(e, "channel_order must be exactly") + e = _valid_event(); e["source"]["sample_rate_hz"] = 128 + _rejects(e, "sample_rate_hz must be") + + +def test_rejects_a_reinterpreted_record(): + e = _valid_event(); e["interpretation"] = "sleep_onset_detected" + _rejects(e, "interpretation must be") + + +def test_rejects_unknown_kind(): + e = _valid_event(); e["kind"] = "sleep_onset" + _rejects(e, "kind must be one of") + + +def test_log_rejects_duplicate_ids_and_mixed_recordings(): + a, b = _valid_event(), _valid_event() + try: + validate_session_event_file([a, b]) + except ContractError as exc: + assert "unique" in str(exc) + else: + raise AssertionError("duplicate event_id must be rejected") + + c = copy.deepcopy(_valid_event()) + c["event_id"] = "c" * 32 + c["source"]["recording_sha256"] = "d" * 64 + try: + validate_session_event_file([_valid_event(), c]) + except ContractError as exc: + assert "one log describes one recording" in str(exc) + return + raise AssertionError("a log spanning two recordings must be rejected") + + +# ── determinism ───────────────────────────────────────────────────────────── + +def test_extraction_is_byte_identical_across_runs(): + with tempfile.TemporaryDirectory() as tmp: + d = Path(tmp) / "rec" + _write_recording(d, clip_channel=1, clip_from=9000, gap_at=5000) + first = extractor.extract(d) + second = extractor.extract(d) + dump = lambda evs: "\n".join( + json.dumps(e, sort_keys=True, separators=(",", ":")) for e in evs) + assert dump(first) == dump(second) + assert first, "the fixture should produce events" + + +def test_editing_one_sample_invalidates_every_reference(): + """The digest is the anchor: change the recording and the old references no + longer resolve, rather than silently pointing at different signal.""" + with tempfile.TemporaryDirectory() as tmp: + d = Path(tmp) / "rec" + path = _write_recording(d, clip_channel=1, clip_from=9000) + before = extractor.extract(d) + digest_before = before[0]["source"]["recording_sha256"] + + rows = path.read_text().splitlines() + head, first_row = rows[0], rows[1].split(",") + first_row[1] = f"{float(first_row[1]) + 1.0:.6f}" + rows[1] = ",".join(first_row) + path.write_text("\n".join(rows) + "\n") + + after = extractor.extract(d) + assert after[0]["source"]["recording_sha256"] != digest_before + + +def test_params_digest_travels_with_every_event(): + """Two logs are comparable only if the rules that produced them match.""" + with tempfile.TemporaryDirectory() as tmp: + d = Path(tmp) / "rec" + _write_recording(d, clip_channel=1, clip_from=9000) + events = extractor.extract(d) + expected = extractor.params_digest() + assert all(e["params_sha256"] == expected for e in events) + + +# ── replay ────────────────────────────────────────────────────────────────── + +def test_observed_recomputes_from_the_reference_alone(): + """The load-bearing property: hold the recording, follow the reference, + recompute the measurement, get the same number.""" + with tempfile.TemporaryDirectory() as tmp: + d = Path(tmp) / "rec" + _write_recording(d, clip_channel=1, clip_from=9000) + events = extractor.extract(d) + health = [e for e in events if e["kind"] == "channel_health_change"] + assert health, "the clipped channel should raise a health event" + + _, channels = extractor.read_eeg(d / "eeg.csv") + for event in health[:5]: + src = event["source"] + index = list(CHANNELS).index(event["observed"]["channel"]) + window = channels[index][src["start_sample"]: + src["start_sample"] + src["sample_count"]] + assert round(extractor.rms(window), 6) == event["observed"]["rms_microvolts"] + + +def test_digest_in_the_event_matches_the_file_it_references(): + with tempfile.TemporaryDirectory() as tmp: + d = Path(tmp) / "rec" + path = _write_recording(d, clip_channel=1, clip_from=9000) + events = extractor.extract(d) + on_disk = hashlib.sha256(path.read_bytes()).hexdigest() + assert all(e["source"]["recording_sha256"] == on_disk for e in events) + + +def test_extractor_output_passes_its_own_contract(): + with tempfile.TemporaryDirectory() as tmp: + d = Path(tmp) / "rec" + _write_recording(d, clip_channel=1, clip_from=9000, gap_at=5000) + events = extractor.extract(d) + total = sum(1 for _ in (d / "eeg.csv").read_text().splitlines()) - 1 + validate_session_event_file(events, recording_sample_count=total) + + +def test_a_clean_recording_yields_no_health_or_burst_events(): + """A quiet fixture must stay quiet — otherwise the triggers are noise.""" + with tempfile.TemporaryDirectory() as tmp: + d = Path(tmp) / "rec" + _write_recording(d) + kinds = {e["kind"] for e in extractor.extract(d)} + assert "channel_health_change" not in kinds + assert "artifact_burst" not in kinds + + +def test_a_clock_gap_is_reported_as_zero_throughput(): + with tempfile.TemporaryDirectory() as tmp: + d = Path(tmp) / "rec" + _write_recording(d, gap_at=5000) + gaps = [e for e in extractor.extract(d) if e["kind"] == "zero_throughput"] + assert len(gaps) == 1 + assert gaps[0]["observed"]["missing_samples_estimate"] > 0 + + +# ── detector eligibility ──────────────────────────────────────────────────── + +def test_a_saturated_channel_is_reported_suppressed_not_clean(): + """The blocker. A saturated channel cannot exceed 6x its own rail, so it + produces no artifact_burst — which must not read as "clean". Silence and + ineligibility are different observations.""" + with tempfile.TemporaryDirectory() as tmp: + d = Path(tmp) / "rec" + _write_recording(d, clip_channel=1, clip_from=0) # AF7 saturated throughout + events, manifest = extractor.extract_with_manifest(d) + + af7 = manifest["channels"]["AF7"] + assert af7["detector_status"] == "suppressed" + assert af7["suppressed_reason"] == "channel_saturated" + assert not [e for e in events + if e["kind"] == "artifact_burst" and e["observed"]["channel"] == "AF7"] + + for other in ("TP9", "AF8", "TP10"): + assert manifest["channels"][other]["detector_status"] == "eligible" + assert manifest["channels"][other]["suppressed_reason"] is None + + +def test_manifest_covers_every_channel(): + """A missing channel would make its silence ambiguous again.""" + with tempfile.TemporaryDirectory() as tmp: + d = Path(tmp) / "rec" + _write_recording(d) + _, manifest = extractor.extract_with_manifest(d) + validate_session_manifest(manifest) + assert set(manifest["channels"]) == set(CHANNELS) + + broken = copy.deepcopy(manifest) + del broken["channels"]["AF7"] + try: + validate_session_manifest(broken) + except ContractError as exc: + assert "detector eligibility" in str(exc) + return + raise AssertionError("a manifest missing a channel must be rejected") + + +def test_manifest_dispositions_are_pinned_negative(): + with tempfile.TemporaryDirectory() as tmp: + d = Path(tmp) / "rec" + _write_recording(d) + _, manifest = extractor.extract_with_manifest(d) + for key, expected in REQUIRED_DISPOSITIONS.items(): + assert manifest[key] == expected, key + for key in REQUIRED_DISPOSITIONS: + flipped = copy.deepcopy(manifest) + flipped[key] = True if manifest[key] is False else "promoted" + try: + validate_session_manifest(flipped) + except ContractError: + continue + raise AssertionError(f"{key} must be pinned") + + +def test_manifest_declares_the_half_open_convention(): + with tempfile.TemporaryDirectory() as tmp: + d = Path(tmp) / "rec" + _write_recording(d) + _, manifest = extractor.extract_with_manifest(d) + assert manifest["sample_interval_convention"].startswith("half_open") + + +# ── content addressing ────────────────────────────────────────────────────── + +def test_same_bytes_at_a_different_path_give_identical_references(): + """Content-addressed means content-addressed: the location is not part of + the identity.""" + with tempfile.TemporaryDirectory() as tmp: + a, b = Path(tmp) / "one", Path(tmp) / "two" / "nested" + _write_recording(a, clip_channel=1, clip_from=9000) + b.mkdir(parents=True) + (b / "eeg.csv").write_bytes((a / "eeg.csv").read_bytes()) + dump = lambda evs: "\n".join( + json.dumps(e, sort_keys=True, separators=(",", ":")) for e in evs) + assert dump(extractor.extract(a)) == dump(extractor.extract(b)) + + +def test_params_digest_is_order_independent(): + """Canonical serialisation — otherwise the digest depends on dict order and + two identical parameter sets could disagree.""" + original = extractor.PARAMS + first = extractor.params_digest() + try: + extractor.PARAMS = dict(reversed(list(original.items()))) + assert extractor.params_digest() == first + finally: + extractor.PARAMS = original + + +# ── fail closed on malformed input ────────────────────────────────────────── + +def _expect_exit(directory: Path, fragment: str) -> None: + try: + extractor.extract(directory) + except SystemExit as exc: + assert fragment in str(exc), f"expected {fragment!r} in: {exc}" + return + raise AssertionError(f"expected a failure mentioning {fragment!r}") + + +def test_non_finite_sample_fails_closed(): + """A NaN would otherwise propagate into a baseline median and move every + threshold derived from it.""" + with tempfile.TemporaryDirectory() as tmp: + d = Path(tmp) / "rec" + path = _write_recording(d, samples=2000) + rows = path.read_text().splitlines() + cells = rows[500].split(","); cells[2] = "nan" + rows[500] = ",".join(cells) + path.write_text("\n".join(rows) + "\n") + _expect_exit(d, "non-finite") + + +def test_unparseable_row_fails_closed(): + with tempfile.TemporaryDirectory() as tmp: + d = Path(tmp) / "rec" + path = _write_recording(d, samples=2000) + rows = path.read_text().splitlines() + cells = rows[500].split(","); cells[1] = "n/a" + rows[500] = ",".join(cells) + path.write_text("\n".join(rows) + "\n") + _expect_exit(d, "unparseable row") + + +def test_sample_rate_disagreement_fails_closed(): + """A 128 Hz file indexed under the 256 Hz contract points every event at the + wrong seconds.""" + with tempfile.TemporaryDirectory() as tmp: + d = Path(tmp) / "rec" + path = _write_recording(d, samples=2000) + rows = path.read_text().splitlines() + out = [rows[0]] + t = 1_000_000.0 + for row in rows[1:]: + cells = row.split(",") + out.append(",".join([f"{t:.9f}", *cells[1:]])) + t += 1.0 / 128.0 + path.write_text("\n".join(out) + "\n") + _expect_exit(d, "disagrees with the") + + +def test_missing_channel_column_fails_closed(): + with tempfile.TemporaryDirectory() as tmp: + d = Path(tmp) / "rec" + path = _write_recording(d, samples=1200) + rows = path.read_text().splitlines() + rows[0] = "t_seconds,TP9,AF7,AF8" + path.write_text("\n".join(rows) + "\n") + _expect_exit(d, "missing channel column") + + +if __name__ == "__main__": + failures = 0 + for name, fn in sorted(globals().items()): + if name.startswith("test_") and callable(fn): + try: + fn() + print(f" ok {name}") + except Exception as exc: # noqa: BLE001 - a test runner reports, not raises + failures += 1 + print(f" FAIL {name}: {type(exc).__name__}: {exc}") + total = sum(1 for n, f in globals().items() if n.startswith("test_") and callable(f)) + print(f"\n{total - failures}/{total} passed") + raise SystemExit(1 if failures else 0) diff --git a/docs/architecture/session-events.md b/docs/architecture/session-events.md new file mode 100644 index 0000000..bc5f6b3 --- /dev/null +++ b/docs/architecture/session-events.md @@ -0,0 +1,151 @@ +# Session events — structured references into a recording + +`nc-eeg-session-event-v0`. One JSON line per observation, in +`/session-events.jsonl`. + +## Why a reference and not a copy + +An event carries the SHA-256 of the `eeg.csv` it points into plus a sample +range. It does not carry signal. Anyone holding the recording can re-derive the +window and recompute what was observed; anyone without it learns nothing. + +That indirection is the whole design. It means one artifact serves every +consumer — a spectral routine today, EEGNet or a future encoder later — because +none of them is baked into the record. Storing a latent vector instead would +bind the log to whichever model produced it, and the log would expire when the +model did. + +It also makes the artifact falsifiable. `observed` is recomputable, so a claim in +the log can be checked against the recording rather than believed. + +## What it does not say + +**These are signal observations. Nothing here is a sleep stage, and nothing is an +intervention outcome.** + +- Detection is gated behind the Sleep Validation Toolkit — `SLEEP_CYCLE_DESIGN.md` + §21, *"Gate to Phase C: 5+ clean sessions through the toolkit, no false + readings, all features observable"*, with *"do not start classifier work until + the toolkit is stable"* stated three times in that document. +- Intervention efficacy is gated behind the D8 OSF pre-registration — + `Evaluation/reports/decision_registry.md` entry 8 calls the hypnagogic loop + *"engineering scaffolding, not a validated intervention"* and the + pre-registration *"non-negotiable"*. +- The Muse montage has no chin EMG (`SLEEP_CYCLE_DESIGN.md` §291, risk R1), so + nothing may imply REM. + +This artifact is **upstream of both gates and touches neither.** It shortens +neither: five clean toolkit sessions still means five. But every session run for +that gate yields a structured event log for free, which is the point — the +gates need data, and this is how a night becomes data. + +`Scripts/session_event_contract.py` enforces the boundary rather than leaving it +to review. A record asserting a stage, an efficacy outcome, or an embedding is +rejected, at any nesting depth. + +## Event kinds + +All four are mechanical and recomputable from the referenced window. + +| kind | fires when | +|---|---| +| `zero_throughput` | a gap in the sample clock exceeds `gap_factor` × nominal — the condition `HealthSnapshot`'s `eeg-zero-throughput` raises live, recorded here as a referenceable span | +| `channel_health_change` | a channel's clipped fraction crosses `clip_fraction_trigger`, or its RMS reaches `rms_ratio_trigger` × its in-session baseline | +| `band_excursion` | theta/alpha/beta power reaches `band_ratio_trigger` × its in-session baseline | +| `artifact_burst` | a channel has ≥ `burst_min_samples` above `burst_sigma` × **its own** baseline RMS | + +Baselines are the median over the opening windows of the *same* session. A +cross-session baseline would silently import another night's headset fit. + +### Detector eligibility + +`artifact_burst` is deliberately per-channel. A cross-channel envelope makes one +bad electrode trip it on every window: the first run against a real recording had +a saturated AF7 (~900 µV against ~20 µV elsewhere) producing a burst almost every +second, restating a fault `channel_health_change` already reported. A burst is a +departure from a channel's own norm, so the norm must be its own. + +**A suppressed detector is not a clean channel.** When a channel's baseline is +itself pathological, the comparison is vacuous — nothing exceeds `burst_sigma` × +a rail — so it emits no `artifact_burst`. On a real recording a saturated AF7 did +exactly that. An absent record is ambiguous between *clean*, *no transient*, and +*never eligible*, so eligibility is stated rather than inferred from silence: + +```json +"channels": { + "AF7": {"detector_status": "suppressed", "suppressed_reason": "channel_saturated"}, + "TP9": {"detector_status": "eligible", "suppressed_reason": null} +} +``` + +in `session-events-manifest.json` beside the log. The validator rejects a +manifest that omits any channel, because a missing entry restores the ambiguity. +`suppressed_reason` is one of `channel_saturated` or `channel_silent`. + +## The manifest's dispositions + +All negative, all pinned, all enforced: + +``` +contains_signal false +science_status pipeline_only +label_status heuristic_observation +live_control false +promotion_status not_eligible +clean_session_gate_credited false +``` + +The last is the important one. Running an extractor over a recording credits +nothing toward the §21 five-clean-session gate — whether a session was *clean* is +a judgement about the recording, not about whether a script parsed it. + +`science_status: pipeline_only` and `label_status: heuristic_observation` set the +standard of proof. The 750-event run over a real recording is **pipeline +validation** — evidence the extractor runs, is deterministic, and replays. It is +not evidence that 750 heuristic observations are individually correct. + +## Sample intervals + +Half-open, `[start_sample, start_sample + sample_count)`, non-empty, and within +`recording_sample_count`. The convention is declared in the manifest as +`sample_interval_convention` rather than left to the reader. Events may overlap — +the window stride is shorter than the window — and are not deduplicated: two +kinds firing on one window are two observations of it, and `event_id` separates +them because the kind and the observation are both in the hash. + +## Determinism + +The same file yields byte-identical output. Nothing consults the clock, the +filesystem order, or a random seed. `event_id` is derived from content, so +re-running produces the same ids and a diff shows only real change. + +Every record carries `params_sha256` over the pinned `PARAMS` block. Two logs are +comparable only if that digest matches — changing a threshold changes the events, +and the record says so. + +Editing one sample changes `recording_sha256`, which invalidates every reference +into that file rather than silently re-pointing it at different signal. + +## Usage + +```sh +python3 Scripts/extract_session_events.py Recordings/ +python3 Scripts/extract_session_events.py Recordings/ --stdout +python3 Tests/eval/test_session_events.py +``` + +Standard library only — no numpy. Band power uses Goertzel over the integer bins +of each band, which is O(n) per bin where a hand-rolled DFT would be O(n²) per +window. The system interpreter on a clean runner has no numpy, and this needs to +stay runnable there. + +## Relationship to `JEPATransition` + +`Sources/BCICore/Models/JEPATransition.swift` stores band energies plus +per-channel RMS² — compact derived state, the right shape for its own purpose and +the wrong one for anything wanting raw signal. It is unchanged; this is additive. + +A future action-conditioned artifact can pair a session-event reference with an +action vector, and any encoder derives its own states offline from the same +recording. That is the same model-agnostic reference the encoder research track +needs, reached from the capture side.