diff --git a/apps/api/openapi.json b/apps/api/openapi.json index 1e8992d03b..472c23de8b 100644 --- a/apps/api/openapi.json +++ b/apps/api/openapi.json @@ -14273,7 +14273,7 @@ "type": "object" }, "RunResponse": { - "description": "Read-side DTO at the API boundary.\n\nCarries primitives, not domain VOs. `status` is the StrEnum's\nstring value. `subject_id` is null for calibration / dark-field\nruns. `raid` is null when no Research Activity Identifier was\nsupplied at start time (additive retrofit).\n\n`override_parameters` and `effective_parameters` carry the\nparameter set: overrides the operator supplied at start\ntime, and the resolved merge of Plan defaults + overrides that\nactually governed this Run. Both default `{}`. `trigger_source`\ncaptures what initiated the Run (None if unrecorded).\n\n`campaign_id` (6i-c) is the Campaign this Run is a member of, set\neither at start time (StartRun.campaign_id) or post-hoc via\nadd_run_to_campaign. None when the Run is standalone (not part of\nany Campaign). Closes design-memo Watch #17 (per Caution-design\ncross-BC consistency precedent).\n\n`capture_code` (slice 13) is the deployment-declared capture\nidentifier a witnessed genesis stamps onto `external_refs`. None\nfor a Conducted Run. NOT personal data.\n\n`observed_capture_path` (slice 13) is the areaDetector file the\ncapture wrote, resolved from the `run_capture_path` PII vault:\n`None` when `capture_code` is `None` (not applicable, a Conducted\nRun); the tombstone literal (`UNOBSERVED_CAPTURE_PATH`) when a\ncapture code exists but the vault has no row yet (never observed,\nor rejected by the dual-clock guard); the real path otherwise.\nThis IS the authorized operator surface meant to let them find the\nfile for `ingest_scan`.", + "description": "Read-side DTO at the API boundary.\n\nCarries primitives, not domain VOs. `status` is the StrEnum's\nstring value. `subject_id` is null for calibration / dark-field\nruns. `raid` is null when no Research Activity Identifier was\nsupplied at start time (additive retrofit).\n\n`override_parameters` and `effective_parameters` carry the\nparameter set: overrides the operator supplied at start\ntime, and the resolved merge of Plan defaults + overrides that\nactually governed this Run. Both default `{}`. `trigger_source`\ncaptures what initiated the Run (None if unrecorded).\n\n`campaign_id` (6i-c) is the Campaign this Run is a member of, set\neither at start time (StartRun.campaign_id) or post-hoc via\nadd_run_to_campaign. None when the Run is standalone (not part of\nany Campaign). Closes design-memo Watch #17 (per Caution-design\ncross-BC consistency precedent).\n\n`capture_code` (slice 13) is the deployment-declared capture\nidentifier a witnessed genesis stamps onto `external_refs`. None\nfor a Conducted Run. NOT personal data.\n\n`observed_capture_path` (slice 13) is the areaDetector file the\ncapture wrote, resolved from the `run_capture_path` PII vault:\n`None` when `capture_code` is `None` (not applicable, a Conducted\nRun); the tombstone literal (`UNOBSERVED_CAPTURE_PATH`) when a\ncapture code exists but the vault has no row yet (never observed,\nor rejected by the dual-clock guard); the real path otherwise.\nThis IS the authorized operator surface meant to let them find the\nfile for `ingest_scan`.\n\n`proposal_number` / `esaf_number` / `esaf_doi_number` (slice 14a), each\npaired with its own `*_observed_at` (the substrate's own reading\ntime, for judging staleness -- these PVs persist across beamtimes\nwith no in-band freshness signal), resolve from the\n`run_experiment_identity` vault under the same `capture_code is\nnot None` condition. No tombstone: `None` here means either \"not\napplicable\" (Conducted Run) or \"nothing recorded yet\"; `capture_code`\nalready distinguishes the two. Institutional identifiers for a\nfunded experiment, not personal data.", "properties": { "campaign_id": { "anyOf": [ @@ -14303,6 +14303,52 @@ "title": "Effective Parameters", "type": "object" }, + "esaf_doi_number": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Esaf Doi Number" + }, + "esaf_doi_number_observed_at": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Esaf Doi Number Observed At" + }, + "esaf_number": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Esaf Number" + }, + "esaf_number_observed_at": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Esaf Number Observed At" + }, "id": { "format": "uuid", "title": "Id", @@ -14334,6 +14380,29 @@ "title": "Plan Id", "type": "string" }, + "proposal_number": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Proposal Number" + }, + "proposal_number_observed_at": { + "anyOf": [ + { + "format": "date-time", + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Proposal Number Observed At" + }, "raid": { "anyOf": [ { diff --git a/apps/api/src/cora/api/_capture_experiment_identity_reader.py b/apps/api/src/cora/api/_capture_experiment_identity_reader.py new file mode 100644 index 0000000000..6e0a0eb3f2 --- /dev/null +++ b/apps/api/src/cora/api/_capture_experiment_identity_reader.py @@ -0,0 +1,280 @@ +"""CaptureExperimentIdentityReader: read a witnessed Run's proposal / ESAF / +ESAF-DOI PVs once, at the instant a capture promotes to a Run. + +Slice 14a. Mirrors `_capture_baseline_reader.py`'s ONE-READ-NOT-A-FEED +shape exactly: invoked exactly once per promotion by +`RunWitnessRecorder._promote`, right after `record_witnessed_run` +returns a `run_id`, alongside (not instead of) the genesis-baseline +read. There is no buffer, no tick, and no ongoing liveness claim. + +## Why the vault, not the event + +See `cora.run.aggregates.run.experiment_identity`'s module docstring +for the full argument (memory/project_witnessed_run_prelive_slices.md, +slice 14a). Short version: `RecordWitnessedRun` has no operator behind +it the way `start_run.external_refs` does, so writing these PVs onto +`RunStarted` would be CORA auto-harvesting a re-identifying fact (a +proposal number plus a timestamp, per D0) into an immutable, +INSERT-only event with no way back. `ESAFDOINumber` was checked and +traced to an internal, authenticated APS API +(`EsafApsDbApi.getStationEsafById`, via the upstream `dmagic` source), +not a DOI registration agency; unconfirmed as a genuinely resolvable +public identifier, so it vaults alongside the other two. + +## Two traps, both silently recording a WRONG fact rather than failing + +1. Every one of these PVs defaults to the substrate literal `"Unknown"` + when `dmagic` has not populated it. An unpopulated PV therefore + reads as a plausible string. `resolved_experiment_identity_text` treats + `"Unknown"`, and an empty string, as ABSENT and returns `None`; the + caller never writes a literal "Unknown" into the vault. +2. Nothing in the IOC populates these PVs; `dmagic` does, from APS + scheduling, so a value PERSISTS ACROSS BEAMTIMES until the next + sync overwrites it. If a value is stale, this reader cannot detect + that -- there is no freshness heuristic to invent, per the design + memo's own instruction. Each `*_observed_at` carries the substrate's + own reading time (`Measurement.produced_at`), the only staleness + evidence available, so a reader downstream can at least see how old + a value is. Whether these PVs are reliably synced at 2-BM before a + beamtime starts is a staff question, not a code question. + +## Per-PV failure posture mirrors `_capture_baseline_reader.py` + +Every exception is caught and logged, never raised into the caller: a +`ControlPort.read()` failure on one PV drops only that PV's reading and +lets the sweep continue over the rest (mirroring +`capture_watch_preflight.py`'s own per-PV independence), and the vault +write's own failure must never unwind or retry the promotion that +already committed (mirroring `_read_baseline`'s exact posture in +`_run_witness.py`). + +Unlike `CaptureBaselineReader`, none of these three values is personal +data, so a write failure's exception text is logged in full (no +`error_class`-only redaction the way `_write_capture_path` requires). +""" + +from __future__ import annotations + +import asyncio +from typing import TYPE_CHECKING + +from cora.infrastructure.logging import get_logger +from cora.operation.ports.control_port import ( + ControlAccessDeniedError, + ControlNotConnectedError, + ControlTimeoutError, + ControlValueCoercionError, +) + +if TYPE_CHECKING: + from collections.abc import Mapping + from datetime import datetime + from uuid import UUID + + from cora.infrastructure.kernel import Kernel + from cora.operation.ports.control_port import ControlPort, Measurement + from cora.run.aggregates.run import ExperimentIdentityStore + +ROLE_PROPOSAL_NUMBER = "proposal_number" +ROLE_ESAF_NUMBER = "esaf_number" +ROLE_ESAF_DOI_NUMBER = "esaf_doi_number" +"""CORA-owned role keys, matching `Settings.capture_experiment_identity_pvs`'s +closed vocabulary. Module-public (not `_`-prefixed): `capture_watch_preflight` +dispatches its own sweep on these same three keys, so a rename here +cannot silently desync from it.""" + +_ROLES: tuple[str, ...] = (ROLE_PROPOSAL_NUMBER, ROLE_ESAF_NUMBER, ROLE_ESAF_DOI_NUMBER) + +UNKNOWN_EXPERIMENT_IDENTITY_LITERAL = "Unknown" +"""The literal `dmagic` / the IOC leaves an unpopulated experiment-identity +PV reading. Treated as ABSENT, never as a plausible value (Trap 1). +Module-public: `capture_watch_preflight` imports this directly (alongside +`resolved_experiment_identity_text`) so its own decode verdict can distinguish +"substrate's own placeholder" from "genuinely empty" without a second +copy of the literal.""" + +_log = get_logger(__name__) + + +def resolved_experiment_identity_text(value: object) -> str | None: + """Resolve one experiment-identity PV reading to a usable string, or + `None` when it must be treated as absent. + + `None` for: a non-string reading, an empty (after stripping) + string, or the substrate's own `"Unknown"` placeholder literal + (matched case-insensitively: the exact casing is an IOC default we + have observed, not a wire-level guarantee, and a differently-cased + variant is still the same "not populated" fact, not a real value). + Otherwise the stripped string. Module-public: `capture_watch_preflight` + imports this directly so its own decode verdict can never drift from + what this reader actually accepts. + """ + if not isinstance(value, str): + return None + stripped = value.strip() + if not stripped or stripped.casefold() == UNKNOWN_EXPERIMENT_IDENTITY_LITERAL.casefold(): + return None + return stripped + + +class CaptureExperimentIdentityReader: + """Reads `experiment_identity_pvs[capture_code]`'s three roles once and vaults + whatever survives. + + `experiment_identity_pvs` is code -> role -> PV, matching + `Settings.capture_experiment_identity_pvs`. A code with no entry (or + an empty one) makes `read` a no-op, mirroring `CaptureBaselineReader`'s + own per-code optionality. A role absent from a code's declared set is + simply never read; it does not stop the other two. + """ + + def __init__( + self, + *, + deps: Kernel, + control_port: ControlPort, + experiment_identity_pvs: Mapping[str, Mapping[str, str]], + store: ExperimentIdentityStore, + ) -> None: + self._deps = deps + self._control_port = control_port + self._experiment_identity_pvs = experiment_identity_pvs + self._store = store + + async def read(self, capture_code: str, run_id: UUID) -> None: + """Read every role declared for `capture_code` CONCURRENTLY, + once, and vault whatever survives as one row against `run_id`. + + Concurrent, not sequential: mirrors `CaptureBaselineReader.read`'s + reasoning exactly -- this runs inline inside + `RunWitnessRecorder._promote`, on `run_witness_loop`'s single + consumer path, so a slow or partially-unreachable control system + must not block the loop from reacting to the next observation. + + Never raises: every failure mode (a dead PV, an unusable + reading, or the vault write itself) is caught and logged here. + """ + roles = self._experiment_identity_pvs.get(capture_code) + if not roles: + return + + ( + (proposal_number, proposal_number_observed_at), + (esaf_number, esaf_number_observed_at), + (esaf_doi_number, esaf_doi_number_observed_at), + ) = await asyncio.gather( + *(self._read_one(capture_code, role, roles.get(role)) for role in _ROLES) + ) + + if proposal_number is None and esaf_number is None and esaf_doi_number is None: + _log.info( + "capture_experiment_identity.nothing_to_record", + capture_code=capture_code, + run_id=str(run_id), + ) + return + + try: + await self._store.upsert( + run_id=run_id, + proposal_number=proposal_number, + proposal_number_observed_at=proposal_number_observed_at, + esaf_number=esaf_number, + esaf_number_observed_at=esaf_number_observed_at, + esaf_doi_number=esaf_doi_number, + esaf_doi_number_observed_at=esaf_doi_number_observed_at, + created_at=self._deps.clock.now(), + ) + except asyncio.CancelledError: + raise + except Exception: + # Unlike `_write_capture_path`, none of these three values + # is personal data, so the full exception (which could + # include the failing row's values in an asyncpg DETAIL + # line) is safe to log. + _log.exception( + "capture_experiment_identity.vault_write_failed", + capture_code=capture_code, + run_id=str(run_id), + ) + return + _log.info( + "capture_experiment_identity.recorded", + capture_code=capture_code, + run_id=str(run_id), + ) + + async def _read_one( + self, capture_code: str, role: str, pv: str | None + ) -> tuple[str | None, datetime | None]: + """One role's reading, or `(None, None)` when the role is + undeclared for this code, unreachable, unusable, or resolves to + `resolved_experiment_identity_text`'s absent case (Trap 1).""" + if pv is None: + return None, None + try: + reading = await self._control_port.read(pv) + except asyncio.CancelledError: + raise + except (ControlNotConnectedError, ControlTimeoutError, ControlAccessDeniedError) as exc: + _log.warning( + "capture_experiment_identity.read_unreachable", + capture_code=capture_code, + role=role, + pv=pv, + detail=str(exc), + ) + return None, None + except ControlValueCoercionError as exc: + _log.warning( + "capture_experiment_identity.read_uncoercible", + capture_code=capture_code, + role=role, + pv=pv, + detail=str(exc), + ) + return None, None + except Exception: + _log.exception( + "capture_experiment_identity.read_failed", + capture_code=capture_code, + role=role, + pv=pv, + ) + return None, None + return self._to_value(capture_code, role, pv, reading) + + def _to_value( + self, capture_code: str, role: str, pv: str, reading: Measurement + ) -> tuple[str | None, datetime | None]: + if reading.produced_at is None: + # The port's dual-clock rule forbids substituting CORA's own + # clock for an absent substrate time (Trap 2: there would be + # no honest staleness evidence to carry). + _log.info( + "capture_experiment_identity.no_substrate_time", + capture_code=capture_code, + role=role, + pv=pv, + ) + return None, None + value = resolved_experiment_identity_text(reading.value) + if value is None: + _log.info( + "capture_experiment_identity.absent_reading", + capture_code=capture_code, + role=role, + pv=pv, + ) + return None, None + return value, reading.produced_at + + +__all__ = [ + "ROLE_ESAF_DOI_NUMBER", + "ROLE_ESAF_NUMBER", + "ROLE_PROPOSAL_NUMBER", + "UNKNOWN_EXPERIMENT_IDENTITY_LITERAL", + "CaptureExperimentIdentityReader", + "resolved_experiment_identity_text", +] diff --git a/apps/api/src/cora/api/_run_witness.py b/apps/api/src/cora/api/_run_witness.py index ecfbffbf2b..d1c5374adb 100644 --- a/apps/api/src/cora/api/_run_witness.py +++ b/apps/api/src/cora/api/_run_witness.py @@ -46,7 +46,10 @@ call the configured `CaptureBaselineReader` (slice 12) exactly once to snapshot the genesis-baseline PVs against the new Run; a failure there is logged and never unwinds the promotion that already - committed (see `RunWitnessRecorder._read_baseline`). + committed (see `RunWitnessRecorder._read_baseline`). Also calls the + configured `CaptureExperimentIdentityReader` (slice 14a) exactly once, same + posture, to vault the proposal / ESAF / ESAF-DOI PVs against the new + Run (see `RunWitnessRecorder._read_experiment_identity`). - `BEGUN` while a Run is already open for this code: the previous terminal was missed (dropped CA transition, or the substrate restarted mid-capture). `TruncateRun` the stale Run first @@ -253,6 +256,7 @@ from cora.agent.seed_capture_progress_feeder import CAPTURE_PROGRESS_FEEDER_AGENT_ID from cora.agent.seed_run_witness import RUN_WITNESS_AGENT_ID from cora.api._capture_baseline_reader import CaptureBaselineReader +from cora.api._capture_experiment_identity_reader import CaptureExperimentIdentityReader from cora.api._capture_observer import ROLE_IMAGES_COLLECTED, ROLE_IMAGES_SAVED from cora.api._capture_progress_feeder import CaptureProgressFeeder, capture_progress_flush_loop from cora.infrastructure.logging import get_logger @@ -283,7 +287,11 @@ from cora.infrastructure.config import Settings from cora.infrastructure.kernel import Kernel from cora.operation.ports.control_port import ControlPort - from cora.run.aggregates.run import CapturePathStore, FeedHeartbeatStore + from cora.run.aggregates.run import ( + CapturePathStore, + ExperimentIdentityStore, + FeedHeartbeatStore, + ) from cora.run.aggregates.run.state import Run from cora.run.features.append_observations.handler import Handler as AppendObservationsHandler from cora.run.features.list_runs.handler import Handler as ListRunsHandler @@ -374,6 +382,7 @@ def __init__( open_captures: dict[str, UUID] | None = None, baseline_reader: CaptureBaselineReader | None = None, capture_path_store: CapturePathStore | None = None, + experiment_identity_reader: CaptureExperimentIdentityReader | None = None, ) -> None: self._deps = deps self._record_witnessed_run = record_witnessed_run @@ -385,6 +394,7 @@ def __init__( self._last_precondition_bypass: dict[str, CapturePreconditionBypassObservation] = {} self._baseline_reader = baseline_reader self._capture_path_store = capture_path_store + self._experiment_identity_reader = experiment_identity_reader self._last_capture_path: dict[str, CapturePathObservation] = {} """Slice 13: the latest `full_file_name` reading retained per capture_code, mirroring `_last_progress`'s retain-latest shape. @@ -635,7 +645,18 @@ async def _promote(self, observation: CaptureLifecycleObservation) -> None: capture_code=observation.capture_code, run_id=str(run_id), ) - await self._read_baseline(observation.capture_code, run_id) + # Concurrent, not sequential: both readers do their own PV + # sweep and neither depends on the other's result, so awaiting + # them back to back would double this loop's stall time for no + # correctness benefit -- exactly what this single-consumer path + # must not do (see this method's own "must not block" framing + # elsewhere in this module). Both readers already catch every + # failure internally, so a plain gather needs no + # return_exceptions. + await asyncio.gather( + self._read_baseline(observation.capture_code, run_id), + self._read_experiment_identity(observation.capture_code, run_id), + ) async def _read_baseline(self, capture_code: str, run_id: UUID) -> None: """Slice 12: read the genesis-baseline PVs once, right after a @@ -665,6 +686,37 @@ async def _read_baseline(self, capture_code: str, run_id: UUID) -> None: run_id=str(run_id), ) + async def _read_experiment_identity(self, capture_code: str, run_id: UUID) -> None: + """Slice 14a: vault the proposal / ESAF / ESAF-DOI PVs once, + right after a successful promotion. + + Same posture as `_read_baseline`: the promotion has already + fully committed by this point, so a read/vault failure must + never unwind or retry it. Gated on BOTH a reader being + configured (main.py wires one whenever + `capture_experiment_identity_pvs` is declared) and the sixth + kill switch, `capture_experiment_identity_recording_enabled`; + `CaptureExperimentIdentityReader` itself catches every failure + internally (see its own module docstring), the outer + try/except here is defense in depth, mirroring + `_read_baseline`'s identical wrapper. + """ + if ( + self._experiment_identity_reader is None + or not self._settings.capture_experiment_identity_recording_enabled + ): + return + try: + await self._experiment_identity_reader.read(capture_code, run_id) + except asyncio.CancelledError: + raise + except Exception: + _log.exception( + "run_witness.experiment_identity_read_failed", + capture_code=capture_code, + run_id=str(run_id), + ) + async def _truncate_stale(self, observation: CaptureLifecycleObservation) -> None: code = observation.capture_code # Pop unconditionally, before attempting the truncate: the new @@ -1041,6 +1093,8 @@ async def run_witness_lifespan( control_port: ControlPort | None = None, capture_baseline_pvs: Mapping[str, Mapping[str, str]] | None = None, capture_path_store: CapturePathStore | None = None, + capture_experiment_identity_pvs: Mapping[str, Mapping[str, str]] | None = None, + experiment_identity_store: ExperimentIdentityStore | None = None, ) -> AsyncGenerator[None]: """Run the watcher as a background task for the app's lifetime. @@ -1083,6 +1137,18 @@ async def run_witness_lifespan( actually happens is gated, same pattern as the fourth switch, inside the recorder by `deps.settings.capture_path_recording_enabled` (the fifth kill switch). + + A non-empty `capture_experiment_identity_pvs` (slice 14a) additionally + requires `record_witnessed_run`, `control_port`, and + `experiment_identity_store`: a `CaptureExperimentIdentityReader` is built + and handed to the recorder, mirroring `capture_baseline_pvs`'s exact + shape (a separate reader object, unlike `capture_path_store`'s + handed-straight-through style, because this reader does its own + `ControlPort.read()` calls rather than consuming an already-pumped + observation). Whether a call to it actually reads and vaults + anything is gated separately, inside the recorder, by + `deps.settings.capture_experiment_identity_recording_enabled` (the + sixth kill switch). """ if not capture_codes: yield @@ -1128,6 +1194,35 @@ async def run_witness_lifespan( principal_id=CAPTURE_BASELINE_READER_AGENT_ID, ) + experiment_identity_reader: CaptureExperimentIdentityReader | None = None + if capture_experiment_identity_pvs: + missing = [ + name + for name, value in ( + ("record_witnessed_run", record_witnessed_run), + ("control_port", control_port), + ("experiment_identity_store", experiment_identity_store), + ) + if value is None + ] + if missing: + msg = ( + "run_witness_lifespan: capture_experiment_identity_pvs requires " + f"{', '.join(missing)}" + ) + raise ValueError(msg) + # Narrowed by the checks above; deps is not None because + # record_witnessed_run is not None (see the first check above). + assert deps is not None + assert control_port is not None + assert experiment_identity_store is not None + experiment_identity_reader = CaptureExperimentIdentityReader( + deps=deps, + control_port=control_port, + experiment_identity_pvs=capture_experiment_identity_pvs, + store=experiment_identity_store, + ) + recorder: RunWitnessRecorder | None = None if record_witnessed_run is not None: # Narrowed by the check above. @@ -1143,6 +1238,7 @@ async def run_witness_lifespan( open_captures=open_captures, baseline_reader=baseline_reader, capture_path_store=capture_path_store, + experiment_identity_reader=experiment_identity_reader, ) feeder: CaptureProgressFeeder | None = None diff --git a/apps/api/src/cora/api/capture_watch_preflight.py b/apps/api/src/cora/api/capture_watch_preflight.py index d1a7ac35f2..178f8bed59 100644 --- a/apps/api/src/cora/api/capture_watch_preflight.py +++ b/apps/api/src/cora/api/capture_watch_preflight.py @@ -1,6 +1,6 @@ -"""Preflight read: report every configured `capture_watch_pvs` PV, and -every `capture_baseline_pvs` PV, against the live control system, before -the recording switch flips. +"""Preflight read: report every configured `capture_watch_pvs` PV, +every `capture_baseline_pvs` PV, and every `capture_experiment_identity_pvs` +PV, against the live control system, before the recording switch flips. `python -m cora.api.capture_watch_preflight` reads every PV named under `Settings.capture_watch_pvs` exactly once and reports, per role: whether it @@ -12,9 +12,17 @@ `kind` / `value` / `units` with verdict `n/a`, EXCEPT that a non-numeric value is flagged BAD -- the one thing checkable ahead of time, since `Observation.value` is `float` and a textual reading would be rejected at -append time anyway. Run this command once the host is reachable and before +append time anyway. It also sweeps `Settings.capture_experiment_identity_pvs` +(slice 14a): the one thing worth flagging ahead of time here is that the +substrate's own `"Unknown"` placeholder reads as a perfectly healthy +string unless this preflight calls it out explicitly, so the verdict +column shows `placeholder` (distinct from `empty` and `text(len=N)`) rather +than letting an unpopulated PV masquerade as a good reading -- see +"Trap 1" in `cora.api._capture_experiment_identity_reader`'s own +docstring. Run this command once the host is reachable and before `RUN_WITNESS_RECORDING_ENABLED` is set, and again after any -`CAPTURE_WATCH_PVS` / `CAPTURE_STATUS_PHASES` / `CAPTURE_BASELINE_PVS` edit. +`CAPTURE_WATCH_PVS` / `CAPTURE_STATUS_PHASES` / `CAPTURE_BASELINE_PVS` / +`CAPTURE_EXPERIMENT_IDENTITY_PVS` edit. ## Why this exists @@ -87,6 +95,18 @@ verdict `n/a`. Not decoding it here does not make it undecodable elsewhere; it means no decoder exists in production for it either. +`capture_experiment_identity_pvs`'s three roles (`proposal_number`, +`esaf_number`, `esaf_doi_number`, slice 14a) are a separate `group="experiment_identity"` +sweep, dispatched on `resolved_experiment_identity_text` from +`cora.api._capture_experiment_identity_reader` so this can never drift +from what the reader actually vaults. None of the three is personal +data, so the printed `value` is the raw reading, unredacted (defensive +path-shape redaction still applies if a `full_file_name` PV were +accidentally declared here instead, mirroring the baseline sweep's own +defense-in-depth). Verdict is `placeholder` for the substrate's own +`"Unknown"` literal, `empty` for a blank string, `text(len=N)` for a +real value, BAD only as `non-text`. + Exit codes: 0 every configured PV connected and decoded clean; 2 anything disconnected, timed out, was access-denied, or a decoder rejected it. """ @@ -100,6 +120,10 @@ from dataclasses import dataclass, field from typing import TYPE_CHECKING, TypeGuard +from cora.api._capture_experiment_identity_reader import ( + UNKNOWN_EXPERIMENT_IDENTITY_LITERAL, + resolved_experiment_identity_text, +) from cora.api._capture_observer import ( FULL_FILE_NAME_TRUNCATION_THRESHOLD, ROLE_ABORT, @@ -143,10 +167,12 @@ class _PvReport: code: str pv_key: str """The `capture_watch_pvs` role (`"status"`, `"abort"`, ...) for a - `group="watch"` line, or the `capture_baseline_pvs` channel_name for - a `group="baseline"` line. Named for what it structurally is (the - PV's inner-dict key) rather than "role", which is only true of half - this report's lines.""" + `group="watch"` line, the `capture_baseline_pvs` channel_name for a + `group="baseline"` line, or the `capture_experiment_identity_pvs` + role (`"proposal_number"`, `"esaf_number"`, `"esaf_doi_number"`) for a + `group="experiment_identity"` line. Named for what it structurally is (the + PV's inner-dict key) rather than "role", which is only true of two + of this report's three groups.""" pv: str ok: bool connected: bool @@ -157,13 +183,14 @@ class _PvReport: units: str | None = None verdict: str = "n/a" group: str = "watch" - """`"watch"` (`capture_watch_pvs`, the default) or `"baseline"` - (`capture_baseline_pvs`, slice 12). Purely a report-rendering - distinction; both groups share every other field.""" + """`"watch"` (`capture_watch_pvs`, the default), `"baseline"` + (`capture_baseline_pvs`, slice 12), or `"experiment_identity"` + (`capture_experiment_identity_pvs`, slice 14a). Purely a + report-rendering distinction; all groups share every other field.""" def render(self) -> str: tag = "OK " if self.ok else "BAD " - key_label = self.pv_key if self.group == "watch" else f"baseline:{self.pv_key}" + key_label = self.pv_key if self.group == "watch" else f"{self.group}:{self.pv_key}" head = f"{tag}{self.code}/{key_label:<17} {self.pv}" if not self.connected: return f"{head} NOT CONNECTED ({self.detail})" @@ -187,15 +214,18 @@ async def preflight_read_capture_pvs( capture_pvs: Mapping[str, Mapping[str, str]], status_phases: Mapping[str, str], baseline_pvs: Mapping[str, Mapping[str, str]] | None = None, + experiment_identity_pvs: Mapping[str, Mapping[str, str]] | None = None, ) -> _Report: """Read every configured `capture_watch_pvs` role, then every - `capture_baseline_pvs` channel (slice 12), once each; report shape. + `capture_baseline_pvs` channel (slice 12), then every + `capture_experiment_identity_pvs` role (slice 14a), once each; + report shape. Iteration order is sorted (code, then role/channel_name), watch - lines before baseline lines, so two runs against an unchanged - config produce line-for-line identical output. Each PV is read - independently: one dead or misconfigured PV does not abort the - sweep, it reports as its own failed line. + lines before baseline lines before experiment-identity lines, so two runs + against an unchanged config produce line-for-line identical output. + Each PV is read independently: one dead or misconfigured PV does + not abort the sweep, it reports as its own failed line. """ report = _Report() for code in sorted(capture_pvs): @@ -204,6 +234,9 @@ async def preflight_read_capture_pvs( for code in sorted(baseline_pvs or {}): for channel_name, pv in sorted((baseline_pvs or {})[code].items()): report.lines.append(await _read_one_baseline(control_port, code, channel_name, pv)) + for code in sorted(experiment_identity_pvs or {}): + for role, pv in sorted((experiment_identity_pvs or {})[code].items()): + report.lines.append(await _read_one_experiment_identity(control_port, code, role, pv)) return report @@ -405,6 +438,95 @@ def _baseline_verdict(reading: Measurement) -> tuple[str, bool]: return "n/a", True +async def _read_one_experiment_identity( + control_port: ControlPort, + code: str, + role: str, + pv: str, +) -> _PvReport: + """A `capture_experiment_identity_pvs` role's preflight result + (slice 14a). + + None of the three roles (`proposal_number`, `esaf_number`, + `esaf_doi_number`) is personal data, so the printed `value` is the raw + reading; the defensive path-shape redaction mirrors the baseline + sweep's own guard against a `full_file_name` PV being declared + under the wrong config key. + """ + try: + reading = await control_port.read(pv) + except (ControlNotConnectedError, ControlTimeoutError, ControlAccessDeniedError) as exc: + return _PvReport( + code=code, + pv_key=role, + pv=pv, + ok=False, + connected=False, + detail=str(exc), + group="experiment_identity", + ) + except ControlValueCoercionError as exc: + return _PvReport( + code=code, + pv_key=role, + pv=pv, + ok=False, + connected=True, + detail=f"adapter could not decode the reading: {exc}", + group="experiment_identity", + ) + element_count = ( + len(reading.value) + if reading.kind == "Array" and hasattr(reading.value, "__len__") + else None + ) + verdict, ok = _experiment_identity_verdict(reading.value) + return _PvReport( + code=code, + pv_key=role, + pv=pv, + ok=ok, + connected=True, + kind=reading.kind, + element_count=element_count, + value=( + f"" + if _looks_like_a_filesystem_path(reading.value) + else reading.value + ), + verdict=verdict, + group="experiment_identity", + ) + + +def _experiment_identity_verdict(value: object) -> tuple[str, bool]: + """The `capture_experiment_identity_pvs` decode check (slice 14a), + dispatched through `resolved_experiment_identity_text` so this can + never drift from what `CaptureExperimentIdentityReader` actually + vaults. + + `non-text` is the only BAD outcome: a non-string reading is a + deployment misconfiguration, not a value to guess at. `placeholder` + (the substrate's own `"Unknown"` literal, Trap 1) and `empty` are + both OK -- they are legitimate substrate states this preflight + exists to make VISIBLE, not defects to fail on -- and are reported + as their own distinct verdicts specifically so an operator does not + mistake either for a healthy value. `placeholder`, not `unknown`: + this report's own `status` role already uses `unrecognized` for a + different failure (CORA's decoder rejected the literal, rather than + the substrate never having populated it), and `unknown` would read + as a synonym for that instead of naming this row's own condition. + """ + if not isinstance(value, str): + return "non-text", False + resolved = resolved_experiment_identity_text(value) + if resolved is not None: + return f"text(len={len(resolved)})", True + if value.strip() == UNKNOWN_EXPERIMENT_IDENTITY_LITERAL: + return "placeholder", True + return "empty", True + + def _finish(report: _Report) -> int: print("capture-watch preflight") if not report.lines: @@ -423,15 +545,17 @@ def build_parser() -> argparse.ArgumentParser: return argparse.ArgumentParser( prog="python -m cora.api.capture_watch_preflight", description=( - "Read every PV configured under CAPTURE_WATCH_PVS, and every " - "channel under CAPTURE_BASELINE_PVS, once against the live " + "Read every PV configured under CAPTURE_WATCH_PVS, every " + "channel under CAPTURE_BASELINE_PVS, and every role under " + "CAPTURE_EXPERIMENT_IDENTITY_PVS, once against the live " "control system and report whether each connects, what shape " "CORA's ControlPort sees it as, the raw value, and (for " - "CAPTURE_WATCH_PVS roles) whether CORA's own decoder for that " - "role accepts it. Read-only; changes nothing. Run once the host " - "is reachable, before RUN_WITNESS_RECORDING_ENABLED is set, and " - "again after any CAPTURE_WATCH_PVS / CAPTURE_STATUS_PHASES / " - "CAPTURE_BASELINE_PVS edit." + "CAPTURE_WATCH_PVS / CAPTURE_EXPERIMENT_IDENTITY_PVS roles) " + "whether CORA's own decoder for that role accepts it. " + "Read-only; changes nothing. Run once the host is reachable, " + "before RUN_WITNESS_RECORDING_ENABLED is set, and again after " + "any CAPTURE_WATCH_PVS / CAPTURE_STATUS_PHASES / " + "CAPTURE_BASELINE_PVS / CAPTURE_EXPERIMENT_IDENTITY_PVS edit." ), ) @@ -450,6 +574,7 @@ async def _run() -> int: capture_pvs=settings.capture_watch_pvs, status_phases=settings.capture_status_phases, baseline_pvs=settings.capture_baseline_pvs, + experiment_identity_pvs=settings.capture_experiment_identity_pvs, ) return _finish(report) finally: diff --git a/apps/api/src/cora/api/main.py b/apps/api/src/cora/api/main.py index 86c4c45fd0..f0d52bd873 100644 --- a/apps/api/src/cora/api/main.py +++ b/apps/api/src/cora/api/main.py @@ -441,6 +441,12 @@ def _enforce_run_witness_recording_gate(settings: Settings) -> None: run_witness_recording_enabled is itself True (slice 13): the write happens at a promoted Run's terminal, so with no promotion there is no run_id to write the observed path against either. + + Also refuses to boot with capture_experiment_identity_recording_enabled=True + unless run_witness_recording_enabled is itself True (slice 14a): the + read happens exactly once, at the instant a capture promotes to a Run + (mirroring the baseline read's timing), so with no promotion there is + no run_id to vault a reading against either. """ if settings.capture_progress_recording_enabled and not settings.run_witness_recording_enabled: msg = ( @@ -463,6 +469,16 @@ def _enforce_run_witness_recording_gate(settings: Settings) -> None: "has no promoted Run's terminal to attach to without it." ) raise RuntimeError(msg) + if ( + settings.capture_experiment_identity_recording_enabled + and not settings.run_witness_recording_enabled + ): + msg = ( + "CAPTURE_EXPERIMENT_IDENTITY_RECORDING_ENABLED=true requires " + "RUN_WITNESS_RECORDING_ENABLED=true. An experiment-identity reading " + "has no promoted Run to vault it against without it." + ) + raise RuntimeError(msg) if not settings.run_witness_recording_enabled: return missing: list[str] = [] @@ -1230,6 +1246,8 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None]: control_port=app.state.operation.control_port, capture_baseline_pvs=settings.capture_baseline_pvs, capture_path_store=app.state.run.capture_path_store, + capture_experiment_identity_pvs=(settings.capture_experiment_identity_pvs), + experiment_identity_store=app.state.run.experiment_identity_store, ), ): yield diff --git a/apps/api/src/cora/infrastructure/config.py b/apps/api/src/cora/infrastructure/config.py index 803e626854..8e20a66c47 100644 --- a/apps/api/src/cora/infrastructure/config.py +++ b/apps/api/src/cora/infrastructure/config.py @@ -17,6 +17,11 @@ _ALLOWED_DATABASE_SCHEMES = ("postgresql://", "postgres://") +# Closed role vocabulary for `Settings.capture_experiment_identity_pvs` +# (slice 14a), dispatched on by name in +# `cora.api._capture_experiment_identity_reader`. +_EXPERIMENT_IDENTITY_ROLES = frozenset({"proposal_number", "esaf_number", "esaf_doi_number"}) + OtelExporter = Literal["otlp", "console", "none"] # ComputePort substrate selector. Deliberately NARROWER than the @@ -935,6 +940,96 @@ class Settings(BaseSettings): # here. See `cora.api._run_witness`'s "Capture path pairing" section. capture_path_recording_enabled: bool = False + # Experiment-identity PVs (slice 14a): a deployment-declared set read + # ONCE, at the instant a capture promotes to a witnessed Run, mirroring + # `capture_baseline_pvs`'s one-shot-at-BEGUN timing exactly. Same + # `code -> inner-key -> PV` shape as `capture_watch_pvs`: a CLOSED + # inner-key vocabulary (`proposal_number`, `esaf_number`, `esaf_doi_number`), + # because these three roles are dispatched on by name in + # `cora.api._capture_experiment_identity_reader` (an unrecognized role + # would silently never be read), unlike `capture_baseline_pvs`'s open + # per-channel vocabulary. + # + # CAPTURE_EXPERIMENT_IDENTITY_PVS='{ + # "2bmb-tomoscan": { + # "proposal_number": "2bmb:TomoScan:ProposalNumber", + # "esaf_number": "2bmb:TomoScan:ESAFNumber", + # "esaf_doi_number": "2bmb:TomoScan:ESAFDOINumber" + # } + # }' + # + # `ProposalNumber` / `ESAFNumber` are `stringout` records (native + # DBR_STRING; no `text_addresses` declaration needed). `ESAFDOINumber` + # is a `waveform` (DBR_CHAR), the identical wire shape as the + # `full_file_name` role's PV: the deployment's `CONTROL_PORT_ROUTES` + # MUST also declare it in `text_addresses`, or it decodes as a + # character array, not text. + # + # These three values are stamped by dmagic from APS scheduling data, + # not by the IOC itself, and are NOT personal data (unlike the `User*` + # PVs under the same `2bmb:TomoScan:` prefix, which slice 14b leaves + # unread): they are institutional identifiers for a funded experiment. + # They default to the substrate literal `"Unknown"` when unpopulated; + # CORA treats that literal, and an empty string, as ABSENT and records + # nothing (see `cora.api._capture_experiment_identity_reader`'s + # `resolved_experiment_identity_text`). + # + # Written to the `run_experiment_identity` PII-vault-shaped table + # (mirroring `run_capture_path`), NEVER onto `RunStarted` or any other + # event: the value read here is auto-harvested off an unauthenticated + # channel with no operator gesture behind it, unlike `start_run`'s + # operator-supplied `external_refs`, and events are immutable and + # INSERT-only, so a harvested proposal/ESAF number written there could + # never be withdrawn. A public-resolvability check against DataCite and + # the upstream `dmagic`/APS-DM-SDK source found `ESAFDOINumber` is + # populated from an internal, authenticated APS API + # (`EsafApsDbApi.getStationEsafById`), with no public DOI-registry + # record found; unconfirmed as a genuinely resolvable DOI, so it vaults + # with the other two rather than riding an event. See + # `cora.api._capture_experiment_identity_reader`'s module docstring for + # the full argument (memory/project_witnessed_run_prelive_slices.md, + # slice 14a). + capture_experiment_identity_pvs: dict[str, dict[str, str]] = {} + + # SIXTH, independent kill switch (slice 14a): gates whether the + # `capture_experiment_identity_pvs` set is actually read and vaulted at + # promotion. Default off. Refuses to boot if True without + # `run_witness_recording_enabled` also True (see + # `_enforce_run_witness_recording_gate`): with no promoted Run there is + # no run_id to vault a reading against. Independently revocable from + # the other five switches for the same reason `capture_path_recording_enabled` + # is: an operator must be able to turn OFF only this write pending a + # privacy/provenance review, without disabling progress, baseline, or + # path recording. Declaring the PVs in `capture_experiment_identity_pvs` + # alone is necessary but not sufficient, mirroring every other switch + # here. + capture_experiment_identity_recording_enabled: bool = False + + @field_validator("capture_experiment_identity_pvs") + @classmethod + def _validate_capture_experiment_identity_pvs( + cls, value: dict[str, dict[str, str]] + ) -> dict[str, dict[str, str]]: + """Refuse an unrecognized role key at boot, not at the first + promotion: `cora.api._capture_experiment_identity_reader` dispatches + on exactly `{"proposal_number", "esaf_number", "esaf_doi_number"}` by name, + so a typo'd role here would otherwise silently never be read, with + no error anywhere -- the same class of silent-misconfiguration risk + `_validate_capture_status_phases` already guards against.""" + bad = { + code: sorted(set(roles) - _EXPERIMENT_IDENTITY_ROLES) + for code, roles in value.items() + if set(roles) - _EXPERIMENT_IDENTITY_ROLES + } + if bad: + msg = ( + "capture_experiment_identity_pvs has roles outside " + f"{sorted(_EXPERIMENT_IDENTITY_ROLES)}: {bad}. An unrecognized role " + "is never read by cora.api._capture_experiment_identity_reader." + ) + raise ValueError(msg) + return value + @field_validator("capture_status_phases") @classmethod def _validate_capture_status_phases(cls, value: dict[str, str]) -> dict[str, str]: diff --git a/apps/api/src/cora/infrastructure/schema_version.py b/apps/api/src/cora/infrastructure/schema_version.py index 15781d5798..1e92c19399 100644 --- a/apps/api/src/cora/infrastructure/schema_version.py +++ b/apps/api/src/cora/infrastructure/schema_version.py @@ -74,7 +74,7 @@ class SchemaCheck: expected: str -EXPECTED_SCHEMA_VERSION: Final = "20260816094314" +EXPECTED_SCHEMA_VERSION: Final = "20260816140958" """The newest migration this build was written against. Hand-maintained, and deliberately not derived at runtime: the image does diff --git a/apps/api/src/cora/run/aggregates/run/__init__.py b/apps/api/src/cora/run/aggregates/run/__init__.py index e319b480cb..5a0597469d 100644 --- a/apps/api/src/cora/run/aggregates/run/__init__.py +++ b/apps/api/src/cora/run/aggregates/run/__init__.py @@ -49,6 +49,13 @@ to_payload, ) from cora.run.aggregates.run.evolver import evolve, fold +from cora.run.aggregates.run.experiment_identity import ( + ExperimentIdentity, + ExperimentIdentityStore, + InMemoryExperimentIdentityStore, + PostgresExperimentIdentityStore, + load_run_experiment_identity, +) from cora.run.aggregates.run.feed_heartbeats import ( FeedHeartbeat, FeedHeartbeatStore, @@ -181,10 +188,13 @@ "ChannelName", "ConductMode", "DecisionDebriefRequested", + "ExperimentIdentity", + "ExperimentIdentityStore", "FeedHeartbeat", "FeedHeartbeatStore", "HoldClaimReleased", "InMemoryCapturePathStore", + "InMemoryExperimentIdentityStore", "InMemoryFeedHeartbeatStore", "InMemoryObservationStore", "InvalidChannelNameError", @@ -206,6 +216,7 @@ "Observation", "ObservationStore", "PostgresCapturePathStore", + "PostgresExperimentIdentityStore", "PostgresFeedHeartbeatStore", "PostgresObservationStore", "Run", @@ -277,6 +288,7 @@ "is_last_active_claim", "load_run", "load_run_capture_path", + "load_run_experiment_identity", "supply_gate_check", "to_payload", "validate_adjusted_parameters_against_method_schema", diff --git a/apps/api/src/cora/run/aggregates/run/experiment_identity.py b/apps/api/src/cora/run/aggregates/run/experiment_identity.py new file mode 100644 index 0000000000..d3104b4315 --- /dev/null +++ b/apps/api/src/cora/run/aggregates/run/experiment_identity.py @@ -0,0 +1,319 @@ +"""ExperimentIdentity vault: a witnessed Run's proposal / ESAF / ESAF-DOI. + +Mirrors `capture_path.py`'s PATTERN exactly (itself mirroring +`actor_profile` / `ProfileStore`, memory/project_pii_vault): a mutable +side table, keyed by an identity the domain already has (here, the +Run's own `run_id`), holding facts that must never reach an event +payload because events are immutable and INSERT-only at the role level. + +A SIBLING table to `run_capture_path`, not a generalization of it, +despite the identical shape: a capture path and a proposal/ESAF number +are different kinds of fact (one is unambiguously personal data by +construction, the other is an institutional identifier that happens to +share this store's write-once-at-promotion timing and PII-vault-shaped +posture for a DIFFERENT reason -- see "Why the vault, not the event" +below). Folding them into one table would make a future, more +permissive disposition on one column read as a disposition on the +other. + +## Why this exists + +`tomoScan_2BM.template` exposes `ProposalNumber`, `ESAFNumber`, and +`ESAFDOINumber` under `2bmb:TomoScan:`, stamped by `dmagic` from APS +scheduling data, not by the IOC itself. `RunWitness` has no operator to +ask for a proposal the way `start_run.external_refs` does (see "Why the +vault, not the event"), so a witnessed Run has no experiment identity +at all today. This closes that gap without touching `RunStarted`. + +## Why the vault, not the event + +`start_run` accepts `external_refs` as an OPERATOR-SUPPLIED command +input (`start_run/command.py`): a human deliberately discloses an +identifier into an append-only, unerasable store. `RecordWitnessedRun` +carries no such field and has no operator behind it; CORA would be +AUTO-HARVESTING these three PVs off an unauthenticated EPICS channel, +every capture, with no human gesture backing the write. +[[project-conjunct-symmetry-design]]'s derivation rule permits exactly +this kind of asymmetry when the missing gesture can be named: here, the +missing gesture is "an operator stamping the proposal", which simply +does not exist on the witnessed path. + +D0 (memory/project_witnessed_run_prelive_plan.md) named the concrete +risk: a beamtime proposal number plus a timestamp is a strong join key +against public APS scheduling data, so an auto-harvested proposal +number is re-identifying via auxiliary public data even though the +field names no person. `RunStarted.external_refs` is already +`drop:opaque` in `record_export/_dispositions.py`, so putting these +values on the event would take the unerasability cost and buy no +publishing benefit at all: the decisive asymmetry is that promoting a +value from the vault to the record later is additive and easy, while +retracting one from an immutable event is impossible. The vault is the +choice that does not pre-empt D0. + +`ESAFDOINumber` was checked for the one way it could differ: a DOI is +meant to be a public, resolvable, third-party-verifiable handle, which +is the exact kind of thing D0 wants for "a facility can check its own +artifact" without disclosing anything about a person. Tracing +`dmagic`'s own source (`dm.py: get_esaf_doi`) shows the value comes +from `EsafApsDbApi.getStationEsafById`, an INTERNAL, authenticated APS +Data Management API, not a DOI registration agency; a DataCite search +for APS ESAF records returned zero results. Unconfirmed as a genuinely +resolvable public identifier, so it vaults with the other two rather +than riding `external_refs` as its own scheme: per the same asymmetry +argument, guessing "public" and being wrong is unerasable, while +guessing "vault" and being wrong later costs one additive follow-up +slice. + +## BC-internal, like CapturePathStore + +Same reasoning as `capture_path.py`'s own docstring: exactly one BC +writes and reads this table, so it is built locally in `wire_run(deps)` +and surfaced on `RunHandlers.experiment_identity_store`, never promoted +to a `Kernel` field. + +## Three independent facts, three independent substrate times + +Unlike `CapturePath` (one PV, one `observed_at`), this row holds THREE +independently-read PVs, each with its own substrate timestamp. Trap: +nothing in the IOC populates any of them; `dmagic` does, from APS +scheduling, so a value PERSISTS ACROSS BEAMTIMES until the next +beamtime's own sync overwrites it. If a value is stale (the current +beamtime's sync has not run yet), CORA cannot detect that from the PV +alone -- `Identifier` (`shared/identifier.py`) is `{scheme, value}` with +no room for a time, which is precisely why this fact lives in a vault +row rather than as an `Identifier` tuple: each `*_observed_at` column +carries the substrate's own reading time (`Measurement.produced_at`), +so a reader can at least SEE how old a value is. No freshness heuristic +is invented here or anywhere in this feature; staleness is a staff +question (memory/project_witnessed_run_prelive_plan.md), not a +computable verdict. + +A field may be present while a sibling is absent (e.g. `ProposalNumber` +populated, `ESAFNumber` still reading the substrate's own `"Unknown"` +placeholder): each pair is independently nullable for exactly this +reason. + +## Read path never redacts + +None of these three values is personal data (institutional identifiers +for a funded experiment, not a person's name or a directory path), so +unlike `CapturePath.observed_path` this dataclass carries no +`repr=False` and no read-path redaction: an operator or the +`get_run` response may show the resolved value directly. +""" + +# pyright: reportUnknownMemberType=false, reportUnknownVariableType=false, reportUnknownArgumentType=false +# asyncpg's stubs are loose; suppress only at module level for the +# adapter classes. Mirrors `run/aggregates/run/capture_path.py`'s +# identical suppress comment. + +from dataclasses import dataclass +from datetime import UTC, datetime +from typing import Protocol +from uuid import UUID + +import asyncpg + + +@dataclass(frozen=True) +class ExperimentIdentity: + """One row in the `run_experiment_identity` vault. + + Each of `proposal_number`, `esaf_number`, `esaf_doi_number` is independently + nullable, paired with its own `*_observed_at` (the substrate's own + reading time, `Measurement.produced_at`), never CORA's clock: a + deployment may configure fewer than three roles for a capture code, + or the substrate may report the `"Unknown"` placeholder / an empty + string for one PV while another reads a real value. + """ + + run_id: UUID + proposal_number: str | None + proposal_number_observed_at: datetime | None + esaf_number: str | None + esaf_number_observed_at: datetime | None + esaf_doi_number: str | None + esaf_doi_number_observed_at: datetime | None + created_at: datetime + updated_at: datetime + + +class ExperimentIdentityStore(Protocol): + """Read / write access to the `run_experiment_identity` table. + + Deliberately no batch/`get_many` read, mirroring `CapturePathStore`: + every consumer (`get_run`'s route/tool) resolves exactly one `run_id` + at a time. Two implementors: `PostgresExperimentIdentityStore` + (production) and `InMemoryExperimentIdentityStore` (tests / + `app_env=test`). + """ + + async def upsert( + self, + *, + run_id: UUID, + proposal_number: str | None, + proposal_number_observed_at: datetime | None, + esaf_number: str | None, + esaf_number_observed_at: datetime | None, + esaf_doi_number: str | None, + esaf_doi_number_observed_at: datetime | None, + created_at: datetime, + ) -> None: + """Insert a new row or overwrite an existing one for `run_id`. + + Idempotent on the run_id PK: `CaptureExperimentIdentityReader` calls + this at most once per promotion (one terminal genesis-read per + Run), but retrying after a partial failure replays cleanly. A + `None` value overwrites a previously-recorded one on retry: the + caller always supplies its own full, freshly-read snapshot, not + a partial patch. + """ + ... + + async def get(self, run_id: UUID) -> ExperimentIdentity | None: + """Fetch a row by run_id; `None` when absent (never read, or + recording disabled).""" + ... + + +async def load_run_experiment_identity( + store: ExperimentIdentityStore, run_id: UUID +) -> ExperimentIdentity | None: + """Resolve the vaulted experiment identity for a run_id, or `None`. + + Unlike `load_run_capture_path`, no tombstone placeholder: none of + these three values is personal data, so a plain `None` per field + (never observed, or the substrate read `"Unknown"`/empty) is honest + on its own, and the caller already knows the difference between "no + row" and "not applicable" from whether `capture_code` is set. + """ + return await store.get(run_id) + + +_UPSERT_SQL = """ +INSERT INTO run_experiment_identity ( + run_id, proposal_number, proposal_number_observed_at, + esaf_number, esaf_number_observed_at, esaf_doi_number, esaf_doi_number_observed_at, + created_at, updated_at +) +VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $8) +ON CONFLICT (run_id) DO UPDATE + SET proposal_number = EXCLUDED.proposal_number, + proposal_number_observed_at = EXCLUDED.proposal_number_observed_at, + esaf_number = EXCLUDED.esaf_number, + esaf_number_observed_at = EXCLUDED.esaf_number_observed_at, + esaf_doi_number = EXCLUDED.esaf_doi_number, + esaf_doi_number_observed_at = EXCLUDED.esaf_doi_number_observed_at, + updated_at = now() +""" + +_GET_SQL = """ +SELECT run_id, proposal_number, proposal_number_observed_at, + esaf_number, esaf_number_observed_at, esaf_doi_number, esaf_doi_number_observed_at, + created_at, updated_at +FROM run_experiment_identity +WHERE run_id = $1 +""" + + +def _row_to_experiment_identity(row: asyncpg.Record) -> ExperimentIdentity: + return ExperimentIdentity( + run_id=row["run_id"], + proposal_number=row["proposal_number"], + proposal_number_observed_at=row["proposal_number_observed_at"], + esaf_number=row["esaf_number"], + esaf_number_observed_at=row["esaf_number_observed_at"], + esaf_doi_number=row["esaf_doi_number"], + esaf_doi_number_observed_at=row["esaf_doi_number_observed_at"], + created_at=row["created_at"], + updated_at=row["updated_at"], + ) + + +class PostgresExperimentIdentityStore: + """asyncpg-backed `ExperimentIdentityStore` implementation.""" + + def __init__(self, pool: asyncpg.Pool) -> None: + self._pool = pool + + async def upsert( + self, + *, + run_id: UUID, + proposal_number: str | None, + proposal_number_observed_at: datetime | None, + esaf_number: str | None, + esaf_number_observed_at: datetime | None, + esaf_doi_number: str | None, + esaf_doi_number_observed_at: datetime | None, + created_at: datetime, + ) -> None: + async with self._pool.acquire() as conn: + await conn.execute( + _UPSERT_SQL, + run_id, + proposal_number, + proposal_number_observed_at, + esaf_number, + esaf_number_observed_at, + esaf_doi_number, + esaf_doi_number_observed_at, + created_at, + ) + + async def get(self, run_id: UUID) -> ExperimentIdentity | None: + async with self._pool.acquire() as conn: + row = await conn.fetchrow(_GET_SQL, run_id) + return _row_to_experiment_identity(row) if row is not None else None + + +class InMemoryExperimentIdentityStore: + """Test / `app_env=test` adapter for `ExperimentIdentityStore`. + + Postgres semantics preserved, mirroring `InMemoryCapturePathStore`: + on insert, `updated_at = created_at` (the caller's clock read); on + update, `updated_at = datetime.now(tz=UTC)` (the DB's own clock at + `ON CONFLICT DO UPDATE` time in the real adapter's `_UPSERT_SQL`, + never the caller-supplied `created_at`). + """ + + def __init__(self) -> None: + self._rows: dict[UUID, ExperimentIdentity] = {} + + async def upsert( + self, + *, + run_id: UUID, + proposal_number: str | None, + proposal_number_observed_at: datetime | None, + esaf_number: str | None, + esaf_number_observed_at: datetime | None, + esaf_doi_number: str | None, + esaf_doi_number_observed_at: datetime | None, + created_at: datetime, + ) -> None: + existing = self._rows.get(run_id) + self._rows[run_id] = ExperimentIdentity( + run_id=run_id, + proposal_number=proposal_number, + proposal_number_observed_at=proposal_number_observed_at, + esaf_number=esaf_number, + esaf_number_observed_at=esaf_number_observed_at, + esaf_doi_number=esaf_doi_number, + esaf_doi_number_observed_at=esaf_doi_number_observed_at, + created_at=existing.created_at if existing is not None else created_at, + updated_at=datetime.now(tz=UTC) if existing is not None else created_at, + ) + + async def get(self, run_id: UUID) -> ExperimentIdentity | None: + return self._rows.get(run_id) + + +__all__ = [ + "ExperimentIdentity", + "ExperimentIdentityStore", + "InMemoryExperimentIdentityStore", + "PostgresExperimentIdentityStore", + "load_run_experiment_identity", +] diff --git a/apps/api/src/cora/run/features/get_run/handler.py b/apps/api/src/cora/run/features/get_run/handler.py index ed6e260202..49adc99c72 100644 --- a/apps/api/src/cora/run/features/get_run/handler.py +++ b/apps/api/src/cora/run/features/get_run/handler.py @@ -4,10 +4,12 @@ / `get_method` / `get_family` / `get_subject` / `get_actor`. Returns `RunView`, not the bare domain `Run` (slice 13): the aggregate -plus `capture_code` (folded from `external_refs`) and -`observed_capture_path` (resolved from the `run_capture_path` PII -vault), mirroring `get_actor`'s `ActorView` composition exactly -- -`bind(deps, *, capture_path_store=...)` resolves both here, in the +plus `capture_code` (folded from `external_refs`), `observed_capture_path` +(resolved from the `run_capture_path` PII vault), and (slice 14a) the +proposal / ESAF / ESAF-DOI experiment identity resolved from the +`run_experiment_identity` vault, mirroring `get_actor`'s `ActorView` +composition exactly -- `bind(deps, *, capture_path_store=..., +experiment_identity_store=...)` resolves all of it here, in the handler, not at the route/tool layer: unlike `list_runs` (one shared handler instance consumed by internal composition-root runtimes that don't need this value), `get_run`'s handler has no internal caller @@ -19,7 +21,9 @@ Query handlers do NOT emit `causation_id` log fields. """ +import asyncio from dataclasses import dataclass +from datetime import datetime from typing import Protocol from uuid import UUID @@ -29,10 +33,12 @@ from cora.infrastructure.routing import NIL_SENTINEL_ID from cora.run.aggregates.run import ( CapturePathStore, + ExperimentIdentityStore, Run, extract_capture_code, load_run, load_run_capture_path, + load_run_experiment_identity, ) from cora.run.errors import UnauthorizedError from cora.run.features.get_run.query import GetRun @@ -44,7 +50,7 @@ @dataclass(frozen=True) class RunView: - """Read-side composition of Run aggregate + capture-path resolution. + """Read-side composition of Run aggregate + vault resolutions. `capture_code` is folded from `run.external_refs`; `None` for a Conducted Run. `observed_capture_path` resolves from the @@ -52,13 +58,29 @@ class RunView: `None` (a Conducted Run never touches the vault at all): `None` when not applicable, `UNOBSERVED_CAPTURE_PATH` (the tombstone) when a capture code exists but the vault has no row yet, the real path - otherwise. Route + MCP-tool layers destructure this into their wire + otherwise. + + `proposal_number` / `esaf_number` / `esaf_doi_number` (slice 14a), each + paired with its own `*_observed_at`, resolve from the + `run_experiment_identity` vault under the SAME `capture_code is not + None` condition. Unlike `observed_capture_path`, no tombstone + placeholder: none of these three values is personal data, so a + plain `None` is already honest, and a Conducted Run (`capture_code + is None`) and a Witnessed Run with nothing recorded yet are both + `None` here -- the caller already has `capture_code` to tell them + apart. Route + MCP-tool layers destructure this into their wire DTOs. """ run: Run capture_code: str | None observed_capture_path: str | None + proposal_number: str | None + proposal_number_observed_at: datetime | None + esaf_number: str | None + esaf_number_observed_at: datetime | None + esaf_doi_number: str | None + esaf_doi_number_observed_at: datetime | None class Handler(Protocol): @@ -74,8 +96,13 @@ async def __call__( ) -> RunView | None: ... -def bind(deps: Kernel, *, capture_path_store: CapturePathStore) -> Handler: - """Build a get_run handler closed over the shared deps + PII vault.""" +def bind( + deps: Kernel, + *, + capture_path_store: CapturePathStore, + experiment_identity_store: ExperimentIdentityStore, +) -> Handler: + """Build a get_run handler closed over the shared deps + both vaults.""" async def handler( query: GetRun, @@ -122,11 +149,17 @@ async def handler( return None capture_code = extract_capture_code(run.external_refs) - observed_capture_path = ( - await load_run_capture_path(capture_path_store, run.id) - if capture_code is not None - else None - ) + observed_capture_path: str | None = None + experiment_identity = None + if capture_code is not None: + # Two independent vault lookups; neither depends on the + # other's result, so run them concurrently rather than + # paying two sequential round trips on every witnessed-Run + # read. + observed_capture_path, experiment_identity = await asyncio.gather( + load_run_capture_path(capture_path_store, run.id), + load_run_experiment_identity(experiment_identity_store, run.id), + ) _log.info( "get_run.success", @@ -137,7 +170,33 @@ async def handler( found=True, ) return RunView( - run=run, capture_code=capture_code, observed_capture_path=observed_capture_path + run=run, + capture_code=capture_code, + observed_capture_path=observed_capture_path, + proposal_number=experiment_identity.proposal_number + if experiment_identity is not None + else None, + proposal_number_observed_at=( + experiment_identity.proposal_number_observed_at + if experiment_identity is not None + else None + ), + esaf_number=experiment_identity.esaf_number + if experiment_identity is not None + else None, + esaf_number_observed_at=( + experiment_identity.esaf_number_observed_at + if experiment_identity is not None + else None + ), + esaf_doi_number=( + experiment_identity.esaf_doi_number if experiment_identity is not None else None + ), + esaf_doi_number_observed_at=( + experiment_identity.esaf_doi_number_observed_at + if experiment_identity is not None + else None + ), ) return handler diff --git a/apps/api/src/cora/run/features/get_run/route.py b/apps/api/src/cora/run/features/get_run/route.py index c578cae38c..93518607fe 100644 --- a/apps/api/src/cora/run/features/get_run/route.py +++ b/apps/api/src/cora/run/features/get_run/route.py @@ -6,13 +6,15 @@ `subject_id` and `raid` are null when not set (calibration runs, or Runs not registered against a research activity respectively). -`capture_code` / `observed_capture_path` (slice 13) are resolved +`capture_code` / `observed_capture_path` (slice 13) and +`proposal_number` / `esaf_number` / `esaf_doi_number` (slice 14a) are resolved inside `get_run`'s own `Handler` (`handler.py`'s `RunView`), mirroring `get_actor`'s `ActorView` exactly: this route only destructures the already-composed view into its wire DTO, the same shape every other field on this DTO already follows. """ +from datetime import datetime from typing import Annotated, Any from uuid import UUID @@ -62,6 +64,16 @@ class RunResponse(BaseModel): or rejected by the dual-clock guard); the real path otherwise. This IS the authorized operator surface meant to let them find the file for `ingest_scan`. + + `proposal_number` / `esaf_number` / `esaf_doi_number` (slice 14a), each + paired with its own `*_observed_at` (the substrate's own reading + time, for judging staleness -- these PVs persist across beamtimes + with no in-band freshness signal), resolve from the + `run_experiment_identity` vault under the same `capture_code is + not None` condition. No tombstone: `None` here means either "not + applicable" (Conducted Run) or "nothing recorded yet"; `capture_code` + already distinguishes the two. Institutional identifiers for a + funded experiment, not personal data. """ id: UUID @@ -76,6 +88,12 @@ class RunResponse(BaseModel): campaign_id: UUID | None = None capture_code: str | None = None observed_capture_path: str | None = None + proposal_number: str | None = None + proposal_number_observed_at: datetime | None = None + esaf_number: str | None = None + esaf_number_observed_at: datetime | None = None + esaf_doi_number: str | None = None + esaf_doi_number_observed_at: datetime | None = None def _get_handler(request: Request) -> Handler: @@ -133,4 +151,10 @@ async def get_runs( campaign_id=run.campaign_id, capture_code=view.capture_code, observed_capture_path=view.observed_capture_path, + proposal_number=view.proposal_number, + proposal_number_observed_at=view.proposal_number_observed_at, + esaf_number=view.esaf_number, + esaf_number_observed_at=view.esaf_number_observed_at, + esaf_doi_number=view.esaf_doi_number, + esaf_doi_number_observed_at=view.esaf_doi_number_observed_at, ) diff --git a/apps/api/src/cora/run/features/get_run/tool.py b/apps/api/src/cora/run/features/get_run/tool.py index 8cee3af554..debc075d0e 100644 --- a/apps/api/src/cora/run/features/get_run/tool.py +++ b/apps/api/src/cora/run/features/get_run/tool.py @@ -1,6 +1,7 @@ """MCP tool for the `get_run` query slice. -`capture_code` / `observed_capture_path` (slice 13) are resolved +`capture_code` / `observed_capture_path` (slice 13) and +`proposal_number` / `esaf_number` / `esaf_doi_number` (slice 14a) are resolved inside `get_run`'s own `Handler` (`handler.py`'s `RunView`), mirroring `get_actor`'s `ActorView` exactly: this tool only destructures the already-composed view into its structured output, the same shape @@ -8,6 +9,7 @@ """ from collections.abc import Callable +from datetime import datetime from typing import Annotated, Any from uuid import UUID @@ -37,6 +39,12 @@ class RunOutput(BaseModel): campaign_id: UUID | None = None capture_code: str | None = None observed_capture_path: str | None = None + proposal_number: str | None = None + proposal_number_observed_at: datetime | None = None + esaf_number: str | None = None + esaf_number_observed_at: datetime | None = None + esaf_doi_number: str | None = None + esaf_doi_number_observed_at: datetime | None = None def register(mcp: FastMCP, *, get_handler: Callable[[], Handler]) -> None: @@ -77,4 +85,10 @@ async def get_run_tool( # pyright: ignore[reportUnusedFunction] campaign_id=run.campaign_id, capture_code=view.capture_code, observed_capture_path=view.observed_capture_path, + proposal_number=view.proposal_number, + proposal_number_observed_at=view.proposal_number_observed_at, + esaf_number=view.esaf_number, + esaf_number_observed_at=view.esaf_number_observed_at, + esaf_doi_number=view.esaf_doi_number, + esaf_doi_number_observed_at=view.esaf_doi_number_observed_at, ) diff --git a/apps/api/src/cora/run/wire.py b/apps/api/src/cora/run/wire.py index ee13600c06..4059cabcf7 100644 --- a/apps/api/src/cora/run/wire.py +++ b/apps/api/src/cora/run/wire.py @@ -101,12 +101,15 @@ from cora.infrastructure.observability import with_tracing from cora.run.aggregates.run import ( CapturePathStore, + ExperimentIdentityStore, FeedHeartbeatStore, InMemoryCapturePathStore, + InMemoryExperimentIdentityStore, InMemoryFeedHeartbeatStore, InMemoryObservationStore, ObservationStore, PostgresCapturePathStore, + PostgresExperimentIdentityStore, PostgresFeedHeartbeatStore, PostgresObservationStore, ) @@ -159,6 +162,12 @@ class RunHandlers: `get_run.bind()` (single-entity, mirroring `get_actor`) reads through the SAME instance; `list_runs` deliberately never touches it at all -- see this class's own module docstring.""" + experiment_identity_store: ExperimentIdentityStore + """Slice 14a's vault store for a witnessed Run's proposal / ESAF / + ESAF-DOI experiment identity. Same surfacing reason and the same + `list_runs`-never-touches-it posture as `capture_path_store`: + `RunWitnessRecorder`'s `CaptureExperimentIdentityReader` writes through it + directly, and `get_run.bind()` reads through the SAME instance.""" def wire_run(deps: Kernel) -> RunHandlers: @@ -174,9 +183,15 @@ def wire_run(deps: Kernel) -> RunHandlers: capture_path_store: CapturePathStore = ( PostgresCapturePathStore(deps.pool) if deps.pool is not None else InMemoryCapturePathStore() ) + experiment_identity_store: ExperimentIdentityStore = ( + PostgresExperimentIdentityStore(deps.pool) + if deps.pool is not None + else InMemoryExperimentIdentityStore() + ) return RunHandlers( feed_heartbeat_store=feed_heartbeat_store, capture_path_store=capture_path_store, + experiment_identity_store=experiment_identity_store, start_run=with_tracing( with_idempotency( start_run.bind(deps), @@ -253,7 +268,11 @@ def wire_run(deps: Kernel) -> RunHandlers: bc=_BC, ), get_run=with_tracing( - get_run.bind(deps, capture_path_store=capture_path_store), + get_run.bind( + deps, + capture_path_store=capture_path_store, + experiment_identity_store=experiment_identity_store, + ), command_name="GetRun", bc=_BC, kind="query", diff --git a/apps/api/tests/architecture/test_run_events_carry_no_pii.py b/apps/api/tests/architecture/test_run_events_carry_no_pii.py index 290dd8ebab..fcb769b9b4 100644 --- a/apps/api/tests/architecture/test_run_events_carry_no_pii.py +++ b/apps/api/tests/architecture/test_run_events_carry_no_pii.py @@ -15,12 +15,16 @@ events that don't carry the `Run` prefix. A name-prefix filter would have silently exempted them. -The deny-list covers both slice 13's own field (`observed_path`, -`capture_path`, plus the wire-level `full_file_name`) and the `User*` -PVs slice 14b already named as blocked (`project_witnessed_run_prelive_slices.md`): -a directory/proposal composition embeds a surname, and those PVs carry -a name, badge, and email directly. Widen it whenever a new identifying -field is found on the substrate. +The deny-list covers slice 13's own field (`observed_path`, +`capture_path`, plus the wire-level `full_file_name`), slice 14a's +proposal/ESAF/ESAF-DOI fields (`proposal_number`, `esaf_number`, +`esaf_doi_number` -- vaulted, never harvested onto an event: see +`cora.run.aggregates.run.experiment_identity`'s module docstring for +the full argument), and the `User*` PVs slice 14b already named as +blocked (`project_witnessed_run_prelive_slices.md`): a directory/proposal +composition embeds a surname, and those PVs carry a name, badge, and +email directly. Widen it whenever a new identifying field is found on +the substrate. """ import ast @@ -44,6 +48,8 @@ "file_path", "surname", "proposal_number", + "esaf_number", + "esaf_doi_number", "user_name", "user_last_name", "user_badge", @@ -76,8 +82,9 @@ def _pii_field_violations(source_path: Path) -> list[str]: @pytest.mark.architecture def test_run_event_payloads_carry_no_pii() -> None: - """Pin: dataclass fields named like PII never land on any event - class in `cora/run/aggregates/run/events.py`. + """Pin: dataclass fields named like PII, PLUS the slice 14a + proposal/ESAF/ESAF-DOI fields, never land on any event class in + `cora/run/aggregates/run/events.py`. The observed capture path is personal data (2-BM's directory layout embeds a surname and a proposal number); it lives in the @@ -86,12 +93,27 @@ class in `cora/run/aggregates/run/events.py`. usually means someone tried to carry the resolved path (or a raw `User*` PV) onto an event for convenience; move it to the vault instead. + + `proposal_number` / `esaf_number` / `esaf_doi_number` are NOT personal + data (institutional identifiers for a funded experiment), so their + presence here widens this test's scope past pure PII: it also + enforces slice 14a's own decision that a value auto-harvested off + an unauthenticated channel, with no operator gesture behind it, + must never ride an immutable, INSERT-only event regardless of + whether it identifies a person. See + `cora.run.aggregates.run.experiment_identity`'s module docstring + for the full argument. A regression here usually means someone + tried to carry one of these three values onto `RunStarted` for + convenience; move it to `run_experiment_identity` via + `ExperimentIdentityStore` instead. """ violations = _pii_field_violations(_EVENTS_FILE) assert not violations, ( - "Run event payloads must carry NO PII; move identifying fields to " - "run_capture_path via CapturePathStore (see " - "memory/project_witnessed_run_prelive_slices.md, slice 13):\n " + "\n ".join(violations) + "Run event payloads must carry NO PII and none of the slice 14a " + "experiment-identity fields; move identifying fields to " + "run_capture_path / run_experiment_identity via their stores (see " + "memory/project_witnessed_run_prelive_slices.md, slices 13 and " + "14a):\n " + "\n ".join(violations) ) diff --git a/apps/api/tests/contract/test_get_run_endpoint.py b/apps/api/tests/contract/test_get_run_endpoint.py index 6b3fe88a7c..55e0630aa4 100644 --- a/apps/api/tests/contract/test_get_run_endpoint.py +++ b/apps/api/tests/contract/test_get_run_endpoint.py @@ -87,6 +87,15 @@ def test_get_run_returns_200_with_running_status_for_sample_run() -> None: # no capture code at all, so neither field resolves. "capture_code": None, "observed_capture_path": None, + # Slice 14a additive response surface: same reasoning as + # capture_code/observed_capture_path above -- a Conducted Run + # never touches the run_experiment_identity vault at all. + "proposal_number": None, + "proposal_number_observed_at": None, + "esaf_number": None, + "esaf_number_observed_at": None, + "esaf_doi_number": None, + "esaf_doi_number_observed_at": None, } diff --git a/apps/api/tests/integration/test_experiment_identity_postgres.py b/apps/api/tests/integration/test_experiment_identity_postgres.py new file mode 100644 index 0000000000..cf694de3bd --- /dev/null +++ b/apps/api/tests/integration/test_experiment_identity_postgres.py @@ -0,0 +1,167 @@ +"""Integration: the `run_experiment_identity` vault against real Postgres. + +Mirrors `test_capture_path_postgres.py`'s shape: exercise +`PostgresExperimentIdentityStore` directly against the migrated table, +no handler involved (this store is a plain composition-root dependency, +not wrapped by a command). Also confirms the RLS posture the init +migration declares. +""" + +# pyright: reportUnknownMemberType=false, reportUnknownVariableType=false, reportUnknownArgumentType=false + +from datetime import UTC, datetime +from uuid import uuid4 + +import asyncpg +import pytest + +from cora.run.aggregates.run import PostgresExperimentIdentityStore + +_NOW = datetime(2026, 8, 16, 12, 0, 0, tzinfo=UTC) + + +@pytest.mark.integration +async def test_upsert_then_get_roundtrips_through_postgres(db_pool: asyncpg.Pool) -> None: + store = PostgresExperimentIdentityStore(db_pool) + run_id = uuid4() + + await store.upsert( + run_id=run_id, + proposal_number="12345", + proposal_number_observed_at=_NOW, + esaf_number="67890", + esaf_number_observed_at=_NOW, + esaf_doi_number="10.1234/esaf.67890", + esaf_doi_number_observed_at=_NOW, + created_at=_NOW, + ) + + row = await store.get(run_id) + assert row is not None + assert row.run_id == run_id + assert row.proposal_number == "12345" + assert row.esaf_number == "67890" + assert row.esaf_doi_number == "10.1234/esaf.67890" + + +@pytest.mark.integration +async def test_upsert_accepts_a_partial_reading_through_postgres(db_pool: asyncpg.Pool) -> None: + store = PostgresExperimentIdentityStore(db_pool) + run_id = uuid4() + + await store.upsert( + run_id=run_id, + proposal_number="12345", + proposal_number_observed_at=_NOW, + esaf_number=None, + esaf_number_observed_at=None, + esaf_doi_number=None, + esaf_doi_number_observed_at=None, + created_at=_NOW, + ) + + row = await store.get(run_id) + assert row is not None + assert row.proposal_number == "12345" + assert row.esaf_number is None + assert row.esaf_doi_number is None + + +@pytest.mark.integration +async def test_get_absent_run_id_returns_none(db_pool: asyncpg.Pool) -> None: + store = PostgresExperimentIdentityStore(db_pool) + assert await store.get(uuid4()) is None + + +@pytest.mark.integration +async def test_upsert_is_idempotent_on_run_id(db_pool: asyncpg.Pool) -> None: + """A retry (same run_id, e.g. after a transient failure) overwrites + rather than duplicating: `run_id` is the PRIMARY KEY, and + `ON CONFLICT (run_id) DO UPDATE` is the whole point of the vault + being mutable, not append-only.""" + store = PostgresExperimentIdentityStore(db_pool) + run_id = uuid4() + await store.upsert( + run_id=run_id, + proposal_number="first", + proposal_number_observed_at=_NOW, + esaf_number=None, + esaf_number_observed_at=None, + esaf_doi_number=None, + esaf_doi_number_observed_at=None, + created_at=_NOW, + ) + await store.upsert( + run_id=run_id, + proposal_number="second", + proposal_number_observed_at=_NOW, + esaf_number=None, + esaf_number_observed_at=None, + esaf_doi_number=None, + esaf_doi_number_observed_at=None, + created_at=_NOW, + ) + + async with db_pool.acquire() as conn: + count = await conn.fetchval( + "SELECT count(*) FROM run_experiment_identity WHERE run_id = $1", run_id + ) + assert count == 1 + row = await store.get(run_id) + assert row is not None + assert row.proposal_number == "second" + + +@pytest.mark.integration +async def test_table_has_force_row_level_security_enabled(db_pool: asyncpg.Pool) -> None: + """Defense-in-depth check on the migration itself: FORCE (not just + ENABLE) means even the table-owner role goes through policy, + mirroring `run_capture_path`'s identical posture.""" + async with db_pool.acquire() as conn: + row = await conn.fetchrow( + "SELECT relrowsecurity, relforcerowsecurity FROM pg_class WHERE relname = $1", + "run_experiment_identity", + ) + assert row is not None + assert row["relrowsecurity"] is True + assert row["relforcerowsecurity"] is True + + +@pytest.mark.integration +async def test_proposal_number_length_constraint_rejects_an_oversized_value( + db_pool: asyncpg.Pool, +) -> None: + """Defense-in-depth CHECK bound, independent of the application-layer + reader (which never fabricates an oversized value on its own but + should not depend solely on that for a value read off an + unauthenticated channel).""" + store = PostgresExperimentIdentityStore(db_pool) + with pytest.raises(asyncpg.CheckViolationError): + await store.upsert( + run_id=uuid4(), + proposal_number="a" * 201, + proposal_number_observed_at=_NOW, + esaf_number=None, + esaf_number_observed_at=None, + esaf_doi_number=None, + esaf_doi_number_observed_at=None, + created_at=_NOW, + ) + + +@pytest.mark.integration +async def test_esaf_doi_number_length_constraint_rejects_an_oversized_value( + db_pool: asyncpg.Pool, +) -> None: + store = PostgresExperimentIdentityStore(db_pool) + with pytest.raises(asyncpg.CheckViolationError): + await store.upsert( + run_id=uuid4(), + proposal_number=None, + proposal_number_observed_at=None, + esaf_number=None, + esaf_number_observed_at=None, + esaf_doi_number="a" * 501, + esaf_doi_number_observed_at=_NOW, + created_at=_NOW, + ) diff --git a/apps/api/tests/unit/api/test_capture_experiment_identity_reader.py b/apps/api/tests/unit/api/test_capture_experiment_identity_reader.py new file mode 100644 index 0000000000..4dac8adbce --- /dev/null +++ b/apps/api/tests/unit/api/test_capture_experiment_identity_reader.py @@ -0,0 +1,377 @@ +"""Unit tests for `CaptureExperimentIdentityReader` +(cora.api._capture_experiment_identity_reader). + +Covers the one-shot read-three-roles-and-vault-once contract, the two +named traps ("Unknown" and empty treated as absent; no substrate time +means skip, never synthesize), that one bad PV does not abort the +sweep over the rest, and that every failure mode -- a dead PV, an +uncoercible reading, or the vault write itself -- is caught and logged +rather than raised. +""" + +from datetime import UTC, datetime +from uuid import UUID, uuid4 + +import pytest + +from cora.api._capture_experiment_identity_reader import ( + ROLE_ESAF_DOI_NUMBER, + ROLE_ESAF_NUMBER, + ROLE_PROPOSAL_NUMBER, + CaptureExperimentIdentityReader, + resolved_experiment_identity_text, +) +from cora.operation.ports.control_port import ( + ControlAccessDeniedError, + ControlNotConnectedError, + ControlTimeoutError, + ControlValueCoercionError, + Measurement, +) +from cora.run.aggregates.run import ExperimentIdentity, InMemoryExperimentIdentityStore +from tests.unit._helpers import build_deps + +_CODE = "2bmb-tomoscan" +_NOW = datetime(2026, 8, 16, 12, 0, 0, tzinfo=UTC) +_RUN_ID = UUID("01900000-0000-7000-8000-000000007114") + +_EXPERIMENT_IDENTITY_PVS = { + _CODE: { + ROLE_PROPOSAL_NUMBER: "2bmb:TomoScan:ProposalNumber", + ROLE_ESAF_NUMBER: "2bmb:TomoScan:ESAFNumber", + ROLE_ESAF_DOI_NUMBER: "2bmb:TomoScan:ESAFDOINumber", + } +} + + +def _reading(value: object, *, produced_at: datetime | None = _NOW) -> Measurement: + return Measurement( # type: ignore[arg-type] + value=value, + kind="Scalar", + quality="Good", # type: ignore[arg-type] + produced_at=produced_at, + units=None, + ) + + +class _FakeControlPort: + """Scripted `read()`-only fake, mirroring `_capture_baseline_reader`'s own.""" + + def __init__(self, script: dict[str, Measurement | Exception]) -> None: + self._script = script + + async def read(self, address: str) -> Measurement: + outcome = self._script[address] + if isinstance(outcome, Exception): + raise outcome + return outcome + + +def _reader( + *, + control_port: _FakeControlPort, + store: InMemoryExperimentIdentityStore | None = None, + experiment_identity_pvs: dict[str, dict[str, str]] | None = None, +) -> tuple[CaptureExperimentIdentityReader, InMemoryExperimentIdentityStore]: + vault = store if store is not None else InMemoryExperimentIdentityStore() + reader = CaptureExperimentIdentityReader( + deps=build_deps(ids=[uuid4() for _ in range(10)], now=_NOW), + control_port=control_port, # type: ignore[arg-type] + experiment_identity_pvs=experiment_identity_pvs + if experiment_identity_pvs is not None + else _EXPERIMENT_IDENTITY_PVS, + store=vault, + ) + return reader, vault + + +# --------------------------------------------------------------------------- +# resolved_experiment_identity_text: the shared absent-value rule (Trap 1) +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +@pytest.mark.parametrize( + "value", + ["Unknown", "UNKNOWN", "unknown", "", " ", " Unknown ", 42, None, 3.14], +) +def test_resolved_experiment_identity_text_treats_these_as_absent(value: object) -> None: + assert resolved_experiment_identity_text(value) is None + + +@pytest.mark.unit +def test_resolved_experiment_identity_text_strips_and_returns_a_real_value() -> None: + assert resolved_experiment_identity_text(" 12345 ") == "12345" + assert resolved_experiment_identity_text("12345") == "12345" + + +# --------------------------------------------------------------------------- +# CaptureExperimentIdentityReader.read +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +async def test_read_with_all_roles_present_vaults_all_three() -> None: + port = _FakeControlPort( + { + "2bmb:TomoScan:ProposalNumber": _reading("12345"), + "2bmb:TomoScan:ESAFNumber": _reading("67890"), + "2bmb:TomoScan:ESAFDOINumber": _reading("10.1234/esaf.67890"), + } + ) + reader, vault = _reader(control_port=port) + + await reader.read(_CODE, _RUN_ID) + + row = await vault.get(_RUN_ID) + assert row is not None + assert row.proposal_number == "12345" + assert row.esaf_number == "67890" + assert row.esaf_doi_number == "10.1234/esaf.67890" + assert row.proposal_number_observed_at == _NOW + + +@pytest.mark.unit +async def test_read_for_an_undeclared_code_is_a_no_op() -> None: + reader, vault = _reader(control_port=_FakeControlPort({})) + + await reader.read("some-other-code", _RUN_ID) + + assert await vault.get(_RUN_ID) is None + + +@pytest.mark.unit +async def test_read_with_no_roles_declared_for_the_code_is_a_no_op() -> None: + reader, vault = _reader(control_port=_FakeControlPort({}), experiment_identity_pvs={_CODE: {}}) + + await reader.read(_CODE, _RUN_ID) + + assert await vault.get(_RUN_ID) is None + + +@pytest.mark.unit +async def test_a_partially_declared_code_reads_only_its_declared_roles() -> None: + port = _FakeControlPort({"2bmb:TomoScan:ProposalNumber": _reading("12345")}) + reader, vault = _reader( + control_port=port, + experiment_identity_pvs={_CODE: {ROLE_PROPOSAL_NUMBER: "2bmb:TomoScan:ProposalNumber"}}, + ) + + await reader.read(_CODE, _RUN_ID) + + row = await vault.get(_RUN_ID) + assert row is not None + assert row.proposal_number == "12345" + assert row.esaf_number is None + assert row.esaf_doi_number is None + + +@pytest.mark.unit +async def test_unknown_literal_is_treated_as_absent_but_the_rest_of_the_sweep_survives() -> None: + """Trap 1: the substrate's own default reads as a plausible string + unless explicitly rejected.""" + port = _FakeControlPort( + { + "2bmb:TomoScan:ProposalNumber": _reading("Unknown"), + "2bmb:TomoScan:ESAFNumber": _reading("67890"), + "2bmb:TomoScan:ESAFDOINumber": _reading("Unknown"), + } + ) + reader, vault = _reader(control_port=port) + + await reader.read(_CODE, _RUN_ID) + + row = await vault.get(_RUN_ID) + assert row is not None + assert row.proposal_number is None + assert row.esaf_number == "67890" + assert row.esaf_doi_number is None + + +@pytest.mark.unit +async def test_empty_string_is_treated_as_absent() -> None: + port = _FakeControlPort( + { + "2bmb:TomoScan:ProposalNumber": _reading(""), + "2bmb:TomoScan:ESAFNumber": _reading("67890"), + "2bmb:TomoScan:ESAFDOINumber": _reading(""), + } + ) + reader, vault = _reader(control_port=port) + + await reader.read(_CODE, _RUN_ID) + + row = await vault.get(_RUN_ID) + assert row is not None + assert row.proposal_number is None + assert row.esaf_number == "67890" + + +@pytest.mark.unit +async def test_reading_with_no_substrate_time_is_skipped_not_synthesized() -> None: + port = _FakeControlPort( + { + "2bmb:TomoScan:ProposalNumber": _reading("12345", produced_at=None), + "2bmb:TomoScan:ESAFNumber": _reading("67890"), + "2bmb:TomoScan:ESAFDOINumber": _reading("Unknown"), + } + ) + reader, vault = _reader(control_port=port) + + await reader.read(_CODE, _RUN_ID) + + row = await vault.get(_RUN_ID) + assert row is not None + assert row.proposal_number is None + assert row.esaf_number == "67890" + + +@pytest.mark.unit +async def test_every_role_absent_writes_nothing() -> None: + port = _FakeControlPort( + { + "2bmb:TomoScan:ProposalNumber": _reading("Unknown"), + "2bmb:TomoScan:ESAFNumber": _reading(""), + "2bmb:TomoScan:ESAFDOINumber": _reading("Unknown"), + } + ) + reader, vault = _reader(control_port=port) + + await reader.read(_CODE, _RUN_ID) + + assert await vault.get(_RUN_ID) is None + + +@pytest.mark.unit +async def test_a_dead_pv_does_not_abort_the_sweep_over_the_rest() -> None: + port = _FakeControlPort( + { + "2bmb:TomoScan:ProposalNumber": ControlNotConnectedError( + "2bmb:TomoScan:ProposalNumber" + ), + "2bmb:TomoScan:ESAFNumber": _reading("67890"), + "2bmb:TomoScan:ESAFDOINumber": _reading("Unknown"), + } + ) + reader, vault = _reader(control_port=port) + + await reader.read(_CODE, _RUN_ID) + + row = await vault.get(_RUN_ID) + assert row is not None + assert row.proposal_number is None + assert row.esaf_number == "67890" + + +@pytest.mark.unit +async def test_a_timed_out_pv_is_skipped() -> None: + port = _FakeControlPort( + { + "2bmb:TomoScan:ProposalNumber": ControlTimeoutError( + "2bmb:TomoScan:ProposalNumber", 5.0 + ), + "2bmb:TomoScan:ESAFNumber": _reading("67890"), + "2bmb:TomoScan:ESAFDOINumber": _reading("Unknown"), + } + ) + reader, vault = _reader(control_port=port) + + await reader.read(_CODE, _RUN_ID) + + row = await vault.get(_RUN_ID) + assert row is not None + assert row.esaf_number == "67890" + + +@pytest.mark.unit +async def test_an_access_denied_pv_is_skipped() -> None: + port = _FakeControlPort( + { + "2bmb:TomoScan:ProposalNumber": ControlAccessDeniedError( + "2bmb:TomoScan:ProposalNumber" + ), + "2bmb:TomoScan:ESAFNumber": _reading("67890"), + "2bmb:TomoScan:ESAFDOINumber": _reading("Unknown"), + } + ) + reader, vault = _reader(control_port=port) + + await reader.read(_CODE, _RUN_ID) + + row = await vault.get(_RUN_ID) + assert row is not None + assert row.esaf_number == "67890" + + +@pytest.mark.unit +async def test_a_value_coercion_error_is_skipped() -> None: + port = _FakeControlPort( + { + "2bmb:TomoScan:ProposalNumber": ControlValueCoercionError( + "2bmb:TomoScan:ProposalNumber", "structured", "Scalar" + ), + "2bmb:TomoScan:ESAFNumber": _reading("67890"), + "2bmb:TomoScan:ESAFDOINumber": _reading("Unknown"), + } + ) + reader, vault = _reader(control_port=port) + + await reader.read(_CODE, _RUN_ID) + + row = await vault.get(_RUN_ID) + assert row is not None + assert row.esaf_number == "67890" + + +@pytest.mark.unit +async def test_an_unexpected_read_exception_is_caught_and_the_sweep_survives() -> None: + port = _FakeControlPort( + { + "2bmb:TomoScan:ProposalNumber": RuntimeError("boom"), + "2bmb:TomoScan:ESAFNumber": _reading("67890"), + "2bmb:TomoScan:ESAFDOINumber": _reading("Unknown"), + } + ) + reader, vault = _reader(control_port=port) + + await reader.read(_CODE, _RUN_ID) + + row = await vault.get(_RUN_ID) + assert row is not None + assert row.esaf_number == "67890" + + +@pytest.mark.unit +async def test_vault_write_failure_is_caught_and_does_not_raise() -> None: + class _FailingStore(InMemoryExperimentIdentityStore): + async def upsert(self, **kwargs: object) -> None: # type: ignore[override] + msg = "boom" + raise RuntimeError(msg) + + port = _FakeControlPort({"2bmb:TomoScan:ProposalNumber": _reading("12345")}) + reader, _vault = _reader( + control_port=port, + store=_FailingStore(), + experiment_identity_pvs={_CODE: {ROLE_PROPOSAL_NUMBER: "2bmb:TomoScan:ProposalNumber"}}, + ) + + await reader.read(_CODE, _RUN_ID) # must not raise + + +@pytest.mark.unit +async def test_read_is_idempotent_on_run_id() -> None: + """A retry (e.g. after a transient promotion-adjacent failure) + overwrites rather than duplicating: mirrors the vault's own PK + contract, exercised here through the reader's own call shape.""" + port = _FakeControlPort({"2bmb:TomoScan:ProposalNumber": _reading("12345")}) + reader, vault = _reader( + control_port=port, + experiment_identity_pvs={_CODE: {ROLE_PROPOSAL_NUMBER: "2bmb:TomoScan:ProposalNumber"}}, + ) + + await reader.read(_CODE, _RUN_ID) + await reader.read(_CODE, _RUN_ID) + + row = await vault.get(_RUN_ID) + assert isinstance(row, ExperimentIdentity) + assert row.proposal_number == "12345" diff --git a/apps/api/tests/unit/api/test_capture_watch_preflight.py b/apps/api/tests/unit/api/test_capture_watch_preflight.py index 60336b0b6c..e76495cb32 100644 --- a/apps/api/tests/unit/api/test_capture_watch_preflight.py +++ b/apps/api/tests/unit/api/test_capture_watch_preflight.py @@ -68,6 +68,7 @@ async def _preflight( *, status_phases: dict[str, str] | None = None, baseline_pvs: dict[str, dict[str, str]] | None = None, + experiment_identity_pvs: dict[str, dict[str, str]] | None = None, ) -> _Report: """`_FakeControlPort` implements `.read()` only (this command never writes or subscribes), so it satisfies `ControlPort` in practice but @@ -78,6 +79,7 @@ async def _preflight( capture_pvs=capture_pvs, status_phases=status_phases if status_phases is not None else _PHASES, baseline_pvs=baseline_pvs, + experiment_identity_pvs=experiment_identity_pvs, ) @@ -521,6 +523,166 @@ async def test_preflight_read_empty_baseline_pvs_reports_no_baseline_lines() -> assert report.lines == [] +# --------------------------------------------------------------------------- +# capture_experiment_identity_pvs sweep (slice 14a): "Unknown" and empty +# are visible, distinct, non-BAD verdicts; only a non-text reading is BAD. +# --------------------------------------------------------------------------- + +_EXPERIMENT_IDENTITY_PVS = { + "2bmb-tomoscan": { + "proposal_number": "2bmb:TomoScan:ProposalNumber", + "esaf_number": "2bmb:TomoScan:ESAFNumber", + "esaf_doi_number": "2bmb:TomoScan:ESAFDOINumber", + } +} + + +@pytest.mark.unit +async def test_preflight_read_identity_real_value_decodes_ok_with_length_verdict() -> None: + port = _FakeControlPort( + { + "2bmb:TomoScan:ProposalNumber": _reading("12345"), + "2bmb:TomoScan:ESAFNumber": _reading("67890"), + "2bmb:TomoScan:ESAFDOINumber": _reading("10.1234/esaf.67890"), + } + ) + + report = await _preflight(port, {}, experiment_identity_pvs=_EXPERIMENT_IDENTITY_PVS) + + assert len(report.lines) == 3 + assert all(line.ok for line in report.lines) + assert all(line.group == "experiment_identity" for line in report.lines) + by_role = {line.pv_key: line for line in report.lines} + assert by_role["proposal_number"].verdict == "text(len=5)" + assert by_role["proposal_number"].value == "12345" + assert by_role["esaf_doi_number"].verdict == "text(len=18)" + + +@pytest.mark.unit +async def test_preflight_read_identity_unknown_literal_is_a_distinct_ok_verdict() -> None: + """Trap 1: the substrate's own placeholder must never look like a + healthy, generic reading -- this is the single highest-value thing + this preflight can say about these PVs.""" + port = _FakeControlPort({"2bmb:TomoScan:ProposalNumber": _reading("Unknown")}) + + report = await _preflight( + port, + {}, + experiment_identity_pvs={"code": {"proposal_number": "2bmb:TomoScan:ProposalNumber"}}, + ) + + (line,) = report.lines + assert line.ok + assert line.verdict == "placeholder" + + +@pytest.mark.unit +async def test_preflight_read_identity_empty_string_is_ok_with_empty_verdict() -> None: + port = _FakeControlPort({"2bmb:TomoScan:ProposalNumber": _reading("")}) + + report = await _preflight( + port, + {}, + experiment_identity_pvs={"code": {"proposal_number": "2bmb:TomoScan:ProposalNumber"}}, + ) + + (line,) = report.lines + assert line.ok + assert line.verdict == "empty" + + +@pytest.mark.unit +async def test_preflight_read_identity_non_text_is_bad() -> None: + port = _FakeControlPort({"2bmb:TomoScan:ProposalNumber": _reading(12345)}) + + report = await _preflight( + port, + {}, + experiment_identity_pvs={"code": {"proposal_number": "2bmb:TomoScan:ProposalNumber"}}, + ) + + (line,) = report.lines + assert not line.ok + assert line.verdict == "non-text" + + +@pytest.mark.unit +async def test_preflight_read_identity_not_connected_pv_reports_bad() -> None: + port = _FakeControlPort({"pv:dead": ControlNotConnectedError("pv:dead")}) + + report = await _preflight( + port, {}, experiment_identity_pvs={"code": {"proposal_number": "pv:dead"}} + ) + + (line,) = report.lines + assert not line.ok + assert not line.connected + assert line.group == "experiment_identity" + + +@pytest.mark.unit +async def test_preflight_read_identity_value_coercion_error_reports_bad_but_connected() -> None: + port = _FakeControlPort({"pv:bad": ControlValueCoercionError("pv:bad", "structured", "Scalar")}) + + report = await _preflight( + port, {}, experiment_identity_pvs={"code": {"proposal_number": "pv:bad"}} + ) + + (line,) = report.lines + assert not line.ok + assert line.connected + + +@pytest.mark.unit +async def test_preflight_read_identity_mis_keyed_role_with_a_path_value_still_redacts() -> None: + """Defense-in-depth: a `full_file_name` PV accidentally declared + under `capture_experiment_identity_pvs` must not print its real, + personal-data-bearing value here, even though none of the three + real identity roles is itself sensitive.""" + port = _FakeControlPort({"pv:misdeclared": _reading("/data/2026-01-Smith-12345/scan.h5")}) + + report = await _preflight( + port, {}, experiment_identity_pvs={"code": {"proposal_number": "pv:misdeclared"}} + ) + + (line,) = report.lines + assert isinstance(line.value, str) + assert line.value.startswith(" ( + None +): + port = _FakeControlPort( + { + "2bmb:TomoScan:ScanStatus": _reading("Scan complete", kind="Categorical"), + "2bmb:TomoScan:ExposureTime": _reading(1.5), + "2bmb:TomoScan:ProposalNumber": _reading("12345"), + } + ) + + report = await _preflight( + port, + {"2bmb-tomoscan": {"status": "2bmb:TomoScan:ScanStatus"}}, + baseline_pvs={"2bmb-tomoscan": {"ExposureTime": "2bmb:TomoScan:ExposureTime"}}, + experiment_identity_pvs={ + "2bmb-tomoscan": {"proposal_number": "2bmb:TomoScan:ProposalNumber"} + }, + ) + + assert len(report.lines) == 3 + groups = {line.group for line in report.lines} + assert groups == {"watch", "baseline", "experiment_identity"} + + +@pytest.mark.unit +async def test_preflight_read_empty_experiment_identity_pvs_reports_no_lines() -> None: + report = await _preflight(_FakeControlPort({}), {}, experiment_identity_pvs={}) + + assert report.lines == [] + + @pytest.mark.unit async def test_finish_exit_code_zero_when_every_line_is_ok( capsys: pytest.CaptureFixture[str], diff --git a/apps/api/tests/unit/api/test_run_witness.py b/apps/api/tests/unit/api/test_run_witness.py index f71917b57a..0a96c23eee 100644 --- a/apps/api/tests/unit/api/test_run_witness.py +++ b/apps/api/tests/unit/api/test_run_witness.py @@ -40,6 +40,7 @@ from cora.run.aggregates.run import ( ConductMode, InMemoryCapturePathStore, + InMemoryExperimentIdentityStore, InMemoryFeedHeartbeatStore, RunStarted, event_type_name, @@ -571,12 +572,17 @@ def _recorder( capture_baseline_recording_enabled: bool = False, capture_path_store: object | None = None, capture_path_recording_enabled: bool = False, + experiment_identity_reader: object | None = None, + capture_experiment_identity_recording_enabled: bool = False, ) -> RunWitnessRecorder: settings = Settings( # type: ignore[call-arg] run_witness_recording_enabled=run_witness_recording_enabled, capture_watch_plan_id=capture_watch_plan_id, capture_baseline_recording_enabled=capture_baseline_recording_enabled, capture_path_recording_enabled=capture_path_recording_enabled, + capture_experiment_identity_recording_enabled=( + capture_experiment_identity_recording_enabled + ), ) outcome = record_witnessed_run_outcome or _FakeRecordWitnessedRunOutcome() truncate = truncate_run or _FakeTruncateRun() @@ -589,6 +595,7 @@ def _recorder( open_captures=open_captures, baseline_reader=baseline_reader, # type: ignore[arg-type] capture_path_store=capture_path_store, # type: ignore[arg-type] + experiment_identity_reader=experiment_identity_reader, # type: ignore[arg-type] ) @@ -715,6 +722,113 @@ async def test_promotion_survives_a_baseline_read_failure() -> None: assert recorder.open_captures() == {_CODE: fake.run_id} +class _FakeExperimentIdentityReader: + """Records every `read()` call, or raises a configured exception instead.""" + + def __init__(self, *, raises: Exception | None = None) -> None: + self.calls: list[tuple[str, UUID]] = [] + self._raises = raises + + async def read(self, capture_code: str, run_id: UUID) -> None: + self.calls.append((capture_code, run_id)) + if self._raises is not None: + raise self._raises + + +@pytest.mark.unit +async def test_promotion_reads_the_experiment_identity_when_configured_and_enabled() -> None: + fake = _FakeRecordWitnessedRun() + identity_reader = _FakeExperimentIdentityReader() + recorder = _recorder( + record_witnessed_run=fake, + experiment_identity_reader=identity_reader, + capture_experiment_identity_recording_enabled=True, + ) + + await recorder.observe_capture(_obs(reported_status="Beginning scan", phase=CapturePhase.BEGUN)) + + assert identity_reader.calls == [(_CODE, fake.run_id)] + + +@pytest.mark.unit +async def test_promotion_does_not_read_the_experiment_identity_when_the_kill_switch_is_off() -> ( + None +): + """Declaring a reader is not sufficient; + capture_experiment_identity_recording_enabled is the sixth, + independent kill switch.""" + fake = _FakeRecordWitnessedRun() + identity_reader = _FakeExperimentIdentityReader() + recorder = _recorder( + record_witnessed_run=fake, + experiment_identity_reader=identity_reader, + capture_experiment_identity_recording_enabled=False, + ) + + await recorder.observe_capture(_obs(reported_status="Beginning scan", phase=CapturePhase.BEGUN)) + + assert identity_reader.calls == [] + + +@pytest.mark.unit +async def test_experiment_identity_kill_switch_is_read_fresh_at_call_time() -> None: + """Mirrors the baseline kill switch's identical freshness contract.""" + fake = _FakeRecordWitnessedRun() + identity_reader = _FakeExperimentIdentityReader() + recorder = _recorder( + record_witnessed_run=fake, + experiment_identity_reader=identity_reader, + capture_experiment_identity_recording_enabled=False, + ) + + await recorder.observe_capture( + _obs(reported_status="Beginning scan", phase=CapturePhase.BEGUN, capture_code="code-a") + ) + assert identity_reader.calls == [] + + recorder._settings.capture_experiment_identity_recording_enabled = True # pyright: ignore[reportPrivateUsage] + + await recorder.observe_capture( + _obs(reported_status="Beginning scan", phase=CapturePhase.BEGUN, capture_code="code-b") + ) + assert identity_reader.calls == [("code-b", fake.run_id)] + + +@pytest.mark.unit +async def test_promotion_with_no_experiment_identity_reader_configured_is_unaffected() -> None: + fake = _FakeRecordWitnessedRun() + recorder = _recorder( + record_witnessed_run=fake, + experiment_identity_reader=None, + capture_experiment_identity_recording_enabled=True, + ) + + await recorder.observe_capture(_obs(reported_status="Beginning scan", phase=CapturePhase.BEGUN)) + + # No exception, and the promotion itself still succeeded. + assert len(fake.calls) == 1 + + +@pytest.mark.unit +async def test_promotion_survives_an_experiment_identity_read_failure() -> None: + """A vault-read failure must never unwind or be mistaken for a + failed promotion: by the time it runs, the promotion already + committed.""" + fake = _FakeRecordWitnessedRun() + identity_reader = _FakeExperimentIdentityReader(raises=RuntimeError("boom")) + recorder = _recorder( + record_witnessed_run=fake, + experiment_identity_reader=identity_reader, + capture_experiment_identity_recording_enabled=True, + ) + + await recorder.observe_capture( + _obs(reported_status="Beginning scan", phase=CapturePhase.BEGUN) + ) # must not raise + + assert recorder.open_captures() == {_CODE: fake.run_id} + + # ---------- Capture path pairing / dual-clock guard (slice 13) ---------- _T0 = _NOW @@ -1865,6 +1979,110 @@ async def test_run_witness_lifespan_declares_baseline_pvs_but_kill_switch_off_ap assert append_observations.calls == [] +@pytest.mark.unit +async def test_run_witness_lifespan_rejects_experiment_identity_pvs_no_control_port() -> None: + deps = dataclasses.replace( + build_deps(ids=[uuid4() for _ in range(5)]), + settings=Settings( # type: ignore[call-arg] + run_witness_recording_enabled=True, + capture_watch_plan_id=_PLAN_ID, + ), + ) + with pytest.raises(ValueError, match="control_port"): + async with run_witness_lifespan( + observer=_FakeObserver([]), + capture_codes=frozenset({_CODE}), + deps=deps, + record_witnessed_run=_FakeRecordWitnessedRun(), + record_witnessed_run_outcome=_FakeRecordWitnessedRunOutcome(), + truncate_run=_FakeTruncateRun(), + capture_experiment_identity_pvs={ + _CODE: {"proposal_number": "2bmb:TomoScan:ProposalNumber"} + }, + experiment_identity_store=InMemoryExperimentIdentityStore(), + ): + pass + + +@pytest.mark.unit +async def test_run_witness_lifespan_with_experiment_identity_pvs_reads_and_vaults() -> None: + """End-to-end wiring check: a BEGUN promotes and the experiment- + identity reader actually reads through the real ControlPort fake + and vaults against the promoted run_id -- proving the reader is + constructed and invoked, not just accepted as a parameter.""" + run_id = uuid4() + genesis = _FakeRecordWitnessedRun(run_id=run_id) + experiment_identity_store = InMemoryExperimentIdentityStore() + control_port = _FakeBaselineControlPort({"2bmb:TomoScan:ProposalNumber": "12345"}) + observer = _FakeObserver([_obs(reported_status="Beginning scan", phase=CapturePhase.BEGUN)]) + deps = dataclasses.replace( + build_deps(ids=[uuid4() for _ in range(20)]), + settings=Settings( # type: ignore[call-arg] + run_witness_recording_enabled=True, + capture_watch_plan_id=_PLAN_ID, + capture_experiment_identity_recording_enabled=True, + ), + ) + + async with run_witness_lifespan( + observer=observer, + capture_codes=frozenset({_CODE}), + deps=deps, + record_witnessed_run=genesis, + record_witnessed_run_outcome=_FakeRecordWitnessedRunOutcome(), + truncate_run=_FakeTruncateRun(), + control_port=control_port, # type: ignore[arg-type] + capture_experiment_identity_pvs={ + _CODE: {"proposal_number": "2bmb:TomoScan:ProposalNumber"} + }, + experiment_identity_store=experiment_identity_store, + ): + await asyncio.sleep(0.02) + + assert len(genesis.calls) == 1 + row = await experiment_identity_store.get(run_id) + assert row is not None + assert row.proposal_number == "12345" + + +@pytest.mark.unit +async def test_run_witness_lifespan_declares_identity_pvs_but_switch_off_vaults_nothing() -> None: + """Declaring capture_experiment_identity_pvs builds a reader; the + sixth kill switch, capture_experiment_identity_recording_enabled, + is what actually gates whether it is called.""" + run_id = uuid4() + genesis = _FakeRecordWitnessedRun(run_id=run_id) + experiment_identity_store = InMemoryExperimentIdentityStore() + control_port = _FakeBaselineControlPort({"2bmb:TomoScan:ProposalNumber": "12345"}) + observer = _FakeObserver([_obs(reported_status="Beginning scan", phase=CapturePhase.BEGUN)]) + deps = dataclasses.replace( + build_deps(ids=[uuid4() for _ in range(20)]), + settings=Settings( # type: ignore[call-arg] + run_witness_recording_enabled=True, + capture_watch_plan_id=_PLAN_ID, + capture_experiment_identity_recording_enabled=False, + ), + ) + + async with run_witness_lifespan( + observer=observer, + capture_codes=frozenset({_CODE}), + deps=deps, + record_witnessed_run=genesis, + record_witnessed_run_outcome=_FakeRecordWitnessedRunOutcome(), + truncate_run=_FakeTruncateRun(), + control_port=control_port, # type: ignore[arg-type] + capture_experiment_identity_pvs={ + _CODE: {"proposal_number": "2bmb:TomoScan:ProposalNumber"} + }, + experiment_identity_store=experiment_identity_store, + ): + await asyncio.sleep(0.02) + + assert len(genesis.calls) == 1 + assert await experiment_identity_store.get(run_id) is None + + @pytest.mark.unit async def test_run_witness_lifespan_flushes_buffered_progress_on_shutdown() -> None: """A reading buffered right before teardown must not be silently diff --git a/apps/api/tests/unit/api/test_run_witness_recording_gate.py b/apps/api/tests/unit/api/test_run_witness_recording_gate.py index d5d843e5d7..0aedabf78d 100644 --- a/apps/api/tests/unit/api/test_run_witness_recording_gate.py +++ b/apps/api/tests/unit/api/test_run_witness_recording_gate.py @@ -18,6 +18,10 @@ Slice 13 adds a FIFTH gate, same shape: `capture_path_recording_enabled=True` requires `run_witness_recording_enabled=True`. + +Slice 14a adds a SIXTH gate, same shape: +`capture_experiment_identity_recording_enabled=True` requires +`run_witness_recording_enabled=True`. """ from uuid import UUID, uuid4 @@ -36,6 +40,7 @@ def _settings( capture_progress_recording_enabled: bool = False, capture_baseline_recording_enabled: bool = False, capture_path_recording_enabled: bool = False, + capture_experiment_identity_recording_enabled: bool = False, ) -> Settings: return Settings( # type: ignore[call-arg] run_witness_enabled=run_witness_enabled, @@ -44,6 +49,9 @@ def _settings( capture_progress_recording_enabled=capture_progress_recording_enabled, capture_baseline_recording_enabled=capture_baseline_recording_enabled, capture_path_recording_enabled=capture_path_recording_enabled, + capture_experiment_identity_recording_enabled=( + capture_experiment_identity_recording_enabled + ), ) @@ -217,3 +225,43 @@ def test_path_recording_enabled_with_run_witness_recording_passes() -> None: capture_path_recording_enabled=True, ) ) + + +def test_experiment_identity_recording_enabled_without_run_witness_recording_refuses_boot() -> None: + with pytest.raises(RuntimeError, match="RUN_WITNESS_RECORDING_ENABLED=true"): + _enforce_run_witness_recording_gate( + _settings( + run_witness_enabled=True, + capture_watch_plan_id=uuid4(), + run_witness_recording_enabled=False, + capture_experiment_identity_recording_enabled=True, + ) + ) + + +def test_experiment_identity_recording_enabled_checked_before_the_first_gates_prerequisites() -> ( + None +): + """Same independence property as the progress / baseline / path + gates: this gate's own message must appear even when the FIRST + gate's prerequisites are also missing.""" + with pytest.raises(RuntimeError, match="RUN_WITNESS_RECORDING_ENABLED=true"): + _enforce_run_witness_recording_gate( + _settings( + run_witness_enabled=False, + capture_watch_plan_id=None, + run_witness_recording_enabled=False, + capture_experiment_identity_recording_enabled=True, + ) + ) + + +def test_experiment_identity_recording_enabled_with_run_witness_recording_passes() -> None: + _enforce_run_witness_recording_gate( + _settings( + run_witness_enabled=True, + capture_watch_plan_id=uuid4(), + run_witness_recording_enabled=True, + capture_experiment_identity_recording_enabled=True, + ) + ) diff --git a/apps/api/tests/unit/run/test_experiment_identity.py b/apps/api/tests/unit/run/test_experiment_identity.py new file mode 100644 index 0000000000..9c9125b01a --- /dev/null +++ b/apps/api/tests/unit/run/test_experiment_identity.py @@ -0,0 +1,155 @@ +"""Unit tests for the `run_experiment_identity` vault's InMemory adapter +and the `load_run_experiment_identity` helper (slice 14a). + +Mirrors `test_capture_path.py`'s shape: exercise the store contract +directly, no reader or recorder involved. +""" + +from datetime import UTC, datetime, timedelta +from uuid import uuid4 + +import pytest + +from cora.run.aggregates.run import ( + InMemoryExperimentIdentityStore, + load_run_experiment_identity, +) + +_T0 = datetime(2026, 8, 16, 12, 0, 0, tzinfo=UTC) + + +def _at(seconds: int) -> datetime: + return _T0 + timedelta(seconds=seconds) + + +@pytest.mark.unit +async def test_upsert_then_get_roundtrips_all_three_fields() -> None: + store = InMemoryExperimentIdentityStore() + run_id = uuid4() + + await store.upsert( + run_id=run_id, + proposal_number="12345", + proposal_number_observed_at=_at(0), + esaf_number="67890", + esaf_number_observed_at=_at(1), + esaf_doi_number="10.1234/esaf.67890", + esaf_doi_number_observed_at=_at(2), + created_at=_at(3), + ) + + row = await store.get(run_id) + assert row is not None + assert row.run_id == run_id + assert row.proposal_number == "12345" + assert row.proposal_number_observed_at == _at(0) + assert row.esaf_number == "67890" + assert row.esaf_number_observed_at == _at(1) + assert row.esaf_doi_number == "10.1234/esaf.67890" + assert row.esaf_doi_number_observed_at == _at(2) + assert row.created_at == _at(3) + assert row.updated_at == _at(3) + + +@pytest.mark.unit +async def test_upsert_accepts_a_partial_reading() -> None: + """Only ProposalNumber was populated at genesis; ESAFNumber / + ESAFDOINumber still read the substrate's own "Unknown" placeholder + (already resolved to None upstream, before this store ever sees + it). Each pair is independently nullable for exactly this reason.""" + store = InMemoryExperimentIdentityStore() + run_id = uuid4() + + await store.upsert( + run_id=run_id, + proposal_number="12345", + proposal_number_observed_at=_at(0), + esaf_number=None, + esaf_number_observed_at=None, + esaf_doi_number=None, + esaf_doi_number_observed_at=None, + created_at=_at(0), + ) + + row = await store.get(run_id) + assert row is not None + assert row.proposal_number == "12345" + assert row.esaf_number is None + assert row.esaf_number_observed_at is None + assert row.esaf_doi_number is None + assert row.esaf_doi_number_observed_at is None + + +@pytest.mark.unit +async def test_get_absent_run_id_returns_none() -> None: + store = InMemoryExperimentIdentityStore() + assert await store.get(uuid4()) is None + + +@pytest.mark.unit +async def test_upsert_overwrites_and_preserves_created_at() -> None: + """A second upsert for the same run_id (a retry) overwrites every + value column but keeps the ORIGINAL created_at, mirroring the + Postgres adapter's `ON CONFLICT DO UPDATE` (which never touches the + column). `updated_at` on the UPDATE branch is the STORE's own clock, + mirroring `InMemoryCapturePathStore`'s identical convention.""" + store = InMemoryExperimentIdentityStore() + run_id = uuid4() + await store.upsert( + run_id=run_id, + proposal_number="111", + proposal_number_observed_at=_at(0), + esaf_number=None, + esaf_number_observed_at=None, + esaf_doi_number=None, + esaf_doi_number_observed_at=None, + created_at=_at(0), + ) + before_second_upsert = datetime.now(tz=UTC) + + await store.upsert( + run_id=run_id, + proposal_number="222", + proposal_number_observed_at=_at(5), + esaf_number="333", + esaf_number_observed_at=_at(5), + esaf_doi_number=None, + esaf_doi_number_observed_at=None, + created_at=_at(5), + ) + + row = await store.get(run_id) + assert row is not None + assert row.proposal_number == "222" + assert row.esaf_number == "333" + assert row.created_at == _at(0) + assert row.updated_at >= before_second_upsert + + +@pytest.mark.unit +async def test_load_run_experiment_identity_returns_the_row_when_present() -> None: + store = InMemoryExperimentIdentityStore() + run_id = uuid4() + await store.upsert( + run_id=run_id, + proposal_number="12345", + proposal_number_observed_at=_at(0), + esaf_number=None, + esaf_number_observed_at=None, + esaf_doi_number=None, + esaf_doi_number_observed_at=None, + created_at=_at(0), + ) + + identity = await load_run_experiment_identity(store, run_id) + assert identity is not None + assert identity.proposal_number == "12345" + + +@pytest.mark.unit +async def test_load_run_experiment_identity_returns_none_when_absent() -> None: + """Unlike `load_run_capture_path`, no tombstone placeholder: none of + these three values is personal data, so a plain `None` is the + honest "nothing recorded" signal.""" + store = InMemoryExperimentIdentityStore() + assert await load_run_experiment_identity(store, uuid4()) is None diff --git a/apps/api/tests/unit/run/test_get_run_handler.py b/apps/api/tests/unit/run/test_get_run_handler.py index 98151aa8f1..296f08ca9e 100644 --- a/apps/api/tests/unit/run/test_get_run_handler.py +++ b/apps/api/tests/unit/run/test_get_run_handler.py @@ -18,6 +18,7 @@ from cora.run import RunHandlers, UnauthorizedError, wire_run from cora.run.aggregates.run import ( InMemoryCapturePathStore, + InMemoryExperimentIdentityStore, Run, RunName, RunStatus, @@ -77,7 +78,11 @@ async def test_handler_returns_run_for_known_id_with_subject() -> None: store = InMemoryEventStore() await _seed_run(store, _RUN_ID, plan_id=_PLAN_ID, subject_id=_SUBJECT_ID) deps = build_deps(ids=[_RUN_ID], now=_NOW, event_store=store) - handler = get_run.bind(deps, capture_path_store=InMemoryCapturePathStore()) + handler = get_run.bind( + deps, + capture_path_store=InMemoryCapturePathStore(), + experiment_identity_store=InMemoryExperimentIdentityStore(), + ) view = await handler( GetRun(run_id=_RUN_ID), principal_id=_PRINCIPAL_ID, @@ -101,7 +106,11 @@ async def test_handler_returns_run_for_known_id_without_subject() -> None: store = InMemoryEventStore() await _seed_run(store, _RUN_ID, plan_id=_PLAN_ID, subject_id=None, name="Dark field") deps = build_deps(ids=[_RUN_ID], now=_NOW, event_store=store) - handler = get_run.bind(deps, capture_path_store=InMemoryCapturePathStore()) + handler = get_run.bind( + deps, + capture_path_store=InMemoryCapturePathStore(), + experiment_identity_store=InMemoryExperimentIdentityStore(), + ) view = await handler( GetRun(run_id=_RUN_ID), principal_id=_PRINCIPAL_ID, @@ -114,7 +123,11 @@ async def test_handler_returns_run_for_known_id_without_subject() -> None: @pytest.mark.unit async def test_handler_returns_none_for_unknown_id() -> None: deps = build_deps(ids=[_RUN_ID], now=_NOW) - handler = get_run.bind(deps, capture_path_store=InMemoryCapturePathStore()) + handler = get_run.bind( + deps, + capture_path_store=InMemoryCapturePathStore(), + experiment_identity_store=InMemoryExperimentIdentityStore(), + ) view = await handler( GetRun(run_id=uuid4()), principal_id=_PRINCIPAL_ID, @@ -128,7 +141,11 @@ async def test_handler_authorizes_with_query_name_and_default_conduit() -> None: tracking = RecordingAuthorize() deps = build_deps(ids=[_RUN_ID], now=_NOW, authz=tracking) - handler = get_run.bind(deps, capture_path_store=InMemoryCapturePathStore()) + handler = get_run.bind( + deps, + capture_path_store=InMemoryCapturePathStore(), + experiment_identity_store=InMemoryExperimentIdentityStore(), + ) await handler( GetRun(run_id=uuid4()), principal_id=_PRINCIPAL_ID, @@ -141,7 +158,11 @@ async def test_handler_authorizes_with_query_name_and_default_conduit() -> None: @pytest.mark.unit async def test_handler_raises_unauthorized_on_deny() -> None: deps = build_deps(ids=[_RUN_ID], now=_NOW, deny=True) - handler = get_run.bind(deps, capture_path_store=InMemoryCapturePathStore()) + handler = get_run.bind( + deps, + capture_path_store=InMemoryCapturePathStore(), + experiment_identity_store=InMemoryExperimentIdentityStore(), + ) with pytest.raises(UnauthorizedError) as exc_info: await handler( GetRun(run_id=uuid4()), diff --git a/apps/api/tests/unit/run/test_get_run_route.py b/apps/api/tests/unit/run/test_get_run_route.py index a2914d583e..9384a4924c 100644 --- a/apps/api/tests/unit/run/test_get_run_route.py +++ b/apps/api/tests/unit/run/test_get_run_route.py @@ -9,12 +9,14 @@ `capture_code` / `observed_capture_path` resolution itself happens inside `get_run`'s `Handler` (`RunView`, per `handler.py`'s own -docstring), not at this layer: `get_run.bind(deps, capture_path_store=...)` -does the vault touch, and this route only destructures the already-composed -`RunView` into its wire DTO. These tests pin THAT destructuring for the -three outcomes -- witnessed + vault row -> real path; witnessed + no -row -> tombstone; conducted -> both fields `None` -- by building the -handler with a store pre-seeded (or not) before calling the route. +docstring), not at this layer: `get_run.bind(deps, capture_path_store=..., +experiment_identity_store=...)` does the vault touch, and this route +only destructures the already-composed `RunView` into its wire DTO. +These tests pin THAT destructuring for the three outcomes -- witnessed ++ vault row -> real path; witnessed + no row -> tombstone; conducted -> +both fields `None` -- by building the handler with a store pre-seeded +(or not) before calling the route. Also covers the slice 14a +proposal/ESAF/ESAF-DOI fields' destructuring the same way. """ from datetime import UTC, datetime @@ -25,7 +27,11 @@ from cora.infrastructure.adapters.in_memory_event_store import InMemoryEventStore from cora.infrastructure.event_envelope import to_new_event from cora.infrastructure.routing import NIL_SENTINEL_ID -from cora.run.aggregates.run import UNOBSERVED_CAPTURE_PATH, InMemoryCapturePathStore +from cora.run.aggregates.run import ( + UNOBSERVED_CAPTURE_PATH, + InMemoryCapturePathStore, + InMemoryExperimentIdentityStore, +) from cora.run.aggregates.run.events import RunStarted, event_type_name, to_payload from cora.run.features import get_run from cora.run.features.get_run.route import get_runs @@ -80,7 +86,11 @@ async def test_get_run_route_resolves_the_real_path_when_the_vault_has_a_row() - observed_at=_NOW, created_at=_NOW, ) - handler = get_run.bind(deps, capture_path_store=capture_path_store) + handler = get_run.bind( + deps, + capture_path_store=capture_path_store, + experiment_identity_store=InMemoryExperimentIdentityStore(), + ) response = await get_runs( run_id, @@ -94,13 +104,79 @@ async def test_get_run_route_resolves_the_real_path_when_the_vault_has_a_row() - assert response.observed_capture_path == "/data/2026-01-Smith-12345/scan_001.h5" +@pytest.mark.unit +async def test_get_run_route_resolves_the_experiment_identity_when_the_vault_has_a_row() -> None: + run_id = uuid4() + store = InMemoryEventStore() + await _seed_run(store, run_id, capture_code="2bmb-tomoscan") + deps = build_deps(ids=[run_id], now=_NOW, event_store=store) + experiment_identity_store = InMemoryExperimentIdentityStore() + await experiment_identity_store.upsert( + run_id=run_id, + proposal_number="12345", + proposal_number_observed_at=_NOW, + esaf_number="67890", + esaf_number_observed_at=_NOW, + esaf_doi_number=None, + esaf_doi_number_observed_at=None, + created_at=_NOW, + ) + handler = get_run.bind( + deps, + capture_path_store=InMemoryCapturePathStore(), + experiment_identity_store=experiment_identity_store, + ) + + response = await get_runs( + run_id, + handler, + _CORRELATION_ID, + _PRINCIPAL_ID, + NIL_SENTINEL_ID, + ) + + assert response.proposal_number == "12345" + assert response.proposal_number_observed_at == _NOW + assert response.esaf_number == "67890" + assert response.esaf_doi_number is None + + +@pytest.mark.unit +async def test_get_run_route_experiment_identity_is_none_when_the_vault_has_no_row() -> None: + run_id = uuid4() + store = InMemoryEventStore() + await _seed_run(store, run_id, capture_code="2bmb-tomoscan-3") + deps = build_deps(ids=[run_id], now=_NOW, event_store=store) + handler = get_run.bind( + deps, + capture_path_store=InMemoryCapturePathStore(), + experiment_identity_store=InMemoryExperimentIdentityStore(), + ) + + response = await get_runs( + run_id, + handler, + _CORRELATION_ID, + _PRINCIPAL_ID, + NIL_SENTINEL_ID, + ) + + assert response.proposal_number is None + assert response.esaf_number is None + assert response.esaf_doi_number is None + + @pytest.mark.unit async def test_get_run_route_resolves_the_tombstone_when_the_vault_has_no_row() -> None: run_id = uuid4() store = InMemoryEventStore() await _seed_run(store, run_id, capture_code="2bmb-tomoscan-2") deps = build_deps(ids=[run_id], now=_NOW, event_store=store) - handler = get_run.bind(deps, capture_path_store=InMemoryCapturePathStore()) + handler = get_run.bind( + deps, + capture_path_store=InMemoryCapturePathStore(), + experiment_identity_store=InMemoryExperimentIdentityStore(), + ) response = await get_runs( run_id, @@ -123,7 +199,11 @@ async def test_get_run_route_a_conducted_run_has_no_capture_code_and_no_tombston store = InMemoryEventStore() await _seed_run(store, run_id, capture_code=None, name="conducted-run") deps = build_deps(ids=[run_id], now=_NOW, event_store=store) - handler = get_run.bind(deps, capture_path_store=InMemoryCapturePathStore()) + handler = get_run.bind( + deps, + capture_path_store=InMemoryCapturePathStore(), + experiment_identity_store=InMemoryExperimentIdentityStore(), + ) response = await get_runs( run_id, @@ -142,7 +222,11 @@ async def test_get_run_route_raises_404_for_unknown_run() -> None: from fastapi import HTTPException deps = build_deps(ids=[uuid4()], now=_NOW) - handler = get_run.bind(deps, capture_path_store=InMemoryCapturePathStore()) + handler = get_run.bind( + deps, + capture_path_store=InMemoryCapturePathStore(), + experiment_identity_store=InMemoryExperimentIdentityStore(), + ) with pytest.raises(HTTPException) as exc_info: await get_runs( diff --git a/apps/api/tests/unit/test_settings.py b/apps/api/tests/unit/test_settings.py index aff52d0fff..992bc227d6 100644 --- a/apps/api/tests/unit/test_settings.py +++ b/apps/api/tests/unit/test_settings.py @@ -413,3 +413,72 @@ def test_settings_capture_baseline_recording_enabled_reads_env( monkeypatch.setenv("CAPTURE_BASELINE_RECORDING_ENABLED", "true") settings = Settings() assert settings.capture_baseline_recording_enabled is True + + +# --------------------------------------------------------------------------- +# capture_experiment_identity_pvs: proposal / ESAF / ESAF-DOI (slice 14a) +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +def test_settings_capture_experiment_identity_defaults_are_empty_and_off( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A generic boot declares no experiment-identity PVs and vaults nothing.""" + monkeypatch.delenv("CAPTURE_EXPERIMENT_IDENTITY_PVS", raising=False) + monkeypatch.delenv("CAPTURE_EXPERIMENT_IDENTITY_RECORDING_ENABLED", raising=False) + + settings = Settings() + + assert settings.capture_experiment_identity_pvs == {} + assert settings.capture_experiment_identity_recording_enabled is False + + +@pytest.mark.unit +def test_settings_capture_experiment_identity_pvs_reads_role_keyed_json( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Outer key is the capture code, inner dict is the closed role -> + PV vocabulary (`proposal_number` / `esaf_number` / `esaf_doi_number`).""" + monkeypatch.setenv( + "CAPTURE_EXPERIMENT_IDENTITY_PVS", + '{"2bmb-tomoscan": {' + '"proposal_number": "2bmb:TomoScan:ProposalNumber", ' + '"esaf_number": "2bmb:TomoScan:ESAFNumber", ' + '"esaf_doi_number": "2bmb:TomoScan:ESAFDOINumber"' + "}}", + ) + settings = Settings() + assert settings.capture_experiment_identity_pvs == { + "2bmb-tomoscan": { + "proposal_number": "2bmb:TomoScan:ProposalNumber", + "esaf_number": "2bmb:TomoScan:ESAFNumber", + "esaf_doi_number": "2bmb:TomoScan:ESAFDOINumber", + } + } + + +@pytest.mark.unit +def test_settings_capture_experiment_identity_pvs_rejects_unrecognized_role( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A typo'd role must fail at boot: the reader dispatches on exactly + the three closed role names and would otherwise silently never read + an unrecognized one, with no error anywhere.""" + import pydantic + + monkeypatch.setenv( + "CAPTURE_EXPERIMENT_IDENTITY_PVS", + '{"2bmb-tomoscan": {"proposal_numberr": "2bmb:TomoScan:ProposalNumber"}}', + ) + with pytest.raises(pydantic.ValidationError, match="capture_experiment_identity_pvs has roles"): + Settings() + + +@pytest.mark.unit +def test_settings_capture_experiment_identity_recording_enabled_reads_env( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("CAPTURE_EXPERIMENT_IDENTITY_RECORDING_ENABLED", "true") + settings = Settings() + assert settings.capture_experiment_identity_recording_enabled is True diff --git a/infra/atlas/migrations/20260816140958_init_run_experiment_identity.sql b/infra/atlas/migrations/20260816140958_init_run_experiment_identity.sql new file mode 100644 index 0000000000..48c10ad48d --- /dev/null +++ b/infra/atlas/migrations/20260816140958_init_run_experiment_identity.sql @@ -0,0 +1,104 @@ +-- Slice 14a (memory/project_witnessed_run_prelive_slices.md): vault for +-- a witnessed Run's proposal / ESAF / ESAF-DOI experiment identity. +-- +-- These three PVs (`ProposalNumber`, `ESAFNumber`, `ESAFDOINumber` under +-- `2bmb:TomoScan:`) are stamped by dmagic from APS scheduling data, not +-- by the IOC, and RunWitness has no operator to ask for a proposal the +-- way start_run's operator-supplied external_refs does. Auto-harvesting +-- them onto RunStarted would put an unerasable, re-identifying fact +-- (D0: a proposal number plus a timestamp is a strong join key against +-- public APS scheduling data) into an immutable, INSERT-only event. +-- ESAFDOINumber was checked and is populated from an internal, +-- authenticated APS API (EsafApsDbApi), not a DOI registration agency; +-- unconfirmed as a genuinely resolvable public identifier, so it vaults +-- alongside the other two rather than riding an event as its own scheme. +-- +-- A SIBLING table to run_capture_path, not a generalization of it: a +-- capture path and a proposal/ESAF number are different kinds of fact +-- (see experiment_identity.py's own module docstring for the full +-- argument). Same PII-vault-shaped posture and write-once-at-promotion +-- timing, for a different reason. +-- +-- Schema decisions (mirroring run_capture_path's init migration): +-- - run_id is both PK and (application-level) link to the Run +-- aggregate's stream_id. No SQL FK to events.stream_id because the +-- events table is INSERT-only at the role level per +-- project_immutability_guarantee; FK enforcement is application +-- discipline. +-- - Each of proposal_number / esaf_number / esaf_doi_number is independently +-- NULLABLE, paired with its own *_observed_at (the substrate's own +-- reading time, never CORA's clock): a deployment may configure +-- fewer than three roles, or the substrate may report "Unknown" / +-- empty for one PV while another reads a real value. CORA's own +-- reader treats "Unknown" and empty as absent and never writes them +-- here (see cora.api._capture_experiment_identity_reader); this +-- table's NULL is the "absent" state, not the substrate's literal. +-- - CHECK bounds are defense-in-depth (a NULL value passes a CHECK +-- unconditionally in Postgres, so these do not force presence): +-- 200 chars for proposal_number / esaf_number (matches +-- shared.identifier.IDENTIFIER_VALUE_MAX_LENGTH's bound for a +-- comparable free-form identifier value), 500 for esaf_doi_number (a DOI +-- suffix can run longer than a bare proposal/ESAF number). +-- - No forgotten_at / soft-delete column, mirroring run_capture_path: +-- it would itself be identifying ("this Run's proposal existed and +-- was erased on Y"). +-- - created_at is application-supplied (the promotion's own clock +-- read). updated_at defaults to now() for a future rename/rewrite +-- path. +-- +-- RLS posture (defense-in-depth, mirroring run_capture_path exactly): +-- - ENABLE + FORCE ROW LEVEL SECURITY. +-- - Two flat permissive policies for cora_app (read + write), both +-- USING (true) for v1, cora_app being the only runtime role. +-- +-- No erasure slice ships in this commit (rule-of-three; nothing calls +-- DELETE yet), but DELETE is granted now so a future forget-style slice +-- needs no follow-up grant migration. + +CREATE TABLE run_experiment_identity ( + run_id UUID PRIMARY KEY, + proposal_number TEXT CHECK (length(proposal_number) BETWEEN 1 AND 200), + proposal_number_observed_at TIMESTAMPTZ, + esaf_number TEXT CHECK (length(esaf_number) BETWEEN 1 AND 200), + esaf_number_observed_at TIMESTAMPTZ, + esaf_doi_number TEXT CHECK (length(esaf_doi_number) BETWEEN 1 AND 500), + esaf_doi_number_observed_at TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL, + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +COMMENT ON TABLE run_experiment_identity IS + 'Vault for a witnessed Run''s proposal/ESAF/ESAF-DOI experiment identity (memory/project_witnessed_run_prelive_slices.md, slice 14a). Mutable. No SQL FK to events (INSERT-only role); run_id matches the Run aggregate stream_id by application discipline.'; +COMMENT ON COLUMN run_experiment_identity.run_id IS + 'Matches Run aggregate stream_id. One row per witnessed Run that has at least one resolved experiment-identity value.'; +COMMENT ON COLUMN run_experiment_identity.proposal_number IS + 'The APS beamtime proposal number read from 2bmb:TomoScan:ProposalNumber. Never on an event: auto-harvested with no operator gesture behind it, and re-identifying alongside a timestamp against public APS scheduling data (D0).'; +COMMENT ON COLUMN run_experiment_identity.proposal_number_observed_at IS + 'The substrate''s own timestamp for this reading (Measurement.produced_at), not CORA''s clock. dmagic writes this PV from APS scheduling and it persists across beamtimes until overwritten; this column is the only staleness evidence available.'; +COMMENT ON COLUMN run_experiment_identity.esaf_number IS + 'The ESAF (Experiment Safety Assessment Form) number read from 2bmb:TomoScan:ESAFNumber. Same posture as proposal_number.'; +COMMENT ON COLUMN run_experiment_identity.esaf_number_observed_at IS + 'The substrate''s own timestamp for this reading; see proposal_number_observed_at.'; +COMMENT ON COLUMN run_experiment_identity.esaf_doi_number IS + 'The value read from 2bmb:TomoScan:ESAFDOINumber. Not confirmed as a publicly resolvable DOI (the upstream dmagic source reads it from an internal, authenticated APS API, not a DOI registry); vaulted alongside the other two rather than treated as a public identifier.'; +COMMENT ON COLUMN run_experiment_identity.esaf_doi_number_observed_at IS + 'The substrate''s own timestamp for this reading; see proposal_number_observed_at.'; + +-- Mutable PII-vault-shaped table: cora_app gets full CRUD. DELETE is the +-- future erasure mechanism; UPDATE lets a later slice correct/rewrite a row. +GRANT SELECT, INSERT, UPDATE, DELETE ON run_experiment_identity TO cora_app; + +-- Row-Level Security: defense-in-depth. +ALTER TABLE run_experiment_identity ENABLE ROW LEVEL SECURITY; +ALTER TABLE run_experiment_identity FORCE ROW LEVEL SECURITY; + +CREATE POLICY run_experiment_identity_cora_app_read + ON run_experiment_identity FOR SELECT + TO cora_app + USING (true); + +CREATE POLICY run_experiment_identity_cora_app_write + ON run_experiment_identity FOR ALL + TO cora_app + USING (true) + WITH CHECK (true); diff --git a/infra/atlas/migrations/atlas.sum b/infra/atlas/migrations/atlas.sum index 21fea24f03..e88b633424 100644 --- a/infra/atlas/migrations/atlas.sum +++ b/infra/atlas/migrations/atlas.sum @@ -1,4 +1,4 @@ -h1:MQxuBVTYB6e2yqkFPUCoiHFLXRHHZvOkJ684l9dXcIw= +h1:bZIKbGxK2kbm8VUiNorP5yM7ZJASeOkKcsFYhWjincI= 20260509120000_init_events.sql h1:GmgCZKfaqXu1m96/cKAks2vhaLWTdEaHTLkFtUo9FXg= 20260509170000_init_idempotency.sql h1:Nbu8DIE4Sv1WiHw3G22+tYffPhKc5Jryw3PMK8wB2zY= 20260510010000_add_event_id.sql h1:RbtYP6uMnOB20zhJ9dNXUi4YVqbmlEzf562pmygnRW8= @@ -166,3 +166,4 @@ h1:MQxuBVTYB6e2yqkFPUCoiHFLXRHHZvOkJ684l9dXcIw= 20260814041035_add_proj_run_summary_conduct_mode.sql h1:WVbP5BfF78sLkryrrmmIMUOpetNnUfGIgsFQpZsKCOw= 20260816094243_init_run_capture_path.sql h1:dXt030Xj03KkoJvRpF7yGFKuRAHQyDfQzlAyoGLWn9k= 20260816094314_add_proj_run_summary_capture_code.sql h1:odo/K4ZB5OmnGRD9V0BY/uvVxBtKUlW+j3WXEIb0F0M= +20260816140958_init_run_experiment_identity.sql h1:puAdTkKEu6pb7Zm7fk0jGFZrrrLtIsTUkswW1W885zA=