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..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, ) @@ -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 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/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/_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/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/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/tests/unit/data/test_data_exchange_scan_reader.py b/apps/api/tests/unit/data/test_data_exchange_scan_reader.py index c0cb54c5d27..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 @@ -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,12 @@ 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: + 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,25 +63,33 @@ 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)) + 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: @@ -87,6 +112,162 @@ 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.captured_at_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 + + +@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: + """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_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) @@ -198,9 +379,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: @@ -212,8 +393,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/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, 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 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)