diff --git a/apps/api/src/cora/api/pilot_seed.py b/apps/api/src/cora/api/pilot_seed.py new file mode 100644 index 00000000000..337b5290a9d --- /dev/null +++ b/apps/api/src/cora/api/pilot_seed.py @@ -0,0 +1,514 @@ +"""The pilot seed ceremony: give a CORA instance the beamline ingest needs. + +`python -m cora.api.pilot_seed` registers, idempotently, the minimum a +deployment must know before `ingest_scan` can record anything: the +beamline root Unit Asset (facility-bound per the anchoring XOR), the +camera Device Asset, its Camera family attachment (whose seed roster +carries Capturing), and one Storage-kind Supply. Inputs are explicit +CLI arguments; the ceremony reads no descriptor. The full +descriptor-reconciling onboarding is a deliberate later slice with its +own trigger (see project_beamline_seeder_design), because 52 of the +descriptor's 53 instances have no production reader in the read-only +pilot and the two things ingest needs are exactly the two the +descriptor cannot provide. + +## Identity is per-aggregate, matching each aggregate's locked design + + - Assets: deterministic `uuid5(ASSET_SEED_NAMESPACE, + "{facility}:{beamline}:asset:{name}")`, path-qualified so no bare + name is reserved repo-wide, appended at expected_version=0 with + ConcurrencyError meaning already-seeded. + - Families: RESOLVED via `family_stream_id(name)`, never defined + here. Family ids are bare-name-derived by design (a Camera at APS + and at MAX IV must share one id or Assembly content hashes fork), + and definitions belong to the seed registry's graduation + governance. + - Enclosures: not touched. They are boot-seeded with minted ids and + an address pre-check precisely because deterministic ids would + collide with tombstones on re-register. + - Supplies: minted id; idempotency comes from an address pre-check + against the supply projection, mirroring the partial-unique + address that makes deregister-then-re-register legal. + +## Why the ceremony drains projections itself + +The registration deciders demand facility lookup results that +production fills from projections, and a standalone kernel has no +projection worker. The ceremony therefore runs the same idempotent +bootstrap hooks the app lifespan runs (federation for the +self-Facility, equipment for roles and families) and drains the +relevant projections between stages; without that, a fresh database +refuses every registration with FacilityNotFound. + +## What a re-run does + +Nothing, loudly. Every instance reports one of: seeded, exists, +retired (the stream folds to Decommissioned; the ceremony never +resurrects a tombstone, since decommission-then-re-register is the +operator's rebind path, not the seeder's), or error. Exit codes: 0 +when everything already existed, 2 when anything was seeded, 1 on any +error. A `--dry-run` prints the same report and writes nothing. +""" + +import argparse +import asyncio +import sys +from dataclasses import dataclass +from uuid import UUID, uuid5 + +from cora.equipment._bootstrap import bootstrap_equipment, bootstrap_families +from cora.equipment._projections import register_equipment_projections +from cora.equipment.aggregates.asset import Asset, AssetLifecycle, AssetTier +from cora.equipment.aggregates.asset.events import event_type_name as asset_event_type_name +from cora.equipment.aggregates.asset.events import to_payload as asset_to_payload +from cora.equipment.aggregates.asset.read import load_asset +from cora.equipment.aggregates.family import Affordance, FamilyName +from cora.equipment.aggregates.family._family_registry import family_stream_id +from cora.equipment.aggregates.family.read import load_family +from cora.equipment.features.add_asset_family.command import AddAssetFamily +from cora.equipment.features.add_asset_family.decider import decide as decide_add_family +from cora.equipment.features.register_asset.command import RegisterAsset +from cora.equipment.features.register_asset.decider import decide as decide_asset +from cora.federation._bootstrap import bootstrap_federation +from cora.federation._projections import register_federation_projections +from cora.federation.adapters.postgres_facility_lookup import PostgresFacilityLookup +from cora.infrastructure.config import Settings +from cora.infrastructure.deps import make_postgres_kernel +from cora.infrastructure.event_envelope import to_new_event +from cora.infrastructure.kernel import Kernel +from cora.infrastructure.ports import ( + AllowAllAuthorize, + SystemClock, + UUIDv7Generator, +) +from cora.infrastructure.ports.event_store import ConcurrencyError +from cora.infrastructure.postgres.pool import create_pool +from cora.infrastructure.projection.drain import drain_projections +from cora.infrastructure.projection.worker import ProjectionRegistry +from cora.infrastructure.routing import SYSTEM_PRINCIPAL_ID +from cora.infrastructure.schema_version import verify_schema_version +from cora.shared.facility_code import FacilityCode +from cora.shared.identity import ActorId +from cora.supply._projections import register_supply_projections +from cora.supply.adapters.postgres_supply_lookup import PostgresSupplyLookup +from cora.supply.aggregates.supply.events import event_type_name as supply_event_type_name +from cora.supply.aggregates.supply.events import to_payload as supply_to_payload +from cora.supply.aggregates.supply.state import Supply, SupplyCannotMarkAvailableError, SupplyStatus +from cora.supply.features.mark_supply_available.command import MarkSupplyAvailable +from cora.supply.features.mark_supply_available.decider import decide as decide_mark_available +from cora.supply.features.register_supply.command import RegisterSupply +from cora.supply.features.register_supply.decider import decide as decide_supply + +#: Namespace for the ceremony's deterministic Asset identities. Path- +#: qualified keys under it ("aps:2-bm:asset:") make re-runs +#: idempotent without reserving bare names repo-wide (the Role/Imager +#: lesson) and keep two beamlines' identically named devices distinct. +ASSET_SEED_NAMESPACE = UUID("6c1f4a52-8f2e-4bb0-9d59-1a4c9be1a23d") + +_COMMAND_NAME = "SeedPilotBeamline" + +_EXIT_CLEAN = 0 +_EXIT_ERROR = 1 +_EXIT_SEEDED = 2 + + +def asset_seed_id(facility_code: str, beamline: str, name: str) -> UUID: + return uuid5(ASSET_SEED_NAMESPACE, f"{facility_code}:{beamline}:asset:{name}") + + +@dataclass +class _Report: + lines: list[str] + seeded: bool = False + failed: bool = False + + def note(self, outcome: str, subject: str, detail: str = "") -> None: + suffix = f" ({detail})" if detail else "" + self.lines.append(f"{outcome:<8} {subject}{suffix}") + if outcome == "seeded": + self.seeded = True + if outcome == "error": + self.failed = True + + +async def seed_pilot_beamline( + *, + facility_code: str, + beamline: str, + root_name: str, + camera_name: str, + camera_family_name: str, + supply_name: str, + dry_run: bool, + database_url: str | None = None, +) -> int: + """Run the ceremony. `database_url` overrides the Settings value so + the integration tier can point a run at its per-test database; the + CLI always uses the deployment's own configuration.""" + settings = Settings() + pool = await create_pool( + database_url if database_url is not None else settings.database_url, + min_size=1, + max_size=4, + ) + report = _Report(lines=[]) + try: + await verify_schema_version(pool) + # The REAL projection-backed lookups, explicitly: the kernel's + # defaults are test stubs (a synthetic-supply lookup and an + # empty in-memory facility lookup), and a ceremony that + # pre-checks idempotency against a stub re-seeds on every run. + kernel = make_postgres_kernel( + pool, + settings=settings, + clock=SystemClock(), + id_generator=UUIDv7Generator(), + authz=AllowAllAuthorize(), + facility_lookup=PostgresFacilityLookup(pool), + supply_lookup=PostgresSupplyLookup(pool), + ) + + registry = ProjectionRegistry() + register_federation_projections(registry, kernel) + register_equipment_projections(registry, kernel) + register_supply_projections(registry, kernel) + + # Prerequisites the app lifespan normally seeds. All + # idempotent; a dry run still runs them so its report reads + # against a database in the state a real run would see. + await bootstrap_federation(kernel) + await bootstrap_equipment(kernel) + await bootstrap_families(kernel) + await drain_projections(pool, registry) + + code = FacilityCode(facility_code) + facility = await kernel.facility_lookup.lookup_by_code(code) + if facility is None: + report.note( + "error", + f"facility {facility_code}", + "not found after bootstrap; is SELF_FACILITY_CODE set to this code?", + ) + return _finish(report, dry_run) + + actor = ActorId(SYSTEM_PRINCIPAL_ID) + clock = kernel.clock + ids = kernel.id_generator + run_correlation_id = ids.new_id() + family_id = family_stream_id(FamilyName(camera_family_name)) + + # Guard the whole point: the camera family must carry Capturing + # or ingest's Acquisition gate refuses every capture. On a fresh + # database the roster seed above guarantees it; on an already- + # seeded database the bootstrap skips existing streams, so an + # older Camera lacks the affordance until version_family runs. + family = await load_family(kernel.event_store, family_id) + if family is None: + report.note("error", f"family {camera_family_name}", "not seeded; unknown family name") + return _finish(report, dry_run) + if Affordance.CAPTURING not in family.affordances: + report.note( + "error", + f"family {camera_family_name}", + "lacks the Capturing affordance; this database predates the " + "roster change. Remedy: version_family adding Capturing.", + ) + return _finish(report, dry_run) + report.note("exists", f"family {camera_family_name}", "resolved, carries Capturing") + + root_id = asset_seed_id(facility_code, beamline, root_name) + camera_id = asset_seed_id(facility_code, beamline, camera_name) + + async def seed_asset(asset_id: UUID, command: RegisterAsset, label: str) -> Asset | None: + state = await load_asset(kernel.event_store, asset_id) + if state is not None: + if state.lifecycle is AssetLifecycle.DECOMMISSIONED: + report.note("retired", label, "tombstoned stream left untouched") + return None + report.note("exists", label) + return state + events = decide_asset( + state=None, + command=command, + now=clock.now(), + new_id=asset_id, + commissioned_by=actor, + facility_lookup_result=facility, + ) + if dry_run: + report.note("seeded", label, "dry-run, not written") + return None + envelopes = [ + to_new_event( + event_type=asset_event_type_name(event), + payload=asset_to_payload(event), + occurred_at=event.occurred_at, + event_id=ids.new_id(), + command_name=_COMMAND_NAME, + correlation_id=run_correlation_id, + principal_id=SYSTEM_PRINCIPAL_ID, + ) + for event in events + ] + try: + await kernel.event_store.append( + stream_type="Asset", + stream_id=asset_id, + expected_version=0, + events=envelopes, + ) + except ConcurrencyError: + report.note("exists", label, "raced another writer; already present") + return await load_asset(kernel.event_store, asset_id) + report.note("seeded", label) + return await load_asset(kernel.event_store, asset_id) + + root = await seed_asset( + root_id, + RegisterAsset( + name=root_name, + tier=AssetTier.UNIT, + parent_id=None, + facility_code=facility_code, + ), + f"asset {root_name} (Unit root)", + ) + + camera = await seed_asset( + camera_id, + RegisterAsset( + name=camera_name, + tier=AssetTier.DEVICE, + parent_id=root_id, + facility_code=None, + ), + f"asset {camera_name} (Device)", + ) + + # Family attachment: strict-not-idempotent decider, so fold + # first. A family absent from the folded set is treated as this + # ceremony completing its own interrupted work and attached. The + # design memo's removed-by-operator discriminator is NOT + # implementable from folded state (the fold keeps no removal + # history), so an operator who deliberately detached the family + # will see it re-attach on the next run; acceptable for the + # pilot camera, revisited if a real detach case appears. + if camera is not None: + if family_id in camera.family_ids: + report.note("exists", f"{camera_name} family attachment") + elif dry_run: + report.note("seeded", f"{camera_name} family attachment", "dry-run, not written") + else: + current_state, current_version = await _load_asset_with_version(kernel, camera_id) + attach_events = decide_add_family( + state=current_state, + command=AddAssetFamily(asset_id=camera_id, family_id=family_id), + now=clock.now(), + ) + attach_envelopes = [ + to_new_event( + event_type=asset_event_type_name(event), + payload=asset_to_payload(event), + occurred_at=event.occurred_at, + event_id=ids.new_id(), + command_name=_COMMAND_NAME, + correlation_id=run_correlation_id, + principal_id=SYSTEM_PRINCIPAL_ID, + ) + for event in attach_events + ] + await kernel.event_store.append( + stream_type="Asset", + stream_id=camera_id, + expected_version=current_version, + events=attach_envelopes, + ) + report.note("seeded", f"{camera_name} family attachment") + + # Storage supply: minted id, address-pre-checked idempotency. + supplies_by_kind = await kernel.supply_lookup.find_supplies_by_kind( + kinds=frozenset({"Storage"}) + ) + existing_storage = [ + supply + for supply in supplies_by_kind.get("Storage", []) + if supply.name == supply_name and supply.facility_code == facility_code + ] + supply_id: UUID | None = None + if existing_storage: + supply_id = existing_storage[0].supply_id + report.note("exists", f"supply {supply_name} (Storage)") + elif dry_run: + report.note("seeded", f"supply {supply_name} (Storage)", "dry-run, not written") + else: + supply_id = ids.new_id() + supply_events = decide_supply( + state=None, + command=RegisterSupply( + kind="Storage", + name=supply_name, + facility_code=facility_code, + ), + now=clock.now(), + new_id=supply_id, + triggered_by=actor, + facility_lookup_result=facility, + asset_lookup_result=None, + ) + supply_envelopes = [ + to_new_event( + event_type=supply_event_type_name(event), + payload=supply_to_payload(event), + occurred_at=event.occurred_at, + event_id=ids.new_id(), + command_name=_COMMAND_NAME, + correlation_id=run_correlation_id, + principal_id=SYSTEM_PRINCIPAL_ID, + ) + for event in supply_events + ] + await kernel.event_store.append( + stream_type="Supply", + stream_id=supply_id, + expected_version=0, + events=supply_envelopes, + ) + report.note("seeded", f"supply {supply_name} (Storage)") + + # Availability: a freshly registered Supply is not Available, + # and the run-preflight gates plus the legacy-Distribution + # backfill (`SELF_FACILITY_DEFAULT_STORAGE_SUPPLY_CODE`) only + # accept an Available one, so a supply left merely registered is + # half-seeded. The transition decider is strict, so fold first + # and only mark when the status still permits it; any other + # status (Degraded, Unavailable) is an operator's statement and + # stays hands-off. + if supply_id is not None: + supply_state, supply_version = await _load_supply_with_version(kernel, supply_id) + if supply_state is None: + report.note("error", f"supply {supply_name} availability", "stream missing") + elif supply_state.status is SupplyStatus.AVAILABLE: + report.note("exists", f"supply {supply_name} availability") + elif dry_run: + report.note("seeded", f"supply {supply_name} availability", "dry-run, not written") + else: + try: + mark_events = decide_mark_available( + state=supply_state, + command=MarkSupplyAvailable( + supply_id=supply_id, + reason="seeded by the pilot ceremony", + ), + now=clock.now(), + triggered_by=actor, + ) + except SupplyCannotMarkAvailableError as exc: + report.note("retired", f"supply {supply_name} availability", str(exc)) + else: + mark_envelopes = [ + to_new_event( + event_type=supply_event_type_name(event), + payload=supply_to_payload(event), + occurred_at=event.occurred_at, + event_id=ids.new_id(), + command_name=_COMMAND_NAME, + correlation_id=run_correlation_id, + principal_id=SYSTEM_PRINCIPAL_ID, + ) + for event in mark_events + ] + await kernel.event_store.append( + stream_type="Supply", + stream_id=supply_id, + expected_version=supply_version, + events=mark_envelopes, + ) + report.note("seeded", f"supply {supply_name} availability") + + _ = root + if not dry_run: + # Leave the projections current so a re-run's supply + # pre-check (and the app, before its worker catches up) + # sees what this run wrote. + await drain_projections(pool, registry) + return _finish(report, dry_run) + except Exception as exc: # the ceremony is a CLI: name it, exit 1 + report.note("error", "ceremony", str(exc)) + return _finish(report, dry_run) + finally: + await pool.close() + + +async def _load_asset_with_version(kernel: Kernel, asset_id: UUID) -> tuple[Asset | None, int]: + from cora.equipment.aggregates.asset.events import from_stored + from cora.equipment.aggregates.asset.evolver import fold + + stored, version = await kernel.event_store.load("Asset", asset_id) + events = [from_stored(event) for event in stored] + return fold(events), version + + +async def _load_supply_with_version(kernel: Kernel, supply_id: UUID) -> tuple[Supply | None, int]: + from cora.supply.aggregates.supply.events import from_stored + from cora.supply.aggregates.supply.evolver import fold + + stored, version = await kernel.event_store.load("Supply", supply_id) + events = [from_stored(event) for event in stored] + return fold(events), version + + +def _finish(report: _Report, dry_run: bool) -> int: + header = "pilot seed (dry run)" if dry_run else "pilot seed" + print(header) + for line in report.lines: + print(f" {line}") + if report.failed: + return _EXIT_ERROR + return _EXIT_SEEDED if report.seeded else _EXIT_CLEAN + + +def build_parser() -> argparse.ArgumentParser: + """The CLI surface, separate from `main` so tests can pin the + defaults and flags without touching a database. + + The facility default is `cora`, matching `SELF_FACILITY_CODE`'s own + default: the ceremony registers under the facility this deployment + self-seeded, and a mismatch is a loud FacilityNotFound rather than a + quietly wrong registration. + """ + parser = argparse.ArgumentParser( + prog="python -m cora.api.pilot_seed", + description=( + "Register the minimum a deployment needs before ingest_scan " + "can record: the beamline root Unit, the camera Device with " + "its Capturing-bearing family, and a Storage supply. " + "Idempotent; re-runs report and change nothing." + ), + ) + parser.add_argument("--facility-code", default="cora") + parser.add_argument("--beamline", default="2-bm") + parser.add_argument("--root-name", default="2-BM") + parser.add_argument("--camera-name", default="Camera") + parser.add_argument("--camera-family-name", default="Camera") + parser.add_argument("--supply-name", default="analysis-tier") + parser.add_argument("--dry-run", action="store_true") + return parser + + +def main(argv: list[str] | None = None) -> int: + args = build_parser().parse_args(argv) + return asyncio.run( + seed_pilot_beamline( + facility_code=args.facility_code, + beamline=args.beamline, + root_name=args.root_name, + camera_name=args.camera_name, + camera_family_name=args.camera_family_name, + supply_name=args.supply_name, + dry_run=args.dry_run, + ) + ) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/apps/api/src/cora/equipment/aggregates/family/_family_seed_registry.py b/apps/api/src/cora/equipment/aggregates/family/_family_seed_registry.py index 252e80c4f37..7e1968022ee 100644 --- a/apps/api/src/cora/equipment/aggregates/family/_family_seed_registry.py +++ b/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 diff --git a/apps/api/tach.toml b/apps/api/tach.toml index aa0f76cadab..5981a7f3636 100644 --- a/apps/api/tach.toml +++ b/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", ] diff --git a/apps/api/tests/integration/test_pilot_seed_postgres.py b/apps/api/tests/integration/test_pilot_seed_postgres.py new file mode 100644 index 00000000000..fe615020baa --- /dev/null +++ b/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 diff --git a/apps/api/tests/unit/api/test_pilot_seed.py b/apps/api/tests/unit/api/test_pilot_seed.py new file mode 100644 index 00000000000..ecaecfb1be3 --- /dev/null +++ b/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 diff --git a/catalog/catalog.yaml b/catalog/catalog.yaml index 3a6595e9e1c..dbc13687f4a 100644 --- a/catalog/catalog.yaml +++ b/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] diff --git a/docs/stack/deployment.md b/docs/stack/deployment.md index 4f3b6ab6cb1..c0b65ef3bca 100644 --- a/docs/stack/deployment.md +++ b/docs/stack/deployment.md @@ -385,6 +385,54 @@ The operator's bootstrap path: The bootstrap seed stays on disk + in the event log forever; you can re-point at it during recovery scenarios. +### Seeding the beamline the pilot needs + +`ingest_scan` refuses to record until two things exist: a registered +Asset whose family affordances include Capturing, and a registered +Storage-kind Supply. Neither appears by itself. The pilot seed +ceremony registers them, idempotently: + +```bash +python -m cora.api.pilot_seed --dry-run +``` + +then the same command without `--dry-run` once the report reads right. +It registers the beamline root Unit Asset, the camera Device Asset, +its Camera family attachment (the Camera seed family carries +Capturing), and one Storage Supply which it also marks Available (a +merely-registered Supply satisfies neither the run-preflight gates nor +the legacy-Distribution backfill). It prints one line per instance: +`seeded`, `exists`, `retired` (a decommissioned stream is reported and +never resurrected; decommission-then-re-register is an operator +rebind, not a seeding concern), or `error`. Exit codes: 0 when +everything already existed, 2 when anything was seeded (or would be, +under `--dry-run`), 1 on any error. Re-running is always safe and +changes nothing. + +The seeded supply's name (default `analysis-tier`) pairs with +`SELF_FACILITY_DEFAULT_STORAGE_SUPPLY_CODE`: the lifespan-time +Distribution backfill resolves that env var against the same supply +name. A deployment that will ever hold legacy Datasets should set the +env var to the seeded name in the same breath as running the ceremony. + +Two facts worth knowing before running it. The ceremony runs the same +idempotent bootstrap hooks the app lifespan runs and drains the +relevant projections itself, so it works against a never-booted +database; and `--facility-code` must name the code this deployment +self-seeds (`SELF_FACILITY_CODE`), or the run stops with a loud +FacilityNotFound rather than registering under the wrong facility. + +A database seeded before the Camera family carried Capturing reports +`error: family Camera lacks the Capturing affordance` and names the +remedy (`version_family` adding it); the ceremony never mutates a +family itself, because family definitions belong to the seed +registry's graduation governance. + +This is deliberately NOT a descriptor import. The full +beamline-descriptor onboarding is a later slice with its own trigger; +the ceremony seeds exactly what the read-only pilot's ingest path +exercises, with explicit arguments and no hidden inputs. + ## Recovery ### Backup and point-in-time recovery