Slice 11: test-vs-real provenance on a witnessed Run

apps/api/src/cora/api/_capture_observer.py

@@ -17,7 +17,12 @@
 claim on top (slice 9); the optional `images_saved` /
 `images_collected` progress roles (slice 10) each pump their own
 `CaptureProgressObservation` onto the same merged stream, siblings of
-the phase pumps.
+the phase pumps. The optional `testing` role (slice 11) pumps its own
+`CapturePreconditionBypassObservation`: a tri-state reading of whether
+the substrate is bypassing its own beam preconditions for this capture
+code, decoded via the same `binary_code` the `abort` role already
+uses (2-BM's `Testing` PV is the identical `DBR_ENUM` record type as
+`AbortScan`).
 
 ## One deliberate inversion from the Enclosure precedent
 
@@ -97,6 +102,7 @@
     CaptureLifecycleObservation,
     CaptureObserverScope,
     CapturePhase,
+    CapturePreconditionBypassObservation,
     CaptureProgressObservation,
 )
 from cora.shared.reach import ReachTier
@@ -111,13 +117,14 @@
 ROLE_ABORT = "abort"
 ROLE_IMAGES_SAVED = "images_saved"
 ROLE_IMAGES_COLLECTED = "images_collected"
+ROLE_TESTING = "testing"
 """CORA-owned role keys, matching `Settings.capture_watch_pvs`'s documented
 example. Module-public (not `_`-prefixed) because other composition-root
 modules read observations back out, or dispatch decoders, by these same
 keys and must not carry their own copy of the literal strings:
 `RunWitnessRecorder._build_progress_snapshot` (`_run_witness.py`) reads
 `ROLE_IMAGES_SAVED` / `ROLE_IMAGES_COLLECTED`, and `capture_watch_preflight`
-dispatches its per-role decode check on all four. Import these, not
+dispatches its per-role decode check on all five. Import these, not
 `_PROGRESS_ROLES` below, so a rename or a new role here cannot silently
 desync from either reader. `server_running` stays declared-and-unread:
 tool liveness is a different concern from capture progress (slice 10)."""
@@ -259,8 +266,13 @@ class ControlPortCaptureObserver:
     A code with no `abort` entry watches `status` only, exactly as
     before this role existed. The `images_saved` / `images_collected`
     progress roles (also optional, independently declared per code)
-    each pump `CaptureProgressObservation` readings; `server_running`
-    stays declared and unread (tool liveness, not capture progress).
+    each pump `CaptureProgressObservation` readings. The `testing` role
+    (also optional, independently declared per code) pumps
+    `CapturePreconditionBypassObservation` readings, a tri-state claim
+    rather than a phase or a counter; see that dataclass's own
+    docstring.
+    `server_running` stays declared and unread (tool liveness, not
+    capture progress).
     """
 
     def __init__(
@@ -278,6 +290,11 @@ def __init__(
         self._abort_pvs = {
             code: roles[ROLE_ABORT] for code, roles in capture_pvs.items() if ROLE_ABORT in roles
         }
+        self._testing_pvs = {
+            code: roles[ROLE_TESTING]
+            for code, roles in capture_pvs.items()
+            if ROLE_TESTING in roles
+        }
         self._progress_pvs = {
             code: filtered
             for code, roles in capture_pvs.items()
@@ -302,6 +319,11 @@ async def _drain(self, scope: CaptureObserverScope) -> AsyncGenerator[AnyCapture
             for code in sorted(scope.capture_codes)
             if code in self._abort_pvs
         ]
+        testing_pvs = [
+            (code, self._testing_pvs[code])
+            for code in sorted(scope.capture_codes)
+            if code in self._testing_pvs
+        ]
         progress_pvs = [
             (code, role, pv)
             for code in sorted(scope.capture_codes)
@@ -312,6 +334,7 @@ async def _drain(self, scope: CaptureObserverScope) -> AsyncGenerator[AnyCapture
         pump_tasks = (
             [asyncio.create_task(self._pump(code, pv, queue)) for code, pv in pvs]
             + [asyncio.create_task(self._pump_abort(code, pv, queue)) for code, pv in abort_pvs]
+            + [asyncio.create_task(self._pump_testing(code, pv, queue)) for code, pv in testing_pvs]
             + [
                 asyncio.create_task(self._pump_progress(code, role, pv, queue))
                 for code, role, pv in progress_pvs
@@ -413,6 +436,33 @@ async def _pump_progress(
         finally:
             queue.put_nowait(_PUMP_DONE)
 
+    async def _pump_testing(
+        self,
+        code: str,
+        pv: str,
+        queue: asyncio.Queue[AnyCaptureObservation | _PumpDone],
+    ) -> None:
+        """Sibling pump for the optional `testing` role.
+
+        Unlike `_pump_abort`, EVERY reading is enqueued, including one
+        that decodes clear or does not decode at all: `testing` is a
+        tri-state reading in its own right (see
+        `CapturePreconditionBypassObservation`), not a phase claim where
+        a clear or unresolvable reading means "nothing happened". No
+        `_unreached` counterpart, mirroring `_pump_progress`: a disconnect must not
+        erase the last reading `RunWitnessRecorder` retained, since the
+        dual-clock (`observed_at`) discipline exists precisely so
+        staleness is visible at genesis time rather than papered over by
+        a synthesized "unknown" on every reconnect.
+        """
+        try:
+            async for reading in self._control_port.subscribe(pv):
+                queue.put_nowait(self._from_testing_reading(code, pv, reading))
+        except ControlNotConnectedError:
+            pass
+        finally:
+            queue.put_nowait(_PUMP_DONE)
+
     async def _poll(
         self,
         code: str,
@@ -506,6 +556,29 @@ def _from_progress_reading(
             source_id=pv,
         )
 
+    def _from_testing_reading(
+        self, code: str, pv: str, reading: Measurement
+    ) -> CapturePreconditionBypassObservation:
+        """A `testing`-role reading, decoded via `binary_code` exactly as
+        the `abort` role's reading is (2-BM's `Testing` PV is the
+        identical `DBR_ENUM` record type as `AbortScan`): `1` -> `True`
+        (asserted: bypassing beam preconditions), `0` -> `False` (clear:
+        a positive claim of a real acquisition), `None` -> `None`
+        (unresolved). Unlike `_from_abort_reading`, every reading
+        constructs an observation; there is no reading here that means
+        "nothing happened".
+        """
+        code_value = binary_code(reading.value)
+        bypassed = None if code_value is None else code_value == 1
+        return CapturePreconditionBypassObservation(
+            capture_code=code,
+            beam_preconditions_bypassed=bypassed,
+            reach_tier=ReachTier.RELAYED,
+            observed_at=reading.produced_at,
+            source_kind=_SOURCE_KIND,
+            source_id=pv,
+        )
+
     def _probe_only(self, code: str, pv: str, reach_tier: ReachTier) -> CaptureLifecycleObservation:
         """A poll tick's result: reach evidence with no status claim."""
         return CaptureLifecycleObservation(
@@ -540,6 +613,7 @@ def _unreached(self, code: str, pv: str) -> CaptureLifecycleObservation:
     "ROLE_IMAGES_COLLECTED",
     "ROLE_IMAGES_SAVED",
     "ROLE_STATUS",
+    "ROLE_TESTING",
     "ControlPortCaptureObserver",
     "binary_code",
     "classify_capture_status",

apps/api/src/cora/api/_run_witness.py

@@ -208,7 +208,10 @@
 from cora.api._capture_observer import ROLE_IMAGES_COLLECTED, ROLE_IMAGES_SAVED
 from cora.api._capture_progress_feeder import CaptureProgressFeeder, capture_progress_flush_loop
 from cora.infrastructure.logging import get_logger
-from cora.run.aggregates.run.state import CaptureProgressSnapshot
+from cora.run.aggregates.run.state import (
+    CapturePreconditionBypassSnapshot,
+    CaptureProgressSnapshot,
+)
 from cora.run.errors import UnauthorizedError
 from cora.run.features.list_runs.query import ListRuns
 from cora.run.features.record_witnessed_run.command import RecordWitnessedRun
@@ -218,6 +221,7 @@
     CaptureLifecycleObservation,
     CaptureObserverScope,
     CapturePhase,
+    CapturePreconditionBypassObservation,
     CaptureProgressObservation,
 )
 from cora.shared.identity import MonitorSourceId
@@ -327,6 +331,7 @@ def __init__(
         self._settings = settings
         self._open_captures: dict[str, UUID] = dict(open_captures or {})
         self._last_progress: dict[str, dict[str, CaptureProgressObservation]] = {}
+        self._last_precondition_bypass: dict[str, CapturePreconditionBypassObservation] = {}
 
     def open_captures(self) -> dict[str, UUID]:
         """A snapshot of every capture_code currently open, mapped to
@@ -405,6 +410,47 @@ def observe_progress(self, observation: CaptureProgressObservation) -> None:
         by_role = self._last_progress.setdefault(observation.capture_code, {})
         by_role[observation.role] = observation
 
+    def observe_precondition_bypass(
+        self, observation: CapturePreconditionBypassObservation
+    ) -> None:
+        """Retain the latest `testing`-role reading per capture_code, so
+        the NEXT `BEGUN` for this code can stamp it onto the witnessed
+        genesis (see `_promote` / `_build_precondition_bypass_snapshot`).
+
+        Gated on `run_witness_recording_enabled`, same as
+        `observe_progress`: shadow mode retains nothing because it
+        writes nothing.
+
+        Deliberately NEVER evicted, unlike `_last_progress`: the
+        `testing` role is a substrate-level flag an operator sets
+        independent of any one capture (TomoScan does not reset it
+        between scans), so the reading retained across a capture
+        boundary is not stale evidence about the WRONG capture the way
+        a leftover progress count would be. It stays the honest answer
+        to "what did `testing` last read" until a fresh reading
+        replaces it, however long ago that was; `observed_at` is what
+        lets a reader judge that gap at genesis time, not eviction.
+        """
+        if not self._settings.run_witness_recording_enabled:
+            return
+        self._last_precondition_bypass[observation.capture_code] = observation
+
+    def _build_precondition_bypass_snapshot(
+        self, code: str
+    ) -> CapturePreconditionBypassSnapshot | None:
+        """The evidence a witnessed genesis carries for `code`: the last
+        `testing` reading `observe_precondition_bypass` retained, or
+        `None` if none has ever arrived (no `testing` role declared for
+        this code, or the substrate has not reported one yet).
+        """
+        observation = self._last_precondition_bypass.get(code)
+        if observation is None:
+            return None
+        return CapturePreconditionBypassSnapshot(
+            beam_preconditions_bypassed=observation.beam_preconditions_bypassed,
+            observed_at=observation.observed_at,
+        )
+
     async def _promote(self, observation: CaptureLifecycleObservation) -> None:
         # A prior capture's retained progress, if any, belongs to that
         # capture's own terminal, never to this one: clear before
@@ -430,6 +476,9 @@ async def _promote(self, observation: CaptureLifecycleObservation) -> None:
             capture_code=observation.capture_code,
             monitor_source_id=RUN_WITNESS_MONITOR_SOURCE_ID,
             trigger="Monitor",
+            capture_precondition_bypass_snapshot=self._build_precondition_bypass_snapshot(
+                observation.capture_code
+            ),
         )
         try:
             run_id = await self._record_witnessed_run(
@@ -694,7 +743,12 @@ async def run_witness_loop(
     ._build_progress_snapshot`) and `feeder.offer()` (buffers it for
     the next `AppendObservations` flush). Order between the two is
     immaterial: both are synchronous and neither raises in normal
-    operation. A `CaptureLifecycleObservation` on a phase in
+    operation. A `CapturePreconditionBypassObservation` goes only to
+    `recorder.observe_precondition_bypass()` (retains the latest
+    reading so the NEXT genesis can stamp it; see `RunWitnessRecorder
+    ._build_precondition_bypass_snapshot`): it has no `feeder`
+    counterpart, since it is never written as an `AppendObservations`
+    row, only carried onto `RunStarted`. A `CaptureLifecycleObservation` on a phase in
     `_FLUSH_TRIGGER_PHASES` triggers `feeder.flush_capture()` BEFORE the
     recorder acts on it, so a capture's buffered progress trail is
     attributed to its Run before that Run can close or be replaced;
@@ -716,6 +770,10 @@ async def run_witness_loop(
                     if feeder is not None:
                         feeder.offer(observation)
                     continue
+                if isinstance(observation, CapturePreconditionBypassObservation):
+                    if recorder is not None:
+                        recorder.observe_precondition_bypass(observation)
+                    continue
                 if feeder is not None and observation.phase in _FLUSH_TRIGGER_PHASES:
                     try:
                         await feeder.flush_capture(observation.capture_code)

apps/api/src/cora/api/capture_watch_preflight.py

@@ -14,10 +14,13 @@
 written from a docstring's assumed shape rather than what the real IOC
 puts on the wire: `NumAngles` a 1-element array not a scalar, `AbortScan`
 an ENUM label `'No'` not `0`, `ImagesSaved` / `ImagesCollected` a
-`"<done>/<total>"` string not a float. All three would have shown up in
-one run of this command: `kind` flags array-vs-scalar and, via
-`classify_capture_status` / `binary_code` / `progress_counts`, a decode
-verdict flags anything a role's real decoder cannot accept.
+`"<done>/<total>"` string not a float. Two of the three would have shown
+up in one run of this command: via `classify_capture_status` /
+`binary_code` / `progress_counts`, a decode verdict flags the `AbortScan`
+enum label and the `ImagesSaved` / `ImagesCollected` string. The third,
+`NumAngles`, would NOT: it is read from the HDF5 file
+(`data_exchange_scan_reader.py:106`), never from a PV, so it is not in
+`CAPTURE_WATCH_PVS` and no channel-access preflight can reach it.
 
 ## Read-only, changes nothing
 
@@ -33,13 +36,19 @@
 `ControlPort.read()` already collapses the wire type into the closed
 `MeasurementKind` set (`_kind_for` in `epics_ca_control_port.py`: DBR_ENUM
 -> Categorical, `element_count > 1` -> Array, else Scalar) before this
-command ever sees a reading. That is sufficient to catch all three defects
-above: `kind` alone flags array-vs-scalar and enum-vs-int, and the raw
-`value` makes a `"12/100"` string visibly not a bare float. Reaching past
-`ControlPort` for the raw aioca `.datatype` would add a substrate-specific
-escape hatch that no other caller needs, breaking the discipline every
-other consumer in this codebase already keeps of never leaking EPICS
-specifics past the adapter.
+command ever sees a reading. That is enough to catch the two wire-shape
+defects this preflight can actually reach: `kind` flags a DBR_ENUM
+masquerading as a scalar (`AbortScan` resolves as Categorical, not
+Scalar) and the raw `value` makes a `"12/100"` string visibly not a bare
+float. It is NOT enough to flag a one-element array as anything but a
+Scalar: `_kind_for` classifies an array by `element_count > 1`, so a
+genuinely 1-element array (`NumAngles`'s own shape, had it been readable
+as a PV here) reports `kind=Scalar`, indistinguishable from a real
+scalar via `kind` alone. Reaching past `ControlPort` for the raw aioca
+`.datatype` would add a substrate-specific escape hatch that no other
+caller needs, breaking the discipline every other consumer in this
+codebase already keeps of never leaking EPICS specifics past the
+adapter.
 
 ## Per-role decode verdict
 
@@ -54,6 +63,9 @@
   - `images_saved` / `images_collected` (`ROLE_IMAGES_SAVED` /
     `ROLE_IMAGES_COLLECTED`): `progress_counts`. BAD when it returns
     `None`.
+  - `testing` (`ROLE_TESTING`): `binary_code`, same decoder as `abort`
+    (2-BM's `Testing` PV is the identical `DBR_ENUM` record type as
+    `AbortScan`). BAD when it returns `None`.
   - any other declared role (e.g. `server_running`, which production
     itself declares and never decodes): reports `kind` / `value` only,
     verdict `n/a`. Not decoding it here does not make it undecodable
@@ -77,6 +89,7 @@
     ROLE_IMAGES_COLLECTED,
     ROLE_IMAGES_SAVED,
     ROLE_STATUS,
+    ROLE_TESTING,
     binary_code,
     classify_capture_status,
     progress_counts,
@@ -213,6 +226,11 @@ def _decode_verdict(
         if code is None:
             return "unrecognized", False
         return ("asserted" if code == 1 else "clear"), True
+    if role == ROLE_TESTING:
+        code = binary_code(reading.value)
+        if code is None:
+            return "unrecognized", False
+        return ("testing" if code == 1 else "real"), True
     if role in _PROGRESS_ROLES:
         counts = progress_counts(reading.value)
         if counts is None:

apps/api/src/cora/infrastructure/config.py

@@ -727,13 +727,18 @@ class Settings(BaseSettings):
     #       "server_running": "2bmb:TomoScan:ServerRunning",
     #       "abort": "2bmb:TomoScan:AbortScan",
     #       "images_saved": "2bmb:TomoScan:ImagesSaved",
-    #       "images_collected": "2bmb:TomoScan:ImagesCollected"
+    #       "images_collected": "2bmb:TomoScan:ImagesCollected",
+    #       "testing": "2bmb:TomoScan:Testing"
     #     }
     #   }'
     #
     # `status` is a DBR_CHAR waveform at 2-BM; the deployment's
     # CONTROL_PORT_ROUTES must declare it in `text_addresses` or it
-    # decodes as raw bytes, not text. See `cora.api._capture_observer`.
+    # decodes as raw bytes, not text. `testing` (slice 11, optional per
+    # code) is a DBR_ENUM, the same record type as `abort`: whether
+    # tomoscan is bypassing its own beam preconditions for this capture,
+    # carried onto the witnessed genesis, never onto
+    # `Observation.is_simulated`. See `cora.api._capture_observer`.
     capture_watch_pvs: dict[str, dict[str, str]] = {}
 
     # The `status` role's raw substrate literal, mapped onto CORA's

apps/api/src/cora/infrastructure/record_export/_dispositions.py

@@ -1636,6 +1636,10 @@
             "workaround_excerpt": "drop:text",
         },
         "campaign_id": "token:uuid",
+        "capture_precondition_bypass_snapshot": {
+            "beam_preconditions_bypassed": "keep:number",
+            "observed_at": "keep:time",
+        },
         "conduct_mode": "keep:enum:ConductMode",
         "decided_by_decision_id": "token:uuid",
         "effective_parameters": "drop:opaque",

apps/api/src/cora/run/aggregates/run/__init__.py

@@ -79,6 +79,7 @@
     RUN_NAME_MAX_LENGTH,
     RUN_PINNED_CALIBRATIONS_MAX_ENTRIES,
     SAMPLING_PROCEDURE_VALUES,
+    CapturePreconditionBypassSnapshot,
     CaptureProgressSnapshot,
     ChannelName,
     ConductMode,
@@ -160,6 +161,7 @@
     "RUN_NAME_MAX_LENGTH",
     "RUN_PINNED_CALIBRATIONS_MAX_ENTRIES",
     "SAMPLING_PROCEDURE_VALUES",
+    "CapturePreconditionBypassSnapshot",
     "CaptureProgressSnapshot",
     "CautionAcknowledgement",
     "ChannelName",