Seed the beamline the pilot needs, and prove ingest against it

apps/api/src/cora/equipment/aggregates/family/_family_seed_registry.py

@@ -101,6 +101,11 @@ def _family(
         # Pass 2: fleet cameras (Eiger / Pilatus) write HDF5 / TIFF to disk
         # (iss, xfm, cdi), so the Camera Family records, not only streams.
         Affordance.RECORDING,
+        # A camera's product IS a capture: the affordance's own contract
+        # ("produces a Data BC Acquisition fact on every capture") is what
+        # record_acquisition's Capturing gate checks, and the 2-BM pilot's
+        # detector must pass it for ingest to record anything.
+        Affordance.CAPTURING,
     }
 )
 # Batch 3 (Pass 1, Sensor cluster). Of the seven measurement/analyzer

apps/api/tach.toml

@@ -463,5 +463,11 @@ depends_on = [
     "cora.safety.aggregates",
     "cora.subject",
     "cora.supply",
+    # The pilot seed ceremony (cora.api.pilot_seed) composes Supply
+    # events the same way the RunSupervisor composes Decision events
+    # (blessed cross-BC aggregate access at the composition root): it
+    # folds the Supply stream and appends register + mark-available
+    # genesis/transition events through the real deciders.
+    "cora.supply.aggregates",
     "cora.trust",
 ]

apps/api/tests/integration/test_pilot_seed_postgres.py

@@ -0,0 +1,246 @@
+"""The pilot seed ceremony, end to end against real Postgres.
+
+Three claims, one flow: a fresh database seeds (exit 2), a re-run
+changes nothing (exit 0), and `ingest_scan` then records a real file
+through the REAL `PostgresAssetLookup` and `PostgresSupplyLookup`,
+which retires the in-memory fake as the only Capturing-bearing asset
+in the test suite. This is the proof the seeder design's gate review
+demanded: the seeded camera must surface the Capturing affordance
+through the projection join, or the ceremony is decoration.
+"""
+
+# pyright: reportUnknownMemberType=false, reportUnknownVariableType=false, reportUnknownArgumentType=false
+
+from dataclasses import replace as dc_replace
+from datetime import UTC, datetime
+from pathlib import Path
+from uuid import uuid4
+
+import asyncpg
+import h5py
+import numpy as np
+import pytest
+import pytest_asyncio
+from testcontainers.postgres import PostgresContainer
+
+from cora.api.pilot_seed import asset_seed_id, seed_pilot_beamline
+from cora.data.adapters.data_exchange_scan_reader import DataExchangeScanReader
+from cora.data.adapters.posix_checksum import PosixChecksumAdapter
+from cora.data.features import ingest_scan
+from cora.data.wire import (
+    _build_dataset_by_checksum_lookup,  # pyright: ignore[reportPrivateUsage]
+)
+from cora.equipment.adapters.postgres_asset_lookup import PostgresAssetLookup
+from cora.infrastructure.postgres.pool import create_pool
+from cora.supply.adapters.postgres_supply_lookup import PostgresSupplyLookup
+from tests._postgres import normalize_async_url
+from tests.integration._helpers import build_postgres_deps
+
+pytestmark = pytest.mark.integration
+
+SeedDatabase = tuple[asyncpg.Pool, str]
+
+_NOW = datetime(2026, 7, 29, 16, 0, 0, tzinfo=UTC)
+_FACILITY = "cora"
+_BEAMLINE = "2-bm"
+_CAMERA = "Camera"
+
+
+@pytest_asyncio.fixture
+async def seed_database(
+    postgres_container: PostgresContainer,
+    template_database: str,
+):
+    """A per-test database plus its URL, because the ceremony builds its
+    own pool from a URL rather than borrowing the fixture's."""
+    test_db = f"seed_{uuid4().hex[:12]}"
+    admin_url = normalize_async_url(postgres_container.get_connection_url(), database="postgres")
+    admin = await asyncpg.connect(admin_url)
+    try:
+        await admin.execute(f'CREATE DATABASE "{test_db}" TEMPLATE "{template_database}"')
+    finally:
+        await admin.close()
+
+    test_url = normalize_async_url(postgres_container.get_connection_url(), database=test_db)
+    pool = await create_pool(test_url, min_size=1, max_size=4)
+    try:
+        yield pool, test_url
+    finally:
+        await pool.close()
+        admin = await asyncpg.connect(admin_url)
+        try:
+            await admin.execute(f'DROP DATABASE "{test_db}"')
+        finally:
+            await admin.close()
+
+
+async def _run_ceremony(url: str, *, dry_run: bool = False) -> int:
+    return await seed_pilot_beamline(
+        facility_code=_FACILITY,
+        beamline=_BEAMLINE,
+        root_name="2-BM",
+        camera_name=_CAMERA,
+        camera_family_name="Camera",
+        supply_name="analysis-tier",
+        dry_run=dry_run,
+        database_url=url,
+    )
+
+
+def _write_scan(path: Path) -> None:
+    with h5py.File(path, "w") as f:
+        f.create_dataset("exchange/data", data=np.zeros((4, 4, 4), dtype=np.uint16))
+        f.create_dataset("exchange/theta", data=np.linspace(0.0, 180.0, 4))
+        f.create_dataset("process/acquisition/rotation/num_angles", data=4)
+        f.create_dataset("process/acquisition/start_date", data="2026-07-29T10:15:30-05:00")
+
+
+async def test_ceremony_seeds_then_rerun_changes_nothing(seed_database: SeedDatabase) -> None:
+    pool, url = seed_database
+
+    first = await _run_ceremony(url)
+    assert first == 2, "first run must report seeded"
+
+    events_after_first = await pool.fetchval("SELECT COUNT(*) FROM events")
+
+    second = await _run_ceremony(url)
+    assert second == 0, "second run must report all-exists"
+
+    events_after_second = await pool.fetchval("SELECT COUNT(*) FROM events")
+    assert events_after_second == events_after_first, "a re-run must append zero events"
+
+
+async def test_dry_run_writes_nothing_beyond_bootstrap(seed_database: SeedDatabase) -> None:
+    pool, url = seed_database
+
+    exit_code = await _run_ceremony(url, dry_run=True)
+    assert exit_code == 2, "dry run against a fresh database reports would-seed"
+
+    camera_streams = await pool.fetchval(
+        "SELECT COUNT(*) FROM events WHERE stream_id = $1",
+        asset_seed_id(_FACILITY, _BEAMLINE, _CAMERA),
+    )
+    assert camera_streams == 0, "dry run must not write the camera asset"
+
+
+async def test_seeded_camera_carries_capturing_through_the_real_lookup(
+    seed_database: SeedDatabase,
+) -> None:
+    pool, url = seed_database
+    assert await _run_ceremony(url) == 2
+
+    asset = await PostgresAssetLookup(pool).lookup(asset_seed_id(_FACILITY, _BEAMLINE, _CAMERA))
+    assert asset is not None
+    assert "Capturing" in asset.family_affordances
+
+
+async def test_ingest_against_the_seeded_beamline_records_the_dataset(
+    seed_database: SeedDatabase, tmp_path: Path
+) -> None:
+    """The whole pipeline with zero fakes: ceremony, then a real HDF5
+    file through the real reader, digest, deciders, projections."""
+    pool, url = seed_database
+    assert await _run_ceremony(url) == 2
+
+    supplies = await PostgresSupplyLookup(pool).find_supplies_by_kind(kinds=frozenset({"Storage"}))
+    assert "Storage" in supplies, "the ceremony must have registered a Storage supply"
+    supply_id = supplies["Storage"][0].supply_id
+
+    scan = tmp_path / "scan_001.h5"
+    _write_scan(scan)
+
+    deps = dc_replace(
+        build_postgres_deps(
+            pool,
+            now=_NOW,
+            ids=[uuid4() for _ in range(12)],
+            asset_lookup=PostgresAssetLookup(pool),
+        ),
+        supply_lookup=PostgresSupplyLookup(pool),
+    )
+    handler = ingest_scan.bind(
+        deps,
+        scan_reader=DataExchangeScanReader(allowed_roots=(str(tmp_path),)),
+        checksum_computer=PosixChecksumAdapter(allowed_roots=(str(tmp_path),)),
+        dataset_by_checksum_lookup=_build_dataset_by_checksum_lookup(deps),
+    )
+
+    dataset_id = await handler(
+        ingest_scan.IngestScan(
+            locator=scan.as_uri(),
+            producing_asset_id=asset_seed_id(_FACILITY, _BEAMLINE, _CAMERA),
+            supply_id=supply_id,
+            access_protocol="POSIX",
+        ),
+        principal_id=uuid4(),
+        correlation_id=uuid4(),
+    )
+
+    events, version = await deps.event_store.load("Dataset", dataset_id)
+    assert version == 1
+    assert events[0].event_type == "DatasetRegistered"
+    assert events[0].payload["name"] == "scan_001.h5"
+
+
+async def test_unknown_facility_code_reports_error_and_exit_one(
+    seed_database: SeedDatabase,
+) -> None:
+    """The facility guard is the loud alternative to registering under
+    the wrong facility; nothing else runs after it."""
+    _, url = seed_database
+
+    exit_code = await seed_pilot_beamline(
+        facility_code="nosuchfacility",
+        beamline=_BEAMLINE,
+        root_name="2-BM",
+        camera_name=_CAMERA,
+        camera_family_name="Camera",
+        supply_name="analysis-tier",
+        dry_run=False,
+        database_url=url,
+    )
+
+    assert exit_code == 1
+
+
+async def test_unknown_family_name_reports_error_and_exit_one(seed_database: SeedDatabase) -> None:
+    _, url = seed_database
+
+    exit_code = await seed_pilot_beamline(
+        facility_code=_FACILITY,
+        beamline=_BEAMLINE,
+        root_name="2-BM",
+        camera_name=_CAMERA,
+        camera_family_name="NoSuchFamily",
+        supply_name="analysis-tier",
+        dry_run=False,
+        database_url=url,
+    )
+
+    assert exit_code == 1
+
+
+async def test_mid_ceremony_exception_reports_error_and_exit_one(
+    seed_database: SeedDatabase, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """The CLI catch: any unexpected failure becomes a named report line
+    and exit 1, never a traceback swallowed into a zero."""
+    _, url = seed_database
+
+    async def explode(*_args: object, **_kwargs: object) -> None:
+        raise RuntimeError("synthetic mid-ceremony failure")
+
+    monkeypatch.setattr("cora.api.pilot_seed.verify_schema_version", explode)
+
+    exit_code = await seed_pilot_beamline(
+        facility_code=_FACILITY,
+        beamline=_BEAMLINE,
+        root_name="2-BM",
+        camera_name=_CAMERA,
+        camera_family_name="Camera",
+        supply_name="analysis-tier",
+        dry_run=False,
+        database_url=url,
+    )
+
+    assert exit_code == 1

apps/api/tests/unit/api/test_pilot_seed.py

@@ -0,0 +1,112 @@
+"""pilot_seed's pure parts: identity stability, report semantics, CLI.
+
+The database-touching flow lives in the integration tier
+(test_pilot_seed_postgres); this tier pins what must never drift
+without ceremony: the deterministic Asset key format (a change orphans
+every previously seeded instance), the exit-code taxonomy operators
+and CI gate on, and the CLI defaults.
+"""
+
+from uuid import UUID
+
+import pytest
+
+from cora.api.pilot_seed import (
+    ASSET_SEED_NAMESPACE,
+    _Report,  # pyright: ignore[reportPrivateUsage]
+    asset_seed_id,
+    build_parser,
+)
+
+pytestmark = pytest.mark.unit
+
+
+def test_asset_seed_id_repeated_calls_return_the_same_id() -> None:
+    first = asset_seed_id("aps", "2-bm", "Camera")
+    second = asset_seed_id("aps", "2-bm", "Camera")
+    assert first == second
+
+
+def test_asset_seed_id_pins_the_key_format() -> None:
+    """The exact uuid5 over "facility:beamline:asset:name". Changing the
+    namespace or the format orphans every previously seeded instance,
+    so this pin must only ever move with a migration story."""
+    from uuid import uuid5
+
+    assert asset_seed_id("aps", "2-bm", "Camera") == uuid5(
+        ASSET_SEED_NAMESPACE, "aps:2-bm:asset:Camera"
+    )
+
+
+def test_asset_seed_id_distinguishes_beamlines_and_names() -> None:
+    ids = {
+        asset_seed_id("aps", "2-bm", "Camera"),
+        asset_seed_id("aps", "7-bm", "Camera"),
+        asset_seed_id("aps", "2-bm", "Mirror"),
+        asset_seed_id("maxiv", "2-bm", "Camera"),
+    }
+    assert len(ids) == 4
+
+
+def test_report_all_exists_leaves_seeded_and_failed_unset() -> None:
+    report = _Report(lines=[])
+    report.note("exists", "asset 2-BM")
+    report.note("exists", "supply analysis tier")
+    assert report.seeded is False
+    assert report.failed is False
+
+
+def test_report_any_seed_marks_seeded() -> None:
+    report = _Report(lines=[])
+    report.note("exists", "asset 2-BM")
+    report.note("seeded", "supply analysis tier")
+    assert report.seeded is True
+    assert report.failed is False
+
+
+def test_report_any_error_marks_failed() -> None:
+    report = _Report(lines=[])
+    report.note("seeded", "asset 2-BM")
+    report.note("error", "family Camera", "lacks Capturing")
+    assert report.failed is True
+
+
+def test_parser_defaults_name_the_pilot() -> None:
+    args = build_parser().parse_args([])
+    assert args.facility_code == "cora"
+    assert args.beamline == "2-bm"
+    assert args.camera_family_name == "Camera"
+    assert args.dry_run is False
+
+
+def test_parser_accepts_overrides() -> None:
+    args = build_parser().parse_args(
+        ["--facility-code", "aps", "--camera-name", "Oryx", "--dry-run"]
+    )
+    assert args.facility_code == "aps"
+    assert args.camera_name == "Oryx"
+    assert args.dry_run is True
+
+
+def test_asset_seed_namespace_is_the_locked_constant() -> None:
+    assert UUID("6c1f4a52-8f2e-4bb0-9d59-1a4c9be1a23d") == ASSET_SEED_NAMESPACE
+
+
+def test_main_parses_argv_and_returns_the_ceremony_exit_code(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from cora.api import pilot_seed
+
+    received: dict[str, object] = {}
+
+    async def fake_ceremony(**kwargs: object) -> int:
+        received.update(kwargs)
+        return 2
+
+    monkeypatch.setattr(pilot_seed, "seed_pilot_beamline", fake_ceremony)
+
+    exit_code = pilot_seed.main(["--camera-name", "Oryx", "--dry-run"])
+
+    assert exit_code == 2
+    assert received["camera_name"] == "Oryx"
+    assert received["dry_run"] is True

catalog/catalog.yaml

@@ -77,7 +77,7 @@ families:
     presents_as: [Positioner]
     note: "Limited-range rotational tilt-stage Family (a goniometer). Distinct from RotaryStage (it carries no PSO Following/Marking for fly-scans) and from LinearStage (the motion is rotational, not translation). 2-BM: the LaminographyPitch Kohzu goniometer, modelled in the SampleTower Assembly."
   - name: Camera
-    affordances: [Imageable, Binnable, Coolable, Triggerable, Streamable, Recording]
+    affordances: [Imageable, Binnable, Coolable, Triggerable, Streamable, Recording, Capturing]
     presents_as: [Detector]
   - name: Scintillator
     affordances: [Consumable]