feat(ingest): ingester identity + versioned releases (raw-ingestion provenance)

server/migrations/versions/b47f9c2e8a31_drop_hook_release_digest_uq.py

@@ -0,0 +1,41 @@
+"""drop hook_releases (name, digest) unique constraint
+
+Hook release idempotency changes from digest-only to definition-equality
+against the live release (#217): a config-only redeploy now mints a new
+release with the same digest, so the digest uniqueness constraint must go.
+Mirrors the ingester registry, which shipped without the constraint.
+
+Safe on populated tables: dropping a unique constraint is metadata-only.
+
+Revision ID: b47f9c2e8a31
+Revises: e1a7d20c4f9b
+Create Date: 2026-08-15
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "b47f9c2e8a31"
+down_revision: Union[str, Sequence[str], None] = "e1a7d20c4f9b"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    op.drop_constraint("uq_hook_releases_hook_digest", "hook_releases", type_="unique")
+
+
+def downgrade() -> None:
+    """Downgrade schema.
+
+    NB: recreating the constraint fails if any hook has minted multiple
+    releases with one digest since upgrading — expected, and correct: those
+    rows are legitimate under the new semantics.
+    """
+    op.create_unique_constraint(
+        "uq_hook_releases_hook_digest", "hook_releases", ["hook_name", "digest"]
+    )

server/migrations/versions/e1a7d20c4f9b_add_ingester_registry.py

@@ -0,0 +1,110 @@
+"""add ingester registry: identities, releases, run snapshot
+
+Ingester identity + immutable versioned releases (#180 §1), mirroring the hook
+registry, plus ``ingest_runs.release_id`` — the release a run resolved at start,
+snapshotted for provenance. Incremental revision (live archives upgrade in
+place).
+
+Greenfield provenance: ``release_id`` is NOT NULL — every run is traceable to
+the exact code that fetched its records. Pre-registry runs carry no release to
+point at, so the upgrade clears ``ingest_runs``: runs are operational state and
+re-runnable (re-ingestion re-creates records with full provenance), not archival
+data.
+
+Revision ID: e1a7d20c4f9b
+Revises: 44a8e3799b97
+Create Date: 2026-08-14
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision: str = "e1a7d20c4f9b"
+down_revision: Union[str, Sequence[str], None] = "44a8e3799b97"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    op.create_table(
+        "ingesters",
+        sa.Column("name", sa.String(length=40), nullable=False),
+        sa.Column("schema_id", sa.String(), nullable=False),
+        sa.Column("live_release_id", sa.UUID(), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.PrimaryKeyConstraint("name"),
+    )
+    op.create_index("idx_ingesters_schema_id", "ingesters", ["schema_id"], unique=False)
+    op.create_table(
+        "ingester_releases",
+        sa.Column("id", sa.UUID(), nullable=False),
+        sa.Column("ingester_name", sa.String(length=40), nullable=False),
+        sa.Column("version", sa.Integer(), nullable=False),
+        sa.Column("image", sa.Text(), nullable=False),
+        sa.Column("digest", sa.Text(), nullable=False),
+        sa.Column(
+            "config",
+            postgresql.JSONB(astext_type=sa.Text()),
+            server_default=sa.text("'{}'"),
+            nullable=False,
+        ),
+        sa.Column("limits", postgresql.JSONB(astext_type=sa.Text()), nullable=False),
+        sa.Column("source_ref", sa.Text(), nullable=False),
+        sa.Column("built_by", sa.Text(), nullable=True),
+        sa.Column("built_at", sa.DateTime(timezone=True), nullable=False),
+        sa.ForeignKeyConstraint(["ingester_name"], ["ingesters.name"]),
+        sa.PrimaryKeyConstraint("id"),
+        # No (name, digest) unique constraint: a config-only redeploy mints a
+        # new release with the same digest (idempotency is definition-equality
+        # against the live release, decided in the adapter).
+        sa.UniqueConstraint(
+            "ingester_name", "version", name="uq_ingester_releases_ingester_version"
+        ),
+    )
+    op.create_index(
+        "idx_ingester_releases_ingester_version",
+        "ingester_releases",
+        ["ingester_name", sa.literal_column("version DESC")],
+        unique=False,
+    )
+    # Same circular-dependency break as hooks.live_release_id in initial_schema.
+    op.create_foreign_key(
+        "fk_ingesters_live_release_id",
+        "ingesters",
+        "ingester_releases",
+        ["live_release_id"],
+        ["id"],
+        deferrable=True,
+        initially="DEFERRED",
+    )
+    # Pre-registry runs have no release to point at; runs are re-runnable
+    # operational state, so clear them rather than carry a nullable column.
+    op.execute("DELETE FROM ingest_runs")
+    op.add_column(
+        "ingest_runs",
+        sa.Column("release_id", sa.UUID(), nullable=False),
+    )
+    op.create_foreign_key(
+        "fk_ingest_runs_release_id",
+        "ingest_runs",
+        "ingester_releases",
+        ["release_id"],
+        ["id"],
+    )
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    op.drop_constraint("fk_ingest_runs_release_id", "ingest_runs", type_="foreignkey")
+    op.drop_column("ingest_runs", "release_id")
+    op.drop_constraint("fk_ingesters_live_release_id", "ingesters", type_="foreignkey")
+    op.drop_index("idx_ingester_releases_ingester_version", table_name="ingester_releases")
+    op.drop_table("ingester_releases")
+    op.drop_index("idx_ingesters_schema_id", table_name="ingesters")
+    op.drop_table("ingesters")

server/osa/domain/deposition/command/create_convention.py

@@ -1,7 +1,7 @@
 from datetime import datetime
 from typing import Any
 
-from pydantic import BaseModel, ConfigDict, Field, model_validator
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
 
 from osa.domain.auth.model.principal import Principal
 from osa.domain.deposition.model.deploy import HookDeploy
@@ -28,6 +28,7 @@
 from osa.domain.shared.model.source import (
     IngesterDefinition,
     IngesterLimits,
+    IngesterName,
     IngesterScheduleConfig,
     InitialRunConfig,
 )
@@ -94,15 +95,25 @@ class DeployConventionIngester(BaseModel):
 
     model_config = ConfigDict(extra="forbid")
 
-    name: str  # build-fan-out key (cloud); not persisted server-side
+    # Build fan-out key (cloud) AND, normalised at this boundary, the ingester's
+    # registry identity (#180 §1). Wire names are free-form (class names,
+    # declared slugs); the domain only ever sees a canonical IngesterName.
+    name: str
     config: dict[str, Any] | None = None
     limits: IngesterLimits = Field(default_factory=IngesterLimits)
     schedule: IngesterScheduleConfig | None = None
     initial_run: InitialRunConfig | None = None
     release: DeployConventionRelease
 
+    @field_validator("name")
+    @classmethod
+    def _name_normalises(cls, v: str) -> str:
+        IngesterName.from_raw(v)  # overlong/unnormalisable → 422 here, not deep in deploy
+        return v
+
     def to_definition(self) -> IngesterDefinition:
         return IngesterDefinition(
+            name=IngesterName.from_raw(self.name),
             image=self.release.image,
             digest=self.release.digest,
             config=self.config,

server/osa/domain/deposition/service/convention.py

@@ -7,11 +7,13 @@
 from osa.domain.deposition.model.docs import ConventionDocs
 from osa.domain.deposition.model.value import FileRequirements
 from osa.domain.deposition.port.convention_repository import ConventionRepository
+from osa.domain.ingest.service.ingester_registry import IngesterRegistryService
 from osa.domain.metadata.service.metadata import MetadataService
 from osa.domain.semantics.model.value import FieldDefinition
 from osa.domain.semantics.service.schema import SchemaService
 from osa.domain.shared.error import NotFoundError
 from osa.domain.shared.event import EventId
+from osa.domain.shared.model.hook import OciConfig, OciLimits
 from osa.domain.shared.model.source import IngesterDefinition
 from osa.domain.shared.model.srn import (
     ConventionSlug,
@@ -30,6 +32,7 @@ class ConventionService(Service):
     schema_service: SchemaService  # TODO: replace with a port?
     metadata_service: MetadataService  # TODO: replace with a port?
     hook_registry: HookRegistryService
+    ingester_registry: IngesterRegistryService
     outbox: Outbox
 
     async def deploy(
@@ -55,7 +58,8 @@ async def deploy(
         are unversioned and mutable, so re-declaring the same state is a no-op and
         a differing declaration updates the convention in place — no caller version,
         no conflict path. The schema (versioned, immutable) and each hook release
-        (idempotent on digest) are reused when already present. Feature-table
+        (idempotent on definition-equality with the live release) are reused
+        when already present. Feature-table
         creation is now inlined at the deploy command handler (#160, decision 9);
         the ``ConventionRegistered`` event this method still appends is audit-only
         (no subscribers) — it lives on for the ``/events`` changefeed.
@@ -84,13 +88,33 @@ async def deploy(
         )
 
         # 2) Hooks: upsert each identity (reject a differing contract) + mint its
-        #    release (idempotent on digest, advancing the live pointer).
+        #    release (idempotent on definition-equality, advancing the live pointer).
         for spec in hooks:
             await self.hook_registry.upsert_identity(spec.identity.name, spec.identity.feature)
             await self.hook_registry.create_release(
                 spec.identity.name, spec.runtime, spec.source_ref, built_by
             )
 
+        # 2b) Ingester: same treatment as hooks (#180 §1) — upsert identity +
+        #     mint release (idempotent on definition-equality, advancing the live pointer).
+        #     Unconditional: name and source_ref are required on the model, so
+        #     every declared ingester carries a provenance chain.
+        if ingester is not None:
+            await self.ingester_registry.upsert_identity(
+                ingester.name, LocalId(created_schema.id.id.root)
+            )
+            await self.ingester_registry.create_release(
+                ingester.name,
+                OciConfig(
+                    image=ingester.image,
+                    digest=ingester.digest,
+                    config=ingester.config if ingester.config is not None else {},
+                    limits=OciLimits(**ingester.limits.model_dump()),
+                ),
+                ingester.source_ref,
+                built_by,
+            )
+
         # 3) Convention referencing hooks by name (upsert by slug).
         convention = Convention(
             id=slug,

server/osa/domain/deposition/util/di/provider.py

@@ -20,6 +20,7 @@
 from osa.domain.deposition.query.list_depositions import ListDepositionsHandler
 from osa.domain.deposition.query.list_ingesters import ListIngestersHandler
 from osa.domain.deposition.service.convention import ConventionService
+from osa.domain.ingest.service.ingester_registry import IngesterRegistryService
 from osa.domain.validation.service.hook_registry import HookRegistryService
 from osa.domain.deposition.service.deposition import DepositionService
 from osa.domain.metadata.service.metadata import MetadataService
@@ -56,13 +57,15 @@ def get_convention_service(
         schema_service: SchemaService,
         metadata_service: MetadataService,
         hook_registry: HookRegistryService,
+        ingester_registry: IngesterRegistryService,
         outbox: Outbox,
     ) -> ConventionService:
         return ConventionService(
             convention_repo=convention_repo,
             schema_service=schema_service,
             metadata_service=metadata_service,
             hook_registry=hook_registry,
+            ingester_registry=ingester_registry,
             outbox=outbox,
         )
 

server/osa/domain/ingest/model/ingest_run.py

@@ -5,6 +5,7 @@
 from enum import StrEnum
 from typing import NewType
 
+from osa.domain.ingest.model.ingester_release import IngesterReleaseId
 from osa.domain.shared.error import InvalidStateError
 from osa.domain.shared.failure import FailureKind
 from osa.domain.shared.model.aggregate import Aggregate
@@ -36,6 +37,10 @@ class IngestRun(Aggregate):
 
     id: IngestRunId
     convention_id: str
+    # The ingester release this run resolved at start (#180 §1) — snapshotted,
+    # exactly as validation snapshots hook releases. Required: every run is
+    # traceable to the exact code that fetched its records.
+    release_id: IngesterReleaseId
     status: IngestStatus = IngestStatus.PENDING
     ingestion_finished: bool = False
     batches_ingested: int = 0

server/osa/domain/ingest/model/ingester.py

@@ -0,0 +1,35 @@
+"""Ingester aggregate — stable identity, target schema, live pointer (#180).
+
+Mirrors :class:`~osa.domain.validation.model.hook.Hook`. Where a hook owns a
+fixed *feature contract*, an ingester owns the *schema it produces records for*;
+both carry a ``live_release_id`` naming the currently active release, advanced
+under a row lock by the registry adapter.
+
+``schema_id`` is the bare schema id, not a versioned :class:`SchemaId`: an
+ingester keeps producing records as its schema gains versions, so binding it to
+one version would break on every field addition.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+from osa.domain.ingest.model.ingester_release import IngesterReleaseId
+from osa.domain.shared.model.aggregate import Aggregate
+from osa.domain.shared.model.source import IngesterName
+from osa.domain.shared.model.srn import LocalId
+
+
+class Ingester(Aggregate):
+    """Stable ingester identity + target schema + live-release pointer."""
+
+    model_config = {"frozen": True}
+
+    name: IngesterName
+    schema_id: LocalId
+    live_release_id: IngesterReleaseId | None = None
+    created_at: datetime
+
+    def with_live_release(self, release_id: IngesterReleaseId) -> "Ingester":
+        """Return a copy whose live pointer references *release_id*."""
+        return self.model_copy(update={"live_release_id": release_id})