From 6d9d8c9e502bbd280a4fe83a11fb739f25d4e630 Mon Sep 17 00:00:00 2001
From: Benjamin Pettit <benjamin.pettit@digimulti.com>
Date: Mon, 6 Jul 2026 22:51:51 +1000
Subject: [PATCH 1/4] feat(experiments): unify event-marker emission behind
 push_marker()/subscribe_marker()
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Route all stimulus-marker emission through a single push_marker() on
BaseExperiment, backed by a subscriber list. One call fans the marker out to the
EEG board, every auxiliary device, and any optional observers under one shared
timestamp — replacing the per-paradigm boilerplate that re-derived the timestamp
and branched on eeg.backend to wrap muse markers.

- subscribe_marker(callback, raise_on_error=True): register any
  callable(marker, timestamp, trial_idx). raise_on_error=True aborts the run on
  failure (essential recorders); False logs-and-swallows (optional telemetry),
  logging each broken subscriber's traceback only once to keep per-onset log I/O
  off the presentation thread.
- push_marker(marker, trial_idx=None) -> float: emit under one shared timestamp
  and return it so callers can stamp their own bookkeeping.
- Built-in essential subscribers _emit_to_eeg / _emit_to_devices preserve the
  existing self.devices contract (dev.push_sample(marker=, timestamp=)).
- muse scalar->[marker] wrap now lives once in EEG._muse_push_sample instead of
  being duplicated across paradigms.
- eoec: LSL outlet + serial trigger box subscribe as inline lambdas; the serial
  write rides push_marker's shared timestamp (best-effort, raise_on_error=False).
- SSVEP emits on the first flip so the recorded onset is the true stimulus onset.

Removes BaseExperiment.send_triggers() (folded into _emit_to_devices).
Adds tests/test_push_marker.py.
---
 eegnb/devices/eeg.py                    |   2 +
 eegnb/experiments/Experiment.py         |  83 ++++++++++-
 eegnb/experiments/rest/eoec.py          |  28 ++--
 eegnb/experiments/visual_n170/n170.py   |  21 +--
 eegnb/experiments/visual_ssvep/ssvep.py |  18 +--
 tests/test_push_marker.py               | 177 ++++++++++++++++++++++++
 6 files changed, 275 insertions(+), 54 deletions(-)
 create mode 100644 tests/test_push_marker.py

diff --git a/eegnb/devices/eeg.py b/eegnb/devices/eeg.py
index 6c017ed21..889f78cdd 100644
--- a/eegnb/devices/eeg.py
+++ b/eegnb/devices/eeg.py
@@ -185,6 +185,8 @@ def _stop_muse(self):
         pass
 
     def _muse_push_sample(self, marker, timestamp):
+        if not isinstance(marker, (list, tuple)):
+            marker = [marker]
         self.muse_StreamOutlet.push_sample(marker, timestamp)
 
     def _muse_get_recent(self, n_samples: int = 256, restart_inlet: bool = False):
diff --git a/eegnb/experiments/Experiment.py b/eegnb/experiments/Experiment.py
index 01427b871..526ecaf5b 100644
--- a/eegnb/experiments/Experiment.py
+++ b/eegnb/experiments/Experiment.py
@@ -16,6 +16,7 @@
 
 import gc
 from time import time
+import logging
 import random
 import json
 
@@ -24,11 +25,13 @@
 
 from eegnb import generate_save_fn
 
+logger = logging.getLogger(__name__)
+
 
 class BaseExperiment(ABC):
 
     def __init__(self, exp_name, duration, eeg, save_fn, n_trials: int, iti: float, soa: float, jitter: float,
-                 use_vr=False, use_fullscr = True, screen_num=0, stereoscopic = False, devices = list):
+                 use_vr=False, use_fullscr = True, screen_num=0, stereoscopic = False, devices=None):
         """ Initializer for the Base Experiment Class
 
         Args:
@@ -51,7 +54,7 @@ def __init__(self, exp_name, duration, eeg, save_fn, n_trials: int, iti: float,
         Press spacebar to continue. \n""".format(self.exp_name)
         self.duration = duration
         self.eeg: EEG = eeg
-        self.devices = devices
+        self.devices = devices if devices is not None else []
         self.save_fn = save_fn
         self.n_trials = n_trials
         self.iti = iti
@@ -77,6 +80,16 @@ def __init__(self, exp_name, duration, eeg, save_fn, n_trials: int, iti: float,
         # Initializing the marker names
         self.markernames = [1, 2]
 
+        # (callback, raise_on_error) pairs invoked on every push_marker().
+        self.marker_subscribers: list = []
+
+        # Per-subscriber failure count: broken subscribers log once, not per trial.
+        self._subscriber_failures: dict = {}
+        
+        # Setting event marker subscribers
+        self.subscribe_marker(self._emit_to_eeg)
+        self.subscribe_marker(self._emit_to_devices)
+
         # Setting up the trial and parameter list
         self.parameter = np.random.binomial(1, 0.5, self.n_trials)
         self.trials = DataFrame(dict(parameter=self.parameter, timestamp=np.zeros(self.n_trials)))
@@ -361,6 +374,10 @@ def run(self, instructions=True):
         # Clearing the screen for the next trial
         event.clearEvents()
 
+        # Surface any optional-subscriber failures whose repeats were suppressed.
+        for callback, count in self._subscriber_failures.items():
+            logger.warning("marker subscriber %s failed on %d marker(s)", callback, count)
+
         # Closing the EEG stream
         if self.eeg:
             self.eeg.stop()
@@ -371,12 +388,68 @@ def run(self, instructions=True):
         # Closing the window
         self.window.close()
 
-    def send_triggers(self, marker):
-        """Send timing triggers to recording device[s]"""
+    def subscribe_marker(self, callback, raise_on_error=True):
+        """Register a marker subscriber: callable(marker, timestamp, trial_idx).
+
+        Invoked on every push_marker(). raise_on_error selects how a raised
+        exception is handled:
+
+        raise_on_error=True (the default) — essential recorders (the EEG board,
+            trigger/serial boxes) where a lost marker in the primary data is
+            unrecoverable. An exception propagates and aborts the run.
+        raise_on_error=False — optional telemetry (flip-time, eyetracker,
+            photodiode) that must never block core data. Exceptions are logged
+            and swallowed, so a bad observer can't take down a live recording.
+            To avoid spamming the log — and adding I/O latency at each marker
+            onset — a given subscriber's traceback is logged only on its first
+            failure; further failures are counted and the per-subscriber total
+            is reported at the end of run().
+        """
+        self.marker_subscribers.append((callback, raise_on_error))
+
+    def _emit_to_eeg(self, marker, timestamp, trial_idx):
+        """Built-in essential subscriber: record the marker on the EEG board."""
+        if self.eeg:
+            self.eeg.push_sample(marker=marker, timestamp=timestamp)
+
+    def _emit_to_devices(self, marker, timestamp, trial_idx):
+        """Built-in essential subscriber: record the marker on every aux device."""
         for dev in self.devices:
-            timestamp = time()
             dev.push_sample(marker=marker, timestamp=timestamp)
 
+    def push_marker(self, marker, trial_idx=None):
+        """Emit a stimulus marker to every subscriber under one shared timestamp.
+
+        A subscriber is any callable(marker, timestamp, trial_idx) registered via
+        subscribe_marker(). The built-in essential subscribers record the marker
+        onto the EEG board and any auxiliary devices (self.eeg, self.devices);
+        optional subscribers (raise_on_error=False) observe timing without emitting —
+        flip-time telemetry, eyetracker fixation, photodiode metadata — and their
+        exceptions are logged, not raised.
+
+        All subscribers share one timestamp so the marker onset is consistent
+        across streams. trial_idx is context for observers; callers without
+        observers may omit it. Returns the shared timestamp so a caller can record
+        the same onset in its own bookkeeping (e.g. a per-trial timestamp column)
+        without minting a second, slightly-skewed time().
+        """
+        timestamp = time()
+        for callback, raise_on_error in self.marker_subscribers:
+            try:
+                callback(marker, timestamp, trial_idx)
+            except Exception:
+                if raise_on_error:
+                    raise
+                # Log the traceback only on first failure (see subscribe_marker).
+                count = self._subscriber_failures.get(callback, 0) + 1
+                self._subscriber_failures[callback] = count
+                if count == 1:
+                    logger.exception(
+                        "marker subscriber failed; suppressing further tracebacks for it: %s",
+                        callback,
+                    )
+        return timestamp
+
     @property
     def name(self) -> str:
         """ This experiment's name """
diff --git a/eegnb/experiments/rest/eoec.py b/eegnb/experiments/rest/eoec.py
index 38ed467db..b7140f063 100644
--- a/eegnb/experiments/rest/eoec.py
+++ b/eegnb/experiments/rest/eoec.py
@@ -1,6 +1,5 @@
 
 from typing import Optional
-from time import time
 
 from psychopy import prefs
 
@@ -87,11 +86,18 @@ def setup(self, instructions: bool = True):
         # LSL outlet for markers
         info = StreamInfo("Markers", "Markers", 1, 0, "int32", "eyeclosure-baseline")
         self.outlet = StreamOutlet(info)
+        self.subscribe_marker(
+            lambda marker, timestamp, _idx: self.outlet.push_sample([marker], timestamp)
+        )
 
         # serial connection for hardware triggers
         if self.serial_port and serial is not None:
             try:
                 self.serial = serial.Serial(self.serial_port, 115200, timeout=1)
+                self.subscribe_marker(
+                    lambda marker, timestamp, _idx: self.serial.write(bytes([marker])),
+                    raise_on_error=False,
+                )
             except Exception:  # pragma: no cover
                 self.serial = None
 
@@ -110,26 +116,8 @@ def present_stimulus(self, idx: int):
 
         label = self.trials["parameter"].iloc[idx]  # 0 open, 1 closed
         if self.trials["timestamp"].iloc[idx] == 0:
-            timestamp = time()
+            timestamp = self.push_marker(self.markernames[label], idx)
             self.trials.at[idx, "timestamp"] = timestamp
-            self.outlet.push_sample([self.markernames[label]], timestamp)
-            
-            if self.eeg:
-              if self.eeg.backend == "muselsl":
-                  marker = [self.markernames[label]]
-              else:
-                  marker = self.markernames[label]
-              self.eeg.push_sample(marker=marker, timestamp=timestamp)
-            
-            if self.devices:
-              marker = self.markernames[label]
-              self.send_triggers(marker)
-
-            if self.serial:
-                try:
-                    self.serial.write(bytes([self.markernames[label]]))
-                except Exception:  # pragma: no cover
-                    pass
 
             if label == 0:
                 self.open_sound.play()
diff --git a/eegnb/experiments/visual_n170/n170.py b/eegnb/experiments/visual_n170/n170.py
index 231381ff7..09c397af3 100644
--- a/eegnb/experiments/visual_n170/n170.py
+++ b/eegnb/experiments/visual_n170/n170.py
@@ -1,7 +1,6 @@
 """  eeg-notebooks/eegnb/experiments/visual_n170/n170.py """
 
 import os
-from time import time
 from glob import glob
 from random import choice
 from psychopy import visual, core, event
@@ -44,23 +43,9 @@ def present_stimulus(self, idx: int):
         # Draw the image
         image.draw()
 
-
-        # Pushing the sample to the EEG
-        if self.eeg:
-            timestamp = time()
-
-            if self.eeg.backend == "muselsl":
-                marker = [self.markernames[label]]
-            else:
-                marker = self.markernames[label]
-            
-            self.eeg.push_sample(marker=marker, timestamp=timestamp)
-      
-
-        if self.devices:
-            marker = self.markernames[label]
-            self.send_triggers(marker)
-
+        # Emit the trial marker to the EEG stream and any auxiliary devices
+        if self.eeg or self.devices:
+            self.push_marker(self.markernames[label], idx)
 
         self.window.flip()
 
diff --git a/eegnb/experiments/visual_ssvep/ssvep.py b/eegnb/experiments/visual_ssvep/ssvep.py
index 94ae91218..87fb98f29 100644
--- a/eegnb/experiments/visual_ssvep/ssvep.py
+++ b/eegnb/experiments/visual_ssvep/ssvep.py
@@ -1,7 +1,6 @@
 
 from eegnb.experiments import Experiment
 import os
-from time import time
 from glob import glob
 from random import choice
 
@@ -95,17 +94,11 @@ def present_stimulus(self, idx: int):
 
         # Select stimulus frequency
         ind = self.trials["parameter"].iloc[idx]
+        marker = self.markernames[ind]
+        marker_pushed = False
 
-        # Push sample
-        if self.eeg:
-            timestamp = time()
-            if self.eeg.backend == "muselsl":
-                marker = [self.markernames[ind]]
-            else:
-                marker = self.markernames[ind]
-            self.eeg.push_sample(marker=marker, timestamp=timestamp)
-
-        # Present flickering stim
+        # Present flickering stim; emit the marker on the first flip so the
+        # captured onset is the true stimulus onset (not one frame early).
         for _ in range(int(self.stim_patterns[ind]["n_cycles"])):
 
             for _ in range(int(self.stim_patterns[ind]["cycle"][0])):
@@ -116,6 +109,9 @@ def present_stimulus(self, idx: int):
                 self.grating.draw()
                 self.fixation.draw()
                 self.window.flip()
+                if not marker_pushed:
+                    self.push_marker(marker, idx)
+                    marker_pushed = True
 
             for _ in range(self.stim_patterns[ind]["cycle"][1]):
                 if self.use_vr:
diff --git a/tests/test_push_marker.py b/tests/test_push_marker.py
new file mode 100644
index 000000000..2ed36fef9
--- /dev/null
+++ b/tests/test_push_marker.py
@@ -0,0 +1,177 @@
+"""Integration tests for the push_marker() event-marker abstraction.
+
+Covers the runnable, display-free subset:
+  1. one shared timestamp across all subscribers (built-in eeg + devices)
+  2. optional-subscriber notification + raise_on_error isolation (both directions)
+  3. the muse scalar->[marker] wrap guard (back-compat with pre-wrapped lists)
+
+The SSVEP first-flip onset test (#4) and full-paradigm smoke (#5) need a real
+PsychoPy window / board and are intentionally not here.
+"""
+
+import logging
+
+import pytest
+
+from eegnb.experiments.Experiment import BaseExperiment
+from eegnb.devices.eeg import EEG
+
+
+class FakeEmitter:
+    """Stands in for self.eeg / a self.devices entry; records push_sample calls."""
+
+    def __init__(self, log):
+        self.log = log
+
+    def push_sample(self, marker, timestamp):
+        self.log.append((marker, timestamp))
+
+
+class _StubExperiment(BaseExperiment):
+    """Concrete BaseExperiment: satisfies the ABC without a real paradigm."""
+
+    def load_stimulus(self):
+        pass
+
+    def present_stimulus(self, *a, **k):
+        pass
+
+
+def _bare_experiment():
+    """A BaseExperiment without running __init__ (no PsychoPy window needed)."""
+    exp = object.__new__(_StubExperiment)
+    exp.eeg = None
+    exp.devices = []
+    exp.marker_subscribers = []
+    exp._subscriber_failures = {}
+    # the built-in essential recorders __init__ would have registered
+    exp.subscribe_marker(exp._emit_to_eeg)
+    exp.subscribe_marker(exp._emit_to_devices)
+    return exp
+
+
+# 1 ---------------------------------------------------------------------------
+def test_push_marker_one_timestamp_across_emitters():
+    log = []
+    exp = _bare_experiment()
+    exp.eeg = FakeEmitter(log)
+    exp.devices = [FakeEmitter(log), FakeEmitter(log)]
+
+    exp.push_marker(1, 0)
+
+    assert len(log) == 3                       # eeg + both devices fired
+    assert {m for m, _ in log} == {1}          # all got the same marker value
+    assert len({t for _, t in log}) == 1       # ...under ONE shared timestamp
+
+
+def test_push_marker_devices_only_no_eeg():
+    log = []
+    exp = _bare_experiment()
+    exp.devices = [FakeEmitter(log)]
+    exp.push_marker(2, 5)
+    assert log == [(2, log[0][1])]             # eeg=None tolerated, device still fired
+
+
+# 2 ---------------------------------------------------------------------------
+def test_optional_subscriber_receives_marker_timestamp_trial_idx():
+    emit_log, seen = [], []
+    exp = _bare_experiment()
+    exp.eeg = FakeEmitter(emit_log)
+    exp.subscribe_marker(
+        lambda marker, timestamp, trial_idx: seen.append((marker, timestamp, trial_idx)),
+        raise_on_error=False,
+    )
+
+    exp.push_marker(1, 7)
+
+    emitted_ts = emit_log[0][1]
+    assert seen == [(1, emitted_ts, 7)]        # subscriber got (marker, same ts, trial_idx)
+
+
+def test_optional_subscriber_exception_is_isolated(caplog):
+    emit_log, seen = [], []
+    exp = _bare_experiment()
+    exp.eeg = FakeEmitter(emit_log)
+
+    def boom(marker, timestamp, trial_idx):
+        raise RuntimeError("bad observer")
+
+    exp.subscribe_marker(boom, raise_on_error=False)
+    exp.subscribe_marker(
+        lambda marker, timestamp, trial_idx: seen.append((trial_idx, timestamp)),
+        raise_on_error=False,
+    )
+
+    with caplog.at_level(logging.ERROR):
+        exp.push_marker(1, 3)                  # must NOT raise
+
+    assert emit_log, "emitter still fired despite a failing optional subscriber"
+    assert seen == [(3, emit_log[0][1])]       # later subscriber still ran
+    assert "marker subscriber failed" in caplog.text
+
+
+def test_repeated_optional_failure_logs_once(caplog):
+    """A subscriber failing every trial logs its traceback once, but counts all failures."""
+    exp = _bare_experiment()
+    exp.eeg = FakeEmitter([])
+
+    def boom(marker, timestamp, trial_idx):
+        raise RuntimeError("bad observer")
+
+    exp.subscribe_marker(boom, raise_on_error=False)
+
+    with caplog.at_level(logging.ERROR):
+        for i in range(3):
+            exp.push_marker(1, i)
+
+    logged = [r for r in caplog.records if "suppressing further tracebacks" in r.getMessage()]
+    assert len(logged) == 1                    # logged once across 3 failures
+    assert exp._subscriber_failures[boom] == 3 # ...but all counted
+
+
+def test_raise_on_error_subscriber_propagates_and_halts():
+    """An essential (raise_on_error=True) subscriber that raises aborts the dispatch:
+    the exception propagates and later subscribers do not run."""
+    exp = _bare_experiment()
+    seen = []
+
+    def boom(marker, timestamp, trial_idx):
+        raise RuntimeError("board died")
+
+    exp.subscribe_marker(boom, raise_on_error=True)
+    exp.subscribe_marker(
+        lambda marker, timestamp, trial_idx: seen.append(trial_idx), raise_on_error=False
+    )
+
+    with pytest.raises(RuntimeError, match="board died"):
+        exp.push_marker(1, 0)
+    assert seen == [], "subscribers after a raise_on_error failure do not run"
+
+
+# 3 ---------------------------------------------------------------------------
+class FakeOutlet:
+    def __init__(self):
+        self.last = None
+
+    def push_sample(self, marker, timestamp):
+        self.last = marker
+
+
+def _bare_eeg(outlet):
+    eeg = object.__new__(EEG)
+    eeg.muse_StreamOutlet = outlet
+    return eeg
+
+
+def test_muse_wraps_scalar_marker():
+    outlet = FakeOutlet()
+    eeg = _bare_eeg(outlet)
+    eeg._muse_push_sample(2, 123.0)
+    assert outlet.last == [2]                  # scalar -> single-channel vector
+
+
+def test_muse_passes_prewrapped_list_unchanged():
+    outlet = FakeOutlet()
+    eeg = _bare_eeg(outlet)
+    eeg._muse_push_sample([2], 123.0)          # P300 back-compat path
+    assert outlet.last == [2]                  # no double-wrap

From 8b9ee02f6632120b211a1f93d1fa217617f596bc Mon Sep 17 00:00:00 2001
From: Benjamin Pettit <benjamin.pettit@digimulti.com>
Date: Wed, 8 Jul 2026 21:31:34 +1000
Subject: [PATCH 2/4] refactor(experiments): finish push_marker migration and
 marker-consistent naming

- Convert p300 and n170 to push_marker(): drop the pre-abstraction self.eeg
  guards and p300's inline muse-wrap/manual timestamp (the wrap now lives in
  EEG._muse_push_sample). p300 was the last raw-push holdout.
- Rename built-in subscribers _emit_to_eeg/_emit_to_devices ->
  _write_marker_to_eeg/_write_marker_to_devices and purge "emit" from
  docstrings/comments in favour of "marker". Note "trigger" as the ERP synonym
  and leave a send_triggers breadcrumb in the push_marker docstring.
- Simplify test_push_marker.py (real construction via a fixture instead of
  object.__new__; one fake sink) and move the muse-wrap unit test to
  test_acquisition.py, alongside the other EEG-device tests.
---
 eegnb/experiments/Experiment.py         |  18 +--
 eegnb/experiments/visual_n170/n170.py   |   4 +-
 eegnb/experiments/visual_p300/p300.py   |  10 +-
 eegnb/experiments/visual_ssvep/ssvep.py |   2 +-
 tests/test_acquisition.py               |  22 +++
 tests/test_push_marker.py               | 174 +++++++-----------------
 6 files changed, 80 insertions(+), 150 deletions(-)

diff --git a/eegnb/experiments/Experiment.py b/eegnb/experiments/Experiment.py
index 526ecaf5b..2baa143b8 100644
--- a/eegnb/experiments/Experiment.py
+++ b/eegnb/experiments/Experiment.py
@@ -87,8 +87,8 @@ def __init__(self, exp_name, duration, eeg, save_fn, n_trials: int, iti: float,
         self._subscriber_failures: dict = {}
         
         # Setting event marker subscribers
-        self.subscribe_marker(self._emit_to_eeg)
-        self.subscribe_marker(self._emit_to_devices)
+        self.subscribe_marker(self._write_marker_to_eeg)
+        self.subscribe_marker(self._write_marker_to_devices)
 
         # Setting up the trial and parameter list
         self.parameter = np.random.binomial(1, 0.5, self.n_trials)
@@ -407,25 +407,25 @@ def subscribe_marker(self, callback, raise_on_error=True):
         """
         self.marker_subscribers.append((callback, raise_on_error))
 
-    def _emit_to_eeg(self, marker, timestamp, trial_idx):
+    def _write_marker_to_eeg(self, marker, timestamp, trial_idx):
         """Built-in essential subscriber: record the marker on the EEG board."""
         if self.eeg:
             self.eeg.push_sample(marker=marker, timestamp=timestamp)
 
-    def _emit_to_devices(self, marker, timestamp, trial_idx):
+    def _write_marker_to_devices(self, marker, timestamp, trial_idx):
         """Built-in essential subscriber: record the marker on every aux device."""
         for dev in self.devices:
             dev.push_sample(marker=marker, timestamp=timestamp)
 
     def push_marker(self, marker, trial_idx=None):
-        """Emit a stimulus marker to every subscriber under one shared timestamp.
+        """Send a stimulus marker to every subscriber under one shared timestamp.
 
         A subscriber is any callable(marker, timestamp, trial_idx) registered via
-        subscribe_marker(). The built-in essential subscribers record the marker
+        subscribe_marker(). The built-in essential subscribers write the marker
         onto the EEG board and any auxiliary devices (self.eeg, self.devices);
-        optional subscribers (raise_on_error=False) observe timing without emitting —
-        flip-time telemetry, eyetracker fixation, photodiode metadata — and their
-        exceptions are logged, not raised.
+        optional subscribers (raise_on_error=False) observe onset timing without
+        writing a marker to any stream — flip-time telemetry, eyetracker fixation,
+        photodiode metadata — and their exceptions are logged, not raised.
 
         All subscribers share one timestamp so the marker onset is consistent
         across streams. trial_idx is context for observers; callers without
diff --git a/eegnb/experiments/visual_n170/n170.py b/eegnb/experiments/visual_n170/n170.py
index 09c397af3..849dda8a7 100644
--- a/eegnb/experiments/visual_n170/n170.py
+++ b/eegnb/experiments/visual_n170/n170.py
@@ -43,9 +43,7 @@ def present_stimulus(self, idx: int):
         # Draw the image
         image.draw()
 
-        # Emit the trial marker to the EEG stream and any auxiliary devices
-        if self.eeg or self.devices:
-            self.push_marker(self.markernames[label], idx)
+        self.push_marker(self.markernames[label], idx)
 
         self.window.flip()
 
diff --git a/eegnb/experiments/visual_p300/p300.py b/eegnb/experiments/visual_p300/p300.py
index d6a8a1df3..3265cd44c 100644
--- a/eegnb/experiments/visual_p300/p300.py
+++ b/eegnb/experiments/visual_p300/p300.py
@@ -2,7 +2,6 @@
 """  eeg-notebooks/eegnb/experiments/visual_p300/p300.py """
 
 import os
-from time import time
 from glob import glob
 from random import choice
 from optparse import OptionParser
@@ -41,13 +40,6 @@ def present_stimulus(self, idx: int):
         image = choice(self.targets if label == 1 else self.nontargets)
         image.draw()
 
-        # Push sample
-        if self.eeg:
-            timestamp = time()
-            if self.eeg.backend == "muselsl":
-                marker = [self.markernames[label]]
-            else:
-                marker = self.markernames[label]
-            self.eeg.push_sample(marker=marker, timestamp=timestamp)
+        self.push_marker(self.markernames[label], idx)
 
         self.window.flip()
\ No newline at end of file
diff --git a/eegnb/experiments/visual_ssvep/ssvep.py b/eegnb/experiments/visual_ssvep/ssvep.py
index 87fb98f29..60be4d023 100644
--- a/eegnb/experiments/visual_ssvep/ssvep.py
+++ b/eegnb/experiments/visual_ssvep/ssvep.py
@@ -97,7 +97,7 @@ def present_stimulus(self, idx: int):
         marker = self.markernames[ind]
         marker_pushed = False
 
-        # Present flickering stim; emit the marker on the first flip so the
+        # Present flickering stim; push the marker on the first flip so the
         # captured onset is the true stimulus onset (not one frame early).
         for _ in range(int(self.stim_patterns[ind]["n_cycles"])):
 
diff --git a/tests/test_acquisition.py b/tests/test_acquisition.py
index 5626f6ac0..efc44f6f7 100644
--- a/tests/test_acquisition.py
+++ b/tests/test_acquisition.py
@@ -54,3 +54,25 @@ def test_synthetic_acquisition(tmp_path):
     assert (data['stim'] != 0).any()
     
     print(f"Acquired {len(data)} samples with columns: {list(data.columns)}")
+
+
+def test_muse_push_sample_wraps_a_scalar_but_leaves_a_list_alone():
+    """EEG._muse_push_sample sends a single-channel [marker] to the LSL outlet:
+    a scalar is wrapped, an already-wrapped list passes through (P300 path)."""
+
+    class FakeOutlet:
+        def __init__(self):
+            self.last = None
+
+        def push_sample(self, marker, timestamp):
+            self.last = marker
+
+    outlet = FakeOutlet()
+    eeg = object.__new__(EEG)                 # skip __init__ (it would connect hardware)
+    eeg.muse_StreamOutlet = outlet
+
+    eeg._muse_push_sample(2, 123.0)
+    assert outlet.last == [2]                 # scalar -> single-channel list
+
+    eeg._muse_push_sample([2], 123.0)
+    assert outlet.last == [2]                 # already a list -> unchanged
diff --git a/tests/test_push_marker.py b/tests/test_push_marker.py
index 2ed36fef9..ddd6e219f 100644
--- a/tests/test_push_marker.py
+++ b/tests/test_push_marker.py
@@ -1,177 +1,95 @@
-"""Integration tests for the push_marker() event-marker abstraction.
-
-Covers the runnable, display-free subset:
-  1. one shared timestamp across all subscribers (built-in eeg + devices)
-  2. optional-subscriber notification + raise_on_error isolation (both directions)
-  3. the muse scalar->[marker] wrap guard (back-compat with pre-wrapped lists)
-
-The SSVEP first-flip onset test (#4) and full-paradigm smoke (#5) need a real
-PsychoPy window / board and are intentionally not here.
-"""
+"""Tests for BaseExperiment.push_marker()."""
 
 import logging
 
 import pytest
 
 from eegnb.experiments.Experiment import BaseExperiment
-from eegnb.devices.eeg import EEG
 
 
-class FakeEmitter:
-    """Stands in for self.eeg / a self.devices entry; records push_sample calls."""
+class Recorder:
+    """Fake eeg / device: remembers every push_sample(marker, timestamp)."""
 
-    def __init__(self, log):
-        self.log = log
+    def __init__(self):
+        self.calls = []
 
     def push_sample(self, marker, timestamp):
-        self.log.append((marker, timestamp))
+        self.calls.append((marker, timestamp))
 
 
-class _StubExperiment(BaseExperiment):
-    """Concrete BaseExperiment: satisfies the ABC without a real paradigm."""
+class StubExperiment(BaseExperiment):
+    """Minimal concrete BaseExperiment — no real stimulus, opens no window."""
 
     def load_stimulus(self):
         pass
 
-    def present_stimulus(self, *a, **k):
+    def present_stimulus(self, idx):
         pass
 
 
-def _bare_experiment():
-    """A BaseExperiment without running __init__ (no PsychoPy window needed)."""
-    exp = object.__new__(_StubExperiment)
-    exp.eeg = None
-    exp.devices = []
-    exp.marker_subscribers = []
-    exp._subscriber_failures = {}
-    # the built-in essential recorders __init__ would have registered
-    exp.subscribe_marker(exp._emit_to_eeg)
-    exp.subscribe_marker(exp._emit_to_devices)
-    return exp
+@pytest.fixture
+def exp():
+    # Real __init__ runs (it opens no window) and registers the two built-in
+    # subscribers that write the marker to self.eeg and self.devices.
+    return StubExperiment("test", duration=1, eeg=None, save_fn=None,
+                          n_trials=1, iti=0, soa=0, jitter=0)
 
 
-# 1 ---------------------------------------------------------------------------
-def test_push_marker_one_timestamp_across_emitters():
-    log = []
-    exp = _bare_experiment()
-    exp.eeg = FakeEmitter(log)
-    exp.devices = [FakeEmitter(log), FakeEmitter(log)]
+def boom(marker, timestamp, trial_idx):
+    raise RuntimeError("boom")
 
-    exp.push_marker(1, 0)
 
-    assert len(log) == 3                       # eeg + both devices fired
-    assert {m for m, _ in log} == {1}          # all got the same marker value
-    assert len({t for _, t in log}) == 1       # ...under ONE shared timestamp
+def test_marker_reaches_eeg_and_every_device_under_one_timestamp(exp):
+    exp.eeg = Recorder()
+    exp.devices = [Recorder(), Recorder()]
 
+    exp.push_marker(1)
 
-def test_push_marker_devices_only_no_eeg():
-    log = []
-    exp = _bare_experiment()
-    exp.devices = [FakeEmitter(log)]
-    exp.push_marker(2, 5)
-    assert log == [(2, log[0][1])]             # eeg=None tolerated, device still fired
+    ts = exp.eeg.calls[0][1]              # the single timestamp push_marker minted
+    for s in [exp.eeg, *exp.devices]:
+        assert s.calls == [(1, ts)]       # each sink recorded marker 1 at that timestamp
 
 
-# 2 ---------------------------------------------------------------------------
-def test_optional_subscriber_receives_marker_timestamp_trial_idx():
-    emit_log, seen = [], []
-    exp = _bare_experiment()
-    exp.eeg = FakeEmitter(emit_log)
-    exp.subscribe_marker(
-        lambda marker, timestamp, trial_idx: seen.append((marker, timestamp, trial_idx)),
-        raise_on_error=False,
-    )
+def test_push_marker_tolerates_no_eeg(exp):
+    exp.devices = [Recorder()]                # exp.eeg stays None
 
-    exp.push_marker(1, 7)
+    exp.push_marker(2)                        # must not crash on eeg=None
 
-    emitted_ts = emit_log[0][1]
-    assert seen == [(1, emitted_ts, 7)]        # subscriber got (marker, same ts, trial_idx)
+    assert exp.devices[0].calls[0][0] == 2
 
 
-def test_optional_subscriber_exception_is_isolated(caplog):
-    emit_log, seen = [], []
-    exp = _bare_experiment()
-    exp.eeg = FakeEmitter(emit_log)
-
-    def boom(marker, timestamp, trial_idx):
-        raise RuntimeError("bad observer")
-
+def test_failing_optional_subscriber_does_not_stop_the_others(exp, caplog):
+    exp.eeg = Recorder()
+    seen = []
     exp.subscribe_marker(boom, raise_on_error=False)
-    exp.subscribe_marker(
-        lambda marker, timestamp, trial_idx: seen.append((trial_idx, timestamp)),
-        raise_on_error=False,
-    )
+    exp.subscribe_marker(lambda *args: seen.append(args), raise_on_error=False)
 
     with caplog.at_level(logging.ERROR):
-        exp.push_marker(1, 3)                  # must NOT raise
+        exp.push_marker(1, trial_idx=3)       # boom is swallowed, not raised
 
-    assert emit_log, "emitter still fired despite a failing optional subscriber"
-    assert seen == [(3, emit_log[0][1])]       # later subscriber still ran
+    ts = exp.eeg.calls[0][1]
+    assert exp.eeg.calls == [(1, ts)]         # essential subscriber still wrote the marker
+    assert seen == [(1, ts, 3)]               # later optional subscriber still ran
     assert "marker subscriber failed" in caplog.text
 
 
-def test_repeated_optional_failure_logs_once(caplog):
-    """A subscriber failing every trial logs its traceback once, but counts all failures."""
-    exp = _bare_experiment()
-    exp.eeg = FakeEmitter([])
-
-    def boom(marker, timestamp, trial_idx):
-        raise RuntimeError("bad observer")
-
+def test_repeatedly_failing_subscriber_logs_once_but_counts_all(exp, caplog):
     exp.subscribe_marker(boom, raise_on_error=False)
 
     with caplog.at_level(logging.ERROR):
         for i in range(3):
-            exp.push_marker(1, i)
+            exp.push_marker(1, trial_idx=i)
 
     logged = [r for r in caplog.records if "suppressing further tracebacks" in r.getMessage()]
-    assert len(logged) == 1                    # logged once across 3 failures
-    assert exp._subscriber_failures[boom] == 3 # ...but all counted
+    assert len(logged) == 1                   # traceback logged once...
+    assert exp._subscriber_failures[boom] == 3  # ...but every failure counted
 
 
-def test_raise_on_error_subscriber_propagates_and_halts():
-    """An essential (raise_on_error=True) subscriber that raises aborts the dispatch:
-    the exception propagates and later subscribers do not run."""
-    exp = _bare_experiment()
+def test_essential_subscriber_failure_propagates_and_halts_dispatch(exp):
     seen = []
+    exp.subscribe_marker(boom, raise_on_error=True)     # essential: failure must surface
+    exp.subscribe_marker(lambda *args: seen.append(args), raise_on_error=False)
 
-    def boom(marker, timestamp, trial_idx):
-        raise RuntimeError("board died")
-
-    exp.subscribe_marker(boom, raise_on_error=True)
-    exp.subscribe_marker(
-        lambda marker, timestamp, trial_idx: seen.append(trial_idx), raise_on_error=False
-    )
-
-    with pytest.raises(RuntimeError, match="board died"):
-        exp.push_marker(1, 0)
-    assert seen == [], "subscribers after a raise_on_error failure do not run"
-
-
-# 3 ---------------------------------------------------------------------------
-class FakeOutlet:
-    def __init__(self):
-        self.last = None
-
-    def push_sample(self, marker, timestamp):
-        self.last = marker
-
-
-def _bare_eeg(outlet):
-    eeg = object.__new__(EEG)
-    eeg.muse_StreamOutlet = outlet
-    return eeg
-
-
-def test_muse_wraps_scalar_marker():
-    outlet = FakeOutlet()
-    eeg = _bare_eeg(outlet)
-    eeg._muse_push_sample(2, 123.0)
-    assert outlet.last == [2]                  # scalar -> single-channel vector
-
-
-def test_muse_passes_prewrapped_list_unchanged():
-    outlet = FakeOutlet()
-    eeg = _bare_eeg(outlet)
-    eeg._muse_push_sample([2], 123.0)          # P300 back-compat path
-    assert outlet.last == [2]                  # no double-wrap
+    with pytest.raises(RuntimeError, match="boom"):
+        exp.push_marker(1)
+    assert seen == []                         # dispatch aborted; later subscriber never ran

From 90d4760bdb3d1704dd3abd4b2b43b4021b1161b1 Mon Sep 17 00:00:00 2001
From: Benjamin Pettit <benjamin.pettit@digimulti.com>
Date: Wed, 8 Jul 2026 22:19:13 +1000
Subject: [PATCH 3/4] refactor(experiments): type-hint marker subscribers via a
 MarkerSubscriber alias

Add MarkerSubscriber = Callable[[int, float, Optional[int]], None] and annotate
subscribe_marker(callback) and marker_subscribers with it, so the subscriber
contract is a single named type: autocomplete and static checks at the call site,
and Optional[int] surfaces that trial_idx may be None.
---
 eegnb/experiments/Experiment.py | 15 ++++++++-------
 1 file changed, 8 insertions(+), 7 deletions(-)

diff --git a/eegnb/experiments/Experiment.py b/eegnb/experiments/Experiment.py
index 2baa143b8..f1564e5cd 100644
--- a/eegnb/experiments/Experiment.py
+++ b/eegnb/experiments/Experiment.py
@@ -9,7 +9,7 @@
 """
 
 from abc import abstractmethod, ABC
-from typing import Callable
+from typing import Callable, Optional
 from eegnb.devices.eeg import EEG
 from eegnb.devices.vr import VR
 from psychopy import prefs, visual, event, core
@@ -28,6 +28,11 @@
 logger = logging.getLogger(__name__)
 
 
+# A marker subscriber: called as callback(marker, timestamp, trial_idx) on every
+# push_marker(); trial_idx may be None. Its return value is ignored.
+MarkerSubscriber = Callable[[int, float, Optional[int]], None]
+
+
 class BaseExperiment(ABC):
 
     def __init__(self, exp_name, duration, eeg, save_fn, n_trials: int, iti: float, soa: float, jitter: float,
@@ -81,7 +86,7 @@ def __init__(self, exp_name, duration, eeg, save_fn, n_trials: int, iti: float,
         self.markernames = [1, 2]
 
         # (callback, raise_on_error) pairs invoked on every push_marker().
-        self.marker_subscribers: list = []
+        self.marker_subscribers: list[tuple[MarkerSubscriber, bool]] = []
 
         # Per-subscriber failure count: broken subscribers log once, not per trial.
         self._subscriber_failures: dict = {}
@@ -388,7 +393,7 @@ def run(self, instructions=True):
         # Closing the window
         self.window.close()
 
-    def subscribe_marker(self, callback, raise_on_error=True):
+    def subscribe_marker(self, callback: MarkerSubscriber, raise_on_error: bool = True):
         """Register a marker subscriber: callable(marker, timestamp, trial_idx).
 
         Invoked on every push_marker(). raise_on_error selects how a raised
@@ -400,10 +405,6 @@ def subscribe_marker(self, callback, raise_on_error=True):
         raise_on_error=False — optional telemetry (flip-time, eyetracker,
             photodiode) that must never block core data. Exceptions are logged
             and swallowed, so a bad observer can't take down a live recording.
-            To avoid spamming the log — and adding I/O latency at each marker
-            onset — a given subscriber's traceback is logged only on its first
-            failure; further failures are counted and the per-subscriber total
-            is reported at the end of run().
         """
         self.marker_subscribers.append((callback, raise_on_error))
 

From 4879d62abbf5cc10bc7df5a9ba03ea076f8624eb Mon Sep 17 00:00:00 2001
From: Benjamin Pettit <benjamin.pettit@digimulti.com>
Date: Thu, 9 Jul 2026 19:31:41 +1000
Subject: [PATCH 4/4] Fix CI: gate push_marker test on psychopy, type eoec
 marker-sink attrs

- conftest: skip tests/test_push_marker.py when psychopy is absent (it
  imports BaseExperiment -> devices.vr -> psychopy), matching the existing
  eegnb/experiments gate. Fixes test collection on the py-3.12/3.13
  streaming CI env.
- eoec: annotate self.outlet/self.serial as Optional[...] and bind locals
  before the subscribe_marker closures so mypy can narrow them. Fixes the
  two attr-defined errors in the typecheck job.
---
 conftest.py                    |  1 +
 eegnb/experiments/rest/eoec.py | 10 ++++++----
 2 files changed, 7 insertions(+), 4 deletions(-)

diff --git a/conftest.py b/conftest.py
index 649b2f0ef..5b71a9893 100644
--- a/conftest.py
+++ b/conftest.py
@@ -14,6 +14,7 @@ def _is_available(module_name: str) -> bool:
     collect_ignore += [
         "eegnb/experiments",
         "eegnb/devices/vr.py",
+        "tests/test_push_marker.py",  # imports BaseExperiment, which pulls in psychopy via devices.vr
     ]
 elif not _is_available("psychxr"):
     collect_ignore += ["eegnb/devices/vr.py"]
diff --git a/eegnb/experiments/rest/eoec.py b/eegnb/experiments/rest/eoec.py
index b7140f063..ea24ff7cc 100644
--- a/eegnb/experiments/rest/eoec.py
+++ b/eegnb/experiments/rest/eoec.py
@@ -48,8 +48,8 @@ def __init__(
         self.use_verbal_cues = use_verbal_cues
         self.open_audio = open_audio
         self.close_audio = close_audio
-        self.serial = None
-        self.outlet = None
+        self.serial: Optional["serial.Serial"] = None
+        self.outlet: Optional[StreamOutlet] = None
         self.open_sound = None
         self.close_sound = None
         super().__init__(
@@ -86,16 +86,18 @@ def setup(self, instructions: bool = True):
         # LSL outlet for markers
         info = StreamInfo("Markers", "Markers", 1, 0, "int32", "eyeclosure-baseline")
         self.outlet = StreamOutlet(info)
+        outlet = self.outlet  # local binding: mypy can't narrow self.outlet inside the closure
         self.subscribe_marker(
-            lambda marker, timestamp, _idx: self.outlet.push_sample([marker], timestamp)
+            lambda marker, timestamp, _idx: outlet.push_sample([marker], timestamp)
         )
 
         # serial connection for hardware triggers
         if self.serial_port and serial is not None:
             try:
                 self.serial = serial.Serial(self.serial_port, 115200, timeout=1)
+                ser = self.serial  # local binding: mypy can't narrow self.serial inside the closure
                 self.subscribe_marker(
-                    lambda marker, timestamp, _idx: self.serial.write(bytes([marker])),
+                    lambda marker, timestamp, _idx: ser.write(bytes([marker])),
                     raise_on_error=False,
                 )
             except Exception:  # pragma: no cover