From ff9a80cc8af494e286d028f96c75d1ceca7d8835 Mon Sep 17 00:00:00 2001 From: xmap <16776958+xmap@users.noreply.github.com> Date: Wed, 29 Jul 2026 16:51:46 +0300 Subject: [PATCH 1/2] Seed the beamline the pilot needs, and prove ingest against it Scan ingest shipped requiring two things no production path registers: an Asset whose family affordances include Capturing, and a Storage-kind Supply. The full descriptor-reconciling onboarding was designed, gate-reviewed, and deliberately deferred: the review showed the descriptor cannot provide either prerequisite (its six supplies are beamline utilities, storage is deployment infrastructure; the Camera family lacked Capturing by roster design) and 52 of its 53 instances have no production reader in a read-only pilot. What ships instead is the minimal honest pair. First, Capturing graduates onto the Camera seed family. The affordance's own contract ("produces a Data BC Acquisition fact on every capture") is what record_acquisition's gate checks, and a camera is precisely the thing it describes. The catalog moves in lockstep. A database seeded before this change reports the version_family remedy rather than being silently mutated: family definitions belong to the seed registry's graduation governance, and the ceremony never defines or edits one. Second, python -m cora.api.pilot_seed registers, idempotently: the beamline root Unit Asset, the camera Device Asset, its Camera family attachment, and one Storage Supply, which it also marks Available, because a merely-registered Supply satisfies neither the run-preflight gates nor the legacy-Distribution backfill that resolves SELF_FACILITY_DEFAULT_STORAGE_SUPPLY_CODE by the same name. Identity is per-aggregate, matching each aggregate's locked design rather than one blanket rule: Assets get deterministic path-qualified uuid5 ids (no bare name reserved repo-wide); Families are resolved via their bare-name-derived ids, never defined (a Camera at APS and at MAX IV must share one id or Assembly content hashes fork); Enclosures are not touched (boot-seeded, minted ids, address pre-check); Supplies keep minted ids with the address pre-check as the idempotency surface. The ceremony runs the same idempotent bootstrap hooks the app lifespan runs and drains projections itself, because the registration deciders demand lookup results that production fills from projections and a standalone kernel has no worker. The kernel wires the REAL projection-backed lookups explicitly; the construction-site defaults are test stubs, and a ceremony that pre-checks idempotency against a synthetic-supply stub re-seeds on every run, which is exactly the bug the re-run test caught. The proof is the integration flow the seeder design's review demanded: a fresh database seeds (exit 2), a re-run reports all-exists and appends zero events (exit 0), a dry run writes nothing, the seeded camera surfaces Capturing through the real PostgresAssetLookup join, and ingest_scan then records a real Data Exchange file with nothing faked anywhere in the chain, retiring the in-memory lookup fake as the suite's only Capturing source. Co-Authored-By: Claude Fable 5 --- apps/api/src/cora/api/pilot_seed.py | 514 ++++++++++++++++++ .../family/_family_seed_registry.py | 5 + apps/api/tach.toml | 6 + .../integration/test_pilot_seed_postgres.py | 182 +++++++ apps/api/tests/unit/api/test_pilot_seed.py | 92 ++++ catalog/catalog.yaml | 2 +- docs/stack/deployment.md | 48 ++ 7 files changed, 848 insertions(+), 1 deletion(-) create mode 100644 apps/api/src/cora/api/pilot_seed.py create mode 100644 apps/api/tests/integration/test_pilot_seed_postgres.py create mode 100644 apps/api/tests/unit/api/test_pilot_seed.py 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..770e2e578f8 --- /dev/null +++ b/apps/api/tests/integration/test_pilot_seed_postgres.py @@ -0,0 +1,182 @@ +"""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" 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..3dbaeff1500 --- /dev/null +++ b/apps/api/tests/unit/api/test_pilot_seed.py @@ -0,0 +1,92 @@ +"""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 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 From fdb4b0a82fe95c066e037ae7f2ebe3534c1311af Mon Sep 17 00:00:00 2001 From: xmap <16776958+xmap@users.noreply.github.com> Date: Wed, 29 Jul 2026 17:14:36 +0300 Subject: [PATCH 2/2] Cover the ceremony's refusal branches The diff-coverage gate flagged the error paths: unknown facility, unknown family, the CLI exception catch, and main itself. Each is now pinned by the test that documents its behavior (loud exit 1 with the instance named), which is worth having independent of the gate: the refusal branches are the operator-facing half of the ceremony. Co-Authored-By: Claude Fable 5 --- .../integration/test_pilot_seed_postgres.py | 64 +++++++++++++++++++ apps/api/tests/unit/api/test_pilot_seed.py | 20 ++++++ 2 files changed, 84 insertions(+) diff --git a/apps/api/tests/integration/test_pilot_seed_postgres.py b/apps/api/tests/integration/test_pilot_seed_postgres.py index 770e2e578f8..fe615020baa 100644 --- a/apps/api/tests/integration/test_pilot_seed_postgres.py +++ b/apps/api/tests/integration/test_pilot_seed_postgres.py @@ -180,3 +180,67 @@ async def test_ingest_against_the_seeded_beamline_records_the_dataset( 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 index 3dbaeff1500..ecaecfb1be3 100644 --- a/apps/api/tests/unit/api/test_pilot_seed.py +++ b/apps/api/tests/unit/api/test_pilot_seed.py @@ -90,3 +90,23 @@ def test_parser_accepts_overrides() -> None: 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