Defects the first real-data run exposed, and the fix that unblocks ingest

apps/api/src/cora/data/adapters/data_exchange_scan_reader.py

@@ -21,6 +21,23 @@
 number of phases the mode names (Start or End contribute one each,
 Both contributes two, None contributes zero).
 
+## Which timestamp is the acquisition time
+
+Not a layout question, a deployment one, so the caller declares it and
+this adapter reads what it is told. The layout offers two, and at 2-BM
+one of them is wrong: measured across six consecutive scan files,
+``start_date`` is the PREVIOUS scan's ``end_date`` every time, while
+``end_date`` matches the file's own close to within five seconds. The
+areaDetector timestamp attribute refreshes only while frames flow, so
+a file opened for scan N carries the value left over from scan N-1's
+last frame, and the value written on close is the current one.
+
+The correctness of the timestamp PV is not the issue and reading it
+live returns the right instant; the staleness is in the attribute
+cache between scans. A deployment whose writer does not have that
+defect keeps naming ``start_date``, which is why this is a per-
+deployment declaration rather than a change of default.
+
 ## Locking, and why it is not optional
 
 The file is opened read-only with HDF5 file locking disabled. Two
@@ -77,6 +94,15 @@
 _THETA_WHITE = "exchange/theta_white"
 _THETA_DARK = "exchange/theta_dark"
 _START_DATE = "process/acquisition/start_date"
+_END_DATE = "process/acquisition/end_date"
+# The timestamps this layout offers, by the name a deployment declares.
+# Closed here rather than taking a raw dataset path from configuration:
+# a caller may choose among the timestamps the layout defines, not point
+# the reader at an arbitrary dataset.
+_CAPTURED_AT_SOURCES: dict[str, str] = {
+    "start_date": _START_DATE,
+    "end_date": _END_DATE,
+}
 _NUM_ANGLES = "process/acquisition/rotation/num_angles"
 _NUM_FLAT_FIELDS = "process/acquisition/flat_fields/num_flat_fields"
 _FLAT_FIELD_MODE = "process/acquisition/flat_fields/flat_field_mode"
@@ -89,10 +115,22 @@ class DataExchangeScanReader:
 
     kind = "DataExchange"
 
-    def __init__(self, *, allowed_roots: tuple[str, ...]) -> None:
+    def __init__(
+        self,
+        *,
+        allowed_roots: tuple[str, ...],
+        captured_at_source: str = "start_date",
+    ) -> None:
         # Same canonicalisation as PosixChecksumAdapter: roots resolve
         # once so containment compares realpath-to-realpath.
         self._allowed_roots = tuple(os.path.realpath(root) for root in allowed_roots)
+        if captured_at_source not in _CAPTURED_AT_SOURCES:
+            raise ValueError(
+                f"captured_at_source {captured_at_source!r} is not a timestamp this "
+                f"layout offers; choose one of {sorted(_CAPTURED_AT_SOURCES)}."
+            )
+        self._captured_at_source = captured_at_source
+        self._captured_at_path = _CAPTURED_AT_SOURCES[captured_at_source]
 
     async def describe(self, *, locator_uri: str) -> ScanReadResult:
         try:
@@ -156,7 +194,7 @@ def _describe_sync(self, locator_uri: str) -> ScanReadResult:
                 _scalar_str(handle, _DARK_FIELD_MODE),
             )
 
-            start_date_raw = _scalar_str(handle, _START_DATE)
+            captured_at_raw = _scalar_str(handle, self._captured_at_path)
 
         dropped: int | None = None
         if commanded_projections is not None:
@@ -183,8 +221,9 @@ def _describe_sync(self, locator_uri: str) -> ScanReadResult:
             projection_angles_deg=projection_angles,
             flat_angles_deg=flat_angles,
             dark_angles_deg=dark_angles,
-            start_date=_parse_aware(start_date_raw),
-            start_date_raw=start_date_raw,
+            captured_at=_parse_aware(captured_at_raw),
+            captured_at_raw=captured_at_raw,
+            captured_at_source=self._captured_at_source,
             byte_size=stat.st_size,
             mtime_ns=stat.st_mtime_ns,
         )
@@ -226,10 +265,36 @@ def _scalar_int(handle: h5py.File, path: str) -> int | None:
         return None
     try:
         value = dataset[()]
-        if isinstance(value, bytes):
-            value = value.decode("utf-8", errors="replace")
-        return int(value)
-    except (TypeError, ValueError, OSError):
+    except OSError:
+        return None
+    if isinstance(value, bytes):
+        value = value.decode("utf-8", errors="replace")
+    parsed = _as_int(value)
+    if parsed is not None:
+        return parsed
+    # Same shape `_scalar_str` already handles: 2-BM writes every
+    # /process/acquisition integer as a 1-element array, and NumPy 2
+    # refuses int() on anything that is not 0-dimensional. Measured on
+    # a real scan file: num_angles is (1,) int32, so reading only the
+    # 0-d form left every commanded count None and made the shortfall
+    # check unable to fire at all. A longer array is NOT a scalar
+    # written oddly, so it stays unreadable rather than being reported
+    # as its own first element.
+    try:
+        if len(value) != 1:  # pyright: ignore[reportArgumentType]
+            return None
+        first = value[0]  # pyright: ignore[reportIndexIssue]
+    except (TypeError, IndexError, KeyError):
+        return None
+    if isinstance(first, bytes):
+        first = first.decode("utf-8", errors="replace")
+    return _as_int(first)
+
+
+def _as_int(value: object) -> int | None:
+    try:
+        return int(value)  # pyright: ignore[reportArgumentType]
+    except (TypeError, ValueError):
         return None
 
 

apps/api/src/cora/data/features/ingest_scan/handler.py

@@ -27,7 +27,7 @@
 
 ## The timestamp policy
 
-A parseable (timezone-aware) file `start_date` always wins; supplying
+A parseable (timezone-aware) file timestamp always wins; supplying
 `captured_at` alongside one is refused as ambiguous rather than
 silently overridden. With no parseable file value, the operator's
 `captured_at` is accepted as caller-asserted provenance, the same trust
@@ -36,6 +36,14 @@
 the only files lacking a timestamp here are complete-but-timestampless
 ones (layout divergence, time-IOC disconnect, format drift), each a
 staff-visible anomaly a fabricated value would bury.
+
+WHICH of the file's timestamps that is comes from the reader, not from
+here, and the record stamps `captured_at_source` with the answer. The
+policy is deliberately unchanged by that: a file value still beats an
+operator's, because the fix for a deployment whose writer emits a bad
+timestamp is to declare the good one, not to let every caller assert
+over the file. See `DataExchangeScanReader` for the 2-BM measurement
+that forced the distinction.
 """
 
 from pathlib import Path
@@ -89,8 +97,11 @@
     "properties": {
         "reader_kind": {"type": "string"},
         "checksum_computer_kind": {"type": "string"},
-        "captured_at_source": {"type": "string", "enum": ["start_date", "operator"]},
-        "start_date_raw": {"type": "string"},
+        "captured_at_source": {
+            "type": "string",
+            "enum": ["start_date", "end_date", "operator"],
+        },
+        "captured_at_raw": {"type": "string"},
         "projection_count": {"type": "integer"},
         "flat_count": {"type": "integer"},
         "dark_count": {"type": "integer"},
@@ -363,19 +374,20 @@ def envelopes(plan: StreamPlan) -> list[Any]:
 
 def _resolve_captured_at(command: IngestScan, described: Description) -> tuple[Any, str]:
     """Apply the locked timestamp policy; see the module docstring."""
-    if described.start_date is not None:
+    if described.captured_at is not None:
         if command.captured_at is not None:
             raise InvalidScanFileError(
                 f"captured_at was supplied but the file carries its own "
-                f"parseable timestamp ({described.start_date_raw}). Drop the "
-                f"supplied value; the file's own timestamp always wins."
+                f"parseable timestamp ({described.captured_at_raw}, from "
+                f"{described.captured_at_source}). Drop the supplied value; "
+                f"the file's own timestamp always wins."
             )
-        return described.start_date, "start_date"
+        return described.captured_at, described.captured_at_source
     if command.captured_at is not None:
         return command.captured_at, "operator"
     detail = (
-        f"present but not parseable as an unambiguous instant: {described.start_date_raw!r}"
-        if described.start_date_raw is not None
+        f"present but not parseable as an unambiguous instant: {described.captured_at_raw!r}"
+        if described.captured_at_raw is not None
         else "absent"
     )
     raise InvalidScanFileError(
@@ -402,8 +414,8 @@ def _build_evidence(
         "dark_count": described.dark_count,
         "invalid_count": described.invalid_count,
     }
-    if described.start_date_raw is not None:
-        evidence["start_date_raw"] = described.start_date_raw
+    if described.captured_at_raw is not None:
+        evidence["captured_at_raw"] = described.captured_at_raw
     if described.commanded_projection_count is not None:
         evidence["commanded_projection_count"] = described.commanded_projection_count
     if described.commanded_flat_count is not None:

apps/api/src/cora/data/ports/scan_reader.py

@@ -70,12 +70,21 @@ class Description:
     digest pass's snapshot to refuse files that changed while being
     read (the live-file guard).
 
-    ``start_date`` is parsed ONLY when the file's timestamp string
+    ``captured_at`` is parsed ONLY when the file's timestamp string
     carries an unambiguous offset; a naive string stays raw in
-    ``start_date_raw`` with ``start_date=None``, because guessing a
+    ``captured_at_raw`` with ``captured_at=None``, because guessing a
     timezone would manufacture provenance. The site timezone rule, once
     staff confirm it, lands in the adapter and turns those raws into
     parsed values.
+
+    ``captured_at_source`` names WHICH of the file's timestamps that
+    came from, and it is not always the obvious one. A layout can offer
+    several, and a deployment's writer can be wrong about one of them:
+    2-BM's ``start_date`` is measurably the PREVIOUS scan's end, while
+    its ``end_date`` is correct to within seconds. So the reader is told
+    which to believe, and the record carries the answer rather than
+    leaving a reader to assume. The field is a plain ``str`` so a new
+    layout can name a timestamp this one has never heard of.
     """
 
     media_type: str
@@ -91,8 +100,9 @@ class Description:
     projection_angles_deg: tuple[float, ...] | None
     flat_angles_deg: tuple[float, ...] | None
     dark_angles_deg: tuple[float, ...] | None
-    start_date: datetime | None
-    start_date_raw: str | None
+    captured_at: datetime | None
+    captured_at_raw: str | None
+    captured_at_source: str
     byte_size: int
     mtime_ns: int
 

apps/api/src/cora/data/wire.py

@@ -316,7 +316,8 @@ def wire_data(deps: Kernel) -> DataHandlers:
                 ingest_scan.bind(
                     deps,
                     scan_reader=DataExchangeScanReader(
-                        allowed_roots=deps.settings.posix_checksum_roots
+                        allowed_roots=deps.settings.posix_checksum_roots,
+                        captured_at_source=deps.settings.scan_captured_at_source,
                     ),
                     checksum_computer=PosixChecksumAdapter(
                         allowed_roots=deps.settings.posix_checksum_roots

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

@@ -510,6 +510,26 @@ class Settings(BaseSettings):
     # See `cora.data.adapters.posix_checksum`.
     posix_checksum_roots: tuple[str, ...] = ()
 
+    # Data BC -- which of a scan file's timestamps is the acquisition
+    # time. `start_date` (the default) preserves the behaviour every
+    # deployment had before this setting existed.
+    #
+    # A deployment overrides it when its writer emits a timestamp that
+    # is wrong rather than merely different. At 2-BM, measured across
+    # six consecutive files, `start_date` is the PREVIOUS scan's
+    # `end_date` because the areaDetector timestamp attribute refreshes
+    # only while frames flow; `end_date` is correct to within seconds of
+    # the file's own close. Ingesting there without `end_date` records a
+    # capture time that is wrong by however long the gap between scans
+    # was, and the policy that a file value beats an operator's means
+    # nobody can correct it afterwards.
+    #
+    # The value is validated against the layout's own timestamp set by
+    # the reader, which refuses a name the layout does not offer rather
+    # than silently reading nothing. Read from
+    # SCAN_CAPTURED_AT_SOURCE.
+    scan_captured_at_source: str = "start_date"
+
     # Equipment BC — PIDINST integration (slice E.1)
     # `facility_publisher` is the institutional `publisher` field emitted
     # on every PIDINST record produced by `GET /assets/{asset_id}/pidinst`

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

@@ -79,7 +79,11 @@
     registered_envelope_classes,
     resolve,
 )
-from cora.infrastructure.record_export._render import render_row, render_value
+from cora.infrastructure.record_export._render import (
+    UndecodedJsonColumnError,
+    render_row,
+    render_value,
+)
 from cora.infrastructure.record_export._stream_types import (
     KNOWN_STREAM_TYPES,
     UnknownStreamTypeError,
@@ -113,6 +117,7 @@
     "Tier1Redactor",
     "TokenMap",
     "TwoTierRecord",
+    "UndecodedJsonColumnError",
     "UnknownEventTypeError",
     "UnknownLogbookKindError",
     "UnknownStreamTypeError",

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

@@ -47,7 +47,7 @@
     "ActorDeactivated": {"actor_id": "token:uuid", "occurred_at": "keep:time"},
     "ActorProfileForgotten": {"actor_id": "token:uuid", "occurred_at": "keep:time"},
     "ActorReactivated": {"actor_id": "token:uuid", "occurred_at": "keep:time"},
-    "ActorRegistered": {
+    "ActorRegisteredV2": {
         "actor_id": "token:uuid",
         "kind": "keep:enum:ActorKind",
         "occurred_at": "keep:time",

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

@@ -51,8 +51,57 @@ def render_value(value: object) -> object:
     return value
 
 
+class UndecodedJsonColumnError(RuntimeError):
+    """A jsonb column arrived as a `str`, so the connection lacks codecs.
+
+    asyncpg returns jsonb as text unless the connection registers the
+    codecs `cora.infrastructure.postgres.pool` installs. An export over
+    such a connection is not merely degraded, it is wrong in a way that
+    hides: every payload becomes one opaque string, the bundle is
+    written, the manifest hashes it, and `verify_record_hash.py` reports
+    OK, because the artifact is perfectly self-consistent about the
+    wrong structure. Redaction is the only stage that would notice, and
+    only by crashing on a `str` where it wanted a mapping.
+
+    An exporter whose product is meant to be verifiable by a stranger
+    cannot leave that to luck, so the shape is checked where rows enter
+    rather than where they happen to break.
+    """
+
+    def __init__(self, column: str) -> None:
+        super().__init__(
+            f"column {column!r} arrived as a str, so this connection has no "
+            "jsonb codec registered; build it through "
+            "cora.infrastructure.postgres.pool.create_pool, or register the "
+            "same codecs before exporting. Exporting now would hash a record "
+            "whose payloads are strings."
+        )
+        self.column = column
+
+
+# The jsonb columns an exported row can carry: `payload` and `metadata`
+# on `events`, `payload` again on the activities entries table. Named
+# rather than sniffed, because only these are known to be jsonb.
+#
+# The check reads a decoded `str` as proof of a missing codec, which
+# holds because every one of these columns is written from a
+# `to_payload()` returning a dict, so a correctly decoded value is
+# always a mapping (or NULL). A jsonb column that legitimately held a
+# bare JSON string would decode to `str` too and trip this; none does,
+# and one arriving is a modelling change that should come here first.
+_JSON_COLUMNS = ("payload", "metadata")
+
+
 def render_row(row: "dict[str, object] | asyncpg.Record") -> dict[str, object]:
     """Render every column of one asyncpg `Record` (production) or plain
     `dict` (unit tests -- `Record` isn't constructible outside asyncpg's
-    own protocol machinery)."""
-    return {key: render_value(value) for key, value in row.items()}
+    own protocol machinery).
+
+    Refuses a jsonb column that arrived as a `str`. See
+    `UndecodedJsonColumnError`.
+    """
+    rendered = {key: render_value(value) for key, value in row.items()}
+    for column in _JSON_COLUMNS:
+        if isinstance(rendered.get(column), str):
+            raise UndecodedJsonColumnError(column)
+    return rendered