From 713171618c9c4a374fd3491ac9915b852265db26 Mon Sep 17 00:00:00 2001
From: xmap <16776958+xmap@users.noreply.github.com>
Date: Wed, 12 Aug 2026 07:05:11 -0500
Subject: [PATCH 1/5] Read the commanded counts 2-BM actually writes

`_scalar_int` read only the 0-dimensional form, so at the beamline
every commanded count came back None: NumPy 2 refuses `int()` on an
array that is not 0-d, the reader caught the TypeError, and the
shortfall check silently had nothing to compare against. Measured on a
real scan file, `/process/acquisition/rotation/num_angles` is a
1-element int32 array, and the flat and dark mode-and-count groups are
absent entirely, so `num_angles` was the only commanded fact available
and it was the one being dropped.

`_scalar_str` already handled exactly this shape, with a comment saying
some writers store scalars as 1-element arrays. That is why the
timestamp parsed while the integers did not, and why the whole module
stayed green: the fixture wrote the 0-d shape the reader could read
rather than the shape the producer emits.

So the shape is now a fixture parameter, not an assumption. Every
`/process/acquisition` scalar can be written either way and the
existing matrix runs against both.

A longer array stays unreadable rather than being reported as its own
first element: that would fabricate a commanded count out of a dataset
whose meaning the reader does not know.

Found by running the reader against test_005.h5, a real 2-BM scan,
after the first scan CORA has ever watched from end to end.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 .../adapters/data_exchange_scan_reader.py     | 34 +++++++-
 .../data/test_data_exchange_scan_reader.py    | 86 +++++++++++++++++--
 2 files changed, 110 insertions(+), 10 deletions(-)

diff --git a/apps/api/src/cora/data/adapters/data_exchange_scan_reader.py b/apps/api/src/cora/data/adapters/data_exchange_scan_reader.py
index 8e44c400276..93ea16945f4 100644
--- a/apps/api/src/cora/data/adapters/data_exchange_scan_reader.py
+++ b/apps/api/src/cora/data/adapters/data_exchange_scan_reader.py
@@ -226,10 +226,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
 
 
diff --git a/apps/api/tests/unit/data/test_data_exchange_scan_reader.py b/apps/api/tests/unit/data/test_data_exchange_scan_reader.py
index c0cb54c5d27..5276af76241 100644
--- a/apps/api/tests/unit/data/test_data_exchange_scan_reader.py
+++ b/apps/api/tests/unit/data/test_data_exchange_scan_reader.py
@@ -23,6 +23,18 @@
 pytestmark = pytest.mark.unit
 
 
+def _scalar(value: int | str, *, boxed: bool) -> object:
+    """A scalar as the two shapes real writers use.
+
+    2-BM writes every `/process/acquisition` scalar as a 1-element
+    array, not as a 0-dimensional dataset, measured against a real scan
+    file. Tests that only ever wrote the 0-d shape were green while the
+    reader could not read a single commanded count at the beamline, so
+    the shape is a fixture parameter rather than an assumption.
+    """
+    return [value] if boxed else value
+
+
 def _write_scan(
     path: Path,
     *,
@@ -36,7 +48,11 @@ def _write_scan(
     dark_mode: str | None = "Start",
     num_dark_fields: int | None = 2,
     start_date: str | None = "2026-07-29T10:15:30-05:00",
+    boxed_scalars: bool = False,
 ) -> None:
+    def scalar(value: int | str) -> object:
+        return _scalar(value, boxed=boxed_scalars)
+
     with h5py.File(path, "w") as f:
         f.create_dataset("exchange/data", data=np.zeros((projections, 4, 4), dtype=np.uint16))
         if flats is not None:
@@ -46,21 +62,25 @@ def _write_scan(
         if theta:
             f.create_dataset("exchange/theta", data=np.linspace(0.0, 180.0, projections))
         if num_angles is not None:
-            f.create_dataset("process/acquisition/rotation/num_angles", data=num_angles)
+            f.create_dataset("process/acquisition/rotation/num_angles", data=scalar(num_angles))
         if num_flat_fields is not None:
             f.create_dataset(
-                "process/acquisition/flat_fields/num_flat_fields", data=num_flat_fields
+                "process/acquisition/flat_fields/num_flat_fields", data=scalar(num_flat_fields)
             )
         if flat_mode is not None:
-            f.create_dataset("process/acquisition/flat_fields/flat_field_mode", data=flat_mode)
+            f.create_dataset(
+                "process/acquisition/flat_fields/flat_field_mode", data=scalar(flat_mode)
+            )
         if num_dark_fields is not None:
             f.create_dataset(
-                "process/acquisition/dark_fields/num_dark_fields", data=num_dark_fields
+                "process/acquisition/dark_fields/num_dark_fields", data=scalar(num_dark_fields)
             )
         if dark_mode is not None:
-            f.create_dataset("process/acquisition/dark_fields/dark_field_mode", data=dark_mode)
+            f.create_dataset(
+                "process/acquisition/dark_fields/dark_field_mode", data=scalar(dark_mode)
+            )
         if start_date is not None:
-            f.create_dataset("process/acquisition/start_date", data=start_date)
+            f.create_dataset("process/acquisition/start_date", data=scalar(start_date))
 
 
 def _reader(tmp_path: Path) -> DataExchangeScanReader:
@@ -87,6 +107,60 @@ async def test_describe_clean_scan_reports_complete_counts(tmp_path: Path) -> No
     assert result.media_type == "application/x-hdf5"
 
 
+async def test_describe_one_element_array_scalars_read_the_same_as_zero_d(tmp_path: Path) -> None:
+    """The shape 2-BM actually writes.
+
+    Measured on a real scan file: `/process/acquisition` scalars are
+    1-element arrays, and NumPy 2 refuses `int()` on those. Before the
+    reader handled them, every commanded count came back None at the
+    beamline while this whole module stayed green, because the fixture
+    only ever wrote the 0-dimensional shape.
+    """
+    scan = tmp_path / "scan_boxed.h5"
+    _write_scan(scan, boxed_scalars=True)
+
+    result = await _reader(tmp_path).describe(locator_uri=scan.as_uri())
+
+    assert isinstance(result, Description)
+    assert result.commanded_projection_count == 5
+    assert result.commanded_flat_count == 2
+    assert result.commanded_dark_count == 2
+    assert result.dropped_frame_count == 0
+    assert result.start_date_raw == "2026-07-29T10:15:30-05:00"
+
+
+async def test_describe_boxed_shortfall_reports_the_dropped_frames(tmp_path: Path) -> None:
+    scan = tmp_path / "scan_boxed_short.h5"
+    _write_scan(scan, projections=3, num_angles=5, boxed_scalars=True)
+
+    result = await _reader(tmp_path).describe(locator_uri=scan.as_uri())
+
+    assert isinstance(result, Description)
+    assert result.commanded_projection_count == 5
+    assert result.dropped_frame_count == 2
+
+
+async def test_describe_multi_element_array_where_a_scalar_belongs_reads_none(
+    tmp_path: Path,
+) -> None:
+    """A longer array is not a scalar written oddly.
+
+    Reporting its first element would fabricate a commanded count from
+    a dataset whose meaning the reader does not know.
+    """
+    scan = tmp_path / "scan_multi.h5"
+    _write_scan(scan)
+    with h5py.File(scan, "a") as f:
+        del f["process/acquisition/rotation/num_angles"]
+        f.create_dataset("process/acquisition/rotation/num_angles", data=[5, 6, 7])
+
+    result = await _reader(tmp_path).describe(locator_uri=scan.as_uri())
+
+    assert isinstance(result, Description)
+    assert result.commanded_projection_count is None
+    assert result.dropped_frame_count is None
+
+
 async def test_describe_theta_absent_reads_structurally_incomplete(tmp_path: Path) -> None:
     scan = tmp_path / "scan_002.h5"
     _write_scan(scan, theta=False)

From 3a07f5da0f35d156f00d0d362a4bf868f796b45e Mon Sep 17 00:00:00 2001
From: xmap <16776958+xmap@users.noreply.github.com>
Date: Wed, 12 Aug 2026 07:09:55 -0500
Subject: [PATCH 2/5] Refuse to export a record whose payloads are strings

asyncpg hands jsonb back as text unless the connection registers the
codecs `postgres.pool` installs. Nothing in the exporter checked, so an
export over a plain `asyncpg.connect` produced a bundle whose every
payload was one opaque string, wrote it, hashed it, and passed
`verify_record_hash.py`. The artifact was perfectly self-consistent
about the wrong structure, which is the one failure a record meant to
be verifiable by a stranger must not have.

Measured, not hypothesised: this is what the first export of the live
2-BM database did. Only redaction noticed, and only by crashing on a
`str` where it wanted a mapping, which is luck rather than a check.

So the shape is checked where rows enter. A decoded `str` in `payload`
or `metadata` can only mean the codec is missing, because both columns
are written from a `to_payload()` that returns a dict, and the error
says which connection to build instead.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 .../infrastructure/record_export/__init__.py  |  7 ++-
 .../infrastructure/record_export/_render.py   | 53 ++++++++++++++++++-
 .../record_export/test_render.py              | 48 ++++++++++++++++-
 3 files changed, 104 insertions(+), 4 deletions(-)

diff --git a/apps/api/src/cora/infrastructure/record_export/__init__.py b/apps/api/src/cora/infrastructure/record_export/__init__.py
index 12630c31c7a..47ddca25f47 100644
--- a/apps/api/src/cora/infrastructure/record_export/__init__.py
+++ b/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",
diff --git a/apps/api/src/cora/infrastructure/record_export/_render.py b/apps/api/src/cora/infrastructure/record_export/_render.py
index 5a3a1aba472..e4e762082d8 100644
--- a/apps/api/src/cora/infrastructure/record_export/_render.py
+++ b/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
diff --git a/apps/api/tests/unit/infrastructure/record_export/test_render.py b/apps/api/tests/unit/infrastructure/record_export/test_render.py
index f6c35bd9e6d..a3ec296b1e4 100644
--- a/apps/api/tests/unit/infrastructure/record_export/test_render.py
+++ b/apps/api/tests/unit/infrastructure/record_export/test_render.py
@@ -7,7 +7,13 @@
 from datetime import UTC, datetime, timedelta, timezone
 from uuid import UUID
 
-from cora.infrastructure.record_export import render_row, render_value
+import pytest
+
+from cora.infrastructure.record_export import (
+    UndecodedJsonColumnError,
+    render_row,
+    render_value,
+)
 
 _SOME_UUID = UUID("12345678-1234-5678-1234-567812345678")
 
@@ -55,6 +61,46 @@ def test_jsonb_decoded_dict_and_list_pass_through_unrendered() -> None:
     assert render_value([1, "a", None]) == [1, "a", None]
 
 
+def test_render_row_refuses_a_payload_that_arrived_as_a_string() -> None:
+    """The shape a connection without jsonb codecs produces.
+
+    Measured against the live 2-BM database: a raw `asyncpg.connect`
+    hands `payload` back as text, and every later stage accepted it. The
+    bundle was written, hashed, and the standalone verifier reported OK
+    on a record whose payloads were strings, because the artifact was
+    self-consistent about the wrong structure.
+    """
+    row: dict[str, object] = {
+        "event_id": _SOME_UUID,
+        "event_type": "EnclosurePermitObserved",
+        "payload": '{"reason": "PSS permit observation", "trigger": "Monitor"}',
+    }
+
+    with pytest.raises(UndecodedJsonColumnError) as caught:
+        render_row(row)
+
+    assert caught.value.column == "payload"
+
+
+def test_render_row_refuses_metadata_that_arrived_as_a_string() -> None:
+    row: dict[str, object] = {
+        "event_id": _SOME_UUID,
+        "payload": {"a": 1},
+        "metadata": '{"command": "ObserveEnclosureStatus"}',
+    }
+
+    with pytest.raises(UndecodedJsonColumnError) as caught:
+        render_row(row)
+
+    assert caught.value.column == "metadata"
+
+
+def test_render_row_accepts_a_null_jsonb_column() -> None:
+    row: dict[str, object] = {"payload": {"a": 1}, "metadata": None}
+
+    assert render_row(row) == {"payload": {"a": 1}, "metadata": None}
+
+
 def test_render_row_applies_render_value_to_every_column() -> None:
     row: dict[str, object] = {
         "event_id": _SOME_UUID,

From ae39e5299eae91291cb6be30dc9586b403c43a88 Mon Sep 17 00:00:00 2001
From: xmap <16776958+xmap@users.noreply.github.com>
Date: Wed, 12 Aug 2026 07:21:09 -0500
Subject: [PATCH 3/5] Key the redaction table on what events are stored under

The table is looked up by `events.event_type`, but the generator keyed
it on the class name and a comment asserted those were the same thing.
They are not: the class is `ActorRegistered` and the append path writes
`"ActorRegisteredV2"`. Redaction refuses an event type it has no entry
for, and every database holds an Actor registration from bootstrap, so
the published-export path could not succeed against ANY real record.
A legacy record would have exported fine, since the V1 string was the
one in the table.

CI could not see it. The drift test compares the generator against its
own committed output, so a generator deriving the wrong key yields a
wrong table and a green test. It took running an export against the
live 2-BM database to surface it.

The generator now asks each module's own `event_type_name`, the same
function the append path calls, so the two cannot drift by
construction. Those functions discriminate on type, so an instance that
was never `__init__`ed answers correctly; one that reads a field aborts
the run rather than falling back to the class name.

The new fitness test deliberately does NOT reuse that route. It reads
the string literals those functions return, statically, without
importing anything or running the generator, because a guard sharing
the generator's mechanism would share its blind spot. Its second test
is a canary on the scanner: a scan that silently matched nothing would
pass forever.

One consequence stated rather than buried: the V1 string is no longer
in the table, so a pre-vault database carrying legacy `ActorRegistered`
rows now fails closed on export. That is the safe direction, since a V1
payload carries the `name` field the PII vault removed and publishing
it needs a deliberate disposition rather than an inherited one, but it
is a behaviour change and retired wire names have no home in a
generated table yet.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 .../record_export/_dispositions.py            |  2 +-
 ..._record_dispositions_cover_stored_names.py | 99 +++++++++++++++++++
 apps/api/tools/gen_record_dispositions.py     | 60 +++++++++--
 3 files changed, 154 insertions(+), 7 deletions(-)
 create mode 100644 apps/api/tests/architecture/test_record_dispositions_cover_stored_names.py

diff --git a/apps/api/src/cora/infrastructure/record_export/_dispositions.py b/apps/api/src/cora/infrastructure/record_export/_dispositions.py
index a1eef305e12..066542efd54 100644
--- a/apps/api/src/cora/infrastructure/record_export/_dispositions.py
+++ b/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",
diff --git a/apps/api/tests/architecture/test_record_dispositions_cover_stored_names.py b/apps/api/tests/architecture/test_record_dispositions_cover_stored_names.py
new file mode 100644
index 00000000000..c9d5a2e71b1
--- /dev/null
+++ b/apps/api/tests/architecture/test_record_dispositions_cover_stored_names.py
@@ -0,0 +1,99 @@
+"""Every name an event is STORED under has a disposition.
+
+The redaction table is keyed on `events.event_type`, and the generator
+builds it by asking each module's own `event_type_name` what that string
+is. This test asks the same question by a deliberately DIFFERENT route:
+it reads the string literals those functions return, straight from the
+source, without importing anything and without running the generator.
+
+The independence is the whole value. The pre-existing drift test
+compares the generator against its own committed output, so a generator
+that derives the wrong key produces a table that is wrong and a drift
+test that is green, which is exactly what happened: the table was keyed
+on class names, `ActorRegistered` writes `"ActorRegisteredV2"`, and the
+mismatch was invisible until a real export refused a real record. Every
+database holds an Actor registration from bootstrap, so the redacted
+export path was unreachable for every real record while CI stayed green.
+
+A static reader cannot be fooled by the same mistake as a runtime one.
+If these two ever disagree, one of them is wrong and this fails.
+
+Scope: only literal returns. `return type(event).__name__` names the
+class and the generator covers it by construction; a literal is the
+case where a stored name and a class name part company, and it is the
+case nothing else checks.
+
+Like the drift test beside it, this reads source files rather than
+importing them, so pytest-tach's impact analysis cannot see that every
+`events.py` is a dependency and `pytest --tach` would skip it after an
+event-only change. CI runs without that flag.
+"""
+
+import ast
+from pathlib import Path
+
+import pytest
+
+from cora.infrastructure.record_export._dispositions import DISPOSITIONS
+
+from .conftest import tracked_python_files
+
+pytestmark = pytest.mark.architecture
+
+
+def _event_type_name_literals(path: Path) -> frozenset[str]:
+    """String literals returned by `event_type_name` in one module."""
+    tree = ast.parse(path.read_text(encoding="utf-8"))
+    literals: set[str] = set()
+    for node in ast.walk(tree):
+        if not isinstance(node, ast.FunctionDef) or node.name != "event_type_name":
+            continue
+        for statement in ast.walk(node):
+            if not isinstance(statement, ast.Return):
+                continue
+            value = statement.value
+            if isinstance(value, ast.Constant) and isinstance(value.value, str):
+                literals.add(value.value)
+    return frozenset(literals)
+
+
+def _stored_names_by_module() -> dict[Path, frozenset[str]]:
+    found: dict[Path, frozenset[str]] = {}
+    for path in sorted(tracked_python_files()):
+        if path.name != "events.py":
+            continue
+        literals = _event_type_name_literals(path)
+        if literals:
+            found[path] = literals
+    return found
+
+
+def test_every_literal_stored_event_name_has_a_disposition() -> None:
+    missing: list[str] = []
+    for path, literals in _stored_names_by_module().items():
+        for name in sorted(literals):
+            if name not in DISPOSITIONS:
+                missing.append(f"{name} (returned by {path.name} in {path.parent.name})")
+
+    assert not missing, (
+        "These strings are written into events.event_type but have no entry in "
+        "DISPOSITIONS, so redaction refuses any record containing one and no "
+        "published export of such a record is possible:\n  "
+        + "\n  ".join(missing)
+        + "\nRegenerate with `make record-dispositions` and review the diff."
+    )
+
+
+def test_the_literal_scan_finds_the_known_renamed_event() -> None:
+    """A canary on the scanner itself.
+
+    A test that silently found nothing would pass forever. `ActorRegistered`
+    is the one event in the tree whose stored name differs from its class
+    name, so the scan must see it. If this fails because the rename was
+    retired, delete this test rather than weakening the one above.
+    """
+    all_literals: set[str] = set()
+    for literals in _stored_names_by_module().values():
+        all_literals |= literals
+
+    assert "ActorRegisteredV2" in all_literals
diff --git a/apps/api/tools/gen_record_dispositions.py b/apps/api/tools/gen_record_dispositions.py
index fe7e2c0d6c9..f7d7018e206 100644
--- a/apps/api/tools/gen_record_dispositions.py
+++ b/apps/api/tools/gen_record_dispositions.py
@@ -216,6 +216,52 @@ def _event_classes(module_name: str) -> Iterable[type]:
         yield obj
 
 
+def _wire_name(module_name: str, cls: type) -> str:
+    """The string this class is STORED under, asked of the real writer.
+
+    The table is keyed on `events.event_type`, and that is not reliably
+    the class name: `ActorRegistered` writes `"ActorRegisteredV2"`. A
+    generator that assumes the two agree produces a table the exporter
+    cannot look up, which is not a degraded export but no export at all,
+    since redaction refuses an unknown event type. Every real database
+    holds an Actor registration from bootstrap, so the redacted export
+    path was unreachable for every real record until this was fixed.
+
+    So the name is taken from each module's own `event_type_name`, the
+    same function the append path calls, rather than re-derived here.
+    Those functions discriminate on type alone, so an instance that was
+    never `__init__`ed answers correctly and no field values have to be
+    invented. If one ever reads a field, this raises and the run ABORTS,
+    matching how the tool treats an annotation it cannot classify: an
+    unanswerable question about the model, not a row to skip.
+    """
+    module = importlib.import_module(module_name)
+    resolve = getattr(module, "event_type_name", None)
+    if resolve is None:
+        raise RuntimeError(
+            f"{module_name} defines event classes but no `event_type_name`, so "
+            "the string they are stored under cannot be established. Add the "
+            "function, or the export table will key on a name that may not be "
+            "what the append path writes."
+        )
+    try:
+        name = resolve(object.__new__(cls))
+    except Exception as exc:
+        raise RuntimeError(
+            f"{module_name}.event_type_name failed on an uninitialised "
+            f"{cls.__name__} ({exc!r}). It reads a field rather than "
+            "discriminating on type, so this tool can no longer establish the "
+            "stored name without inventing values. Teach the tool about it "
+            "deliberately rather than falling back to the class name."
+        ) from exc
+    if not isinstance(name, str) or not name:
+        raise RuntimeError(
+            f"{module_name}.event_type_name returned {name!r} for "
+            f"{cls.__name__}; expected the non-empty string it is stored under."
+        )
+    return name
+
+
 def build_table(survey: bool = False) -> tuple[dict[str, dict[str, Any]], list[str]]:
     """Disposition per (event type, field) across every bounded context.
 
@@ -228,17 +274,19 @@ def build_table(survey: bool = False) -> tuple[dict[str, dict[str, Any]], list[s
     unclassified: list[str] = []
     for module_name in _event_modules():
         for cls in _event_classes(module_name):
-            if cls.__name__ in table:
+            wire_name = _wire_name(module_name, cls)
+            if wire_name in table:
                 raise RuntimeError(
-                    f"Duplicate event class name {cls.__name__!r}; the table is "
-                    "keyed on the bare name because that is what `events.event_type` "
-                    "stores. Rename one, or key on the qualified name."
+                    f"Duplicate stored event type {wire_name!r} (from class "
+                    f"{cls.__name__}); the table is keyed on what "
+                    "`events.event_type` stores. Rename one, or key on the "
+                    "qualified name."
                 )
             if not survey:
-                table[cls.__name__] = _resolve_fields(cls)
+                table[wire_name] = _resolve_fields(cls)
                 continue
             try:
-                table[cls.__name__] = _resolve_fields(cls)
+                table[wire_name] = _resolve_fields(cls)
             except UnclassifiedAnnotationError as exc:
                 unclassified.append(str(exc).split(".", 1)[0] + ": " + str(exc))
     return dict(sorted(table.items())), unclassified

From b1c51735df4c848622cd2468e23fe2d35d9bcba0 Mon Sep 17 00:00:00 2001
From: xmap <16776958+xmap@users.noreply.github.com>
Date: Wed, 12 Aug 2026 07:51:32 -0500
Subject: [PATCH 4/5] Let a deployment say which timestamp is the acquisition
 time

2-BM's scan files carry two, and the obvious one is wrong. Measured
across six consecutive files, every file's `start_date` equals the
PREVIOUS file's `end_date`, while every `end_date` falls within five
seconds of its own close. The areaDetector timestamp attribute
refreshes only while frames flow, so a file opened for scan N carries
the value left from scan N-1's last frame. The timestamp PV is healthy:
read live it returns the correct instant.

The reader only ever read `start_date`, and the ingest policy is that a
parseable file timestamp beats an operator's and supplying both is
refused as ambiguous. Together those would have recorded a scan watched
from end to end this morning as captured the previous evening, with no
way to correct it. That is a false fact in a record whose whole purpose
is to be true, so ingest at 2-BM stayed blocked rather than writing it.

Which timestamp to believe is a deployment fact, not a layout fact, so
the reader is told and reads what it is told. The default is unchanged,
`start_date`, so no other deployment moves. There is no fallback: a
deployment that declared `end_date` and got a file without one gets a
refusal, because silently reading the other timestamp would hand back
exactly the value the declaration exists to avoid.

`Description.start_date` becomes `captured_at`, since the field no
longer always holds a start date, and gains `captured_at_source` so the
record states which fact it used instead of leaving a reader to assume.
The policy itself is deliberately unchanged: a file value still beats
an operator's, because the fix for a bad writer is to declare the good
timestamp, not to let every caller assert over the file.

The descriptor records the choice for humans and the setting drives the
runtime, since nothing reads the descriptor at runtime yet. They agree
by hand for now, which the descriptor docstring says out loud.

Naming reviewed: `_CAPTURED_AT_SOURCES` made private to match every
sibling constant in the module.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 .../adapters/data_exchange_scan_reader.py     | 47 +++++++++++-
 .../cora/data/features/ingest_scan/handler.py | 34 ++++++---
 apps/api/src/cora/data/ports/scan_reader.py   | 18 ++++-
 apps/api/src/cora/data/wire.py                |  3 +-
 apps/api/src/cora/infrastructure/config.py    | 20 +++++
 .../data/test_data_exchange_scan_reader.py    | 73 +++++++++++++++++--
 .../unit/data/test_ingest_scan_handler.py     | 31 ++++++--
 deployments/2-bm/beamline.yaml                | 13 ++++
 scripts/beamline_descriptor.py                | 23 ++++--
 9 files changed, 222 insertions(+), 40 deletions(-)

diff --git a/apps/api/src/cora/data/adapters/data_exchange_scan_reader.py b/apps/api/src/cora/data/adapters/data_exchange_scan_reader.py
index 93ea16945f4..89b657b9046 100644
--- a/apps/api/src/cora/data/adapters/data_exchange_scan_reader.py
+++ b/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,
         )
diff --git a/apps/api/src/cora/data/features/ingest_scan/handler.py b/apps/api/src/cora/data/features/ingest_scan/handler.py
index caf57b8a25a..02f0b4dc0a4 100644
--- a/apps/api/src/cora/data/features/ingest_scan/handler.py
+++ b/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:
diff --git a/apps/api/src/cora/data/ports/scan_reader.py b/apps/api/src/cora/data/ports/scan_reader.py
index b18acbf82de..6f83964ce37 100644
--- a/apps/api/src/cora/data/ports/scan_reader.py
+++ b/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
 
diff --git a/apps/api/src/cora/data/wire.py b/apps/api/src/cora/data/wire.py
index 81421da7007..0c0a0e3b669 100644
--- a/apps/api/src/cora/data/wire.py
+++ b/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
diff --git a/apps/api/src/cora/infrastructure/config.py b/apps/api/src/cora/infrastructure/config.py
index 0b47d8ee365..32da2dcf89f 100644
--- a/apps/api/src/cora/infrastructure/config.py
+++ b/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`
diff --git a/apps/api/tests/unit/data/test_data_exchange_scan_reader.py b/apps/api/tests/unit/data/test_data_exchange_scan_reader.py
index 5276af76241..eb3d420aa26 100644
--- a/apps/api/tests/unit/data/test_data_exchange_scan_reader.py
+++ b/apps/api/tests/unit/data/test_data_exchange_scan_reader.py
@@ -48,6 +48,7 @@ def _write_scan(
     dark_mode: str | None = "Start",
     num_dark_fields: int | None = 2,
     start_date: str | None = "2026-07-29T10:15:30-05:00",
+    end_date: str | None = None,
     boxed_scalars: bool = False,
 ) -> None:
     def scalar(value: int | str) -> object:
@@ -81,10 +82,14 @@ def scalar(value: int | str) -> object:
             )
         if start_date is not None:
             f.create_dataset("process/acquisition/start_date", data=scalar(start_date))
+        if end_date is not None:
+            f.create_dataset("process/acquisition/end_date", data=scalar(end_date))
 
 
-def _reader(tmp_path: Path) -> DataExchangeScanReader:
-    return DataExchangeScanReader(allowed_roots=(str(tmp_path),))
+def _reader(tmp_path: Path, captured_at_source: str = "start_date") -> DataExchangeScanReader:
+    return DataExchangeScanReader(
+        allowed_roots=(str(tmp_path),), captured_at_source=captured_at_source
+    )
 
 
 async def test_describe_clean_scan_reports_complete_counts(tmp_path: Path) -> None:
@@ -126,7 +131,7 @@ async def test_describe_one_element_array_scalars_read_the_same_as_zero_d(tmp_pa
     assert result.commanded_flat_count == 2
     assert result.commanded_dark_count == 2
     assert result.dropped_frame_count == 0
-    assert result.start_date_raw == "2026-07-29T10:15:30-05:00"
+    assert result.captured_at_raw == "2026-07-29T10:15:30-05:00"
 
 
 async def test_describe_boxed_shortfall_reports_the_dropped_frames(tmp_path: Path) -> None:
@@ -161,6 +166,58 @@ async def test_describe_multi_element_array_where_a_scalar_belongs_reads_none(
     assert result.dropped_frame_count is None
 
 
+async def test_describe_reads_the_declared_timestamp_not_the_first_one(tmp_path: Path) -> None:
+    """The 2-BM case, reproduced in miniature.
+
+    A file whose start_date belongs to the previous scan and whose
+    end_date is its own. A deployment declaring end_date gets the right
+    instant, and the Description says which fact it used so a reader of
+    the record never has to assume.
+    """
+    scan = tmp_path / "scan_two_stamps.h5"
+    _write_scan(
+        scan,
+        start_date="2026-08-11T18:51:06-05:00",
+        end_date="2026-08-12T06:21:17-05:00",
+    )
+
+    from_start = await _reader(tmp_path).describe(locator_uri=scan.as_uri())
+    from_end = await _reader(tmp_path, "end_date").describe(locator_uri=scan.as_uri())
+
+    assert isinstance(from_start, Description)
+    assert isinstance(from_end, Description)
+    assert from_start.captured_at_raw == "2026-08-11T18:51:06-05:00"
+    assert from_start.captured_at_source == "start_date"
+    assert from_end.captured_at_raw == "2026-08-12T06:21:17-05:00"
+    assert from_end.captured_at_source == "end_date"
+
+
+async def test_describe_declared_timestamp_absent_reads_none_not_the_other_one(
+    tmp_path: Path,
+) -> None:
+    """No silent fallback.
+
+    A deployment that declared end_date and got a file without one has
+    a broken assumption, and the ingest refusing is how it finds out.
+    Falling back to start_date would hand back the exact wrong value
+    the declaration exists to avoid.
+    """
+    scan = tmp_path / "scan_no_end.h5"
+    _write_scan(scan, start_date="2026-08-11T18:51:06-05:00", end_date=None)
+
+    result = await _reader(tmp_path, "end_date").describe(locator_uri=scan.as_uri())
+
+    assert isinstance(result, Description)
+    assert result.captured_at is None
+    assert result.captured_at_raw is None
+    assert result.captured_at_source == "end_date"
+
+
+def test_reader_refuses_a_timestamp_the_layout_does_not_offer() -> None:
+    with pytest.raises(ValueError, match="not a timestamp this layout offers"):
+        DataExchangeScanReader(allowed_roots=("/tmp",), captured_at_source="acquired_on")
+
+
 async def test_describe_theta_absent_reads_structurally_incomplete(tmp_path: Path) -> None:
     scan = tmp_path / "scan_002.h5"
     _write_scan(scan, theta=False)
@@ -272,9 +329,9 @@ async def test_describe_aware_start_date_parses(tmp_path: Path) -> None:
     result = await _reader(tmp_path).describe(locator_uri=scan.as_uri())
 
     assert isinstance(result, Description)
-    assert result.start_date is not None
-    assert result.start_date.utcoffset() is not None
-    assert result.start_date_raw == "2026-07-29T10:15:30-05:00"
+    assert result.captured_at is not None
+    assert result.captured_at.utcoffset() is not None
+    assert result.captured_at_raw == "2026-07-29T10:15:30-05:00"
 
 
 async def test_describe_naive_start_date_stays_raw(tmp_path: Path) -> None:
@@ -286,8 +343,8 @@ async def test_describe_naive_start_date_stays_raw(tmp_path: Path) -> None:
     result = await _reader(tmp_path).describe(locator_uri=scan.as_uri())
 
     assert isinstance(result, Description)
-    assert result.start_date is None
-    assert result.start_date_raw == "2026-07-29T10:15:30"
+    assert result.captured_at is None
+    assert result.captured_at_raw == "2026-07-29T10:15:30"
 
 
 async def test_describe_non_hdf5_bytes_reads_unreadable(tmp_path: Path) -> None:
diff --git a/apps/api/tests/unit/data/test_ingest_scan_handler.py b/apps/api/tests/unit/data/test_ingest_scan_handler.py
index 62dd4f57176..229ac94f725 100644
--- a/apps/api/tests/unit/data/test_ingest_scan_handler.py
+++ b/apps/api/tests/unit/data/test_ingest_scan_handler.py
@@ -68,8 +68,9 @@ def _description(**overrides: object) -> Description:
         projection_angles_deg=(0.0, 45.0, 90.0, 135.0, 180.0),
         flat_angles_deg=None,
         dark_angles_deg=None,
-        start_date=datetime.fromisoformat(_AWARE_RAW),
-        start_date_raw=_AWARE_RAW,
+        captured_at=datetime.fromisoformat(_AWARE_RAW),
+        captured_at_raw=_AWARE_RAW,
+        captured_at_source="start_date",
         byte_size=4096,
         mtime_ns=111,
     )
@@ -229,7 +230,8 @@ async def test_ingest_incomplete_file_refusal_leaves_zero_events() -> None:
 async def test_ingest_timestampless_file_without_operator_value_refuses() -> None:
     store = InMemoryEventStore()
     handler = _bind(
-        _deps(store), described=_description(start_date=None, start_date_raw="2026-07-29T10:15:30")
+        _deps(store),
+        described=_description(captured_at=None, captured_at_raw="2026-07-29T10:15:30"),
     )
 
     with pytest.raises(InvalidScanFileError, match="captured_at"):
@@ -241,7 +243,8 @@ async def test_ingest_timestampless_file_without_operator_value_refuses() -> Non
 async def test_ingest_timestampless_file_accepts_operator_captured_at() -> None:
     store = InMemoryEventStore()
     handler = _bind(
-        _deps(store), described=_description(start_date=None, start_date_raw="2026-07-29T10:15:30")
+        _deps(store),
+        described=_description(captured_at=None, captured_at_raw="2026-07-29T10:15:30"),
     )
     operator_time = datetime(2026, 7, 29, 10, 20, 0, tzinfo=UTC)
 
@@ -255,7 +258,25 @@ async def test_ingest_timestampless_file_accepts_operator_captured_at() -> None:
     payload = events[0].payload
     assert payload["captured_at"] == operator_time.isoformat()
     assert payload["evidence"]["captured_at_source"] == "operator"
-    assert payload["evidence"]["start_date_raw"] == "2026-07-29T10:15:30"
+    assert payload["evidence"]["captured_at_raw"] == "2026-07-29T10:15:30"
+
+
+async def test_ingest_records_the_source_the_reader_used_not_a_fixed_name() -> None:
+    """The record says WHICH timestamp it believed.
+
+    A deployment whose writer emits a bad `start_date` declares another
+    source, and the record has to carry that rather than a hardcoded
+    label, or a reader cannot tell which fact the capture time came
+    from. This is the 2-BM posture: `end_date`, because `start_date`
+    there is measurably the previous scan's.
+    """
+    store = InMemoryEventStore()
+    handler = _bind(_deps(store), described=_description(captured_at_source="end_date"))
+
+    await handler(_command(), principal_id=_PRINCIPAL_ID, correlation_id=_CORRELATION_ID)
+
+    events, _ = await store.load("Acquisition", _ACQUISITION_ID)
+    assert events[0].payload["evidence"]["captured_at_source"] == "end_date"
 
 
 async def test_ingest_operator_value_alongside_file_timestamp_refuses_as_ambiguous() -> None:
diff --git a/deployments/2-bm/beamline.yaml b/deployments/2-bm/beamline.yaml
index ceccd41884c..82abf0de95b 100644
--- a/deployments/2-bm/beamline.yaml
+++ b/deployments/2-bm/beamline.yaml
@@ -795,5 +795,18 @@ data:
     independently. The transport axis (mount vs fetch) is deliberately
     NOT declared: it is the open HOST-2 staff question, and the
     descriptor records answers, not assumptions.
+
+    The acquisition time is read from end_date, not start_date, and
+    that is a measurement rather than a preference. Across six
+    consecutive files written 2026-08-05 to 2026-08-12, every file's
+    start_date equalled the PREVIOUS file's end_date, while every
+    file's end_date fell within five seconds of its own close. The
+    areaDetector timestamp attribute refreshes only while frames flow,
+    so a file opened for scan N carries the value left from scan N-1's
+    last frame. The timestamp PV itself is healthy: read live it
+    returns the correct instant. Until the IOC's attribute
+    configuration is corrected upstream, end_date is the only
+    trustworthy instant the file carries.
   layout: data_exchange
   finality: transfer_status
+  captured_at_source: end_date
diff --git a/scripts/beamline_descriptor.py b/scripts/beamline_descriptor.py
index 1a1f05c08d5..9539025ffd2 100644
--- a/scripts/beamline_descriptor.py
+++ b/scripts/beamline_descriptor.py
@@ -321,13 +321,21 @@ class Resources(BaseModel):
 class Data(BaseModel):
     """The beamline's scan-data choices on the ingest seam's axes.
 
-    Declares VERIFIED facts only (`layout`, `finality`); an open
-    assumption stays out of the descriptor rather than in it. Nothing
-    at runtime reads this section yet: the adapters are wired
-    unconditionally while one layout exists, and the first consumer is
-    the discovery slice (or a second layout adapter). Declaring it now
-    records the per-beamline choice where the docs and a future seeder
-    read from, without building selection machinery ahead of need.
+    Declares VERIFIED facts only (`layout`, `finality`,
+    `captured_at_source`); an open assumption stays out of the
+    descriptor rather than in it. Nothing at runtime reads this section
+    yet: the adapters are wired unconditionally while one layout
+    exists, and the first consumer is the discovery slice (or a second
+    layout adapter). Declaring it here records the per-beamline choice
+    where the docs and a future seeder read from, without building
+    selection machinery ahead of need.
+
+    `captured_at_source` is the one whose runtime twin already exists,
+    as the `SCAN_CAPTURED_AT_SOURCE` setting, because a wrong
+    acquisition time is a wrong fact in the record rather than a
+    missing feature. Until the descriptor is read at runtime the two
+    have to agree by hand, so a deployment declaring it here must set
+    the environment variable to match.
     """
 
     model_config = _MODEL_CONFIG
@@ -335,6 +343,7 @@ class Data(BaseModel):
     intro: str | None = None
     layout: str
     finality: str
+    captured_at_source: str | None = None
 
 
 @dataclass(frozen=True)

From 4f1e20fa72acea712282095305ba5ff99852bca7 Mon Sep 17 00:00:00 2001
From: xmap <16776958+xmap@users.noreply.github.com>
Date: Wed, 12 Aug 2026 08:10:52 -0500
Subject: [PATCH 5/5] Cover the scalar reader's refusal branches

The diff-coverage gate counts refusal paths, and reworking `_scalar_int`
added three it had no fixture for: a count stored as text, a count
stored as text inside a one-element array, and a value that is neither
a number nor a sized thing.

Worth testing rather than waving through. The first two are the same
decode the timestamp path already does, so a count the reader could
have read but silently dropped would be this commit's parent bug in a
different costume. The third pins the never-raise contract: an
unreadable count reports unknowable rather than escaping the worker
thread as a failed describe.

Real HDF5 files rather than mocks, since the whole lesson of the day is
that a fixture built to match the reader proves nothing about a file
built by a writer.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 .../data/test_data_exchange_scan_reader.py    | 50 +++++++++++++++++++
 1 file changed, 50 insertions(+)

diff --git a/apps/api/tests/unit/data/test_data_exchange_scan_reader.py b/apps/api/tests/unit/data/test_data_exchange_scan_reader.py
index eb3d420aa26..5afcddcc741 100644
--- a/apps/api/tests/unit/data/test_data_exchange_scan_reader.py
+++ b/apps/api/tests/unit/data/test_data_exchange_scan_reader.py
@@ -145,6 +145,56 @@ async def test_describe_boxed_shortfall_reports_the_dropped_frames(tmp_path: Pat
     assert result.dropped_frame_count == 2
 
 
+@pytest.mark.parametrize(
+    ("written", "expected"),
+    [
+        pytest.param("1501", 1501, id="text_scalar"),
+        pytest.param(["1501"], 1501, id="text_in_a_one_element_array"),
+    ],
+)
+async def test_describe_reads_a_commanded_count_written_as_text(
+    tmp_path: Path, written: object, expected: int
+) -> None:
+    """h5py hands string datasets back as bytes.
+
+    A writer that stores a count as text is not a shape this beamline
+    uses today, but the reader already decodes bytes for the timestamp
+    and the same value can arrive on either path; a count it could
+    decode but silently dropped would be the `_scalar_int` bug again in
+    a different costume.
+    """
+    scan = tmp_path / "scan_text_count.h5"
+    _write_scan(scan, projections=1501)
+    with h5py.File(scan, "a") as f:
+        del f["process/acquisition/rotation/num_angles"]
+        f.create_dataset("process/acquisition/rotation/num_angles", data=written)
+
+    result = await _reader(tmp_path).describe(locator_uri=scan.as_uri())
+
+    assert isinstance(result, Description)
+    assert result.commanded_projection_count == expected
+
+
+async def test_describe_uncountable_commanded_value_reads_none(tmp_path: Path) -> None:
+    """A value that is neither a number nor a sized thing.
+
+    The reader's contract is never to raise, so an unreadable count
+    reports unknowable rather than propagating out of the worker
+    thread as a failed describe.
+    """
+    scan = tmp_path / "scan_nan_count.h5"
+    _write_scan(scan)
+    with h5py.File(scan, "a") as f:
+        del f["process/acquisition/rotation/num_angles"]
+        f.create_dataset("process/acquisition/rotation/num_angles", data=float("nan"))
+
+    result = await _reader(tmp_path).describe(locator_uri=scan.as_uri())
+
+    assert isinstance(result, Description)
+    assert result.commanded_projection_count is None
+    assert result.dropped_frame_count is None
+
+
 async def test_describe_multi_element_array_where_a_scalar_belongs_reads_none(
     tmp_path: Path,
 ) -> None: