diff --git a/apps/api/src/cora/api/_capture_scan_ingestor.py b/apps/api/src/cora/api/_capture_scan_ingestor.py index f4460be1261..4c2f77f9f5b 100644 --- a/apps/api/src/cora/api/_capture_scan_ingestor.py +++ b/apps/api/src/cora/api/_capture_scan_ingestor.py @@ -39,13 +39,20 @@ re-selects the same oldest row. `tick()` instead excludes each candidate it gives up on THIS tick and tries the next-oldest, up to `_MAX_CANDIDATES_PER_TICK` attempts, stopping at the first real success -(or the first systemic failure -- see below). A persistently-failing -candidate is still retried every tick, forever, by design: the failure -is logged loudly each time rather than parked in a dead-letter table -this slice does not build, so fixing the root cause (adding the -binding, or supplying `captured_at` by hand through the ordinary POST -route) is what actually clears it. It just no longer starves its -siblings while it waits. +(or the first systemic failure -- see below). Most persistently-failing +candidates are retried every tick, forever, by design: the failure is +logged loudly each time rather than parked in a dead-letter table this +slice does not build, so fixing the root cause (adding the binding, or +supplying `captured_at` by hand through the ordinary POST route) is +what actually clears it. It just no longer starves its siblings while +it waits. The one case that is NOT retried forever is a structurally +incomplete file whose Run has already ended: `ingest_scan` records +that as a Shortfall (see its own module docstring's "The second +outcome"), and once that fact lands, `_CANDIDATE_SQL` excludes the +candidate permanently, because retrying would only re-confirm a +verdict that is already final. Every other stuck reason (`no_binding`, +`camera_unconfirmed`, a structurally incomplete file whose Run has NOT +yet ended, etc.) keeps the forever-retried behaviour described above. ## Never blocks, never raises past the tick @@ -194,6 +201,17 @@ SELECT 1 FROM proj_data_dataset_summary dds WHERE dds.producing_run_id = rcp.run_id ) + -- A recorded Shortfall is a terminal verdict on this OBSERVATION + -- (it can never become a Dataset), unlike the other stuck reasons + -- below: retrying would only re-confirm the same fact forever, so + -- this retires the candidate permanently instead of leaving it to + -- spin. Keyed on capture_path_id, not run_id, matching the + -- `exclude` key's own per-location rationale (see + -- `ScanIngestCandidateLookup`'s docstring). + AND NOT EXISTS ( + SELECT 1 FROM proj_data_shortfall_summary dss + WHERE dss.capture_path_id = rcp.capture_path_id + ) ORDER BY rcp.created_at ASC LIMIT 1 """ diff --git a/apps/api/src/cora/data/_projections.py b/apps/api/src/cora/data/_projections.py index a9591defb8c..6dbcf172474 100644 --- a/apps/api/src/cora/data/_projections.py +++ b/apps/api/src/cora/data/_projections.py @@ -6,6 +6,7 @@ DatasetSummaryProjection, DistributionSummaryProjection, EditionSummaryProjection, + ShortfallSummaryProjection, ) from cora.infrastructure.kernel import Kernel from cora.infrastructure.projection import ProjectionRegistry @@ -22,6 +23,7 @@ def register_data_projections( registry.register(DistributionSummaryProjection()) registry.register(EditionSummaryProjection()) registry.register(AttestationSummaryProjection()) + registry.register(ShortfallSummaryProjection()) __all__ = ["register_data_projections"] diff --git a/apps/api/src/cora/data/adapters/capture_path_locator.py b/apps/api/src/cora/data/adapters/capture_path_locator.py index fd306a644ea..66ffc99147a 100644 --- a/apps/api/src/cora/data/adapters/capture_path_locator.py +++ b/apps/api/src/cora/data/adapters/capture_path_locator.py @@ -93,6 +93,7 @@ from __future__ import annotations +from dataclasses import dataclass from pathlib import Path from typing import TYPE_CHECKING, Protocol from urllib.parse import quote, unquote, urlparse @@ -103,6 +104,42 @@ from cora.run.aggregates.run import CapturePath +@dataclass(frozen=True) +class CapturePathReference: + """The non-personal identity of a resolved `run_capture_path` row. + + Deliberately NOT the `CapturePath` row itself. That row carries + `observed_path`, personal data, and this value travels out to + `ingest_scan`'s handler, which must never hold one: handing over + the whole row would put a path within reach of every log line and + every event payload downstream, protected by nothing but care. + These four fields are the ones a caller can carry onto an immutable + record, and `capture_path_locator`'s own docstring is where the + argument for `host` / `root` being safe already lives. + """ + + capture_path_id: UUID + run_id: UUID + host: str + root: str + + +@dataclass(frozen=True) +class ResolvedLocator: + """What an `IngestScan` locator resolved to. + + `reference` is `None` for a pass-through locator (an ordinary + `file://` URI a human supplied), because there is no vault row + behind one. It is present exactly when the locator was an indirect + `cora-capture-path://` reference, which is also precisely the + condition under which the reader's error text must be withheld from + the caller, so it doubles as the redaction signal. + """ + + uri: str + reference: CapturePathReference | None + + class CapturePathLookup(Protocol): """The one method this module ever calls on Run BC's `CapturePathStore`. @@ -209,14 +246,16 @@ async def resolve_capture_path_locator( locator: str, *, capture_path_store: CapturePathLookup, -) -> str | None: +) -> ResolvedLocator | None: """Resolve `locator` to the real `file://` URI the scan reader / - checksum computer can act on. + checksum computer can act on, plus the vault row it came from. Pass-through for every scheme other than `cora-capture-path`: the manual `ingest_scan` POST route and MCP tool keep sending real `file://` URIs directly, unaffected by this module, per the scope decision - that only the automated sweep mints indirect locators. + that only the automated sweep mints indirect locators. A + pass-through resolves to `ResolvedLocator(uri=locator, + reference=None)`: there is no row behind a caller-supplied path. Returns `None` -- never a reason string -- on every failure mode (malformed locator; no vault row at all; a row that exists but not @@ -227,10 +266,16 @@ async def resolve_capture_path_locator( once a forget-style slice calls `CapturePathStore`'s (currently unused) `DELETE` grant, so refusing quietly here is already the right behavior for that future, not a placeholder for it. + + The `reference` is carried out so a caller can key a durable record + on the OBSERVATION this locator named, without a second store read + and without ever holding the path. `ingest_scan` uses it to key a + `Shortfall`; see that aggregate's `_stream_id` for why the + surrogate rather than the path is the only safe seed. """ parsed = urlparse(locator) if parsed.scheme != CAPTURE_PATH_SCHEME: - return locator + return ResolvedLocator(uri=locator, reference=None) segments = [segment for segment in parsed.path.split("/") if segment] if len(segments) < 2: @@ -257,12 +302,30 @@ async def resolve_capture_path_locator( if Path(row.observed_path).name != unquote(filename_segment): return None - return "file://" + quote(row.observed_path) + return ResolvedLocator( + uri="file://" + quote(row.observed_path), + # `host` / `root` are the parsed lookup key rather than + # `row.host` / `row.root`, which `get` documents as an EXACT + # match on that same pair: reading them back off the row would + # look like a second source and is the key echoed back. Using + # the parse keeps that honest, and keeps them `str` rather than + # the row's `str | None`, which is nullable for rows written + # before the vault tracked location and could never have + # matched a concrete key anyway. + reference=CapturePathReference( + capture_path_id=row.capture_path_id, + run_id=row.run_id, + host=host, + root=root, + ), + ) __all__ = [ "CAPTURE_PATH_SCHEME", "CapturePathLookup", + "CapturePathReference", + "ResolvedLocator", "active_scan_transport", "mint_capture_path_locator", "resolve_capture_path_locator", diff --git a/apps/api/src/cora/data/aggregates/shortfall/__init__.py b/apps/api/src/cora/data/aggregates/shortfall/__init__.py new file mode 100644 index 00000000000..23ba0490b1f --- /dev/null +++ b/apps/api/src/cora/data/aggregates/shortfall/__init__.py @@ -0,0 +1,47 @@ +"""Shortfall aggregate: state, status/reason enums, events, evolver, read repo. + +The Shortfall is a slim recorded-fact-chain in the Data BC: the fact +that a capture produced something that can never become a Dataset. +Terminal at genesis, one stream per `capture_path_id`, exactly one +`ShortfallRecorded` event ever. + +No vertical slice of its own. Unlike `Acquisition`, which the +`record_acquisition` slice owns, a Shortfall is appended from the +EXISTING `ingest_scan` slice as its second outcome: the frame counts +that make the fact worth recording are computed by the same reader +pass that refuses the file, and a separate command would need its own +grant, route and MCP tool to record a fact no human asks for. See +`cora.data.features.ingest_scan.handler` for the append site and +`_stream_id.py` for why the stream id is derived. +""" + +from cora.data.aggregates.shortfall._stream_id import shortfall_stream_id +from cora.data.aggregates.shortfall.events import ( + ShortfallEvent, + ShortfallRecorded, + event_type_name, + from_stored, + to_payload, +) +from cora.data.aggregates.shortfall.evolver import evolve, fold +from cora.data.aggregates.shortfall.read import load_shortfall +from cora.data.aggregates.shortfall.state import ( + Shortfall, + ShortfallReason, + ShortfallStatus, +) + +__all__ = [ + "Shortfall", + "ShortfallEvent", + "ShortfallReason", + "ShortfallRecorded", + "ShortfallStatus", + "event_type_name", + "evolve", + "fold", + "from_stored", + "load_shortfall", + "shortfall_stream_id", + "to_payload", +] diff --git a/apps/api/src/cora/data/aggregates/shortfall/_stream_id.py b/apps/api/src/cora/data/aggregates/shortfall/_stream_id.py new file mode 100644 index 00000000000..62bd3d278ed --- /dev/null +++ b/apps/api/src/cora/data/aggregates/shortfall/_stream_id.py @@ -0,0 +1,55 @@ +"""Deterministic stream-id derivation for Shortfall. + +A Shortfall is keyed on the capture observation it judges, one stream +per `capture_path_id`, and the id is derived rather than minted so the +EVENT STORE is what enforces "at most one Shortfall per observation". + +## Why derived, when Acquisition and Attestation mint a fresh UUIDv7 + +Those two are reached once, by a caller who has already decided to +record. This one is reached by a sweep that retries the same candidate +every tick. `_CANDIDATE_SQL` does drop a candidate once its Shortfall +lands, but that read goes through `proj_data_shortfall_summary`, and a +projection lags: between the append and the projection catching up, the +next tick re-selects the same candidate and tries again. With a fresh +id per attempt each of those retries would mint ANOTHER Shortfall +stream, and the record would accumulate one duplicate per tick of +projection lag. + +Deriving the id makes the second append a version conflict against a +stream that already exists, so uniqueness is guaranteed by +`expected_version=0` inside the same transaction rather than by a read +that can be stale. Idempotency is not a projection read. + +## Derived from the surrogate, never from the path + +The input is `capture_path_id`, the opaque surrogate key of the +`run_capture_path` vault row. Never the observed path and never the +filename: both embed `{UserLastName}-{ProposalNumber}` at 2-BM, and a +path-seeded uuid5 would be a confirmation oracle, letting anyone who +could guess a path test that guess against the derived stream id. + +`_DATA_SHORTFALL_NAMESPACE` is a fixed sentinel chosen once and frozen; +it MUST NOT change, or existing Shortfall streams become unreachable. +Mirrors `_FEDERATION_SEAL_NAMESPACE` and the +`_DATA_DISTRIBUTION_BACKFILL_NAMESPACE` precedent in this same BC. +""" + +from uuid import UUID, uuid5 + +_DATA_SHORTFALL_NAMESPACE = UUID("01900000-0000-7000-8000-0000da5f0001") + + +def shortfall_stream_id(capture_path_id: UUID) -> UUID: + """Derive the deterministic Shortfall stream UUID from the vault + row's surrogate key. + + The result is used as BOTH the stream id and the `shortfall_id` on + the payload, same as every other aggregate whose stream id is its + own identity; the difference here is only that the value is derived + instead of minted. + """ + return uuid5(_DATA_SHORTFALL_NAMESPACE, str(capture_path_id)) + + +__all__ = ["shortfall_stream_id"] diff --git a/apps/api/src/cora/data/aggregates/shortfall/events.py b/apps/api/src/cora/data/aggregates/shortfall/events.py new file mode 100644 index 00000000000..0c6dcf76ef3 --- /dev/null +++ b/apps/api/src/cora/data/aggregates/shortfall/events.py @@ -0,0 +1,179 @@ +"""Domain events emitted by the Shortfall aggregate, plus the union. + +Mirrors the locked event-module shape: event classes, discriminated +union, `event_type_name`, `to_payload`, `from_stored`. The +persistence-envelope construction (`NewEvent`) lives at +`cora.infrastructure.event_envelope.to_new_event`. + +Single event ever emitted on a Shortfall stream: + + - `ShortfallRecorded` (genesis-and-terminal): identity, the two + cross-aggregate bindings, the frame accounting, the closed reason, + the finality pair and the `recorded_by` attribution. + +## Payload conventions + + - UUIDs serialize as strings; the two optional counts serialize as + null when None. + - `reason` serializes as its `StrEnum` value, a closed vocabulary + that can never embed a path (see `ShortfallReason`). + - Datetimes serialize via `.isoformat()`. + - Status is NOT carried in the payload; the event type encodes it + (ShortfallRecorded -> RECORDED), same precedent as the rest of the + codebase. + +## Wire payload key ordering (pinned) + +`shortfall_id`, `producing_run_id`, `capture_path_id`, `host`, `root`, +`projection_count`, `commanded_projection_count`, `dropped_frame_count`, +`reason`, `file_modified_at`, `run_ended_at`, `occurred_at`, +`recorded_by`. +""" + +from dataclasses import dataclass +from datetime import datetime +from typing import Any, assert_never +from uuid import UUID + +from cora.data.aggregates.shortfall.state import ShortfallReason +from cora.infrastructure.event_payload import deserialize_or_raise +from cora.infrastructure.ports.event_store import StoredEvent +from cora.shared.identity import ActorId + + +@dataclass(frozen=True) +class ShortfallRecorded: + """A capture was found to have produced something that can never + become a Dataset. + + Status is implicit (`Recorded`); the evolver sets it. This is the + only event the Shortfall aggregate ever emits. + + Every field is a primitive (str, int, UUID, datetime) or the closed + `ShortfallReason` enum. There is no carrier dict and no + free-text field anywhere on this payload, deliberately: the refusal + messages this fact replaces embed the opened path verbatim, and the + row cannot be erased, so the payload is kept to shapes that cannot + carry one. + + `producing_run_id` is NOT optional, unlike `AcquisitionRecorded`'s. + The finality judgement is defined against the Run's terminal, so a + Shortfall without a Run has no evidence that the file is done + changing and must not be recorded at all. + + Fold-symmetry attribution (every-fact-has-an-actor): + - `recorded_by: ActorId`: the envelope `principal_id` of the + caller whose ingest attempt surfaced this. In practice the + `CaptureScanIngestor` agent, but a human hitting the same + refusal through the ordinary route records it identically. + """ + + shortfall_id: UUID + producing_run_id: UUID + capture_path_id: UUID + host: str + root: str + projection_count: int + commanded_projection_count: int | None + dropped_frame_count: int | None + reason: ShortfallReason + file_modified_at: datetime + run_ended_at: datetime + occurred_at: datetime + recorded_by: ActorId + + +# Discriminated union of every event the Shortfall aggregate emits. +# Single-arm today; widening only happens if a retraction event ever +# fires (deliberately unfilled extension space, same posture as +# `AcquisitionEvent`). +ShortfallEvent = ShortfallRecorded + + +def event_type_name(event: ShortfallEvent) -> str: + """Discriminator string written into StoredEvent.event_type.""" + return type(event).__name__ + + +def to_payload(event: ShortfallEvent) -> dict[str, Any]: + """Serialize a Shortfall event to a JSON-friendly dict for jsonb.""" + match event: + case ShortfallRecorded( + shortfall_id=shortfall_id, + producing_run_id=producing_run_id, + capture_path_id=capture_path_id, + host=host, + root=root, + projection_count=projection_count, + commanded_projection_count=commanded_projection_count, + dropped_frame_count=dropped_frame_count, + reason=reason, + file_modified_at=file_modified_at, + run_ended_at=run_ended_at, + occurred_at=occurred_at, + recorded_by=recorded_by, + ): + return { + "shortfall_id": str(shortfall_id), + "producing_run_id": str(producing_run_id), + "capture_path_id": str(capture_path_id), + "host": host, + "root": root, + "projection_count": projection_count, + "commanded_projection_count": commanded_projection_count, + "dropped_frame_count": dropped_frame_count, + "reason": reason.value, + "file_modified_at": file_modified_at.isoformat(), + "run_ended_at": run_ended_at.isoformat(), + "occurred_at": occurred_at.isoformat(), + "recorded_by": str(recorded_by), + } + case _: # pragma: no cover # exhaustiveness guard + assert_never(event) + + +def from_stored(stored: StoredEvent) -> ShortfallEvent: + """Rebuild a Shortfall event from a StoredEvent loaded from the store. + + Dispatches on `stored.event_type`; raises ValueError on unknown + discriminators so a stream contaminated with foreign event types + fails loud rather than being silently dropped by the evolver. + """ + payload = stored.payload + match stored.event_type: + case "ShortfallRecorded": + + def _build_recorded() -> ShortfallRecorded: + raw_commanded = payload["commanded_projection_count"] + raw_dropped = payload["dropped_frame_count"] + return ShortfallRecorded( + shortfall_id=UUID(payload["shortfall_id"]), + producing_run_id=UUID(payload["producing_run_id"]), + capture_path_id=UUID(payload["capture_path_id"]), + host=payload["host"], + root=payload["root"], + projection_count=int(payload["projection_count"]), + commanded_projection_count=( + int(raw_commanded) if raw_commanded is not None else None + ), + dropped_frame_count=(int(raw_dropped) if raw_dropped is not None else None), + reason=ShortfallReason(payload["reason"]), + file_modified_at=datetime.fromisoformat(payload["file_modified_at"]), + run_ended_at=datetime.fromisoformat(payload["run_ended_at"]), + occurred_at=datetime.fromisoformat(payload["occurred_at"]), + recorded_by=ActorId(UUID(payload["recorded_by"])), + ) + + return deserialize_or_raise("ShortfallRecorded", _build_recorded, extra=(ValueError,)) + case _: + msg = f"Unknown ShortfallEvent event_type: {stored.event_type!r}" + raise ValueError(msg) + + +__all__ = [ + "ShortfallEvent", + "ShortfallRecorded", + "event_type_name", + "from_stored", + "to_payload", +] diff --git a/apps/api/src/cora/data/aggregates/shortfall/evolver.py b/apps/api/src/cora/data/aggregates/shortfall/evolver.py new file mode 100644 index 00000000000..e2884f88946 --- /dev/null +++ b/apps/api/src/cora/data/aggregates/shortfall/evolver.py @@ -0,0 +1,69 @@ +"""Evolver: replay events to reconstruct Shortfall state. + +The Shortfall aggregate is terminal at genesis: `ShortfallRecorded` is +the only event type, and the only state-producing arm. A capture that +is later ingested by hand is NOT a correction to this fact and emits no +event here; both facts stand (see `Shortfall`'s module docstring), so +there is no transition arm and no retraction arm. + +Same single-arm shape as the Acquisition and Decision evolvers. The +terminal `assert_never` forces pyright (and the runtime) to error if a +new event type is ever added to `ShortfallEvent` without a matching arm +here. + +The genesis arm ignores prior state: a duplicate genesis on the same +stream is prevented at append time by `expected_version=0`, not here. +""" + +from collections.abc import Sequence +from typing import assert_never + +from cora.data.aggregates.shortfall.events import ShortfallEvent, ShortfallRecorded +from cora.data.aggregates.shortfall.state import Shortfall, ShortfallStatus + + +def evolve(state: Shortfall | None, event: ShortfallEvent) -> Shortfall: + """Apply one event to the current state.""" + match event: + case ShortfallRecorded( + shortfall_id=shortfall_id, + producing_run_id=producing_run_id, + capture_path_id=capture_path_id, + host=host, + root=root, + projection_count=projection_count, + commanded_projection_count=commanded_projection_count, + dropped_frame_count=dropped_frame_count, + reason=reason, + file_modified_at=file_modified_at, + run_ended_at=run_ended_at, + occurred_at=occurred_at, + recorded_by=recorded_by, + ): + _ = state # ShortfallRecorded is the genesis event; prior state ignored. + return Shortfall( + id=shortfall_id, + producing_run_id=producing_run_id, + capture_path_id=capture_path_id, + host=host, + root=root, + projection_count=projection_count, + commanded_projection_count=commanded_projection_count, + dropped_frame_count=dropped_frame_count, + reason=reason, + file_modified_at=file_modified_at, + run_ended_at=run_ended_at, + recorded_at=occurred_at, + recorded_by=recorded_by, + status=ShortfallStatus.RECORDED, + ) + case _: # pragma: no cover # exhaustiveness guard + assert_never(event) + + +def fold(events: Sequence[ShortfallEvent]) -> Shortfall | None: + """Replay a stream of events from the empty initial state.""" + state: Shortfall | None = None + for event in events: + state = evolve(state, event) + return state diff --git a/apps/api/src/cora/data/aggregates/shortfall/read.py b/apps/api/src/cora/data/aggregates/shortfall/read.py new file mode 100644 index 00000000000..996c4a0ed65 --- /dev/null +++ b/apps/api/src/cora/data/aggregates/shortfall/read.py @@ -0,0 +1,31 @@ +"""Read repository for the Shortfall aggregate. + +`load_shortfall(event_store, shortfall_id) -> Shortfall | None` mirrors +`load_acquisition` / `load_attestation` / `load_dataset` / etc. The +aggregate is terminal at genesis, so a load returns either the +single-event genesis state or None. + +Unlike `Acquisition`, the stream id is NOT a freshly minted UUIDv7: it +is derived from `capture_path_id` via `shortfall_stream_id`, so a +caller holding a vault row's surrogate key can address the stream +without a lookup. See `_stream_id.py` for why the derivation is +load-bearing rather than a convenience. List / filter across Shortfalls +runs against `proj_data_shortfall_summary`, not this single-aggregate +read. +""" + +from uuid import UUID + +from cora.data.aggregates.shortfall.events import from_stored +from cora.data.aggregates.shortfall.evolver import fold +from cora.data.aggregates.shortfall.state import Shortfall +from cora.infrastructure.ports import EventStore + +_STREAM_TYPE = "Shortfall" + + +async def load_shortfall(event_store: EventStore, shortfall_id: UUID) -> Shortfall | None: + """Load and fold a Shortfall's event stream into current state.""" + stored, _version = await event_store.load(_STREAM_TYPE, shortfall_id) + events = [from_stored(s) for s in stored] + return fold(events) diff --git a/apps/api/src/cora/data/aggregates/shortfall/state.py b/apps/api/src/cora/data/aggregates/shortfall/state.py new file mode 100644 index 00000000000..138fb55eacc --- /dev/null +++ b/apps/api/src/cora/data/aggregates/shortfall/state.py @@ -0,0 +1,143 @@ +"""State, status enum, reason enum and errors for the Shortfall aggregate. + +A Shortfall is the recorded fact that a capture PRODUCED SOMETHING THAT +CAN NEVER BECOME A DATASET. It is not a re-judgement of the act: the +Run's own terminal stands untouched, faithfully transcribing what the +substrate reported. This records a fact about the PRODUCT. + +## Why this exists + +Before this aggregate, CORA's record could not distinguish "no scan +ran" from "a scan ran and produced nothing". Both looked identical: a +terminated Run with no Dataset naming it. Three real 2-BM Runs hold +HDF5 files carrying 1 projection of a commanded 1541 and no +`/exchange/theta`, and the sweep retried them 387 times in one recent +log window, at a measured 6.3s per attempt, writing nothing each time. +The reader had already computed the frame counts and discarded them at +the refusal, so CORA threw away its only non-substrate evidence +precisely when that evidence DISAGREED with the substrate. See +tomoscan#181 for why "Scan complete" is reported on every exit path, +and therefore why a `Completed` Run is a faithful transcript and a +false record at the same time. + +## Terminal at genesis + +Same fact-chain shape as `Acquisition` and `Attestation`: one stream per +Shortfall, exactly one `ShortfallRecorded` event ever, no lifecycle and +no transition arm. A capture that is later ingested by hand does NOT +retract this: both facts stand, because both are true. The file really +did lack its rotation angles when CORA looked, and a human really did +supply what was missing afterwards. + +## Finality is what makes the claim safe to make + +"Can never become a Dataset" is unfalsifiable unless the record says +how it was reached, so the state and the event carry BOTH sides of the +judgement (`file_modified_at`, `run_ended_at`) rather than only the +verdict. A reader can re-derive the verdict from the payload without +going back to the file, which matters because the file sits on a +detector control PC that may not exist in five years, and because this +row can never be erased. + +## Personal data + +`capture_path_id` is the surrogate key of the `run_capture_path` vault +row, NOT the path. `host` and `root` are the facility-level storage +tier (`tomdet`, `/local1/2BM`), which is provenance a reader benefits +from and which `capture_path_locator` already argues is safe to carry +in the clear. Everything strictly between the root and the filename, +which at 2-BM embeds `{UserLastName}-{ProposalNumber}`, never reaches +this aggregate at all. + +Both land as `drop:text` in the generated redaction table, because they +are bare `str` and the generator cannot tell a storage tier from free +text. That is not a defect to work around: the same already holds for +`AttestationRecorded.kind` / `.outcome`, and dropping a facility's +internal hostname from the PUBLISHED record is the right default even +though carrying it on the internal event is safe. Anyone tempted to add +an override should have a reason to publish them, not just a wish for +symmetry. +""" + +from dataclasses import dataclass +from datetime import datetime +from enum import StrEnum +from uuid import UUID + +from cora.shared.identity import ActorId + + +class ShortfallStatus(StrEnum): + """The Shortfall's lifecycle state. + + Single-valued and terminal at genesis, mirroring + `AcquisitionStatus`. Not carried in the event payload: the event + type encodes it, same precedent as the rest of the codebase. + """ + + RECORDED = "Recorded" + + +class ShortfallReason(StrEnum): + """Why this capture can never become a Dataset. + + A CLOSED enum, deliberately: the free-text refusal messages + `InvalidScanFileError` carries embed the opened path verbatim (they + are built from `os.stat` / h5py error strings), and this row is + unerasable. A closed enum cannot carry a path by construction, + which is the whole reason the payload names a member here instead + of a rendered message. + + One member today, and the space is deliberately unfilled rather + than overlooked. `ScanFileInvalidReason` marks several of its nine + members permanent, but only this one has ever been OBSERVED, and + only this one can supply the frame counts that make the record + worth writing: an `Unrecognized` file yields no `Description` at + all, so a Shortfall for it would carry a verdict and no evidence. + Adding that member is a modelling decision about what a countless + Shortfall means, not a mechanical widening, so it waits for a real + instance to reason from. + """ + + STRUCTURALLY_INCOMPLETE = "StructurallyIncomplete" + + +@dataclass(frozen=True) +class Shortfall: + """Aggregate root: one capture that can never become a Dataset. + + `capture_path_id` identifies the observation this verdict is about, + and is what the stream is keyed on, so a Run observed under two + storage locations can carry a separate verdict per location. That + matches `ScanIngestCandidateLookup`'s own `exclude` key, which is + scoped the same way and for the same reason. + + `projection_count` is what the file actually holds; + `commanded_projection_count` is what the scan was told to collect, + and is `None` when the file does not record it. The pair is the + substance of the fact: 1 of a commanded 1541 is the shape that + motivated this aggregate. `dropped_frame_count` is the detector's + own count of frames it discarded, again `None` when absent, and is + a SEPARATE fact from the shortfall between the other two: a scan + can fall short with zero drops (it stopped early) or drop frames + and still meet its total. + + `file_modified_at` and `run_ended_at` are the two sides of the + finality judgement, kept so the verdict stays checkable. See the + module docstring. + """ + + id: UUID + producing_run_id: UUID + capture_path_id: UUID + host: str + root: str + projection_count: int + commanded_projection_count: int | None + dropped_frame_count: int | None + reason: ShortfallReason + file_modified_at: datetime + run_ended_at: datetime + recorded_at: datetime + recorded_by: ActorId + status: ShortfallStatus = ShortfallStatus.RECORDED diff --git a/apps/api/src/cora/data/features/ingest_scan/handler.py b/apps/api/src/cora/data/features/ingest_scan/handler.py index 50e599c600c..15da3c01e73 100644 --- a/apps/api/src/cora/data/features/ingest_scan/handler.py +++ b/apps/api/src/cora/data/features/ingest_scan/handler.py @@ -21,7 +21,11 @@ (unreadable, unrecognized, structurally incomplete), the timestamp policy, the digest pass, the changed-under-read guard, the natural-key duplicate check, and the cross-aggregate pre-loads. A refusal at any -point leaves zero events. Decider rejections (Capturing gate, future +point leaves zero events ON THE DATASET CHAIN, which is the guarantee +this slice exists to make: never a Dataset without its Distribution and +Acquisition. Exactly one refusal additionally writes a fact of its own +first, on a SEPARATE stream that is complete in itself and is no part +of that chain: see "The second outcome" below. Decider rejections (Capturing gate, future captured_at, non-Storage supply, and now evidence shape -- `AcquisitionEvidence` validation moved from a pre-decider check here to `record_acquisition.decide`'s `validate_evidence` call, reached last @@ -33,6 +37,26 @@ all-or-nothing guarantee and the resulting HTTP 400 either way are unaffected. +## The second outcome + +A structurally incomplete file whose Run has already ended is not a +file to retry: it is a finished capture that produced nothing +ingestable, and re-reading it costs a 6.3s round trip to the detector +host to learn the same thing again. When the evidence supports saying +so, the refusal records a `Shortfall` (its own terminal-at-genesis +stream, keyed on the observation) and then raises exactly as before. + +The 400 is unchanged deliberately. The caller asked for a Dataset and +did not get one, so the refusal is still the truthful answer to their +question; the Shortfall is CORA writing down what it learned while +answering, which is a different act. Nothing about the response shape, +the route or the MCP tool moves, and `CaptureScanIngestor` needs no new +grant: it already catches this exception, and the candidate simply +stops being selected once the fact lands. + +See `_record_shortfall_if_final` for the three preconditions and for +why the finality rule is an invariant rather than a retry threshold. + ## The timestamp policy A parseable (timezone-aware) file timestamp always wins; supplying @@ -54,19 +78,35 @@ that forced the distinction. """ +from datetime import UTC, datetime from pathlib import Path from typing import Any, Protocol from urllib.parse import unquote, urlparse from uuid import UUID from cora.data._ingest import StreamPlan, decide_ingest -from cora.data.adapters.capture_path_locator import CapturePathLookup, resolve_capture_path_locator +from cora.data.adapters.capture_path_locator import ( + CapturePathLookup, + CapturePathReference, + resolve_capture_path_locator, +) from cora.data.aggregates.acquisition import AcquisitionAssetNotFoundError from cora.data.aggregates.dataset import ( DatasetAlreadyIngestedError, ProducingRunNotFoundError, ) from cora.data.aggregates.distribution import DistributionSupplyNotFoundError +from cora.data.aggregates.shortfall import ( + ShortfallReason, + ShortfallRecorded, + shortfall_stream_id, +) +from cora.data.aggregates.shortfall import ( + event_type_name as shortfall_event_type_name, +) +from cora.data.aggregates.shortfall import ( + to_payload as shortfall_to_payload, +) from cora.data.errors import InvalidScanFileError, ScanFileInvalidReason, UnauthorizedError from cora.data.features.ingest_scan.command import IngestScan from cora.data.ports.checksum_computer import ChecksumComputer @@ -76,15 +116,16 @@ from cora.infrastructure.kernel import Kernel from cora.infrastructure.logging import get_logger from cora.infrastructure.ports import Deny -from cora.infrastructure.ports.event_store import StreamAppend +from cora.infrastructure.ports.event_store import ConcurrencyError, StreamAppend from cora.infrastructure.routing import NIL_SENTINEL_ID -from cora.run.aggregates.run import load_run +from cora.run.aggregates.run import load_run, load_run_ended_at from cora.shared.identity import ActorId _COMMAND_NAME = "IngestScan" _DATASET_STREAM = "Dataset" _DISTRIBUTION_STREAM = "Distribution" _ACQUISITION_STREAM = "Acquisition" +_SHORTFALL_STREAM = "Shortfall" _log = get_logger(__name__) @@ -203,15 +244,16 @@ async def handler( # POST route and MCP tool are unaffected. `command.locator` # itself is left UNCHANGED below: it is what gets recorded on # the Dataset/Distribution events, indirect form intact. - resolved_locator = await resolve_capture_path_locator( + resolved = await resolve_capture_path_locator( command.locator, capture_path_store=capture_path_store ) - if resolved_locator is None: + if resolved is None: raise InvalidScanFileError( "scan file locator could not be resolved: the referenced " "run's capture path is missing or no longer matches.", reason=ScanFileInvalidReason.LOCATOR_UNRESOLVED, ) + resolved_locator = resolved.uri # Whether resolution actually substituted a DIFFERENT string: # `described.reason` / `computed.error_detail` below come from # `os.stat`/h5py error text and embed whatever path the reader @@ -221,7 +263,10 @@ async def handler( # caller does not necessarily have any prior right to the # REAL path it resolved to (that is the whole point of the # indirection), so that detail must never reach the response. - locator_was_resolved = resolved_locator != command.locator + # `reference` is present exactly when the scheme was indirect, + # which is the same condition, stated directly rather than + # inferred from the two strings differing. + locator_was_resolved = resolved.reference is not None # 1. Read the file's facts. Any non-Description is a refusal # with the reader's reason; an incomplete file is refused too, @@ -241,6 +286,14 @@ async def handler( reason=ScanFileInvalidReason.UNRECOGNIZED, ) if not described.structurally_complete: + await _record_shortfall_if_final( + deps, + described=described, + reference=resolved.reference, + correlation_id=correlation_id, + causation_id=causation_id, + principal_id=principal_id, + ) raise InvalidScanFileError( "scan file is structurally incomplete: the rotation-angle " "dataset is absent, meaning post-processing has not " @@ -392,6 +445,157 @@ def envelopes(plan: StreamPlan) -> list[Any]: return handler +def _instant_of(mtime_ns: int) -> datetime: + """A filesystem mtime in integer nanoseconds as a UTC datetime. + + Integer arithmetic throughout: `mtime_ns / 1e9` loses precision + well before the nanosecond, because a float64 carries about 16 + significant digits and a modern epoch-nanosecond value needs 19. + Truncating to `datetime`'s microsecond resolution is the only loss, + and it is the reason `_record_shortfall_if_final`'s comparison + documents its own granularity. + """ + seconds, nanos = divmod(mtime_ns, 1_000_000_000) + return datetime.fromtimestamp(seconds, tz=UTC).replace(microsecond=nanos // 1000) + + +async def _record_shortfall_if_final( + deps: Kernel, + *, + described: Description, + reference: CapturePathReference | None, + correlation_id: UUID, + causation_id: UUID | None, + principal_id: UUID, +) -> None: + """Record that this capture can never become a Dataset, when the + evidence supports saying so. Otherwise do nothing at all. + + Called from the structurally-incomplete refusal, BEFORE it raises. + The refusal still raises either way: the caller asked for a Dataset + and did not get one, so a 400 remains the truthful answer to their + question. This writes down what CORA learned while answering it, + which is a different thing from answering it. A failed login + returning 401 and writing an audit row is the same shape. + + ## Three preconditions, none of them about who is calling + + A `reference` is required: the stream is keyed on + `capture_path_id`, so a caller-supplied `file://` path has no key + to be recorded under. A terminal Run is required: `ended_at` is one + half of the finality test. And the file must be OLDER than that + terminal. + + Together these mean the automated sweep always qualifies (its + candidates are terminal Runs with vault rows) and a hand-typed path + never does, without either being named. A human who POSTs an + indirect locator gets identical treatment to the sweep, which is + the point: the rule is about what CORA can prove, not about whom it + is talking to. + + ## Why "older than the Run's terminal" is the whole finality rule + + Measured, not assumed: across 329 scan files still resident on the + 2-BM detector host, every single one stopped changing 7.0 to 8.8 + seconds BEFORE its Run's terminal event was recorded, zero + exceptions, with host clock skew bounded under one second. So a + file that is still older than its Run's terminal is one nothing + will write to again, and the absent rotation-angle dataset it has + now is the one it will have forever. + + The comparison is at microsecond resolution (`datetime`'s), against + a measured margin of seconds. No stability window and no attempt + counter: both would need a fresh mtime read, and `ScanReader` + exposes only `describe()`, so reading one costs the same 6.3s round + trip this whole change exists to stop paying. + + The two sides are independent by construction, which is what makes + the check worth running: `file_modified_at` comes from the + filesystem via the reader, `run_ended_at` from CORA's own event + stream. Neither is derived from the other, and neither is derived + from `deps.clock`. + """ + if reference is None: + return + + run_ended_at = await load_run_ended_at(deps.event_store, reference.run_id) + if run_ended_at is None: + return + if run_ended_at.tzinfo is None: + # Every Run terminal is written tz-aware; a naive one means the + # stream is malformed, not that the file is final. Refusing to + # judge is the fail-closed answer, but a silent refusal here + # would be indistinguishable from an open Run, so it is logged. + _log.warning( + "ingest_scan.shortfall_run_ended_at_naive", + run_id=str(reference.run_id), + ) + return + + file_modified_at = _instant_of(described.mtime_ns) + if file_modified_at >= run_ended_at: + return + + shortfall_id = shortfall_stream_id(reference.capture_path_id) + event = ShortfallRecorded( + shortfall_id=shortfall_id, + producing_run_id=reference.run_id, + capture_path_id=reference.capture_path_id, + host=reference.host, + root=reference.root, + projection_count=described.projection_count, + commanded_projection_count=described.commanded_projection_count, + dropped_frame_count=described.dropped_frame_count, + reason=ShortfallReason.STRUCTURALLY_INCOMPLETE, + file_modified_at=file_modified_at, + run_ended_at=run_ended_at, + occurred_at=deps.clock.now(), + recorded_by=ActorId(principal_id), + ) + try: + await deps.event_store.append( + _SHORTFALL_STREAM, + shortfall_id, + 0, + [ + to_new_event( + event_type=shortfall_event_type_name(event), + payload=shortfall_to_payload(event), + occurred_at=event.occurred_at, + event_id=deps.id_generator.new_id(), + command_name=_COMMAND_NAME, + correlation_id=correlation_id, + causation_id=causation_id, + principal_id=principal_id, + ) + ], + ) + except ConcurrencyError: + # Expected, not exceptional. `_CANDIDATE_SQL` drops a candidate + # once its Shortfall lands, but that read goes through a + # projection: until the projection catches up, the sweep + # re-selects this same candidate and arrives back here. The + # derived stream id is what turns that into a no-op instead of + # a duplicate fact, and this is where the no-op is absorbed. + # Surfacing it would convert a routine lag window into an error + # the operator cannot act on. + _log.info( + "ingest_scan.shortfall_already_recorded", + run_id=str(reference.run_id), + shortfall_id=str(shortfall_id), + ) + return + + _log.warning( + "ingest_scan.shortfall_recorded", + run_id=str(reference.run_id), + shortfall_id=str(shortfall_id), + reason=ShortfallReason.STRUCTURALLY_INCOMPLETE.value, + projection_count=described.projection_count, + commanded_projection_count=described.commanded_projection_count, + ) + + def _redacted_suffix(detail: str, *, redact: bool) -> str: """`": {detail}"` normally, or a fixed placeholder with no `detail` at all when `redact` is True. diff --git a/apps/api/src/cora/data/projections/__init__.py b/apps/api/src/cora/data/projections/__init__.py index 7f3e83ee69c..5dcefaebf23 100644 --- a/apps/api/src/cora/data/projections/__init__.py +++ b/apps/api/src/cora/data/projections/__init__.py @@ -1,6 +1,6 @@ """Data BC projections. -Five projection writers: +Six projection writers: - DatasetSummaryProjection: folds Dataset lifecycle events into proj_data_dataset_summary. - AcquisitionSummaryProjection: folds AcquisitionRecorded into @@ -12,12 +12,15 @@ proj_data_edition_summary. - AttestationSummaryProjection: folds AttestationRecorded into proj_data_attestation_summary. + - ShortfallSummaryProjection: folds ShortfallRecorded into + proj_data_shortfall_summary. """ from cora.data.projections.acquisition_summary import AcquisitionSummaryProjection from cora.data.projections.attestation_summary import AttestationSummaryProjection from cora.data.projections.distribution_summary import DistributionSummaryProjection from cora.data.projections.edition_summary import EditionSummaryProjection +from cora.data.projections.shortfall_summary import ShortfallSummaryProjection from cora.data.projections.summary import DatasetSummaryProjection __all__ = [ @@ -26,4 +29,5 @@ "DatasetSummaryProjection", "DistributionSummaryProjection", "EditionSummaryProjection", + "ShortfallSummaryProjection", ] diff --git a/apps/api/src/cora/data/projections/shortfall_summary.py b/apps/api/src/cora/data/projections/shortfall_summary.py new file mode 100644 index 00000000000..35373c8109d --- /dev/null +++ b/apps/api/src/cora/data/projections/shortfall_summary.py @@ -0,0 +1,76 @@ +"""ShortfallSummaryProjection: folds the Shortfall aggregate's single +ShortfallRecorded event into the `proj_data_shortfall_summary` read +model. + +Subscribed events: + - ShortfallRecorded -> INSERT (status='Recorded') + +The Shortfall is terminal at genesis (one event ever per stream), so +this projection only ever inserts. INSERT ... ON CONFLICT +(shortfall_id) DO NOTHING keeps replay idempotent. + +Dual-time columns: `file_modified_at` / `run_ended_at` are the +finality evidence pair the reason judgement was made from; `recorded_at` +is the event's `occurred_at` payload key (CORA-side wall-clock). +`commanded_projection_count` and `dropped_frame_count` are nullable +because the FILE may not record them, not because the read fell short: +`Description` carries each as None when the corresponding dataset is +absent from the layout. A null commanded count is what makes the +shortfall unquantifiable rather than absent, so it must stay +distinguishable from zero. +""" + +# pyright: reportUnknownMemberType=false, reportUnknownVariableType=false, reportUnknownArgumentType=false + +from datetime import datetime +from uuid import UUID + +from cora.infrastructure.ports.event_store import StoredEvent +from cora.infrastructure.projection.handler import ConnectionLike + +_INSERT_SHORTFALL_SQL = """ +INSERT INTO proj_data_shortfall_summary + (shortfall_id, producing_run_id, capture_path_id, host, root, + projection_count, commanded_projection_count, dropped_frame_count, + reason, file_modified_at, run_ended_at, recorded_at, recorded_by, + status) +VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, 'Recorded') +ON CONFLICT (shortfall_id) DO NOTHING +""" + + +class ShortfallSummaryProjection: + """Maintains the `proj_data_shortfall_summary` read model.""" + + name = "proj_data_shortfall_summary" + subscribed_event_types = frozenset({"ShortfallRecorded"}) + + async def apply( + self, + event: StoredEvent, + conn: ConnectionLike, + ) -> None: + match event.event_type: + case "ShortfallRecorded": + payload = event.payload + await conn.execute( + _INSERT_SHORTFALL_SQL, + UUID(payload["shortfall_id"]), + UUID(payload["producing_run_id"]), + UUID(payload["capture_path_id"]), + payload["host"], + payload["root"], + payload["projection_count"], + payload["commanded_projection_count"], + payload["dropped_frame_count"], + payload["reason"], + datetime.fromisoformat(payload["file_modified_at"]), + datetime.fromisoformat(payload["run_ended_at"]), + datetime.fromisoformat(payload["occurred_at"]), + UUID(payload["recorded_by"]), + ) + case _: + pass + + +__all__ = ["ShortfallSummaryProjection"] diff --git a/apps/api/src/cora/infrastructure/record_export/_dispositions.py b/apps/api/src/cora/infrastructure/record_export/_dispositions.py index b4dbef6a374..b40cb2b1787 100644 --- a/apps/api/src/cora/infrastructure/record_export/_dispositions.py +++ b/apps/api/src/cora/infrastructure/record_export/_dispositions.py @@ -1765,6 +1765,21 @@ "reason": "drop:text", "started_by": "token:uuid", }, + "ShortfallRecorded": { + "capture_path_id": "token:uuid", + "commanded_projection_count": "keep:number", + "dropped_frame_count": "keep:number", + "file_modified_at": "keep:time", + "host": "drop:text", + "occurred_at": "keep:time", + "producing_run_id": "token:uuid", + "projection_count": "keep:number", + "reason": "keep:enum:ShortfallReason", + "recorded_by": "token:uuid", + "root": "drop:text", + "run_ended_at": "keep:time", + "shortfall_id": "token:uuid", + }, "SteeringDesignRecorded": { "brain": { "handoff_brain": { diff --git a/apps/api/src/cora/infrastructure/record_export/_stream_types.py b/apps/api/src/cora/infrastructure/record_export/_stream_types.py index 677b5682cf5..dd96f58a4b6 100644 --- a/apps/api/src/cora/infrastructure/record_export/_stream_types.py +++ b/apps/api/src/cora/infrastructure/record_export/_stream_types.py @@ -56,6 +56,7 @@ "Role", "Run", "Seal", + "Shortfall", "Subject", "Supply", "Surface", diff --git a/apps/api/src/cora/infrastructure/schema_version.py b/apps/api/src/cora/infrastructure/schema_version.py index 0ab78af5415..62c3f357887 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 = "20260904120000" +EXPECTED_SCHEMA_VERSION: Final = "20260910222120" """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 4cf0ae41907..c149707b029 100644 --- a/apps/api/src/cora/run/aggregates/run/__init__.py +++ b/apps/api/src/cora/run/aggregates/run/__init__.py @@ -80,7 +80,7 @@ validate_adjusted_parameters_against_method_schema, validate_effective_parameters_against_method_schema, ) -from cora.run.aggregates.run.read import load_run +from cora.run.aggregates.run.read import load_run, load_run_ended_at from cora.run.aggregates.run.safety_envelope import ( beam_gate_refusal, check_safety_envelope, @@ -302,6 +302,7 @@ "is_last_active_claim", "load_run", "load_run_capture_path", + "load_run_ended_at", "load_run_experiment_identity", "supply_gate_check", "to_payload", diff --git a/apps/api/src/cora/run/aggregates/run/read.py b/apps/api/src/cora/run/aggregates/run/read.py index 1c7e15c8691..3607229deab 100644 --- a/apps/api/src/cora/run/aggregates/run/read.py +++ b/apps/api/src/cora/run/aggregates/run/read.py @@ -4,20 +4,62 @@ `load_practice` / `load_method` / `load_family` / `load_actor` / `load_subject` / `load_asset`. Used by the `get_run` query slice (6f-1) and any future update-style commands (6f-2+). + +`load_run_ended_at` answers a question the fold cannot: WHEN the Run +reached its terminal. `Run` state carries the terminal STATUS but no +terminal timestamp, so the fact survives only on the event envelope +and `load_run` discards it. See that function for why the answer is +not taken from `proj_run_summary` instead. """ +from datetime import datetime from uuid import UUID from cora.infrastructure.ports import EventStore -from cora.run.aggregates.run.events import from_stored +from cora.run.aggregates.run.events import ( + RunAborted, + RunCompleted, + RunStopped, + RunTruncated, + from_stored, +) from cora.run.aggregates.run.evolver import fold from cora.run.aggregates.run.state import Run _STREAM_TYPE = "Run" +# The four reachable terminals (see `RunStatus`). Matched by TYPE rather +# than by taking the last event on the stream: "last event" would +# silently start answering a different question the day any event is +# appended after a terminal. +_TERMINAL_EVENTS = (RunCompleted, RunAborted, RunStopped, RunTruncated) + async def load_run(event_store: EventStore, run_id: UUID) -> Run | None: """Load and fold a Run's event stream into current state.""" stored, _version = await event_store.load(_STREAM_TYPE, run_id) events = [from_stored(s) for s in stored] return fold(events) + + +async def load_run_ended_at(event_store: EventStore, run_id: UUID) -> datetime | None: + """`occurred_at` of the Run's terminal event, or None when the Run + is still open, absent, or holds no terminal event. + + Deliberately NOT read from `proj_run_summary.updated_at`: that + column is `now()` at PROJECTION-WRITE time, which is a different + instant from the terminal's `occurred_at`, is mutable, and is + re-derived on every replay. A caller comparing it against a fact + from outside CORA would be comparing against CORA's own bookkeeping + rather than against the record. + + Walks from the END of the stream so a terminal Run deserializes one + event rather than the whole history; an open Run pays a full walk, + which is the case that returns None and does no further work. + """ + stored, _version = await event_store.load(_STREAM_TYPE, run_id) + for raw in reversed(stored): + event = from_stored(raw) + if isinstance(event, _TERMINAL_EVENTS): + return event.occurred_at + return None diff --git a/apps/api/tests/architecture/test_command_name_derives_event_name.py b/apps/api/tests/architecture/test_command_name_derives_event_name.py index dd064a6aa3a..9887dad0e02 100644 --- a/apps/api/tests/architecture/test_command_name_derives_event_name.py +++ b/apps/api/tests/architecture/test_command_name_derives_event_name.py @@ -155,6 +155,19 @@ # states the act of recording it. Symmetric with StartRun -> RunStarted, # the driven genesis's own sanctioned pair. "run/record_witnessed_run": "witnessed genesis; states the fact, not the recording verb", + # A discovery artifact rather than a naming judgement. `ingest_scan` + # is a composition slice: its three genesis events reach the store + # as `(event_type, payload, occurred_at)` plans built by + # `decide_ingest`, so this file's scan finds no event CLASS + # constructed in the handler and the slice was invisible here. The + # handler now constructs exactly one event class directly, its + # second outcome, and that lone construction makes the slice look + # single-event to the scan. `ShortfallRecorded` is not the event + # `IngestScan` is named after and could not be: the command names + # the act attempted, this event names what was found instead. + # Renaming either side would be renaming to satisfy a scan that has + # mis-classified the slice. + "data/ingest_scan": "composition slice; discovered event is its second outcome", } _KNOWN_DRIFT: dict[str, str] = { diff --git a/apps/api/tests/architecture/test_event_class_defined_vs_registered.py b/apps/api/tests/architecture/test_event_class_defined_vs_registered.py index 837e9c01a98..1840da6b763 100644 --- a/apps/api/tests/architecture/test_event_class_defined_vs_registered.py +++ b/apps/api/tests/architecture/test_event_class_defined_vs_registered.py @@ -132,6 +132,16 @@ "state it produces and erase the award semantic the whole " "allocation arc is built on." ), + ("data", "shortfall"): ( + "ShortfallRecorded is a terminal-at-genesis recorded-fact-chain " + "(one stream per capture_path_id; a single ShortfallRecorded " + "event). The fact is that an observed capture can never become a " + "Dataset, stated once as of the producing Run's terminal; the " + "verb records that finality judgement rather than registering a " + "long-lived instance. Renaming to ShortfallRegistered would " + "misframe a fact as an entity. Mirrors the acquisition / " + "attestation recorded-fact-chain precedent in this same BC." + ), } diff --git a/apps/api/tests/architecture/test_fold_symmetry.py b/apps/api/tests/architecture/test_fold_symmetry.py index cea006d1425..8ad7692a3f8 100644 --- a/apps/api/tests/architecture/test_fold_symmetry.py +++ b/apps/api/tests/architecture/test_fold_symmetry.py @@ -111,6 +111,21 @@ class name (e.g. SubjectRegistered -> registered_by). "CORA-side recording act folds as the proper `recorded_at` / " "`recorded_by` pair on the same dataclass" ), + "data.Shortfall.file_modified_at": ( + "filesystem mtime, same shape as data.Acquisition.captured_at: the " + "entity that last wrote the file is the external tool's file-writer, " + "not a CORA Actor, so there is no `file_modified_by` fact-act partner. " + "The CORA-side recording act folds as the proper `recorded_at` / " + "`recorded_by` pair on the same dataclass" + ), + "data.Shortfall.run_ended_at": ( + "cited from ANOTHER aggregate's stream, not an act on this one: it is " + "the producing Run's terminal `occurred_at`, and the actor who ended " + "that Run folds onto the Run's own terminal event. Carried here so the " + "finality judgement stays checkable from this payload alone; a " + "`run_ended_by` here would duplicate an attribution that already has " + "an owner and could drift from it" + ), "data.Edition.Creator.actor_id": ( "publication-author identity-ref (credited creator on the citable Edition); " "ordered tuple semantics, NOT a fact-act fold" diff --git a/apps/api/tests/integration/test_capture_scan_ingestor_postgres.py b/apps/api/tests/integration/test_capture_scan_ingestor_postgres.py index 3872221c97e..07bdd56ffb1 100644 --- a/apps/api/tests/integration/test_capture_scan_ingestor_postgres.py +++ b/apps/api/tests/integration/test_capture_scan_ingestor_postgres.py @@ -74,6 +74,32 @@ async def _insert_capture_path( ) +async def _insert_shortfall( + pool: asyncpg.Pool, *, producing_run_id: UUID, capture_path_id: UUID +) -> None: + """Record a Shortfall against ONE capture observation. + + Written straight to the projection for the same reason every other + helper here is: what these tests exercise is the raw join, not the + fold that would normally populate it. + """ + await pool.execute( + """ + INSERT INTO proj_data_shortfall_summary + (shortfall_id, producing_run_id, capture_path_id, host, root, + projection_count, commanded_projection_count, dropped_frame_count, + reason, file_modified_at, run_ended_at, recorded_at, recorded_by, status) + VALUES ($1, $2, $3, 'tomdet', '/local1/2BM', 1, 1541, 0, + 'StructurallyIncomplete', $4, $4, $4, $5, 'Recorded') + """, + uuid4(), + producing_run_id, + capture_path_id, + _NOW, + uuid4(), + ) + + async def _insert_dataset(pool: asyncpg.Pool, *, producing_run_id: UUID) -> None: await pool.execute( """ @@ -313,3 +339,60 @@ async def test_a_run_whose_location_was_never_recorded_is_not_a_candidate( assert candidate is not None assert candidate.run_id == run_id assert candidate.root == "/local1/2BM" + + +@pytest.mark.integration +async def test_a_capture_path_with_a_recorded_shortfall_is_not_a_candidate( + db_pool: asyncpg.Pool, +) -> None: + """A Shortfall is a terminal verdict on the observation: the file + can never become a Dataset, so re-selecting it would buy nothing + but another 6.3s round trip to the detector host, forever. This is + the clause that turns the retry loop off.""" + run_id = uuid4() + await _insert_run_summary(db_pool, run_id=run_id, capture_code="2bmb-tomoscan") + capture_path_id = await _insert_capture_path( + db_pool, run_id=run_id, observed_path="/local1/2BM/scan.h5", created_at=_NOW + ) + await _insert_shortfall(db_pool, producing_run_id=run_id, capture_path_id=capture_path_id) + + lookup = PostgresScanIngestCandidateLookup(db_pool) + + assert await lookup.next_candidate() is None + + +@pytest.mark.integration +async def test_a_shortfall_on_one_location_still_leaves_the_other_a_candidate( + db_pool: asyncpg.Pool, +) -> None: + """The discriminating case for the exclusion's KEY. Keyed on + `capture_path_id`, a verdict about the acquisition-tier copy says + nothing about the archive-tier copy, which is a separate file that + may well be complete. Keyed on `run_id` instead, this test fails + and one bad copy would silently condemn every other copy of the + same Run: the plain single-location test above passes either way, + so it alone would not catch that.""" + run_id = uuid4() + await _insert_run_summary(db_pool, run_id=run_id, capture_code="2bmb-tomoscan") + local_capture_path_id = await _insert_capture_path( + db_pool, + run_id=run_id, + observed_path="/local1/2BM/scan.h5", + created_at=_NOW, + root="/local1/2BM", + ) + await _insert_capture_path( + db_pool, + run_id=run_id, + observed_path="/gdata/dm/2BM/scan.h5", + created_at=_NOW, + root="/gdata/dm/2BM", + ) + await _insert_shortfall(db_pool, producing_run_id=run_id, capture_path_id=local_capture_path_id) + + lookup = PostgresScanIngestCandidateLookup(db_pool) + candidate = await lookup.next_candidate() + + assert candidate is not None + assert candidate.run_id == run_id + assert candidate.root == "/gdata/dm/2BM" diff --git a/apps/api/tests/unit/api/test_durable_distribution_driver.py b/apps/api/tests/unit/api/test_durable_distribution_driver.py index f798877e553..e5df8026db4 100644 --- a/apps/api/tests/unit/api/test_durable_distribution_driver.py +++ b/apps/api/tests/unit/api/test_durable_distribution_driver.py @@ -246,7 +246,8 @@ async def test_the_registered_locator_resolves_back_to_the_recorded_path() -> No str(registrar.calls[0]["locator"]), capture_path_store=recorder.store ) - assert resolved == f"file://{_FOUND}" + assert resolved is not None + assert resolved.uri == f"file://{_FOUND}" @pytest.mark.parametrize( @@ -297,7 +298,8 @@ async def test_a_root_spelled_with_a_trailing_slash_still_round_trips() -> None: resolved = await resolve_capture_path_locator( str(registrar.calls[0]["locator"]), capture_path_store=recorder.store ) - assert resolved == f"file://{_FOUND}" + assert resolved is not None + assert resolved.uri == f"file://{_FOUND}" async def test_the_recorded_observation_time_is_the_files_own_not_the_clocks() -> None: @@ -639,7 +641,8 @@ async def test_the_locator_names_the_file_the_probe_found_not_the_one_it_searche str(registrar.calls[0]["locator"]), capture_path_store=recorder.store ) - assert resolved == f"file://{renamed}" + assert resolved is not None + assert resolved.uri == f"file://{renamed}" async def test_a_tick_stopped_by_a_dead_transport_retries_the_same_candidate() -> None: diff --git a/apps/api/tests/unit/data/test_capture_path_locator.py b/apps/api/tests/unit/data/test_capture_path_locator.py index 7f443a37992..13feb549a49 100644 --- a/apps/api/tests/unit/data/test_capture_path_locator.py +++ b/apps/api/tests/unit/data/test_capture_path_locator.py @@ -14,6 +14,7 @@ from __future__ import annotations +import dataclasses from datetime import UTC, datetime from urllib.parse import quote from uuid import UUID @@ -109,7 +110,8 @@ async def test_mint_and_resolve_round_trip_a_filename_with_spaces() -> None: resolved = await resolve_capture_path_locator(locator, capture_path_store=store) - assert resolved == "file://" + quote(observed_path) + assert resolved is not None + assert resolved.uri == "file://" + quote(observed_path) async def test_resolve_passes_through_a_non_vault_scheme_unchanged() -> None: @@ -121,7 +123,23 @@ async def test_resolve_passes_through_a_non_vault_scheme_unchanged() -> None: resolved = await resolve_capture_path_locator(real_locator, capture_path_store=store) - assert resolved == real_locator + assert resolved is not None + assert resolved.uri == real_locator + + +async def test_resolve_pass_through_locator_carries_no_reference() -> None: + """A caller-supplied `file://` URI has no vault row behind it, so + `reference` must be `None` rather than a value object built from + nothing. `ingest_scan` uses `reference is not None` to decide + whether to record a Shortfall, so a pass-through must never look + like a resolved vault row.""" + store = await _seeded_store() + real_locator = "file:///local/cora-scans/test_005.h5" + + resolved = await resolve_capture_path_locator(real_locator, capture_path_store=store) + + assert resolved is not None + assert resolved.reference is None async def test_resolve_recovers_the_real_path_as_a_file_uri() -> None: @@ -133,7 +151,52 @@ async def test_resolve_recovers_the_real_path_as_a_file_uri() -> None: resolved = await resolve_capture_path_locator(locator, capture_path_store=store) - assert resolved == "file://" + _OBSERVED_PATH + assert resolved is not None + assert resolved.uri == "file://" + _OBSERVED_PATH + + +async def test_resolve_indirect_locator_carries_a_reference_matching_the_vault_row() -> None: + """`reference` is what `ingest_scan` keys a Shortfall on without a + second store read, so its four fields must agree with the actual + vault row, not merely be present.""" + store = await _seeded_store() + locator = mint_capture_path_locator( + observed_path=_OBSERVED_PATH, run_id=_RUN_ID, host="tomdet", root="/local1/2BM" + ) + assert locator is not None + + resolved = await resolve_capture_path_locator(locator, capture_path_store=store) + + assert resolved is not None + assert resolved.reference is not None + row = await store.get(_RUN_ID, host=_HOST, root=_ROOT) + assert row is not None + assert resolved.reference.capture_path_id == row.capture_path_id + assert resolved.reference.run_id == row.run_id + assert resolved.reference.host == _HOST + assert resolved.reference.root == _ROOT + + +async def test_resolve_reference_never_carries_the_observed_path() -> None: + """`CapturePathReference` travels out to `ingest_scan`'s handler and + onto an immutable Shortfall event; none of its fields may equal or + contain the personal-data-bearing observed path, not even the + identity fields that merely LOOK like they could echo a fragment of + it.""" + store = await _seeded_store() + locator = mint_capture_path_locator( + observed_path=_OBSERVED_PATH, run_id=_RUN_ID, host="tomdet", root="/local1/2BM" + ) + assert locator is not None + + resolved = await resolve_capture_path_locator(locator, capture_path_store=store) + + assert resolved is not None + assert resolved.reference is not None + for field in dataclasses.fields(resolved.reference): + value = str(getattr(resolved.reference, field.name)) + assert value != _OBSERVED_PATH + assert _PERSONAL_PATH_FRAGMENT not in value async def test_resolve_refuses_when_the_vault_row_is_absent() -> None: @@ -236,14 +299,15 @@ async def test_two_locations_for_one_run_each_resolve_to_their_own_path() -> Non assert archive_locator is not None assert acquisition_locator != archive_locator - assert ( - await resolve_capture_path_locator(acquisition_locator, capture_path_store=store) - == "file://" + _OBSERVED_PATH - ) - assert ( - await resolve_capture_path_locator(archive_locator, capture_path_store=store) - == "file://" + archive_path + acquisition_resolved = await resolve_capture_path_locator( + acquisition_locator, capture_path_store=store ) + archive_resolved = await resolve_capture_path_locator(archive_locator, capture_path_store=store) + + assert acquisition_resolved is not None + assert acquisition_resolved.uri == "file://" + _OBSERVED_PATH + assert archive_resolved is not None + assert archive_resolved.uri == "file://" + archive_path async def test_resolve_refuses_a_location_the_run_was_never_observed_on() -> None: @@ -286,10 +350,11 @@ async def test_resolve_refuses_a_locator_naming_a_different_host() -> None: observed_path=_OBSERVED_PATH, run_id=_RUN_ID, host=_HOST, root=_ROOT ) assert right_host_locator is not None - assert ( - await resolve_capture_path_locator(right_host_locator, capture_path_store=store) - == "file://" + _OBSERVED_PATH + right_host_resolved = await resolve_capture_path_locator( + right_host_locator, capture_path_store=store ) + assert right_host_resolved is not None + assert right_host_resolved.uri == "file://" + _OBSERVED_PATH async def test_resolve_refuses_a_legacy_row_whose_location_was_never_recorded() -> None: diff --git a/apps/api/tests/unit/data/test_ingest_scan_shortfall.py b/apps/api/tests/unit/data/test_ingest_scan_shortfall.py new file mode 100644 index 00000000000..53c0f0dd23c --- /dev/null +++ b/apps/api/tests/unit/data/test_ingest_scan_shortfall.py @@ -0,0 +1,516 @@ +"""ingest_scan's second outcome: recording a capture that can never +become a Dataset. + +What these tests pin is the SHAPE of the decision, not just that a row +appears: which refusals qualify, what evidence is required before CORA +will make a permanent claim, and that the refusal itself is unchanged. + +## The three instants are deliberately distinct + +`deps.clock` (`_NOW`), the Run's terminal (`_RUN_ENDED_AT`) and the +file's mtime (`_FINAL_MTIME_AT`) are three different times, and +`test_shortfall_fixture_instants_are_independent` fails if any two are +ever collapsed. The finality check compares the file against the Run's +terminal; a fixture that let one clock supply both sides would agree by +construction and pass no matter what the production code did (see +[[project_independent_check_principle]]). + +The nanosecond conversion is derived differently here than in +production for the same reason: `_ns` walks a `timedelta` from the +epoch, while `_instant_of` uses `divmod` and `datetime.fromtimestamp`. +A rounding bug in one is not reproduced by the other. + +The 8-second gap in `_FINAL_MTIME_AT` is not arbitrary: across 329 scan +files on the 2-BM detector host, every one settled 7.0 to 8.8 seconds +before its Run's terminal was recorded. +""" + +from dataclasses import replace as dc_replace +from datetime import UTC, datetime, timedelta +from typing import Any +from uuid import UUID, uuid4 + +import pytest + +from cora.data.adapters.capture_path_locator import mint_capture_path_locator +from cora.data.aggregates.shortfall import ShortfallReason, shortfall_stream_id +from cora.data.errors import InvalidScanFileError +from cora.data.features import ingest_scan +from cora.data.features.ingest_scan import IngestScan +from cora.data.ports.checksum_computer import ComputedChecksum, ConfiguredChecksumComputer +from cora.data.ports.scan_reader import ( + ConfiguredScanReader, + Description, + ScanReadResult, + Unreadable, +) +from cora.infrastructure.adapters.in_memory_asset_lookup import InMemoryAssetLookup +from cora.infrastructure.adapters.in_memory_event_store import InMemoryEventStore +from cora.infrastructure.event_envelope import to_new_event +from cora.infrastructure.kernel import Kernel +from cora.infrastructure.ports.event_store import StoredEvent +from cora.infrastructure.ports.supply_lookup import SingleSupplyLookup, SupplyLookupResult +from cora.run.aggregates.run import ( + InMemoryCapturePathStore, + RunCompleted, + RunStarted, +) +from cora.run.aggregates.run import event_type_name as run_event_type_name +from cora.run.aggregates.run import to_payload as run_to_payload +from tests.unit._helpers import build_deps + +pytestmark = pytest.mark.unit + +_EPOCH = datetime(1970, 1, 1, tzinfo=UTC) + +_NOW = datetime(2026, 7, 29, 16, 0, 0, tzinfo=UTC) +_RUN_ENDED_AT = datetime(2026, 7, 29, 15, 0, 0, tzinfo=UTC) +_FINAL_MTIME_AT = _RUN_ENDED_AT - timedelta(seconds=8) +_STILL_WRITING_MTIME_AT = _RUN_ENDED_AT + timedelta(seconds=5) + +_HOST = "tomdet" +_ROOT = "/local1/2BM" +_OBSERVED_PATH = f"{_ROOT}/2026-08-Smith-1015116/scan_042.h5" +_DIRECT_LOCATOR = "file:///data2/2026-07/doe-12345/scan_001.h5" + +_PRINCIPAL_ID = uuid4() +_CORRELATION_ID = uuid4() +_ASSET_ID = uuid4() +_SUPPLY_ID = uuid4() +_RUN_ID = UUID("01900000-0000-7000-8000-0000000090a1") +_SHA = "b" * 64 + +_IDS = [uuid4() for _ in range(8)] + + +def _ns(moment: datetime) -> int: + """Nanoseconds since the epoch, by timedelta arithmetic. + + Deliberately a different derivation from production's + `_instant_of`; see this module's docstring. + """ + delta = moment - _EPOCH + return (delta.days * 86_400 + delta.seconds) * 1_000_000_000 + delta.microseconds * 1_000 + + +def _incomplete(**overrides: object) -> Description: + """A real 2-BM shortfall: 1 projection of a commanded 1541, and no + rotation-angle dataset, so `structurally_complete` is False.""" + base = Description( + media_type="application/x-hdf5", + structurally_complete=False, + projection_count=1, + flat_count=0, + dark_count=0, + invalid_count=0, + commanded_projection_count=1541, + commanded_flat_count=20, + commanded_dark_count=20, + dropped_frame_count=0, + projection_angles_deg=None, + flat_angles_deg=None, + dark_angles_deg=None, + captured_at=None, + captured_at_raw=None, + captured_at_source="end_date", + byte_size=4096, + mtime_ns=_ns(_FINAL_MTIME_AT), + ) + return dc_replace(base, **overrides) # type: ignore[arg-type] + + +class _NoDuplicate: + async def __call__(self, *, checksum_algorithm: str, checksum_value: str) -> UUID | None: + return None + + +def _deps(store: InMemoryEventStore) -> Kernel: + lookup = InMemoryAssetLookup() + lookup.register( + asset_id=_ASSET_ID, + name="Oryx Detector", + tier="Device", + lifecycle="Active", + family_affordances=frozenset({"Capturing"}), + ) + base = build_deps(ids=list(_IDS), now=_NOW, event_store=store, asset_lookup=lookup) + return dc_replace( + base, + supply_lookup=SingleSupplyLookup( + SupplyLookupResult( + supply_id=_SUPPLY_ID, + kind="Storage", + name="analysis tier", + status="Available", + facility_code="aps", + ) + ), + ) + + +async def _seed_run( + store: InMemoryEventStore, *, ended: bool, naive_terminal: bool = False +) -> None: + started = RunStarted( + run_id=_RUN_ID, + name="Shortfall Run", + plan_id=UUID("01900000-0000-7000-8000-000000000401"), + subject_id=None, + occurred_at=_RUN_ENDED_AT - timedelta(hours=1), + ) + await store.append( + stream_type="Run", + stream_id=_RUN_ID, + expected_version=0, + events=[ + to_new_event( + event_type=run_event_type_name(started), + payload=run_to_payload(started), + occurred_at=started.occurred_at, + event_id=uuid4(), + command_name="StartRun", + correlation_id=_CORRELATION_ID, + causation_id=None, + principal_id=_PRINCIPAL_ID, + ) + ], + ) + if not ended: + return + if naive_terminal: + # A malformed stream, built the only way it can arise: + # `to_payload` isoformats without an offset, and + # `datetime.fromisoformat` reads that back naive. + await _append_naive_terminal(store) + return + finished = RunCompleted( + run_id=_RUN_ID, + occurred_at=_RUN_ENDED_AT, + observed_at=_RUN_ENDED_AT, + ) + await store.append( + stream_type="Run", + stream_id=_RUN_ID, + expected_version=1, + events=[ + to_new_event( + event_type=run_event_type_name(finished), + payload=run_to_payload(finished), + occurred_at=_RUN_ENDED_AT, + event_id=uuid4(), + command_name="CompleteRun", + correlation_id=_CORRELATION_ID, + causation_id=None, + principal_id=_PRINCIPAL_ID, + ) + ], + ) + + +async def _append_naive_terminal(store: InMemoryEventStore) -> None: + naive = _RUN_ENDED_AT.replace(tzinfo=None) + finished = RunCompleted(run_id=_RUN_ID, occurred_at=naive, observed_at=None) + await store.append( + stream_type="Run", + stream_id=_RUN_ID, + expected_version=1, + events=[ + to_new_event( + event_type=run_event_type_name(finished), + payload=run_to_payload(finished), + occurred_at=_RUN_ENDED_AT, + event_id=uuid4(), + command_name="CompleteRun", + correlation_id=_CORRELATION_ID, + causation_id=None, + principal_id=_PRINCIPAL_ID, + ) + ], + ) + + +async def _vault(store: InMemoryCapturePathStore) -> UUID: + """Record the capture observation and hand back its surrogate key.""" + await store.upsert( + run_id=_RUN_ID, + observed_path=_OBSERVED_PATH, + observed_at=_RUN_ENDED_AT, + created_at=_RUN_ENDED_AT, + host=_HOST, + root=_ROOT, + ) + row = await store.get(_RUN_ID, host=_HOST, root=_ROOT) + assert row is not None + return row.capture_path_id + + +def _bind( + deps: Kernel, + *, + locator: str, + described: ScanReadResult, + capture_path_store: InMemoryCapturePathStore, +) -> ingest_scan.Handler: + resolved_uri = "file://" + _OBSERVED_PATH if locator != _DIRECT_LOCATOR else _DIRECT_LOCATOR + return ingest_scan.bind( + deps, + scan_reader=ConfiguredScanReader({resolved_uri: described}), + checksum_computer=ConfiguredChecksumComputer( + { + resolved_uri: ComputedChecksum( + algorithm="sha256", + value=_SHA, + byte_size=4096, + mtime_ns=_ns(_FINAL_MTIME_AT), + ) + } + ), + dataset_by_checksum_lookup=_NoDuplicate(), + capture_path_store=capture_path_store, + ) + + +def _command(locator: str) -> IngestScan: + return IngestScan( + locator=locator, + producing_asset_id=_ASSET_ID, + supply_id=_SUPPLY_ID, + access_protocol="POSIX", + producing_run_id=_RUN_ID, + ) + + +async def _shortfall_events(store: InMemoryEventStore, capture_path_id: UUID) -> list[StoredEvent]: + events, _ = await store.load("Shortfall", shortfall_stream_id(capture_path_id)) + return list(events) + + +async def _shortfall_payload(store: InMemoryEventStore, capture_path_id: UUID) -> dict[str, Any]: + events = await _shortfall_events(store, capture_path_id) + assert len(events) == 1 + return events[0].payload + + +async def _dataset_chain_counts(store: InMemoryEventStore) -> tuple[int, int, int]: + counts: list[int] = [] + for stream_type, stream_id in ( + ("Dataset", _IDS[0]), + ("Distribution", _IDS[1]), + ("Acquisition", _IDS[2]), + ): + events, _ = await store.load(stream_type, stream_id) + counts.append(len(events)) + return counts[0], counts[1], counts[2] + + +async def _run_incomplete_ingest( + store: InMemoryEventStore, + *, + locator: str, + described: ScanReadResult | None = None, + capture_path_store: InMemoryCapturePathStore, +) -> None: + """Drive one ingest that is expected to refuse, and swallow only + the refusal itself so each test asserts on the RECORD rather than + on the exception it already knows is coming.""" + handler = _bind( + _deps(store), + locator=locator, + described=described if described is not None else _incomplete(), + capture_path_store=capture_path_store, + ) + with pytest.raises(InvalidScanFileError): + await handler(_command(locator), principal_id=_PRINCIPAL_ID, correlation_id=_CORRELATION_ID) + + +def _indirect_locator() -> str: + locator = mint_capture_path_locator( + observed_path=_OBSERVED_PATH, run_id=_RUN_ID, host=_HOST, root=_ROOT + ) + assert locator is not None + return locator + + +def test_shortfall_fixture_instants_are_independent() -> None: + """Guard the fixture, not the code: the finality check compares the + file against the Run's terminal, so a fixture that ever collapses + those two (or sources either from `deps.clock`) would make every + test below pass by construction.""" + assert len({_NOW, _RUN_ENDED_AT, _FINAL_MTIME_AT, _STILL_WRITING_MTIME_AT}) == 4 + assert _FINAL_MTIME_AT < _RUN_ENDED_AT + assert _STILL_WRITING_MTIME_AT > _RUN_ENDED_AT + + +async def test_ingest_incomplete_file_from_ended_run_records_a_shortfall() -> None: + """The motivating 2-BM case: 1 projection of a commanded 1541, no + rotation angles, on a Run that finished eight seconds after the + file stopped changing.""" + store = InMemoryEventStore() + vault = InMemoryCapturePathStore() + await _seed_run(store, ended=True) + capture_path_id = await _vault(vault) + + await _run_incomplete_ingest(store, locator=_indirect_locator(), capture_path_store=vault) + + events = await _shortfall_events(store, capture_path_id) + assert len(events) == 1 + payload = events[0].payload + assert payload["projection_count"] == 1 + assert payload["commanded_projection_count"] == 1541 + assert payload["reason"] == ShortfallReason.STRUCTURALLY_INCOMPLETE.value + assert payload["producing_run_id"] == str(_RUN_ID) + assert payload["capture_path_id"] == str(capture_path_id) + assert payload["host"] == _HOST + assert payload["root"] == _ROOT + + +async def test_ingest_shortfall_carries_both_sides_of_the_finality_judgement() -> None: + """The verdict has to stay checkable from the payload alone: the + file outlives neither the detector host nor this row.""" + store = InMemoryEventStore() + vault = InMemoryCapturePathStore() + await _seed_run(store, ended=True) + capture_path_id = await _vault(vault) + + await _run_incomplete_ingest(store, locator=_indirect_locator(), capture_path_store=vault) + + payload = await _shortfall_payload(store, capture_path_id) + assert datetime.fromisoformat(payload["file_modified_at"]) == _FINAL_MTIME_AT + assert datetime.fromisoformat(payload["run_ended_at"]) == _RUN_ENDED_AT + assert datetime.fromisoformat(payload["occurred_at"]) == _NOW + + +async def test_ingest_incomplete_file_still_being_written_records_nothing() -> None: + """THE guard. A file whose mtime is NEWER than the Run's terminal + is one something is still writing to, so the absent rotation angles + may yet arrive. Condemning it would be permanent and wrong, and + this is the assertion that fails if the comparison is ever + loosened or inverted.""" + store = InMemoryEventStore() + vault = InMemoryCapturePathStore() + await _seed_run(store, ended=True) + capture_path_id = await _vault(vault) + + await _run_incomplete_ingest( + store, + locator=_indirect_locator(), + described=_incomplete(mtime_ns=_ns(_STILL_WRITING_MTIME_AT)), + capture_path_store=vault, + ) + + assert await _shortfall_events(store, capture_path_id) == [] + + +async def test_ingest_incomplete_file_from_open_run_records_nothing() -> None: + """No terminal means no finality evidence at all, so there is + nothing to base a permanent claim on.""" + store = InMemoryEventStore() + vault = InMemoryCapturePathStore() + await _seed_run(store, ended=False) + capture_path_id = await _vault(vault) + + await _run_incomplete_ingest(store, locator=_indirect_locator(), capture_path_store=vault) + + assert await _shortfall_events(store, capture_path_id) == [] + + +async def test_ingest_transient_refusal_records_nothing() -> None: + """An unreadable file is the documented TRANSIENT refusal. Keying a + Shortfall on the exception class rather than the specific cause + would permanently condemn every file caught mid-transfer, which is + the one mistake in this design that loses data silently.""" + store = InMemoryEventStore() + vault = InMemoryCapturePathStore() + await _seed_run(store, ended=True) + capture_path_id = await _vault(vault) + + await _run_incomplete_ingest( + store, + locator=_indirect_locator(), + described=Unreadable(reason="half-transferred"), + capture_path_store=vault, + ) + + assert await _shortfall_events(store, capture_path_id) == [] + + +async def test_ingest_incomplete_file_at_a_direct_path_records_nothing() -> None: + """A hand-typed `file://` path has no vault row, so there is no + observation surrogate to key the stream on. The rule is about what + CORA can prove, not about who is calling: a human POSTing an + INDIRECT locator is treated exactly like the sweep.""" + store = InMemoryEventStore() + vault = InMemoryCapturePathStore() + await _seed_run(store, ended=True) + capture_path_id = await _vault(vault) + + await _run_incomplete_ingest(store, locator=_DIRECT_LOCATOR, capture_path_store=vault) + + assert await _shortfall_events(store, capture_path_id) == [] + + +async def test_ingest_repeated_incomplete_reads_record_one_shortfall() -> None: + """The sweep re-selects a candidate until `proj_data_shortfall_summary` + catches up, so the second attempt is routine, not exceptional. It + must leave the record unchanged and still raise the ordinary + refusal rather than a conflict the operator cannot act on.""" + store = InMemoryEventStore() + vault = InMemoryCapturePathStore() + await _seed_run(store, ended=True) + capture_path_id = await _vault(vault) + locator = _indirect_locator() + + await _run_incomplete_ingest(store, locator=locator, capture_path_store=vault) + await _run_incomplete_ingest(store, locator=locator, capture_path_store=vault) + + assert len(await _shortfall_events(store, capture_path_id)) == 1 + + +async def test_ingest_shortfall_leaves_the_dataset_chain_empty() -> None: + """The all-or-nothing guarantee is about the Dataset chain, and it + is untouched: a Shortfall is a complete fact on its own stream, not + half of a Dataset.""" + store = InMemoryEventStore() + vault = InMemoryCapturePathStore() + await _seed_run(store, ended=True) + await _vault(vault) + + await _run_incomplete_ingest(store, locator=_indirect_locator(), capture_path_store=vault) + + assert await _dataset_chain_counts(store) == (0, 0, 0) + + +async def test_ingest_shortfall_payload_carries_no_part_of_the_observed_path() -> None: + """`observed_path` embeds a surname and a proposal number at 2-BM, + and this row can never be erased. The tier segments are carried + deliberately and are not personal data; the experiment folder and + the filename must appear nowhere.""" + store = InMemoryEventStore() + vault = InMemoryCapturePathStore() + await _seed_run(store, ended=True) + capture_path_id = await _vault(vault) + + await _run_incomplete_ingest(store, locator=_indirect_locator(), capture_path_store=vault) + + payload = await _shortfall_payload(store, capture_path_id) + rendered = repr(payload) + assert "Smith" not in rendered + assert "1015116" not in rendered + assert "scan_042" not in rendered + assert _OBSERVED_PATH not in rendered + + +async def test_ingest_with_a_naive_run_terminal_records_nothing() -> None: + """A terminal without an offset means the Run stream is malformed, + not that the file is final, so the judgement is refused. Failing + closed is the easy half; the branch also LOGS, because a silent + refusal here is indistinguishable from an open Run and the + difference is a corrupt stream nobody would go looking for.""" + store = InMemoryEventStore() + vault = InMemoryCapturePathStore() + await _seed_run(store, ended=True, naive_terminal=True) + capture_path_id = await _vault(vault) + + await _run_incomplete_ingest(store, locator=_indirect_locator(), capture_path_store=vault) + + assert await _shortfall_events(store, capture_path_id) == [] diff --git a/apps/api/tests/unit/data/test_shortfall_aggregate.py b/apps/api/tests/unit/data/test_shortfall_aggregate.py new file mode 100644 index 00000000000..5867b8377c3 --- /dev/null +++ b/apps/api/tests/unit/data/test_shortfall_aggregate.py @@ -0,0 +1,184 @@ +"""Shortfall aggregate: the write side and the read side agree. + +`ingest_scan` only ever APPENDS a Shortfall, so nothing in the slice's +own tests ever folds one back. That leaves `from_stored`, `evolve`, +`fold` and `load_shortfall` written but unexercised, which in an +event-sourced system is a latent bug rather than a coverage statistic: +an event that cannot be read back is an event that is not really +recorded. `test_event_union_from_stored_coverage.py` does not close +this either, since it skips single-event aggregates, which is exactly +this shape. + +The round trip asserts the WHOLE folded state against an explicitly +constructed `Shortfall`, not field by field. A per-field assertion is +blind to a field the writer forgot to carry, because the reader's +default quietly fills it in (see [[project_field_drop_bug_class]]); +comparing the whole dataclass fails the moment `to_payload` / +`from_stored` / `evolve` drop one between them. +""" + +from dataclasses import replace as dc_replace +from datetime import UTC, datetime +from uuid import UUID, uuid4 + +import pytest + +from cora.data.aggregates.shortfall import ( + Shortfall, + ShortfallReason, + ShortfallRecorded, + ShortfallStatus, + event_type_name, + from_stored, + load_shortfall, + shortfall_stream_id, + to_payload, +) +from cora.infrastructure.adapters.in_memory_event_store import InMemoryEventStore +from cora.infrastructure.event_envelope import to_new_event +from cora.infrastructure.ports.event_store import StoredEvent +from cora.shared.identity import ActorId + +pytestmark = pytest.mark.unit + +_CAPTURE_PATH_ID = UUID("01900000-0000-7000-8000-0000000091a1") +_RUN_ID = UUID("01900000-0000-7000-8000-0000000091b2") +_ACTOR_ID = ActorId(UUID("01900000-0000-7000-8000-0000000091c3")) + +_FILE_MODIFIED_AT = datetime(2026, 7, 29, 14, 59, 52, tzinfo=UTC) +_RUN_ENDED_AT = datetime(2026, 7, 29, 15, 0, 0, tzinfo=UTC) +_RECORDED_AT = datetime(2026, 7, 29, 16, 0, 0, tzinfo=UTC) + + +def _recorded(**overrides: object) -> ShortfallRecorded: + values: dict[str, object] = { + "shortfall_id": shortfall_stream_id(_CAPTURE_PATH_ID), + "producing_run_id": _RUN_ID, + "capture_path_id": _CAPTURE_PATH_ID, + "host": "tomdet", + "root": "/local1/2BM", + "projection_count": 1, + "commanded_projection_count": 1541, + "dropped_frame_count": 3, + "reason": ShortfallReason.STRUCTURALLY_INCOMPLETE, + "file_modified_at": _FILE_MODIFIED_AT, + "run_ended_at": _RUN_ENDED_AT, + "occurred_at": _RECORDED_AT, + "recorded_by": _ACTOR_ID, + } + values.update(overrides) + return ShortfallRecorded(**values) # type: ignore[arg-type] + + +def _stored(event: ShortfallRecorded) -> StoredEvent: + return StoredEvent( + position=1, + event_id=uuid4(), + stream_type="Shortfall", + stream_id=event.shortfall_id, + version=1, + event_type=event_type_name(event), + schema_version=1, + payload=to_payload(event), + correlation_id=uuid4(), + causation_id=None, + occurred_at=event.occurred_at, + recorded_at=event.occurred_at, + ) + + +async def _append(store: InMemoryEventStore, event: ShortfallRecorded) -> None: + await store.append( + stream_type="Shortfall", + stream_id=event.shortfall_id, + expected_version=0, + events=[ + to_new_event( + event_type=event_type_name(event), + payload=to_payload(event), + occurred_at=event.occurred_at, + event_id=uuid4(), + command_name="IngestScan", + correlation_id=uuid4(), + causation_id=None, + principal_id=UUID(str(_ACTOR_ID)), + ) + ], + ) + + +async def test_a_recorded_shortfall_folds_back_to_the_state_that_was_written() -> None: + store = InMemoryEventStore() + event = _recorded() + await _append(store, event) + + loaded = await load_shortfall(store, event.shortfall_id) + + assert loaded == Shortfall( + id=event.shortfall_id, + producing_run_id=_RUN_ID, + capture_path_id=_CAPTURE_PATH_ID, + host="tomdet", + root="/local1/2BM", + projection_count=1, + commanded_projection_count=1541, + dropped_frame_count=3, + reason=ShortfallReason.STRUCTURALLY_INCOMPLETE, + file_modified_at=_FILE_MODIFIED_AT, + run_ended_at=_RUN_ENDED_AT, + recorded_at=_RECORDED_AT, + recorded_by=_ACTOR_ID, + status=ShortfallStatus.RECORDED, + ) + + +async def test_loading_a_stream_that_holds_nothing_returns_none() -> None: + store = InMemoryEventStore() + assert await load_shortfall(store, shortfall_stream_id(_CAPTURE_PATH_ID)) is None + + +def test_absent_counts_survive_the_round_trip_as_none_not_zero() -> None: + """A file that records no commanded total is UNQUANTIFIED, which is + a different fact from one that commanded zero frames. Serializing + None to null and back has to preserve that.""" + event = _recorded(commanded_projection_count=None, dropped_frame_count=None) + + payload = to_payload(event) + assert payload["commanded_projection_count"] is None + assert payload["dropped_frame_count"] is None + + assert from_stored(_stored(event)) == event + + +def test_the_closed_reason_survives_the_round_trip_as_the_enum() -> None: + """The payload carries the string value, and the reader has to hand + back the enum member: a bare string here would defeat the closed + vocabulary the whole design rests on.""" + event = _recorded() + assert to_payload(event)["reason"] == "StructurallyIncomplete" + + rebuilt = from_stored(_stored(event)) + assert rebuilt.reason is ShortfallReason.STRUCTURALLY_INCOMPLETE + + +def test_a_foreign_event_type_on_the_stream_is_refused_loudly() -> None: + """A stream contaminated with another aggregate's event must fail + the fold rather than be silently skipped, or the state that comes + back is quietly wrong.""" + # `replace` rather than a second constructor call: a hand-rebuilt + # StoredEvent silently stops tracking new fields on that dataclass. + contaminated = dc_replace(_stored(_recorded()), event_type="AcquisitionRecorded") + + with pytest.raises(ValueError, match="Unknown ShortfallEvent event_type"): + from_stored(contaminated) + + +def test_the_stream_id_derives_from_the_capture_path_and_nothing_else() -> None: + """Determinism is what makes `expected_version=0` the dedupe rather + than a projection read, and the seed is the opaque surrogate so the + id can never be a confirmation oracle for a guessed path.""" + again = shortfall_stream_id(_CAPTURE_PATH_ID) + other = shortfall_stream_id(UUID("01900000-0000-7000-8000-0000000091ff")) + + assert again == shortfall_stream_id(_CAPTURE_PATH_ID) + assert again != other diff --git a/apps/api/tests/unit/data/test_shortfall_summary_projection.py b/apps/api/tests/unit/data/test_shortfall_summary_projection.py new file mode 100644 index 00000000000..95169e26465 --- /dev/null +++ b/apps/api/tests/unit/data/test_shortfall_summary_projection.py @@ -0,0 +1,123 @@ +"""Unit tests for ShortfallSummaryProjection. + +Pins the subscribed-event-types frozenset (projection-metadata +assertion) and the single INSERT-on-genesis apply path. +""" + +from datetime import UTC, datetime +from typing import Any +from unittest.mock import AsyncMock +from uuid import uuid4 + +import pytest + +from cora.data.aggregates.shortfall.state import ShortfallReason +from cora.data.projections import ShortfallSummaryProjection +from cora.infrastructure.ports.event_store import StoredEvent + +_SHORTFALL_ID = uuid4() +_RUN_ID = uuid4() +_CAPTURE_PATH_ID = uuid4() +_RECORDED_BY = uuid4() +_EVENT_ID = uuid4() +_CORRELATION_ID = uuid4() +_HOST = "tomdet" +_ROOT = "/local1/2BM" +_FILE_MODIFIED_AT = datetime(2026, 6, 10, 9, 0, 0, tzinfo=UTC) +_RUN_ENDED_AT = datetime(2026, 6, 10, 9, 5, 0, tzinfo=UTC) +_OCCURRED_AT = datetime(2026, 6, 11, 12, 0, 0, tzinfo=UTC) + + +def _stored(event_type: str, payload: dict[str, Any]) -> StoredEvent: + return StoredEvent( + position=1, + event_id=_EVENT_ID, + stream_type="Shortfall", + stream_id=_SHORTFALL_ID, + version=1, + event_type=event_type, + schema_version=1, + payload=payload, + correlation_id=_CORRELATION_ID, + causation_id=None, + occurred_at=_OCCURRED_AT, + recorded_at=_OCCURRED_AT, + ) + + +def _recorded_payload( + *, + commanded_projection_count: int | None = 1541, + dropped_frame_count: int | None = 0, +) -> dict[str, Any]: + return { + "shortfall_id": str(_SHORTFALL_ID), + "producing_run_id": str(_RUN_ID), + "capture_path_id": str(_CAPTURE_PATH_ID), + "host": _HOST, + "root": _ROOT, + "projection_count": 1, + "commanded_projection_count": commanded_projection_count, + "dropped_frame_count": dropped_frame_count, + "reason": ShortfallReason.STRUCTURALLY_INCOMPLETE.value, + "file_modified_at": _FILE_MODIFIED_AT.isoformat(), + "run_ended_at": _RUN_ENDED_AT.isoformat(), + "occurred_at": _OCCURRED_AT.isoformat(), + "recorded_by": str(_RECORDED_BY), + } + + +@pytest.mark.unit +def test_projection_metadata() -> None: + proj = ShortfallSummaryProjection() + assert proj.name == "proj_data_shortfall_summary" + assert proj.subscribed_event_types == frozenset({"ShortfallRecorded"}) + + +@pytest.mark.unit +async def test_shortfall_recorded_inserts_with_counts() -> None: + proj = ShortfallSummaryProjection() + conn = AsyncMock() + event = _stored("ShortfallRecorded", _recorded_payload()) + await proj.apply(event, conn) + + args = conn.execute.await_args + assert args is not None + sql = args.args[0] + assert "INSERT INTO proj_data_shortfall_summary" in sql + assert "ON CONFLICT (shortfall_id) DO NOTHING" in sql + assert "'Recorded'" in sql + assert args.args[1] == _SHORTFALL_ID + assert args.args[2] == _RUN_ID + assert args.args[3] == _CAPTURE_PATH_ID + assert args.args[4] == _HOST + assert args.args[5] == _ROOT + assert args.args[6] == 1 + assert args.args[7] == 1541 + assert args.args[8] == 0 + assert args.args[9] == ShortfallReason.STRUCTURALLY_INCOMPLETE.value + assert args.args[10] == _FILE_MODIFIED_AT + assert args.args[11] == _RUN_ENDED_AT + assert args.args[12] == _OCCURRED_AT # recorded_at <- occurred_at + assert args.args[13] == _RECORDED_BY + + +@pytest.mark.unit +async def test_shortfall_recorded_inserts_with_null_counts() -> None: + proj = ShortfallSummaryProjection() + conn = AsyncMock() + payload = _recorded_payload(commanded_projection_count=None, dropped_frame_count=None) + await proj.apply(_stored("ShortfallRecorded", payload), conn) + + args = conn.execute.await_args + assert args is not None + assert args.args[7] is None # commanded_projection_count + assert args.args[8] is None # dropped_frame_count + + +@pytest.mark.unit +async def test_unknown_event_type_falls_through() -> None: + proj = ShortfallSummaryProjection() + conn = AsyncMock() + await proj.apply(_stored("UnrelatedEvent", {}), conn) + conn.execute.assert_not_awaited() diff --git a/apps/api/tests/unit/deployments/test_architecture_introspect.py b/apps/api/tests/unit/deployments/test_architecture_introspect.py index c81ee5bcd46..83240b6fbbb 100644 --- a/apps/api/tests/unit/deployments/test_architecture_introspect.py +++ b/apps/api/tests/unit/deployments/test_architecture_introspect.py @@ -75,13 +75,13 @@ def test_introspection_aggregates_match_filesystem() -> None: assert generated == _filesystem_aggregates() -def test_counts_are_eighteen_bcs_and_forty_three_aggregates() -> None: +def test_counts_are_eighteen_bcs_and_forty_four_aggregates() -> None: # Anti-drift pins for the model.md headline; bump deliberately on a BC/aggregate add. - # 18 BCs / 43 aggregates: the budget BC landed with Allocation (the - # beamline's spending envelope, the budget BC). + # 18 BCs / 44 aggregates: the Data BC gained Shortfall (a capture that + # can never become a Dataset), with no new BC. model = ai.introspect(_CORA) assert model.bc_count == 18 - assert model.aggregate_count == 43 + assert model.aggregate_count == 44 def test_enclosure_bc_and_equipment_role_are_present() -> None: @@ -144,7 +144,7 @@ def test_bc_table_group_map_covers_every_bc() -> None: def test_count_renderer() -> None: assert ap.render_count(_MODEL, {"kind": "bc", "spell": "true", "cap": "true"}) == "Eighteen" - assert ap.render_count(_MODEL, {"kind": "aggregate", "spell": "true"}) == "forty-three" + assert ap.render_count(_MODEL, {"kind": "aggregate", "spell": "true"}) == "forty-four" assert ap.render_count(_MODEL, {"kind": "bc"}) == "18" assert ap.render_count(_MODEL, {"kind": "event", "bc": "decision"}) == "4" assert ap.render_count(_MODEL, {"kind": "slice", "bc": "equipment"}) == "61" diff --git a/infra/atlas/migrations/20260910222120_init_proj_data_shortfall_summary.sql b/infra/atlas/migrations/20260910222120_init_proj_data_shortfall_summary.sql new file mode 100644 index 00000000000..34cee23aba0 --- /dev/null +++ b/infra/atlas/migrations/20260910222120_init_proj_data_shortfall_summary.sql @@ -0,0 +1,82 @@ +-- Data BC's newest projection: shortfall summary. +-- +-- Folds the Shortfall aggregate's single ShortfallRecorded event into +-- the `proj_data_shortfall_summary` read model. The Shortfall is a +-- recorded-fact-chain: terminal at genesis, one stream per +-- capture_path_id, exactly one event ever. The projection mirrors the +-- fact that an observed capture can never become a Dataset, a +-- judgement made once as of the producing Run's terminal. +-- +-- Subscribed events: +-- - ShortfallRecorded -> INSERT (status='Recorded') +-- +-- Dual-time columns: +-- - file_modified_at / run_ended_at: the two sides of the finality +-- judgement, kept so the verdict stays checkable without going +-- back to the file (see cora.data.aggregates.shortfall.state's +-- module docstring). +-- - recorded_at: CORA-side wall-clock when the Shortfall was +-- recorded (the event's occurred_at payload key). +-- +-- Frame accounting: projection_count is what the file actually holds; +-- commanded_projection_count is what the scan was told to collect and +-- is NULL when the file does not record it; dropped_frame_count is +-- the detector's own discard count, also NULL when absent and a +-- SEPARATE fact from the shortfall between the other two. +-- +-- Personal data: host/root are the facility-level storage tier +-- (tomdet, /local1/2BM), never the path itself. capture_path_id is the +-- surrogate key of the run_capture_path vault row. Everything strictly +-- between the root and the filename never reaches this table. +-- +-- NO CHECK on reason: ShortfallReason is closed at the aggregate tier +-- (a StrEnum) with a single member today; a DB-tier CHECK would need +-- editing in lockstep with every future member and buys little the +-- aggregate does not already guarantee. +-- +-- UNIQUE INDEX on capture_path_id: one Shortfall per observation is +-- the aggregate's core invariant, mirroring ScanIngestCandidateLookup's +-- own per-location exclude key (scoped the same way, for the same +-- reason). Enforced here at the DB tier in addition to the +-- deterministic (uuid5) stream id. +-- +-- Mutable read model. cora_app gets full DML. + +CREATE TABLE proj_data_shortfall_summary ( + shortfall_id UUID PRIMARY KEY, + producing_run_id UUID NOT NULL, + capture_path_id UUID NOT NULL, + host TEXT NOT NULL, + root TEXT NOT NULL, + projection_count INTEGER NOT NULL CHECK (projection_count >= 0), + commanded_projection_count INTEGER CHECK (commanded_projection_count >= 0), + dropped_frame_count INTEGER CHECK (dropped_frame_count >= 0), + reason TEXT NOT NULL, + file_modified_at TIMESTAMPTZ NOT NULL, + run_ended_at TIMESTAMPTZ NOT NULL, + recorded_at TIMESTAMPTZ NOT NULL, + recorded_by UUID NOT NULL, + status TEXT NOT NULL CHECK ( + status IN ('Recorded') + ), + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +-- One Shortfall per observation is the aggregate's core invariant +-- (mirrors ScanIngestCandidateLookup's own exclude key). Enforced here +-- at the DB tier in addition to the deterministic (uuid5) stream id. +CREATE UNIQUE INDEX proj_data_shortfall_summary_capture_path_idx + ON proj_data_shortfall_summary (capture_path_id); + +-- "Shortfalls recorded against Run X, newest first" (per-run finality +-- review). +CREATE INDEX proj_data_shortfall_summary_run_idx + ON proj_data_shortfall_summary (producing_run_id, recorded_at DESC); + +GRANT SELECT, INSERT, UPDATE, DELETE + ON proj_data_shortfall_summary TO cora_app; + +INSERT INTO projection_bookmarks (name) +VALUES ('proj_data_shortfall_summary') +ON CONFLICT DO NOTHING; diff --git a/infra/atlas/migrations/atlas.sum b/infra/atlas/migrations/atlas.sum index 083037a0a3c..87731a6618a 100644 --- a/infra/atlas/migrations/atlas.sum +++ b/infra/atlas/migrations/atlas.sum @@ -1,4 +1,4 @@ -h1:E+EVeweNOUnBetEfIlhH5HCpEcETJQVOf07yX3WDKu8= +h1:m9WL1j5CeVOYtyYi9kxsQ8HLMh1mLF6x/t2NdHDAJIg= 20260509120000_init_events.sql h1:GmgCZKfaqXu1m96/cKAks2vhaLWTdEaHTLkFtUo9FXg= 20260509170000_init_idempotency.sql h1:Nbu8DIE4Sv1WiHw3G22+tYffPhKc5Jryw3PMK8wB2zY= 20260510010000_add_event_id.sql h1:RbtYP6uMnOB20zhJ9dNXUi4YVqbmlEzf562pmygnRW8= @@ -176,3 +176,4 @@ h1:E+EVeweNOUnBetEfIlhH5HCpEcETJQVOf07yX3WDKu8= 20260831140000_seed_local_zone_conduit_verdict_logbook.sql h1:ufCypY1LKNxFJuxzMat6zw8cuPuXAuNeJPH4KNqdjXU= 20260831150000_seed_in_process_surface.sql h1:QVyvB18DHwE6gIagfeu5DuC2P8uqqzz8dD65uSd32AU= 20260904120000_add_proj_operation_procedure_iterations_advice_latency.sql h1:MR1m7EoZ1Bf51f+A6qwf7M/WCpv2SusKVs8schf79Ko= +20260910222120_init_proj_data_shortfall_summary.sql h1:+3EbIqdQY5J76TtmUGW+AgbK+qA1lsajN5GEeb6HT8s=