From f18adc3bfedd7c41a8aa7f0fc92e16734eb9e717 Mon Sep 17 00:00:00 2001 From: xmap <16776958+xmap@users.noreply.github.com> Date: Thu, 10 Sep 2026 18:10:40 -0500 Subject: [PATCH 1/2] Record a capture that can never become a Dataset as a Shortfall CORA's record could not distinguish "no scan ran" from "a scan ran and produced nothing": both looked like a terminated Run with no Dataset. Three real 2-BM Runs hold files with 1 projection of a commanded 1541 and no rotation angles, and the sweep retried them 387 times in one log window at a measured 6.3s each, writing nothing. The reader had already computed the frame counts and threw them away at the refusal, so the only non-substrate evidence was discarded precisely when it disagreed with the substrate. Shortfall is a terminal-at-genesis fact-chain beside Acquisition and Attestation, appended from ingest_scan as its second outcome so it needs no new command, grant, route or MCP tool. The 400 still raises: the caller asked for a Dataset and did not get one, so the refusal remains the truthful answer, and recording what CORA learned while answering is a separate act. The sweep's code is unchanged; the candidate simply stops being selected once the fact lands. Three preconditions gate it, all evidentiary rather than identity-based, so a human POSTing an indirect locator is treated exactly like the sweep: a resolved capture_path_id to key the stream on, a terminal Run, and a file whose mtime is strictly older than that terminal. Finality rests on a measurement, not a belief: across 329 files on the detector host every one settled 7.0 to 8.8s before its Run's terminal, so no stability window or attempt counter is needed, and neither could be afforded anyway since reading an mtime costs the same round trip this change exists to stop paying. The design memo assumed both sides of that comparison were already in hand. They were not: Run state carries no terminal timestamp and load_run discards the envelopes, so load_run_ended_at reads it from the stream. Reading it lazily inside the refusal branch made the memo's handler-reordering unnecessary and leaves every existing error precedence untouched. The stream id derives from capture_path_id rather than being minted, so uniqueness comes from expected_version=0 rather than from a projection read: the sweep re-selects a candidate until proj_data_shortfall_summary catches up, and that duplicate append is an expected race, swallowed quietly rather than surfaced as a conflict nobody can act on. Only StructurallyIncomplete records a Shortfall. Unrecognized is permanent too but yields no Description, so its Shortfall would carry a verdict and no evidence; that is a modelling question with no observed instance yet. Keying on the exception class instead of the cause would have permanently condemned every file caught mid-transfer, which is the one mistake here that loses data silently. Co-Authored-By: Claude Opus 5 (1M context) --- .../src/cora/api/_capture_scan_ingestor.py | 32 +- apps/api/src/cora/data/_projections.py | 2 + .../data/adapters/capture_path_locator.py | 73 ++- .../data/aggregates/shortfall/__init__.py | 47 ++ .../data/aggregates/shortfall/_stream_id.py | 55 ++ .../cora/data/aggregates/shortfall/events.py | 179 +++++++ .../cora/data/aggregates/shortfall/evolver.py | 69 +++ .../cora/data/aggregates/shortfall/read.py | 31 ++ .../cora/data/aggregates/shortfall/state.py | 143 ++++++ .../cora/data/features/ingest_scan/handler.py | 218 +++++++- .../api/src/cora/data/projections/__init__.py | 6 +- .../data/projections/shortfall_summary.py | 76 +++ .../record_export/_dispositions.py | 15 + .../record_export/_stream_types.py | 1 + .../src/cora/infrastructure/schema_version.py | 2 +- .../src/cora/run/aggregates/run/__init__.py | 3 +- apps/api/src/cora/run/aggregates/run/read.py | 44 +- .../test_command_name_derives_event_name.py | 13 + .../test_event_class_defined_vs_registered.py | 10 + .../tests/architecture/test_fold_symmetry.py | 15 + .../test_capture_scan_ingestor_postgres.py | 83 ++++ .../api/test_durable_distribution_driver.py | 9 +- .../unit/data/test_capture_path_locator.py | 91 +++- .../unit/data/test_ingest_scan_shortfall.py | 470 ++++++++++++++++++ .../data/test_shortfall_summary_projection.py | 123 +++++ .../test_architecture_introspect.py | 10 +- ...22120_init_proj_data_shortfall_summary.sql | 82 +++ infra/atlas/migrations/atlas.sum | 3 +- 28 files changed, 1860 insertions(+), 45 deletions(-) create mode 100644 apps/api/src/cora/data/aggregates/shortfall/__init__.py create mode 100644 apps/api/src/cora/data/aggregates/shortfall/_stream_id.py create mode 100644 apps/api/src/cora/data/aggregates/shortfall/events.py create mode 100644 apps/api/src/cora/data/aggregates/shortfall/evolver.py create mode 100644 apps/api/src/cora/data/aggregates/shortfall/read.py create mode 100644 apps/api/src/cora/data/aggregates/shortfall/state.py create mode 100644 apps/api/src/cora/data/projections/shortfall_summary.py create mode 100644 apps/api/tests/unit/data/test_ingest_scan_shortfall.py create mode 100644 apps/api/tests/unit/data/test_shortfall_summary_projection.py create mode 100644 infra/atlas/migrations/20260910222120_init_proj_data_shortfall_summary.sql 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..704173e290c --- /dev/null +++ b/apps/api/tests/unit/data/test_ingest_scan_shortfall.py @@ -0,0 +1,470 @@ +"""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) -> 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 + 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 _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 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= From ce9e4ac7b9c0d317626b7a5f6b81ea9a837d513c Mon Sep 17 00:00:00 2001 From: xmap <16776958+xmap@users.noreply.github.com> Date: Thu, 10 Sep 2026 18:39:34 -0500 Subject: [PATCH 2/2] Fold a Shortfall back, so the read side is exercised too ingest_scan only ever appends a Shortfall, so from_stored, evolve, fold and load_shortfall shipped written but never run. In an event-sourced system that is a latent bug rather than a coverage statistic: an event that cannot be read back is not really recorded. The generic net does not catch it either, since test_event_union_from_stored_coverage.py skips single-event aggregates, which is exactly this shape. The round trip compares the WHOLE folded state against an explicitly constructed Shortfall. A per-field assertion is blind to a field the writer forgot to carry, because the reader's default quietly fills it in; whole-dataclass equality fails the moment to_payload, from_stored or evolve drops one between them. Also covers the naive-terminal guard, built the only way that stream can actually arise: to_payload isoformats without an offset and fromisoformat reads it back naive. Diff coverage 86% to 100%. Co-Authored-By: Claude Opus 5 (1M context) --- .../unit/data/test_ingest_scan_shortfall.py | 48 ++++- .../unit/data/test_shortfall_aggregate.py | 184 ++++++++++++++++++ 2 files changed, 231 insertions(+), 1 deletion(-) create mode 100644 apps/api/tests/unit/data/test_shortfall_aggregate.py diff --git a/apps/api/tests/unit/data/test_ingest_scan_shortfall.py b/apps/api/tests/unit/data/test_ingest_scan_shortfall.py index 704173e290c..53c0f0dd23c 100644 --- a/apps/api/tests/unit/data/test_ingest_scan_shortfall.py +++ b/apps/api/tests/unit/data/test_ingest_scan_shortfall.py @@ -148,7 +148,9 @@ def _deps(store: InMemoryEventStore) -> Kernel: ) -async def _seed_run(store: InMemoryEventStore, *, ended: bool) -> None: +async def _seed_run( + store: InMemoryEventStore, *, ended: bool, naive_terminal: bool = False +) -> None: started = RunStarted( run_id=_RUN_ID, name="Shortfall Run", @@ -175,6 +177,12 @@ async def _seed_run(store: InMemoryEventStore, *, ended: bool) -> None: ) 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, @@ -199,6 +207,28 @@ async def _seed_run(store: InMemoryEventStore, *, ended: bool) -> None: ) +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( @@ -468,3 +498,19 @@ async def test_ingest_shortfall_payload_carries_no_part_of_the_observed_path() - 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