diff --git a/apps/api/openapi.json b/apps/api/openapi.json index b3e3340245..1e8992d03b 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).", + "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`.", "properties": { "campaign_id": { "anyOf": [ @@ -14287,6 +14287,17 @@ ], "title": "Campaign Id" }, + "capture_code": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Capture Code" + }, "effective_parameters": { "additionalProperties": true, "title": "Effective Parameters", @@ -14302,6 +14313,17 @@ "title": "Name", "type": "string" }, + "observed_capture_path": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Observed Capture Path" + }, "override_parameters": { "additionalProperties": true, "title": "Override Parameters", @@ -14378,6 +14400,18 @@ "description": "Campaign this Run is a member of. Set at start time (StartRun.campaign_id) or post-hoc (add_run_to_campaign). NULL for standalone Runs. Campaign membership snapshot.", "title": "Campaign Id" }, + "capture_code": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "description": "Deployment-declared capture identifier a witnessed genesis stamps onto external_refs (slice 13). NULL for a Conducted Run. NOT the observed capture file path: that value is personal data and is resolved per-run via `GET /runs/{run_id}`, never surfaced on this bulk list.", + "title": "Capture Code" + }, "created_at": { "format": "date-time", "title": "Created At", diff --git a/apps/api/src/cora/api/_capture_observer.py b/apps/api/src/cora/api/_capture_observer.py index 9ae3e717e1..9049f5afbe 100644 --- a/apps/api/src/cora/api/_capture_observer.py +++ b/apps/api/src/cora/api/_capture_observer.py @@ -22,7 +22,14 @@ the substrate is bypassing its own beam preconditions for this capture code, decoded via the same `binary_code` the `abort` role already uses (2-BM's `Testing` PV is the identical `DBR_ENUM` record type as -`AbortScan`). +`AbortScan`). The optional `full_file_name` role (slice 13) pumps its +own `CapturePathObservation`: a text reading of the areaDetector file +plugin's own filename readback (`2bmSP2:HDF1:FullFileName_RBV`, NOT +tomoscan's own `FullFileName`, which is written too late relative to +CORA's terminal -- see `_run_witness.py`'s "Capture path pairing" +section for the full argument). `observed_path` is PERSONAL DATA; this +module never logs it, only its length when rejecting a suspect +reading. ## One deliberate inversion from the Enclosure precedent @@ -96,11 +103,13 @@ import math from typing import TYPE_CHECKING +from cora.infrastructure.logging import get_logger from cora.operation.ports.control_port import ControlNotConnectedError from cora.run.ports.capture_observer import ( AnyCaptureObservation, CaptureLifecycleObservation, CaptureObserverScope, + CapturePathObservation, CapturePhase, CapturePreconditionBypassObservation, CaptureProgressObservation, @@ -112,24 +121,52 @@ from cora.operation.ports.control_port import ControlPort, Measurement +_log = get_logger(__name__) + _SOURCE_KIND = "EpicsPv" ROLE_STATUS = "status" ROLE_ABORT = "abort" ROLE_IMAGES_SAVED = "images_saved" ROLE_IMAGES_COLLECTED = "images_collected" ROLE_TESTING = "testing" +ROLE_FULL_FILE_NAME = "full_file_name" """CORA-owned role keys, matching `Settings.capture_watch_pvs`'s documented example. Module-public (not `_`-prefixed) because other composition-root modules read observations back out, or dispatch decoders, by these same keys and must not carry their own copy of the literal strings: `RunWitnessRecorder._build_progress_snapshot` (`_run_witness.py`) reads `ROLE_IMAGES_SAVED` / `ROLE_IMAGES_COLLECTED`, and `capture_watch_preflight` -dispatches its per-role decode check on all five. Import these, not +dispatches its per-role decode check on all six. Import these, not `_PROGRESS_ROLES` below, so a rename or a new role here cannot silently desync from either reader. `server_running` stays declared-and-unread: -tool liveness is a different concern from capture progress (slice 10).""" +tool liveness is a different concern from capture progress (slice 10). + +`ROLE_FULL_FILE_NAME` (slice 13) names the raw substrate PV +(`FullFileName_RBV`) rather than the semantic concept it feeds +(`CapturePathObservation`, a `run_capture_path` vault row): matches the +existing style, where a role key mirrors the wire (`abort`, `testing`, +`images_saved`) and the domain type it produces is named for the fact, +not the PV. +""" _PROGRESS_ROLES = (ROLE_IMAGES_SAVED, ROLE_IMAGES_COLLECTED) +FULL_FILE_NAME_TRUNCATION_THRESHOLD = 511 +"""The areaDetector file plugin's `FullFileName_RBV` is a DBR_CHAR +waveform with `NELM=512` (confirmed against ADCore's `NDFile.template`, +2026-08-16 -- NOT the 256 tomoscan's own `FullFileName` uses; the two +PVs share no wire-shape assumption). Up to 511 usable characters remain +after the NUL terminator. EPICS gives no separate "this got truncated" +flag: a decoded string that fills the whole buffer is indistinguishable +from one that was cut off mid-path. Rather than risk a truncated path +silently looking like a good one, ANY decoded value at or over this +length is treated as suspect and rejected (see +`_from_full_file_name_reading`). Conservative: a genuine 511-character +path would false-reject, but real 2-BM paths are far under this bound. + +Module-public (not `_`-prefixed): `capture_watch_preflight` imports this +directly so its own truncation check can never drift from production's. +""" + # Conventional EPICS binary state labels, mirroring # `_enclosure_permit_observer._PERMITTED_LABELS` / `_NOT_PERMITTED_LABELS` # exactly: a DBR_ENUM reading through `EpicsCaControlPort` reaches this @@ -270,7 +307,10 @@ class ControlPortCaptureObserver: (also optional, independently declared per code) pumps `CapturePreconditionBypassObservation` readings, a tri-state claim rather than a phase or a counter; see that dataclass's own - docstring. + docstring. The `full_file_name` role (slice 13, also optional, + independently declared per code) pumps `CapturePathObservation` + readings, a text claim carrying personal data; see that dataclass's + own docstring. `server_running` stays declared and unread (tool liveness, not capture progress). """ @@ -295,6 +335,11 @@ def __init__( for code, roles in capture_pvs.items() if ROLE_TESTING in roles } + self._full_file_name_pvs = { + code: roles[ROLE_FULL_FILE_NAME] + for code, roles in capture_pvs.items() + if ROLE_FULL_FILE_NAME in roles + } self._progress_pvs = { code: filtered for code, roles in capture_pvs.items() @@ -324,6 +369,11 @@ async def _drain(self, scope: CaptureObserverScope) -> AsyncGenerator[AnyCapture for code in sorted(scope.capture_codes) if code in self._testing_pvs ] + full_file_name_pvs = [ + (code, self._full_file_name_pvs[code]) + for code in sorted(scope.capture_codes) + if code in self._full_file_name_pvs + ] progress_pvs = [ (code, role, pv) for code in sorted(scope.capture_codes) @@ -335,6 +385,10 @@ async def _drain(self, scope: CaptureObserverScope) -> AsyncGenerator[AnyCapture [asyncio.create_task(self._pump(code, pv, queue)) for code, pv in pvs] + [asyncio.create_task(self._pump_abort(code, pv, queue)) for code, pv in abort_pvs] + [asyncio.create_task(self._pump_testing(code, pv, queue)) for code, pv in testing_pvs] + + [ + asyncio.create_task(self._pump_full_file_name(code, pv, queue)) + for code, pv in full_file_name_pvs + ] + [ asyncio.create_task(self._pump_progress(code, role, pv, queue)) for code, role, pv in progress_pvs @@ -463,6 +517,37 @@ async def _pump_testing( finally: queue.put_nowait(_PUMP_DONE) + async def _pump_full_file_name( + self, + code: str, + pv: str, + queue: asyncio.Queue[AnyCaptureObservation | _PumpDone], + ) -> None: + """Sibling pump for the optional `full_file_name` role (slice 13). + + Mirrors `_pump_testing` exactly, for the same reason: a + disconnect must not erase the last retained reading, since the + dual-clock guard `RunWitnessRecorder` applies (comparing this + reading's `observed_at` against the Run's own BEGUN time) needs + the last GOOD reading to survive a reconnect, not be replaced by + a synthesized "unreached". Unlike `_pump_abort` / `_pump`, no + `_from_full_file_name_reading` result is dropped for being + "no claim" the way an abort's clear reading is -- it can return + `None` (see that function), but that is a REJECTION (empty + string, suspected truncation, non-str value), not a "nothing + happened" no-op, so it is still correct to enqueue nothing for + it: there is no partial fact to carry. + """ + try: + async for reading in self._control_port.subscribe(pv): + observation = self._from_full_file_name_reading(code, pv, reading) + if observation is not None: + queue.put_nowait(observation) + except ControlNotConnectedError: + pass + finally: + queue.put_nowait(_PUMP_DONE) + async def _poll( self, code: str, @@ -579,6 +664,49 @@ def _from_testing_reading( source_id=pv, ) + def _from_full_file_name_reading( + self, code: str, pv: str, reading: Measurement + ) -> CapturePathObservation | None: + """A `full_file_name`-role reading (slice 13), rejected rather + than enqueued for three reasons, in order. + + 1. Not a string: the adapter's own text-waveform decode + (`text_addresses`) failed to apply or produced something + else. Never coerced; a non-text reading on this role is a + deployment misconfiguration, not a value to guess at. + 2. Empty string: the fresh-IOC-boot state (the file plugin has + never opened a file since the IOC started). A fine, ordinary + outcome, not an error -- just nothing to enqueue. + 3. Length at or over `FULL_FILE_NAME_TRUNCATION_THRESHOLD`: + indistinguishable from a wire truncation (see that + constant's own docstring). Logs the length only, NEVER the + value -- `observed_path` is personal data. + + `None` return means the caller enqueues nothing, matching + `_from_abort_reading` / `_from_progress_reading`'s fail-toward- + silence posture. + """ + value = reading.value + if not isinstance(value, str): + return None + if not value: + return None + if len(value) >= FULL_FILE_NAME_TRUNCATION_THRESHOLD: + _log.warning( + "capture_observer.full_file_name_suspected_truncated", + capture_code=code, + length=len(value), + ) + return None + return CapturePathObservation( + capture_code=code, + observed_path=value, + reach_tier=ReachTier.RELAYED, + observed_at=reading.produced_at, + source_kind=_SOURCE_KIND, + source_id=pv, + ) + def _probe_only(self, code: str, pv: str, reach_tier: ReachTier) -> CaptureLifecycleObservation: """A poll tick's result: reach evidence with no status claim.""" return CaptureLifecycleObservation( @@ -609,7 +737,9 @@ def _unreached(self, code: str, pv: str) -> CaptureLifecycleObservation: __all__ = [ + "FULL_FILE_NAME_TRUNCATION_THRESHOLD", "ROLE_ABORT", + "ROLE_FULL_FILE_NAME", "ROLE_IMAGES_COLLECTED", "ROLE_IMAGES_SAVED", "ROLE_STATUS", diff --git a/apps/api/src/cora/api/_run_witness.py b/apps/api/src/cora/api/_run_witness.py index 69eef5a8d2..ecfbffbf2b 100644 --- a/apps/api/src/cora/api/_run_witness.py +++ b/apps/api/src/cora/api/_run_witness.py @@ -184,6 +184,48 @@ fix but, again, a deliberate design decision to make later if this is ever observed in practice, not a wiring change. +## Capture path pairing (slice 13) + +Closes the "which file did this Run produce" findability gap (auto- +correlation stays deferred; this only makes the manual pairing +possible). Source is the areaDetector file plugin's own filename +readback (`full_file_name` role, `_capture_observer.py`), NOT +tomoscan's own `FullFileName`: upstream `tomoscan.py`'s `end_scan()` +writes `ScanStatus='Scan complete'` (CORA's terminal) FOUR statements +BEFORE it writes `FullFileName`, so a synchronous read at the terminal +against that PV would return the PREVIOUS scan's filename. The +areaDetector PV is written at file OPEN, i.e. before the terminal, so +it does not have this race -- but "written before the terminal" is not +by itself sufficient: the file plugin also fires independently of any +one capture, so a value retained from a PREVIOUS capture (or a +reconnect redelivering one) could still be sitting in +`_last_capture_path` when THIS capture's terminal arrives. + +The guard: `_promote` records the BEGUN observation's OWN substrate +time into `_begun_at[code]` (never CORA's clock -- comparing two +substrate timestamps from the same control system avoids a +CORA-host-vs-IOC clock-skew question this guard does not need to +answer). At the terminal, `_resolve_capture_path` accepts the retained +reading only if its `observed_at` is at or after that value. This is +the freshness guard the "CA's redeliver-on-resubscribe" residual above +names as a real fix "if this is ever observed in practice" for +progress readings -- implemented here, scoped to this one role, because +Finding A makes it load-bearing rather than optional: without it, this +feature would silently pair Runs with the WRONG file on every +reconnect-during-BEGUN window, not just degrade one row's ordering. + +`observed_path` is personal data (2-BM's directory layout embeds a +surname and a proposal number,`tomoscan_2bm.py:474-477`), so it is +never written to `RunCompleted` / `RunAborted` or any other event: it +goes to the `run_capture_path` PII vault (`CapturePathStore`, +mirroring `actor_profile` / `ProfileStore`) via `_write_capture_path`, +called at the end of `_record_outcome`'s success branch -- the outcome +has already committed by then, so (mirroring `_read_baseline`'s exact +posture) a vault-write failure is logged and never unwinds it. Gated +on the fifth kill switch, `capture_path_recording_enabled`. No log +line in this section ever includes `observed_path` itself; only +`capture_code` / `run_id` / lengths. + ## Retry + resilience Mirrors `run_enclosure_permit_monitor`: `observe()` ending (stream @@ -217,6 +259,7 @@ from cora.run.aggregates.run.state import ( CapturePreconditionBypassSnapshot, CaptureProgressSnapshot, + extract_capture_code, ) from cora.run.errors import UnauthorizedError from cora.run.features.list_runs.query import ListRuns @@ -226,6 +269,7 @@ from cora.run.ports.capture_observer import ( CaptureLifecycleObservation, CaptureObserverScope, + CapturePathObservation, CapturePhase, CapturePreconditionBypassObservation, CaptureProgressObservation, @@ -239,7 +283,7 @@ 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 FeedHeartbeatStore + from cora.run.aggregates.run import CapturePathStore, 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 @@ -249,12 +293,10 @@ ) from cora.run.features.truncate_run.handler import Handler as TruncateRunHandler from cora.run.ports.capture_observer import CaptureObserver - from cora.shared.identifier import Identifier _RECONNECT_DELAY_SECONDS = 5.0 _CAPTURE_PROGRESS_DEFAULT_FLUSH_TICK_SECONDS = 10.0 _PAGE_LIMIT = 100 -_CAPTURE_CODE_SCHEME = "capture-code" _log = get_logger(__name__) @@ -331,6 +373,7 @@ def __init__( settings: Settings, open_captures: dict[str, UUID] | None = None, baseline_reader: CaptureBaselineReader | None = None, + capture_path_store: CapturePathStore | None = None, ) -> None: self._deps = deps self._record_witnessed_run = record_witnessed_run @@ -341,6 +384,19 @@ def __init__( self._last_progress: dict[str, dict[str, CaptureProgressObservation]] = {} self._last_precondition_bypass: dict[str, CapturePreconditionBypassObservation] = {} self._baseline_reader = baseline_reader + self._capture_path_store = capture_path_store + 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. + Evicted in lockstep with `_begun_at` below in `_promote` / + `_truncate_stale` / `_record_outcome`'s success path.""" + self._begun_at: dict[str, datetime] = {} + """Slice 13: the BEGUN observation's OWN substrate time per + capture_code, recorded in `_promote`. The dual-clock guard in + `_resolve_capture_path` compares a retained + `CapturePathObservation.observed_at` against this value, never + against CORA's own clock (see this module's "Capture path + pairing" docstring section).""" def open_captures(self) -> dict[str, UUID]: """A snapshot of every capture_code currently open, mapped to @@ -460,12 +516,72 @@ def _build_precondition_bypass_snapshot( observed_at=observation.observed_at, ) + def observe_capture_path(self, observation: CapturePathObservation) -> None: + """Retain the latest `full_file_name`-role reading per + capture_code (slice 13), so a terminal can resolve it through + `_resolve_capture_path`. + + Gated on `run_witness_recording_enabled`, same as + `observe_progress`: shadow mode retains nothing because it + writes nothing. + + Evicted WITH `_begun_at`, unlike `_last_precondition_bypass`: + `full_file_name` describes one specific capture's output file, + so a reading retained across a capture boundary IS stale + evidence about the wrong capture, the same reasoning + `_last_progress` already applies. See `_promote` / + `_truncate_stale` / `_record_outcome`'s success path. + + Note this method does NOT itself apply the dual-clock guard: it + retains whatever arrives, latest-wins, exactly like + `observe_progress`. The guard is `_resolve_capture_path`'s job, + applied once, at the terminal -- not here, on every reading. + """ + if not self._settings.run_witness_recording_enabled: + return + self._last_capture_path[observation.capture_code] = observation + + def _resolve_capture_path(self, code: str) -> tuple[str, datetime] | None: + """The dual-clock guard (Finding A, slice 13): a retained + `full_file_name` reading is usable for `code`'s terminal only if + its OWN substrate time is at or after this code's own BEGUN + substrate time. Returns `(observed_path, observed_at)` when the + guard passes, `None` when there is nothing retained, `code` was + never promoted with a substrate time to compare against + (`_begun_at` has no entry), or the retained reading predates it. + + `None` is a legitimate, expected outcome (never observed yet, + or correctly rejected as belonging to a previous capture), not + an error; callers must not log it as one. + """ + observation = self._last_capture_path.get(code) + begun_at = self._begun_at.get(code) + if observation is None or begun_at is None: + return None + if observation.observed_at is None or observation.observed_at < begun_at: + return None + return observation.observed_path, observation.observed_at + async def _promote(self, observation: CaptureLifecycleObservation) -> None: # A prior capture's retained progress, if any, belongs to that # capture's own terminal, never to this one: clear before # promoting so a stale carry-over cannot ride onto the new # Run's eventual outcome. self._last_progress.pop(observation.capture_code, None) + # Slice 13: same reasoning for a retained full_file_name + # reading -- it describes the PREVIOUS capture's file, not this + # one's, until a fresh reading arrives. + self._last_capture_path.pop(observation.capture_code, None) + if observation.observed_at is not None: + # The dual-clock guard's reference point: recorded here, + # from this BEGUN's OWN substrate time, never CORA's clock. + self._begun_at[observation.capture_code] = observation.observed_at + else: + # No substrate time on this BEGUN means the guard has + # nothing to compare against; a stale prior entry must not + # be left standing to be compared against the WRONG + # promotion (see `_resolve_capture_path`). + self._begun_at.pop(observation.capture_code, None) plan_id = self._settings.capture_watch_plan_id if plan_id is None: # Unreachable when `_enforce_run_witness_recording_gate` @@ -570,6 +686,10 @@ async def _truncate_stale(self, observation: CaptureLifecycleObservation) -> Non # different capture and must not inherit counts that describe # the one being truncated. self._last_progress.pop(code, None) + # Slice 13: same reasoning, for the retained full_file_name + # reading and its BEGUN reference point. + self._last_capture_path.pop(code, None) + self._begun_at.pop(code, None) if stale_run_id is None: return @@ -656,12 +776,19 @@ async def _record_outcome(self, observation: CaptureLifecycleObservation) -> Non run_id=str(run_id), ) return + # Resolve and write BEFORE evicting: `_write_capture_path` reads + # `_last_capture_path` / `_begun_at` via `_resolve_capture_path`, + # so eviction must come after, mirroring the success-only + # ordering `_last_progress` already follows below. + await self._write_capture_path(code, run_id) self._open_captures.pop(code, None) # Success path only, mirroring `_open_captures` immediately # above: on failure both dicts stay populated so the next # BEGUN's truncate-then-promote clears them together, keeping # the two eviction states in lockstep. self._last_progress.pop(code, None) + self._last_capture_path.pop(code, None) + self._begun_at.pop(code, None) _log.info( "run_witness.outcome_recorded", capture_code=code, @@ -669,6 +796,65 @@ async def _record_outcome(self, observation: CaptureLifecycleObservation) -> Non observed_phase=str(observation.phase), ) + async def _write_capture_path(self, code: str, run_id: UUID) -> None: + """Slice 13: resolve and write the observed capture path for + `code`'s just-recorded terminal, if the dual-clock guard passes. + + By this point `record_witnessed_run_outcome` has already + succeeded, so (mirroring `_read_baseline`'s exact posture) a + failure here must be logged and must never unwind it. Gated on + BOTH a store being configured (`run_witness_lifespan` wires one + whenever `capture_watch_pvs` declares the `full_file_name` role + for at least one code) and the fifth kill switch, + `capture_path_recording_enabled`. + + No resolved value is a normal, expected outcome (never + observed, or correctly rejected by `_resolve_capture_path`), + logged at `info`, not `warning`: a missing filename is a fine + outcome, per the slice's own design lock; a wrong one is not, + which is exactly what the guard exists to prevent. + """ + if self._capture_path_store is None or not self._settings.capture_path_recording_enabled: + return + resolved = self._resolve_capture_path(code) + if resolved is None: + _log.info( + "run_witness.capture_path_unresolved", + capture_code=code, + run_id=str(run_id), + ) + return + observed_path, observed_at = resolved + try: + await self._capture_path_store.upsert( + run_id=run_id, + observed_path=observed_path, + observed_at=observed_at, + created_at=self._deps.clock.now(), + ) + except asyncio.CancelledError: + raise + except Exception as exc: + # `_log.exception` (unlike `_log.error`) renders the full + # traceback, whose final line is `str(exc)` -- and asyncpg + # appends a `DETAIL:` line to a CHECK-violation error that + # can include the failing row's own column VALUES. That + # value is `observed_path`, personal data, and this log + # sink is not the vault: it cannot be erased. Log the + # exception's class name only, never the object itself. + _log.error( + "run_witness.capture_path_write_failed", + capture_code=code, + run_id=str(run_id), + error_class=type(exc).__name__, + ) + return + _log.info( + "run_witness.capture_path_recorded", + capture_code=code, + run_id=str(run_id), + ) + def _build_progress_snapshot(self, code: str) -> CaptureProgressSnapshot | None: """The evidence a witnessed terminal carries for `code`: the last `collected` / `saved` progress readings `observe_progress` @@ -715,19 +901,6 @@ def _progress_fields( return observation.value, observation.commanded_total, observation.observed_at -def _extract_capture_code(external_refs: frozenset[Identifier]) -> str | None: - """Find the `Identifier(scheme="capture-code", ...)` entry's value. - - Defensive: `record_witnessed_run`'s decider always stamps exactly one, - so `None` should not happen for a Witnessed Run, but a missing ref - must not crash the boot-time rebuild. - """ - for ref in external_refs: - if ref.scheme == _CAPTURE_CODE_SCHEME: - return ref.value - return None - - async def rebuild_open_captures(deps: Kernel, *, list_runs: ListRunsHandler) -> dict[str, UUID]: """Page through every Running, Witnessed Run and return capture_code -> run_id for each one's `external_refs`. @@ -756,7 +929,7 @@ async def rebuild_open_captures(deps: Kernel, *, list_runs: ListRunsHandler) -> run: Run | None = await load_run(deps.event_store, item.run_id) if run is None: continue - capture_code = _extract_capture_code(run.external_refs) + capture_code = extract_capture_code(run.external_refs) if capture_code is not None: open_captures[capture_code] = item.run_id if page.next_cursor is None: @@ -786,7 +959,13 @@ async def run_witness_loop( reading so the NEXT genesis can stamp it; see `RunWitnessRecorder ._build_precondition_bypass_snapshot`): it has no `feeder` counterpart, since it is never written as an `AppendObservations` - row, only carried onto `RunStarted`. A `CaptureLifecycleObservation` on a phase in + row, only carried onto `RunStarted`. A `CapturePathObservation` + (slice 13) likewise goes only to `recorder.observe_capture_path()` + (retains the latest reading so a terminal can resolve it through + the dual-clock guard; see `RunWitnessRecorder._resolve_capture_path`): + no `feeder` counterpart either, since it never rides + `AppendObservations` -- it goes to the `run_capture_path` PII + vault, not the observation logbook. A `CaptureLifecycleObservation` on a phase in `_FLUSH_TRIGGER_PHASES` triggers `feeder.flush_capture()` BEFORE the recorder acts on it, so a capture's buffered progress trail is attributed to its Run before that Run can close or be replaced; @@ -812,6 +991,10 @@ async def run_witness_loop( if recorder is not None: recorder.observe_precondition_bypass(observation) continue + if isinstance(observation, CapturePathObservation): + if recorder is not None: + recorder.observe_capture_path(observation) + continue if feeder is not None and observation.phase in _FLUSH_TRIGGER_PHASES: try: await feeder.flush_capture(observation.capture_code) @@ -857,6 +1040,7 @@ async def run_witness_lifespan( capture_progress_flush_tick_seconds: float = _CAPTURE_PROGRESS_DEFAULT_FLUSH_TICK_SECONDS, control_port: ControlPort | None = None, capture_baseline_pvs: Mapping[str, Mapping[str, str]] | None = None, + capture_path_store: CapturePathStore | None = None, ) -> AsyncGenerator[None]: """Run the watcher as a background task for the app's lifetime. @@ -889,6 +1073,16 @@ async def run_witness_lifespan( fourth kill switch) -- declaring the PVs here is necessary but not sufficient, mirroring how declaring `capture_watch_pvs` alone does not turn on recording either. + + A supplied `capture_path_store` (slice 13) is handed straight to + the recorder with no extra required-params check: unlike + `capture_baseline_pvs`, there is no separate reader object to build + here (the observer already pumps `CapturePathObservation` whenever + a code's `capture_watch_pvs` declares `full_file_name`; the store + is only where the recorder writes the RESULT). Whether a write + actually happens is gated, same pattern as the fourth switch, + inside the recorder by `deps.settings.capture_path_recording_enabled` + (the fifth kill switch). """ if not capture_codes: yield @@ -948,6 +1142,7 @@ async def run_witness_lifespan( settings=deps.settings, open_captures=open_captures, baseline_reader=baseline_reader, + capture_path_store=capture_path_store, ) 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 d22506a411..d1a7ac35f2 100644 --- a/apps/api/src/cora/api/capture_watch_preflight.py +++ b/apps/api/src/cora/api/capture_watch_preflight.py @@ -74,6 +74,14 @@ - `testing` (`ROLE_TESTING`): `binary_code`, same decoder as `abort` (2-BM's `Testing` PV is the identical `DBR_ENUM` record type as `AbortScan`). BAD when it returns `None`. + - `full_file_name` (`ROLE_FULL_FILE_NAME`, slice 13): the value is + PERSONAL DATA (see `_capture_observer.py`), so this is the one role + whose printed `value` field is REDACTED to a length-only placeholder + rather than the raw reading -- `kind` and `element_count` still + print real. Verdict reports `text(len=N)`, `empty`, or + `suspected-truncated` (mirroring `_from_full_file_name_reading`'s + own truncation threshold); BAD only for `suspected-truncated`. A + non-str reading is BAD as `non-text`. - any other declared role (e.g. `server_running`, which production itself declares and never decodes): reports `kind` / `value` only, verdict `n/a`. Not decoding it here does not make it undecodable @@ -90,10 +98,12 @@ import contextlib import sys from dataclasses import dataclass, field -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, TypeGuard from cora.api._capture_observer import ( + FULL_FILE_NAME_TRUNCATION_THRESHOLD, ROLE_ABORT, + ROLE_FULL_FILE_NAME, ROLE_IMAGES_COLLECTED, ROLE_IMAGES_SAVED, ROLE_STATUS, @@ -235,11 +245,45 @@ async def _read_one( connected=True, kind=reading.kind, element_count=element_count, - value=reading.value, + value=_redacted_value(role, reading.value), verdict=verdict, ) +def _looks_like_a_filesystem_path(value: object) -> TypeGuard[str]: + """True for a string shaped like an absolute filesystem path. + + Defense-in-depth redaction trigger, independent of role/channel + key: a `capture_watch_pvs` or `capture_baseline_pvs` operator typo + on the intended role/channel name (`"full_filename"`, + `"fullFileName"`, or the `full_file_name` PV accidentally declared + under `capture_baseline_pvs`) would otherwise fall through to a + generic branch and print the real path -- precisely the shape of + mistake this tool exists to catch, in the one case where catching + it here matters most: this preflight command is what an operator + runs, screenshots, and pastes into a ticket while debugging exactly + this kind of misconfiguration. + """ + return isinstance(value, str) and value.startswith("/") + + +def _redacted_value(role: str, value: object) -> object: + """The printed `value` for a `capture_watch_pvs` `_PvReport` line: + real for every role except `full_file_name` (slice 13, PERSONAL + DATA), which prints a length-only placeholder instead. Also + redacts defensively when the role doesn't match but the value is + still path-shaped; see `_looks_like_a_filesystem_path`. Production + (`_capture_observer.py`) fails safe on a role-key typo (no pump is + created for an undeclared role), so the defensive branch here is + this tool's own, not a mirror of a production check. + """ + if role == ROLE_FULL_FILE_NAME: + return f"" if isinstance(value, str) else "" + if _looks_like_a_filesystem_path(value): + return f"" + return value + + def _decode_verdict( role: str, reading: Measurement, status_phases: Mapping[str, str] ) -> tuple[str, bool]: @@ -253,6 +297,8 @@ def _decode_verdict( if code is None: return "unrecognized", False return ("asserted" if code == 1 else "clear"), True + if role == ROLE_FULL_FILE_NAME: + return _full_file_name_verdict(reading.value) if role == ROLE_TESTING: code = binary_code(reading.value) if code is None: @@ -267,6 +313,22 @@ def _decode_verdict( return "n/a", True +def _full_file_name_verdict(value: object) -> tuple[str, bool]: + """The `full_file_name` role's decode check (slice 13): reports a + length-only verdict, NEVER the value, mirroring the same three + rejection reasons `_from_full_file_name_reading` applies in + production (`cora.api._capture_observer`) so this can never drift + from what the running system actually accepts. + """ + if not isinstance(value, str): + return "non-text", False + if not value: + return "empty", True + if len(value) >= FULL_FILE_NAME_TRUNCATION_THRESHOLD: + return "suspected-truncated", False + return f"text(len={len(value)})", True + + async def _read_one_baseline( control_port: ControlPort, code: str, @@ -317,7 +379,15 @@ async def _read_one_baseline( connected=True, kind=reading.kind, element_count=element_count, - value=reading.value, + # Defense-in-depth, not the primary guard: a `full_file_name` PV + # accidentally declared under `capture_baseline_pvs` (the wrong + # dict) would otherwise print its real, personal-data-bearing + # value here. See `_looks_like_a_filesystem_path`. + value=( + f"" + if _looks_like_a_filesystem_path(reading.value) + else reading.value + ), units=reading.units, verdict=verdict, group="baseline", diff --git a/apps/api/src/cora/api/main.py b/apps/api/src/cora/api/main.py index 19cb04fd8e..86c4c45fd0 100644 --- a/apps/api/src/cora/api/main.py +++ b/apps/api/src/cora/api/main.py @@ -436,6 +436,11 @@ def _enforce_run_witness_recording_gate(settings: Settings) -> None: baseline read happens exactly once, at the instant a capture promotes to a Run, so with no promotion there is nothing to attach a baseline reading to either. + + Also refuses to boot with capture_path_recording_enabled=True unless + 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. """ if settings.capture_progress_recording_enabled and not settings.run_witness_recording_enabled: msg = ( @@ -451,6 +456,13 @@ def _enforce_run_witness_recording_gate(settings: Settings) -> None: "promoted Run to attach to without it." ) raise RuntimeError(msg) + if settings.capture_path_recording_enabled and not settings.run_witness_recording_enabled: + msg = ( + "CAPTURE_PATH_RECORDING_ENABLED=true requires " + "RUN_WITNESS_RECORDING_ENABLED=true. An observed capture path " + "has no promoted Run's terminal to attach to without it." + ) + raise RuntimeError(msg) if not settings.run_witness_recording_enabled: return missing: list[str] = [] @@ -1217,6 +1229,7 @@ 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, ), ): yield diff --git a/apps/api/src/cora/infrastructure/config.py b/apps/api/src/cora/infrastructure/config.py index 79c072dcaf..803e626854 100644 --- a/apps/api/src/cora/infrastructure/config.py +++ b/apps/api/src/cora/infrastructure/config.py @@ -728,7 +728,8 @@ class Settings(BaseSettings): # "abort": "2bmb:TomoScan:AbortScan", # "images_saved": "2bmb:TomoScan:ImagesSaved", # "images_collected": "2bmb:TomoScan:ImagesCollected", - # "testing": "2bmb:TomoScan:Testing" + # "testing": "2bmb:TomoScan:Testing", + # "full_file_name": "2bmSP2:HDF1:FullFileName_RBV" # } # }' # @@ -739,6 +740,21 @@ class Settings(BaseSettings): # tomoscan is bypassing its own beam preconditions for this capture, # carried onto the witnessed genesis, never onto # `Observation.is_simulated`. See `cora.api._capture_observer`. + # + # `full_file_name` (slice 13, optional per code) is ALSO a DBR_CHAR + # waveform needing a `text_addresses` declaration, but deliberately + # NOT `2bmb:TomoScan:FullFileName` (tomoscan's own mirror of it): + # upstream `end_scan()` writes that PV four statements AFTER the + # `ScanStatus='Scan complete'` write that fires CORA's terminal, so a + # read there returns the PREVIOUS scan's filename. The areaDetector + # file plugin's own readback is written at file OPEN, before the + # terminal, and CORA's conducted path already reads the same PV + # family (`operation/acquisitions.py`), so this is the correct + # source, not a workaround. The value is PERSONAL DATA (2-BM's + # directory layout embeds a surname and a proposal number): it is + # never logged in full and never lands on an event; it goes to the + # `run_capture_path` PII vault via `RunWitnessRecorder`'s dual-clock + # guard. See `_run_witness.py`'s "Capture path pairing" section. capture_watch_pvs: dict[str, dict[str, str]] = {} # Genesis-baseline PVs (slice 12): a deployment-declared set read @@ -903,6 +919,22 @@ class Settings(BaseSettings): # `cora.api._capture_baseline_reader`. capture_baseline_recording_enabled: bool = False + # FIFTH, independent kill switch (slice 13): gates whether the + # `full_file_name` role's observed path is actually resolved and + # written to the `run_capture_path` PII vault at a witnessed Run's + # terminal. Default off. Refuses to boot if True without + # `run_witness_recording_enabled` also True (see + # `_enforce_run_witness_recording_gate`): the write happens at a + # promoted Run's terminal, so with no promotion there is no run_id + # to write against. Independently revocable from the other four + # switches because it is the one that writes personal data: an + # operator must be able to turn OFF only this write (e.g. pending a + # privacy review) without also disabling progress or baseline + # recording. Declaring `full_file_name` in `capture_watch_pvs` alone + # is necessary but not sufficient, mirroring every other switch + # here. See `cora.api._run_witness`'s "Capture path pairing" section. + capture_path_recording_enabled: bool = False + @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 2ec5a0cc6c..15781d5798 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 = "20260814041035" +EXPECTED_SCHEMA_VERSION: Final = "20260816094314" """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 3bbb8c2e59..e319b480cb 100644 --- a/apps/api/src/cora/run/aggregates/run/__init__.py +++ b/apps/api/src/cora/run/aggregates/run/__init__.py @@ -7,6 +7,14 @@ start_run needs it today). """ +from cora.run.aggregates.run.capture_path import ( + UNOBSERVED_CAPTURE_PATH, + CapturePath, + CapturePathStore, + InMemoryCapturePathStore, + PostgresCapturePathStore, + load_run_capture_path, +) from cora.run.aggregates.run.entries import ( InMemoryObservationStore, Observation, @@ -69,6 +77,7 @@ witness_safety_envelope, ) from cora.run.aggregates.run.state import ( + CAPTURE_CODE_EXTERNAL_REF_SCHEME, LOGBOOK_KIND_OBSERVATION, OBSERVATION_LOGBOOK_SCHEMA, READING_CHANNEL_NAME_MAX_LENGTH, @@ -139,11 +148,13 @@ RunTruncateReason, SafetyEnvelopeVerdict, SamplingProcedure, + extract_capture_code, validate_input_dataset_ids, validate_pinned_calibration_ids, ) __all__ = [ + "CAPTURE_CODE_EXTERNAL_REF_SCHEME", "HOLD_CAUSES", "HOLD_CAUSE_AUTHORITY_REVOCATION", "HOLD_CAUSE_OPERATOR", @@ -161,6 +172,9 @@ "RUN_NAME_MAX_LENGTH", "RUN_PINNED_CALIBRATIONS_MAX_ENTRIES", "SAMPLING_PROCEDURE_VALUES", + "UNOBSERVED_CAPTURE_PATH", + "CapturePath", + "CapturePathStore", "CapturePreconditionBypassSnapshot", "CaptureProgressSnapshot", "CautionAcknowledgement", @@ -170,6 +184,7 @@ "FeedHeartbeat", "FeedHeartbeatStore", "HoldClaimReleased", + "InMemoryCapturePathStore", "InMemoryFeedHeartbeatStore", "InMemoryObservationStore", "InvalidChannelNameError", @@ -190,6 +205,7 @@ "InvalidSamplingProcedureError", "Observation", "ObservationStore", + "PostgresCapturePathStore", "PostgresFeedHeartbeatStore", "PostgresObservationStore", "Run", @@ -253,12 +269,14 @@ "enclosure_gate_refusal", "event_type_name", "evolve", + "extract_capture_code", "fold", "fold_hold_claims", "fold_hold_claims_from_stored", "from_stored", "is_last_active_claim", "load_run", + "load_run_capture_path", "supply_gate_check", "to_payload", "validate_adjusted_parameters_against_method_schema", diff --git a/apps/api/src/cora/run/aggregates/run/capture_path.py b/apps/api/src/cora/run/aggregates/run/capture_path.py new file mode 100644 index 0000000000..aa4b49053d --- /dev/null +++ b/apps/api/src/cora/run/aggregates/run/capture_path.py @@ -0,0 +1,232 @@ +"""CapturePath vault: a witnessed Run's observed capture file path. + +Mirrors the `actor_profile` / `ProfileStore` PATTERN (memory/project_pii_vault, +memory/project_pii_vault_implementation_design), not its code +organization (see "BC-internal" below for the deliberate divergence +there): a mutable side table, keyed by an identity the domain already +has (here, the Run's own `run_id`), holding a value that must never +reach an event payload because events are immutable and INSERT-only at +the role level, so personal data written there could never be erased. + +## Why this exists + +`RunWitnessRecorder` observes the areaDetector file plugin's own +`FullFileName_RBV` readback and, once it verifies the reading postdates +the Run's own BEGUN time, needs somewhere to put the result. 2-BM's +directory layout embeds `{UserLastName}-{ProposalNumber}` +(`tomoscan_2bm.py:474-477`), so the observed path is personal data by +construction; it goes here, never onto `RunCompleted` / `RunAborted`. + +## BC-internal, like ObservationStore and FeedHeartbeatStore + +Unlike `ProfileStore` (Kernel-level because Access BC AND Agent BC both +write to it), this store has exactly one BC. Per `wire.py`'s own +"BC-internal ObservationStore + FeedHeartbeatStore wiring" convention, +it is built locally in `wire_run(deps)` from `deps.pool` and surfaced on +the `RunHandlers` bundle (`RunHandlers.capture_path_store`), not +promoted to a `Kernel` field. `main.py`'s composition-root lifespan +(`RunWitnessRecorder`, which is outside the Run BC) reads it off +`app.state.run.capture_path_store`, the same route +`app.state.run.feed_heartbeat_store` already takes. + +## Read path never redacts; write path never logs + +Unlike `load_actor_display_name`'s tombstone (which fires on +*erasure*), `load_run_capture_path`'s fallback fires on *absence* +(never observed, or rejected by the dual-clock guard): there is no +erasure slice yet. The resolved value IS the real path: an operator +reads it specifically to locate the file for `ingest_scan`, so nothing +in this module redacts it. Redaction belongs to logs, exception text, +and `capture_watch_preflight`, never to this authorized read. +""" + +# pyright: reportUnknownMemberType=false, reportUnknownVariableType=false, reportUnknownArgumentType=false +# asyncpg's stubs are loose; suppress only at module level for the +# adapter classes. The Protocol + dataclass + tombstone helper stay +# strictly typed for every caller above the boundary. Mirrors +# `run/aggregates/run/entries.py`'s identical suppress comment. + +from dataclasses import dataclass, field +from datetime import UTC, datetime +from typing import Protocol +from uuid import UUID + +import asyncpg + +UNOBSERVED_CAPTURE_PATH = "" + + +@dataclass(frozen=True) +class CapturePath: + """One row in the `run_capture_path` PII vault. + + `observed_at` is the substrate's own timestamp for the reading + (`Measurement.produced_at`), carried through so a reader can see + when the file plugin actually opened the file, distinct from + `created_at` (when this row was written). + """ + + run_id: UUID + observed_path: str = field(repr=False) + """Personal data. `repr=False` so an accidental bare `_log.info(..., + row=row)` or assertion-failure message renders this dataclass + without it; deliberate defense-in-depth, not the primary guard + (nothing should be logging a `CapturePath` at all).""" + observed_at: datetime + created_at: datetime + updated_at: datetime + + +class CapturePathStore(Protocol): + """Read / write access to the `run_capture_path` table. + + Deliberately no batch/`get_many` read: every consumer resolves + exactly one `run_id` at a time (`get_run`'s route/tool, mirroring + `get_actor`'s single-entity `load_actor_display_name`). `list_runs` + (the bulk, cursor-paginated query) never touches this store at all; + see `list_runs.bind`'s own docstring for why a batch method here + would be an attractive nuisance toward reintroducing that mistake. + + Two implementors: `PostgresCapturePathStore` (production) and + `InMemoryCapturePathStore` (tests / `app_env=test`). No `delete` / + `scrub_and_delete` method yet: no erasure slice calls one (see this + module's docstring); a future slice adds it alongside the SQL + `DELETE` grant the init migration already carries. + """ + + async def upsert( + self, + *, + run_id: UUID, + observed_path: str, + observed_at: datetime, + created_at: datetime, + ) -> None: + """Insert a new row or overwrite an existing one for `run_id`. + + Idempotent on the run_id PK: `RunWitnessRecorder` calls this at + most once per promotion (one terminal per Run), but retrying + after a partial failure replays cleanly. + """ + ... + + async def get(self, run_id: UUID) -> CapturePath | None: + """Fetch a row by run_id; `None` when absent (never observed, + rejected by the dual-clock guard, or recording disabled).""" + ... + + +async def load_run_capture_path(store: CapturePathStore, run_id: UUID) -> str: + """Resolve the observed path for a run_id; fallback when absent. + + Read-path convention mirroring `load_actor_display_name`: any + handler surfacing this value calls this helper rather than + inlining the `None` check. Returns `UNOBSERVED_CAPTURE_PATH` when + no row exists, which the caller should treat as "not yet observed + or rejected by the dual-clock guard," never as an error. + """ + row = await store.get(run_id) + return row.observed_path if row else UNOBSERVED_CAPTURE_PATH + + +_UPSERT_SQL = """ +INSERT INTO run_capture_path (run_id, observed_path, observed_at, created_at, updated_at) +VALUES ($1, $2, $3, $4, $4) +ON CONFLICT (run_id) DO UPDATE + SET observed_path = EXCLUDED.observed_path, + observed_at = EXCLUDED.observed_at, + updated_at = now() +""" + +_GET_SQL = """ +SELECT run_id, observed_path, observed_at, created_at, updated_at +FROM run_capture_path +WHERE run_id = $1 +""" + + +def _row_to_capture_path(row: asyncpg.Record) -> CapturePath: + return CapturePath( + run_id=row["run_id"], + observed_path=row["observed_path"], + observed_at=row["observed_at"], + created_at=row["created_at"], + updated_at=row["updated_at"], + ) + + +class PostgresCapturePathStore: + """asyncpg-backed `CapturePathStore` implementation.""" + + def __init__(self, pool: asyncpg.Pool) -> None: + self._pool = pool + + async def upsert( + self, + *, + run_id: UUID, + observed_path: str, + observed_at: datetime, + created_at: datetime, + ) -> None: + async with self._pool.acquire() as conn: + await conn.execute(_UPSERT_SQL, run_id, observed_path, observed_at, created_at) + + async def get(self, run_id: UUID) -> CapturePath | None: + async with self._pool.acquire() as conn: + row = await conn.fetchrow(_GET_SQL, run_id) + return _row_to_capture_path(row) if row is not None else None + + +class InMemoryCapturePathStore: + """Test / `app_env=test` adapter for `CapturePathStore`. + + Postgres semantics preserved, mirroring `InMemoryProfileStore`: + 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`, which the real adapter + never even sends on the UPDATE branch). + """ + + def __init__(self) -> None: + self._rows: dict[UUID, CapturePath] = {} + + async def upsert( + self, + *, + run_id: UUID, + observed_path: str, + observed_at: datetime, + created_at: datetime, + ) -> None: + existing = self._rows.get(run_id) + if existing is None: + self._rows[run_id] = CapturePath( + run_id=run_id, + observed_path=observed_path, + observed_at=observed_at, + created_at=created_at, + updated_at=created_at, + ) + else: + self._rows[run_id] = CapturePath( + run_id=run_id, + observed_path=observed_path, + observed_at=observed_at, + created_at=existing.created_at, + updated_at=datetime.now(tz=UTC), + ) + + async def get(self, run_id: UUID) -> CapturePath | None: + return self._rows.get(run_id) + + +__all__ = [ + "UNOBSERVED_CAPTURE_PATH", + "CapturePath", + "CapturePathStore", + "InMemoryCapturePathStore", + "PostgresCapturePathStore", + "load_run_capture_path", +] diff --git a/apps/api/src/cora/run/aggregates/run/state.py b/apps/api/src/cora/run/aggregates/run/state.py index 3f8ea99498..bd2e7396b0 100644 --- a/apps/api/src/cora/run/aggregates/run/state.py +++ b/apps/api/src/cora/run/aggregates/run/state.py @@ -1898,3 +1898,32 @@ def validate_input_dataset_ids(value: frozenset[UUID]) -> frozenset[UUID]: if len(value) > RUN_INPUT_DATASETS_MAX_ENTRIES: raise InvalidInputDatasetsError(len(value)) return value + + +CAPTURE_CODE_EXTERNAL_REF_SCHEME = "capture-code" +"""The `Identifier.scheme` a witnessed genesis stamps onto +`Run.external_refs` (`record_witnessed_run`'s decider). Public so every +reader of `external_refs` looking for the capture code uses the exact +same literal, never a second copy that can drift from the writer's own.""" + + +def extract_capture_code(external_refs: frozenset[Identifier]) -> str | None: + """Find the `Identifier(scheme="capture-code", ...)` entry's value. + + `None` for a Conducted Run (no such ref at all) and, defensively, + for a Witnessed Run whose genesis somehow lacked one: + `record_witnessed_run`'s decider always stamps exactly one, so this + should not happen in practice, but a missing ref must never raise -- + only degrade to "capture code unknown," e.g. at the boot-time + dedup rebuild (`rebuild_open_captures`) or a read-model query. + + Single source of truth for this lookup against the FOLDED aggregate + state's `frozenset[Identifier]` shape; `RunSummaryProjection`'s + sibling `_extract_capture_code` (`run/projections/summary.py`) + performs the same lookup against a raw JSON event payload list, a + genuinely different input type, so it stays a separate function. + """ + for ref in external_refs: + if ref.scheme == CAPTURE_CODE_EXTERNAL_REF_SCHEME: + return ref.value + return None 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 233e64d85c..ed6e260202 100644 --- a/apps/api/src/cora/run/features/get_run/handler.py +++ b/apps/api/src/cora/run/features/get_run/handler.py @@ -3,12 +3,23 @@ Cross-BC query-handler shape mirroring `get_plan` / `get_practice` / `get_method` / `get_family` / `get_subject` / `get_actor`. -Returns the domain `Run`, not a DTO. The route + tool layers do -their own DTO mapping (primitives only). +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 +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 +today, so nothing forces the vault touch out of it, and keeping the +resolve-and-guard logic in ONE place (not duplicated across `route.py` +and `tool.py`) is strictly better. The route + tool layers do their own +DTO mapping (primitives only) off the returned `RunView`. Query handlers do NOT emit `causation_id` log fields. """ +from dataclasses import dataclass from typing import Protocol from uuid import UUID @@ -16,7 +27,13 @@ from cora.infrastructure.logging import get_logger from cora.infrastructure.ports import Deny from cora.infrastructure.routing import NIL_SENTINEL_ID -from cora.run.aggregates.run import Run, load_run +from cora.run.aggregates.run import ( + CapturePathStore, + Run, + extract_capture_code, + load_run, + load_run_capture_path, +) from cora.run.errors import UnauthorizedError from cora.run.features.get_run.query import GetRun @@ -25,6 +42,25 @@ _log = get_logger(__name__) +@dataclass(frozen=True) +class RunView: + """Read-side composition of Run aggregate + capture-path resolution. + + `capture_code` is folded from `run.external_refs`; `None` for a + Conducted Run. `observed_capture_path` resolves from the + `run_capture_path` PII vault ONLY when `capture_code` is not + `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 + DTOs. + """ + + run: Run + capture_code: str | None + observed_capture_path: str | None + + class Handler(Protocol): """Callable interface every get_run handler implements.""" @@ -35,11 +71,11 @@ async def __call__( principal_id: UUID, correlation_id: UUID, surface_id: UUID = NIL_SENTINEL_ID, - ) -> Run | None: ... + ) -> RunView | None: ... -def bind(deps: Kernel) -> Handler: - """Build a get_run handler closed over the shared deps.""" +def bind(deps: Kernel, *, capture_path_store: CapturePathStore) -> Handler: + """Build a get_run handler closed over the shared deps + PII vault.""" async def handler( query: GetRun, @@ -47,7 +83,7 @@ async def handler( principal_id: UUID, correlation_id: UUID, surface_id: UUID = NIL_SENTINEL_ID, - ) -> Run | None: + ) -> RunView | None: _log.info( "get_run.start", query_name=_QUERY_NAME, @@ -74,6 +110,23 @@ async def handler( raise UnauthorizedError(decision.reason) run = await load_run(deps.event_store, query.run_id) + if run is None: + _log.info( + "get_run.success", + query_name=_QUERY_NAME, + run_id=str(query.run_id), + principal_id=str(principal_id), + correlation_id=str(correlation_id), + found=False, + ) + 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 + ) _log.info( "get_run.success", @@ -81,8 +134,17 @@ async def handler( run_id=str(query.run_id), principal_id=str(principal_id), correlation_id=str(correlation_id), - found=run is not None, + found=True, + ) + return RunView( + run=run, capture_code=capture_code, observed_capture_path=observed_capture_path ) - return run return handler + + +__all__ = [ + "Handler", + "RunView", + "bind", +] 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 4452c59670..c578cae38c 100644 --- a/apps/api/src/cora/run/features/get_run/route.py +++ b/apps/api/src/cora/run/features/get_run/route.py @@ -5,6 +5,12 @@ Response shape: `{id, name, plan_id, subject_id, raid, status}`. `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 +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 typing import Annotated, Any @@ -43,6 +49,19 @@ class RunResponse(BaseModel): add_run_to_campaign. None when the Run is standalone (not part of any Campaign). Closes design-memo Watch #17 (per Caution-design cross-BC consistency precedent). + + `capture_code` (slice 13) is the deployment-declared capture + identifier a witnessed genesis stamps onto `external_refs`. None + for a Conducted Run. NOT personal data. + + `observed_capture_path` (slice 13) is the areaDetector file the + capture wrote, resolved from the `run_capture_path` PII vault: + `None` when `capture_code` is `None` (not applicable, a Conducted + Run); the tombstone literal (`UNOBSERVED_CAPTURE_PATH`) when a + capture code exists but the vault has no row yet (never observed, + 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`. """ id: UUID @@ -55,6 +74,8 @@ class RunResponse(BaseModel): effective_parameters: dict[str, Any] = Field(default_factory=dict) trigger_source: str | None = None campaign_id: UUID | None = None + capture_code: str | None = None + observed_capture_path: str | None = None def _get_handler(request: Request) -> Handler: @@ -87,17 +108,18 @@ async def get_runs( principal_id: Annotated[UUID, Depends(get_principal_id)], surface_id: Annotated[UUID, Depends(get_surface_id)], ) -> RunResponse: - run = await handler( + view = await handler( GetRun(run_id=run_id), principal_id=principal_id, correlation_id=cid, surface_id=surface_id, ) - if run is None: + if view is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Run {run_id} not found", ) + run = view.run return RunResponse( id=run.id, name=run.name.value, @@ -109,4 +131,6 @@ async def get_runs( effective_parameters=run.effective_parameters, trigger_source=run.trigger_source, campaign_id=run.campaign_id, + capture_code=view.capture_code, + observed_capture_path=view.observed_capture_path, ) 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 9ddd5d4330..8cee3af554 100644 --- a/apps/api/src/cora/run/features/get_run/tool.py +++ b/apps/api/src/cora/run/features/get_run/tool.py @@ -1,4 +1,11 @@ -"""MCP tool for the `get_run` query slice.""" +"""MCP tool for the `get_run` query slice. + +`capture_code` / `observed_capture_path` (slice 13) 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 +every other field already follows. +""" from collections.abc import Callable from typing import Annotated, Any @@ -28,6 +35,8 @@ class RunOutput(BaseModel): effective_parameters: dict[str, Any] = Field(default_factory=dict) trigger_source: str | None = None campaign_id: UUID | None = None + capture_code: str | None = None + observed_capture_path: str | None = None def register(mcp: FastMCP, *, get_handler: Callable[[], Handler]) -> None: @@ -45,15 +54,16 @@ async def get_run_tool( # pyright: ignore[reportUnusedFunction] ], ) -> RunOutput: handler = get_handler() - run = await handler( + view = await handler( GetRun(run_id=run_id), principal_id=get_mcp_principal_id(ctx), correlation_id=current_correlation_id(), surface_id=get_mcp_surface_id(), ) - if run is None: + if view is None: msg = f"Run {run_id} not found" raise ValueError(msg) + run = view.run return RunOutput( id=run.id, name=run.name.value, @@ -65,4 +75,6 @@ async def get_run_tool( # pyright: ignore[reportUnusedFunction] effective_parameters=run.effective_parameters, trigger_source=run.trigger_source, campaign_id=run.campaign_id, + capture_code=view.capture_code, + observed_capture_path=view.observed_capture_path, ) diff --git a/apps/api/src/cora/run/features/list_runs/handler.py b/apps/api/src/cora/run/features/list_runs/handler.py index e0d94d4884..1d81d35afc 100644 --- a/apps/api/src/cora/run/features/list_runs/handler.py +++ b/apps/api/src/cora/run/features/list_runs/handler.py @@ -71,6 +71,20 @@ class RunSummaryItem: genesis. The RunSupervisor and RunInitiator runtimes filter on this to skip Witnessed Runs: those are not theirs to hold, resume, truncate, or count toward in-flight limits.""" + capture_code: str | None + """The `capture-code` external_ref a witnessed genesis stamps onto + `RunStarted.external_refs` (slice 13), folded onto the projection at + insert time. `None` for a Conducted Run, which has no capture code + at all. NOT personal data (a deployment-declared identifier), unlike + the resolved capture path: that value is deliberately NOT resolved + here. This handler instance is shared by every internal + composition-root caller (`rebuild_open_captures`, the supervisor + and initiator watchdogs) as well as the REST/MCP route, none of + which need the observed path; resolving it here would touch the + vault on every one of those callers' every page, for no benefit. + `get_run` resolves it instead (single-entity, no internal caller of + its own today), mirroring why `list_actors` never touches + `ProfileStore` while `get_actor` does.""" @dataclass(frozen=True) @@ -97,7 +111,7 @@ async def __call__( _SELECT_COLUMNS = ( "run_id, name, plan_id, subject_id, raid, status, created_at, running_since, " "override_parameters_present, campaign_id, snr_limit, " - "expected_observation_interval_seconds, conduct_mode" + "expected_observation_interval_seconds, conduct_mode, capture_code" ) @@ -116,6 +130,7 @@ def _row_to_item(row: Any) -> RunSummaryItem: snr_limit=row["snr_limit"], expected_observation_interval_seconds=row["expected_observation_interval_seconds"], conduct_mode=str(row["conduct_mode"]), + capture_code=str(row["capture_code"]) if row["capture_code"] is not None else None, ) @@ -129,7 +144,17 @@ def _log_fields(query: ListRuns) -> dict[str, Any]: def bind(deps: Kernel) -> Handler: - """Build a list_runs handler closed over the shared deps.""" + """Build a list_runs handler closed over the shared deps. + + Deliberately never touches `CapturePathStore`: this handler instance + is shared by every internal composition-root caller + (`RunWitnessRecorder.rebuild_open_captures`, the run-supervisor and + run-initiator watchdogs) as well as the REST/MCP route, all under + one coarse `ListRuns` authorization grant. Resolving the observed + capture path here would hand personal data to every one of them. + `get_run` (single-entity) resolves it instead; see + `RunSummaryItem.capture_code`'s own docstring for the full argument. + """ return make_list_query_handler( deps, query_name="ListRuns", diff --git a/apps/api/src/cora/run/features/list_runs/route.py b/apps/api/src/cora/run/features/list_runs/route.py index b762cdb933..396e3f83d2 100644 --- a/apps/api/src/cora/run/features/list_runs/route.py +++ b/apps/api/src/cora/run/features/list_runs/route.py @@ -52,6 +52,16 @@ class RunSummaryDTO(BaseModel): "Campaign membership snapshot." ), ) + capture_code: str | None = Field( + default=None, + description=( + "Deployment-declared capture identifier a witnessed genesis " + "stamps onto external_refs (slice 13). NULL for a Conducted " + "Run. NOT the observed capture file path: that value is " + "personal data and is resolved per-run via `GET " + "/runs/{run_id}`, never surfaced on this bulk list." + ), + ) class RunListResponse(BaseModel): @@ -148,6 +158,7 @@ async def list_runs( created_at=item.created_at, override_parameters_present=item.override_parameters_present, campaign_id=item.campaign_id, + capture_code=item.capture_code, ) for item in page.items ], diff --git a/apps/api/src/cora/run/features/list_runs/tool.py b/apps/api/src/cora/run/features/list_runs/tool.py index 0e0667ab69..821a4b91f3 100644 --- a/apps/api/src/cora/run/features/list_runs/tool.py +++ b/apps/api/src/cora/run/features/list_runs/tool.py @@ -40,6 +40,16 @@ class RunSummaryRow(BaseModel): "Campaign this Run is a member of (at-start or post-hoc). NULL for standalone Runs." ), ) + capture_code: str | None = Field( + default=None, + description=( + "Deployment-declared capture identifier a witnessed genesis " + "stamps onto external_refs (slice 13). NULL for a Conducted " + "Run. NOT the observed capture file path: that value is " + "personal data, resolved per-run via the `get_run` tool, " + "never surfaced on this bulk list." + ), + ) class RunListOutput(BaseModel): @@ -112,6 +122,7 @@ async def list_runs_tool( # pyright: ignore[reportUnusedFunction] created_at=item.created_at, override_parameters_present=item.override_parameters_present, campaign_id=item.campaign_id, + capture_code=item.capture_code, ) for item in page.items ], diff --git a/apps/api/src/cora/run/ports/capture_observer.py b/apps/api/src/cora/run/ports/capture_observer.py index 7a0d64d281..c51bcfcf39 100644 --- a/apps/api/src/cora/run/ports/capture_observer.py +++ b/apps/api/src/cora/run/ports/capture_observer.py @@ -17,16 +17,17 @@ cross-BC consumer (rule-of-three), exactly the `RunChannelLookup` precedent. -## Three reading kinds, one stream +## Four reading kinds, one stream `observe()` yields `AnyCaptureObservation`, the union of `CaptureLifecycleObservation` (a phase claim: BEGUN / PROGRESSING / ENDED / ABORTED / UNRECOGNIZED, or no claim at all on a probe-only or disconnect reading), `CaptureProgressObservation` (a numeric -progress counter, e.g. `ImagesSaved`), and -`CapturePreconditionBypassObservation` (the optional `testing` role's -tri-state reading, slice 11). A consumer narrows with `isinstance`. -The three are peers, not a supertype and subtypes: a single reading is +progress counter, e.g. `ImagesSaved`), `CapturePreconditionBypassObservation` +(the optional `testing` role's tri-state reading, slice 11), and +`CapturePathObservation` (the optional `full_file_name` role's text +reading, slice 13). A consumer narrows with `isinstance`. +The four are peers, not a supertype and subtypes: a single reading is never more than one of these at once, so one closed-over `CaptureObservation` name for "the default kind" would have made an isinstance check on the lifecycle kind read as a supertype check @@ -79,6 +80,22 @@ orthogonal question from whether the facility had beam. See [[project_run_witness_test_provenance_slice11]] for the full argument against collapsing the two. +- `CapturePathObservation` (slice 13): one reading of the optional + `full_file_name` role, the areaDetector file plugin's own filename + readback (`FullFileName_RBV`), drained continuously and independently + of any one capture (the file plugin fires this at file OPEN, which + can land before, during, or after any particular capture's own BEGUN + observation reaches this port). **`observed_path` is personal data**: + 2-BM's directory layout embeds `{UserLastName}-{ProposalNumber}` + (`tomoscan_2bm.py:474-477`), so every real reading of this role + carries a person's surname. A consumer of this observation MUST NOT + log it, put it on an event, or persist it anywhere but the dedicated + `run_capture_path` PII vault (`CapturePathStore`, mirroring + `actor_profile` / `ProfileStore`). Recording an observed path is + NOT a claim the file is complete, so it does not violate "No terminal + claims about a file" below; it says only that the file plugin opened + a file at this substrate time, nothing about what it contains or + whether writing to it has finished. - `CaptureObserverScope`: the set of capture codes the substrate adapter should subscribe to. Empty scope is valid and yields no observations. @@ -115,7 +132,7 @@ """ from collections.abc import AsyncGenerator, AsyncIterator -from dataclasses import dataclass +from dataclasses import dataclass, field from datetime import datetime from typing import Protocol, runtime_checkable @@ -245,8 +262,54 @@ class CapturePreconditionBypassObservation: source_id: str +@dataclass(frozen=True) +class CapturePathObservation: + """One reading of the optional `full_file_name` role: the + areaDetector file plugin's own filename readback (`FullFileName_RBV`). + + `observed_path` is PERSONAL DATA (see this module's docstring, + "Domain vocabulary" bullet on this class). It MUST NOT be logged, + placed on an event payload, or persisted anywhere but the + `run_capture_path` PII vault. A consumer that retains this + observation (e.g. to compare against a Run's own BEGUN time before + writing it to the vault) must audit every log line it emits along + that path for this field. + + Not a phase claim, not a numeric counter, not a tri-state flag: a + plain text reading, the substrate's file plugin's own string, taken + as-is once it passes the length/emptiness checks in + `_capture_observer.py` (see `_from_full_file_name_reading`). + + `role` is not carried (mirrors `CapturePreconditionBypassObservation`): + `full_file_name` is singular per capture code. + + `reach_tier`, `observed_at`, `source_kind`, `source_id` carry the + same meaning as on `CaptureLifecycleObservation`. `observed_at` is + the substrate's OWN time for this reading, never CORA's clock: it + is the timestamp a consumer compares against a Run's own BEGUN time + to decide whether this reading belongs to that Run at all (Finding + A, memory/project_witnessed_run_prelive_slices.md slice 13): a + reading whose file plugin fired before this capture even started + almost certainly describes the PREVIOUS capture's file. + """ + + capture_code: str + observed_path: str = field(repr=False) + """Personal data. `repr=False` so an accidental bare `_log.info(..., + observation=observation)` or assertion-failure message renders this + dataclass without it; deliberate defense-in-depth, not the primary + guard (nothing should be logging this observation at all).""" + reach_tier: ReachTier + observed_at: datetime | None + source_kind: str + source_id: str + + AnyCaptureObservation = ( - CaptureLifecycleObservation | CaptureProgressObservation | CapturePreconditionBypassObservation + CaptureLifecycleObservation + | CaptureProgressObservation + | CapturePreconditionBypassObservation + | CapturePathObservation ) """The union `observe()` yields. Named explicitly rather than reusing any one member's name for the union, so a caller that has not yet been @@ -313,6 +376,7 @@ async def _drain(self, scope: CaptureObserverScope) -> AsyncGenerator[AnyCapture "CaptureLifecycleObservation", "CaptureObserver", "CaptureObserverScope", + "CapturePathObservation", "CapturePhase", "CapturePreconditionBypassObservation", "CaptureProgressObservation", diff --git a/apps/api/src/cora/run/projections/summary.py b/apps/api/src/cora/run/projections/summary.py index 519863637e..2a7a5e6b51 100644 --- a/apps/api/src/cora/run/projections/summary.py +++ b/apps/api/src/cora/run/projections/summary.py @@ -9,7 +9,8 @@ override_parameters_present + campaign_id? + pinned_calibration_ids + - conduct_mode from payload) + conduct_mode + capture_code? from + payload) - RunHeld -> UPDATE status=Held - RunResumed -> UPDATE status=Running + running_since reset - RunCompleted -> UPDATE status=Completed (terminal) @@ -58,6 +59,19 @@ Forward-compat: legacy RunStarted payloads lack the key entirely; `.get("pinned_calibration_ids", [])` returns `[]` so legacy rows land with an empty UUID array. + +`capture_code` (slice 13) surfaces the `Identifier(scheme="capture-code")` +a witnessed genesis already stamps onto `RunStarted.external_refs`, so +`list_runs` can filter/resolve without folding the Run stream. NULL for +a Conducted Run (no external_refs entry at all) and, defensively, for +any Witnessed row whose genesis somehow lacked one: this module's own +`_extract_capture_code` returns `None` in both cases, mirroring the +same "absent, not erroring" posture the public sibling function of the +same name (`cora.run.aggregates.run.state.extract_capture_code`) +already takes against the folded aggregate state. Deliberately NOT the +resolved capture PATH: that value is +personal data and lives only in the `run_capture_path` PII vault, +resolved at query time, never folded into this projection. """ # pyright: reportUnknownMemberType=false, reportUnknownVariableType=false, reportUnknownArgumentType=false @@ -74,11 +88,34 @@ INSERT INTO proj_run_summary (run_id, name, plan_id, subject_id, raid, status, created_at, running_since, override_parameters_present, campaign_id, pinned_calibration_ids, - snr_limit, expected_observation_interval_seconds, conduct_mode) -VALUES ($1, $2, $3, $4, $5, 'Running', $6, $6, $7, $8, $9::uuid[], $10, $11, $12) + snr_limit, expected_observation_interval_seconds, conduct_mode, capture_code) +VALUES ($1, $2, $3, $4, $5, 'Running', $6, $6, $7, $8, $9::uuid[], $10, $11, $12, $13) ON CONFLICT (run_id) DO NOTHING """ +_CAPTURE_CODE_SCHEME = "capture-code" + + +def _extract_capture_code(external_refs: list[dict[str, Any]]) -> str | None: + """Find the `Identifier(scheme="capture-code")` entry's value in a + raw `RunStarted.external_refs` payload list. + + Mirrors the public sibling `extract_capture_code` in + `cora.run.aggregates.run.state` (which does the same lookup against + the folded aggregate state's `frozenset[Identifier]`, and is what + `_run_witness.py` and `get_run`'s handler both call); this one + operates on the raw JSON list a stored event payload carries, a + genuinely different input type, so it stays a separate function. + `None` for a Conducted Run (empty list) and, defensively, for a + Witnessed row that somehow lacks the ref: never an error. + """ + for ref in external_refs: + if ref.get("scheme") == _CAPTURE_CODE_SCHEME: + value = ref.get("value") + return str(value) if value is not None else None + return None + + _UPDATE_STATUS_SQL = """ UPDATE proj_run_summary SET status = $2, updated_at = now() @@ -197,6 +234,10 @@ async def apply( # historical default so legacy rows land as Conducted, which # is true of every Run started before this field existed. conduct_mode = payload.get("conduct_mode", "Conducted") + # Forward-compat: legacy RunStarted payloads have no + # external_refs key; .get(..., []) returns [] so legacy rows + # (and every Conducted Run) land with capture_code IS NULL. + capture_code = _extract_capture_code(payload.get("external_refs", [])) await conn.execute( _INSERT_RUN_SQL, UUID(payload["run_id"]), @@ -211,6 +252,7 @@ async def apply( snr_limit, expected_interval, conduct_mode, + capture_code, ) return if event.event_type == "RunResumed": diff --git a/apps/api/src/cora/run/wire.py b/apps/api/src/cora/run/wire.py index f40a98756d..ee13600c06 100644 --- a/apps/api/src/cora/run/wire.py +++ b/apps/api/src/cora/run/wire.py @@ -48,7 +48,7 @@ cross-loads Plan → Practice → Method to surface the Method's `parameters_schema` for merged-result validation. -## BC-internal ObservationStore + FeedHeartbeatStore wiring +## BC-internal ObservationStore + FeedHeartbeatStore + CapturePathStore wiring `append_observations` needs a `ObservationStore` adapter. Per the per-category-writer pattern (mirrors Decision BC's InferenceStore @@ -61,6 +61,32 @@ routes/tools never touch it, but the composition-root lifespan (`_capture_progress_feeder.py`'s feeder, slice 10) needs the write store directly, not wrapped behind a command. + +`capture_path_store` (slice 13) follows the identical construction +shape (built locally from `deps.pool`), but NOT the identical wiring: +it is surfaced on the bundle for the composition root's +`RunWitnessRecorder` to write through (it verifies the observed path +against the Run's own BEGUN time before writing, so the write cannot be +a plain command) AND passed into `get_run.bind(deps, capture_path_store=...)`, +which resolves it inside the handler exactly the way `get_actor.bind(deps, +profile_store=...)` resolves `ProfileStore` -- see `get_run/handler.py`'s +`RunView`. It is deliberately NEVER passed into `list_runs.bind()`. +`list_runs` is one shared handler instance read by every internal +composition-root caller (`rebuild_open_captures`, the supervisor and +initiator watchdogs) as well as the REST/MCP route; resolving personal +data there would expose it to callers that only need `run_id` off each +page item, and would do so under one bulk, cursor-paginated grant no +different in kind from the coarse `ListRuns` authorization this BC's +own `list_query.py` already documents as unscoped-per-row (BOLA +deferred until ReBAC). `get_run` avoids the FIRST problem outright (no +internal caller exists today) regardless of how its own authorization +eventually gets scoped, mirroring why `list_actors` never touches +`ProfileStore` while `get_actor` does. Kernel-level placement is still +wrong for the same reason as before: this is a PII vault, so +`ProfileStore` would be the closer precedent by subject matter, but +`ProfileStore` is Kernel-level specifically because it is genuinely +cross-BC-shared (Access + Agent); this store has exactly one +BC. """ from dataclasses import dataclass @@ -74,10 +100,13 @@ from cora.infrastructure.kernel import Kernel from cora.infrastructure.observability import with_tracing from cora.run.aggregates.run import ( + CapturePathStore, FeedHeartbeatStore, + InMemoryCapturePathStore, InMemoryFeedHeartbeatStore, InMemoryObservationStore, ObservationStore, + PostgresCapturePathStore, PostgresFeedHeartbeatStore, PostgresObservationStore, ) @@ -122,6 +151,14 @@ class RunHandlers: not a handler, mirroring `EnclosureHandlers.permit_probe_store` exactly: a composition-root lifespan needs this dependency directly, and it isn't itself a command handler.""" + capture_path_store: CapturePathStore + """Slice 13's PII vault store for a witnessed Run's observed + capture file path. Surfaced on the bundle for the same reason as + `feed_heartbeat_store`: `RunWitnessRecorder` (composition root) + writes through it directly after its own dual-clock guard passes. + `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.""" def wire_run(deps: Kernel) -> RunHandlers: @@ -134,8 +171,12 @@ def wire_run(deps: Kernel) -> RunHandlers: if deps.pool is not None else InMemoryFeedHeartbeatStore() ) + capture_path_store: CapturePathStore = ( + PostgresCapturePathStore(deps.pool) if deps.pool is not None else InMemoryCapturePathStore() + ) return RunHandlers( feed_heartbeat_store=feed_heartbeat_store, + capture_path_store=capture_path_store, start_run=with_tracing( with_idempotency( start_run.bind(deps), @@ -212,7 +253,7 @@ def wire_run(deps: Kernel) -> RunHandlers: bc=_BC, ), get_run=with_tracing( - get_run.bind(deps), + get_run.bind(deps, capture_path_store=capture_path_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 new file mode 100644 index 0000000000..290dd8ebab --- /dev/null +++ b/apps/api/tests/architecture/test_run_events_carry_no_pii.py @@ -0,0 +1,130 @@ +"""Run event payloads carry NO PII. The observed capture path lives in +the `run_capture_path` table per memory/project_witnessed_run_prelive_slices.md +slice 13 (mirroring `actor_profile` / [[project_pii_vault]]). + +Fitness function: AST-walks `cora/run/aggregates/run/events.py` and +rejects any dataclass field on ANY class in that file whose name +appears in the PII deny-list. Mirrors `test_actor_events_carry_no_pii.py`'s +mechanism (main check + file-presence guard + seeded-violation +meta-test); see that file for the full rationale on why this lives +separately from structural fitness tests. + +Deliberately scans every class in the file, NOT only classes whose name +starts with "Run": this module also defines `CautionAcknowledgement`, +`DecisionDebriefRequested`, and `HoldClaimReleased`, real Run-stream +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. +""" + +import ast +from pathlib import Path + +import pytest + +from tests.architecture.conftest import CORA_ROOT + +_EVENTS_FILE = CORA_ROOT / "run" / "aggregates" / "run" / "events.py" + +# Any dataclass field anywhere in the events file whose annotation +# target name matches one of these strings counts as a violation. +_PII_FIELD_NAMES = frozenset( + { + "observed_path", + "capture_path", + "full_file_name", + "path", + "directory", + "file_path", + "surname", + "proposal_number", + "user_name", + "user_last_name", + "user_badge", + "user_email", + "user_institution", + } +) + + +def _pii_field_violations(source_path: Path) -> list[str]: + """AST-walk every class in `source_path`'s dataclass fields for a + PII deny-list hit. Takes a path (not a hardcoded file) so the + seeded-violation meta-test below can call this SAME function + against a temp file, rather than maintaining a second copy of the + walk that could silently drift from what actually runs. + """ + tree = ast.parse(source_path.read_text()) + violations: list[str] = [] + for node in tree.body: + if not isinstance(node, ast.ClassDef): + continue + for stmt in node.body: + if not isinstance(stmt, ast.AnnAssign): + continue + target = stmt.target + if isinstance(target, ast.Name) and target.id in _PII_FIELD_NAMES: + violations.append(f"line {stmt.lineno}: {node.name}.{target.id}") + return violations + + +@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`. + + The observed capture path is personal data (2-BM's directory layout + embeds a surname and a proposal number); it lives in the + `run_capture_path` vault accessed via `CapturePathStore`, never on + `RunCompleted` / `RunAborted` or any other event. A regression here + 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. + """ + 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) + ) + + +@pytest.mark.architecture +def test_run_events_file_is_present() -> None: + """Sanity: the events.py file must exist; the file move below the + aggregate folder would silently make the PII-deny scan a no-op + without this guard.""" + msg = f"Expected Run events file at {_EVENTS_FILE}" + assert _EVENTS_FILE.exists(), msg + + +@pytest.mark.architecture +def test_pii_deny_list_actually_finds_violations_when_seeded(tmp_path: Path) -> None: + """Meta-test: confirm the ACTUAL production walker (`_pii_field_violations`, + not a re-implemented copy) flags a seeded violation, on a class with + no `Run` prefix -- the exact shape `CautionAcknowledgement` / + `DecisionDebriefRequested` / `HoldClaimReleased` already have in the + real file, which a name-prefix filter would have missed. + + Guards against two silent-pass failure modes at once: a future + refactor moving the event classes to a sub-module (the walker + quietly stops seeing them), and a future re-introduction of a + name-prefix filter (the walker stops seeing non-`Run`-prefixed + events). + """ + seed_file = tmp_path / "seed_events.py" + seed_file.write_text( + "from dataclasses import dataclass\n" + "@dataclass\n" + "class HoldClaimReleased:\n" + " run_id: int\n" + " observed_path: str # PII violation seeded by the meta-test\n" + ) + violations = _pii_field_violations(seed_file) + assert violations, "seeded `observed_path` field must be flagged by the deny-list walker" diff --git a/apps/api/tests/contract/test_get_run_endpoint.py b/apps/api/tests/contract/test_get_run_endpoint.py index af1efcf1cc..6b3fe88a7c 100644 --- a/apps/api/tests/contract/test_get_run_endpoint.py +++ b/apps/api/tests/contract/test_get_run_endpoint.py @@ -82,6 +82,11 @@ def test_get_run_returns_200_with_running_status_for_sample_run() -> None: # None for standalone runs (no Campaign membership set at # start time or via add_run_to_campaign). "campaign_id": None, + # Slice 13 additive response surface: a Conducted Run (started + # via /runs, not the in-process-only record_witnessed_run) has + # no capture code at all, so neither field resolves. + "capture_code": None, + "observed_capture_path": None, } diff --git a/apps/api/tests/integration/test_capture_path_postgres.py b/apps/api/tests/integration/test_capture_path_postgres.py new file mode 100644 index 0000000000..350aa8cffa --- /dev/null +++ b/apps/api/tests/integration/test_capture_path_postgres.py @@ -0,0 +1,105 @@ +"""Integration: the `run_capture_path` PII vault against real Postgres. + +Mirrors `test_feed_heartbeats_postgres.py`'s shape: exercise +`PostgresCapturePathStore` 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: `cora_app` (the role `db_pool` connects as in +tests) can read and write; FORCE ROW LEVEL SECURITY is asserted at the +catalog level since there is no second role easily reachable in-test to +prove a bypass attempt fails. +""" + +# 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 PostgresCapturePathStore + +_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 = PostgresCapturePathStore(db_pool) + run_id = uuid4() + + await store.upsert( + run_id=run_id, + observed_path="/data/2026-01-Smith-12345/scan_001.h5", + 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.observed_path == "/data/2026-01-Smith-12345/scan_001.h5" + assert row.observed_at == _NOW + + +@pytest.mark.integration +async def test_get_absent_run_id_returns_none(db_pool: asyncpg.Pool) -> None: + store = PostgresCapturePathStore(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 = PostgresCapturePathStore(db_pool) + run_id = uuid4() + await store.upsert( + run_id=run_id, observed_path="/data/first.h5", observed_at=_NOW, created_at=_NOW + ) + await store.upsert( + run_id=run_id, observed_path="/data/second.h5", observed_at=_NOW, created_at=_NOW + ) + + async with db_pool.acquire() as conn: + count = await conn.fetchval( + "SELECT count(*) FROM run_capture_path WHERE run_id = $1", run_id + ) + assert count == 1 + row = await store.get(run_id) + assert row is not None + assert row.observed_path == "/data/second.h5" + + +@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 `actor_profile`'s identical posture.""" + async with db_pool.acquire() as conn: + row = await conn.fetchrow( + "SELECT relrowsecurity, relforcerowsecurity FROM pg_class WHERE relname = $1", + "run_capture_path", + ) + assert row is not None + assert row["relrowsecurity"] is True + assert row["relforcerowsecurity"] is True + + +@pytest.mark.integration +async def test_observed_path_length_constraint_rejects_an_oversized_value( + db_pool: asyncpg.Pool, +) -> None: + """Defense-in-depth on the same fact the application-layer + truncation guard checks (NELM=512 on the real PV, 511 usable + chars): the CHECK constraint is a second, independent backstop.""" + store = PostgresCapturePathStore(db_pool) + with pytest.raises(asyncpg.CheckViolationError): + await store.upsert( + run_id=uuid4(), + observed_path="a" * 512, + observed_at=_NOW, + created_at=_NOW, + ) diff --git a/apps/api/tests/integration/test_list_runs_handler_postgres.py b/apps/api/tests/integration/test_list_runs_handler_postgres.py index 42642581a9..28bfbd9da4 100644 --- a/apps/api/tests/integration/test_list_runs_handler_postgres.py +++ b/apps/api/tests/integration/test_list_runs_handler_postgres.py @@ -493,3 +493,64 @@ async def test_conduct_mode_filter_narrows_to_recorded_runs_only(db_pool: asyncp assert len(page.items) == 1 assert page.items[0].run_id == run_recorded assert page.items[0].conduct_mode == "Witnessed" + + +@pytest.mark.integration +async def test_witnessed_run_surfaces_its_capture_code(db_pool: asyncpg.Pool) -> None: + """Slice 13: `capture_code` folds from `RunStarted.external_refs` + onto `proj_run_summary` and surfaces via `list_runs` -- it is a + deployment-declared identifier, not personal data, so it is safe on + this bulk, cursor-paginated, coarsely-authorized query. The + OBSERVED PATH is deliberately NOT resolved here (`list_runs.bind` + never touches `CapturePathStore` at all; see that module's + docstring): `get_run` resolves it instead, mirroring why + `list_actors` never touches `ProfileStore` while `get_actor` does.""" + deps = _build_deps(db_pool, [*_chain_ids(), uuid4(), uuid4()]) + plan_id = await _seed_plan(deps, family_name="TomographyCaptureCode") + run_id = await bind_record_witnessed_run(deps)( + RecordWitnessedRun( + name="witnessed-with-code", + plan_id=plan_id, + capture_code="2bmb-tomoscan", + monitor_source_id=MonitorSourceId(uuid4()), + trigger="Monitor", + ), + principal_id=_PRINCIPAL_ID, + correlation_id=_CORRELATION_ID, + ) + await _drain(db_pool) + + handler = bind_list(deps) + page = await handler( + ListRuns(conduct_mode="Witnessed", limit=10), + principal_id=_PRINCIPAL_ID, + correlation_id=_CORRELATION_ID, + ) + + item = next(item for item in page.items if item.run_id == run_id) + assert item.capture_code == "2bmb-tomoscan" + + +@pytest.mark.integration +async def test_conducted_run_has_no_capture_code(db_pool: asyncpg.Pool) -> None: + """A Conducted Run has no capture_code at all: `None`, distinct from + a Witnessed Run whose genesis did stamp one.""" + run_id = uuid4() + deps = _build_deps(db_pool, [*_chain_ids(), run_id, uuid4()]) + plan_id = await _seed_plan(deps, family_name="TomographyConductedNoCode") + await bind_start(deps)( + StartRun(name="conducted-no-code", plan_id=plan_id, subject_id=None), + principal_id=_PRINCIPAL_ID, + correlation_id=_CORRELATION_ID, + ) + await _drain(db_pool) + + handler = bind_list(deps) + page = await handler( + ListRuns(plan_id=plan_id, limit=10), + principal_id=_PRINCIPAL_ID, + correlation_id=_CORRELATION_ID, + ) + + item = next(item for item in page.items if item.run_id == run_id) + assert item.capture_code is None diff --git a/apps/api/tests/unit/api/test_capture_observer.py b/apps/api/tests/unit/api/test_capture_observer.py index 3b4f372ca3..cef20a8b00 100644 --- a/apps/api/tests/unit/api/test_capture_observer.py +++ b/apps/api/tests/unit/api/test_capture_observer.py @@ -17,12 +17,17 @@ import pytest -from cora.api._capture_observer import ControlPortCaptureObserver, classify_capture_status +from cora.api._capture_observer import ( + FULL_FILE_NAME_TRUNCATION_THRESHOLD, + ControlPortCaptureObserver, + classify_capture_status, +) from cora.operation.ports.control_port import ControlNotConnectedError, Measurement from cora.run.ports.capture_observer import ( AnyCaptureObservation, CaptureLifecycleObservation, CaptureObserverScope, + CapturePathObservation, CapturePhase, CapturePreconditionBypassObservation, CaptureProgressObservation, @@ -898,3 +903,136 @@ async def test_observe_server_running_role_stays_declared_and_unread() -> None: observations = await _collect_any(observer, {"tomoscan"}) assert not any(o.source_id == "pvRunning" for o in observations) + + +# ---------- Full file name role (slice 13) ---------- + + +@pytest.mark.unit +async def test_observe_a_code_with_no_full_file_name_role_is_unaffected() -> None: + """A code declaring no `full_file_name` role behaves exactly as + before this role existed: no pump, no extra observation.""" + port = _ScriptedControlPort(readings={"pvA": [_reading("Beginning scan")]}) + observer = _observer(port, {"tomoscan": {"status": "pvA"}}) + + observations = await _collect_any(observer, {"tomoscan"}) + + assert not any(isinstance(o, CapturePathObservation) for o in observations) + + +@pytest.mark.unit +async def test_observe_a_full_file_name_reading_is_a_path_observation() -> None: + port = _ScriptedControlPort( + readings={ + "pvA": [_reading("Beginning scan")], + "pvFile": [_reading("/data/2026-01-Smith-12345/scan_001.h5")], + } + ) + observer = _observer(port, {"tomoscan": {"status": "pvA", "full_file_name": "pvFile"}}) + + observations = await _collect_any(observer, {"tomoscan"}) + + paths = [o for o in observations if isinstance(o, CapturePathObservation)] + assert len(paths) == 1 + assert paths[0].capture_code == "tomoscan" + assert paths[0].observed_path == "/data/2026-01-Smith-12345/scan_001.h5" + assert paths[0].reach_tier is ReachTier.RELAYED + assert paths[0].source_id == "pvFile" + assert paths[0].observed_at == _T + + +@pytest.mark.unit +async def test_observe_an_empty_full_file_name_reading_emits_nothing() -> None: + """The fresh-IOC-boot state: a fine, ordinary outcome, not an error, + but not a fact to enqueue either.""" + port = _ScriptedControlPort( + readings={"pvA": [_reading("Beginning scan")], "pvFile": [_reading("")]} + ) + observer = _observer(port, {"tomoscan": {"status": "pvA", "full_file_name": "pvFile"}}) + + observations = await _collect_any(observer, {"tomoscan"}) + + assert not any(isinstance(o, CapturePathObservation) for o in observations) + + +@pytest.mark.unit +async def test_observe_a_non_text_full_file_name_reading_emits_nothing() -> None: + """A non-str value means the adapter's text-waveform decode did not + apply; never coerced or guessed at.""" + port = _ScriptedControlPort( + readings={"pvA": [_reading("Beginning scan")], "pvFile": [_reading(0)]} + ) + observer = _observer(port, {"tomoscan": {"status": "pvA", "full_file_name": "pvFile"}}) + + observations = await _collect_any(observer, {"tomoscan"}) + + assert not any(isinstance(o, CapturePathObservation) for o in observations) + + +@pytest.mark.unit +async def test_observe_a_suspected_truncated_full_file_name_reading_emits_nothing() -> None: + """A decoded value at or over the truncation threshold is + indistinguishable from a wire truncation (NELM=512 on the real PV, + 511 usable chars): rejected rather than recorded as if complete.""" + long_path = "/" + ("a" * (FULL_FILE_NAME_TRUNCATION_THRESHOLD - 1)) + assert len(long_path) == FULL_FILE_NAME_TRUNCATION_THRESHOLD + port = _ScriptedControlPort( + readings={"pvA": [_reading("Beginning scan")], "pvFile": [_reading(long_path)]} + ) + observer = _observer(port, {"tomoscan": {"status": "pvA", "full_file_name": "pvFile"}}) + + observations = await _collect_any(observer, {"tomoscan"}) + + assert not any(isinstance(o, CapturePathObservation) for o in observations) + + +@pytest.mark.unit +async def test_observe_a_full_file_name_reading_just_under_the_threshold_is_accepted() -> None: + ok_path = "/" + ("a" * (FULL_FILE_NAME_TRUNCATION_THRESHOLD - 2)) + assert len(ok_path) == FULL_FILE_NAME_TRUNCATION_THRESHOLD - 1 + port = _ScriptedControlPort( + readings={"pvA": [_reading("Beginning scan")], "pvFile": [_reading(ok_path)]} + ) + observer = _observer(port, {"tomoscan": {"status": "pvA", "full_file_name": "pvFile"}}) + + observations = await _collect_any(observer, {"tomoscan"}) + + paths = [o for o in observations if isinstance(o, CapturePathObservation)] + assert len(paths) == 1 + assert paths[0].observed_path == ok_path + + +@pytest.mark.unit +async def test_observe_full_file_name_pump_has_no_unreached_counterpart() -> None: + """A disconnect or clean stream end simply stops the pump: it must + NOT synthesize a `CapturePathObservation`. Mirrors the `testing` + role's identical guarantee, for the identical reason: erasing the + last retained reading on every reconnect would defeat the + dual-clock discipline `observed_at` exists to provide.""" + port = _ScriptedControlPort( + readings={"pvA": [_reading("Beginning scan")], "pvFile": []}, + disconnect=frozenset({"pvFile"}), + ) + observer = _observer(port, {"tomoscan": {"status": "pvA", "full_file_name": "pvFile"}}) + + observations = await _collect_any(observer, {"tomoscan"}) + + assert not any(isinstance(o, CapturePathObservation) for o in observations) + + +@pytest.mark.unit +async def test_observe_full_file_name_reading_with_no_substrate_time_is_none_not_synthesized() -> ( + None +): + port = _ScriptedControlPort( + readings={ + "pvA": [_reading("Beginning scan")], + "pvFile": [_reading("/data/scan.h5", produced_at=None)], + } + ) + observer = _observer(port, {"tomoscan": {"status": "pvA", "full_file_name": "pvFile"}}) + + observations = await _collect_any(observer, {"tomoscan"}) + + path = next(o for o in observations if isinstance(o, CapturePathObservation)) + assert path.observed_at is None 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 6ce6b5368b..60336b0b6c 100644 --- a/apps/api/tests/unit/api/test_capture_watch_preflight.py +++ b/apps/api/tests/unit/api/test_capture_watch_preflight.py @@ -214,6 +214,112 @@ async def test_preflight_read_testing_role_unrecognized_token_is_bad() -> None: assert line.verdict == "unrecognized" +@pytest.mark.unit +async def test_preflight_read_full_file_name_role_redacts_the_real_value() -> None: + """The one role whose printed `value` is REDACTED: `observed_path` + is personal data (see `_capture_observer.py`). `kind` stays real.""" + port = _FakeControlPort({"pv:file": _reading("/data/2026-01-Smith-12345/scan_001.h5")}) + + report = await _preflight(port, {"code": {"full_file_name": "pv:file"}}) + + (line,) = report.lines + assert line.ok + assert "Smith" not in str(line.value) + assert "Smith" not in line.render() + assert line.value == "" + assert line.verdict == "text(len=37)" + + +@pytest.mark.unit +async def test_preflight_read_full_file_name_role_empty_string_is_ok() -> None: + """The fresh-IOC-boot state: a fine, ordinary outcome, not a + connectivity or decode problem.""" + port = _FakeControlPort({"pv:file": _reading("")}) + + report = await _preflight(port, {"code": {"full_file_name": "pv:file"}}) + + (line,) = report.lines + assert line.ok + assert line.verdict == "empty" + assert line.value == "" + + +@pytest.mark.unit +async def test_preflight_read_full_file_name_role_suspected_truncated_is_bad() -> None: + from cora.api._capture_observer import FULL_FILE_NAME_TRUNCATION_THRESHOLD + + long_path = "/" + ("a" * (FULL_FILE_NAME_TRUNCATION_THRESHOLD - 1)) + port = _FakeControlPort({"pv:file": _reading(long_path)}) + + report = await _preflight(port, {"code": {"full_file_name": "pv:file"}}) + + (line,) = report.lines + assert not line.ok + assert line.verdict == "suspected-truncated" + assert "a" * 10 not in str(line.value) # never the real (redacted) content + + +@pytest.mark.unit +async def test_preflight_read_full_file_name_role_just_under_the_threshold_is_ok() -> None: + from cora.api._capture_observer import FULL_FILE_NAME_TRUNCATION_THRESHOLD + + ok_path = "/" + ("a" * (FULL_FILE_NAME_TRUNCATION_THRESHOLD - 2)) + port = _FakeControlPort({"pv:file": _reading(ok_path)}) + + report = await _preflight(port, {"code": {"full_file_name": "pv:file"}}) + + (line,) = report.lines + assert line.ok + assert line.verdict == f"text(len={FULL_FILE_NAME_TRUNCATION_THRESHOLD - 1})" + + +@pytest.mark.unit +async def test_preflight_read_full_file_name_role_non_text_is_bad() -> None: + port = _FakeControlPort({"pv:file": _reading(0)}) + + report = await _preflight(port, {"code": {"full_file_name": "pv:file"}}) + + (line,) = report.lines + assert not line.ok + assert line.verdict == "non-text" + assert line.value == "" + + +@pytest.mark.unit +async def test_preflight_read_a_mis_keyed_role_with_a_path_value_still_redacts() -> None: + """Defense-in-depth: a `capture_watch_pvs` role-key typo + (`"full_filename"` instead of `"full_file_name"`) falls through to + the generic unknown-role branch (verdict `n/a`), which must NOT + print the real path just because the role name didn't match -- + this preflight tool is exactly what an operator runs, screenshots, + and pastes into a ticket while debugging that kind of typo.""" + port = _FakeControlPort({"pv:file": _reading("/data/2026-01-Smith-12345/scan_001.h5")}) + + report = await _preflight(port, {"code": {"full_filename": "pv:file"}}) + + (line,) = report.lines + assert line.pv_key == "full_filename" + assert line.verdict == "n/a" + assert "Smith" not in str(line.value) + assert "Smith" not in line.render() + assert line.value == "" + + +@pytest.mark.unit +async def test_preflight_read_baseline_full_file_name_pv_misdeclared_still_redacts() -> None: + """Same defense, for the OTHER wrong-dict mistake: the + `full_file_name` PV declared under `capture_baseline_pvs` instead + of `capture_watch_pvs`.""" + port = _FakeControlPort({"pv:file": _reading("/data/2026-01-Smith-12345/scan_001.h5")}) + + report = await _preflight(port, {}, baseline_pvs={"code": {"FullFileName": "pv:file"}}) + + (line,) = report.lines + assert "Smith" not in str(line.value) + assert "Smith" not in line.render() + assert line.value == "" + + @pytest.mark.unit async def test_preflight_read_progress_role_non_numeric_is_bad() -> None: port = _FakeControlPort({"pv:saved": _reading("garbled")}) diff --git a/apps/api/tests/unit/api/test_run_supervisor.py b/apps/api/tests/unit/api/test_run_supervisor.py index 4a1be9c075..e5b79a5e20 100644 --- a/apps/api/tests/unit/api/test_run_supervisor.py +++ b/apps/api/tests/unit/api/test_run_supervisor.py @@ -266,6 +266,7 @@ def _running_item( snr_limit=snr_limit, expected_observation_interval_seconds=expected_observation_interval_seconds, conduct_mode=conduct_mode, + capture_code=None, ) @@ -840,6 +841,7 @@ def _held_item(run_id: UUID) -> RunSummaryItem: snr_limit=None, expected_observation_interval_seconds=None, conduct_mode="Conducted", + capture_code=None, ) diff --git a/apps/api/tests/unit/api/test_run_witness.py b/apps/api/tests/unit/api/test_run_witness.py index 6dc8e50ad5..f71917b57a 100644 --- a/apps/api/tests/unit/api/test_run_witness.py +++ b/apps/api/tests/unit/api/test_run_witness.py @@ -39,6 +39,7 @@ from cora.infrastructure.routing import NIL_SENTINEL_ID from cora.run.aggregates.run import ( ConductMode, + InMemoryCapturePathStore, InMemoryFeedHeartbeatStore, RunStarted, event_type_name, @@ -54,6 +55,7 @@ AnyCaptureObservation, CaptureLifecycleObservation, CaptureObserverScope, + CapturePathObservation, CapturePhase, CapturePreconditionBypassObservation, CaptureProgressObservation, @@ -121,6 +123,22 @@ def _testing_obs( ) +def _path_obs( + *, + observed_path: str = "/data/2026-01-Smith-12345/scan_001.h5", + capture_code: str = _CODE, + observed_at: datetime | None = _NOW, +) -> CapturePathObservation: + return CapturePathObservation( + capture_code=capture_code, + observed_path=observed_path, + reach_tier=ReachTier.RELAYED, + observed_at=observed_at, + source_kind="EpicsPv", + source_id="2bmSP2:HDF1:FullFileName_RBV", + ) + + class _FakeObserver: """Yields a fixed observation sequence once, then ends the stream.""" @@ -551,11 +569,14 @@ def _recorder( open_captures: dict[str, UUID] | None = None, baseline_reader: object | None = None, capture_baseline_recording_enabled: bool = False, + capture_path_store: object | None = None, + capture_path_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, ) outcome = record_witnessed_run_outcome or _FakeRecordWitnessedRunOutcome() truncate = truncate_run or _FakeTruncateRun() @@ -567,6 +588,7 @@ def _recorder( settings=settings, open_captures=open_captures, baseline_reader=baseline_reader, # type: ignore[arg-type] + capture_path_store=capture_path_store, # type: ignore[arg-type] ) @@ -693,6 +715,274 @@ async def test_promotion_survives_a_baseline_read_failure() -> None: assert recorder.open_captures() == {_CODE: fake.run_id} +# ---------- Capture path pairing / dual-clock guard (slice 13) ---------- + +_T0 = _NOW +_T1 = datetime(2026, 8, 13, 12, 5, 0, tzinfo=UTC) + + +class _FakeCapturePathStore: + """Records every `upsert()` call, or raises a configured exception + instead. `get()` mirrors `InMemoryCapturePathStore` for tests that + need to read back what was (or wasn't) written.""" + + def __init__(self, *, raises: Exception | None = None) -> None: + self.calls: list[dict[str, Any]] = [] + self._raises = raises + + async def upsert( + self, *, run_id: UUID, observed_path: str, observed_at: datetime, created_at: datetime + ) -> None: + self.calls.append( + { + "run_id": run_id, + "observed_path": observed_path, + "observed_at": observed_at, + "created_at": created_at, + } + ) + if self._raises is not None: + raise self._raises + + +@pytest.mark.unit +async def test_capture_path_recorded_when_observed_after_begun() -> None: + """The normal case: the file plugin opens a file sometime after + BEGUN, before the terminal. The guard passes and the vault gets the + real path.""" + run_id = uuid4() + genesis = _FakeRecordWitnessedRun(run_id=run_id) + store = InMemoryCapturePathStore() + recorder = _recorder( + record_witnessed_run=genesis, + capture_path_store=store, + capture_path_recording_enabled=True, + ) + + await recorder.observe_capture( + _obs(reported_status="Beginning scan", phase=CapturePhase.BEGUN, observed_at=_T0) + ) + recorder.observe_capture_path(_path_obs(observed_at=_T1)) + await recorder.observe_capture(_obs(reported_status="Scan complete", phase=CapturePhase.ENDED)) + + row = await store.get(run_id) + assert row is not None + assert row.observed_path == "/data/2026-01-Smith-12345/scan_001.h5" + assert row.observed_at == _T1 + + +@pytest.mark.unit +async def test_capture_path_not_recorded_when_observed_before_begun() -> None: + """Finding A's exact race: a retained reading whose OWN substrate + time predates this capture's BEGUN almost certainly describes the + PREVIOUS capture's file (e.g. a CA redelivery on reconnect). The + guard must reject it, not attach it to the wrong Run.""" + run_id = uuid4() + genesis = _FakeRecordWitnessedRun(run_id=run_id) + store = InMemoryCapturePathStore() + recorder = _recorder( + record_witnessed_run=genesis, + capture_path_store=store, + capture_path_recording_enabled=True, + ) + + await recorder.observe_capture( + _obs(reported_status="Beginning scan", phase=CapturePhase.BEGUN, observed_at=_T1) + ) + # Arrives AFTER promotion (so _promote's own clear cannot catch it) + # but carries an OLDER substrate time than BEGUN's own. + recorder.observe_capture_path(_path_obs(observed_at=_T0)) + await recorder.observe_capture(_obs(reported_status="Scan complete", phase=CapturePhase.ENDED)) + + assert await store.get(run_id) is None + + +@pytest.mark.unit +async def test_capture_path_not_recorded_when_never_observed() -> None: + """A missing filename is a fine outcome, per the slice's own design + lock: no reading arrived, so nothing is written, and nothing raises.""" + run_id = uuid4() + genesis = _FakeRecordWitnessedRun(run_id=run_id) + store = InMemoryCapturePathStore() + recorder = _recorder( + record_witnessed_run=genesis, + capture_path_store=store, + capture_path_recording_enabled=True, + ) + + await recorder.observe_capture(_obs(reported_status="Beginning scan", phase=CapturePhase.BEGUN)) + await recorder.observe_capture(_obs(reported_status="Scan complete", phase=CapturePhase.ENDED)) + + assert await store.get(run_id) is None + + +@pytest.mark.unit +async def test_capture_path_not_recorded_when_the_kill_switch_is_off() -> None: + """Declaring a store is not sufficient; capture_path_recording_enabled + is the fifth, independent kill switch.""" + run_id = uuid4() + genesis = _FakeRecordWitnessedRun(run_id=run_id) + store = InMemoryCapturePathStore() + recorder = _recorder( + record_witnessed_run=genesis, + capture_path_store=store, + capture_path_recording_enabled=False, + ) + + await recorder.observe_capture( + _obs(reported_status="Beginning scan", phase=CapturePhase.BEGUN, observed_at=_T0) + ) + recorder.observe_capture_path(_path_obs(observed_at=_T1)) + await recorder.observe_capture(_obs(reported_status="Scan complete", phase=CapturePhase.ENDED)) + + assert await store.get(run_id) is None + + +@pytest.mark.unit +async def test_capture_path_with_no_store_configured_is_unaffected() -> None: + run_id = uuid4() + genesis = _FakeRecordWitnessedRun(run_id=run_id) + recorder = _recorder( + record_witnessed_run=genesis, + capture_path_store=None, + capture_path_recording_enabled=True, + ) + + await recorder.observe_capture( + _obs(reported_status="Beginning scan", phase=CapturePhase.BEGUN, observed_at=_T0) + ) + recorder.observe_capture_path(_path_obs(observed_at=_T1)) + await recorder.observe_capture( + _obs(reported_status="Scan complete", phase=CapturePhase.ENDED) + ) # must not raise + + +@pytest.mark.unit +async def test_capture_path_retention_does_not_carry_into_the_next_promotion() -> None: + """A path retained (and written) for one capture must not silently + re-attach to a LATER capture of the same code that never got its own + fresh reading -- the eviction on outcome success must actually run.""" + first_run_id = uuid4() + second_run_id = uuid4() + genesis = _FakeRecordWitnessedRun(run_id=first_run_id) + store = InMemoryCapturePathStore() + recorder = _recorder( + record_witnessed_run=genesis, + capture_path_store=store, + capture_path_recording_enabled=True, + ) + + await recorder.observe_capture( + _obs(reported_status="Beginning scan", phase=CapturePhase.BEGUN, observed_at=_T0) + ) + recorder.observe_capture_path(_path_obs(observed_at=_T1)) + await recorder.observe_capture(_obs(reported_status="Scan complete", phase=CapturePhase.ENDED)) + assert await store.get(first_run_id) is not None + + genesis.run_id = second_run_id + await recorder.observe_capture( + _obs(reported_status="Beginning scan", phase=CapturePhase.BEGUN, observed_at=_T1) + ) + await recorder.observe_capture(_obs(reported_status="Scan complete", phase=CapturePhase.ENDED)) + + assert await store.get(second_run_id) is None + + +@pytest.mark.unit +async def test_capture_path_write_failure_does_not_unwind_the_outcome() -> None: + """By the time this runs, record_witnessed_run_outcome has already + succeeded (mirrors `_read_baseline`'s exact posture): a vault-write + failure must be logged and must never unwind it.""" + run_id = uuid4() + genesis = _FakeRecordWitnessedRun(run_id=run_id) + outcome = _FakeRecordWitnessedRunOutcome() + store = _FakeCapturePathStore(raises=RuntimeError("boom")) + recorder = _recorder( + record_witnessed_run=genesis, + record_witnessed_run_outcome=outcome, + capture_path_store=store, + capture_path_recording_enabled=True, + ) + + await recorder.observe_capture( + _obs(reported_status="Beginning scan", phase=CapturePhase.BEGUN, observed_at=_T0) + ) + recorder.observe_capture_path(_path_obs(observed_at=_T1)) + await recorder.observe_capture( + _obs(reported_status="Scan complete", phase=CapturePhase.ENDED) + ) # must not raise + + assert len(outcome.calls) == 1 + assert len(store.calls) == 1 + + +@pytest.mark.unit +async def test_capture_path_write_never_logs_the_observed_path() -> None: + """`observed_path` is personal data: no log line this feature emits + may ever contain it, across the whole promote -> terminal -> write + flow.""" + run_id = uuid4() + genesis = _FakeRecordWitnessedRun(run_id=run_id) + store = InMemoryCapturePathStore() + recorder = _recorder( + record_witnessed_run=genesis, + capture_path_store=store, + capture_path_recording_enabled=True, + ) + marker = "2026-01-Smith-12345" + + with structlog.testing.capture_logs() as logs: + await recorder.observe_capture( + _obs(reported_status="Beginning scan", phase=CapturePhase.BEGUN, observed_at=_T0) + ) + recorder.observe_capture_path(_path_obs(observed_path=f"/data/{marker}/scan_001.h5")) + await recorder.observe_capture( + _obs(reported_status="Scan complete", phase=CapturePhase.ENDED) + ) + + for entry in logs: + for value in entry.values(): + assert marker not in str(value), f"leaked observed_path fragment in log entry: {entry}" + + +@pytest.mark.unit +async def test_capture_path_with_no_substrate_time_is_never_recorded() -> None: + """`_resolve_capture_path`'s `observed_at is None` short-circuit + (checked BEFORE the `<` comparison, to avoid `None < datetime`): + a retained reading with no substrate time can never satisfy the + guard, since there is nothing to compare against BEGUN.""" + run_id = uuid4() + genesis = _FakeRecordWitnessedRun(run_id=run_id) + store = InMemoryCapturePathStore() + recorder = _recorder( + record_witnessed_run=genesis, + capture_path_store=store, + capture_path_recording_enabled=True, + ) + + await recorder.observe_capture( + _obs(reported_status="Beginning scan", phase=CapturePhase.BEGUN, observed_at=_T0) + ) + recorder.observe_capture_path(_path_obs(observed_at=None)) + await recorder.observe_capture( + _obs(reported_status="Scan complete", phase=CapturePhase.ENDED) + ) # must not raise (no None < datetime comparison) + + assert await store.get(run_id) is None + + +@pytest.mark.unit +async def test_observe_capture_path_is_a_noop_when_recording_disabled() -> None: + """Shadow mode must retain nothing at all, mirroring + `test_observe_precondition_bypass_is_a_noop_when_recording_disabled`.""" + genesis = _FakeRecordWitnessedRun() + recorder = _recorder(record_witnessed_run=genesis, run_witness_recording_enabled=False) + + recorder.observe_capture_path(_path_obs()) + + assert recorder._last_capture_path == {} + + @pytest.mark.unit async def test_run_witness_recorder_truncates_stale_run_and_repromotes_on_a_second_begun() -> None: """A second BEGUN for a code that is already open means the previous @@ -1653,6 +1943,7 @@ def _summary_item(*, run_id: UUID, conduct_mode: str = "Witnessed") -> RunSummar snr_limit=None, expected_observation_interval_seconds=None, conduct_mode=conduct_mode, + capture_code=None, ) 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 058da3e8f8..d5d843e5d7 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 @@ -15,6 +15,9 @@ Slice 12 adds a FOURTH gate, same shape: `capture_baseline_recording_enabled=True` requires `run_witness_recording_enabled=True`. + +Slice 13 adds a FIFTH gate, same shape: `capture_path_recording_enabled=True` +requires `run_witness_recording_enabled=True`. """ from uuid import UUID, uuid4 @@ -32,6 +35,7 @@ def _settings( run_witness_recording_enabled: bool = False, capture_progress_recording_enabled: bool = False, capture_baseline_recording_enabled: bool = False, + capture_path_recording_enabled: bool = False, ) -> Settings: return Settings( # type: ignore[call-arg] run_witness_enabled=run_witness_enabled, @@ -39,6 +43,7 @@ def _settings( run_witness_recording_enabled=run_witness_recording_enabled, capture_progress_recording_enabled=capture_progress_recording_enabled, capture_baseline_recording_enabled=capture_baseline_recording_enabled, + capture_path_recording_enabled=capture_path_recording_enabled, ) @@ -174,3 +179,41 @@ def test_baseline_recording_enabled_with_run_witness_recording_passes() -> None: capture_baseline_recording_enabled=True, ) ) + + +def test_path_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_path_recording_enabled=True, + ) + ) + + +def test_path_recording_enabled_checked_before_the_first_gates_prerequisites() -> None: + """Same independence property as the progress / baseline gates: the + path 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_path_recording_enabled=True, + ) + ) + + +def test_path_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_path_recording_enabled=True, + ) + ) diff --git a/apps/api/tests/unit/run/test_capture_path.py b/apps/api/tests/unit/run/test_capture_path.py new file mode 100644 index 0000000000..35a194d42b --- /dev/null +++ b/apps/api/tests/unit/run/test_capture_path.py @@ -0,0 +1,101 @@ +"""Unit tests for the `run_capture_path` PII vault's InMemory adapter +and the `load_run_capture_path` display-fallback helper (slice 13). + +Mirrors `test_feed_heartbeats.py`'s shape: exercise the store contract +directly, no recorder or observer involved. +""" + +from datetime import UTC, datetime, timedelta +from uuid import uuid4 + +import pytest + +from cora.run.aggregates.run import ( + UNOBSERVED_CAPTURE_PATH, + InMemoryCapturePathStore, + load_run_capture_path, +) + +_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() -> None: + store = InMemoryCapturePathStore() + run_id = uuid4() + + await store.upsert( + run_id=run_id, + observed_path="/data/2026-01-Smith-12345/scan_001.h5", + observed_at=_at(0), + created_at=_at(1), + ) + + row = await store.get(run_id) + assert row is not None + assert row.run_id == run_id + assert row.observed_path == "/data/2026-01-Smith-12345/scan_001.h5" + assert row.observed_at == _at(0) + assert row.created_at == _at(1) + assert row.updated_at == _at(1) + + +@pytest.mark.unit +async def test_get_absent_run_id_returns_none() -> None: + store = InMemoryCapturePathStore() + 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 rewrite, e.g. a retry) + updates the path and observed_at 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 (`datetime.now(tz=UTC)`), mirroring + `InMemoryProfileStore`'s identical Postgres-semantics-preserved + convention (the real adapter's `ON CONFLICT DO UPDATE` sets + `updated_at = now()`, the DB's clock, never the caller-supplied + `created_at` parameter) -- never the fixed `_at(5)` passed in. + """ + store = InMemoryCapturePathStore() + run_id = uuid4() + await store.upsert( + run_id=run_id, observed_path="/data/first.h5", observed_at=_at(0), created_at=_at(0) + ) + before_second_upsert = datetime.now(tz=UTC) + + await store.upsert( + run_id=run_id, observed_path="/data/second.h5", observed_at=_at(5), created_at=_at(5) + ) + + row = await store.get(run_id) + assert row is not None + assert row.observed_path == "/data/second.h5" + assert row.observed_at == _at(5) + assert row.created_at == _at(0) + assert row.updated_at >= before_second_upsert + + +@pytest.mark.unit +async def test_load_run_capture_path_returns_the_real_path_when_present() -> None: + store = InMemoryCapturePathStore() + run_id = uuid4() + await store.upsert( + run_id=run_id, observed_path="/data/a.h5", observed_at=_at(0), created_at=_at(0) + ) + + assert await load_run_capture_path(store, run_id) == "/data/a.h5" + + +@pytest.mark.unit +async def test_load_run_capture_path_falls_back_when_absent() -> None: + """Absence here means 'never observed or rejected by the dual-clock + guard', not erasure -- there is no erasure slice yet -- but the + fallback shape is the same as `load_actor_display_name`'s.""" + store = InMemoryCapturePathStore() + assert await load_run_capture_path(store, uuid4()) == UNOBSERVED_CAPTURE_PATH 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 95c7da003a..98151aa8f1 100644 --- a/apps/api/tests/unit/run/test_get_run_handler.py +++ b/apps/api/tests/unit/run/test_get_run_handler.py @@ -1,7 +1,11 @@ """Unit tests for the `get_run` query handler. Mirrors `test_get_plan_handler.py`. Round-trip seed + get verifies -fold-on-read returns the started Run. +fold-on-read returns the started Run, composed into a `RunView` +(slice 13) mirroring `get_actor`'s `ActorView`. `capture_code` / +`observed_capture_path` resolution itself is covered by +`test_get_run_route.py` and the `get_run` contract tests; these tests +pin the handler's OWN contract (authz, fold-on-read, RunView shape). """ from datetime import UTC, datetime @@ -13,6 +17,7 @@ from cora.infrastructure.event_envelope import to_new_event from cora.run import RunHandlers, UnauthorizedError, wire_run from cora.run.aggregates.run import ( + InMemoryCapturePathStore, Run, RunName, RunStatus, @@ -72,19 +77,22 @@ 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) - run = await handler( + handler = get_run.bind(deps, capture_path_store=InMemoryCapturePathStore()) + view = await handler( GetRun(run_id=_RUN_ID), principal_id=_PRINCIPAL_ID, correlation_id=_CORRELATION_ID, ) - assert run == Run( + assert view is not None + assert view.run == Run( id=_RUN_ID, name=RunName("32-ID FlyScan"), plan_id=_PLAN_ID, subject_id=_SUBJECT_ID, status=RunStatus.RUNNING, ) + assert view.capture_code is None + assert view.observed_capture_path is None @pytest.mark.unit @@ -93,26 +101,26 @@ 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) - run = await handler( + handler = get_run.bind(deps, capture_path_store=InMemoryCapturePathStore()) + view = await handler( GetRun(run_id=_RUN_ID), principal_id=_PRINCIPAL_ID, correlation_id=_CORRELATION_ID, ) - assert run is not None - assert run.subject_id is None + assert view is not None + assert view.run.subject_id is 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) - run = await handler( + handler = get_run.bind(deps, capture_path_store=InMemoryCapturePathStore()) + view = await handler( GetRun(run_id=uuid4()), principal_id=_PRINCIPAL_ID, correlation_id=_CORRELATION_ID, ) - assert run is None + assert view is None @pytest.mark.unit @@ -120,7 +128,7 @@ 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) + handler = get_run.bind(deps, capture_path_store=InMemoryCapturePathStore()) await handler( GetRun(run_id=uuid4()), principal_id=_PRINCIPAL_ID, @@ -133,7 +141,7 @@ 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) + handler = get_run.bind(deps, capture_path_store=InMemoryCapturePathStore()) 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 new file mode 100644 index 0000000000..a2914d583e --- /dev/null +++ b/apps/api/tests/unit/run/test_get_run_route.py @@ -0,0 +1,155 @@ +"""Unit tests for `get_run`'s route-level capture-path DTO mapping +(slice 13). + +Calls `route.get_runs` directly (a plain async function; FastAPI's +`Depends`/`Annotated` wrapping does not prevent direct invocation in a +unit test), seeding a real `Run` via `InMemoryEventStore` so +`run.external_refs` is genuinely folded, not faked. Mirrors +`test_get_run_handler.py`'s `_seed_run` shape. + +`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. +""" + +from datetime import UTC, datetime +from uuid import UUID, uuid4 + +import pytest + +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.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 +from tests.unit._helpers import build_deps + +_NOW = datetime(2026, 8, 16, 12, 0, 0, tzinfo=UTC) +_PLAN_ID = UUID("01900000-0000-7000-8000-00000000ff02") +_PRINCIPAL_ID = UUID("01900000-0000-7000-8000-000000000099") +_CORRELATION_ID = UUID("01900000-0000-7000-8000-0000000000aa") + + +async def _seed_run( + store: InMemoryEventStore, + run_id: UUID, + *, + capture_code: str | None, + name: str = "witnessed-run", +) -> None: + external_refs = ( + ({"scheme": "capture-code", "value": capture_code},) if capture_code is not None else () + ) + event = RunStarted( + run_id=run_id, + name=name, + plan_id=_PLAN_ID, + subject_id=None, + external_refs=external_refs, + occurred_at=_NOW, + ) + new_event = to_new_event( + event_type=event_type_name(event), + payload=to_payload(event), + occurred_at=_NOW, + event_id=uuid4(), + command_name="StartRun", + correlation_id=_CORRELATION_ID, + principal_id=uuid4(), + ) + await store.append(stream_type="Run", stream_id=run_id, expected_version=0, events=[new_event]) + + +@pytest.mark.unit +async def test_get_run_route_resolves_the_real_path_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) + capture_path_store = InMemoryCapturePathStore() + await capture_path_store.upsert( + run_id=run_id, + observed_path="/data/2026-01-Smith-12345/scan_001.h5", + observed_at=_NOW, + created_at=_NOW, + ) + handler = get_run.bind(deps, capture_path_store=capture_path_store) + + response = await get_runs( + run_id, + handler, + _CORRELATION_ID, + _PRINCIPAL_ID, + NIL_SENTINEL_ID, + ) + + assert response.capture_code == "2bmb-tomoscan" + assert response.observed_capture_path == "/data/2026-01-Smith-12345/scan_001.h5" + + +@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()) + + response = await get_runs( + run_id, + handler, + _CORRELATION_ID, + _PRINCIPAL_ID, + NIL_SENTINEL_ID, + ) + + assert response.capture_code == "2bmb-tomoscan-2" + assert response.observed_capture_path == UNOBSERVED_CAPTURE_PATH + + +@pytest.mark.unit +async def test_get_run_route_a_conducted_run_has_no_capture_code_and_no_tombstone() -> None: + """Never touches the vault at all when there's no capture_code: + 'not applicable' (bare None) is a different fact from 'expected but + missing' (the tombstone).""" + run_id = uuid4() + 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()) + + response = await get_runs( + run_id, + handler, + _CORRELATION_ID, + _PRINCIPAL_ID, + NIL_SENTINEL_ID, + ) + + assert response.capture_code is None + assert response.observed_capture_path is None + + +@pytest.mark.unit +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()) + + with pytest.raises(HTTPException) as exc_info: + await get_runs( + uuid4(), + handler, + _CORRELATION_ID, + _PRINCIPAL_ID, + NIL_SENTINEL_ID, + ) + assert exc_info.value.status_code == 404 diff --git a/infra/atlas/migrations/20260816094243_init_run_capture_path.sql b/infra/atlas/migrations/20260816094243_init_run_capture_path.sql new file mode 100644 index 0000000000..18a9a39f32 --- /dev/null +++ b/infra/atlas/migrations/20260816094243_init_run_capture_path.sql @@ -0,0 +1,87 @@ +-- Slice 13 (memory/project_witnessed_run_prelive_slices.md): PII vault +-- for a witnessed Run's observed capture file path. +-- +-- 2-BM's directory layout embeds a surname and a proposal number +-- (tomoscan_2bm.py:474-477, `{ExperimentYearMonth}-{UserLastName}- +-- {ProposalNumber}`), so the full observed path is personal data by +-- construction. It must never land on an event (events are immutable +-- and INSERT-only at the role level, so PII written there could never +-- be erased). This table mirrors actor_profile (memory/project_pii_vault) +-- exactly: single mutable table, one row per Run, erasable by a future +-- slice via DELETE. +-- +-- Naming: aggregate-prefixed (run_), not proj_-prefixed: this is a +-- mutable vault table, not a projection, same distinction actor_profile +-- draws from proj_access_actor_summary. +-- +-- Schema decisions (mirroring actor_profile'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. +-- - observed_path CHECK upper bound is 511: the areaDetector file +-- plugin's FullFileName_RBV is a DBR_CHAR waveform with NELM=512 +-- (confirmed against ADCore's NDFile.template), so a decoded string +-- of 511+ chars already reads as network-truncated at the +-- application layer and is never appended by RunWitnessRecorder; +-- this CHECK is defense-in-depth against that same fact, not the +-- primary guard. +-- - No forgotten_at / soft-delete column: it would itself be PII +-- ("this Run's capture path existed and was erased on Y"). +-- - No UNIQUE constraint on observed_path: two Runs writing to the +-- same directory on the same day is realistic (multiple scans, one +-- proposal); UNIQUE under RLS also leaks via constraint-violation +-- timing, same reasoning as actor_profile. +-- - created_at is application-supplied (the promotion's own clock +-- read, matching the observation's dual-clock discipline elsewhere +-- in this feature). updated_at defaults to now() for a future +-- rename/rewrite path. +-- +-- RLS posture (defense-in-depth on a mutable PII surface): +-- - ENABLE ROW LEVEL SECURITY: default-deny; the two CREATE POLICY +-- statements below grant the access cora_app needs. +-- - FORCE ROW LEVEL SECURITY: defense-in-depth against the +-- table-owner role bypassing policy. +-- - 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_capture_path ( + run_id UUID PRIMARY KEY, + observed_path TEXT NOT NULL CHECK (length(observed_path) BETWEEN 1 AND 511), + observed_at TIMESTAMPTZ NOT NULL, + created_at TIMESTAMPTZ NOT NULL, + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +COMMENT ON TABLE run_capture_path IS + 'PII vault for a witnessed Run''s observed capture file path (memory/project_witnessed_run_prelive_slices.md, slice 13). Mutable. No SQL FK to events (INSERT-only role); run_id matches the Run aggregate stream_id by application discipline.'; +COMMENT ON COLUMN run_capture_path.run_id IS + 'Matches Run aggregate stream_id. One row per witnessed Run that has a resolved capture path.'; +COMMENT ON COLUMN run_capture_path.observed_path IS + 'The full path read from the areaDetector file plugin''s FullFileName_RBV readback. Personal data (embeds a surname + proposal number at 2-BM): never referenced from an event payload, never logged in full.'; +COMMENT ON COLUMN run_capture_path.observed_at IS + 'The substrate''s own timestamp for this reading (Measurement.produced_at), not CORA''s clock. Used at write time to prove the reading postdates the Run''s own BEGUN time.'; + +-- Mutable PII vault: 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_capture_path TO cora_app; + +-- Row-Level Security: defense-in-depth. +ALTER TABLE run_capture_path ENABLE ROW LEVEL SECURITY; +ALTER TABLE run_capture_path FORCE ROW LEVEL SECURITY; + +CREATE POLICY run_capture_path_cora_app_read + ON run_capture_path FOR SELECT + TO cora_app + USING (true); + +CREATE POLICY run_capture_path_cora_app_write + ON run_capture_path FOR ALL + TO cora_app + USING (true) + WITH CHECK (true); diff --git a/infra/atlas/migrations/20260816094314_add_proj_run_summary_capture_code.sql b/infra/atlas/migrations/20260816094314_add_proj_run_summary_capture_code.sql new file mode 100644 index 0000000000..19e3be56f2 --- /dev/null +++ b/infra/atlas/migrations/20260816094314_add_proj_run_summary_capture_code.sql @@ -0,0 +1,9 @@ +-- Slice 13: surface the capture code a witnessed Run's genesis already +-- stamps onto RunStarted.external_refs (Identifier(scheme="capture-code")), +-- so list_runs can filter/join on it without folding the Run stream. +-- Nullable: Conducted runs (and, defensively, a Witnessed row whose +-- genesis somehow lacked the ref) have none. Sourced from an event the +-- RunSummaryProjection already subscribes to (RunStarted); no new +-- subscription, so the projection-metadata frozenset is unaffected. + +ALTER TABLE proj_run_summary ADD COLUMN capture_code text; diff --git a/infra/atlas/migrations/atlas.sum b/infra/atlas/migrations/atlas.sum index 1586c68a9c..21fea24f03 100644 --- a/infra/atlas/migrations/atlas.sum +++ b/infra/atlas/migrations/atlas.sum @@ -1,4 +1,4 @@ -h1:mlpLiUcmVMHvVR7QS4DPygTUY9ngJqkQyzR5F+aGUW8= +h1:MQxuBVTYB6e2yqkFPUCoiHFLXRHHZvOkJ684l9dXcIw= 20260509120000_init_events.sql h1:GmgCZKfaqXu1m96/cKAks2vhaLWTdEaHTLkFtUo9FXg= 20260509170000_init_idempotency.sql h1:Nbu8DIE4Sv1WiHw3G22+tYffPhKc5Jryw3PMK8wB2zY= 20260510010000_add_event_id.sql h1:RbtYP6uMnOB20zhJ9dNXUi4YVqbmlEzf562pmygnRW8= @@ -164,3 +164,5 @@ h1:mlpLiUcmVMHvVR7QS4DPygTUY9ngJqkQyzR5F+aGUW8= 20260810000000_init_entries_enclosure_permit_probes.sql h1:AgExM2HGWE6XsXJbKyNI3/DCoy8224PH+GQjCL3itkI= 20260810120000_grant_cora_app_entries_table_access.sql h1:f8IxkQu8R7AUVyaItHjCQxDnt/nda2Y7mtRC8blqvB4= 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=