Record a capture that can never become a Dataset as a Shortfall

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
 """

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"]

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",

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",
+]

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"]