ssvep.runtime — sessions and acquisition

Part of the API reference, generated from the docstrings in src/ssvep.

Run plan

ssvep.runtime.run_plan

Run plan: expand a run manifest into a deterministic, ordered list of trials + markers.

The runtime runs ONLY built, saved, versioned protocols. Given a manifest, this derives the condition × block × trial structure, assigns a cued target (frequency) per trial, and produces the marker sequence — all reproducibly from a seed (recorded in provenance). The runner attaches real LSL timestamps when each marker fires.

Condition order is the manifest's, never drawn here (#31). Blocks play in the order design.trial_order declares (generated at design time by :mod:ssvep.stim.trial_order and fingerprinted with the manifest); a manifest that declares none plays its conditions in design.conditions order, blocked, as every run always has. The seed only shuffles which target each trial inside a block cues.

RunPlan dataclass

Source code in src/ssvep/runtime/run_plan.py

@dataclass
class RunPlan:
    trials: list[Trial]
    timing: dict               # {cue_s, stim_s, iti_s, onset_offset_s}
    marker_template: str
    seed: int | None
    resting: dict | None = None   # design.resting when this is a resting run; None for SSVEP
    order_scheme: str = trial_order_mod.SCHEME_FIXED   # design.order.scheme as declared (#31)
    order_seed: int | None = None                      # design.order.seed (blocked_randomized only)
    feedback: dict | None = None   # design.feedback ({mode, wait_s, show_s}) when the run shows any
    cue_mode: str = manifest_io.CUE_CUED   # design.cue_mode: 'free' = the participant chooses (#268)
    #: design.early_stop as :func:`ssvep.io.manifest.early_stop_config` reads it (#23), or None: a
    #: run whose trials may end before ``stim_s``. The plan's trials and markers are unchanged by it;
    #: only how long the run can take is (:meth:`estimated_duration_range_s`).
    early_stop: dict | None = None
    #: design.fpvs when this is an FPVS run (#494); None otherwise. Its "trials" are its sequences.
    fpvs: dict | None = None

    @classmethod
    def from_manifest(cls, manifest: dict, seed: int | None = 0) -> "RunPlan":
        design = manifest["design"]
        timing = design["timing"]
        template = manifest["markers"].get(
            "schema", "cond/{condition}/freq/{target_freq_hz}/block/{block}/trial/{trial}/{segment}")
        # Resting is a block paradigm with no targets and no cued frequencies: its "trials" are the
        # eyes-open/eyes-closed blocks, and the real schedule is expanded frame-wise by the renderer
        # from design.resting (ssvep.stim.resting.build_resting_schedule). Planning it as SSVEP would
        # divide by an empty frequency set.
        if manifest_io.is_resting(manifest):
            return cls(trials=_resting_trials(design["resting"]), timing=timing,
                       marker_template=template, seed=seed, resting=dict(design["resting"]))
        # FPVS (#494): one "trial" per sequence, at the base rate. The frame schedule is expanded by
        # its renderer from design.fpvs (ssvep.stim.fpvs.build_fpvs_schedule); nothing is drawn here.
        if manifest_io.is_fpvs(manifest):
            f = design["fpvs"]
            base = float(f["base_hz"])
            n = int(f["n_sequences"])
            return cls(trials=[Trial("fpvs", i, 1, "stream", base) for i in range(1, n + 1)],
                       timing=timing, marker_template=template, seed=seed, fpvs=dict(f))

        freq_to_id = {round(float(s["flicker"]["freq_hz"]), 6): s["id"] for s in manifest["stimuli"]}
        rng = random.Random(seed) if seed is not None else None
        # The block sequence the manifest declares, verified (TrialOrderError if it cannot be played
        # as declared). With no declaration this is design.conditions order, blocked — the same
        # sequence, and so the same draws from ``rng``, as before trial_order was honoured.
        declared = trial_order_mod.declared_order(design)
        conds = {str(c["id"]): c for c in design["conditions"]}

        free = not manifest_io.is_cued(manifest)
        trials: list[Trial] = []
        for cid, b in declared.blocks:
            cond = conds[cid]
            freqs = [float(f) for f in cond["stimulus_freqs_hz"]]
            per_block = design.get("trials_per_block", len(freqs))
            if free:
                # Free choice (#268): the block still has its trials, but nothing is cued - no
                # target, no label, and no draw from the seed (there is nothing to shuffle).
                trials.extend(Trial(cond["id"], b, i, None, None)
                              for i in range(1, int(per_block) + 1))
                continue
            order = _block_order(freqs, per_block, rng)
            for i, f in enumerate(order, start=1):
                tid = freq_to_id.get(round(f, 6), _nearest_id(freq_to_id, f))
                trials.append(Trial(cond["id"], b, i, tid, f))
        return cls(trials=trials, timing=timing, marker_template=template, seed=seed,
                   order_scheme=declared.scheme, order_seed=declared.seed,
                   feedback=manifest_io.feedback_config(manifest),
                   cue_mode=manifest_io.cue_mode(manifest),
                   early_stop=manifest_io.early_stop_config(manifest))

    # -- convenience --------------------------------------------------------
    @property
    def is_resting(self) -> bool:
        return self.resting is not None

    @property
    def is_free(self) -> bool:
        """True for a free-choice run: no trial has a cued target or a label (#268)."""
        return self.cue_mode == manifest_io.CUE_FREE

    @property
    def n_trials(self) -> int:
        return len(self.trials)

    def played_order(self, n_trials: int | None = None) -> dict:
        """The order this plan plays, for the sidecar (#31): the scheme and one row per trial.

        Each row is ``position`` (1-based in the run), ``condition``, ``block`` (the condition's own
        1-based block number), ``block_position`` (1-based position of that block in the run),
        ``trial`` (1-based within the block), ``target_id`` and ``target_freq_hz``. ``n_trials``
        keeps only the first that many — the trials a run stopped early actually played.
        """
        rows, seen = [], {}
        for pos, t in enumerate(self.trials, start=1):
            key = (t.condition, t.block)
            if key not in seen:
                seen[key] = len(seen) + 1
            rows.append({"position": pos, "condition": t.condition, "block": t.block,
                         "block_position": seen[key], "trial": t.trial,
                         "target_id": t.target_id, "target_freq_hz": t.target_freq_hz})
        if n_trials is not None:
            rows = rows[:max(0, int(n_trials))]
        out = {"scheme": self.order_scheme, "trials": rows}
        if self.order_seed is not None:
            out["seed"] = self.order_seed
        return out

    def estimated_duration_s(self) -> float:
        if self.is_resting:      # contiguous blocks, no cue/ITI, plus the quiet lead-in
            return (float(self.resting.get("lead_in_s", 0.0))
                    + float(self.resting["n_blocks"]) * float(self.resting["block_s"]))
        if self.fpvs is not None:
            from ..stim import fpvs as fpvs_mod
            return fpvs_mod.total_s(self.fpvs)
        per = (self.timing["cue_s"]
               + presented_stim_s(self.timing, self.early_stop)   # incl. the lead-in presented
               + self.timing["iti_s"])
        if self.feedback:         # the two fixed feedback slots (#268)
            per += self.feedback["wait_s"] + self.feedback["show_s"]
        return per * self.n_trials

    def estimated_duration_range_s(self) -> tuple[float, float]:
        """``(shortest, longest)`` the run can take (s). Equal for a fixed-length run.

        With early stopping (#23) the longest is :meth:`estimated_duration_s` (every trial runs to
        ``stim_s``; what a progress bar plans for, so it finishes early rather than late) and the
        shortest has every trial stop at its first look, ``min_window_s``. The shortest is a floor
        the run cannot reach: the flicker also runs on for the stop latency after each decision.
        """
        longest = self.estimated_duration_s()
        if not self.early_stop or self.is_resting:
            return longest, longest
        saved = max(0.0, float(self.timing["stim_s"]) - float(self.early_stop["min_window_s"]))
        return longest - saved * self.n_trials, longest

    def markers_for_trial(self, trial: Trial) -> list[Marker]:
        """cue → stim_on → stim_off markers for one trial (block_start/end added by the runner)."""
        common = dict(condition=trial.condition, trial=trial.trial, block=trial.block,
                      target_id=trial.target_id, target_freq_hz=trial.target_freq_hz,
                      cued=not self.is_free)
        return [
            Marker(segment="cue", **common),
            Marker(segment="stim_on",
                   trial_dur_s=presented_stim_s(self.timing, self.early_stop),
                   **common),
            Marker(segment="stim_off", **common),
        ]

is_free property

is_free

True for a free-choice run: no trial has a cued target or a label (#268).

played_order

played_order(n_trials=None)

The order this plan plays, for the sidecar (#31): the scheme and one row per trial.

Each row is position (1-based in the run), condition, block (the condition's own 1-based block number), block_position (1-based position of that block in the run), trial (1-based within the block), target_id and target_freq_hz. n_trials keeps only the first that many — the trials a run stopped early actually played.

Source code in src/ssvep/runtime/run_plan.py

def played_order(self, n_trials: int | None = None) -> dict:
    """The order this plan plays, for the sidecar (#31): the scheme and one row per trial.

    Each row is ``position`` (1-based in the run), ``condition``, ``block`` (the condition's own
    1-based block number), ``block_position`` (1-based position of that block in the run),
    ``trial`` (1-based within the block), ``target_id`` and ``target_freq_hz``. ``n_trials``
    keeps only the first that many — the trials a run stopped early actually played.
    """
    rows, seen = [], {}
    for pos, t in enumerate(self.trials, start=1):
        key = (t.condition, t.block)
        if key not in seen:
            seen[key] = len(seen) + 1
        rows.append({"position": pos, "condition": t.condition, "block": t.block,
                     "block_position": seen[key], "trial": t.trial,
                     "target_id": t.target_id, "target_freq_hz": t.target_freq_hz})
    if n_trials is not None:
        rows = rows[:max(0, int(n_trials))]
    out = {"scheme": self.order_scheme, "trials": rows}
    if self.order_seed is not None:
        out["seed"] = self.order_seed
    return out

estimated_duration_range_s

estimated_duration_range_s()

(shortest, longest) the run can take (s). Equal for a fixed-length run.

With early stopping (#23) the longest is :meth:estimated_duration_s (every trial runs to stim_s; what a progress bar plans for, so it finishes early rather than late) and the shortest has every trial stop at its first look, min_window_s. The shortest is a floor the run cannot reach: the flicker also runs on for the stop latency after each decision.

Source code in src/ssvep/runtime/run_plan.py

def estimated_duration_range_s(self) -> tuple[float, float]:
    """``(shortest, longest)`` the run can take (s). Equal for a fixed-length run.

    With early stopping (#23) the longest is :meth:`estimated_duration_s` (every trial runs to
    ``stim_s``; what a progress bar plans for, so it finishes early rather than late) and the
    shortest has every trial stop at its first look, ``min_window_s``. The shortest is a floor
    the run cannot reach: the flicker also runs on for the stop latency after each decision.
    """
    longest = self.estimated_duration_s()
    if not self.early_stop or self.is_resting:
        return longest, longest
    saved = max(0.0, float(self.timing["stim_s"]) - float(self.early_stop["min_window_s"]))
    return longest - saved * self.n_trials, longest

markers_for_trial

markers_for_trial(trial)

cue → stim_on → stim_off markers for one trial (block_start/end added by the runner).

Source code in src/ssvep/runtime/run_plan.py

def markers_for_trial(self, trial: Trial) -> list[Marker]:
    """cue → stim_on → stim_off markers for one trial (block_start/end added by the runner)."""
    common = dict(condition=trial.condition, trial=trial.trial, block=trial.block,
                  target_id=trial.target_id, target_freq_hz=trial.target_freq_hz,
                  cued=not self.is_free)
    return [
        Marker(segment="cue", **common),
        Marker(segment="stim_on",
               trial_dur_s=presented_stim_s(self.timing, self.early_stop),
               **common),
        Marker(segment="stim_off", **common),
    ]

early_stop_lead_in_s

early_stop_lead_in_s(timing)

The visual-latency lead-in (s) a run that stops early trims before its first look (#23): design.timing.onset_offset_s, or :data:ssvep.DEFAULT_ONSET_OFFSET_S when the design does not state one - as the offline batch reads it, so the live looks and the replay start their growing windows on the same sample.

Source code in src/ssvep/runtime/run_plan.py

def early_stop_lead_in_s(timing: dict) -> float:
    """The visual-latency lead-in (s) a run that stops early trims before its first look (#23):
    ``design.timing.onset_offset_s``, or :data:`ssvep.DEFAULT_ONSET_OFFSET_S` when the design does
    not state one - as the offline batch reads it, so the live looks and the replay start their
    growing windows on the same sample."""
    from .. import DEFAULT_ONSET_OFFSET_S
    v = (timing or {}).get("onset_offset_s")
    return float(DEFAULT_ONSET_OFFSET_S if v is None else v)

presented_stim_s

presented_stim_s(timing, early_stop=None)

How long one trial flickers when it runs to its end (s), before rounding to frames.

A fixed-length run presents stim_s plus the onset_offset_s the design itself states (0 for one that predates the field), exactly as it always has. A run that may stop early (#23) presents :func:ssvep.classification.decoding.early_stop_span_s: the lead-in its decoder trims, the maximum window and a small guard, so a trial that runs to its maximum always holds the samples its last look needs (:func:ssvep.runtime.presentation.stim_frames then rounds UP).

Source code in src/ssvep/runtime/run_plan.py

def presented_stim_s(timing: dict, early_stop: dict | None = None) -> float:
    """How long one trial flickers when it runs to its end (s), before rounding to frames.

    A fixed-length run presents ``stim_s`` plus the ``onset_offset_s`` the design itself states (0
    for one that predates the field), exactly as it always has. A run that may stop early (#23)
    presents :func:`ssvep.classification.decoding.early_stop_span_s`: the lead-in its decoder trims,
    the maximum window and a small guard, so a trial that runs to its maximum always holds the
    samples its last look needs (:func:`ssvep.runtime.presentation.stim_frames` then rounds UP).
    """
    if early_stop:
        from ..classification.decoding import early_stop_span_s
        return early_stop_span_s(timing["stim_s"], early_stop_lead_in_s(timing))
    return float(timing["stim_s"]) + float(timing.get("onset_offset_s", 0.0) or 0.0)

plan_summary

plan_summary(plan, manifest, *, seed, n_trials=None)

The sidecar's plan block for plan: size, seed, duration, conditions, and the play order.

trial_order (#31) is the order the runtime played, trial by trial (:meth:RunPlan.played_order), with the scheme the manifest declared — so a recording says which condition each trial was and where in the run it fell, without re-deriving it. A run stopped early played the first provenance.trials_recorded of these. A resting run has no conditions to order and gets no trial_order. Shared with the synthetic generator so the two sidecars cannot drift (#348).

Source code in src/ssvep/runtime/run_plan.py

def plan_summary(plan: RunPlan, manifest: dict, *, seed, n_trials: int | None = None) -> dict:
    """The sidecar's ``plan`` block for ``plan``: size, seed, duration, conditions, and the play order.

    ``trial_order`` (#31) is the order the runtime played, trial by trial
    (:meth:`RunPlan.played_order`), with the scheme the manifest declared — so a recording says
    which condition each trial was and where in the run it fell, without re-deriving it. A run
    stopped early played the first ``provenance.trials_recorded`` of these. A resting run has no
    conditions to order and gets no ``trial_order``. Shared with the synthetic generator so the two
    sidecars cannot drift (#348).
    """
    out = {
        "n_trials": plan.n_trials if n_trials is None else n_trials, "seed": seed,
        "estimated_duration_s": plan.estimated_duration_s(),
        "conditions": [c["id"] for c in (manifest.get("design") or {}).get("conditions", [])],
    }
    if plan.early_stop and not plan.is_resting:
        # Only for a run that may stop trials early (#23), so every other sidecar is unchanged.
        out["estimated_min_duration_s"] = plan.estimated_duration_range_s()[0]
    if not plan.is_resting:
        out["trial_order"] = plan.played_order()
    return out

Session protocol and resolver

ssvep.runtime.session

The session level: a protocol (template) → a concrete, ordered, frozen set of runs (#43).

Terminology (AGENTS.md §2.1): a run is one continuous recording (one run manifest, planned by :mod:ssvep.runtime.run_plan); a session is an ordered set of runs; a protocol is the instructions for running a session on a participant. This module is the protocol layer that #42 unblocked — it reuses the module name freed when the old session.py (a run planner) became run_plan.py.

A :class:SessionProtocol is a template: authored once, instantiated per participant. It contains its run manifests and groups them; each group declares how its runs are ordered for a given participant. :func:resolve_session turns (protocol, sub-XXX, ses-YYY) into an ordered list of :class:ResolvedRun — each carrying the frozen run manifest, a derived run_index, and the position block stamped into its recording's sidecar. The order is derived and recorded, never inferred from files on disk (the manifest-spine rule, one level up). A protocol never redefines a run's design.

Hardware is the session's, not a run's (#484). A protocol carries one acquisition block (schema 1.7, :attr:SessionProtocol.acquisition) and :func:resolve_session stamps it into every resolved run, replacing whatever acquisition block an embedded run manifest still carries — so everything downstream that reads run.manifest["acquisition"] (impedance, preview, decoder, recorder, sidecar) reads the session's hardware, whichever run it is looking at. A protocol from before 1.7 has no block of its own: its runs' blocks are lifted when they all agree, and when they disagree the session has no hardware and cannot be locked (:attr:ResolvedSession.acquisition_problem). A protocol with no hardware anywhere takes the hardware Set up Session states for it (resolve_session(..., session_hardware=...)): the operator chooses it once, for the whole session, which is the same decision the protocol would otherwise have carried. To run the same design on different hardware, record another session.

Protocol schema v1.2 (#115) embeds each run manifest in the protocol; before it, a protocol held a path + pinned fingerprint and the resolver loaded the file, hard-stopping on drift. Embedding makes a protocol one self-contained artifact — copy it to an acquisition PC and the whole design goes with it — and means a saved protocol always runs, where before it could be silently broken by anyone editing a run manifest it pointed at. What was lost with the resolve-time drift stop is regained by :meth:SessionProtocol.source_report, which asks the same question at design time and can be acted on. Pre-1.2 protocols still resolve exactly as they did, drift stop included.

The core (protocol model + ordering + resolver) is pure and unit-tested: it scans no sourcedata, and for an embedded protocol it touches no filesystem at all. The helpers :func:next_session / :func:session_exists are separate, so the resolver stays pure.

ProtocolDriftError

Bases: RuntimeError

A referenced run manifest's on-disk fingerprint no longer matches the protocol's pin.

Raised by :func:resolve_session (unless allow_drift): the design a participant would run has silently diverged from the design the protocol was authored against. A design change must be a deliberate new protocol version, never a side effect of editing a referenced run manifest — so this is a hard stop, not a warning (AGENTS.md §2: declared, never inferred).

Source code in src/ssvep/runtime/session.py

class ProtocolDriftError(RuntimeError):
    """A referenced run manifest's on-disk fingerprint no longer matches the protocol's pin.

    Raised by :func:`resolve_session` (unless ``allow_drift``): the design a participant would run has
    silently diverged from the design the protocol was authored against. A design change must be a
    deliberate new protocol version, never a side effect of editing a referenced run manifest — so
    this is a hard stop, not a warning (AGENTS.md §2: declared, never inferred).
    """

RunRef dataclass

One run of a group: the run manifest itself, plus where it was copied from.

Since protocol schema v1.2 (#115) the manifest is embedded and authoritative — the resolver runs :attr:manifest and never reads the store. That makes a protocol a single self-contained file (copy it to an acquisition PC and the whole design travels with it) and makes a saved protocol un-breakable by someone editing a run manifest it happens to point at.

:attr:run_manifest / :attr:fingerprint survive as the source pointer: where this copy came from and what it fingerprinted at embed time. Nothing at record time reads them. They exist so :meth:source_status can answer "has the file I copied this from moved on?" at design time, which is where a design question belongs — the previous scheme could only answer it at resolve time, i.e. by refusing to start a session with the participant already in the chair.

A pre-1.2 protocol has no :attr:manifest. Those still load, and :func:resolve_session still resolves them the old way — store lookup plus a fingerprint check that hard-stops on drift.

Source code in src/ssvep/runtime/session.py

@dataclass
class RunRef:
    """One run of a group: the run manifest itself, plus where it was copied from.

    Since protocol schema v1.2 (#115) the manifest is **embedded and authoritative** — the resolver
    runs :attr:`manifest` and never reads the store. That makes a protocol a single self-contained
    file (copy it to an acquisition PC and the whole design travels with it) and makes a saved
    protocol un-breakable by someone editing a run manifest it happens to point at.

    :attr:`run_manifest` / :attr:`fingerprint` survive as the **source pointer**: where this copy came
    from and what it fingerprinted at embed time. Nothing at record time reads them. They exist so
    :meth:`source_status` can answer "has the file I copied this from moved on?" at *design* time,
    which is where a design question belongs — the previous scheme could only answer it at resolve
    time, i.e. by refusing to start a session with the participant already in the chair.

    A pre-1.2 protocol has no :attr:`manifest`. Those still load, and :func:`resolve_session` still
    resolves them the old way — store lookup plus a fingerprint check that hard-stops on drift.
    """
    run_manifest: str = ""             # source path, relative to the run-manifest store
    fingerprint: str = ""              # the manifest's fingerprint when it was embedded/pinned
    label: str | None = None           # optional per-position label; falls back to run.name
    manifest: dict | None = None       # the embedded run manifest (v1.2+); None ⇒ legacy reference

    @property
    def embedded(self) -> bool:
        return self.manifest is not None

    def to_dict(self) -> dict:
        if not self.embedded:                              # legacy reference — round-trip unchanged
            d = {"run_manifest": self.run_manifest, "fingerprint": self.fingerprint}
            if self.label:
                d["label"] = self.label
            return d
        d: dict = {"manifest": self.manifest}
        if self.run_manifest:
            d["source"] = {"run_manifest": self.run_manifest, "fingerprint": self.fingerprint}
        if self.label:
            d["label"] = self.label
        return d

    @classmethod
    def from_dict(cls, d: dict) -> "RunRef":
        src = d.get("source") or {}
        return cls(
            # Read the source block first (v1.2), then the legacy top-level keys — a v1.2 run keeps
            # its provenance in 'source', a v1.0/v1.1 run has it at the top level and no manifest.
            run_manifest=src.get("run_manifest") or d.get("run_manifest") or "",
            fingerprint=src.get("fingerprint") or d.get("fingerprint") or "",
            label=d.get("label"),
            manifest=(manifest_io.canonicalize_manifest(d["manifest"])
                      if d.get("manifest") is not None else None))

    @classmethod
    def pin(cls, run_manifest: str, store_dir, label: str | None = None) -> "RunRef":
        """Embed the run manifest at ``store_dir / run_manifest``, recording where it came from.

        The manifest is loaded through :func:`ssvep.io.manifest.load_manifest`, so it is canonicalized
        to v1.3 shape and schema-validated here — an unloadable or invalid run manifest fails while
        someone is authoring, not while someone is recording.
        """
        m = manifest_io.load_manifest(locate_in_store(store_dir, run_manifest))
        return cls(run_manifest=run_manifest, fingerprint=manifest_io.manifest_fingerprint(m),
                   label=label, manifest=m)

    def resolve_manifest(self, store_dir, *, allow_drift: bool = False,
                         protocol_name: str = "") -> tuple[dict, str]:
        """This run's manifest and its fingerprint — embedded copy, or the legacy store lookup.

        Embedded (v1.2+): returned as-is. There is nothing to verify, because there is nothing else
        it could have come from; the protocol *is* the design.

        Legacy (v1.0/v1.1): loaded from ``store_dir`` and re-fingerprinted, raising
        :class:`ProtocolDriftError` on a mismatch unless ``allow_drift``.
        """
        if self.embedded:
            return self.manifest, (self.fingerprint or manifest_io.manifest_fingerprint(self.manifest))
        if store_dir is None:
            raise ValueError(
                f"run {self.run_manifest!r} is a pre-1.2 reference, so resolving it needs the "
                f"run-manifest store it points into — pass store_dir.")
        manifest = manifest_io.load_manifest(locate_in_store(store_dir, self.run_manifest))
        actual = manifest_io.manifest_fingerprint(manifest)
        if actual != self.fingerprint and not allow_drift:
            raise ProtocolDriftError(
                f"run manifest {self.run_manifest!r} has drifted from protocol "
                f"{protocol_name!r}: pinned {self.fingerprint}, on disk {actual}. "
                f"Re-pin deliberately (a design change is a new protocol version), or pass "
                f"allow_drift=True to override.")
        return manifest, actual

    def source_status(self, store_dir) -> str:
        """Whether the file this run was embedded from still matches it — a *design-time* question.

        One of :data:`SOURCE_CURRENT` / :data:`SOURCE_DRIFTED` / :data:`SOURCE_MISSING` /
        :data:`SOURCE_UNKNOWN`. Never raises: an unreadable or invalid source file is ``missing``,
        because the answer this feeds is a report, and a report that throws is a report nobody sees.
        """
        if not self.run_manifest or store_dir is None:
            return SOURCE_UNKNOWN
        try:
            current = manifest_io.manifest_fingerprint(
                manifest_io.load_manifest(locate_in_store(store_dir, self.run_manifest)))
        except Exception:
            return SOURCE_MISSING
        mine = self.fingerprint or (manifest_io.manifest_fingerprint(self.manifest)
                                    if self.embedded else "")
        return SOURCE_CURRENT if current == mine else SOURCE_DRIFTED

pin classmethod

pin(run_manifest, store_dir, label=None)

Embed the run manifest at store_dir / run_manifest, recording where it came from.

The manifest is loaded through :func:ssvep.io.manifest.load_manifest, so it is canonicalized to v1.3 shape and schema-validated here — an unloadable or invalid run manifest fails while someone is authoring, not while someone is recording.

Source code in src/ssvep/runtime/session.py

@classmethod
def pin(cls, run_manifest: str, store_dir, label: str | None = None) -> "RunRef":
    """Embed the run manifest at ``store_dir / run_manifest``, recording where it came from.

    The manifest is loaded through :func:`ssvep.io.manifest.load_manifest`, so it is canonicalized
    to v1.3 shape and schema-validated here — an unloadable or invalid run manifest fails while
    someone is authoring, not while someone is recording.
    """
    m = manifest_io.load_manifest(locate_in_store(store_dir, run_manifest))
    return cls(run_manifest=run_manifest, fingerprint=manifest_io.manifest_fingerprint(m),
               label=label, manifest=m)

resolve_manifest

resolve_manifest(store_dir, *, allow_drift=False, protocol_name='')

This run's manifest and its fingerprint — embedded copy, or the legacy store lookup.

Embedded (v1.2+): returned as-is. There is nothing to verify, because there is nothing else it could have come from; the protocol is the design.

Legacy (v1.0/v1.1): loaded from store_dir and re-fingerprinted, raising :class:ProtocolDriftError on a mismatch unless allow_drift.

Source code in src/ssvep/runtime/session.py

def resolve_manifest(self, store_dir, *, allow_drift: bool = False,
                     protocol_name: str = "") -> tuple[dict, str]:
    """This run's manifest and its fingerprint — embedded copy, or the legacy store lookup.

    Embedded (v1.2+): returned as-is. There is nothing to verify, because there is nothing else
    it could have come from; the protocol *is* the design.

    Legacy (v1.0/v1.1): loaded from ``store_dir`` and re-fingerprinted, raising
    :class:`ProtocolDriftError` on a mismatch unless ``allow_drift``.
    """
    if self.embedded:
        return self.manifest, (self.fingerprint or manifest_io.manifest_fingerprint(self.manifest))
    if store_dir is None:
        raise ValueError(
            f"run {self.run_manifest!r} is a pre-1.2 reference, so resolving it needs the "
            f"run-manifest store it points into — pass store_dir.")
    manifest = manifest_io.load_manifest(locate_in_store(store_dir, self.run_manifest))
    actual = manifest_io.manifest_fingerprint(manifest)
    if actual != self.fingerprint and not allow_drift:
        raise ProtocolDriftError(
            f"run manifest {self.run_manifest!r} has drifted from protocol "
            f"{protocol_name!r}: pinned {self.fingerprint}, on disk {actual}. "
            f"Re-pin deliberately (a design change is a new protocol version), or pass "
            f"allow_drift=True to override.")
    return manifest, actual

source_status

source_status(store_dir)

Whether the file this run was embedded from still matches it — a design-time question.

One of :data:SOURCE_CURRENT / :data:SOURCE_DRIFTED / :data:SOURCE_MISSING / :data:SOURCE_UNKNOWN. Never raises: an unreadable or invalid source file is missing, because the answer this feeds is a report, and a report that throws is a report nobody sees.

Source code in src/ssvep/runtime/session.py

def source_status(self, store_dir) -> str:
    """Whether the file this run was embedded from still matches it — a *design-time* question.

    One of :data:`SOURCE_CURRENT` / :data:`SOURCE_DRIFTED` / :data:`SOURCE_MISSING` /
    :data:`SOURCE_UNKNOWN`. Never raises: an unreadable or invalid source file is ``missing``,
    because the answer this feeds is a report, and a report that throws is a report nobody sees.
    """
    if not self.run_manifest or store_dir is None:
        return SOURCE_UNKNOWN
    try:
        current = manifest_io.manifest_fingerprint(
            manifest_io.load_manifest(locate_in_store(store_dir, self.run_manifest)))
    except Exception:
        return SOURCE_MISSING
    mine = self.fingerprint or (manifest_io.manifest_fingerprint(self.manifest)
                                if self.embedded else "")
    return SOURCE_CURRENT if current == mine else SOURCE_DRIFTED

SessionProtocol dataclass

Source code in src/ssvep/runtime/session.py

@dataclass
class SessionProtocol:
    name: str
    groups: list[Group] = field(default_factory=list)
    version: str | None = None         # author's design version (bump deliberately on a design change)
    description: str | None = None
    seed: int = 0                      # base seed for randomized groups
    protocol_version: str = protocol_io.PROTOCOL_VERSION   # schema version
    end_questionnaire: Q.Questionnaire | None = None       # administered once, at the very end (#20)
    # Pre-declared between-condition comparisons (#269, stage 1). Raw, schema-validated dicts — the
    # analysis layer (not this module) parses them into whatever it needs to run a contrast, so this
    # class stays symmetric with how a run manifest's own schema-shaped blocks are carried here.
    contrasts: list[dict] = field(default_factory=list)
    # The participant-facing blocks (#286, schema 1.5): the data-use wording, the safety screen's
    # extra statements, and the session-start questionnaire. Raw, schema-validated dicts (the
    # questionnaire parsed, like the others), carried verbatim so the fingerprint covers every word.
    # The policy that reads ``data_use`` is ssvep.runtime.data_use; nothing here interprets it.
    data_use: dict | None = None
    safety_screen: dict | None = None
    start_questionnaire: Q.Questionnaire | None = None
    # The study this protocol belongs to (#396, schema 1.6): a declared study's id, which decides the
    # data root its sessions are recorded into (ssvep.runtime.study). None = the install's implicit
    # single study. Stamped into every recording's session_position, and covered by the fingerprint.
    study: str | None = None
    # The session's acquisition hardware (#484, schema 1.7): one block, the same shape as a run
    # manifest's, which every run of the session is recorded on. None = a protocol from before 1.7,
    # whose runs' own blocks are lifted when they agree (:meth:`session_acquisition`).
    acquisition: dict | None = None

    def _questionnaires(self) -> list[Q.Questionnaire]:
        qs = [self.start_questionnaire, self.end_questionnaire] + [g.questionnaire for g in self.groups]
        return [q for q in qs if q is not None]

    def has_participant_features(self) -> bool:
        """True once any 1.5 feature is in use — i.e. this file needs a 1.5 reader (#286): a
        ``data_use``, ``safety_screen`` or ``start_questionnaire`` block. A closed-choice item can
        only be in the start questionnaire (validation refuses one anywhere else), so it needs no
        separate check."""
        return (self.data_use is not None or self.safety_screen is not None
                or self.start_questionnaire is not None)

    def has_questionnaires(self) -> bool:
        return self.end_questionnaire is not None or any(g.questionnaire for g in self.groups)

    def has_embedded_runs(self) -> bool:
        """True once any run carries its manifest inline — i.e. this file needs a 1.2 reader (#115)."""
        return any(r.embedded for g in self.groups for r in g.runs)

    def has_per_session_fpvs(self) -> bool:
        """True once any embedded run is an FPVS design whose item order is drawn per session
        (``design.fpvs.realise: "per_session"``, #530) — i.e. this file needs a 1.8 reader. A legacy
        path reference is not inspected: its manifest lives in the store, not in this file."""
        for g in self.groups:
            for r in g.runs:
                m = r.manifest
                if (isinstance(m, dict) and manifest_io.is_fpvs(m)
                        and (manifest_io.fpvs_params(m) or {}).get("realise") == "per_session"):
                    return True
        return False

    def source_report(self, store_dir) -> list[tuple[Group, RunRef, str]]:
        """Every run paired with whether its source file still matches it (:meth:`RunRef.source_status`).

        The design-time replacement for the old resolve-time drift stop: authoring can ask "what have
        I fallen behind on?" and act on it, instead of finding out when a session refuses to start.
        """
        return [(g, r, r.source_status(store_dir)) for g in self.groups for r in g.runs]

    def duplicate_run_labels(self, store_dir=None) -> list[dict]:
        """Runs in this protocol that share a ``run.name`` but are different designs (#467).

        The **name** identifies a run; a description need not be unique (two variants of one design
        legitimately share it), so a shared description is no longer reported. Two runs with one name
        and *different* designs is the copy-paste that was never updated for the run it landed on —
        what happened in the 2026-08-28 pilot, where runs at three frequency bands were all cloned
        under one name. The same design twice under one name (one run repeated on purpose) is not a
        collision. Prose differences (``run.description``) do not make a design different
        (:func:`ssvep.io.manifest.design_fingerprint`). The same check runs earlier, at Save in
        Build run (:func:`ssvep.io.manifest.name_clashes`); this one catches a pair that was
        never saved side by side.

        Returns one entry per colliding name — ``{"field": "name", "value": ..., "runs":
        [(group_label, run_label), ...]}`` — worst (most runs sharing it) first. A run whose manifest
        can't be resolved (a legacy reference with no store, or a source file that has moved) is
        skipped rather than raising.
        """
        by_name: dict[str, list[tuple[str, str, str]]] = {}
        for g in self.groups:
            for ref in g.runs:
                try:
                    manifest, _ = ref.resolve_manifest(store_dir)
                except Exception:
                    continue
                run = manifest_io.run_meta(manifest)
                label = ref.label or run.get("name") or ref.run_manifest or "(unnamed)"
                if run.get("name"):
                    by_name.setdefault(run["name"], []).append(
                        (g.label, label, manifest_io.design_fingerprint(manifest)))
        out = [{"field": "name", "value": name, "runs": [(g, lab) for g, lab, _ in runs]}
               for name, runs in by_name.items()
               if len(runs) > 1 and len({fp for _, _, fp in runs}) > 1]
        out.sort(key=lambda d: -len(d["runs"]))
        return out

    def run_labels(self, store_dir=None) -> list[str]:
        """Every run's label, in protocol order — computed exactly as :func:`resolve_session` does.

        Used to check ``contrasts[*].levels`` name real runs (:meth:`contrast_problems`). A run whose
        manifest can't be resolved (a legacy reference with no store, or a source file that has moved)
        is skipped rather than raising, the same convention as :meth:`duplicate_run_labels`.
        """
        labels = []
        for g in self.groups:
            for ref in g.runs:
                try:
                    manifest, _ = ref.resolve_manifest(store_dir)
                except Exception:
                    continue
                labels.append(ref.label or manifest_io.run_meta(manifest).get("name", ref.run_manifest))
        return labels

    #: participants.tsv columns a contrast's ``between.column`` may never name (#269 stage 2).
    #: ``participant_id`` and ``species`` are identity/administrative, not a manipulation; ``group``
    #: is the exclusion declaration pilot/test are dropped through (D13) — using it as a between
    #: factor would conflate "excluded from every contrast" with "a level of this one".
    _RESERVED_BETWEEN_COLUMNS = frozenset({"participant_id", "species", "group"})

    def contrast_problems(self, store_dir=None) -> list[str]:
        """Human-readable problems with this protocol's ``contrasts`` block. Empty ⇒ fine.

        Checked here, at design time, rather than only when an analysis tries to run a contrast: a
        typo'd run label or a copy-pasted duplicate name is something the Build Protocol tab's
        "Validate protocol…" action flags while someone is authoring (:meth:`_validate_clicked` in
        ``ssvep.ui.session_builder``), the same way :meth:`duplicate_run_labels` already does for
        runs. Schema-shaped problems (unknown key, bad name pattern, wrong level count, which fields
        a design requires/forbids) are the schema's job; what's checked here is cross-item and
        cross-object — things a JSON Schema item definition cannot express, because they depend on
        *this* protocol's own runs (``run``/``levels`` naming a real run label) or on values the
        schema has no opinion on (``between.column`` being reserved, ``between.levels`` differing).
        """
        problems: list[str] = []
        labels = self.run_labels(store_dir)
        seen_names: set[str] = set()
        for c in self.contrasts:
            name = c.get("name", "(unnamed)")
            if name in seen_names:
                problems.append(f"contrast {name!r}: duplicate name")
            seen_names.add(name)

            levels = c.get("levels", [])
            for level in levels:
                if level not in labels:
                    problems.append(
                        f"contrast {name!r}: level {level!r} names no run in this protocol "
                        f"(valid labels: {labels})")
            if len(levels) == 2 and levels[0] == levels[1]:
                problems.append(f"contrast {name!r}: both levels are the same run ({levels[0]!r})")

            run = c.get("run")
            if run is not None and run not in labels:
                problems.append(
                    f"contrast {name!r}: run {run!r} names no run in this protocol "
                    f"(valid labels: {labels})")

            between = c.get("between")
            if between:
                column = between.get("column")
                if column in self._RESERVED_BETWEEN_COLUMNS:
                    problems.append(
                        f"contrast {name!r}: between.column {column!r} is reserved — 'group' is the "
                        f"exclusion declaration pilot/test are dropped through, and 'participant_id'/"
                        f"'species' are identity/administrative, not a manipulation")
                between_levels = between.get("levels", [])
                if len(between_levels) == 2 and between_levels[0] == between_levels[1]:
                    problems.append(
                        f"contrast {name!r}: between.levels are the same value "
                        f"({between_levels[0]!r})")
                else:
                    try:
                        from ..io import declarations
                        declarations.level_tokens([v for v in between_levels if isinstance(v, str)])
                    except Exception as exc:
                        problems.append(f"contrast {name!r}: between.levels: {exc}")
        return problems

    def run_acquisition_blocks(self, store_dir=None) -> list[tuple[str, dict]]:
        """``(run label, acquisition block)`` for every run whose own manifest still carries one.

        Read only to migrate a protocol from before 1.7 (#484): such a block is never used for a
        recording once the protocol has its own. A run that cannot be resolved is skipped, the
        convention of :meth:`run_labels`."""
        out = []
        for g in self.groups:
            for ref in g.runs:
                try:
                    manifest, _ = ref.resolve_manifest(store_dir, allow_drift=True)
                except Exception:
                    continue
                if manifest.get("acquisition"):
                    out.append((ref.label or manifest_io.run_meta(manifest).get(
                        "name", ref.run_manifest), manifest["acquisition"]))
        return out

    def session_acquisition(self, store_dir=None) -> dict | None:
        """The hardware this protocol's sessions are recorded on, or ``None`` if it has none (#484).

        Its own block when it has one. Otherwise (a protocol from before 1.7) the block its runs
        carry, lifted, when every run that carries one carries the same one; ``None`` when none does
        or they disagree — :meth:`acquisition_problem` says which."""
        return _session_hardware(self, self.run_acquisition_blocks(store_dir))[0]

    def acquisition_problem(self, store_dir=None) -> str | None:
        """Why this protocol has no session hardware, in words an operator can act on, or ``None``."""
        return _session_hardware(self, self.run_acquisition_blocks(store_dir))[1]

    def schema_version(self) -> str:
        """The schema version this protocol **is**, not the one it was loaded as.

        Derived, not stored: a protocol carrying questionnaire blocks uses 1.1 features and says so,
        while one without them round-trips byte-identically as whatever it already was. The stamp
        then means "this file needs a 1.1 reader", never "a 1.1 build opened it" — so loading a pilot
        protocol to look at it does not rewrite its version.

        Everything that reports a schema version must come through here. It did not, once: the file
        said 1.1 and every recording made from it said 1.0, because ``to_dict`` derived the value and
        the sidecar stamp read the stale ``protocol_version`` attribute off an in-memory object that
        had been loaded as 1.0 and had questionnaires added since (caught on a hardware
        check, 2026-08-01). One protocol, two answers, depending on which artifact you read.

        Highest feature in use wins: an embedded FPVS run whose item order is drawn per session
        (1.8, #530: an older build would play the design's sample order to everyone, so it must
        refuse the file) outranks a session ``acquisition`` block (1.7, #484), which outranks a
        ``study`` (1.6, #396), which outranks a participant-facing block
        (1.5: ``data_use``, ``safety_screen``, ``start_questionnaire``, the one slot a ``choice`` item
        may be in), which outranks a between/mixed contrast (1.4), which outranks a within-only contrasts block (1.3),
        which outranks embedded runs (1.2), which outranks questionnaires (1.1), since each reader
        necessarily understands every version below it. A protocol whose contrasts are all
        ``"within"`` therefore stays 1.3 — adding stage 2 (#269) must not silently bump every
        existing within-only protocol, including the teaching protocol, to a version they don't use.
        """
        if self.has_per_session_fpvs():
            return protocol_io.PROTOCOL_VERSION_FPVS_PER_SESSION
        if self.acquisition is not None:
            return protocol_io.PROTOCOL_VERSION_SESSION_HARDWARE
        if self.study is not None:
            return protocol_io.PROTOCOL_VERSION_STUDY
        if self.has_participant_features():
            return protocol_io.PROTOCOL_VERSION_DATA_USE
        if any(c.get("design") in ("between", "mixed") for c in self.contrasts):
            return protocol_io.PROTOCOL_VERSION_BETWEEN
        if self.contrasts:
            return protocol_io.PROTOCOL_VERSION_CONTRASTS
        if self.has_embedded_runs():
            return protocol_io.PROTOCOL_VERSION_EMBEDDED_RUNS
        if self.has_questionnaires():
            return protocol_io.PROTOCOL_VERSION_QUESTIONNAIRES
        return self.protocol_version

    def to_dict(self) -> dict:
        d: dict = {"protocol_version": self.schema_version(), "name": self.name}
        if self.version:
            d["version"] = self.version
        if self.study is not None:
            d["study"] = self.study
        if self.description:
            d["description"] = self.description
        d["seed"] = self.seed
        if self.acquisition is not None:
            d["acquisition"] = copy.deepcopy(self.acquisition)
        if self.data_use is not None:
            d["data_use"] = copy.deepcopy(self.data_use)
        if self.safety_screen is not None:
            d["safety_screen"] = copy.deepcopy(self.safety_screen)
        if self.start_questionnaire is not None:
            d["start_questionnaire"] = self.start_questionnaire.to_dict()
        if self.end_questionnaire is not None:
            d["end_questionnaire"] = self.end_questionnaire.to_dict()
        d["groups"] = [g.to_dict() for g in self.groups]
        if self.contrasts:
            d["contrasts"] = copy.deepcopy(self.contrasts)
        return d

    @classmethod
    def from_dict(cls, d: dict) -> "SessionProtocol":
        return cls(
            name=d["name"],
            groups=[Group.from_dict(g) for g in d["groups"]],
            version=d.get("version"),
            study=d.get("study"),
            description=d.get("description"),
            seed=int(d.get("seed", 0)),
            protocol_version=d.get("protocol_version", protocol_io.PROTOCOL_VERSION),
            end_questionnaire=(Q.Questionnaire.from_dict(d["end_questionnaire"])
                               if d.get("end_questionnaire") else None),
            contrasts=copy.deepcopy(d.get("contrasts", [])),
            data_use=copy.deepcopy(d["data_use"]) if "data_use" in d else None,
            safety_screen=copy.deepcopy(d["safety_screen"]) if "safety_screen" in d else None,
            start_questionnaire=(Q.Questionnaire.from_dict(d["start_questionnaire"])
                                 if d.get("start_questionnaire") else None),
            acquisition=(copy.deepcopy(d["acquisition"])
                         if d.get("acquisition") is not None else None),
        )

    def validate(self) -> None:
        """Raise if this protocol is not one a session can run: the schema and its cross-field rules
        (:func:`ssvep.io.protocol.validate_protocol`), then each questionnaire's own checks
        (:meth:`Questionnaire.validate`, e.g. two options of one choice item with the same value),
        so a broken instrument is caught at design time rather than at administration."""
        protocol_io.validate_protocol(self.to_dict())
        for q in self._questionnaires():
            try:
                q.validate()
            except ValueError as exc:
                raise ValueError(f"questionnaire {q.id!r}: {exc}") from exc

    def save(self, path) -> Path:
        return protocol_io.save_protocol(self.to_dict(), path)

    @classmethod
    def load(cls, path) -> "SessionProtocol":
        return cls.from_dict(protocol_io.load_protocol(path))

    def fingerprint(self) -> str:
        """Semantic fingerprint of the protocol itself — its identity in each recording's sidecar."""
        return manifest_io.manifest_fingerprint(self.to_dict())

has_participant_features

has_participant_features()

True once any 1.5 feature is in use — i.e. this file needs a 1.5 reader (#286): a data_use, safety_screen or start_questionnaire block. A closed-choice item can only be in the start questionnaire (validation refuses one anywhere else), so it needs no separate check.

Source code in src/ssvep/runtime/session.py

def has_participant_features(self) -> bool:
    """True once any 1.5 feature is in use — i.e. this file needs a 1.5 reader (#286): a
    ``data_use``, ``safety_screen`` or ``start_questionnaire`` block. A closed-choice item can
    only be in the start questionnaire (validation refuses one anywhere else), so it needs no
    separate check."""
    return (self.data_use is not None or self.safety_screen is not None
            or self.start_questionnaire is not None)

has_embedded_runs

has_embedded_runs()

True once any run carries its manifest inline — i.e. this file needs a 1.2 reader (#115).

Source code in src/ssvep/runtime/session.py

def has_embedded_runs(self) -> bool:
    """True once any run carries its manifest inline — i.e. this file needs a 1.2 reader (#115)."""
    return any(r.embedded for g in self.groups for r in g.runs)

has_per_session_fpvs

has_per_session_fpvs()

True once any embedded run is an FPVS design whose item order is drawn per session (design.fpvs.realise: "per_session", #530) — i.e. this file needs a 1.8 reader. A legacy path reference is not inspected: its manifest lives in the store, not in this file.

Source code in src/ssvep/runtime/session.py

def has_per_session_fpvs(self) -> bool:
    """True once any embedded run is an FPVS design whose item order is drawn per session
    (``design.fpvs.realise: "per_session"``, #530) — i.e. this file needs a 1.8 reader. A legacy
    path reference is not inspected: its manifest lives in the store, not in this file."""
    for g in self.groups:
        for r in g.runs:
            m = r.manifest
            if (isinstance(m, dict) and manifest_io.is_fpvs(m)
                    and (manifest_io.fpvs_params(m) or {}).get("realise") == "per_session"):
                return True
    return False

source_report

source_report(store_dir)

Every run paired with whether its source file still matches it (:meth:RunRef.source_status).

The design-time replacement for the old resolve-time drift stop: authoring can ask "what have I fallen behind on?" and act on it, instead of finding out when a session refuses to start.

Source code in src/ssvep/runtime/session.py

def source_report(self, store_dir) -> list[tuple[Group, RunRef, str]]:
    """Every run paired with whether its source file still matches it (:meth:`RunRef.source_status`).

    The design-time replacement for the old resolve-time drift stop: authoring can ask "what have
    I fallen behind on?" and act on it, instead of finding out when a session refuses to start.
    """
    return [(g, r, r.source_status(store_dir)) for g in self.groups for r in g.runs]

duplicate_run_labels

duplicate_run_labels(store_dir=None)

Runs in this protocol that share a run.name but are different designs (#467).

The name identifies a run; a description need not be unique (two variants of one design legitimately share it), so a shared description is no longer reported. Two runs with one name and different designs is the copy-paste that was never updated for the run it landed on — what happened in the 2026-08-28 pilot, where runs at three frequency bands were all cloned under one name. The same design twice under one name (one run repeated on purpose) is not a collision. Prose differences (run.description) do not make a design different (:func:ssvep.io.manifest.design_fingerprint). The same check runs earlier, at Save in Build run (:func:ssvep.io.manifest.name_clashes); this one catches a pair that was never saved side by side.

Returns one entry per colliding name — {"field": "name", "value": ..., "runs": [(group_label, run_label), ...]} — worst (most runs sharing it) first. A run whose manifest can't be resolved (a legacy reference with no store, or a source file that has moved) is skipped rather than raising.

Source code in src/ssvep/runtime/session.py

def duplicate_run_labels(self, store_dir=None) -> list[dict]:
    """Runs in this protocol that share a ``run.name`` but are different designs (#467).

    The **name** identifies a run; a description need not be unique (two variants of one design
    legitimately share it), so a shared description is no longer reported. Two runs with one name
    and *different* designs is the copy-paste that was never updated for the run it landed on —
    what happened in the 2026-08-28 pilot, where runs at three frequency bands were all cloned
    under one name. The same design twice under one name (one run repeated on purpose) is not a
    collision. Prose differences (``run.description``) do not make a design different
    (:func:`ssvep.io.manifest.design_fingerprint`). The same check runs earlier, at Save in
    Build run (:func:`ssvep.io.manifest.name_clashes`); this one catches a pair that was
    never saved side by side.

    Returns one entry per colliding name — ``{"field": "name", "value": ..., "runs":
    [(group_label, run_label), ...]}`` — worst (most runs sharing it) first. A run whose manifest
    can't be resolved (a legacy reference with no store, or a source file that has moved) is
    skipped rather than raising.
    """
    by_name: dict[str, list[tuple[str, str, str]]] = {}
    for g in self.groups:
        for ref in g.runs:
            try:
                manifest, _ = ref.resolve_manifest(store_dir)
            except Exception:
                continue
            run = manifest_io.run_meta(manifest)
            label = ref.label or run.get("name") or ref.run_manifest or "(unnamed)"
            if run.get("name"):
                by_name.setdefault(run["name"], []).append(
                    (g.label, label, manifest_io.design_fingerprint(manifest)))
    out = [{"field": "name", "value": name, "runs": [(g, lab) for g, lab, _ in runs]}
           for name, runs in by_name.items()
           if len(runs) > 1 and len({fp for _, _, fp in runs}) > 1]
    out.sort(key=lambda d: -len(d["runs"]))
    return out

run_labels

run_labels(store_dir=None)

Every run's label, in protocol order — computed exactly as :func:resolve_session does.

Used to check contrasts[*].levels name real runs (:meth:contrast_problems). A run whose manifest can't be resolved (a legacy reference with no store, or a source file that has moved) is skipped rather than raising, the same convention as :meth:duplicate_run_labels.

Source code in src/ssvep/runtime/session.py

def run_labels(self, store_dir=None) -> list[str]:
    """Every run's label, in protocol order — computed exactly as :func:`resolve_session` does.

    Used to check ``contrasts[*].levels`` name real runs (:meth:`contrast_problems`). A run whose
    manifest can't be resolved (a legacy reference with no store, or a source file that has moved)
    is skipped rather than raising, the same convention as :meth:`duplicate_run_labels`.
    """
    labels = []
    for g in self.groups:
        for ref in g.runs:
            try:
                manifest, _ = ref.resolve_manifest(store_dir)
            except Exception:
                continue
            labels.append(ref.label or manifest_io.run_meta(manifest).get("name", ref.run_manifest))
    return labels

contrast_problems

contrast_problems(store_dir=None)

Human-readable problems with this protocol's contrasts block. Empty ⇒ fine.

Checked here, at design time, rather than only when an analysis tries to run a contrast: a typo'd run label or a copy-pasted duplicate name is something the Build Protocol tab's "Validate protocol…" action flags while someone is authoring (:meth:_validate_clicked in ssvep.ui.session_builder), the same way :meth:duplicate_run_labels already does for runs. Schema-shaped problems (unknown key, bad name pattern, wrong level count, which fields a design requires/forbids) are the schema's job; what's checked here is cross-item and cross-object — things a JSON Schema item definition cannot express, because they depend on this protocol's own runs (run/levels naming a real run label) or on values the schema has no opinion on (between.column being reserved, between.levels differing).

Source code in src/ssvep/runtime/session.py

def contrast_problems(self, store_dir=None) -> list[str]:
    """Human-readable problems with this protocol's ``contrasts`` block. Empty ⇒ fine.

    Checked here, at design time, rather than only when an analysis tries to run a contrast: a
    typo'd run label or a copy-pasted duplicate name is something the Build Protocol tab's
    "Validate protocol…" action flags while someone is authoring (:meth:`_validate_clicked` in
    ``ssvep.ui.session_builder``), the same way :meth:`duplicate_run_labels` already does for
    runs. Schema-shaped problems (unknown key, bad name pattern, wrong level count, which fields
    a design requires/forbids) are the schema's job; what's checked here is cross-item and
    cross-object — things a JSON Schema item definition cannot express, because they depend on
    *this* protocol's own runs (``run``/``levels`` naming a real run label) or on values the
    schema has no opinion on (``between.column`` being reserved, ``between.levels`` differing).
    """
    problems: list[str] = []
    labels = self.run_labels(store_dir)
    seen_names: set[str] = set()
    for c in self.contrasts:
        name = c.get("name", "(unnamed)")
        if name in seen_names:
            problems.append(f"contrast {name!r}: duplicate name")
        seen_names.add(name)

        levels = c.get("levels", [])
        for level in levels:
            if level not in labels:
                problems.append(
                    f"contrast {name!r}: level {level!r} names no run in this protocol "
                    f"(valid labels: {labels})")
        if len(levels) == 2 and levels[0] == levels[1]:
            problems.append(f"contrast {name!r}: both levels are the same run ({levels[0]!r})")

        run = c.get("run")
        if run is not None and run not in labels:
            problems.append(
                f"contrast {name!r}: run {run!r} names no run in this protocol "
                f"(valid labels: {labels})")

        between = c.get("between")
        if between:
            column = between.get("column")
            if column in self._RESERVED_BETWEEN_COLUMNS:
                problems.append(
                    f"contrast {name!r}: between.column {column!r} is reserved — 'group' is the "
                    f"exclusion declaration pilot/test are dropped through, and 'participant_id'/"
                    f"'species' are identity/administrative, not a manipulation")
            between_levels = between.get("levels", [])
            if len(between_levels) == 2 and between_levels[0] == between_levels[1]:
                problems.append(
                    f"contrast {name!r}: between.levels are the same value "
                    f"({between_levels[0]!r})")
            else:
                try:
                    from ..io import declarations
                    declarations.level_tokens([v for v in between_levels if isinstance(v, str)])
                except Exception as exc:
                    problems.append(f"contrast {name!r}: between.levels: {exc}")
    return problems

run_acquisition_blocks

run_acquisition_blocks(store_dir=None)

(run label, acquisition block) for every run whose own manifest still carries one.

Read only to migrate a protocol from before 1.7 (#484): such a block is never used for a recording once the protocol has its own. A run that cannot be resolved is skipped, the convention of :meth:run_labels.

Source code in src/ssvep/runtime/session.py

def run_acquisition_blocks(self, store_dir=None) -> list[tuple[str, dict]]:
    """``(run label, acquisition block)`` for every run whose own manifest still carries one.

    Read only to migrate a protocol from before 1.7 (#484): such a block is never used for a
    recording once the protocol has its own. A run that cannot be resolved is skipped, the
    convention of :meth:`run_labels`."""
    out = []
    for g in self.groups:
        for ref in g.runs:
            try:
                manifest, _ = ref.resolve_manifest(store_dir, allow_drift=True)
            except Exception:
                continue
            if manifest.get("acquisition"):
                out.append((ref.label or manifest_io.run_meta(manifest).get(
                    "name", ref.run_manifest), manifest["acquisition"]))
    return out

session_acquisition

session_acquisition(store_dir=None)

The hardware this protocol's sessions are recorded on, or None if it has none (#484).

Its own block when it has one. Otherwise (a protocol from before 1.7) the block its runs carry, lifted, when every run that carries one carries the same one; None when none does or they disagree — :meth:acquisition_problem says which.

Source code in src/ssvep/runtime/session.py

def session_acquisition(self, store_dir=None) -> dict | None:
    """The hardware this protocol's sessions are recorded on, or ``None`` if it has none (#484).

    Its own block when it has one. Otherwise (a protocol from before 1.7) the block its runs
    carry, lifted, when every run that carries one carries the same one; ``None`` when none does
    or they disagree — :meth:`acquisition_problem` says which."""
    return _session_hardware(self, self.run_acquisition_blocks(store_dir))[0]

acquisition_problem

acquisition_problem(store_dir=None)

Why this protocol has no session hardware, in words an operator can act on, or None.

Source code in src/ssvep/runtime/session.py

def acquisition_problem(self, store_dir=None) -> str | None:
    """Why this protocol has no session hardware, in words an operator can act on, or ``None``."""
    return _session_hardware(self, self.run_acquisition_blocks(store_dir))[1]

schema_version

schema_version()

The schema version this protocol is, not the one it was loaded as.

Derived, not stored: a protocol carrying questionnaire blocks uses 1.1 features and says so, while one without them round-trips byte-identically as whatever it already was. The stamp then means "this file needs a 1.1 reader", never "a 1.1 build opened it" — so loading a pilot protocol to look at it does not rewrite its version.

Everything that reports a schema version must come through here. It did not, once: the file said 1.1 and every recording made from it said 1.0, because to_dict derived the value and the sidecar stamp read the stale protocol_version attribute off an in-memory object that had been loaded as 1.0 and had questionnaires added since (caught on a hardware check, 2026-08-01). One protocol, two answers, depending on which artifact you read.

Highest feature in use wins: an embedded FPVS run whose item order is drawn per session (1.8, #530: an older build would play the design's sample order to everyone, so it must refuse the file) outranks a session acquisition block (1.7, #484), which outranks a study (1.6, #396), which outranks a participant-facing block (1.5: data_use, safety_screen, start_questionnaire, the one slot a choice item may be in), which outranks a between/mixed contrast (1.4), which outranks a within-only contrasts block (1.3), which outranks embedded runs (1.2), which outranks questionnaires (1.1), since each reader necessarily understands every version below it. A protocol whose contrasts are all "within" therefore stays 1.3 — adding stage 2 (#269) must not silently bump every existing within-only protocol, including the teaching protocol, to a version they don't use.

Source code in src/ssvep/runtime/session.py

def schema_version(self) -> str:
    """The schema version this protocol **is**, not the one it was loaded as.

    Derived, not stored: a protocol carrying questionnaire blocks uses 1.1 features and says so,
    while one without them round-trips byte-identically as whatever it already was. The stamp
    then means "this file needs a 1.1 reader", never "a 1.1 build opened it" — so loading a pilot
    protocol to look at it does not rewrite its version.

    Everything that reports a schema version must come through here. It did not, once: the file
    said 1.1 and every recording made from it said 1.0, because ``to_dict`` derived the value and
    the sidecar stamp read the stale ``protocol_version`` attribute off an in-memory object that
    had been loaded as 1.0 and had questionnaires added since (caught on a hardware
    check, 2026-08-01). One protocol, two answers, depending on which artifact you read.

    Highest feature in use wins: an embedded FPVS run whose item order is drawn per session
    (1.8, #530: an older build would play the design's sample order to everyone, so it must
    refuse the file) outranks a session ``acquisition`` block (1.7, #484), which outranks a
    ``study`` (1.6, #396), which outranks a participant-facing block
    (1.5: ``data_use``, ``safety_screen``, ``start_questionnaire``, the one slot a ``choice`` item
    may be in), which outranks a between/mixed contrast (1.4), which outranks a within-only contrasts block (1.3),
    which outranks embedded runs (1.2), which outranks questionnaires (1.1), since each reader
    necessarily understands every version below it. A protocol whose contrasts are all
    ``"within"`` therefore stays 1.3 — adding stage 2 (#269) must not silently bump every
    existing within-only protocol, including the teaching protocol, to a version they don't use.
    """
    if self.has_per_session_fpvs():
        return protocol_io.PROTOCOL_VERSION_FPVS_PER_SESSION
    if self.acquisition is not None:
        return protocol_io.PROTOCOL_VERSION_SESSION_HARDWARE
    if self.study is not None:
        return protocol_io.PROTOCOL_VERSION_STUDY
    if self.has_participant_features():
        return protocol_io.PROTOCOL_VERSION_DATA_USE
    if any(c.get("design") in ("between", "mixed") for c in self.contrasts):
        return protocol_io.PROTOCOL_VERSION_BETWEEN
    if self.contrasts:
        return protocol_io.PROTOCOL_VERSION_CONTRASTS
    if self.has_embedded_runs():
        return protocol_io.PROTOCOL_VERSION_EMBEDDED_RUNS
    if self.has_questionnaires():
        return protocol_io.PROTOCOL_VERSION_QUESTIONNAIRES
    return self.protocol_version

validate

validate()

Raise if this protocol is not one a session can run: the schema and its cross-field rules (:func:ssvep.io.protocol.validate_protocol), then each questionnaire's own checks (:meth:Questionnaire.validate, e.g. two options of one choice item with the same value), so a broken instrument is caught at design time rather than at administration.

Source code in src/ssvep/runtime/session.py

def validate(self) -> None:
    """Raise if this protocol is not one a session can run: the schema and its cross-field rules
    (:func:`ssvep.io.protocol.validate_protocol`), then each questionnaire's own checks
    (:meth:`Questionnaire.validate`, e.g. two options of one choice item with the same value),
    so a broken instrument is caught at design time rather than at administration."""
    protocol_io.validate_protocol(self.to_dict())
    for q in self._questionnaires():
        try:
            q.validate()
        except ValueError as exc:
            raise ValueError(f"questionnaire {q.id!r}: {exc}") from exc

fingerprint

fingerprint()

Semantic fingerprint of the protocol itself — its identity in each recording's sidecar.

Source code in src/ssvep/runtime/session.py

def fingerprint(self) -> str:
    """Semantic fingerprint of the protocol itself — its identity in each recording's sidecar."""
    return manifest_io.manifest_fingerprint(self.to_dict())

ResolvedCheckpoint dataclass

One questionnaire administration in a resolved session (#20).

A checkpoint sits between runs — after a group's last run, or after the whole session — so nothing is streaming while it happens and there are no markers to write, only a timestamp. It is resolved rather than improvised for the same reason a run's run_index is: where in the session a participant was asked is part of the record, not something to reconstruct from file times.

Source code in src/ssvep/runtime/session.py

@dataclass
class ResolvedCheckpoint:
    """One questionnaire administration in a resolved session (#20).

    A checkpoint sits **between** runs — after a group's last run, or after the whole session — so
    nothing is streaming while it happens and there are no markers to write, only a timestamp. It is
    resolved rather than improvised for the same reason a run's ``run_index`` is: where in the session
    a participant was asked is part of the record, not something to reconstruct from file times.
    """
    questionnaire: Q.Questionnaire
    group: str | None                  # the group it follows; None for the end-of-session instrument
    after_run_index: int               # the run_index it comes after (0 ⇒ before any run)
    index: int | None                  # administration ordinal → BIDS run-NNN; None if once per session
    position: dict                     # stamped into the response sidecar, like a run's

    @property
    def task(self) -> str:
        return self.questionnaire.task

    @property
    def label(self) -> str:
        return self.questionnaire.id

ResolvedSession dataclass

Source code in src/ssvep/runtime/session.py

@dataclass
class ResolvedSession:
    protocol_name: str
    protocol_version: str | None
    protocol_fingerprint: str
    subject: str
    session: str
    runs: list[ResolvedRun] = field(default_factory=list)
    checkpoints: list[ResolvedCheckpoint] = field(default_factory=list)
    #: The protocol's study (#396), or ``None`` for the install's implicit single study.
    study: str | None = None
    #: The session's hardware (#484), already stamped into every run's manifest, or ``None`` with
    #: :attr:`acquisition_problem` saying why (no hardware set, or a pre-1.7 protocol whose runs
    #: disagree). A session without it cannot be locked.
    acquisition: dict | None = None
    acquisition_problem: str | None = None
    #: True when :attr:`acquisition` is Set up Session's, because the protocol had none to give.
    acquisition_from_session: bool = False

    def __len__(self) -> int:
        return len(self.runs)

    def __iter__(self):
        return iter(self.runs)

    def sequence(self) -> list[ResolvedRun | ResolvedCheckpoint]:
        """Runs and checkpoints interleaved into the literal running order of the session.

        A checkpoint lands immediately after the run whose index it follows, so this is what the
        operator's checklist shows and what "the next thing to do" means.
        """
        out: list[ResolvedRun | ResolvedCheckpoint] = []
        for r in self.runs:
            out.append(r)
            out.extend(c for c in self.checkpoints if c.after_run_index == r.run_index)
        # A checkpoint on an empty group (or a protocol with no runs at all) has nothing to follow;
        # keep it rather than dropping it silently — a missing item is how a checkpoint gets skipped.
        placed = {id(c) for c in out if isinstance(c, ResolvedCheckpoint)}
        out.extend(c for c in self.checkpoints if id(c) not in placed)
        return out

sequence

sequence()

Runs and checkpoints interleaved into the literal running order of the session.

A checkpoint lands immediately after the run whose index it follows, so this is what the operator's checklist shows and what "the next thing to do" means.

Source code in src/ssvep/runtime/session.py

def sequence(self) -> list[ResolvedRun | ResolvedCheckpoint]:
    """Runs and checkpoints interleaved into the literal running order of the session.

    A checkpoint lands immediately after the run whose index it follows, so this is what the
    operator's checklist shows and what "the next thing to do" means.
    """
    out: list[ResolvedRun | ResolvedCheckpoint] = []
    for r in self.runs:
        out.append(r)
        out.extend(c for c in self.checkpoints if c.after_run_index == r.run_index)
    # A checkpoint on an empty group (or a protocol with no runs at all) has nothing to follow;
    # keep it rather than dropping it silently — a missing item is how a checkpoint gets skipped.
    placed = {id(c) for c in out if isinstance(c, ResolvedCheckpoint)}
    out.extend(c for c in self.checkpoints if id(c) not in placed)
    return out

RunStatus dataclass

Source code in src/ssvep/runtime/session.py

@dataclass
class RunStatus:
    run_index: int
    label: str
    group: str
    state: str                         # pending | partial | done | foreign (see session_progress)
    sidecar: Path
    trials_recorded: int | None = None  # from the sidecar; None if it predates the field
    trials_planned: int | None = None
    foreign: P.ForeignFile | None = None  # whose file it is, when state is foreign
    no_eeg: bool = False               # saved with no EEG at all (#482): markers only, re-run it
    eeg_stalls: dict | None = None     # {"count", "longest_s", "total_s"}: EEG stopped arriving mid-run (#486)

    @property
    def done(self) -> bool:
        """Recorded **and complete**. A partial run is not done — re-recording it is still owed."""
        return self.state == STATE_DONE

    @property
    def recorded(self) -> bool:
        """This protocol's data is on disk for this run (complete or not) — re-recording would
        overwrite it. Not true of a *foreign* run: that file is not this protocol's to replace."""
        return self.state in (STATE_DONE, STATE_PARTIAL)

done property

done

Recorded and complete. A partial run is not done — re-recording it is still owed.

recorded property

recorded

This protocol's data is on disk for this run (complete or not) — re-recording would overwrite it. Not true of a foreign run: that file is not this protocol's to replace.

ItemStatus dataclass

One row of the session checklist — a run or a checkpoint, with its state derived from disk.

Source code in src/ssvep/runtime/session.py

@dataclass
class ItemStatus:
    """One row of the session checklist — a run or a checkpoint, with its state derived from disk."""
    kind: str                          # ITEM_RUN | ITEM_CHECKPOINT
    label: str
    group: str
    state: str                         # pending | partial | done | skipped | foreign
    path: Path                         # the file whose existence decides the state
    run_index: int | None = None       # runs only
    trials_recorded: int | None = None
    trials_planned: int | None = None
    run: ResolvedRun | None = None
    checkpoint: ResolvedCheckpoint | None = None
    foreign: P.ForeignFile | None = None  # whose file sits at ``path``, when state is foreign
    no_eeg: bool = False               # runs only: saved with no EEG at all (#482)
    eeg_stalls: dict | None = None     # runs only: EEG stopped arriving mid-run (#486)

    @property
    def done(self) -> bool:
        return self.state == STATE_DONE

fpvs_session_seed

fpvs_session_seed(seed, subject, session, group, run_label, occurrence=1)

The seed one session's FPVS item order is drawn from (design.fpvs.realise: "per_session", #530).

The randomized-group recipe (:func:_rng) with the session and the run in the key: a one-way SHA-256 of (protocol seed, sub-XXX, ses-YYY, group label, run label), so every participant, every visit and every FPVS run of a session draws its own order, and the same inputs always give the same seed (a relock or a resumed session plays the order it started with; nothing is drawn at lock). occurrence (2, 3, ...) enters the key only when one run label appears more than once in a group (the same design repeated on purpose), so the repeats differ too.

It carries nothing beyond its inputs (AGENTS.md §4): the subject and session codes are the de-identified sub-/ses- labels already in every file name and in session_position; no date, time, operator or other identifier enters it. It is a function of the codes, so it is only as de-identified as they are (docs/FPVS.md says what that means for an export that ever re-codes participants). Truncated to 52 bits so it is exact in any JSON reader.

Source code in src/ssvep/runtime/session.py

def fpvs_session_seed(seed: int, subject: str, session: str, group: str, run_label: str,
                      occurrence: int = 1) -> int:
    """The seed one session's FPVS item order is drawn from (``design.fpvs.realise: "per_session"``, #530).

    The randomized-group recipe (:func:`_rng`) with the session and the run in the key: a one-way
    SHA-256 of ``(protocol seed, sub-XXX, ses-YYY, group label, run label)``, so every participant,
    every visit and every FPVS run of a session draws its own order, and the same inputs always give
    the same seed (a relock or a resumed session plays the order it started with; nothing is drawn
    at lock). ``occurrence`` (2, 3, ...) enters the key only when one run label appears more than
    once in a group (the same design repeated on purpose), so the repeats differ too.

    **It carries nothing beyond its inputs** (AGENTS.md §4): the subject and session codes are the
    de-identified ``sub-``/``ses-`` labels already in every file name and in ``session_position``;
    no date, time, operator or other identifier enters it. It is a function of the codes, so it is
    only as de-identified as they are (docs/FPVS.md says what that means for an export that ever
    re-codes participants). Truncated to 52 bits so it is exact in any JSON reader.
    """
    key = f"{seed}|{subject}|{session}|{group}|{run_label}"
    if occurrence > 1:
        key += f"|{occurrence}"
    return int(hashlib.sha256(key.encode("utf-8")).hexdigest()[:13], 16)

realise_run_for_session

realise_run_for_session(manifest, seed)

manifest with a "per_session" FPVS run's item order drawn from seed (#530), as a new dict; any other manifest is returned as it is. The drawn order is then checked by the rules the loader applies to every FPVS block (the #529 sequence, adjacency and shuffled-cycle checks), so a session is never locked on an order the loader would refuse.

Source code in src/ssvep/runtime/session.py

def realise_run_for_session(manifest: dict, seed: int) -> dict:
    """``manifest`` with a ``"per_session"`` FPVS run's item order drawn from ``seed`` (#530), as a
    new dict; any other manifest is returned as it is. The drawn order is then checked by the rules
    the loader applies to every FPVS block (the #529 sequence, adjacency and shuffled-cycle checks),
    so a session is never locked on an order the loader would refuse."""
    if not manifest_io.is_fpvs(manifest):
        return manifest
    from ..stim import fpvs as _fpvs
    block = manifest_io.fpvs_params(manifest)
    realised = _fpvs.realise_for_session(block, seed)
    if realised is block:
        return manifest
    out = dict(manifest)
    out["design"] = {**manifest["design"], "fpvs": realised}
    problem = manifest_io._fpvs_problem(out)
    if problem:
        raise ValueError(f"the per-session FPVS item order drawn for this session breaks the "
                         f"design's rules: {problem}")
    return out

order_indices

order_indices(order, n, *, subject, seed, label)

This participant's ordering of n runs in a group, plus the counterbalance arm (or None).

Pure and deterministic in (order, n, subject, seed, label). fixed keeps authored order; counterbalanced picks a Williams sequence by subject_number mod #sequences; randomized shuffles with a per-group seed.

Source code in src/ssvep/runtime/session.py

def order_indices(order: str, n: int, *, subject: str, seed: int, label: str) -> tuple[list[int], int | None]:
    """This participant's ordering of ``n`` runs in a group, plus the counterbalance arm (or ``None``).

    Pure and deterministic in ``(order, n, subject, seed, label)``. ``fixed`` keeps authored order;
    ``counterbalanced`` picks a Williams sequence by ``subject_number mod #sequences``; ``randomized``
    shuffles with a per-group seed.
    """
    if order == ORDER_FIXED:
        return list(range(n)), None
    if order == ORDER_COUNTERBALANCED:
        seqs = williams_sequences(n)
        arm = _subject_number(subject) % len(seqs)
        return list(seqs[arm]), arm
    if order == ORDER_RANDOMIZED:
        idx = list(range(n))
        _rng(seed, subject, label).shuffle(idx)
        return idx, None
    raise ValueError(f"unknown order {order!r}; expected one of {ORDERS}")

locate_in_store

locate_in_store(store_dir, run_manifest)

Where a run reference points: run_manifest under store_dir.

store_dir is one directory — the path is simply joined, exactly as it always was — or an ordered search path of them: the run-manifest store plus a site's extra design directories (paths.designs, #298). Against a search path the reference must name exactly one file: present in two directories, it is refused rather than resolved to whichever is listed first, because a pre-1.2 run is loaded from here and silently picking the wrong copy is choosing a different design. Present in none, the first directory's path is returned, so the caller's own missing-file error reads as it always did.

Source code in src/ssvep/runtime/session.py

def locate_in_store(store_dir, run_manifest: str) -> Path:
    """Where a run reference points: ``run_manifest`` under ``store_dir``.

    ``store_dir`` is one directory — the path is simply joined, exactly as it always was — or an
    ordered **search path** of them: the run-manifest store plus a site's extra design directories
    (``paths.designs``, #298). Against a search path the reference must name **exactly one** file:
    present in two directories, it is refused rather than resolved to whichever is listed first,
    because a pre-1.2 run is loaded from here and silently picking the wrong copy is choosing a
    different design. Present in none, the first directory's path is returned, so the caller's own
    missing-file error reads as it always did.
    """
    dirs = _store_dirs(store_dir)
    if not dirs:
        raise ValueError(f"no run-manifest store to look for {run_manifest!r} in")
    if len(dirs) == 1:
        return dirs[0] / run_manifest
    hits = [d / run_manifest for d in dirs if (d / run_manifest).is_file()]
    if len(hits) > 1:
        raise ValueError(
            f"run manifest {run_manifest!r} exists in more than one design directory "
            f"({', '.join(str(h.parent) for h in hits)}), so which one this protocol means is "
            f"ambiguous. Rename or remove all but one.")
    return hits[0] if hits else dirs[0] / run_manifest

acquisition_summary

acquisition_summary(acq)

One line naming a hardware block: amplifier, carrier (with its wiring hash), rate, prep.

Source code in src/ssvep/runtime/session.py

def acquisition_summary(acq: dict | None) -> str:
    """One line naming a hardware block: amplifier, carrier (with its wiring hash), rate, prep."""
    acq = acq or {}
    prep = f", prep {acq['prep']}" if acq.get("prep") else ""
    return (f"{acq.get('device') or acq.get('board') or '?'}, "
            f"{manifest_io.carrier_label(acq)}, {acq.get('sfreq_hz', '?')} Hz{prep}")

with_session_hardware

with_session_hardware(manifest, hardware)

manifest with its acquisition block replaced by the session's (#484), as a new dict.

A run's own block, if it still has one, is dropped rather than merged: hardware is not a run fact, and a merge is how a run's default carrier survived under the session's amplifier. With no session hardware the run carries none at all, so nothing downstream can fall back on the run's.

Source code in src/ssvep/runtime/session.py

def with_session_hardware(manifest: dict, hardware: dict | None) -> dict:
    """``manifest`` with its ``acquisition`` block replaced by the session's (#484), as a new dict.

    A run's own block, if it still has one, is dropped rather than merged: hardware is not a run
    fact, and a merge is how a run's default carrier survived under the session's amplifier. With no
    session hardware the run carries none at all, so nothing downstream can fall back on the run's."""
    out = {k: v for k, v in manifest.items() if k != "acquisition"}
    if hardware is not None:
        out["acquisition"] = copy.deepcopy(hardware)
    return out

resolve_session

resolve_session(protocol, subject, session, store_dir=None, *, allow_drift=False, session_hardware=None)

Instantiate protocol for one participant's session into an ordered, frozen run list.

Inputs are the template, the participant sub-XXX, and the visit ses-YYY. Groups run in authored order; within each group the runs are reordered per its scheme (derived from the participant), then flattened and stamped with run_index 1..N.

store_dir is needed only for a pre-1.2 protocol, whose runs are paths relative to the run-manifest store: those are loaded and re-fingerprinted, raising :class:ProtocolDriftError on a mismatch unless allow_drift. A v1.2 protocol carries its manifests, so store_dir is unused and may be omitted — the whole resolution is then pure in-memory, which is the point: no file outside the protocol can change what a participant runs.

session_hardware is Set up Session's hardware, used only when the protocol has none of its own and none of its runs carry any (:func:_session_hardware); :attr:ResolvedSession.acquisition_from_session then says so.

The resolver never scans sourcedata and never assigns ses-YYY — that is the caller's, kept out so this stays a pure function of its inputs.

Source code in src/ssvep/runtime/session.py

def resolve_session(protocol: SessionProtocol, subject: str, session: str, store_dir=None, *,
                    allow_drift: bool = False,
                    session_hardware: dict | None = None) -> ResolvedSession:
    """Instantiate ``protocol`` for one participant's session into an ordered, frozen run list.

    Inputs are the template, the participant ``sub-XXX``, and the visit ``ses-YYY``. Groups run in
    authored order; within each group the runs are reordered per its scheme (derived from the
    participant), then flattened and stamped with ``run_index`` 1..N.

    ``store_dir`` is needed **only** for a pre-1.2 protocol, whose runs are paths relative to the
    run-manifest store: those are loaded and re-fingerprinted, raising :class:`ProtocolDriftError` on
    a mismatch unless ``allow_drift``. A v1.2 protocol carries its manifests, so ``store_dir`` is
    unused and may be omitted — the whole resolution is then pure in-memory, which is the point: no
    file outside the protocol can change what a participant runs.

    ``session_hardware`` is Set up Session's hardware, used only when the protocol has none of its
    own and none of its runs carry any (:func:`_session_hardware`);
    :attr:`ResolvedSession.acquisition_from_session` then says so.

    The resolver never scans ``sourcedata`` and never assigns ``ses-YYY`` — that is the caller's, kept
    out so this stays a pure function of its inputs.
    """
    if not _SUBJECT_RE.match(subject or ""):
        raise ValueError(f"subject must be a de-identified code like 'sub-105', got {subject!r}")
    # Every file the session writes is named with both, so a date-like one (sub-20310517,
    # ses-20310517) would put the real date on all of them (#286). Refused here, before any run
    # order exists.
    dates.check_subject_label(subject)
    dates.check_session_label(session)
    # One directory or a search path (#298) — RunRef.resolve_manifest / locate_in_store take both.
    store = store_dir
    pfp = protocol.fingerprint()
    runs: list[ResolvedRun] = []
    checkpoints: list[ResolvedCheckpoint] = []
    run_index = 0
    for group in protocol.groups:
        n = len(group.runs)
        idx_order, arm = order_indices(group.order, n, subject=subject, seed=protocol.seed,
                                       label=group.label)
        # Resolved in authored order (as a legacy reference's drift stop always checked them), so
        # which repeat of its label each run is does not depend on this participant's order (#530).
        authored = []
        seen: dict[str, int] = {}
        for ref in group.runs:
            m, fp = ref.resolve_manifest(store, allow_drift=allow_drift,
                                         protocol_name=protocol.name)
            lab = ref.label or manifest_io.run_meta(m).get("name", ref.run_manifest)
            seen[lab] = seen.get(lab, 0) + 1
            authored.append((m, fp, lab, seen[lab]))
        for pos in idx_order:
            ref = group.runs[pos]
            manifest, actual, label, occurrence = authored[pos]
            run_index += 1
            # A per-session FPVS run draws this session's item order now (#530): from a seed derived
            # from the protocol seed and the de-identified codes, so the recording holds exactly what
            # played and a relock draws the same order. ``actual`` stays the design's fingerprint
            # (the protocol's pin); the recording's own manifest is the realised one.
            fpvs_seed = None
            if manifest_io.is_fpvs(manifest):
                from ..stim import fpvs as _fpvs
                if _fpvs.realise_mode(manifest_io.fpvs_params(manifest)) == _fpvs.REALISE_PER_SESSION:
                    fpvs_seed = fpvs_session_seed(protocol.seed, subject, session, group.label,
                                                  label, occurrence)
                    try:
                        manifest = realise_run_for_session(manifest, fpvs_seed)
                    except ValueError as exc:
                        raise ValueError(f"group {group.label!r}, run {label!r}: {exc}") from exc
            # A declared condition order that cannot be played is refused here, before the session
            # locks, not by RunRecorder when Record is pressed (#31). validate() makes the same check
            # on embedded runs; this also covers a legacy path reference and a caller that skipped it.
            try:
                _declared_order(manifest.get("design") or {})
            except TrialOrderError as exc:
                raise TrialOrderError(f"group {group.label!r}, run {label!r}: {exc}") from exc
            position = {
                "protocol": protocol.name,
                "protocol_version": protocol.version,
                "protocol_schema_version": protocol.schema_version(),
                "protocol_fingerprint": pfp,
                "subject": subject,
                "session": session,
                "group": group.label,
                "order": group.order,
                "run_index": run_index,
                "arm": arm,
                "seed": protocol.seed,
                "label": label,
            }
            if protocol.study is not None:
                position["study"] = protocol.study          # which dataset root it belongs to (#396)
            if fpvs_seed is not None:
                position["fpvs_seed"] = fpvs_seed           # this session's FPVS item order (#530)
            runs.append(ResolvedRun(
                manifest=manifest, run_index=run_index, group=group.label, order=group.order,
                run_manifest=ref.run_manifest, fingerprint=actual, label=label, arm=arm,
                position=position))
        if group.questionnaire is not None:
            checkpoints.append(_checkpoint(
                group.questionnaire, protocol, subject, session, pfp,
                group=group.label, after_run_index=run_index,
                # BIDS run-NNN for a repeated instrument = which administration this is. Counted over
                # administrations of THIS task, so a protocol mixing instruments numbers each cleanly.
                index=1 + sum(c.task == group.questionnaire.task for c in checkpoints)))
    # The session's hardware replaces each run's own (#484) — after every run is resolved, so a
    # pre-1.7 protocol is lifted from exactly the manifests this session will run.
    run_blocks = [(r.label, r.manifest["acquisition"]) for r in runs if r.manifest.get("acquisition")]
    hardware, hardware_problem = _session_hardware(protocol, run_blocks, session_hardware)
    from_session = hardware is not None and protocol.acquisition is None and not run_blocks
    for r in runs:
        r.manifest = with_session_hardware(r.manifest, hardware)
    if protocol.end_questionnaire is not None:
        # Once per session ⇒ no BIDS run entity. Unless a group happens to use the same task, in
        # which case it needs one or the two would write the same filename.
        same_task = sum(c.task == protocol.end_questionnaire.task for c in checkpoints)
        checkpoints.append(_checkpoint(
            protocol.end_questionnaire, protocol, subject, session, pfp,
            group=None, after_run_index=run_index,
            index=(same_task + 1) if same_task else None))
    return ResolvedSession(
        protocol_name=protocol.name, protocol_version=protocol.version, protocol_fingerprint=pfp,
        subject=subject, session=session, runs=runs, checkpoints=checkpoints, study=protocol.study,
        acquisition=hardware, acquisition_problem=hardware_problem,
        acquisition_from_session=from_session)

checkpoint_position

checkpoint_position(checkpoint)

The session_position block to save with one checkpoint's answers: the resolved position.

Whether the session was a pilot is no longer written here (#161 wrote pilot: true): the answer carries the session's data-use stamp instead (#286 P3, data_use.purpose), which :func:ssvep.io.bids.is_pilot_session reads. The answer still carries it itself rather than relying on a recording beside it, for the reason #372 found: a questionnaire given before an aborted first run has no recording to speak for it.

Source code in src/ssvep/runtime/session.py

def checkpoint_position(checkpoint: ResolvedCheckpoint) -> dict:
    """The ``session_position`` block to save with one checkpoint's answers: the resolved position.

    Whether the session was a pilot is no longer written here (#161 wrote ``pilot: true``): the
    answer carries the session's data-use stamp instead (#286 P3, ``data_use.purpose``), which
    :func:`ssvep.io.bids.is_pilot_session` reads. The answer still carries it itself rather than
    relying on a recording beside it, for the reason #372 found: a questionnaire given before an
    aborted first run has no recording to speak for it.
    """
    return dict(checkpoint.position)

eeg_stall_phrase

eeg_stall_phrase(stalls)

EEG stopped for 42 s / EEG stopped 3 times, 58 s in all (longest 42 s) from the sidecar's eeg_stalls (#486). Sidecars written before total_s existed say stopped 3 times (longest 42 s): the longest gap is never presented as if it were each gap's length.

Source code in src/ssvep/runtime/session.py

def eeg_stall_phrase(stalls: dict) -> str:
    """``EEG stopped for 42 s`` / ``EEG stopped 3 times, 58 s in all (longest 42 s)`` from the sidecar's
    ``eeg_stalls`` (#486). Sidecars written before ``total_s`` existed say ``stopped 3 times (longest
    42 s)``: the longest gap is never presented as if it were each gap's length."""
    n, longest, total = stalls.get("count"), stalls.get("longest_s"), stalls.get("total_s")
    if not (isinstance(n, int) and n > 1):
        gap = total if total is not None else longest
        return f"EEG stopped for {gap:g} s" if gap is not None else "EEG stopped mid-run"
    if total is not None and longest is not None:
        return f"EEG stopped {n} times, {total:g} s in all (longest {longest:g} s)"
    return f"EEG stopped {n} times" + (f" (longest {longest:g} s)" if longest is not None else "")

session_progress

session_progress(resolved, out_dir)

Which runs of a resolved session are already recorded — derived from disk, no progress file.

Three states, because two were not enough. A run is pending when its *.session.json sidecar is absent, so a crash before saving is correctly offered again. It is done when the sidecar shows every planned trial was recorded and the run was not stopped early. In between it is partial: the sidecar exists, but the run was aborted (Abort or Esc) or lost its stream part-way.

That middle state is the lesson from #86 (docs/PROTOCOLS.md). An 11-second abort still saves (deliberately — partial data is data), and when "the sidecar exists" was the whole test, the checklist called that run recorded and moved on. The operator, who could see perfectly well that it had not been, worked around the checklist by hand-numbering the next run — which took it out of the protocol entirely and cost it its fingerprint pin. Showing partial is what removes the reason to improvise.

The recordings remain the state; there is still no second artifact to desync (a lesson from the same review, one artifact over) — completeness is read out of the sidecars themselves.

A sidecar is read before it is counted (#279). The path is predicted from sub/ses/task/run alone, so a second protocol that reuses a participant code — and a run task label, which ssvep as the Build run default makes the usual case — finds the first protocol's sidecars at its own paths. One whose session_position does not carry this protocol's fingerprint is foreign: never done, never partial (so never offered for a confirmed overwrite), and the lock refuses the session label while it is there. See :func:ssvep.runtime.provenance.foreign_file for the rule, including why a sidecar that names no protocol at all is foreign too.

(An observer-only session persists nothing, so every run reads pending here — the record UI runs those forward-only rather than resuming, since there is nothing on disk to resume from.)

Source code in src/ssvep/runtime/session.py

def session_progress(resolved: ResolvedSession, out_dir) -> list[RunStatus]:
    """Which runs of a resolved session are already recorded — **derived from disk**, no progress file.

    Three states, because two were not enough. A run is **pending** when its ``*.session.json``
    sidecar is absent, so a crash *before* saving is correctly offered again. It is **done** when the
    sidecar shows every planned trial was recorded and the run was not stopped early. In between it
    is **partial**: the sidecar exists, but the run was aborted (Abort or Esc) or lost its stream
    part-way.

    That middle state is the lesson from #86 (`docs/PROTOCOLS.md`). An 11-second abort still saves (deliberately — partial
    data is data), and when "the sidecar exists" was the whole test, the checklist called that run
    recorded and moved on. The operator, who could see perfectly well that it had not been, worked
    around the checklist by hand-numbering the next run — which took it out of the protocol entirely
    and cost it its fingerprint pin. Showing partial is what removes the reason to improvise.

    The recordings remain the state; there is still no second artifact to desync (a lesson from the
    same review, one artifact over) — completeness is read out of the sidecars themselves.

    **A sidecar is read before it is counted (#279).** The path is predicted from
    ``sub``/``ses``/``task``/``run`` alone, so a second protocol that reuses a participant code — and a
    run task label, which ``ssvep`` as the Build run default makes the usual case — finds the first
    protocol's sidecars at its own paths. One whose ``session_position`` does not carry this protocol's
    fingerprint is **foreign**: never done, never partial (so never offered for a confirmed
    overwrite), and the lock refuses the session label while it is there. See
    :func:`ssvep.runtime.provenance.foreign_file` for the rule, including why a sidecar that names no
    protocol at all is foreign too.

    (An observer-only session persists nothing, so every run reads pending here — the record UI runs
    those forward-only rather than resuming, since there is nothing on disk to resume from.)
    """
    from .runner import run_output_base, foreign_run_output
    out: list[RunStatus] = []
    for r in resolved.runs:
        task = manifest_io.run_meta(r.manifest).get("task", "ssvep")
        base = run_output_base(out_dir, resolved.subject, resolved.session, task, r.run_index)
        sidecar = base.parent / (base.name + ".session.json")
        # The recorder's own answer for this stem, so checklist, lock and recorder agree — including
        # on an XDF left without its sidecar, which is foreign (unpaired), not pending: pending would
        # offer a run the recorder then refuses to record.
        foreign = foreign_run_output(base, r.position)
        if foreign is not None:
            out.append(RunStatus(run_index=r.run_index, label=r.label, group=r.group,
                                 state=STATE_FOREIGN, sidecar=sidecar, foreign=foreign))
            continue
        state, got, planned = _run_state(sidecar)
        out.append(RunStatus(run_index=r.run_index, label=r.label, group=r.group,
                             state=state, sidecar=sidecar,
                             trials_recorded=got, trials_planned=planned,
                             no_eeg=state == STATE_PARTIAL and _run_has_no_eeg(sidecar),
                             eeg_stalls=(_run_eeg_stalls(sidecar)
                                         if state in (STATE_DONE, STATE_PARTIAL) else None)))
    return out

next_pending

next_pending(progress)

The first unrecorded run (lowest run_index with no sidecar), or None if none is left.

Deliberately skips partial runs rather than re-offering them: something is on disk for those, and re-recording would overwrite it. They are shown as partial in the checklist and re-run through the confirmed path, so replacing collected data stays a decision rather than a default.

A foreign run is not skipped: it is where this session stops, because nothing of this protocol's has been recorded there and nothing can be (#279). Stepping over it is how another protocol's data used to read as this session being further along — or complete.

Source code in src/ssvep/runtime/session.py

def next_pending(progress: list[RunStatus]) -> RunStatus | None:
    """The first **unrecorded** run (lowest run_index with no sidecar), or None if none is left.

    Deliberately skips *partial* runs rather than re-offering them: something is on disk for those,
    and re-recording would overwrite it. They are shown as partial in the checklist and re-run through
    the confirmed path, so replacing collected data stays a decision rather than a default.

    A *foreign* run is **not** skipped: it is where this session stops, because nothing of this
    protocol's has been recorded there and nothing can be (#279). Stepping over it is how another
    protocol's data used to read as this session being further along — or complete.
    """
    return next((s for s in progress if s.state in (STATE_PENDING, STATE_FOREIGN)), None)

questionnaire_mode

questionnaire_mode(protocol, *, enabled)

Which stamp this session's runs carry. Declared-and-off is not the same as never-declared.

Absence of a response file has to mean one thing for the checklist to resume correctly ("not asked yet"), so the reason there will never be one gets written down instead of inferred. A rig test with no participant and a session whose responses were lost must not read the same way later — that ambiguity is the same lesson, one artifact over.

Source code in src/ssvep/runtime/session.py

def questionnaire_mode(protocol: SessionProtocol | None, *, enabled: bool) -> str:
    """Which stamp this session's runs carry. Declared-and-off is *not* the same as never-declared.

    Absence of a response file has to mean one thing for the checklist to resume correctly ("not
    asked yet"), so the reason there will never be one gets written down instead of inferred. A rig
    test with no participant and a session whose responses were lost must not read the same way later
    — that ambiguity is the same lesson, one artifact over.
    """
    if protocol is None or not protocol.has_questionnaires():
        return Q_NONE
    return Q_ADMINISTERED if enabled else Q_SKIPPED

refresh_ack_check

refresh_ack_check(resolved, advertised_hz, *, screen_index=None, n_screens=None, tol=REFRESH_ACK_TOL)

Compare the stimulus screen's advertised mode rate against every resolved run's design rate.

Pure — the caller supplies advertised_hz (or None when the screen could not be queried), so this is testable offscreen with a mocked rate and never touches a real pyglet display itself (#314). screen_index / n_screens are carried through only to be stamped (finding 4,

352 review) — they say which screen this compared, on a multi-monitor rig, without changing

the comparison itself.

What a mismatch actually breaks — not the presented frequency: the renderer evaluates the ideal waveform at t = frame / clock with clock = the screen's advertised mode (it is never re-pointed at the design's refresh_hz — see ssvep.stim.renderer.FlickerRenderer, flicker.py's module docstring), so the mean presented frequency is exactly the declared one regardless of which mode the panel is actually in. What is lost is frame-exactness: the design's frames-per-cycle arithmetic, its frame-exactness and photosensitivity checks were all computed for the design's refresh_hz and describe nothing about a run on a different clock. A square-wave target that was an exact, jitter-free divisor at the design's rate (e.g. 240/10 = 24 frames/cycle at 240 Hz) can become a non-divisor at the screen's real rate (144/10 = 14.4, so its cycles alternate 14 and 15 frames — interval jitter the design never validated). A driver that misreports the panel's real rate (a 59.94 Hz mode advertised as 59) does not shift the presented frequency (the flicker follows time, #544) but makes it repeat a sample about once a second to stay on time — that is invisible to this check, which only ever compares two advertised numbers; the flik-o-meter (photodiode) measures the real, physical frequency and is the only thing here that would catch it.

FPVS runs (#494) are not a mismatch on a panel whose rate their base rate divides: their design is declared in Hz, seconds and items, and the FPVS renderer rebuilds the frame schedule for the panel and re-checks every rule there (:func:ssvep.stim.fpvs.clock_for_panel), so nothing this acknowledgement warns about is lost. One the panel cannot play is still listed.

Returns {"status", "advertised_hz", "mismatched_runs", "undeclared_runs", "screen_index", "n_screens"}. status is one of :data:REFRESH_MATCH / :data:REFRESH_MISMATCH / :data:REFRESH_UNKNOWN. mismatched_runs is [(run_index, design_hz), …] for every run whose display.refresh_hz differs from advertised_hz by more than tol (fractional). undeclared_runs is [run_index, …] for every run with no positive display.refresh_hz to compare at all — schema requires the key, so this should not happen in practice, but a run this check silently skipped must never let the overall status read as "match" for it (finding 5, #352 review): with no mismatches but at least one undeclared run, status is "unknown", not "match".

Source code in src/ssvep/runtime/session.py

def refresh_ack_check(resolved: ResolvedSession, advertised_hz: float | None,
                      *, screen_index: int | None = None, n_screens: int | None = None,
                      tol: float = REFRESH_ACK_TOL) -> dict:
    """Compare the stimulus screen's advertised mode rate against every resolved run's design rate.

    Pure — the caller supplies ``advertised_hz`` (or ``None`` when the screen could not be queried),
    so this is testable offscreen with a mocked rate and never touches a real pyglet display itself
    (#314). ``screen_index`` / ``n_screens`` are carried through only to be stamped (finding 4,
    #352 review) — they say *which* screen this compared, on a multi-monitor rig, without changing
    the comparison itself.

    What a mismatch actually breaks — **not** the presented frequency: the renderer evaluates the
    ideal waveform at ``t = frame / clock`` with ``clock`` = the screen's advertised mode (it is
    never re-pointed at the design's ``refresh_hz`` — see ``ssvep.stim.renderer.FlickerRenderer``,
    ``flicker.py``'s module docstring), so the *mean* presented frequency is exactly the declared one
    regardless of which mode the panel is actually in. What is lost is **frame-exactness**: the
    design's frames-per-cycle arithmetic, its frame-exactness and photosensitivity checks were all
    computed for the design's ``refresh_hz`` and describe nothing about a run on a different clock. A
    square-wave target that was an exact, jitter-free divisor at the design's rate (e.g. 240/10 = 24
    frames/cycle at 240 Hz) can become a non-divisor at the screen's real rate (144/10 = 14.4, so its
    cycles alternate 14 and 15 frames — interval jitter the design never validated). A driver that
    *misreports* the panel's real rate (a 59.94 Hz mode advertised as 59) does not shift the
    presented frequency (the flicker follows time, #544) but makes it repeat a sample about once a
    second to stay on time — that is invisible to this check, which only ever compares two
    advertised numbers; the flik-o-meter (photodiode) measures the real, physical frequency and is
    the only thing here that would catch it.

    **FPVS runs** (#494) are not a mismatch on a panel whose rate their base rate divides: their
    design is declared in Hz, seconds and items, and the FPVS renderer rebuilds the frame schedule
    for the panel and re-checks every rule there (:func:`ssvep.stim.fpvs.clock_for_panel`), so
    nothing this acknowledgement warns about is lost. One the panel cannot play is still listed.

    Returns ``{"status", "advertised_hz", "mismatched_runs", "undeclared_runs", "screen_index",
    "n_screens"}``. ``status`` is one of :data:`REFRESH_MATCH` / :data:`REFRESH_MISMATCH` /
    :data:`REFRESH_UNKNOWN`. ``mismatched_runs`` is ``[(run_index, design_hz), …]`` for every run
    whose ``display.refresh_hz`` differs from ``advertised_hz`` by more than ``tol`` (fractional).
    ``undeclared_runs`` is ``[run_index, …]`` for every run with no positive ``display.refresh_hz`` to
    compare at all — schema requires the key, so this should not happen in practice, but a run this
    check silently skipped must never let the overall status read as "match" for it (finding 5, #352
    review): with no mismatches but at least one undeclared run, status is "unknown", not "match".
    """
    if advertised_hz is None:
        return {"status": REFRESH_UNKNOWN, "advertised_hz": None, "mismatched_runs": [],
                "undeclared_runs": [], "screen_index": screen_index, "n_screens": n_screens}
    mismatched = []
    undeclared = []
    for r in resolved.runs:
        design_hz = float((r.manifest.get("display") or {}).get("refresh_hz") or 0.0)
        if design_hz <= 0:
            undeclared.append(r.run_index)
            continue
        if abs(advertised_hz - design_hz) / design_hz > tol:
            if _fpvs_playable_at(r.manifest, design_hz, advertised_hz, tol):
                continue        # #494: an FPVS schedule is rebuilt frame-exactly for this panel
            mismatched.append((r.run_index, design_hz))
    if mismatched:
        status = REFRESH_MISMATCH
    elif undeclared:
        status = REFRESH_UNKNOWN
    else:
        status = REFRESH_MATCH
    return {"status": status, "advertised_hz": advertised_hz, "mismatched_runs": mismatched,
            "undeclared_runs": undeclared, "screen_index": screen_index, "n_screens": n_screens}

refresh_ack_stamp

refresh_ack_stamp(check, *, acknowledged=False)

The block :func:recorder_for stamps into every run's session_position (#314).

So a run recorded under a screen-mode mismatch stays describable from its own sidecar without depending on the renderer's after-the-fact provenance.advertised_refresh_hz (#230) — that is a per-run measurement, taken during the run itself; this is the lock-time check plus what the operator was told and did about it, decided once for the whole session. acknowledged only means anything for a "mismatch" status: there was nothing to acknowledge for "match" or "unknown", so it is omitted there rather than stamped False. screen_index / n_screens are included only when known, so a stamp made before #352's fix (or in a test that never supplied them) keeps comparing equal to its old, shorter shape. undeclared_runs is included only when non-empty — the run(s), if any, this check could not compare at all (finding 5).

Source code in src/ssvep/runtime/session.py

def refresh_ack_stamp(check: dict, *, acknowledged: bool = False) -> dict:
    """The block :func:`recorder_for` stamps into every run's ``session_position`` (#314).

    So a run recorded under a screen-mode mismatch stays describable from its own sidecar without
    depending on the renderer's after-the-fact ``provenance.advertised_refresh_hz`` (#230) — that is
    a per-run *measurement*, taken during the run itself; this is the lock-time check plus what the
    operator was told and did about it, decided once for the whole session. ``acknowledged`` only
    means anything for a "mismatch" status: there was nothing to acknowledge for "match" or "unknown",
    so it is omitted there rather than stamped ``False``. ``screen_index`` / ``n_screens`` are
    included only when known, so a stamp made before #352's fix (or in a test that never supplied
    them) keeps comparing equal to its old, shorter shape. ``undeclared_runs`` is included only when
    non-empty — the run(s), if any, this check could not compare at all (finding 5).
    """
    out = {"status": check["status"], "advertised_hz": check["advertised_hz"]}
    if check.get("screen_index") is not None:
        out["screen_index"] = check["screen_index"]
    if check.get("n_screens") is not None:
        out["n_screens"] = check["n_screens"]
    if check.get("undeclared_runs"):
        out["undeclared_runs"] = list(check["undeclared_runs"])
    if check["status"] == REFRESH_MISMATCH:
        out["mismatched_runs"] = [{"run_index": i, "design_hz": hz}
                                  for i, hz in check["mismatched_runs"]]
        out["acknowledged"] = bool(acknowledged)
    return out

carrier_confirm_stamp

carrier_confirm_stamp(acquisition, *, operator_id, at, differs_from_protocol=False)

The :data:CARRIER_CONFIRM_KEY block: the operator confirmed, at lock, that the carrier named here is the one on the participant (#195).

carrier and carrier_hash identify the carrier the session records (the as-run one, which labels the channels); differs_from_protocol says it is not the protocol's. operator is the operator id provenance already records (the typed name, on a profile with no whitelist); at must be date-shifted (:func:ssvep.runtime.dates.stamp), a real date is never written.

Source code in src/ssvep/runtime/session.py

def carrier_confirm_stamp(acquisition: dict, *, operator_id: str, at: str,
                          differs_from_protocol: bool = False) -> dict:
    """The :data:`CARRIER_CONFIRM_KEY` block: the operator confirmed, at lock, that the carrier
    named here is the one on the participant (#195).

    ``carrier`` and ``carrier_hash`` identify the carrier the session records (the as-run one, which
    labels the channels); ``differs_from_protocol`` says it is not the protocol's. ``operator`` is the
    operator id provenance already records (the typed name, on a profile with no whitelist); ``at`` must be **date-shifted**
    (:func:`ssvep.runtime.dates.stamp`), a real date is never written.
    """
    from . import dates
    if not (operator_id and str(operator_id).strip()):
        raise ValueError("a carrier confirmation needs the operator's id")
    if not str(acquisition.get("headset") or "").strip():
        raise ValueError("a carrier confirmation needs the carrier it confirms (the headset is blank)")
    if not (at and dates.is_shifted(at)):
        raise ValueError("a carrier confirmation needs its time, date-shifted "
                         "(ssvep.runtime.dates.stamp): a real date is never written")
    out = {"format": 1, "confirmed": True, "carrier": acquisition.get("headset"),
           "differs_from_protocol": bool(differs_from_protocol),
           "operator": str(operator_id).strip(), "at": at}
    if acquisition.get("carrier_hash"):
        out["carrier_hash"] = acquisition["carrier_hash"]
    return out

session_continuation_stamp

session_continuation_stamp(*, session, kind, changes, prior_recordings, operator_id, at)

The :data:SESSION_CONTINUATION_KEY block (#490): this session was unlocked and locked again while it already held recordings, and what differed from the lock those recordings were made under.

kind is "continued" when the operator was asked at Lock and chose to carry on (the protocol was the same; the purpose / data-use option differed), or "hardware_changed" when only the as-run hardware differed, which keeps the session without asking. changes says what differed, in words; at must be date-shifted (:func:ssvep.runtime.dates.stamp).

Source code in src/ssvep/runtime/session.py

def session_continuation_stamp(*, session: str, kind: str, changes: list[str],
                               prior_recordings: int, operator_id: str, at: str) -> dict:
    """The :data:`SESSION_CONTINUATION_KEY` block (#490): this session was unlocked and locked again
    while it already held recordings, and what differed from the lock those recordings were made under.

    ``kind`` is ``"continued"`` when the operator was asked at Lock and chose to carry on (the
    protocol was the same; the purpose / data-use option differed), or ``"hardware_changed"`` when
    only the as-run hardware differed, which keeps the session without asking. ``changes`` says what
    differed, in words; ``at`` must be **date-shifted** (:func:`ssvep.runtime.dates.stamp`).
    """
    from . import dates
    if kind not in (CONTINUATION_CONTINUED, CONTINUATION_HARDWARE):
        raise ValueError(f"unknown session continuation kind {kind!r}")
    if not (operator_id and str(operator_id).strip()):
        raise ValueError("a session continuation record needs the operator's id")
    if not (at and dates.is_shifted(at)):
        raise ValueError("a session continuation record needs its time, date-shifted "
                         "(ssvep.runtime.dates.stamp): a real date is never written")
    return {"format": 1, "kind": kind, "session": session, "changes": list(changes),
            "prior_recordings": int(prior_recordings), "operator": str(operator_id).strip(), "at": at}

recorder_for

recorder_for(run, resolved, *, operator, data_use, out_dir, device=None, notes=None, seed=0, questionnaires=None, acquisition_override=None, asset_dir=None, refresh_ack=None, approval=None, flicker_ack=None, carrier_confirm=None, session_continuation=None)

Build a :class:RunRecorder for one resolved run — the only place resolver meets recorder.

run_index, session, and the position block all come from the resolved run/session, so they are never hand-set at record time (the whole point of #43). data_use (the session's purpose and tier, :class:ssvep.runtime.data_use.DataUse, or None for observer only) is passed per session by the caller: it is decided at lock. Returns an un-run recorder; the record path drives it.

questionnaires (see :func:questionnaire_mode) rides in the position block rather than in the protocol, because whether they were administered is a fact about this session, decided at the bench — not part of the design. The resolver stays pure and unaware of it.

acquisition_override is the same kind of fact: what headset / electrode type / skin prep physically ran, which the protocol cannot know because electrodes get swapped at the bench. It was threaded through the ad-hoc record path only, so until #115 a run recorded from a protocol — the path the lab actually uses — silently dropped it, which is the wet/dry electrode-metadata gap reappearing one level up. Removing the ad-hoc path made fixing this compulsory.

asset_dir is the locked protocol's asset bundle, where the run's image targets' pictures are; the recorder copies them beside the recording when it saves (#258).

refresh_ack (see :func:refresh_ack_stamp) is the same kind of session-level fact as questionnaires: whether the stimulus screen's mode matched this session's designs at lock time, and what the operator did about it — decided once, at the bench, not part of the protocol (#314). None only for a recorder built with no lock-time check at all (should not happen from the GUI lock path; kept so a caller that never checked still gets a sidecar, just without the stamp).

approval (see :func:ssvep.runtime.approval.stamp) is the approval envelope the session was locked under (#161): its id, version, reference and document hash. Absent when none gated the lock — no envelope, or a profile that is not a lab one — which is itself the statement that none did; the sidecar's site_profile stamp says which kind of deployment it was.

flicker_ack (see :func:ssvep.runtime.flicker_risk.ack_stamp) is the session's highest flicker-risk grade as the operator was shown it at lock, and, for a high grade with a human present, their acknowledgement (#286 P5). Session-level, like refresh_ack; absent when no run presents flicker.

carrier_confirm (see :func:carrier_confirm_stamp) records that the operator, at lock, chose the carrier physically on the participant and it was the session's (#195). Session-level, like refresh_ack; absent only for a recorder built with no lock-time confirmation.

session_continuation (see :func:session_continuation_stamp) records that this lock carried on a session that already held recordings, and what changed since (#490). Absent for a fresh session and for an abort-and-retry relock where nothing changed.

A pilot session is declared by data_use.purpose (#286 P3), not by session_position.pilot (#161's stamp, no longer written; still read, only to restrict). Every between-condition contrast excludes such a recording, and the BIDS converter mirrors it as session_type = pilot.

Source code in src/ssvep/runtime/session.py

def recorder_for(run: ResolvedRun, resolved: ResolvedSession, *, operator, data_use, out_dir,
                 device: dict | None = None, notes: str | None = None, seed: int | None = 0,
                 questionnaires: str | None = None, acquisition_override: dict | None = None,
                 asset_dir=None, refresh_ack: dict | None = None,
                 approval: dict | None = None, flicker_ack: dict | None = None,
                 carrier_confirm: dict | None = None,
                 session_continuation: dict | None = None):
    """Build a :class:`RunRecorder` for one resolved run — the only place resolver meets recorder.

    ``run_index``, ``session``, and the ``position`` block all come from the resolved run/session, so
    they are never hand-set at record time (the whole point of #43). ``data_use`` (the session's
    purpose and tier, :class:`ssvep.runtime.data_use.DataUse`, or ``None`` for observer only) is
    passed per session by the caller: it is decided at lock. Returns an un-run recorder; the record
    path drives it.

    ``questionnaires`` (see :func:`questionnaire_mode`) rides in the position block rather than in the
    protocol, because whether they were administered is a fact about *this session*, decided at the
    bench — not part of the design. The resolver stays pure and unaware of it.

    ``acquisition_override`` is the same kind of fact: what headset / electrode type / skin prep
    **physically** ran, which the protocol cannot know because electrodes get swapped at the bench.
    It was threaded through the ad-hoc record path only, so until #115 a run recorded *from a
    protocol* — the path the lab actually uses — silently dropped it, which is the wet/dry
    electrode-metadata gap reappearing one level up. Removing the ad-hoc path made fixing this compulsory.

    ``asset_dir`` is the locked protocol's asset bundle, where the run's image targets' pictures are;
    the recorder copies them beside the recording when it saves (#258).

    ``refresh_ack`` (see :func:`refresh_ack_stamp`) is the same kind of session-level fact as
    ``questionnaires``: whether the stimulus screen's mode matched this session's designs at lock
    time, and what the operator did about it — decided once, at the bench, not part of the protocol
    (#314). ``None`` only for a recorder built with no lock-time check at all (should not happen from
    the GUI lock path; kept so a caller that never checked still gets a sidecar, just without the stamp).

    ``approval`` (see :func:`ssvep.runtime.approval.stamp`) is the approval envelope the session was
    locked under (#161): its id, version, reference and document hash. Absent when none gated the
    lock — no envelope, or a profile that is not a ``lab`` one — which is itself the statement that
    none did; the sidecar's ``site_profile`` stamp says which kind of deployment it was.

    ``flicker_ack`` (see :func:`ssvep.runtime.flicker_risk.ack_stamp`) is the session's highest
    flicker-risk grade as the operator was shown it at lock, and, for a ``high`` grade with a human
    present, their acknowledgement (#286 P5). Session-level, like ``refresh_ack``; absent when no run
    presents flicker.

    ``carrier_confirm`` (see :func:`carrier_confirm_stamp`) records that the operator, at lock, chose
    the carrier physically on the participant and it was the session's (#195). Session-level, like
    ``refresh_ack``; absent only for a recorder built with no lock-time confirmation.

    ``session_continuation`` (see :func:`session_continuation_stamp`) records that this lock carried
    on a session that already held recordings, and what changed since (#490). Absent for a fresh
    session and for an abort-and-retry relock where nothing changed.

    A pilot session is declared by ``data_use.purpose`` (#286 P3), not by ``session_position.pilot``
    (#161's stamp, no longer written; still read, only to restrict). Every between-condition contrast
    excludes such a recording, and the BIDS converter mirrors it as ``session_type = pilot``.
    """
    from . import site, study
    from .runner import RunRecorder
    # The declared data root is checked where a recording is about to be written into it (#298):
    # missing is an error here, never a directory quietly created or a recording sent elsewhere.
    # ...and is this session's study's own root (#396): the protocol names its study, and a study's
    # sessions are recorded into its root only. A declared study's root is marked the first time.
    # One read of the tree's stamps serves both checks (one_scan).
    with study.one_scan():
        site.check_data_root(out_dir)
        study.claim_root(out_dir, resolved.study)
    position = dict(run.position)
    if questionnaires:
        position["questionnaires"] = questionnaires
    if refresh_ack is not None:
        position["refresh_ack"] = refresh_ack
    if approval is not None:
        position["approval"] = dict(approval)
    if flicker_ack is not None:
        from .flicker_risk import ACK_KEY
        position[ACK_KEY] = dict(flicker_ack)
    if carrier_confirm is not None:
        position[CARRIER_CONFIRM_KEY] = dict(carrier_confirm)
    if session_continuation is not None:
        position[SESSION_CONTINUATION_KEY] = dict(session_continuation)
    return RunRecorder(
        manifest=run.manifest, operator=operator, data_use=data_use, out_dir=str(out_dir),
        subject=resolved.subject, session=resolved.session, run_index=run.run_index,
        seed=seed, device=device, notes=notes, position=position,
        acquisition_override=acquisition_override,
        asset_dir=str(asset_dir) if asset_dir is not None else None)

session_checklist

session_checklist(resolved, out_dir, *, questionnaires=True)

The whole session as an ordered to-do list — runs and checkpoints interleaved.

Completion is derived from disk for both kinds, by the same rule and for the same reason: a run is done when its *.session.json exists (and reports every planned trial), a checkpoint when its _beh.json exists — and, for both, when that file's own session_position says it is this protocol's (#279). There is still no separate progress file to desync from the data.

The last clause is the symptom checkpoints' safeguard. Their built-in instruments use fixed task labels (postcondition, endofsession), so a second protocol reusing a participant code predicts the first one's answer files even when it gave its runs a distinct task label. Counted by existence, those read as already answered and were never presented. A file that is not this protocol's is foreign instead: not done, not skipped over by :func:next_pending_item, and a reason the lock refuses the session label. Foreign outranks skipped — switching questionnaires off does not make another protocol's data this session's.

questionnaires=False (the operator's "don't run questionnaires" switch — a rig test with no participant) marks checkpoints skipped rather than hiding them: they stay visible, so nobody has to wonder later whether the protocol had any.

Source code in src/ssvep/runtime/session.py

def session_checklist(resolved: ResolvedSession, out_dir, *, questionnaires: bool = True
                      ) -> list[ItemStatus]:
    """The whole session as an ordered to-do list — runs and checkpoints interleaved.

    Completion is derived from disk for both kinds, by the same rule and for the same reason: a run
    is done when its ``*.session.json`` exists (and reports every planned trial), a checkpoint when
    its ``_beh.json`` exists — and, for both, when that file's own ``session_position`` says it is
    **this protocol's** (#279). There is still no separate progress file to desync from the data.

    The last clause is the symptom checkpoints' safeguard. Their built-in instruments use fixed task
    labels (``postcondition``, ``endofsession``), so a second protocol reusing a participant code
    predicts the first one's answer files even when it gave its runs a distinct task label. Counted by
    existence, those read as *already answered* and were never presented. A file that is not this
    protocol's is **foreign** instead: not done, not skipped over by :func:`next_pending_item`, and a
    reason the lock refuses the session label. Foreign outranks *skipped* — switching questionnaires
    off does not make another protocol's data this session's.

    ``questionnaires=False`` (the operator's "don't run questionnaires" switch — a rig test with no
    participant) marks checkpoints **skipped** rather than hiding them: they stay visible, so nobody
    has to wonder later whether the protocol had any.
    """
    runs = {p.run_index: p for p in session_progress(resolved, out_dir)}
    out: list[ItemStatus] = []
    for item in resolved.sequence():
        if isinstance(item, ResolvedRun):
            p = runs[item.run_index]
            out.append(ItemStatus(kind=ITEM_RUN, label=p.label, group=p.group, state=p.state,
                                  path=p.sidecar, run_index=p.run_index,
                                  trials_recorded=p.trials_recorded,
                                  trials_planned=p.trials_planned, run=item, foreign=p.foreign,
                                  no_eeg=p.no_eeg, eeg_stalls=p.eeg_stalls))
        else:
            path = Q.response_path(out_dir, resolved.subject, resolved.session,
                                   item.task, item.index)
            foreign = P.foreign_file(path, item.position)
            if foreign is not None:
                state = STATE_FOREIGN
            elif not questionnaires:
                state = STATE_SKIPPED
            else:
                # Q.is_answered is the one definition of "answered" (exists AND is this protocol's).
                answered = Q.is_answered(out_dir, resolved.subject, resolved.session, item.task,
                                         item.index, position=item.position)
                state = STATE_DONE if answered else STATE_PENDING
            out.append(ItemStatus(kind=ITEM_CHECKPOINT, label=item.questionnaire.id,
                                  group=item.group or "—", state=state, path=path,
                                  checkpoint=item, foreign=foreign))
    return out

next_pending_item

next_pending_item(checklist)

The next thing to do: the first item that is neither done, partial, nor skipped.

Partial runs are stepped over for the same reason :func:next_pending steps over them — something is on disk, so replacing it stays a deliberate re-run rather than the default.

A foreign item is not stepped over (#279): the session stops there, and says why. Walking past it is precisely how a symptom checkpoint went unasked, and how a session holding nothing but another protocol's data read "Session complete".

Source code in src/ssvep/runtime/session.py

def next_pending_item(checklist: list[ItemStatus]) -> ItemStatus | None:
    """The next thing to do: the first item that is neither done, partial, nor skipped.

    Partial runs are stepped over for the same reason :func:`next_pending` steps over them — something
    is on disk, so replacing it stays a deliberate re-run rather than the default.

    A *foreign* item is **not** stepped over (#279): the session stops there, and says why. Walking
    past it is precisely how a symptom checkpoint went unasked, and how a session holding nothing but
    another protocol's data read "Session complete".
    """
    return next((i for i in checklist if i.state in (STATE_PENDING, STATE_FOREIGN)), None)

foreign_items

foreign_items(checklist)

The items whose path already holds another protocol's file — what the lock refuses on (#279).

Source code in src/ssvep/runtime/session.py

def foreign_items(checklist: list[ItemStatus]) -> list[ItemStatus]:
    """The items whose path already holds another protocol's file — what the lock refuses on (#279)."""
    return [i for i in checklist if i.state == STATE_FOREIGN]

session_dir_foreign_files

session_dir_foreign_files(out_dir, resolved)

Every foreign or unreadable file anywhere under this resolved session's own directory (#279).

:func:foreign_items (via :func:session_checklist) only looks at the paths this protocol would itself predict, from its own run and questionnaire task labels. A visit that ran a different protocol under this same sub/ses label may have used entirely different task labels for both, so its files sit at paths this protocol never predicts and never sees — the gap "one session folder = one protocol" closes: not just the paths a protocol cares about, but the directory it would be locked into.

Every *.session.json (a run's sidecar) and *_beh.json (a checkpoint's) anywhere under sourcedata/sub-XXX/ses-YYY/ is read and judged the same way any single one would be at its own predicted path (:func:ssvep.runtime.provenance.foreign_file, against this resolved session's own fingerprint) — the same identity rule, and the same fail-closed treatment of a file that names no protocol or cannot be read. An .xdf with no sidecar beside it counts too, for the same reason :func:ssvep.runtime.runner.foreign_run_output already treats one that way at a predicted path: nobody has shown it belongs to any protocol, let alone this one.

A file that is this protocol's own (own fingerprint, including one it rescued earlier) is not returned. Returns [] if the directory does not exist yet — a brand new session label.

The stimuli/ subtree (:data:ssvep.io.assets.STIMULI_DIR) is skipped for the two JSON globs: it holds copied pictures, named by the design (pins puts no extension restriction on a pinned file), not run/checkpoint sidecars — a picture that happens to be named e.g. pic.session.json is not one, and judging it as one only produces an unreadable-file refusal nobody can act on.

Source code in src/ssvep/runtime/session.py

def session_dir_foreign_files(out_dir, resolved: ResolvedSession) -> list[P.ForeignFile]:
    """Every foreign or unreadable file anywhere under this resolved session's own directory (#279).

    :func:`foreign_items` (via :func:`session_checklist`) only looks at the paths *this* protocol
    would itself predict, from its own run and questionnaire task labels. A visit that ran a
    *different* protocol under this same ``sub``/``ses`` label may have used entirely different task
    labels for both, so its files sit at paths this protocol never predicts and never sees — the gap
    "one session folder = one protocol" closes: not just the paths a protocol cares about, but the
    directory it would be locked into.

    Every ``*.session.json`` (a run's sidecar) and ``*_beh.json`` (a checkpoint's) anywhere under
    ``sourcedata/sub-XXX/ses-YYY/`` is read and judged the same way any single one would be at its own
    predicted path (:func:`ssvep.runtime.provenance.foreign_file`, against this resolved session's own
    fingerprint) — the same identity rule, and the same fail-closed treatment of a file that names no
    protocol or cannot be read. An ``.xdf`` with no sidecar beside it counts too, for the same reason
    :func:`ssvep.runtime.runner.foreign_run_output` already treats one that way at a predicted path:
    nobody has shown it belongs to any protocol, let alone this one.

    A file that *is* this protocol's own (own fingerprint, including one it rescued earlier) is not
    returned. Returns ``[]`` if the directory does not exist yet — a brand new session label.

    The ``stimuli/`` subtree (:data:`ssvep.io.assets.STIMULI_DIR`) is skipped for the two JSON globs:
    it holds copied pictures, named by the design (``pins`` puts no extension restriction on a pinned
    file), not run/checkpoint sidecars — a picture that happens to be named e.g. ``pic.session.json``
    is not one, and judging it as one only produces an unreadable-file refusal nobody can act on.
    """
    root = _session_dir(out_dir, resolved.subject, resolved.session)
    if not root.is_dir():
        return []
    stimuli_dir = root / assets_io.STIMULI_DIR
    position = {"protocol": resolved.protocol_name, "protocol_version": resolved.protocol_version,
                "protocol_fingerprint": resolved.protocol_fingerprint}
    found: dict[str, P.ForeignFile] = {}
    for side in sorted(root.rglob("*.session.json")):
        if stimuli_dir in side.parents:
            continue
        f = P.foreign_file(side, position)
        if f is not None:
            found[str(side)] = f
    for side in sorted(root.rglob("*_beh.json")):
        if stimuli_dir in side.parents:
            continue
        f = P.foreign_file(side, position)
        if f is not None:
            found[str(side)] = f
    for xdf in sorted(root.rglob("*.xdf")):
        side = xdf.with_name(xdf.name[:-4] + ".session.json")   # strip ".xdf", not a generic stem
        if side.exists():
            continue          # already read above (whatever it said) — don't double-count the run
        found.setdefault(str(xdf), P.ForeignFile(xdf, P.FOREIGN_UNPAIRED))
    return list(found.values())

session_recording_count

session_recording_count(out_dir, subject, session)

How many recordings (run sidecars, and XDFs with no sidecar) a session folder holds (#490).

Counts files on disk whatever protocol they belong to: it answers "is there anything here a relock could mix with", not "how far has this protocol got" (that is :func:session_progress). Zero for a session folder that does not exist yet and for an observer-only session, which writes nothing.

Source code in src/ssvep/runtime/session.py

def session_recording_count(out_dir, subject: str, session: str) -> int:
    """How many recordings (run sidecars, and XDFs with no sidecar) a session folder holds (#490).

    Counts files on disk whatever protocol they belong to: it answers "is there anything here a
    relock could mix with", not "how far has this protocol got" (that is :func:`session_progress`).
    Zero for a session folder that does not exist yet and for an observer-only session, which writes
    nothing.
    """
    root = _session_dir(out_dir, subject, session)
    if not root.is_dir():
        return 0
    stimuli_dir = root / assets_io.STIMULI_DIR
    n = 0
    for side in root.rglob("*.session.json"):
        if stimuli_dir not in side.parents:
            n += 1
    for xdf in root.rglob("*.xdf"):
        if stimuli_dir not in xdf.parents and not xdf.with_name(xdf.name[:-4] + ".session.json").exists():
            n += 1
    return n

session_exists

session_exists(out_dir, subject, session)

True if a recording session dir already exists at all.

Not wired to the GUI, and not what protects a session's files: an existing directory is the normal state of a session being resumed. What the lock checks instead is whose files sit at the paths this protocol would write (:func:foreign_items, #279), and whose files sit anywhere else in the directory too (:func:session_dir_foreign_files) — one session folder holds one protocol's data.

Source code in src/ssvep/runtime/session.py

def session_exists(out_dir, subject: str, session: str) -> bool:
    """True if a recording session dir already exists at all.

    Not wired to the GUI, and not what protects a session's files: an existing directory is the
    normal state of a session being resumed. What the lock checks instead is *whose* files sit at the
    paths this protocol would write (:func:`foreign_items`, #279), and whose files sit anywhere else
    in the directory too (:func:`session_dir_foreign_files`) — one session folder holds one
    protocol's data.
    """
    return _session_dir(out_dir, subject, session).is_dir()

next_session

next_session(out_dir, subject)

Suggest the next ses-NNN for a participant by scanning existing session dirs (ses-001 if none).

A convenience for prefilling the operator's session field — deliberately separate from the pure resolver, which takes ses-YYY as an input rather than deriving it from disk.

Source code in src/ssvep/runtime/session.py

def next_session(out_dir, subject: str) -> str:
    """Suggest the next ``ses-NNN`` for a participant by scanning existing session dirs (``ses-001`` if none).

    A convenience for prefilling the operator's session field — deliberately separate from the pure
    resolver, which takes ``ses-YYY`` as an input rather than deriving it from disk.
    """
    subj_dir = Path(out_dir) / "sourcedata" / subject
    nums = [int(m.group(1)) for p in (subj_dir.iterdir() if subj_dir.is_dir() else [])
            if (m := _SES_RE.match(p.name)) and p.is_dir()]
    return f"ses-{(max(nums) + 1) if nums else 1:03d}"

Session purpose and data-use tier

ssvep.runtime.data_use

Session purpose and data-use tier: the one policy every gate reads (#286 P3, D4/D5/D11).

A recording's fate (whether it is kept, mirrored to lab storage, converted, pooled into a contrast, or exported) depends on two facts stamped into it at lock: why the session was run (its purpose) and what the participant agreed to (its tier). This module holds both vocabularies, the single matrix that maps them to what may happen, and the fail-closed reader that turns a sidecar's stamp back into a decision. It is pure and stdlib-only on purpose, so every gate can import it and none can drift from another by re-deriving the rule.

Purposes (D5, D11): research (participant consent), pilot (a team-member volunteer, no consent screen, the operator attests), teaching (a class; a fixed course-use-only tier) and bench (no human present). Which ones a deployment offers depends on its site profile's kind (:data:PURPOSES_BY_KIND).

Tiers (D4) are cumulative: 0 observer only (the participant takes part, and nothing from the session is saved), 1 this study only, 2 this study and future research by this team, 3 share openly. Only research asks for one of these; the other purposes have one fixed tier each (:data:TIERS_BY_PURPOSE), plus 0 so that "keep nothing" exists everywhere. Tier 0 permits nothing at all: under it nothing is ever written, not even temporarily.

Unstamped means refused (D11), with one exception (#286 Q1): a recording with no stamp counts as teaching on a teaching machine, unless it says it was made on a lab machine. That is the most restrictive purpose (never pushed, never exported), so inferring it widens nothing. There is no migration: an older sidecar's three consent flags are never translated into a tier, because "public yes, future no" has no tier and inventing one would be consent nobody gave.

The matrix and the reader are the policy; the session lock, the writers and the push, export and conversion gates are where it is applied.

Purpose

Bases: str, Enum

Why a session was run. Set at lock, stamped into every file the session writes (D5).

Source code in src/ssvep/runtime/data_use.py

class Purpose(str, Enum):
    """Why a session was run. Set at lock, stamped into every file the session writes (D5)."""
    RESEARCH = "research"
    PILOT = "pilot"
    TEACHING = "teaching"
    BENCH = "bench"

Action

Bases: str, Enum

Something that may or may not happen to a recording.

Source code in src/ssvep/runtime/data_use.py

class Action(str, Enum):
    """Something that may or may not happen to a recording."""
    PERSIST = "persist"                          # written to disk at all
    PUSH = "push"                                # mirrored to the lab's own storage
    CONVERT = "convert"                          # turned into a local BIDS view
    POOL = "pool"                                # pooled into a contrast with other recordings
    EXPORT_FUTURE_USE = "export_future_use"      # exported for reuse by this team
    EXPORT_PUBLIC_SHARE = "export_public_share"  # exported for a public repository

PurposeRules dataclass

What a session of one purpose involves before anything is recorded.

Source code in src/ssvep/runtime/data_use.py

@dataclass(frozen=True)
class PurposeRules:
    """What a session of one purpose involves before anything is recorded."""
    tier_question: bool      # the participant is asked their tier (research; teaching's 0 vs course)
    safety_screen: bool      # the photosensitivity safety screen runs (D10: whenever a human is present)
    attestation: bool        # the operator attests the two pilot booleans (D5)
    envelope: str            # the approval envelope at lock: "refuse", "warn" or "not_applied"

WordingChoice dataclass

The wording a research session will use, or why it can't be locked (wording is None).

Source code in src/ssvep/runtime/data_use.py

@dataclass(frozen=True)
class WordingChoice:
    """The wording a research session will use, or why it can't be locked (``wording`` is None)."""
    wording: dict | None
    source: str | None = None        # "protocol" or "builtin"
    sha256: str | None = None
    refused: str | None = None

DataUse dataclass

One recording's purpose and tier, as stamped (or, for Q1 only, inferred).

mode, wording ({"source", "sha256"}), offered and minimum describe a research session's consent and are absent otherwise; attestation is a pilot session's two booleans and absent otherwise. inferred is true only for an unstamped recording read as teaching under the Q1 rule; it is never written (:meth:to_stamp refuses one).

Source code in src/ssvep/runtime/data_use.py

@dataclass(frozen=True)
class DataUse:
    """One recording's purpose and tier, as stamped (or, for Q1 only, inferred).

    ``mode``, ``wording`` (``{"source", "sha256"}``), ``offered`` and ``minimum`` describe a
    research session's consent and are absent otherwise; ``attestation`` is a pilot session's two
    booleans and absent otherwise. ``inferred`` is true only for an unstamped recording read as
    teaching under the Q1 rule; it is never written (:meth:`to_stamp` refuses one).
    """
    purpose: str
    tier: object
    entered_by: str = "operator"
    mode: str | None = None
    wording: dict | None = None
    offered: tuple | None = None
    minimum: int | None = None
    attestation: dict | None = None
    inferred: bool = False

    def __post_init__(self):
        if not self.inferred:
            problems = stamp_problems(self.to_stamp())
            if problems:
                raise ValueError("invalid data-use stamp: " + "; ".join(problems))

    def to_stamp(self) -> dict:
        """The block a writer puts under :data:`STAMP_KEY` (absent fields left out).

        Raises ``ValueError`` for an inferred record: an inference about an unstamped recording is
        not a stamp, and writing it as one would turn a guess into a durable claim.
        """
        if self.inferred:
            raise ValueError("an inferred data-use record (an unstamped recording read as teaching) "
                             "is never written as a stamp")
        d: dict = {"purpose": self.purpose, "tier": self.tier, "entered_by": self.entered_by}
        if self.mode is not None:
            d["mode"] = self.mode
        if self.wording is not None:
            d["wording"] = dict(self.wording)
        if self.offered is not None:
            d["offered"] = list(self.offered)
        if self.minimum is not None:
            d["minimum"] = self.minimum
        if self.attestation is not None:
            d["attestation"] = dict(self.attestation)
        return d

    def allowed(self) -> frozenset:
        return allowed(self.purpose, self.tier)

to_stamp

to_stamp()

The block a writer puts under :data:STAMP_KEY (absent fields left out).

Raises ValueError for an inferred record: an inference about an unstamped recording is not a stamp, and writing it as one would turn a guess into a durable claim.

Source code in src/ssvep/runtime/data_use.py

def to_stamp(self) -> dict:
    """The block a writer puts under :data:`STAMP_KEY` (absent fields left out).

    Raises ``ValueError`` for an inferred record: an inference about an unstamped recording is
    not a stamp, and writing it as one would turn a guess into a durable claim.
    """
    if self.inferred:
        raise ValueError("an inferred data-use record (an unstamped recording read as teaching) "
                         "is never written as a stamp")
    d: dict = {"purpose": self.purpose, "tier": self.tier, "entered_by": self.entered_by}
    if self.mode is not None:
        d["mode"] = self.mode
    if self.wording is not None:
        d["wording"] = dict(self.wording)
    if self.offered is not None:
        d["offered"] = list(self.offered)
    if self.minimum is not None:
        d["minimum"] = self.minimum
    if self.attestation is not None:
        d["attestation"] = dict(self.attestation)
    return d

Reading dataclass

What :func:data_use_of made of a sidecar: a record, or why it has none (refused).

Source code in src/ssvep/runtime/data_use.py

@dataclass(frozen=True)
class Reading:
    """What :func:`data_use_of` made of a sidecar: a record, or why it has none (``refused``)."""
    record: DataUse | None
    refused: str | None = None

LockDecision dataclass

What a session lock decided about data use: a stamp, "write nothing", or a refusal.

• record set: the session is recorded, and every file it writes carries record.to_stamp().
• observer: the session runs (the person under the flicker is real, so the safety checks and questionnaires still happen) but nothing is written, not even temporarily, and no stamp is ever built (tier 0, D4/D7).
• refused: the lock must not happen, and this says why in the operator's words.

Source code in src/ssvep/runtime/data_use.py

@dataclass(frozen=True)
class LockDecision:
    """What a session lock decided about data use: a stamp, "write nothing", or a refusal.

    * ``record`` set: the session is recorded, and every file it writes carries ``record.to_stamp()``.
    * ``observer``: the session runs (the person under the flicker is real, so the safety checks and
      questionnaires still happen) but **nothing is written**, not even temporarily, and no stamp is
      ever built (tier 0, D4/D7).
    * ``refused``: the lock must not happen, and this says why in the operator's words.
    """
    purpose: str | None
    record: DataUse | None = None
    observer: bool = False
    refused: str | None = None
    #: Set with ``refused`` when the only thing missing is the participant's own answer from the
    #: participant screens (#286 P4): the lock runs them and decides again with their answer. Still
    #: a refusal to anyone who does not know this flag, so it fails closed.
    screens_pending: bool = False

    @property
    def persists(self) -> bool:
        """Whether anything this session produces may be written to disk at all."""
        return (self.refused is None and self.record is not None
                and Action.PERSIST in self.record.allowed())

    def describe(self) -> str:
        """One line for the lock status: the purpose and tier, or that nothing is saved."""
        if self.screens_pending:
            return f"{self.purpose}: the participant chooses their option on screen at lock"
        if self.refused is not None:
            return f"not lockable: {self.refused}"
        if self.observer:
            return f"{self.purpose}, OBSERVER ONLY: nothing from this session is saved"
        rec = self.record
        if rec is None:
            return "no data-use decision"
        tier = f"option {rec.tier}" if type(rec.tier) is int else str(rec.tier).replace("_", " ")
        who = ""
        if rec.purpose in ("research", "teaching"):
            who = (" (entered by the operator)" if rec.entered_by == "operator"
                   else " (chosen by the participant on screen)")
        return f"{rec.purpose}, {tier}{who}"

persists property

persists

Whether anything this session produces may be written to disk at all.

describe

describe()

One line for the lock status: the purpose and tier, or that nothing is saved.

Source code in src/ssvep/runtime/data_use.py

def describe(self) -> str:
    """One line for the lock status: the purpose and tier, or that nothing is saved."""
    if self.screens_pending:
        return f"{self.purpose}: the participant chooses their option on screen at lock"
    if self.refused is not None:
        return f"not lockable: {self.refused}"
    if self.observer:
        return f"{self.purpose}, OBSERVER ONLY: nothing from this session is saved"
    rec = self.record
    if rec is None:
        return "no data-use decision"
    tier = f"option {rec.tier}" if type(rec.tier) is int else str(rec.tier).replace("_", " ")
    who = ""
    if rec.purpose in ("research", "teaching"):
        who = (" (entered by the operator)" if rec.entered_by == "operator"
               else " (chosen by the participant on screen)")
    return f"{rec.purpose}, {tier}{who}"

purposes_for

purposes_for(kind)

The purposes a deployment of this kind offers at lock (D11).

Source code in src/ssvep/runtime/data_use.py

def purposes_for(kind: str) -> tuple[str, ...]:
    """The purposes a deployment of this kind offers at lock (D11)."""
    _check_kind(kind)
    return PURPOSES_BY_KIND[kind]

allowed

allowed(purpose, tier)

The matrix cell for (purpose, tier): every :class:Action it allows (empty = none).

An unknown purpose, or a tier that purpose never carries, allows nothing. True is not the tier 1 (True == 1 in Python): only a real int is a research tier.

Source code in src/ssvep/runtime/data_use.py

def allowed(purpose: str, tier) -> frozenset:
    """The matrix cell for ``(purpose, tier)``: every :class:`Action` it allows (empty = none).

    An unknown purpose, or a tier that purpose never carries, allows nothing. ``True`` is not the
    tier 1 (``True == 1`` in Python): only a real ``int`` is a research tier.
    """
    if purpose not in TIERS_BY_PURPOSE or not _is_tier_of(purpose, tier):
        return frozenset()
    return MATRIX.get((purpose, tier), frozenset())

wording_sha256

wording_sha256(block)

SHA-256 of a wording block in canonical JSON (sorted keys, no whitespace, UTF-8).

What a research recording is stamped with, so the words the participant was shown can be identified later without copying them into every file.

Source code in src/ssvep/runtime/data_use.py

def wording_sha256(block: dict) -> str:
    """SHA-256 of a wording block in canonical JSON (sorted keys, no whitespace, UTF-8).

    What a research recording is stamped with, so the words the participant was shown can be
    identified later without copying them into every file.
    """
    canonical = json.dumps(block, sort_keys=True, ensure_ascii=False, separators=(",", ":"))
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()

wording_problems

wording_problems(block)

Cross-field problems with a data_use block that JSON Schema cannot express. Empty = fine.

The schema checks each field's shape; this checks they agree: the minimum is one of the tiers offered, every offered tier has its text and no other tier does, and an on_screen protocol carries the form it shows. :func:ssvep.io.protocol.validate_protocol runs it on load and save.

Source code in src/ssvep/runtime/data_use.py

def wording_problems(block) -> list[str]:
    """Cross-field problems with a ``data_use`` block that JSON Schema cannot express. Empty = fine.

    The schema checks each field's shape; this checks they agree: the minimum is one of the tiers
    offered, every offered tier has its text and no other tier does, and an ``on_screen`` protocol
    carries the form it shows. :func:`ssvep.io.protocol.validate_protocol` runs it on load and save.
    """
    if not isinstance(block, dict):
        return ["data_use must be an object"]
    problems: list[str] = []
    offered = block.get("tiers_offered")
    if not isinstance(offered, list) or not offered or not all(
            type(t) is int and t in RESEARCH_TIERS for t in offered):
        problems.append(f"tiers_offered must list tiers from {list(RESEARCH_TIERS)}")
        offered = []
    elif len(set(offered)) != len(offered):
        problems.append("tiers_offered lists a tier twice")
    minimum = block.get("minimum_tier")
    if type(minimum) is not int or minimum not in offered:
        problems.append(f"minimum_tier {minimum!r} is not one of the tiers offered ({offered})")
    options = block.get("options")
    if isinstance(options, dict):
        want = {_tier_key(t) for t in offered}
        missing = sorted(want - set(options))
        extra = sorted(set(options) - want)
        if missing:
            problems.append(f"options has no text for offered tier(s) {', '.join(missing)}")
        if extra:
            problems.append(f"options has text for tier(s) not offered: {', '.join(extra)}")
    else:
        problems.append("options must map each offered tier to its text")
    mode = block.get("consent_mode")
    if mode not in CONSENT_MODES:
        problems.append(f"consent_mode {mode!r} is not one of {CONSENT_MODES}")
    form = block.get("consent_form")
    if mode == "on_screen" and not (isinstance(form, str) and form.strip()):
        problems.append("consent_mode 'on_screen' needs the consent_form it shows")
    if mode != "on_screen" and form is not None:
        problems.append(f"consent_form is shown only on screen, but consent_mode is {mode!r}")
    return problems

is_builtin_wording

is_builtin_wording(block)

Whether block is a built-in generic wording, unchanged: the same canonical form, and so the same :func:wording_sha256, as one in :data:KNOWN_BUILTIN_SHA256. False for anything that cannot be hashed.

Source code in src/ssvep/runtime/data_use.py

def is_builtin_wording(block) -> bool:
    """Whether ``block`` is a built-in generic wording, unchanged: the same canonical form, and so
    the same :func:`wording_sha256`, as one in :data:`KNOWN_BUILTIN_SHA256`. False for anything
    that cannot be hashed."""
    if not isinstance(block, dict):
        return False
    try:
        return wording_sha256(block) in KNOWN_BUILTIN_SHA256
    except (TypeError, ValueError):
        return False

protocol_wording_problems

protocol_wording_problems(block)

:func:wording_problems, plus the one rule only a protocol's own block is held to: it is not the built-in generic wording, unchanged (#548).

Build Protocol's editor starts a new block from the built-in wording, so an author who ticks "declares its own wording" and changes nothing would otherwise declare it as the protocol's own. The stamp would then say source: "protocol", a claim that someone approved those words for this study, and a lab machine, which refuses a research session without wording of the protocol's own, would run one under the toolbox's generic text. The built-in wording is used by leaving the block out, on a machine that allows it, and is stamped builtin there. :func:ssvep.io.protocol.validate_protocol (every load and save) and :func:research_wording (the lock) run this. The comparison is exact (:func:is_builtin_wording, the stamp's own canonical form): it stops the untouched pre-filled wording, or an exact copy of it, being declared as the protocol's own. A copy with any edit is the protocol's own wording, and the stamp's hash identifies exactly which words those were.

Source code in src/ssvep/runtime/data_use.py

def protocol_wording_problems(block) -> list[str]:
    """:func:`wording_problems`, plus the one rule only a *protocol's own* block is held to: it is
    not the built-in generic wording, unchanged (#548).

    Build Protocol's editor starts a new block from the built-in wording, so an author who ticks
    "declares its own wording" and changes nothing would otherwise declare it as the protocol's
    own. The stamp would then say ``source: "protocol"``, a claim that someone approved those words
    for this study, and a lab machine, which refuses a research session without wording of the
    protocol's own, would run one under the toolbox's generic text. The built-in wording is used
    by leaving the block out, on a machine that allows it, and is stamped ``builtin`` there.
    :func:`ssvep.io.protocol.validate_protocol` (every load and save) and :func:`research_wording`
    (the lock) run this. The comparison is exact (:func:`is_builtin_wording`, the stamp's own
    canonical form): it stops the untouched pre-filled wording, or an exact copy of it, being
    declared as the protocol's own. A copy with any edit is the protocol's own wording, and the
    stamp's hash identifies exactly which words those were.
    """
    problems = wording_problems(block)
    if is_builtin_wording(block):
        problems.append(BUILTIN_AS_OWN)
    return problems

research_wording

research_wording(block, kind)

Which wording a research session on a kind machine uses, given its protocol's block (Q5).

A protocol's own block always wins, once it passes :func:protocol_wording_problems (which refuses the built-in wording declared, unchanged, as the protocol's own). Without one, a lab machine refuses the lock (a lab's research is run under wording someone approved, and the toolbox has no business guessing it); a generic machine falls back to :data:BUILTIN_GENERIC_WORDING, stamped as built-in with its hash. A teaching machine never offers research at all.

Source code in src/ssvep/runtime/data_use.py

def research_wording(block, kind: str) -> WordingChoice:
    """Which wording a research session on a ``kind`` machine uses, given its protocol's block (Q5).

    A protocol's own block always wins, once it passes :func:`protocol_wording_problems` (which
    refuses the built-in wording declared, unchanged, as the protocol's own). Without one, a ``lab`` machine refuses the lock (a lab's
    research is run under wording someone approved, and the toolbox has no business guessing it); a
    ``generic`` machine falls back to :data:`BUILTIN_GENERIC_WORDING`, stamped as built-in with its
    hash. A ``teaching`` machine never offers research at all.
    """
    _check_kind(kind)
    if Purpose.RESEARCH.value not in PURPOSES_BY_KIND[kind]:
        return WordingChoice(None, refused=f"a {kind} deployment does not offer research sessions")
    if block is not None:
        problems = protocol_wording_problems(block)
        if problems:
            return WordingChoice(None, refused="the protocol's data_use block: " + "; ".join(problems))
        return WordingChoice(block, "protocol", wording_sha256(block))
    if kind == "lab":
        return WordingChoice(None, refused=(
            "this protocol declares no data-use wording (its data_use block), and a lab deployment "
            "runs research only under the wording its protocol carries"))
    return WordingChoice(BUILTIN_GENERIC_WORDING, "builtin", BUILTIN_GENERIC_WORDING_SHA256)

tier_choice

tier_choice(tier, wording)

Classify a research tier against the wording's offered tiers and minimum.

Source code in src/ssvep/runtime/data_use.py

def tier_choice(tier, wording: dict) -> str:
    """Classify a research tier against the wording's offered tiers and minimum."""
    offered = wording.get("tiers_offered") or []
    if isinstance(tier, bool) or type(tier) is not int or tier not in offered:
        return CHOICE_NOT_OFFERED
    if tier < wording.get("minimum_tier", 0):
        return CHOICE_DECLINED
    return CHOICE_OBSERVER if tier == 0 else CHOICE_KEEP

stamp_problems

stamp_problems(stamp)

Everything wrong with a stamp as read from disk. Empty = a stamp this toolbox can act on.

Strict in the fail-closed direction: an unknown key (a newer toolbox's stamp), a value of the wrong JSON type ("2" for 2, true for 1), a tier the purpose never carries, tier 0, a research tier below the minimum it was stamped with, or a pilot attestation that is not exactly two true booleans each make the stamp unreadable, and an unreadable stamp allows nothing.

Tier 0 and a tier below the minimum are refused for the same reason: under either, nothing is written, so a recording that carries one should not exist, and what it says cannot be trusted.

Source code in src/ssvep/runtime/data_use.py

def stamp_problems(stamp) -> list[str]:
    """Everything wrong with a stamp as read from disk. Empty = a stamp this toolbox can act on.

    Strict in the fail-closed direction: an unknown key (a newer toolbox's stamp), a value of the
    wrong JSON type (``"2"`` for 2, ``true`` for 1), a tier the purpose never carries, tier 0, a
    research tier below the minimum it was stamped with, or a pilot attestation that is not exactly
    two ``true`` booleans each make the stamp unreadable, and an unreadable stamp allows nothing.

    Tier 0 and a tier below the minimum are refused for the same reason: under either, nothing is
    written, so a recording that carries one should not exist, and what it says cannot be trusted.
    """
    if not isinstance(stamp, dict):
        return [f"{STAMP_KEY} is not an object"]
    problems: list[str] = []
    unknown = sorted(set(stamp) - _STAMP_KEYS)
    if unknown:
        problems.append(f"unknown key(s) {', '.join(unknown)} (written by a newer toolbox?)")
    purpose = stamp.get("purpose")
    if purpose not in PURPOSES:
        return problems + [f"purpose {purpose!r} is not one of {PURPOSES}"]
    tier = stamp.get("tier")
    if not _is_tier_of(purpose, tier):
        problems.append(f"tier {tier!r} is not one a {purpose} session carries "
                        f"({list(TIERS_BY_PURPOSE[purpose])})")
    elif tier == 0 and type(tier) is int:
        problems.append("tier 0 means nothing is written: a recording stamped 0 should not exist")
    entered_by = stamp.get("entered_by")
    if entered_by not in ENTERED_BY:
        problems.append(f"entered_by {entered_by!r} is not one of {ENTERED_BY}")
    elif purpose in ("pilot", "bench") and entered_by != "operator":
        problems.append(f"a {purpose} session's tier is the operator's, not the participant's")

    if purpose == "research":
        if stamp.get("mode") not in CONSENT_MODES:
            problems.append(f"mode {stamp.get('mode')!r} is not one of {CONSENT_MODES}")
        wording = stamp.get("wording")
        if not isinstance(wording, dict) or set(wording) != {"source", "sha256"}:
            problems.append("wording must be exactly {source, sha256}")
        else:
            sha = wording.get("sha256")
            if wording.get("source") not in WORDING_SOURCES:
                problems.append(f"wording.source {wording.get('source')!r} is not one of {WORDING_SOURCES}")
            if not (isinstance(sha, str) and len(sha) == 64
                    and all(c in "0123456789abcdef" for c in sha)):
                problems.append("wording.sha256 is not a SHA-256 hex digest")
            elif wording.get("source") == "builtin" and sha not in KNOWN_BUILTIN_SHA256:
                problems.append("wording.sha256 names no built-in wording this toolbox knows")
        offered = stamp.get("offered")
        if not (isinstance(offered, list) and offered and all(_is_tier_int(t) for t in offered)
                and len(set(offered)) == len(offered)):
            problems.append("offered must list the tiers offered")
            offered = None
        minimum = stamp.get("minimum")
        if not _is_tier_int(minimum) or (offered is not None and minimum not in offered):
            problems.append(f"minimum {minimum!r} is not one of the tiers offered")
        elif _is_tier_int(tier) and offered is not None:
            if tier not in offered:
                problems.append(f"tier {tier} was not offered ({offered})")
            elif tier < minimum and tier != 0:      # tier 0 is refused above
                problems.append(f"tier {tier} is below the minimum {minimum}: nothing should exist")
    else:
        for key in _RESEARCH_ONLY:
            if key in stamp:
                problems.append(f"{key} describes research consent, not a {purpose} session")

    attestation = stamp.get("attestation")
    if purpose == "pilot" and tier == TIER_PILOT_ATTESTED:
        if not (isinstance(attestation, dict) and set(attestation) == set(ATTESTATION_KEYS)
                and all(attestation[k] is True for k in ATTESTATION_KEYS)):
            problems.append(f"a pilot session needs its attestation: exactly "
                            f"{{{', '.join(ATTESTATION_KEYS)}}}, each true")
    elif attestation is not None:
        problems.append("only a pilot session carries an attestation")
    return problems

data_use_of

data_use_of(sidecar, active_kind)

The purpose and tier of the recording this sidecar describes, or why it has none. Fail-closed.

active_kind is the kind of the machine doing the reading (the site profile in force).

• A data_use stamp is read strictly (:func:stamp_problems); a malformed one is refused and never inferred around. Its purpose must be one the recording's own kind offers, when the recording says which kind it was made under.
• No stamp: refused (D11), unless the recording counts as teaching (#286 Q1). On a teaching machine (active_kind teaching) it does when its own site_profile.kind is teaching, generic or absent: generic is what :func:ssvep.runtime.site.recording_stamp writes when the profile does not resolve, which is how a class recording made with the profile unset reads, and a copy made before the kind was stamped names none. A recording whose own kind is lab is research data under its own rules and stays refused, wherever it is read. Off a teaching machine, only a recording whose own kind is teaching counts.
• A legacy observer-only flag refuses either way, and a legacy pilot: true blocks the teaching inference and contradicts any non-pilot stamp.

Source code in src/ssvep/runtime/data_use.py

def data_use_of(sidecar, active_kind: str) -> Reading:
    """The purpose and tier of the recording this sidecar describes, or why it has none. Fail-closed.

    ``active_kind`` is the kind of the machine doing the reading (the site profile in force).

    * A ``data_use`` stamp is read strictly (:func:`stamp_problems`); a malformed one is refused and
      never inferred around. Its purpose must be one the recording's own kind offers, when the
      recording says which kind it was made under.
    * No stamp: refused (D11), unless the recording counts as teaching (#286 Q1). On a teaching
      machine (``active_kind`` ``teaching``) it does when its own ``site_profile.kind`` is
      ``teaching``, ``generic`` or absent: ``generic`` is what :func:`ssvep.runtime.site.recording_stamp`
      writes when the profile does not resolve, which is how a class recording made with the profile
      unset reads, and a copy made before the kind was stamped names none. A recording whose own
      kind is ``lab`` is research data under its own rules and stays refused, wherever it is read.
      Off a teaching machine, only a recording whose own kind is ``teaching`` counts.
    * A legacy observer-only flag refuses either way, and a legacy ``pilot: true`` blocks the
      teaching inference and contradicts any non-pilot stamp.
    """
    _check_kind(active_kind)
    if not isinstance(sidecar, dict):
        return Reading(None, "the sidecar is not readable as a JSON object")
    veto = _legacy_veto(sidecar)
    if veto:
        return Reading(None, veto)
    own = _own_kind(sidecar)
    if own is False:
        return Reading(None, "its site_profile.kind is not a kind this toolbox knows")
    stamp = sidecar.get(STAMP_KEY)
    if stamp is not None:
        problems = stamp_problems(stamp)
        if problems:
            return Reading(None, "its data-use stamp cannot be read: " + "; ".join(problems))
        if own is not None and stamp["purpose"] not in PURPOSES_BY_KIND[own]:
            return Reading(None, f"it is stamped {stamp['purpose']!r}, which a {own} deployment "
                                 f"does not offer")
        if _legacy_pilot(sidecar) and stamp["purpose"] != "pilot":
            return Reading(None, f"it is stamped {stamp['purpose']!r} but its session position "
                                 f"says pilot")
        record = DataUse(
            purpose=stamp["purpose"], tier=stamp["tier"], entered_by=stamp["entered_by"],
            mode=stamp.get("mode"), wording=stamp.get("wording"),
            offered=tuple(stamp["offered"]) if "offered" in stamp else None,
            minimum=stamp.get("minimum"), attestation=stamp.get("attestation"))
        return Reading(record)
    if STAMP_KEY in sidecar:
        return Reading(None, f"its {STAMP_KEY} stamp is null")
    teaching = own == "teaching" or (active_kind == "teaching" and own in (None, "generic"))
    if teaching and not _legacy_pilot(sidecar):
        return Reading(DataUse(purpose="teaching", tier=TIER_COURSE_USE_ONLY, inferred=True))
    return Reading(None, "it carries no data-use stamp (recorded before purpose and tier were "
                         "stamped); there is no migration, so it is held back")

research_needs_screens

research_needs_screens(tier_entry, consent_mode)

Why operator entry is not honest for this research session, or None when it is. One rule for both reasons: the site has the participant choose (tier_entry participant, or anything unreadable: :data:PARTICIPANT_ENTRY, which the participant screens answer), or the protocol shows its consent form on screen under operator entry (:data:ON_SCREEN_NEEDS_PARTICIPANT_ENTRY, which is refused: no screens run for it).

Source code in src/ssvep/runtime/data_use.py

def research_needs_screens(tier_entry, consent_mode) -> str | None:
    """Why operator entry is not honest for this research session, or ``None`` when it is. One rule
    for both reasons: the site has the participant choose (``tier_entry`` ``participant``, or
    anything unreadable: :data:`PARTICIPANT_ENTRY`, which the participant screens answer), or the
    protocol shows its consent form on screen under operator entry
    (:data:`ON_SCREEN_NEEDS_PARTICIPANT_ENTRY`, which is refused: no screens run for it)."""
    if tier_entry != "operator":
        return PARTICIPANT_ENTRY
    if consent_mode == "on_screen":
        return ON_SCREEN_NEEDS_PARTICIPANT_ENTRY
    return None

lock_decision

lock_decision(kind, purpose, *, protocol_block=None, tier=None, attestation=None, tier_entry=DEFAULT_TIER_ENTRY, participant_tier=None)

Turn the operator's choices at lock into a :class:LockDecision. Pure; fail-closed.

kind is the active site profile's kind, purpose the one chosen, protocol_block the loaded protocol's data_use block (or None), tier the participant's stated tier as the operator entered it (research: an int; teaching: :data:TIER_COURSE_USE_ONLY or 0), and attestation a pilot session's {"team_member": bool, "oral_consent": bool}, and tier_entry the site profile's data_use.tier_entry (default participant, fail-closed), and participant_tier the option the participant chose on the participant screens (#286 P4), or None when they have not run.

• A purpose this kind does not offer is refused (D11).
• Research: the wording is the protocol's block, else built-in on a generic machine, else the lock is refused (#286 Q5). Then :func:research_needs_screens: on a site whose tier_entry is participant (:data:PARTICIPANT_ENTRY) the option is the participant's own from the screens: without participant_tier the decision is refused with screens_pending set (the operator's tier is ignored, never stamped as the participant's), and with it the stamp says entered_by: "participant". Under operator entry a protocol whose consent_mode is on_screen is refused outright (:data:ON_SCREEN_NEEDS_PARTICIPANT_ENTRY); otherwise the operator's tier is used, stamped entered_by: "operator" (#286 Q4), and a participant_tier is refused (nobody asked). The tier must be one the wording offers. Below the minimum the participant did not consent: refused, and nothing runs (D7). Tier 0 at a minimum of 0 (the built-in wording's) is observer only: the session runs and nothing is saved.
• Teaching: course use only, or 0 (observer only, #286 Q6); tier_entry does not apply. The lock still shows the student the safety screen (and only that) before it freezes this decision (Aaron, #560); nothing about it is stamped.
• Pilot: both attestation booleans must be true (D5), or refused. No name is ever taken.
• Bench: nothing to ask (#286 Q3).
• Pilot and bench take no tier; one passed in is refused (a confused caller).

Source code in src/ssvep/runtime/data_use.py

def lock_decision(kind: str, purpose, *, protocol_block=None, tier=None,
                  attestation=None, tier_entry: str = DEFAULT_TIER_ENTRY,
                  participant_tier=None) -> LockDecision:
    """Turn the operator's choices at lock into a :class:`LockDecision`. Pure; fail-closed.

    ``kind`` is the active site profile's kind, ``purpose`` the one chosen, ``protocol_block`` the
    loaded protocol's ``data_use`` block (or ``None``), ``tier`` the participant's stated tier as
    the operator entered it (research: an int; teaching: :data:`TIER_COURSE_USE_ONLY` or 0), and
    ``attestation`` a pilot session's ``{"team_member": bool, "oral_consent": bool}``, and
    ``tier_entry`` the site profile's ``data_use.tier_entry`` (default ``participant``, fail-closed),
    and ``participant_tier`` the option the participant chose on the participant screens (#286 P4),
    or ``None`` when they have not run.

    * A purpose this kind does not offer is refused (D11).
    * Research: the wording is the protocol's block, else built-in on a generic machine, else the
      lock is refused (#286 Q5). Then :func:`research_needs_screens`: on a site whose ``tier_entry``
      is ``participant`` (:data:`PARTICIPANT_ENTRY`) the option is the participant's own from the
      screens: without ``participant_tier`` the decision is refused with ``screens_pending`` set
      (the operator's ``tier`` is ignored, never stamped as the participant's), and with it the
      stamp says ``entered_by: "participant"``. Under operator entry a protocol whose
      ``consent_mode`` is ``on_screen`` is refused outright
      (:data:`ON_SCREEN_NEEDS_PARTICIPANT_ENTRY`); otherwise the operator's ``tier`` is used,
      stamped ``entered_by: "operator"`` (#286 Q4), and a ``participant_tier`` is refused (nobody
      asked).
      The tier must be one the wording offers. Below the minimum the participant did not consent:
      refused, and nothing runs (D7). Tier 0 at a minimum of 0 (the built-in wording's) is observer
      only: the session runs and nothing is saved.
    * Teaching: course use only, or 0 (observer only, #286 Q6); ``tier_entry`` does not apply. The
      lock still shows the student the safety screen (and only that) before it freezes this
      decision (Aaron, #560); nothing about it is stamped.
    * Pilot: both attestation booleans must be true (D5), or refused. No name is ever taken.
    * Bench: nothing to ask (#286 Q3).
    * Pilot and bench take no tier; one passed in is refused (a confused caller).
    """
    _check_kind(kind)
    if purpose not in PURPOSES:
        return LockDecision(None, refused="choose this session's purpose")
    if purpose not in PURPOSES_BY_KIND[kind]:
        return LockDecision(purpose, refused=f"a {kind} deployment does not offer {purpose} sessions")
    if purpose == Purpose.RESEARCH.value:
        wc = research_wording(protocol_block, kind)
        if wc.wording is None:
            return LockDecision(purpose, refused=wc.refused)
        # The participant's own answer, from the participant screens (#286 P4), wherever the site or
        # the protocol needs it; an operator-entered option is never stamped as theirs.
        needs = research_needs_screens(tier_entry, wc.wording.get("consent_mode"))
        if needs == ON_SCREEN_NEEDS_PARTICIPANT_ENTRY:
            return LockDecision(purpose, refused=needs)
        if needs is not None:
            if participant_tier is None:
                return LockDecision(purpose, refused=needs, screens_pending=True)
            tier, entered_by = participant_tier, "participant"
        elif participant_tier is not None:
            return LockDecision(purpose, refused=PARTICIPANT_ANSWER_NOT_ASKED)
        else:
            entered_by = "operator"
        if tier is None:
            return LockDecision(purpose, refused="enter the option the participant chose")
        choice = tier_choice(tier, wc.wording)
        if choice == CHOICE_NOT_OFFERED:
            return LockDecision(purpose, refused=f"option {tier!r} is not one this protocol offers "
                                                 f"({wc.wording.get('tiers_offered')})")
        if choice == CHOICE_DECLINED:
            return LockDecision(purpose, refused=(
                f"option {tier} is below this protocol's minimum ({wc.wording.get('minimum_tier')}): "
                f"the participant has not consented to take part, so the session does not run and "
                f"nothing is saved"))
        if choice == CHOICE_OBSERVER:
            return LockDecision(purpose, observer=True)
        return LockDecision(purpose, DataUse(
            purpose=purpose, tier=tier, entered_by=entered_by, mode=wc.wording["consent_mode"],
            wording={"source": wc.source, "sha256": wc.sha256},
            offered=tuple(wc.wording["tiers_offered"]), minimum=wc.wording["minimum_tier"]))
    if participant_tier is not None:
        return LockDecision(purpose, refused=f"a {purpose} session asks no participant screens' "
                                             f"option (got {participant_tier!r})")
    if purpose == Purpose.TEACHING.value:
        if isinstance(tier, bool) or tier not in (TIER_COURSE_USE_ONLY, 0):
            return LockDecision(purpose, refused="enter the option the student chose")
        if tier == 0:
            return LockDecision(purpose, observer=True)
        return LockDecision(purpose, DataUse(purpose=purpose, tier=TIER_COURSE_USE_ONLY))
    if tier is not None:
        return LockDecision(purpose, refused=f"a {purpose} session has no participant option to "
                                             f"enter (got {tier!r})")
    if purpose == Purpose.PILOT.value:
        att = attestation if isinstance(attestation, dict) else {}
        if not all(att.get(k) is True for k in ATTESTATION_KEYS):
            return LockDecision(purpose, refused=(
                "a pilot session needs both attestations: the volunteer is a team member named in "
                "the protocol, and gave oral consent"))
        return LockDecision(purpose, DataUse(purpose=purpose, tier=TIER_PILOT_ATTESTED,
                                             attestation={k: True for k in ATTESTATION_KEYS}))
    return LockDecision(purpose, DataUse(purpose=purpose, tier=TIER_NO_PARTICIPANT))

lower_targets

lower_targets(decision)

The options a locked session may be lowered to, highest first, 0 always last. Pure.

Lowering is downward only and never raises an option on anyone's behalf. A research session may go to any offered option below its current one that is still at or above the minimum it was stamped with (an option below the minimum means "did not consent", and that is option 0: delete what was saved). Every lowerable purpose may go to 0 (:data:LOWERABLE_PURPOSES). An observer-only session, a refused one, a bench one or no decision at all has nothing to lower to; whether an observer-only session still has saved files to delete is a question about the disk, not about this policy.

Source code in src/ssvep/runtime/data_use.py

def lower_targets(decision: "LockDecision | None") -> tuple:
    """The options a locked session may be lowered to, highest first, ``0`` always last. Pure.

    Lowering is downward only and never raises an option on anyone's behalf. A research session may
    go to any offered option below its current one that is still at or above the minimum it was
    stamped with (an option below the minimum means "did not consent", and that is option 0: delete
    what was saved). Every lowerable purpose may go to 0 (:data:`LOWERABLE_PURPOSES`). An
    observer-only session, a refused one, a bench one or no decision at all has nothing to lower to;
    whether an observer-only session still has saved files to delete is a question about the disk,
    not about this policy.
    """
    if decision is None or decision.refused is not None or decision.observer:
        return ()
    rec = decision.record
    if rec is None or rec.purpose not in LOWERABLE_PURPOSES:
        return ()
    if rec.purpose == Purpose.RESEARCH.value and type(rec.tier) is int:
        lo = max(rec.minimum or 0, 1)
        offered = rec.offered or ()
        return tuple(sorted((t for t in offered if lo <= t < rec.tier), reverse=True)) + (0,)
    return (0,)

lowered

lowered(decision, tier)

The lock decision after lowering decision to tier. Raises ValueError otherwise.

tier must be one of :func:lower_targets. 0 gives an observer-only decision: from now on nothing more is written (the caller deletes what was already saved). Any other tier keeps every other field of the stamp (wording, offered, minimum, entered_by) and changes the tier alone, so a lowered stamp is exactly the stamp the lock would have made for that option.

Source code in src/ssvep/runtime/data_use.py

def lowered(decision: "LockDecision", tier) -> "LockDecision":
    """The lock decision after lowering ``decision`` to ``tier``. Raises ``ValueError`` otherwise.

    ``tier`` must be one of :func:`lower_targets`. ``0`` gives an observer-only decision: from now
    on nothing more is written (the caller deletes what was already saved). Any other tier keeps
    every other field of the stamp (wording, offered, minimum, ``entered_by``) and changes the tier
    alone, so a lowered stamp is exactly the stamp the lock would have made for that option.
    """
    if isinstance(tier, bool) or tier not in lower_targets(decision):
        raise ValueError(f"cannot lower this session to {tier!r}: the options it may be lowered to "
                         f"are {list(lower_targets(decision))}")
    if tier == 0:
        return LockDecision(decision.purpose, observer=True)
    return LockDecision(decision.purpose, replace(decision.record, tier=tier))

lowered_stamp

lowered_stamp(stamp, tier)

stamp (as read from a file) lowered to research option tier, or None when it is already at or below it. Pure; never raises an option.

Raises ValueError when the stamp cannot be lowered to tier: it is unreadable, it is not a research stamp, or tier is not an option it can carry (not offered, below its minimum, or 0, which is deletion and never a stamp).

Source code in src/ssvep/runtime/data_use.py

def lowered_stamp(stamp, tier: int) -> dict | None:
    """``stamp`` (as read from a file) lowered to research option ``tier``, or ``None`` when it is
    already at or below it. Pure; never raises an option.

    Raises ``ValueError`` when the stamp cannot be lowered to ``tier``: it is unreadable, it is not
    a research stamp, or ``tier`` is not an option it can carry (not offered, below its minimum, or
    0, which is deletion and never a stamp).
    """
    problems = stamp_problems(stamp)
    if problems:
        raise ValueError("its data-use stamp cannot be read: " + "; ".join(problems))
    if stamp["purpose"] != Purpose.RESEARCH.value:
        raise ValueError(f"it is stamped {stamp['purpose']!r}, not research")
    if stamp["tier"] <= tier:
        return None
    new = dict(stamp)
    new["tier"] = tier
    problems = stamp_problems(new)
    if problems:
        raise ValueError(f"option {tier!r} is not one it can carry: " + "; ".join(problems))
    return new

may

may(action, record, *, active_kind=None)

Whether action may happen to the recording record describes. None = never.

POOL also needs active_kind: a contrast on that machine pools only :data:POOL_PURPOSE_BY_KIND's purpose.

Source code in src/ssvep/runtime/data_use.py

def may(action, record: DataUse | None, *, active_kind: str | None = None) -> bool:
    """Whether ``action`` may happen to the recording ``record`` describes. ``None`` = never.

    POOL also needs ``active_kind``: a contrast on that machine pools only
    :data:`POOL_PURPOSE_BY_KIND`'s purpose.
    """
    return refusal(action, Reading(record) if record is not None else None,
                   active_kind=active_kind) is None

refusal

refusal(action, reading, *, active_kind=None)

Why action may not happen to this recording, or None when it may. Fail-closed.

Source code in src/ssvep/runtime/data_use.py

def refusal(action, reading: Reading | None, *, active_kind: str | None = None) -> str | None:
    """Why ``action`` may not happen to this recording, or ``None`` when it may. Fail-closed."""
    action = Action(action)
    if action is Action.POOL:
        if active_kind is None:
            raise ValueError("POOL needs active_kind: which purpose pools depends on the machine")
        _check_kind(active_kind)
    if reading is None or reading.record is None:
        return (reading.refused if reading is not None and reading.refused
                else "no data-use record")
    rec = reading.record
    if action not in rec.allowed():
        return f"a {rec.purpose} recording at tier {rec.tier!r} may not {action.value.replace('_', ' ')}"
    if action is Action.POOL:
        want = POOL_PURPOSE_BY_KIND[active_kind]
        if rec.purpose != want:
            return f"a contrast on a {active_kind} machine pools only {want} recordings"
    return None

pool_purpose

pool_purpose(active_kind, *, synthetic=False)

The one purpose a contrast pools (#286 Q7): :data:POOL_PURPOSE_BY_KIND's for this machine, or :data:SYNTHETIC_POOL_PURPOSE on a synthetic dataset.

Source code in src/ssvep/runtime/data_use.py

def pool_purpose(active_kind: str, *, synthetic: bool = False) -> str:
    """The one purpose a contrast pools (#286 Q7): :data:`POOL_PURPOSE_BY_KIND`'s for this machine,
    or :data:`SYNTHETIC_POOL_PURPOSE` on a synthetic dataset."""
    _check_kind(active_kind)
    return SYNTHETIC_POOL_PURPOSE if synthetic else POOL_PURPOSE_BY_KIND[active_kind]

pool_refusal

pool_refusal(reading, *, active_kind, synthetic=False)

Why this recording may not be pooled into a contrast, or None when it may. Fail-closed.

A contrast pools exactly one purpose (:func:pool_purpose). On a real dataset that is the matrix's POOL action (:func:refusal). On a synthetic dataset it is a readable bench stamp: the matrix never pools bench, because a real bench recording (a phantom, a photodiode) beside participants' recordings is not one of them, but a synthetic dataset holds nothing else.

Source code in src/ssvep/runtime/data_use.py

def pool_refusal(reading: Reading | None, *, active_kind: str, synthetic: bool = False) -> str | None:
    """Why this recording may not be pooled into a contrast, or ``None`` when it may. Fail-closed.

    A contrast pools exactly one purpose (:func:`pool_purpose`). On a real dataset that is the
    matrix's POOL action (:func:`refusal`). On a synthetic dataset it is a readable bench stamp: the
    matrix never pools bench, because a real bench recording (a phantom, a photodiode) beside
    participants' recordings is not one of them, but a synthetic dataset holds nothing else.
    """
    if not synthetic:
        return refusal(Action.POOL, reading, active_kind=active_kind)
    _check_kind(active_kind)
    if reading is None or reading.record is None:
        return (reading.refused if reading is not None and reading.refused
                else "no data-use record")
    rec = reading.record
    if rec.purpose != SYNTHETIC_POOL_PURPOSE or Action.PERSIST not in rec.allowed():
        return (f"a contrast on a synthetic dataset pools only {SYNTHETIC_POOL_PURPOSE} recordings "
                f"(nobody was recorded), and this one is {rec.purpose}")
    return None

Session lifecycle: lowering and deletion

ssvep.runtime.lifecycle

The session lifecycle on disk: lowering a session's option, and deleting it (#286 P3c, D7).

A participant may lower their data-use option before they leave, never raise it, and never after (:mod:ssvep.runtime.data_use's :func:~ssvep.runtime.data_use.lower_targets is the policy). This module is what that decision does to the files the session has already written:

• Lowering a research option (say 3 to 1) rewrites the tier in every stamp this sub/ses carries: each run sidecar (a rescued one included), each questionnaire answer and each impedance-log entry. Nothing else in a file changes, every file is checked before any is written, and the new files replace the old ones only once all of them are staged (:func:lower_session). The tier lives only in these stamps (the BIDS view, the derivatives and every gate read it back from them), so nothing derived needs rewriting.
• Switching to option 0 (observer only) deletes the whole sub/ses output: the sourcedata session folder (recordings, sidecars, answers, impedance log, rescued files, declarations, stimulus copies), its BIDS view, its per-session derivatives, its row in the participant's sessions table, every contrast report that pooled it (each report lists whom it pooled), any other derived page that names it, and its lines in any log or table (the rest of which, other participants' records, is kept). An older report nothing can rule out is named for a check by hand. A participant with nothing left anywhere loses their participants.tsv row too (:func:delete_session).

Both check the subject and session labels first (:func:checked_session), and refuse, before touching anything, to work through a symlink or junction: following one could rewrite or delete files that belong to someone else (the same rule as push, #341). Neither reaches copies outside this data root: a file already pushed to lab storage, or an export made earlier. That is why a session whose option can still be lowered pushes nothing until it ends (:data:ssvep.runtime.data_use.PUSH_AT_SESSION_END).

LifecycleError

Bases: Exception

Lowering or deleting was refused, and nothing was changed (the message says why).

Source code in src/ssvep/runtime/lifecycle.py

class LifecycleError(Exception):
    """Lowering or deleting was refused, and nothing was changed (the message says why)."""

LowerResult dataclass

What :func:lower_session did: the files rewritten, and the ones it left as they were.

Source code in src/ssvep/runtime/lifecycle.py

@dataclass
class LowerResult:
    """What :func:`lower_session` did: the files rewritten, and the ones it left as they were."""
    rewritten: list[Path] = field(default_factory=list)
    unchanged: list[tuple[Path, str]] = field(default_factory=list)   # (path, why it was left)
    cleaned: list[Path] = field(default_factory=list)   # staging files an interrupted run left

DeleteResult dataclass

What :func:delete_session removed, and anything worth telling the operator.

Source code in src/ssvep/runtime/lifecycle.py

@dataclass
class DeleteResult:
    """What :func:`delete_session` removed, and anything worth telling the operator."""
    deleted: list[Path] = field(default_factory=list)
    notes: list[str] = field(default_factory=list)

session_dir

session_dir(out_dir, subject, session)

<out_dir>/sourcedata/sub-XXX/ses-YYY: where a session's own files live.

Source code in src/ssvep/runtime/lifecycle.py

def session_dir(out_dir, subject: str, session: str) -> Path:
    """``<out_dir>/sourcedata/sub-XXX/ses-YYY``: where a session's own files live."""
    return Path(out_dir) / "sourcedata" / subject / session

is_within

is_within(path, folder)

Whether path is folder or inside it, compared as written (normalised, never resolved, case-insensitively where the platform is).

Source code in src/ssvep/runtime/lifecycle.py

def is_within(path, folder) -> bool:
    """Whether ``path`` is ``folder`` or inside it, compared as written (normalised, never resolved,
    case-insensitively where the platform is)."""
    def norm(p):
        return os.path.normcase(os.path.normpath(os.path.abspath(str(p))))
    p, d = norm(path), norm(folder)
    return p == d or p.startswith(d.rstrip(os.sep) + os.sep)

stamped_files

stamped_files(sdir)

Every JSON file in a session's eeg/ and beh/ folders: where stamps are written.

Source code in src/ssvep/runtime/lifecycle.py

def stamped_files(sdir: Path) -> list[Path]:
    """Every JSON file in a session's ``eeg/`` and ``beh/`` folders: where stamps are written."""
    out: list[Path] = []
    for mod in ("eeg", "beh"):
        d = Path(sdir) / mod
        if d.is_dir():
            out += sorted(p for p in d.iterdir() if p.suffix == ".json" and p.is_file()
                          and ".part-" not in p.name)
    return out

lower_session

lower_session(out_dir, subject, session, tier)

Lower every research stamp of this sub/ses to option tier (1 or above).

All or nothing, checked first. Every stamp-bearing file is read and its new content worked out before any is written; a stamp that cannot be lowered to tier (unreadable, another purpose, or tier not an option it can carry) refuses the whole change with :class:LifecycleError, naming the file, and leaves every file as it was. A stamp already at or below tier stays as it is (lowering never raises one). A file with no stamp at all is left alone and listed: it is already held back by every gate, which is stricter than any option.

The new files are staged beside the old ones and then swapped in. If a swap itself fails, the files already swapped carry the lower option and the error names the rest; running the same lowering again finishes it.

Source code in src/ssvep/runtime/lifecycle.py

def lower_session(out_dir, subject: str, session: str, tier: int) -> LowerResult:
    """Lower every research stamp of this ``sub``/``ses`` to option ``tier`` (1 or above).

    All or nothing, checked first. Every stamp-bearing file is read and its new content worked out
    before any is written; a stamp that cannot be lowered to ``tier`` (unreadable, another purpose,
    or ``tier`` not an option it can carry) refuses the whole change with :class:`LifecycleError`,
    naming the file, and leaves every file as it was. A stamp already at or below ``tier`` stays as
    it is (lowering never raises one). A file with no stamp at all is left alone and listed: it is
    already held back by every gate, which is stricter than any option.

    The new files are staged beside the old ones and then swapped in. If a swap itself fails, the
    files already swapped carry the lower option and the error names the rest; running the same
    lowering again finishes it.
    """
    if isinstance(tier, bool) or type(tier) is not int or tier < 1:
        raise ValueError(f"lower_session lowers to a research option of 1 or above, not {tier!r}; "
                         f"option 0 is delete_session")
    root, subject, session, sdir = checked_session(out_dir, subject, session)
    result = LowerResult()
    if not sdir.exists():
        return result
    for mod in ("eeg", "beh"):
        _check_no_link_down_to(root, sdir / mod)
    # A lowering interrupted by a crash can leave its staging files; no gate reads them, but they
    # are not the session's, so the next lowering clears them before it stages its own.
    for mod in ("eeg", "beh"):
        d = sdir / mod
        if d.is_dir():
            for stale in sorted(d.glob("*.lowering")):
                if stale.is_file() and _link_reason(stale) is None:
                    stale.unlink()
                    result.cleaned.append(stale)
    staged: list[tuple[Path, str]] = []
    for path in stamped_files(sdir):
        reason = _link_reason(path)
        if reason is not None:
            raise LifecycleError(f"{path.name} is {reason}; nothing was changed")
        try:
            text = path.read_text(encoding="utf-8")
            doc = json.loads(text)
        except (OSError, UnicodeDecodeError, ValueError) as exc:
            raise LifecycleError(f"{path.name} cannot be read ({exc}); nothing was changed") from exc
        holders = _stamp_holders(doc)
        if holders is None:
            result.unchanged.append((path, "it carries no data-use stamp, so every gate already "
                                           "holds it back"))
            continue
        changed = False
        for h in holders:
            try:
                new = DU.lowered_stamp(h[DU.STAMP_KEY], tier)
            except ValueError as exc:
                raise LifecycleError(f"{path.name}: {exc}; nothing was changed") from exc
            if new is not None:
                h[DU.STAMP_KEY] = new
                changed = True
        if not changed:
            result.unchanged.append((path, f"already at option {tier} or below"))
            continue
        staged.append((path, json.dumps(doc, indent=2) + ("\n" if text.endswith("\n") else "")))
    temps: list[tuple[Path, Path]] = []
    try:
        for path, content in staged:
            tmp = path.with_name(path.name + ".lowering")
            tmp.write_text(content, encoding="utf-8")
            temps.append((tmp, path))
    except OSError as exc:
        for tmp, _ in temps:
            tmp.unlink(missing_ok=True)
        raise LifecycleError(f"could not stage the lowered files ({exc}); nothing was changed") from exc
    for i, (tmp, path) in enumerate(temps):
        try:
            os.replace(tmp, path)
        except OSError as exc:
            for left, _ in temps[i:]:
                left.unlink(missing_ok=True)
            done = ", ".join(p.name for p in result.rewritten) or "none"
            rest = ", ".join(p.name for _, p in temps[i:])
            raise LifecycleError(f"could not replace {path.name} ({exc}). Lowered already: {done}. "
                                 f"Still at the old option: {rest}. Lower again to finish.") from exc
        result.rewritten.append(path)
    return result

checked_session

checked_session(out_dir, subject, session)

(root, subject, session, session folder) once the labels are ones the recorder itself would write, or :class:LifecycleError. Every entry point here goes through this first: an empty or .. label would otherwise turn "this session" into the participant's every session, or the whole data root.

Source code in src/ssvep/runtime/lifecycle.py

def checked_session(out_dir, subject, session) -> tuple[Path, str, str, Path]:
    """``(root, subject, session, session folder)`` once the labels are ones the recorder itself
    would write, or :class:`LifecycleError`. Every entry point here goes through this first: an empty
    or ``..`` label would otherwise turn "this session" into the participant's every session, or the
    whole data root.
    """
    from . import dates
    if not isinstance(subject, str) or not isinstance(session, str):
        raise LifecycleError(f"not a subject and session label: {subject!r}, {session!r}")
    sub, ses = subject.strip(), session.strip()
    if not _SUBJECT_SHAPE.match(sub):
        raise LifecycleError(f"the subject code must be 'sub-' followed by letters and digits only "
                             f"(got {subject!r}); nothing was changed")
    try:
        dates.check_subject_label(sub)
        dates.check_session_label(ses)
    except ValueError as exc:
        raise LifecycleError(f"{exc}; nothing was changed") from exc
    root = Path(out_dir)
    sdir = session_dir(root, sub, ses)
    src = root / "sourcedata"
    if not is_within(sdir, src / sub) or sdir.parent.parent != src or sdir.parent == src:
        raise LifecycleError(f"{sdir} is not a session folder inside {src}; nothing was changed")
    return root, sub, ses, sdir

session_protocols

session_protocols(sdir)

The protocols a sourcedata session's run sidecars name (session_position.protocol): the only ones a contrast can have pooled it into. None when that cannot be known (no run sidecar, or one that cannot be read), so the caller cannot rule out any older report.

Source code in src/ssvep/runtime/lifecycle.py

def session_protocols(sdir: Path) -> set[str] | None:
    """The protocols a ``sourcedata`` session's run sidecars name (``session_position.protocol``):
    the only ones a contrast can have pooled it into. ``None`` when that cannot be known (no run
    sidecar, or one that cannot be read), so the caller cannot rule out any older report."""
    eeg = Path(sdir) / "eeg"
    sidecars = sorted(eeg.glob("*.session.json")) if eeg.is_dir() else []
    if not sidecars:
        return None
    out: set[str] = set()
    for s in sidecars:
        try:
            doc = json.loads(s.read_text(encoding="utf-8-sig"))
        except (OSError, UnicodeDecodeError, ValueError):
            return None
        pos = doc.get("session_position") if isinstance(doc, dict) else None
        if isinstance(pos, dict) and pos.get("protocol"):
            out.add(str(pos["protocol"]))
    return out

lowest_research_tier

lowest_research_tier(out_dir, subject, session)

The lowest research option any file of this sub/ses already carries, or None. A lock reads it so that a re-lock never stamps later files above earlier, lowered ones.

Source code in src/ssvep/runtime/lifecycle.py

def lowest_research_tier(out_dir, subject: str, session: str) -> int | None:
    """The lowest research option any file of this ``sub``/``ses`` already carries, or ``None``.
    A lock reads it so that a re-lock never stamps later files above earlier, lowered ones."""
    try:
        _, _, _, sdir = checked_session(out_dir, subject, session)
    except LifecycleError:
        return None
    low = None
    for path in stamped_files(sdir):
        try:
            holders = _stamp_holders(json.loads(path.read_text(encoding="utf-8-sig"))) or []
        except (OSError, UnicodeDecodeError, ValueError):
            continue
        for h in holders:
            st = h.get(DU.STAMP_KEY)
            if (isinstance(st, dict) and st.get("purpose") == DU.Purpose.RESEARCH.value
                    and type(st.get("tier")) is int):
                low = st["tier"] if low is None else min(low, st["tier"])
    return low

deletion_targets

deletion_targets(out_dir, subject, session)

The folders :func:delete_session removes whole, those that exist: the sourcedata session, its BIDS view, and its folder in each derivatives pipeline. What a confirmation names. Refuses labels that are not a session's (:func:checked_session).

Source code in src/ssvep/runtime/lifecycle.py

def deletion_targets(out_dir, subject: str, session: str) -> list[Path]:
    """The folders :func:`delete_session` removes whole, those that exist: the ``sourcedata``
    session, its BIDS view, and its folder in each derivatives pipeline. What a confirmation names.
    Refuses labels that are not a session's (:func:`checked_session`)."""
    root, subject, session, sdir = checked_session(out_dir, subject, session)
    out = [sdir, root / subject / session]
    deriv = root / "derivatives"
    if deriv.is_dir():
        out += sorted(p / subject / session for p in deriv.iterdir() if p.is_dir())
    return [p for p in out if p.exists()]

delete_session

delete_session(out_dir, subject, session)

Delete everything this data root holds of one sub/ses (option 0 after saving).

Removes :func:deletion_targets; the session's row in sub-XXX/sub-XXX_sessions.tsv; the dataset-level derived files that carry it (:func:_plan_derived: every contrast report that pooled it, any other page that names it, and its lines in any log or table, which are kept for everyone else); the participant's now-empty folders; and, when the participant has no session left anywhere here, their participants.tsv row. Then rebuilds the derivatives index, if there is one. A contrast report deleted here is rebuilt by running the contrasts again. A file nothing here can rule out is not kept silently: the result names it for a check by hand.

The labels are checked first (:func:checked_session), and every folder is checked for links: a bad label or a link anywhere refuses the whole deletion with :class:LifecycleError before anything is removed. The sourcedata session goes last, so a deletion interrupted part-way leaves the session to be deleted again rather than derived files of a session that is gone. Copies outside this data root (lab storage, an earlier export) are not reachable from here.

Source code in src/ssvep/runtime/lifecycle.py

def delete_session(out_dir, subject: str, session: str) -> DeleteResult:
    """Delete everything this data root holds of one ``sub``/``ses`` (option 0 after saving).

    Removes :func:`deletion_targets`; the session's row in ``sub-XXX/sub-XXX_sessions.tsv``; the
    dataset-level derived files that carry it (:func:`_plan_derived`: every contrast report that
    pooled it, any other page that names it, and its lines in any log or table, which are kept for
    everyone else); the participant's now-empty folders; and, when the participant has no session
    left anywhere here, their ``participants.tsv`` row. Then rebuilds the derivatives index, if
    there is one. A contrast report deleted here is rebuilt by running the contrasts again. A file
    nothing here can rule out is not kept silently: the result names it for a check by hand.

    The labels are checked first (:func:`checked_session`), and every folder is checked for links:
    a bad label or a link anywhere refuses the whole deletion with :class:`LifecycleError` before
    anything is removed. The ``sourcedata`` session goes last, so a deletion interrupted part-way
    leaves the session to be deleted again rather than derived files of a session that is gone.
    Copies outside this data root (lab storage, an earlier export) are not reachable from here.
    """
    root, subject, session, src = checked_session(out_dir, subject, session)
    targets = deletion_targets(root, subject, session)
    for t in targets:
        _check_tree_has_no_links(root, t)
    others = [c for d in (root / "sourcedata" / subject, root / subject) if d.is_dir()
              for c in d.iterdir() if c.name.startswith("ses-") and c.name != session]
    whole_subject = not others
    deriv = root / "derivatives"
    plan = _Plan()
    if deriv.is_dir():
        _check_no_link_down_to(root, deriv)
        plan = _plan_derived(deriv, targets, subject, session, session_protocols(src),
                             whole_subject)
    result = DeleteResult()
    # A redaction interrupted by a crash can leave its staging file; it is not a record of anyone's.
    if deriv.is_dir():
        for dirpath, dirnames, filenames in os.walk(deriv, followlinks=False):
            for name in filenames:
                stale = Path(dirpath) / name
                if name.endswith(".redacting") and _link_reason(stale) is None:
                    stale.unlink(missing_ok=True)
                    result.notes.append(f"removed a stale {name} left by an interrupted redaction")
    sessions_tsv = root / subject / f"{subject}_sessions.tsv"
    if sessions_tsv.is_file() and _link_reason(sessions_tsv) is None:
        try:
            if _drop_session_row(sessions_tsv, session):
                result.notes.append(f"removed {session} from {sessions_tsv.name}")
        except (OSError, UnicodeDecodeError) as exc:
            result.notes.append(f"{sessions_tsv.name} could not be updated ({exc}); remove the "
                                f"{session} row by hand")
    for p in plan.delete:
        p.unlink(missing_ok=True)
        result.deleted.append(p)
    for p, text in plan.redact:
        rel = p.relative_to(root).as_posix()
        tmp = p.with_name(p.name + ".redacting")
        try:
            # Staged, then swapped in whole: a crash mid-write leaves the original (other
            # participants' records) intact, never a truncated file.
            with tmp.open("w", encoding="utf-8", errors="surrogateescape", newline="") as fh:
                fh.write(text)
            os.replace(tmp, p)
            result.notes.append(f"removed the lines naming {subject} {session} from {rel}")
        except OSError as exc:
            tmp.unlink(missing_ok=True)
            result.notes.append(f"{rel} could not be updated ({exc}); remove the lines naming "
                                f"{subject} {session} by hand")
    _drop_ledger_lines(root, subject, session, result)
    if plan.contrasts:
        result.notes.append(f"deleted {plan.contrasts} contrast report(s) that pooled this session; "
                            f"run the contrasts again to rebuild them from the recordings that remain")
    for p in plan.review:
        result.notes.append(f"check by hand: {p.relative_to(root).as_posix()} may include this "
                            f"session, and nothing here can tell (it was written before reports "
                            f"listed whom they pooled, or it cannot be read)")
    # sourcedata last (see the docstring)
    for t in sorted(targets, key=lambda t: t == src):
        shutil.rmtree(t)
        result.deleted.append(t)
    # The participant's folders, once they hold no session.
    for parent in [root / "sourcedata" / subject, root / subject,
                   *((p / subject) for p in (deriv.iterdir() if deriv.is_dir() else ()) if p.is_dir())]:
        if not parent.is_dir() or _link_reason(parent) is not None:
            continue
        if any(c.name.startswith("ses-") for c in parent.iterdir()):
            continue
        if parent == root / subject:
            if not _only_sessions_tables(parent):
                continue
        elif parent == root / "sourcedata" / subject:
            if any(parent.iterdir()):
                continue
        elif any(c.is_dir() for c in parent.iterdir()):
            continue
        shutil.rmtree(parent)
        result.deleted.append(parent)
    if not (root / "sourcedata" / subject).exists() and not (root / subject).exists():
        try:
            if _drop_participant_row(root, subject):
                result.notes.append(f"removed {subject} from participants.tsv")
        except (OSError, UnicodeDecodeError) as exc:
            result.notes.append(f"participants.tsv could not be updated ({exc}); remove the "
                                f"{subject} row by hand")
    if whole_subject:
        _retire_code(root, subject, result)
    index = deriv / "ssvep-analysis"
    if index.is_dir():
        try:
            from ..analysis import derivatives
            derivatives.build_index(index, bids_root=root)
        except Exception as exc:          # the index is a convenience page, rebuilt on next analysis
            result.notes.append(f"the derivatives index could not be rebuilt ({exc}); it is "
                                f"rebuilt on the next analysis")
    return result