diff --git a/mpcontribs-api/src/mpcontribs_api/domains/_shared/models.py b/mpcontribs-api/src/mpcontribs_api/domains/_shared/models.py
index 2b4b840e7..16b4dddb5 100644
--- a/mpcontribs-api/src/mpcontribs_api/domains/_shared/models.py
+++ b/mpcontribs-api/src/mpcontribs_api/domains/_shared/models.py
@@ -9,7 +9,7 @@
 from pymongo.results import DeleteResult
 
 from mpcontribs_api import pagination
-from mpcontribs_api.domains._shared.types import MD5Hash, NFKCStr
+from mpcontribs_api.domains._shared.types import Identity, MD5Hash, NFKCStr
 from mpcontribs_api.projection import SparseFieldsModel
 
 
@@ -25,24 +25,25 @@ class BaseDocumentWithInput[TId](Document):
     """
 
     HAS_DERIVED_FIELDS: ClassVar[bool] = False
+    # The domain's ``Identity`` subclass. Describes how documents are uniquely addressed
+    identity_model: ClassVar[type[Identity]] = Identity
     # Required, non-null, resource-specific id. Overrides Document's optional ``PydanticObjectId`` id.
     id: TId = Field(alias="_id")  # pyright: ignore[reportGeneralTypeIssues, reportIncompatibleVariableOverride]
 
-    @classmethod
-    def identifier_fields(cls) -> frozenset[str]:
-        """Field names that uniquely identify a document in this collection.
-
-        This is the natural/unique key a caller can supply without first knowing the Mongo ``_id``
-        (e.g. ``{"name", "owner"}`` for a project group). The repository pairs these names with
-        caller-supplied values to locate a single resource, and rejects any value dict whose keys
-        don't match this set. Defaults to the primary key; subclasses with a meaningful compound key
-        override it.
+    def identifiers(self) -> dict[str, Any]:
+        """This document's natural-key field values, keyed by ``identity_model.model_fields``.
+
+        The natural/unique key is a caller-suppliable alternative to the Mongo ``_id`` (e.g.
+        ``{"name", "owner"}`` for a project group); it comes straight from the domain's ``Identity``
+        subclass so each domain declares its key exactly once.
         """
-        return frozenset({"id"})
+        return {field: getattr(self, field) for field in self.identity_model.model_fields}
 
-    def identifiers(self) -> dict[str, Any]:
-        """This document's identifier field values, keyed by :meth:`identifier_fields`."""
-        return {field: getattr(self, field) for field in self.identifier_fields()}
+    def identity(self) -> Identity:
+        """This document's identity as a concrete :class:`Identity`, built from its own fields."""
+        return self.identity_model.from_document(
+            {name: getattr(self, name) for name in self.identity_model.model_fields}
+        )
 
     def derived_field_updates(self) -> dict[str, Any]:
         """Server-derived fields to persist alongside a patch.
@@ -103,6 +104,15 @@ def canonical_md5(payload: Mapping[str, Any]) -> str:
     return hashlib.md5(normalized.encode("utf-8")).hexdigest()
 
 
+class ComponentIdentity(Identity):
+    """Identity of a content-addressed component: its ``md5`` content hash.
+
+    Shared by every component domain (structures/tables/attachments).
+    """
+
+    md5: MD5Hash
+
+
 class ComponentIn(BaseModel):
     """Base for component input payloads.
 
@@ -124,17 +134,13 @@ class Component(BaseDocumentWithInput[PydanticObjectId]):
     """
 
     HAS_DERIVED_FIELDS: ClassVar[bool] = True
+    identity_model: ClassVar[type[Identity]] = ComponentIdentity
     name: NFKCStr
     # Server-computed; the placeholder default is overwritten by ``_recompute_md5`` on validation.
     md5: MD5Hash = Field(default="0" * 32)
 
     hash_fields: ClassVar[frozenset[str]]
 
-    @classmethod
-    def identifier_fields(cls) -> frozenset[str]:
-        """A component is content-addressed: its ``md5`` uniquely identifies its content."""
-        return frozenset({"md5"})
-
     def derived_field_updates(self) -> dict[str, Any]:
         """Recompute ``md5`` from the (patched-in-memory) content so the write stays authoritative."""
         return {"md5": self.compute_md5()}
diff --git a/mpcontribs-api/src/mpcontribs_api/domains/_shared/repository.py b/mpcontribs-api/src/mpcontribs_api/domains/_shared/repository.py
index fec8bef1a..63d82779d 100644
--- a/mpcontribs-api/src/mpcontribs_api/domains/_shared/repository.py
+++ b/mpcontribs-api/src/mpcontribs_api/domains/_shared/repository.py
@@ -7,7 +7,7 @@
 from abc import ABC
 from collections.abc import AsyncIterable, AsyncIterator, Callable, Iterable, Mapping
 from contextlib import AbstractAsyncContextManager
-from typing import Any, ClassVar
+from typing import Any, ClassVar, cast
 
 import structlog
 from beanie import PydanticObjectId, UpdateResponse
@@ -23,7 +23,7 @@
 from mpcontribs_api.config import get_settings
 from mpcontribs_api.domains._shared.bulk import BulkFailure, BulkWriteSummary, bulk_failure_from_exception
 from mpcontribs_api.domains._shared.models import BaseDocumentWithInput, DeleteResponse, DocumentOut
-from mpcontribs_api.domains._shared.types import DownloadFormat, ShortMimeFormat
+from mpcontribs_api.domains._shared.types import DownloadFormat, Identity, ShortMimeFormat
 from mpcontribs_api.exceptions import ConflictError, DownloadError, NotFoundError, ValidationError
 from mpcontribs_api.pagination import CursorParams, Page, encode_cursor
 from mpcontribs_api.scope import Scope
@@ -124,18 +124,18 @@ async def read_many(
     def _identifier_query(self, identifiers: dict[str, Any]) -> dict[str, Any]:
         """Turn a ``{field: value}`` identifier dict into a scoped Mongo query fragment.
 
-        The keys must be either the model's :meth:`identifier_fields` exactly, or the bare
-        primary-key form ``{"id": ...}`` (which addresses any document by its ``_id`` regardless of
-        its semantic identifier). ``id`` is remapped to Mongo's ``_id`` (mirroring
+        The keys must be either the model's natural key (``identity_model.model_fields``) exactly, or
+        the bare primary-key form ``{"id": ...}`` (which addresses any document by its ``_id``
+        regardless of its semantic identifier). ``id`` is remapped to Mongo's ``_id`` (mirroring
         ``BaseFilter._get_filter_conditions``) since a raw dict query does not go through Beanie's
         alias resolution.
 
         Args:
-            identifiers (dict[str, Any]): identifier field values keyed by ``identifier_fields``,
+            identifiers (dict[str, Any]): identifier field values keyed by the model's natural key,
                 or ``{"id": <primary key>}``
         """
         identifiers = self.coerce_identifiers(identifiers)
-        expected = self.document_model.identifier_fields()
+        expected = self.document_model.identity_model.model_fields.keys()
         if identifiers.keys() != expected and identifiers.keys() != {"id"}:
             raise ValidationError(
                 "identifiers must match the model's identifier fields, or be a bare {'id': ...}",
@@ -153,7 +153,8 @@ async def read_one(
         """Return the single scoped document matching ``identifiers``, projected to ``fields``.
 
         Args:
-            identifiers (dict[str, Any]): identifier field values keyed by ``identifier_fields``
+            identifiers (dict[str, Any]): identifier field values keyed by the model's natural key, or the bare
+                primary-key form ``{"id": <primary key>}``
             fields (frozenset[str] | None): fields to project; if None the full document is returned
             session (AsyncClientSession | None): optional client session for transactions
         """
@@ -217,15 +218,49 @@ async def insert_many(self, documents: list[TDoc], session: AsyncClientSession |
         """
         return await self.document_model.insert_many(documents, ordered=False, session=session)
 
+    def _scoped_identity_match(self, identity: Identity) -> dict[str, Any]:
+        """Scoped Mongo match locating the single document with ``identity`` (its full natural key)."""
+        match = {("_id" if key == "id" else key): value for key, value in identity.as_dict().items()}
+        return {"$and": [self._scope, match]} if self._scope else match
+
     async def upsert_one(self, document: TDoc, session: AsyncClientSession | None = None) -> TDoc:
-        """Insert ``document`` or replace the existing one with the same ``_id`` (PUT semantics).
+        """Insert ``document`` or merge it into the existing one with the same identity (natural key).
 
-        Domains whose upsert key is a compound identity rather than ``_id`` (e.g. contributions) override this.
+        Null fields are dropped from the ``$set`` (``keep_nulls`` parity); on insert the full ``document`` is written.
+        For PUT-by-``_id`` replace semantics use :meth:`replace_one`.
 
         Args:
             document (TDoc): the fully-built document to persist
             session (AsyncClientSession | None): optional client session for transactions
         """
+        match = self._scoped_identity_match(document.identity())
+        update_data = document.model_dump(exclude={"id"}, exclude_none=True)
+        try:
+            result = await self.document_model.find_one(match, session=session).upsert(  # pyright: ignore[reportGeneralTypeIssues] # beanie UpdateQuery is awaitable
+                Set(update_data),
+                on_insert=document,
+                response_type=UpdateResponse.NEW_DOCUMENT,
+                session=session,
+            )
+        except DuplicateKeyError as exc:
+            raise ConflictError(
+                f"Cannot upsert {self.document_model.__name__}: a conflicting document already exists",
+                identifiers=document.identifiers(),
+            ) from exc
+        return cast(TDoc, result)  # upsert always returns the resulting document
+
+    async def replace_one(self, id: Any, document: TDoc, session: AsyncClientSession | None = None) -> TDoc:
+        """Insert ``document`` under ``id`` or fully replace the existing one at that ``_id`` (PUT).
+
+        This replaces the whole document at a primary key — omitted/null fields are cleared — matching HTTP PUT.
+        ``document.id`` is set to ``id`` so the write always lands under the caller's key.
+
+        Args:
+            id (Any): the primary key to write under
+            document (TDoc): the fully-built replacement document
+            session (AsyncClientSession | None): optional client session for transactions
+        """
+        document.id = id
         try:
             return await document.save(session=session)
         except DuplicateKeyError as exc:
@@ -291,7 +326,8 @@ async def delete_one(
         """Delete the single scoped document matching ``identifiers``.
 
         Args:
-            identifiers (dict[str, Any]): identifier field values keyed by ``identifier_fields``
+            identifiers (dict[str, Any]): identifier field values keyed by the model's natural key, or the bare
+                primary-key form ``{"id": <primary key>}``
             session (AsyncClientSession | None): optional client session for transactions
         """
         query = self._identifier_query(identifiers)
@@ -367,7 +403,8 @@ async def update_one(
         """Partially update the single scoped document matching ``identifiers``.
 
         Args:
-            identifiers (dict[str, Any]): identifier field values keyed by ``identifier_fields``
+            identifiers (dict[str, Any]): identifier field values keyed by the model's natural key, or the bare
+                primary-key form ``{"id": <primary key>}``
             update (TPatch): the partial update to apply; unset fields are dropped
             session (AsyncClientSession | None): optional client session for transactions
             extra_set (dict[str, Any] | None): server-resolved fields to merge into the ``$set``
diff --git a/mpcontribs-api/src/mpcontribs_api/domains/_shared/types.py b/mpcontribs-api/src/mpcontribs_api/domains/_shared/types.py
index c04615b29..6cfd3d4f3 100644
--- a/mpcontribs-api/src/mpcontribs_api/domains/_shared/types.py
+++ b/mpcontribs-api/src/mpcontribs_api/domains/_shared/types.py
@@ -1,15 +1,12 @@
 import re
 import unicodedata
-from abc import ABC
 from collections.abc import Callable, Mapping
-from dataclasses import MISSING, dataclass, fields
 from enum import StrEnum
-from functools import cache
-from typing import Annotated, Any, Self, get_args, get_type_hints
+from typing import Annotated, Any, Self
 
 import polars as pl
 from fastapi import Query
-from pydantic import BeforeValidator, Field, PlainSerializer, WithJsonSchema
+from pydantic import BaseModel, BeforeValidator, ConfigDict, Field, PlainSerializer, WithJsonSchema
 from pymatgen.core import Element
 from pymongo import ASCENDING, IndexModel
 
@@ -424,62 +421,45 @@ def map_keys(value: Any, *, coerce: Callable[[Any], str], on_scalar: Callable[[A
     return on_scalar(value)
 
 
-@cache
-def _optional_field_names(cls: type) -> frozenset[str]:
-    """Names of ``cls``'s dataclass fields whose type admits ``None`` (resolved through string annotations)."""
-    hints = get_type_hints(cls)
-    return frozenset(f.name for f in fields(cls) if type(None) in get_args(hints.get(f.name)))
-
-
-# dataclass construction is cheaper than Pydantic.BaseModel
-@dataclass(frozen=True, slots=True)
-class Identity(ABC):  # noqa: B024  # base kept abstract as a marker; from_document is a shared concrete helper
+class Identity(BaseModel):
     """The full identity of a document model.
 
     Field declaration order IS the identity/index column order: ``index_model`` and ``projection``
-    iterate ``dataclasses.fields`` in that order, so the order is declared exactly once (below).
+    iterate ``model_fields`` (which preserves definition order) so the order is declared exactly once
+    on each subclass.
     """
 
     # WARNING: the order the fields are specified in reflects their ordering for indices. Changing the order
     # creates index migration. Only change intentionally
+    model_config = ConfigDict(frozen=True)
 
     def as_dict(self) -> dict[str, Any]:
         """Identity as a flat dict keyed by field name (for Mongo match clauses and upsert)."""
-        return {f.name: getattr(self, f.name) for f in fields(self)}
-
-    @classmethod
-    def model_fields(cls) -> frozenset[str]:
-        """Returns the field names as a frozenset"""
-        return frozenset(f.name for f in fields(cls))
+        return {name: getattr(self, name) for name in type(self).model_fields}
 
     @classmethod
     def from_document(cls, doc: Mapping[str, Any]) -> Self:
         """Build from a raw Mongo document/projection, tolerating null-stripped fields.
 
-        Generic over any ``@dataclass`` subclass: iterates the concrete class's own fields,
-        falling back to each field's default (or ``None`` for a defaultless Optional field) when the
-        document omits it, since Mongo strips nulls (``keep_nulls=False``). Required non-null fields
-        that are absent surface as a ``TypeError`` from the constructor.
+        Uses `model_construct` to avoid having to revalidate data that is already valid (stored), and to avoid
+        errors on legacy documents.
         """
-        optional = _optional_field_names(cls)
-        kwargs: dict[str, Any] = {}
-        for f in fields(cls):
-            if doc.get(f.name) is not None:
-                kwargs[f.name] = doc[f.name]
-            elif f.default is not MISSING:
-                kwargs[f.name] = f.default
-            elif f.default_factory is not MISSING:
-                kwargs[f.name] = f.default_factory()
-            elif f.name in optional:
-                kwargs[f.name] = None
-        return cls(**kwargs)
+        present = {name: doc[name] for name in cls.model_fields if doc.get(name) is not None}
+        missing = [name for name, field in cls.model_fields.items() if field.is_required() and name not in present]
+        if missing:
+            raise TypeError(f"{cls.__name__}.from_document missing required field(s): {sorted(missing)}")
+        return cls.model_construct(**present)
 
     @classmethod
-    def index_model(cls, name: str = "project_identity", *, unique: bool = True) -> IndexModel:
-        """The unique index enforcing identity — keys follow the field order so they can't drift."""
-        return IndexModel(keys=[(f.name, ASCENDING) for f in fields(cls)], name=name, unique=unique)
+    def index_model(cls, name: str = "identity", *, unique: bool = True) -> IndexModel:
+        """The unique index enforcing identity — keys follow the field order so they can't drift.
+
+        ``name`` should be passed explicitly to match the collection's deployed index name; renaming
+        a live index forces a drop/recreate migration.
+        """
+        return IndexModel(keys=[(name_, ASCENDING) for name_ in cls.model_fields], name=name, unique=unique)
 
     @classmethod
     def projection(cls) -> dict[str, int]:
         """A Mongo projection selecting exactly the identity fields."""
-        return {f.name: 1 for f in fields(cls)}
+        return {name: 1 for name in cls.model_fields}
diff --git a/mpcontribs-api/src/mpcontribs_api/domains/attachments/models.py b/mpcontribs-api/src/mpcontribs_api/domains/attachments/models.py
index e379c7317..8ba32842e 100644
--- a/mpcontribs-api/src/mpcontribs_api/domains/attachments/models.py
+++ b/mpcontribs-api/src/mpcontribs_api/domains/attachments/models.py
@@ -2,7 +2,7 @@
 from pydantic import field_validator
 
 from mpcontribs_api.domains._shared.filters import BaseFilter
-from mpcontribs_api.domains._shared.models import Component, ComponentIn, DocumentOut
+from mpcontribs_api.domains._shared.models import Component, ComponentIdentity, ComponentIn, DocumentOut
 from mpcontribs_api.domains._shared.types import FileLike, MD5Hash, MimeFormat, NFKCStr
 from mpcontribs_api.exceptions import ValidationError
 from mpcontribs_api.projection import SparseFieldsModel
@@ -27,6 +27,7 @@ class Attachment(Component):
 
     class Settings:
         name = "attachments"
+        indexes = [ComponentIdentity.index_model(name="md5")]
 
     @field_validator("name", mode="before")
     @classmethod
diff --git a/mpcontribs-api/src/mpcontribs_api/domains/attachments/router.py b/mpcontribs-api/src/mpcontribs_api/domains/attachments/router.py
index e042ed172..ceb1f0a6b 100644
--- a/mpcontribs-api/src/mpcontribs_api/domains/attachments/router.py
+++ b/mpcontribs-api/src/mpcontribs_api/domains/attachments/router.py
@@ -5,7 +5,7 @@
 from fastapi_filter import FilterDepends
 
 from mpcontribs_api.dependencies import S3Dep, require_user
-from mpcontribs_api.domains._shared.models import ComponentDeleteResponse
+from mpcontribs_api.domains._shared.models import ComponentDeleteResponse, ComponentIdentity
 from mpcontribs_api.domains._shared.types import (
     DownloadFormat,
     FieldSelector,
@@ -30,6 +30,33 @@ async def read_many(
     return await service.read_many(filter=filter, fields=selected, pagination=pagination)
 
 
+@router.get("/item")
+async def read_one_by_identity(
+    service: AttachmentServiceDep,
+    identity: Annotated[ComponentIdentity, Depends()],
+    fields: FieldSelector = None,
+):
+    """Return a single attachment addressed by its content ``md5`` (its natural key)."""
+    selected = AttachmentOut.parse_fields(fields)
+    return await service.read_one(identifiers=identity.as_dict(), fields=selected)
+
+
+@router.delete("/item", response_model=ComponentDeleteResponse, dependencies=[Depends(require_user)])
+async def delete_one_by_identity(service: AttachmentServiceDep, identity: Annotated[ComponentIdentity, Depends()]):
+    """Delete a single attachment addressed by its content ``md5`` (its natural key)."""
+    return await service.delete_one(identifiers=identity.as_dict())
+
+
+@router.patch("/item", dependencies=[Depends(require_user)])
+async def update_one_by_identity(
+    service: AttachmentServiceDep,
+    identity: Annotated[ComponentIdentity, Depends()],
+    update: AttachmentPatch,
+):
+    """Patch a single attachment addressed by its content ``md5`` (its natural key)."""
+    return await service.update_one(identifiers=identity.as_dict(), update=update)
+
+
 @router.get("/{id}")
 async def read_one(
     service: AttachmentServiceDep,
@@ -85,5 +112,5 @@ async def update_one(
     id: str,
     update: AttachmentPatch,
 ):
-    """Patch a single attachment addressed by its ``_id`` or its content ``md5``."""
+    """Patch a single attachment addressed by its ``_id``."""
     return await service.update_one(identifiers={"id": id}, update=update)
diff --git a/mpcontribs-api/src/mpcontribs_api/domains/consumers/models.py b/mpcontribs-api/src/mpcontribs_api/domains/consumers/models.py
index 6e6d218ae..f47fa908e 100644
--- a/mpcontribs-api/src/mpcontribs_api/domains/consumers/models.py
+++ b/mpcontribs-api/src/mpcontribs_api/domains/consumers/models.py
@@ -1,11 +1,19 @@
+from typing import ClassVar
+
 from beanie import PydanticObjectId
 from fastapi_filter import FilterDepends, with_prefix
 from pydantic import BaseModel, Field
-from pymongo import ASCENDING, IndexModel
 
 from mpcontribs_api.config import get_settings
 from mpcontribs_api.domains._shared.filters import BaseFilter
 from mpcontribs_api.domains._shared.models import BaseDocumentWithInput, DocumentOut
+from mpcontribs_api.domains._shared.types import Identity
+
+
+class ConsumerIdentity(Identity):
+    """A consumer override's identity: Kong's unique ``consumer_id``."""
+
+    consumer_id: str
 
 
 class ConsumerSettings(BaseModel):
@@ -49,14 +57,10 @@ class Consumer(BaseDocumentWithInput[PydanticObjectId]):
     they leave unset inherits the env-backed default, snapshotted onto the document at insert time.
     """
 
+    identity_model: ClassVar[type[Identity]] = ConsumerIdentity
     consumer_id: str
     settings: ConsumerSettings = Field(default_factory=ConsumerSettings)
 
-    @classmethod
-    def identifier_fields(cls) -> frozenset[str]:
-        """A consumer override is keyed by Kong's ``consumer_id`` (its unique natural key)."""
-        return frozenset({"consumer_id"})
-
     @classmethod
     def with_defaults(cls, consumer_id: str = "") -> Consumer:
         """In-memory Consumer whose ``settings`` carry the env-backed default limits.
@@ -78,7 +82,7 @@ def from_input_model(cls, data: ConsumerIn) -> Consumer:
     class Settings:
         name = "mp_consumers"
         keep_nulls = False
-        indexes = [IndexModel([("consumer_id", ASCENDING)], name="consumer_id", unique=True)]
+        indexes = [ConsumerIdentity.index_model(name="consumer_id")]
 
 
 class ConsumerIn(BaseModel):
diff --git a/mpcontribs-api/src/mpcontribs_api/domains/consumers/router.py b/mpcontribs-api/src/mpcontribs_api/domains/consumers/router.py
index c63fcd197..ec095c257 100644
--- a/mpcontribs-api/src/mpcontribs_api/domains/consumers/router.py
+++ b/mpcontribs-api/src/mpcontribs_api/domains/consumers/router.py
@@ -8,6 +8,7 @@
 from mpcontribs_api.domains.consumers.dependencies import ConsumerServiceDep
 from mpcontribs_api.domains.consumers.models import (
     ConsumerFilter,
+    ConsumerIdentity,
     ConsumerIn,
     ConsumerOut,
     ConsumerPatch,
@@ -32,6 +33,37 @@ async def read_many(
     return await service.read_many(filter=filter, pagination=pagination, fields=selected)
 
 
+@router.get("/item")
+async def read_one_by_identity(
+    identity: Annotated[ConsumerIdentity, Depends()],
+    service: ConsumerServiceDep,
+    fields: FieldSelector = None,
+):
+    """Get a single consumer override by its natural key, Kong's ``consumer_id`` (admin only)."""
+    selected = ConsumerOut.parse_fields(fields)
+    return await service.read_one(identity.as_dict(), fields=selected)
+
+
+@router.patch("/item", response_model=ConsumerOut)
+async def update_one_by_identity(
+    service: ConsumerServiceDep,
+    identity: Annotated[ConsumerIdentity, Depends()],
+    update: ConsumerPatch,
+):
+    """Partially update a consumer override by its ``consumer_id`` natural key (admin only)."""
+    return await service.update_one(identity.as_dict(), update)
+
+
+@router.delete("/item", status_code=status.HTTP_204_NO_CONTENT)
+async def delete_one_by_identity(
+    service: ConsumerServiceDep,
+    identity: Annotated[ConsumerIdentity, Depends()],
+):
+    """Delete a consumer override by its ``consumer_id`` natural key (admin only)."""
+    await service.delete_one(identity.as_dict())
+    return Response(status_code=status.HTTP_204_NO_CONTENT)
+
+
 @router.get("/{id}")
 async def read_one(
     id: str,
@@ -40,7 +72,7 @@ async def read_one(
 ):
     """Get a single consumer override by document id (admin only)."""
     selected = ConsumerOut.parse_fields(fields)
-    return await service.read_one(id, fields=selected)
+    return await service.read_one({"id": id}, fields=selected)
 
 
 @router.post("", response_model=ConsumerOut, status_code=status.HTTP_201_CREATED)
@@ -59,7 +91,7 @@ async def update_one(
     update: ConsumerPatch,
 ):
     """Partially update a consumer override by document id (admin only)."""
-    return await service.update_one(id, update)
+    return await service.update_one({"id": id}, update)
 
 
 @router.delete("/{id}", status_code=status.HTTP_204_NO_CONTENT)
@@ -68,5 +100,5 @@ async def delete_one(
     id: str,
 ):
     """Delete a consumer override by document id (admin only)."""
-    await service.delete_one(id)
+    await service.delete_one({"id": id})
     return Response(status_code=status.HTTP_204_NO_CONTENT)
diff --git a/mpcontribs-api/src/mpcontribs_api/domains/consumers/service.py b/mpcontribs-api/src/mpcontribs_api/domains/consumers/service.py
index c4051ef9c..b1e0082fc 100644
--- a/mpcontribs-api/src/mpcontribs_api/domains/consumers/service.py
+++ b/mpcontribs-api/src/mpcontribs_api/domains/consumers/service.py
@@ -1,3 +1,5 @@
+from typing import Any
+
 from mpcontribs_api.domains.consumers.models import (
     Consumer,
     ConsumerFilter,
@@ -31,15 +33,16 @@ async def read_many(
     ) -> Page[ConsumerOut]:
         return await self._consumer.read_many(filter=filter, pagination=pagination, fields=fields)
 
-    async def read_one(self, id: str, fields: frozenset[str] | None) -> ConsumerOut | None:
-        return await self._consumer.read_one(identifiers={"id": id}, fields=fields)
+    async def read_one(self, identifiers: dict[str, Any], fields: frozenset[str] | None) -> ConsumerOut | None:
+        """Read one override by its identity — the bare ``{"id": ...}`` or ``{"consumer_id": ...}``."""
+        return await self._consumer.read_one(identifiers=identifiers, fields=fields)
 
     async def insert_one(self, consumer: ConsumerIn) -> Consumer:
         document = self._consumer.document_model.from_input_model(consumer)
         return await self._consumer.insert_one(document)
 
-    async def update_one(self, id: str, update: ConsumerPatch) -> Consumer:
-        return await self._consumer.update_one(identifiers={"id": id}, update=update)
+    async def update_one(self, identifiers: dict[str, Any], update: ConsumerPatch) -> Consumer:
+        return await self._consumer.update_one(identifiers=identifiers, update=update)
 
-    async def delete_one(self, id: str) -> None:
-        await self._consumer.delete_one(identifiers={"id": id})
+    async def delete_one(self, identifiers: dict[str, Any]) -> None:
+        await self._consumer.delete_one(identifiers=identifiers)
diff --git a/mpcontribs-api/src/mpcontribs_api/domains/contributions/models.py b/mpcontribs-api/src/mpcontribs_api/domains/contributions/models.py
index b557693e9..1c9192824 100644
--- a/mpcontribs-api/src/mpcontribs_api/domains/contributions/models.py
+++ b/mpcontribs-api/src/mpcontribs_api/domains/contributions/models.py
@@ -1,4 +1,3 @@
-from dataclasses import dataclass
 from datetime import UTC, datetime
 from typing import Any, ClassVar
 
@@ -68,20 +67,19 @@ def extract_unique_value(data: dict[str, Any] | None, unique_column: str) -> Sca
     return value
 
 
-@dataclass(frozen=True, slots=True)
 class ContributionIdentity(Identity):
     """The full identity of a Contribution.
 
     Field declaration order IS the identity/index column order: ``index_model`` and ``projection``
-    iterate ``dataclasses.fields`` in that order, so the order is declared exactly once (below).
+    iterate ``model_fields`` in that order, so the order is declared exactly once (below).
     """
 
     # WARNING: the order the fields are specified in reflects their ordering for indices. Changing the order
     # creates index migration. Only change intentionally
     project: str
-    material_id: str | None
-    chemical_system_id: str
-    formula: str | None
+    material_id: MaterialId | None = None
+    chemical_system_id: ChemicalSystemId
+    formula: Formula | None = None
     unique_value: Scalar | None = None
     condition_key: str = ""
 
@@ -103,6 +101,18 @@ def check_hierarchy(material_id: str | None, chemical_system_id: str | None, for
                 material_id=material_id,
             )
 
+    @model_validator(mode="after")
+    def _check_identifier_hierarchy(self) -> ContributionIdentity:
+        """Enforce ``chemical_system_id`` > ``formula`` > ``material_id`` whenever an identity is built.
+
+        Bound directly as the ``/item`` query params, so every selector verb (read, patch, delete)
+        rejects a bad hierarchy with a 422 at the query boundary rather than relying on a downstream
+        ``read_one`` resolution to catch it. ``from_document`` uses ``model_construct`` and skips
+        validators, so trusted stored documents are unaffected.
+        """
+        ContributionIdentity.check_hierarchy(self.material_id, self.chemical_system_id, self.formula)
+        return self
+
 
 class ContributionBase(BaseModel):
     """Shared fields for Contribution, ContributionIn, and ContributionOut.
@@ -129,7 +139,8 @@ class Settings:
         name = "contributions"
         keep_nulls = False
         indexes = [
-            ContributionIdentity.index_model(),
+            # Keep the deployed index name — renaming forces a drop/recreate on a large collection.
+            ContributionIdentity.index_model(name="project_identity"),
             # Multikey indexes over each Link field's DBRef id so the component-delete
             # reference check (referenced_component_ids) is index-served, not a COLLSCAN.
             IndexModel(keys=[("structures.$id", ASCENDING)], name="ref_structures"),
@@ -137,15 +148,11 @@ class Settings:
             IndexModel(keys=[("attachments.$id", ASCENDING)], name="ref_attachments"),
         ]
 
-    @classmethod
-    def identifier_fields(cls) -> frozenset[str]:
-        """A contribution's natural key is its full :class:`ContributionIdentity` composite."""
-        return frozenset({"project", "material_id", "chemical_system_id", "formula", "unique_value", "condition_key"})
-
 
 class Contribution(ContributionBase, BaseDocumentWithInput[PydanticObjectId]):
     """Models what is actually stored in the database."""
 
+    identity_model: ClassVar[type[Identity]] = ContributionIdentity
     # Strict validation over stored data rather than coercsion
     data: ContributionStoredData
 
@@ -171,18 +178,6 @@ def from_input_model(cls, data: ContributionIn) -> Contribution:
     def set_last_modified(self):
         self.last_modified = datetime.now(UTC)
 
-    @property
-    def identity(self) -> ContributionIdentity:
-        """This document's identity, read straight off its own stored fields."""
-        return ContributionIdentity(
-            project=self.project,
-            material_id=self.material_id,
-            chemical_system_id=self.chemical_system_id,
-            formula=self.formula,
-            unique_value=self.unique_value,
-            condition_key=self.condition_key,
-        )
-
 
 class ContributionIn(ContributionBase):
     """Fields that users are allowed to submit when adding a Contribution.
diff --git a/mpcontribs-api/src/mpcontribs_api/domains/contributions/repository.py b/mpcontribs-api/src/mpcontribs_api/domains/contributions/repository.py
index 3f146754b..b25c61b03 100644
--- a/mpcontribs-api/src/mpcontribs_api/domains/contributions/repository.py
+++ b/mpcontribs-api/src/mpcontribs_api/domains/contributions/repository.py
@@ -1,4 +1,4 @@
-from typing import Any, cast
+from typing import Any
 
 from beanie import PydanticObjectId, UpdateResponse
 from beanie.operators import Set
@@ -259,51 +259,64 @@ async def aggregate_project_stats(self, project_id: str) -> ProjectAggregate:
         agg.columns = finalize_columns(acc)
         return agg
 
-    async def upsert_one(  # pyright: ignore[reportIncompatibleMethodOverride]
+    async def read_one(
         self,
         identifiers: dict[str, Any],
-        contribution: ContributionIn,
-        unique_value: Scalar | None = _UNSET,
+        fields: frozenset[str] | None = None,
         session: AsyncClientSession | None = None,
-    ) -> Contribution:
-        """Atomically upsert a single Contribution, keyed by the shape of ``identifiers``.
+    ) -> ContributionOut | None:
+        """Read one contribution by its Mongo ``_id`` or by a (possibly partial) natural identity.
+
+        Prefers the id: a bare ``{"id": ...}`` takes the exact, index-covered base path. Any other
+        key set is treated as a natural identity — only part of :class:`ContributionIdentity` is
+        user-suppliable over HTTP (``project``, ``material_id``, ``chemical_system_id``, ``formula``,
+        and optional ``unique_value``; ``condition_key`` is server-owned), so that subset is not
+        covered by the unique index. It builds its own scoped match (deliberately bypassing
+        :meth:`_identifier_query`, which requires the exact key set), validates the hierarchy, and
+        rejects an ambiguous match rather than silently returning the first document.
 
-        Relies on the unique index over the identity fields as the concurrency tiebreaker.
-        On insert a fresh document is written with ``is_public=False``.
+        Args:
+            identifiers: ``{"id": ...}`` for the exact key, or identity field values (a subset of
+                ``ContributionIdentity``); ``None`` values match null-or-absent stored fields
+                (``keep_nulls=False``)
+            fields: fields to project; if None the full document is returned
+            session: optional client session for transactions
+
+        Raises:
+            ConflictError: if more than one in-scope contribution matches a partial identity
         """
-        identifiers = self.coerce_identifiers(identifiers)
         if identifiers.keys() == {"id"}:
-            return await self._upsert_by_id(
-                identifiers["id"], contribution, None if unique_value is _UNSET else unique_value
-            )
-
-        doc = self.document_model.from_input_model(contribution)
-        doc.unique_value = identifiers["unique_value"]
-        doc.condition_key = identifiers["condition_key"]
-        update_data = doc.model_dump(exclude={"id"}, exclude_none=True)
-        query = self.document_model.find_one(
-            self._scope,
-            self.document_model.project == identifiers["project"],
-            self.document_model.material_id == identifiers["material_id"],
-            self.document_model.chemical_system_id == identifiers["chemical_system_id"],
-            self.document_model.formula == identifiers["formula"],
-            self.document_model.unique_value == identifiers["unique_value"],
-            self.document_model.condition_key == identifiers["condition_key"],
-        ).upsert(
-            Set(update_data),
-            on_insert=doc,
-            response_type=UpdateResponse.NEW_DOCUMENT,
+            return await super().read_one(identifiers, fields, session=session)
+        ContributionIdentity.check_hierarchy(
+            identifiers.get("material_id"), identifiers.get("chemical_system_id"), identifiers.get("formula")
         )
-        result = await query  # pyright: ignore[reportGeneralTypeIssues] # beanie UpdateQuery is awaitable, but pyright doesn't see it
-        return cast(Contribution, result)  # upsert always returns the resulting document
+        match: dict[str, Any] = dict(identifiers)
+        if self._scope:
+            match = {"$and": [self._scope, match]}
+        projection = self.out_model.projection(fields)
+        # Fetch up to two so an ambiguous key is detected without scanning the whole match.
+        docs = await self.document_model.find(match).limit(2).project(projection).to_list()  # pyright: ignore[reportArgumentType]
+        if len(docs) > 1:
+            raise ConflictError(
+                "identifiers match more than one contribution; supply unique_value to disambiguate",
+                identifiers=identifiers,
+            )
+        return docs[0] if docs else None
 
-    async def _upsert_by_id(
+    async def upsert_by_id(
         self,
         id: str,
         contribution: ContributionIn,
         unique_value: Scalar | None = None,
     ) -> Contribution:
-        """Upsert a single Contribution keyed on its Mongo ``_id`` (see :meth:`upsert_one`)."""
+        """Upsert a single Contribution keyed on its Mongo ``_id``.
+
+        Distinct from the base identity-keyed :meth:`upsert_one`: a PUT addresses a specific existing
+        document by ``_id`` and may *change* its identity fields, so the match must stay on ``_id``.
+        Non-null fields merge into the stored document (so component links absent from the payload
+        survive), while the server-owned ``unique_value`` is written explicitly — even when ``None`` —
+        so the identity index tracks the new data. A resulting duplicate identity surfaces as a 409.
+        """
         oid = self._convert_object_id(id)
         doc = self.document_model.from_input_model(contribution)
         # from_input_model mints a fresh id; upsert-by-id must key on the caller-supplied id so the
diff --git a/mpcontribs-api/src/mpcontribs_api/domains/contributions/router.py b/mpcontribs-api/src/mpcontribs_api/domains/contributions/router.py
index 9dca54cd1..8659ea5e7 100644
--- a/mpcontribs-api/src/mpcontribs_api/domains/contributions/router.py
+++ b/mpcontribs-api/src/mpcontribs_api/domains/contributions/router.py
@@ -18,6 +18,7 @@
 from mpcontribs_api.domains.contributions.models import (
     Contribution,
     ContributionFilter,
+    ContributionIdentity,
     ContributionIn,
     ContributionOut,
     ContributionPatch,
@@ -133,6 +134,38 @@ async def download_contributions(
     )
 
 
+# Declared before the ``/{id}`` routes so the literal ``item`` is never captured as an id.
+@router.get("/item")
+async def read_one_by_identity(
+    service: ContributionServiceDep,
+    identity: Annotated[ContributionIdentity, Depends()],
+    fields: FieldSelector = None,
+):
+    """Return the single contribution addressed by its natural identity (409 if ambiguous)."""
+    selected = ContributionOut.parse_fields(fields)
+    return await service.read_one(identity.as_dict(), fields=selected)
+
+
+@router.delete("/item", dependencies=[Depends(require_user)])
+async def delete_one_by_identity(
+    service: ContributionServiceDep,
+    identity: Annotated[ContributionIdentity, Depends()],
+) -> BulkDeleteSummary:
+    """Delete the single contribution addressed by its natural identity, cascading to components."""
+    return await service.delete_one(identity.as_dict())
+
+
+@router.patch("/item", dependencies=[Depends(require_user)])
+async def update_one_by_identity(
+    service: ContributionServiceDep,
+    update: ContributionPatch,
+    identity: Annotated[ContributionIdentity, Depends()],
+    replace_data: bool = False,
+):
+    """Patch the single contribution addressed by its natural identity (409 if ambiguous)."""
+    return await service.update_one(identity.as_dict(), update=update, replace_data=replace_data)
+
+
 @router.delete("/{id}", dependencies=[Depends(require_user)])
 async def delete_one(
     service: ContributionServiceDep,
diff --git a/mpcontribs-api/src/mpcontribs_api/domains/contributions/service.py b/mpcontribs-api/src/mpcontribs_api/domains/contributions/service.py
index a3bea70d7..fa36f72ab 100644
--- a/mpcontribs-api/src/mpcontribs_api/domains/contributions/service.py
+++ b/mpcontribs-api/src/mpcontribs_api/domains/contributions/service.py
@@ -685,9 +685,14 @@ async def upsert_many(self, contributions: list[ContributionIn]) -> BulkWriteSum
         async def _bounded_upsert(item: PreparedWrite) -> Contribution | BulkFailure:
             contrib = item.contribution
             identifiers = contrib.identity_dict(item.unique_value, item.condition_key)
+            # Build the full document with its server-resolved identity parts stamped on, then upsert
+            # keyed on that identity (the repository reads it off ``document.identity()``).
+            doc = self._contributions.document_model.from_input_model(contrib)
+            doc.unique_value = item.unique_value
+            doc.condition_key = item.condition_key
             async with sem:
                 try:
-                    return await self._contributions.upsert_one(identifiers, contrib)
+                    return await self._contributions.upsert_one(doc)
                 except Exception as exc:
                     logger.error("upsert_contribution_failed", index=item.index, identifier=identifiers, exc_info=True)
                     return bulk_failure_from_exception(item.index, identifiers, exc)
@@ -738,7 +743,7 @@ async def update_many(
             else filter.model_copy(update={"project__in": sorted(self._user.writable(*PROJECT_PATH))})
         )
 
-        touches_identity = bool(ContributionIdentity.model_fields() & fields.keys()) or "data" in fields
+        touches_identity = bool(ContributionIdentity.model_fields.keys() & fields.keys()) or "data" in fields
         if not touches_identity:
             # No identity/unique_value recompute needed, so a uniform $set is safe.
             summary = await self._contributions.update_many(filter, fields)
@@ -811,7 +816,7 @@ async def upsert_one(self, identifiers: dict[str, Any], contribution: Contributi
                     max_allowed=cap,
                 )
         unique_value = await self._resolve_unique_value(contribution.project, contribution.data)
-        return await self._contributions.upsert_one(identifiers, contribution, unique_value)
+        return await self._contributions.upsert_by_id(identifiers["id"], contribution, unique_value)
 
     async def update_one(
         self, identifiers: dict[str, Any], update: ContributionPatch, *, replace_data: bool = False
@@ -830,6 +835,11 @@ async def update_one(
         re-validated strictly (the permissive patch validator allows leaf fragments a full doc may not).
         ``unique_value`` is resolved against the same post-write view the repository will persist.
         """
+        if identifiers.keys() != {"id"}:
+            existing = await self._contributions.read_one(identifiers, frozenset({"id"}))
+            if existing is None:
+                raise NotFoundError("contribution not found", identifiers=identifiers)
+            identifiers = {"id": str(existing.id)}
         if not self._user.is_admin(*ROOT_PATH):
             target = await self._contributions.read_one(identifiers, frozenset({"id", "project"}))
             if target is None:
diff --git a/mpcontribs-api/src/mpcontribs_api/domains/initiatives/models.py b/mpcontribs-api/src/mpcontribs_api/domains/initiatives/models.py
index 3f5a3db92..30c14504e 100644
--- a/mpcontribs-api/src/mpcontribs_api/domains/initiatives/models.py
+++ b/mpcontribs-api/src/mpcontribs_api/domains/initiatives/models.py
@@ -1,4 +1,4 @@
-from typing import Self
+from typing import ClassVar, Self
 
 from beanie import PydanticObjectId
 from bson.errors import InvalidId
@@ -7,11 +7,17 @@
 
 from mpcontribs_api.domains._shared.filters import BaseFilter
 from mpcontribs_api.domains._shared.models import BaseDocumentWithInput, DocumentOut
-from mpcontribs_api.domains._shared.types import NFKCStr, PrefixedEmail, Slug
+from mpcontribs_api.domains._shared.types import Identity, NFKCStr, PrefixedEmail, Slug
 from mpcontribs_api.exceptions import ValidationError
 from mpcontribs_api.projection import SparseFieldsModel
 
 
+class InitiativeIdentity(Identity):
+    """An initiative's identity: its globally-unique ``slug``."""
+
+    slug: Slug
+
+
 class Initiative(BaseDocumentWithInput[PydanticObjectId]):
     """A canonical, authoritative grouping of projects into a larger organizational effort.
 
@@ -25,6 +31,7 @@ class Initiative(BaseDocumentWithInput[PydanticObjectId]):
     the ``initiative:<slug>`` role.
     """
 
+    identity_model: ClassVar[type[Identity]] = InitiativeIdentity
     slug: Slug
     name: NFKCStr = Field(max_length=100)
     owner: PrefixedEmail
@@ -35,7 +42,7 @@ class Settings:
         name = "initiatives"
         keep_nulls = False
         indexes = [
-            IndexModel(keys=[("slug", ASCENDING)], name="slug", unique=True),
+            InitiativeIdentity.index_model(name="slug"),
             IndexModel(
                 keys=[("owner", ASCENDING), ("is_approved", ASCENDING), ("is_public", ASCENDING)],
                 name="owner_is_approved_is_public",
@@ -47,11 +54,6 @@ class Settings:
     def from_input_model(cls, data: InitiativeIn, owner: PrefixedEmail) -> Self:  # pyright: ignore[reportIncompatibleMethodOverride]
         return cls(_id=PydanticObjectId(), **data.model_dump(), owner=owner)
 
-    @classmethod
-    def identifier_fields(cls) -> frozenset[str]:
-        """An ``Initiative`` is uniquely identified by its globally-unique ``slug``."""
-        return frozenset({"slug"})
-
     @model_validator(mode="after")
     def _public_requires_approved(self) -> Self:
         """An initiative cannot be public until it has been approved."""
diff --git a/mpcontribs-api/src/mpcontribs_api/domains/initiatives/router.py b/mpcontribs-api/src/mpcontribs_api/domains/initiatives/router.py
index f8b4298c2..0b8a515ce 100644
--- a/mpcontribs-api/src/mpcontribs_api/domains/initiatives/router.py
+++ b/mpcontribs-api/src/mpcontribs_api/domains/initiatives/router.py
@@ -8,6 +8,7 @@
 from mpcontribs_api.domains.initiatives.dependencies import InitiativeServiceDep
 from mpcontribs_api.domains.initiatives.models import (
     InitiativeFilter,
+    InitiativeIdentity,
     InitiativeIn,
     InitiativeOut,
     InitiativePatch,
@@ -29,6 +30,37 @@ async def read_many(
     return await service.read_many(pagination=pagination, filter=filter, fields=selected)
 
 
+@router.get("/item")
+async def read_one_by_identity(
+    service: InitiativeServiceDep,
+    identity: Annotated[InitiativeIdentity, Depends()],
+    fields: FieldSelector = None,
+):
+    """Return the single initiative by its natural key ``slug`` (the uniform ``/item`` entrypoint)."""
+    selected = InitiativeOut.parse_fields(fields)
+    return await service.read_one(identity.as_dict(), fields=selected)
+
+
+@router.patch("/item", response_model=InitiativeOut, dependencies=[Depends(require_user)])
+async def update_one_by_identity(
+    service: InitiativeServiceDep,
+    identity: Annotated[InitiativeIdentity, Depends()],
+    update: InitiativePatch,
+):
+    """Partially update the initiative by its natural key ``slug`` (the uniform ``/item`` entrypoint)."""
+    return await service.update_one(identity.as_dict(), update=update)
+
+
+@router.delete("/item", status_code=status.HTTP_204_NO_CONTENT, dependencies=[Depends(require_user)])
+async def delete_one_by_identity(
+    service: InitiativeServiceDep,
+    identity: Annotated[InitiativeIdentity, Depends()],
+):
+    """Delete the initiative by its natural key ``slug`` (the uniform ``/item`` entrypoint)."""
+    await service.delete_one(identity.as_dict())
+    return Response(status_code=status.HTTP_204_NO_CONTENT)
+
+
 @router.get("/{slug}")
 async def read_one(
     service: InitiativeServiceDep,
diff --git a/mpcontribs-api/src/mpcontribs_api/domains/initiatives/service.py b/mpcontribs-api/src/mpcontribs_api/domains/initiatives/service.py
index 5bfaf4cb0..27af83731 100644
--- a/mpcontribs-api/src/mpcontribs_api/domains/initiatives/service.py
+++ b/mpcontribs-api/src/mpcontribs_api/domains/initiatives/service.py
@@ -76,7 +76,7 @@ async def insert_one(self, data: InitiativeIn) -> Initiative:
                 )
 
         # The repository translates the unique-slug DuplicateKeyError into a ConflictError whose
-        # context carries the slug (Initiative.identifier_fields() == {"slug"}).
+        # context carries the slug (Initiative.identity_model.model_fields == {"slug"}).
         initiative = self._initiatives.document_model.from_input_model(data, owner=self._user.username)
         return await self._initiatives.insert_one(initiative)
 
diff --git a/mpcontribs-api/src/mpcontribs_api/domains/project_groups/models.py b/mpcontribs-api/src/mpcontribs_api/domains/project_groups/models.py
index f3ba3379f..3fcd9a8bc 100644
--- a/mpcontribs-api/src/mpcontribs_api/domains/project_groups/models.py
+++ b/mpcontribs-api/src/mpcontribs_api/domains/project_groups/models.py
@@ -1,3 +1,5 @@
+from typing import ClassVar
+
 from beanie import Link, PydanticObjectId
 from bson import DBRef
 from bson.errors import InvalidId
@@ -6,13 +8,21 @@
 
 from mpcontribs_api.domains._shared.filters import BaseFilter
 from mpcontribs_api.domains._shared.models import BaseDocumentWithInput, DocumentOut
-from mpcontribs_api.domains._shared.types import PrefixedEmail, SearchStr, ShortStr
+from mpcontribs_api.domains._shared.types import Identity, PrefixedEmail, SearchStr, ShortStr
 from mpcontribs_api.domains.projects.models import Project
 from mpcontribs_api.exceptions import ValidationError
 from mpcontribs_api.projection import SparseFieldsModel
 
 
+class ProjectGroupIdentity(Identity):
+    """A project group's identity: its ``name`` scoped to its ``owner`` (declaration order = index order)."""
+
+    name: SearchStr
+    owner: PrefixedEmail
+
+
 class ProjectGroup(BaseDocumentWithInput[PydanticObjectId]):
+    identity_model: ClassVar[type[Identity]] = ProjectGroupIdentity
     name: SearchStr = Field(max_length=50)
     owner: PrefixedEmail
     description: str = Field(max_length=100)
@@ -22,11 +32,7 @@ class ProjectGroup(BaseDocumentWithInput[PydanticObjectId]):
     class Settings:
         name = "project_groups"
         indexes = [
-            IndexModel(
-                keys=[("name", ASCENDING), ("owner", ASCENDING)],
-                name="name_owner",
-                unique=True,
-            ),
+            ProjectGroupIdentity.index_model(name="name_owner"),
             IndexModel(
                 keys=[("name", ASCENDING), ("owner", ASCENDING), ("is_public", ASCENDING)],
                 name="name_owner_is_public",
@@ -34,11 +40,6 @@ class Settings:
         ]
         validate_on_save = True
 
-    @classmethod
-    def identifier_fields(cls) -> frozenset[str]:
-        """A ``ProjectGroup`` is uniquely identified by its ``name`` + ``owner``."""
-        return frozenset({"name", "owner"})
-
     @classmethod
     def from_input_model(cls, data: ProjectGroupIn) -> ProjectGroup:
         """Build a stored group from input, assigning a fresh ``_id`` and resolving member ids to links"""
diff --git a/mpcontribs-api/src/mpcontribs_api/domains/project_groups/router.py b/mpcontribs-api/src/mpcontribs_api/domains/project_groups/router.py
index 0689c26a8..d112023a0 100644
--- a/mpcontribs-api/src/mpcontribs_api/domains/project_groups/router.py
+++ b/mpcontribs-api/src/mpcontribs_api/domains/project_groups/router.py
@@ -6,10 +6,11 @@
 from mpcontribs_api.dependencies import require_user
 from mpcontribs_api.domains._shared.bulk import BulkWriteSummary
 from mpcontribs_api.domains._shared.models import DeleteResponse
-from mpcontribs_api.domains._shared.types import FieldSelector, PrefixedEmail, SearchStr
+from mpcontribs_api.domains._shared.types import FieldSelector
 from mpcontribs_api.domains.project_groups.dependencies import ProjectGroupServiceDep
 from mpcontribs_api.domains.project_groups.models import (
     ProjectGroupFilter,
+    ProjectGroupIdentity,
     ProjectGroupIn,
     ProjectGroupOut,
     ProjectGroupPatch,
@@ -40,22 +41,20 @@ async def read_many(
 
 
 @router.get("/item")
-async def read_one(
+async def read_one_by_identity(
     service: ProjectGroupServiceDep,
-    name: SearchStr,
-    owner: PrefixedEmail,
+    identity: Annotated[ProjectGroupIdentity, Depends()],
     fields: FieldSelector = None,
 ):
-    """Return the single project group identified by ``name`` + ``owner``.
+    """Return the single project group identified by its ``name`` + ``owner`` natural key.
 
     Args:
         service (ProjectGroupServiceDep): the project group service we depend on
-        name (SearchStr): the project group's name
-        owner (PrefixedEmail): the project group's owner
+        identity (ProjectGroupIdentity): the group's natural key (``name`` + ``owner``) query params
         fields (FieldSelector): the fields to return to a user
     """
     selected = ProjectGroupOut.parse_fields(fields)
-    return await service.read_one({"name": name, "owner": owner}, fields=selected)
+    return await service.read_one(identity.as_dict(), fields=selected)
 
 
 @router.post(
@@ -78,39 +77,35 @@ async def insert_one(
 
 
 @router.patch("/item", response_model=ProjectGroupOut, dependencies=[Depends(require_user)])
-async def update_one(
+async def update_one_by_identity(
     service: ProjectGroupServiceDep,
-    name: SearchStr,
-    owner: PrefixedEmail,
+    identity: Annotated[ProjectGroupIdentity, Depends()],
     update: ProjectGroupPatch,
 ):
-    """Partially update the project group identified by ``name`` + ``owner``.
+    """Partially update the project group identified by its ``name`` + ``owner`` natural key.
 
     Args:
         service (ProjectGroupServiceDep): the project group service we depend on
-        name (SearchStr): the project group's name
-        owner (PrefixedEmail): the project group's owner
+        identity (ProjectGroupIdentity): the group's natural key (``name`` + ``owner``) query params
         update (ProjectGroupPatch): the partial update to apply - unset fields are dropped
     """
-    return await service.update_one({"name": name, "owner": owner}, update=update)
+    return await service.update_one(identity.as_dict(), update=update)
 
 
 @router.delete("/item", status_code=status.HTTP_204_NO_CONTENT, dependencies=[Depends(require_user)])
-async def delete_one(
+async def delete_one_by_identity(
     service: ProjectGroupServiceDep,
-    name: SearchStr,
-    owner: PrefixedEmail,
+    identity: Annotated[ProjectGroupIdentity, Depends()],
 ):
-    """Delete the single project group identified by ``name`` + ``owner``.
+    """Delete the single project group identified by its ``name`` + ``owner`` natural key.
 
     Raises 404 if no such group is visible to the caller, 409 if the identifiers are ambiguous.
 
     Args:
         service (ProjectGroupServiceDep): the project group service we depend on
-        name (SearchStr): the project group's name
-        owner (PrefixedEmail): the project group's owner
+        identity (ProjectGroupIdentity): the group's natural key (``name`` + ``owner``) query params
     """
-    await service.delete_one({"name": name, "owner": owner})
+    await service.delete_one(identity.as_dict())
     return Response(status_code=status.HTTP_204_NO_CONTENT)
 
 
@@ -129,32 +124,30 @@ async def delete_many(
 
 
 @router.post("/item/projects", response_model=BulkWriteSummary[str], dependencies=[Depends(require_user)])
-async def add_projects_by_identifiers(
+async def add_projects_by_identity(
     service: ProjectGroupServiceDep,
-    name: SearchStr,
-    owner: PrefixedEmail,
+    identity: Annotated[ProjectGroupIdentity, Depends()],
     body: ProjectRefs,
 ):
-    """Add projects to the group identified by ``name`` + ``owner``.
+    """Add projects to the group identified by its ``name`` + ``owner`` natural key.
 
     Each project is verified against the projects collection (scoped to the caller); unknown or
     invisible projects are reported per-item in the response rather than failing the whole request.
     """
-    return await service.add_projects({"name": name, "owner": owner}, body.project_ids)
+    return await service.add_projects(identity.as_dict(), body.project_ids)
 
 
 @router.delete("/item/projects", response_model=BulkWriteSummary[str], dependencies=[Depends(require_user)])
-async def delete_projects_by_identifiers(
+async def delete_projects_by_identity(
     service: ProjectGroupServiceDep,
-    name: SearchStr,
-    owner: PrefixedEmail,
+    identity: Annotated[ProjectGroupIdentity, Depends()],
     body: ProjectRefs,
 ):
-    """Delete projects from the group identified by ``name`` + ``owner``.
+    """Delete projects from the group identified by its ``name`` + ``owner`` natural key.
 
     Ids that are not members of the group are reported per-item in the response.
     """
-    return await service.delete_projects({"name": name, "owner": owner}, body.project_ids)
+    return await service.delete_projects(identity.as_dict(), body.project_ids)
 
 
 @router.post("/{id}/projects", response_model=BulkWriteSummary[str], dependencies=[Depends(require_user)])
@@ -175,3 +168,36 @@ async def delete_projects_by_id(
 ):
     """Delete projects from the group identified by ``id``. See ``delete_projects``."""
     return await service.delete_projects({"id": id}, body.project_ids)
+
+
+# Primary-key CRUD, symmetric to the ``/item`` (name+owner) routes above. Declared after ``/item`` so
+# the literal path is never captured as an ``{id}``.
+@router.get("/{id}")
+async def read_one(
+    service: ProjectGroupServiceDep,
+    id: str,
+    fields: FieldSelector = None,
+):
+    """Return the single project group identified by its ``_id``."""
+    selected = ProjectGroupOut.parse_fields(fields)
+    return await service.read_one({"id": id}, fields=selected)
+
+
+@router.patch("/{id}", response_model=ProjectGroupOut, dependencies=[Depends(require_user)])
+async def update_one(
+    service: ProjectGroupServiceDep,
+    id: str,
+    update: ProjectGroupPatch,
+):
+    """Partially update the project group identified by its ``_id``."""
+    return await service.update_one({"id": id}, update=update)
+
+
+@router.delete("/{id}", status_code=status.HTTP_204_NO_CONTENT, dependencies=[Depends(require_user)])
+async def delete_one(
+    service: ProjectGroupServiceDep,
+    id: str,
+):
+    """Delete the project group identified by its ``_id``. Restricted to its owner or an admin."""
+    await service.delete_one({"id": id})
+    return Response(status_code=status.HTTP_204_NO_CONTENT)
diff --git a/mpcontribs-api/src/mpcontribs_api/domains/projects/models.py b/mpcontribs-api/src/mpcontribs_api/domains/projects/models.py
index 6fb52ac9c..99a9e63f2 100644
--- a/mpcontribs-api/src/mpcontribs_api/domains/projects/models.py
+++ b/mpcontribs-api/src/mpcontribs_api/domains/projects/models.py
@@ -1,5 +1,5 @@
 from enum import StrEnum
-from typing import Any, Literal
+from typing import Any, ClassVar, Literal
 
 from beanie import Link
 from pydantic import BaseModel, ConfigDict, Field, HttpUrl, field_validator
@@ -7,11 +7,22 @@
 from mpcontribs_api import pagination
 from mpcontribs_api.domains._shared.filters import BaseFilter
 from mpcontribs_api.domains._shared.models import BaseDocumentWithInput, DocumentOut
-from mpcontribs_api.domains._shared.types import PrefixedEmail, SearchStr, ShortStr
+from mpcontribs_api.domains._shared.types import Identity, PrefixedEmail, SearchStr, ShortStr
 from mpcontribs_api.domains.initiatives.models import Initiative
 from mpcontribs_api.exceptions import ValidationError
 
 
+class ProjectIdentity(Identity):
+    """A project's identity: its human-chosen short name, which is also its Mongo ``_id``.
+
+    Because the identity is the primary key, no separate unique index is declared — Mongo's implicit
+    ``_id`` index enforces it. (``index_model`` must not be added to ``Settings``: it would key on a
+    literal ``"id"`` field that does not exist in the stored document.)
+    """
+
+    id: ShortStr
+
+
 def _validate_unique_column(value: str | None) -> str | None:
     """Shape-only check for ``unique_column``: a non-empty, non-blank dotted-path string or None."""
     if value is None:
@@ -136,6 +147,7 @@ class Settings:
 class Project(ProjectBase, BaseDocumentWithInput[ShortStr]):
     """Document model of what is actually stored."""
 
+    identity_model: ClassVar[type[Identity]] = ProjectIdentity
     # Server-owned: derived from the project's contributions
     stats: Stats = Field(default_factory=Stats)
     columns: list[Column] = Field(default_factory=list)
diff --git a/mpcontribs-api/src/mpcontribs_api/domains/projects/service.py b/mpcontribs-api/src/mpcontribs_api/domains/projects/service.py
index c2aed64f1..294e89eff 100644
--- a/mpcontribs-api/src/mpcontribs_api/domains/projects/service.py
+++ b/mpcontribs-api/src/mpcontribs_api/domains/projects/service.py
@@ -87,7 +87,7 @@ async def upsert_one(self, identifiers: dict[str, Any], data: ProjectIn) -> Proj
 
         if project.is_public and not project.is_approved:
             raise ValidationError("a project cannot be public until it is approved", id=id)
-        return await self._projects.upsert_one(project)
+        return await self._projects.replace_one(id, project)
 
     async def update_one(self, identifiers: dict[str, Any], update: ProjectPatch) -> Project:
         """Apply a project patch, enforcing approval rules and routing ``initiative`` changes.
diff --git a/mpcontribs-api/src/mpcontribs_api/domains/structures/models.py b/mpcontribs-api/src/mpcontribs_api/domains/structures/models.py
index 916ae65d2..fb95c7c0b 100644
--- a/mpcontribs-api/src/mpcontribs_api/domains/structures/models.py
+++ b/mpcontribs-api/src/mpcontribs_api/domains/structures/models.py
@@ -5,7 +5,7 @@
 from pymatgen.core import Element
 
 from mpcontribs_api.domains._shared.filters import BaseFilter
-from mpcontribs_api.domains._shared.models import Component, ComponentIn, DocumentOut
+from mpcontribs_api.domains._shared.models import Component, ComponentIdentity, ComponentIn, DocumentOut
 from mpcontribs_api.domains._shared.types import MD5Hash, NFKCStr
 from mpcontribs_api.exceptions import ValidationError
 from mpcontribs_api.projection import SparseFieldsModel
@@ -99,6 +99,7 @@ class Structure(Component):
 
     class Settings:
         name = "structures"
+        indexes = [ComponentIdentity.index_model(name="md5")]
 
 
 class StructureIn(ComponentIn):
diff --git a/mpcontribs-api/src/mpcontribs_api/domains/structures/router.py b/mpcontribs-api/src/mpcontribs_api/domains/structures/router.py
index dde1c623b..b8fb1ca80 100644
--- a/mpcontribs-api/src/mpcontribs_api/domains/structures/router.py
+++ b/mpcontribs-api/src/mpcontribs_api/domains/structures/router.py
@@ -6,7 +6,7 @@
 
 from mpcontribs_api.dependencies import S3Dep, require_user, require_writer
 from mpcontribs_api.domains._shared.bulk import BulkWriteSummary
-from mpcontribs_api.domains._shared.models import ComponentDeleteResponse
+from mpcontribs_api.domains._shared.models import ComponentDeleteResponse, ComponentIdentity
 from mpcontribs_api.domains._shared.types import (
     DownloadFormat,
     FieldSelector,
@@ -31,13 +31,40 @@ async def read_many(
     return await service.read_many(filter=filter, fields=selected, pagination=pagination)
 
 
+@router.get("/item")
+async def read_one_by_identity(
+    service: StructureServiceDep,
+    identity: Annotated[ComponentIdentity, Depends()],
+    fields: FieldSelector = None,
+):
+    """Return a single structure addressed by its content ``md5`` (its natural key)."""
+    selected = StructureOut.parse_fields(fields)
+    return await service.read_one(identifiers=identity.as_dict(), fields=selected)
+
+
+@router.delete("/item", response_model=ComponentDeleteResponse, dependencies=[Depends(require_user)])
+async def delete_one_by_identity(service: StructureServiceDep, identity: Annotated[ComponentIdentity, Depends()]):
+    """Delete a single structure addressed by its content ``md5`` (its natural key)."""
+    return await service.delete_one(identifiers=identity.as_dict())
+
+
+@router.patch("/item", dependencies=[Depends(require_user)])
+async def update_one_by_identity(
+    service: StructureServiceDep,
+    identity: Annotated[ComponentIdentity, Depends()],
+    update: StructurePatch,
+):
+    """Patch a single structure addressed by its content ``md5`` (its natural key)."""
+    return await service.update_one(identifiers=identity.as_dict(), update=update)
+
+
 @router.get("/{id}")
 async def read_one(
     service: StructureServiceDep,
     id: str,
     fields: FieldSelector = None,
 ):
-    """Return a single structure addressed by its ``_id`` or its content ``md5``."""
+    """Return a single structure addressed by its ``_id``."""
     selected = StructureOut.parse_fields(fields)
     return await service.read_one(identifiers={"id": id}, fields=selected)
 
@@ -84,7 +111,7 @@ async def delete_many(service: StructureServiceDep, filter: StructureFilter = Fi
 
 @router.delete("/{id}", response_model=ComponentDeleteResponse, dependencies=[Depends(require_user)])
 async def delete_one(service: StructureServiceDep, id: str):
-    """Delete a single structure addressed by its ``_id`` or its content ``md5``."""
+    """Delete a single structure addressed by its ``_id``."""
     return await service.delete_one(identifiers={"id": id})
 
 
@@ -94,5 +121,5 @@ async def update_one(
     id: str,
     update: StructurePatch,
 ):
-    """Patch a single structure addressed by its ``_id`` or its content ``md5``."""
+    """Patch a single structure addressed by its ``_id``."""
     return await service.update_one(identifiers={"id": id}, update=update)
diff --git a/mpcontribs-api/src/mpcontribs_api/domains/tables/models.py b/mpcontribs-api/src/mpcontribs_api/domains/tables/models.py
index c2a6e22ca..5effdb27f 100644
--- a/mpcontribs-api/src/mpcontribs_api/domains/tables/models.py
+++ b/mpcontribs-api/src/mpcontribs_api/domains/tables/models.py
@@ -11,7 +11,7 @@
 )
 
 from mpcontribs_api.domains._shared.filters import BaseFilter
-from mpcontribs_api.domains._shared.models import Component, ComponentIn, DocumentOut
+from mpcontribs_api.domains._shared.models import Component, ComponentIdentity, ComponentIn, DocumentOut
 from mpcontribs_api.domains._shared.types import DisplayStr, MD5Hash, NFKCStr, PolarsFrame, nfc_normalize
 from mpcontribs_api.projection import SparseFieldsModel
 
@@ -71,6 +71,7 @@ class Table(Component):
 
     class Settings:
         name = "tables"
+        indexes = [ComponentIdentity.index_model(name="md5")]
 
     @classmethod
     def from_input(cls, input: TableIn) -> Self:  # pyright: ignore[reportIncompatibleMethodOverride]
diff --git a/mpcontribs-api/src/mpcontribs_api/domains/tables/router.py b/mpcontribs-api/src/mpcontribs_api/domains/tables/router.py
index 6a594634f..0fce7c52f 100644
--- a/mpcontribs-api/src/mpcontribs_api/domains/tables/router.py
+++ b/mpcontribs-api/src/mpcontribs_api/domains/tables/router.py
@@ -6,7 +6,7 @@
 
 from mpcontribs_api.dependencies import S3Dep, require_user, require_writer
 from mpcontribs_api.domains._shared.bulk import BulkWriteSummary
-from mpcontribs_api.domains._shared.models import ComponentDeleteResponse
+from mpcontribs_api.domains._shared.models import ComponentDeleteResponse, ComponentIdentity
 from mpcontribs_api.domains._shared.types import (
     DownloadFormat,
     FieldSelector,
@@ -31,13 +31,40 @@ async def read_many(
     return await service.read_many(filter=filter, fields=selected, pagination=pagination)
 
 
+@router.get("/item")
+async def read_one_by_identity(
+    service: TableServiceDep,
+    identity: Annotated[ComponentIdentity, Depends()],
+    fields: FieldSelector = None,
+):
+    """Return a single table addressed by its content ``md5`` (its natural key)."""
+    selected = TableOut.parse_fields(fields)
+    return await service.read_one(identifiers=identity.as_dict(), fields=selected)
+
+
+@router.delete("/item", response_model=ComponentDeleteResponse, dependencies=[Depends(require_user)])
+async def delete_one_by_identity(service: TableServiceDep, identity: Annotated[ComponentIdentity, Depends()]):
+    """Delete a single table addressed by its content ``md5`` (its natural key)."""
+    return await service.delete_one(identifiers=identity.as_dict())
+
+
+@router.patch("/item", dependencies=[Depends(require_user)])
+async def update_one_by_identity(
+    service: TableServiceDep,
+    identity: Annotated[ComponentIdentity, Depends()],
+    update: TablePatch,
+):
+    """Patch a single table addressed by its content ``md5`` (its natural key)."""
+    return await service.update_one(identifiers=identity.as_dict(), update=update)
+
+
 @router.get("/{id}")
 async def read_one(
     service: TableServiceDep,
     id: str,
     fields: FieldSelector = None,
 ):
-    """Return a single table addressed by its ``_id`` or its content ``md5``."""
+    """Return a single table addressed by its ``_id``."""
     selected = TableOut.parse_fields(fields)
     return await service.read_one(identifiers={"id": id}, fields=selected)
 
diff --git a/mpcontribs-api/tests/integration/db/test_contributions_repository.py b/mpcontribs-api/tests/integration/db/test_contributions_repository.py
index 14cd3a8b8..8fd5bc1bd 100644
--- a/mpcontribs-api/tests/integration/db/test_contributions_repository.py
+++ b/mpcontribs-api/tests/integration/db/test_contributions_repository.py
@@ -86,7 +86,7 @@ def _identity(
     unique_value=None,
     condition_key="",
 ) -> dict:
-    """The full composite natural key (see ``Contribution.identifier_fields``) for a semantic lookup.
+    """The full composite natural key (see ``ContributionIdentity.model_fields``) for a semantic lookup.
 
     Mirrors the defaults ``_insert``/``_contrib_in`` seed, so ``_identity(material_id=...)`` addresses
     a document created with the matching ``identifier=...``.
@@ -400,11 +400,29 @@ async def test_composite_identity_is_unique_lookup(self, db):
         assert result is not None
         assert result.material_id == "id-a"
 
-    async def test_partial_identifier_set_is_rejected(self, db):
-        # The semantic set must be the complete composite key, not a subset.
-        await _insert(project="partial-proj", identifier="partial-id")
+    async def test_partial_identity_resolves_matching_doc(self, db):
+        # A partial identity (a subset of the composite key) is accepted, as long as it satisfies the
+        # hierarchy, and resolves the single matching row.
+        await _insert(project="partial-proj", identifier="partial-id", chemical_system_id="Fe-O")
+        result = await _repo(ADMIN).read_one(
+            {"project": "partial-proj", "chemical_system_id": "Fe-O"}, fields=None
+        )
+        assert result is not None
+        assert result.material_id == "partial-id"
+
+    async def test_partial_identity_ambiguous_raises_conflict(self, db):
+        # A partial identity that matches more than one in-scope row is rejected rather than silently
+        # returning the first — the caller must disambiguate (e.g. supply material_id/formula).
+        await _insert(project="ambig-proj", identifier="mp-1", chemical_system_id="Fe-O")
+        await _insert(project="ambig-proj", identifier="mp-2", chemical_system_id="Fe-O")
+        with pytest.raises(ConflictError):
+            await _repo(ADMIN).read_one({"project": "ambig-proj", "chemical_system_id": "Fe-O"}, fields=None)
+
+    async def test_identity_without_chemical_system_is_rejected(self, db):
+        # The identifier hierarchy requires chemical_system_id; a partial lacking it is invalid.
+        await _insert(project="no-chemsys", identifier="mp-1")
         with pytest.raises(ValidationError):
-            await _repo(ADMIN).read_one({"project": "partial-proj", "material_id": "partial-id"}, fields=None)
+            await _repo(ADMIN).read_one({"project": "no-chemsys", "material_id": "mp-1"}, fields=None)
 
 
 # ---------------------------------------------------------------------------
@@ -533,7 +551,7 @@ class TestUpsertContributionById:
     async def test_insert_when_id_absent_persists_document(self, db):
         new_id = PydanticObjectId()
         payload = _contrib_in(identifier="mp-4002", _id=new_id)
-        result = await _repo(ADMIN).upsert_one({"id": str(new_id)}, payload)
+        result = await _repo(ADMIN).upsert_by_id(str(new_id), payload)
         # Must be the resolved document, not an un-awaited query object.
         assert isinstance(result, Contribution)
         stored = await Contribution.find_one(Contribution.id == new_id)
@@ -543,7 +561,7 @@ async def test_insert_when_id_absent_persists_document(self, db):
     async def test_update_when_id_present_applies_change(self, db):
         existing = await _insert(identifier="mp-4001")
         payload = _contrib_in(identifier="mp-4001", formula="Li2O", _id=existing.id)
-        result = await _repo(ADMIN).upsert_one({"id": str(existing.id)}, payload)
+        result = await _repo(ADMIN).upsert_by_id(str(existing.id), payload)
         assert isinstance(result, Contribution)
         stored = await Contribution.find_one(Contribution.id == existing.id)
         assert stored is not None
@@ -554,7 +572,8 @@ async def test_upsert_by_identity_updates_in_place(self, db):
         # updates the matching document in place rather than creating a duplicate.
         await _insert(project="ups-sem", identifier="mp-5001", formula="Fe2O3")
         payload = _contrib_in(project="ups-sem", identifier="mp-5001", formula="Fe2O3")
-        result = await _repo(ADMIN).upsert_one(_identity(project="ups-sem", material_id="mp-5001"), payload)
+        doc = _repo(ADMIN).document_model.from_input_model(payload)
+        result = await _repo(ADMIN).upsert_one(doc)
         assert isinstance(result, Contribution)
         count = await Contribution.find(Contribution.project == "ups-sem").count()
         assert count == 1
@@ -565,7 +584,7 @@ async def test_update_clears_unique_value_when_resolved_to_none(self, db):
         existing = await _insert(identifier="mp-4004", unique_value="batch-A")
         assert existing.unique_value == "batch-A"
         payload = _contrib_in(identifier="mp-4004", _id=existing.id)
-        await _repo(ADMIN).upsert_one({"id": str(existing.id)}, payload, unique_value=None)
+        await _repo(ADMIN).upsert_by_id(str(existing.id), payload, unique_value=None)
         stored = await Contribution.find_one(Contribution.id == existing.id)
         assert stored is not None
         assert stored.unique_value is None
@@ -577,7 +596,7 @@ async def test_upsert_onto_existing_identity_raises_conflict(self, db):
         victim = await _insert(project="uid-dup", identifier="mp-200")
         payload = _contrib_in(project="uid-dup", identifier="mp-100", _id=victim.id)
         with pytest.raises(ConflictError):
-            await _repo(ADMIN).upsert_one({"id": str(victim.id)}, payload)
+            await _repo(ADMIN).upsert_by_id(str(victim.id), payload)
 
 
 class TestDeleteByIdsScope:
diff --git a/mpcontribs-api/tests/integration/test_component_routes.py b/mpcontribs-api/tests/integration/test_component_routes.py
index 84b193f43..5da9ca21c 100644
--- a/mpcontribs-api/tests/integration/test_component_routes.py
+++ b/mpcontribs-api/tests/integration/test_component_routes.py
@@ -144,6 +144,41 @@ def test_download_conventional_path(self, client, structure_service):
         assert client.get("/api/v1/structures/download/gz?format=csv").status_code == 200
 
 
+class TestStructuresByMd5Routing:
+    """The ``/item?md5=`` path addresses a component by its content hash — previously advertised in
+    docstrings but unreachable (a 32-hex md5 was rejected as a bad ObjectId on ``/{id}``)."""
+
+    MD5 = "a" * 32
+
+    def test_get_by_md5_forwards_identifiers(self, client, structure_service):
+        structure_service.read_one.return_value = SAMPLE_STRUCTURE
+        r = client.get(f"/api/v1/structures/item?md5={self.MD5}")
+        assert r.status_code == 200
+        assert structure_service.read_one.await_args.kwargs["identifiers"] == {"md5": self.MD5}
+
+    def test_delete_by_md5_forwards_identifiers(self, client, structure_service):
+        structure_service.delete_one.return_value = ComponentDeleteResponse(num_deleted=1)
+        r = client.delete(f"/api/v1/structures/item?md5={self.MD5}")
+        assert r.status_code == 200
+        assert structure_service.delete_one.await_args.kwargs["identifiers"] == {"md5": self.MD5}
+
+    def test_patch_by_md5_forwards_identifiers(self, client, structure_service):
+        structure_service.update_one.return_value = SAMPLE_STRUCTURE
+        r = client.patch(f"/api/v1/structures/item?md5={self.MD5}", json={"name": "renamed"})
+        assert r.status_code == 200
+        assert structure_service.update_one.await_args.kwargs["identifiers"] == {"md5": self.MD5}
+
+    def test_malformed_md5_returns_422(self, client, structure_service):
+        # Typed with MD5Hash, so a non-32-hex value is rejected before reaching the service.
+        assert client.get("/api/v1/structures/item?md5=nothex").status_code == 422
+
+    def test_item_is_not_captured_as_an_id(self, client, structure_service):
+        # ``/item`` must be routed to the md5 handler, not matched as ``/{id}`` with id="item".
+        structure_service.read_one.return_value = SAMPLE_STRUCTURE
+        client.get(f"/api/v1/structures/item?md5={self.MD5}")
+        assert structure_service.read_one.await_args.kwargs["identifiers"] == {"md5": self.MD5}
+
+
 # ===========================================================================
 # TABLES
 # ===========================================================================
diff --git a/mpcontribs-api/tests/integration/test_contributions_routes.py b/mpcontribs-api/tests/integration/test_contributions_routes.py
index 8195d7d30..6f9ee71e2 100644
--- a/mpcontribs-api/tests/integration/test_contributions_routes.py
+++ b/mpcontribs-api/tests/integration/test_contributions_routes.py
@@ -4,7 +4,7 @@
 from mpcontribs_api.domains._shared.bulk import BulkDeleteSummary, BulkUpdateSummary, BulkWriteSummary
 from mpcontribs_api.domains.contributions.dependencies import get_contribution_service
 from mpcontribs_api.domains.contributions.models import ContributionOut
-from mpcontribs_api.exceptions import NotFoundError
+from mpcontribs_api.exceptions import ConflictError, NotFoundError
 from tests.integration.conftest import AUTHED_HEADERS, FORCE_ANON_HEADERS
 
 # ---------------------------------------------------------------------------
@@ -150,6 +150,98 @@ def test_download_route_conventional_path(self, client, contribution_service):
         assert client.get("/api/v1/contributions/download/gz").status_code == 200
 
 
+class TestContributionByIdentityRouting:
+    """The ``/item`` path addresses a contribution by its user-suppliable natural identity; both it and
+    ``/{id}`` funnel through the unified ``read_one``/``delete_one``/``update_one`` (which take an
+    identity or an id, preferring the id). The server resolves the rest and 409s on an ambiguous subset."""
+
+    def test_get_by_identity_defaults_unsupplied_subset(self, client, contribution_service):
+        contribution_service.read_one.return_value = SAMPLE_OUT
+        r = client.get("/api/v1/contributions/item?project=p&chemical_system_id=Fe-O")
+        assert r.status_code == 200
+        identifiers = contribution_service.read_one.await_args.args[0]
+        # The literal ``item`` must reach ``read_one`` as an identity dict, not ``{"id": "item"}``.
+        assert identifiers["project"] == "p"
+        assert "id" not in identifiers
+        # Unsupplied hierarchy/tiebreaker fields default to None; the service pins/relaxes them.
+        assert identifiers["material_id"] is None
+        assert identifiers["formula"] is None
+        # The identity dict always carries the full natural key; an unsupplied condition_key defaults to
+        # "" (matching the empty-condition row) and an unsupplied unique_value is present as None, which
+        # Mongo-matches null-or-absent stored values (keep_nulls=False) and satisfies the repository's
+        # exact-identifier-key check.
+        assert identifiers["unique_value"] is None
+        assert identifiers["condition_key"] == ""
+
+    def test_get_by_identity_forwards_condition_key(self, client, contribution_service):
+        contribution_service.read_one.return_value = SAMPLE_OUT
+        client.get(
+            "/api/v1/contributions/item",
+            params={"project": "p", "chemical_system_id": "Fe-O", "condition_key": "T=300K"},
+        )
+        identifiers = contribution_service.read_one.await_args.args[0]
+        # condition_key is a caller-suppliable selector for a specific pivoted row (no longer forced to
+        # ""), so the caller's value reaches the service verbatim to address that row.
+        assert identifiers["condition_key"] == "T=300K"
+
+    def test_get_by_identity_forwards_full_subset(self, client, contribution_service):
+        contribution_service.read_one.return_value = SAMPLE_OUT
+        client.get(
+            "/api/v1/contributions/item?project=p&chemical_system_id=Fe-O&material_id=mp-1&formula=Fe2O3&unique_value=A"
+        )
+        identifiers = contribution_service.read_one.await_args.args[0]
+        assert identifiers["material_id"] == "mp-1"
+        assert identifiers["unique_value"] == "A"
+
+    def test_missing_required_chemical_system_returns_422(self, client, contribution_service):
+        assert client.get("/api/v1/contributions/item?project=p").status_code == 422
+
+    def test_ambiguous_identity_returns_409(self, client, contribution_service):
+        contribution_service.read_one.side_effect = ConflictError("ambiguous")
+        r = client.get("/api/v1/contributions/item?project=p&chemical_system_id=Fe-O")
+        assert r.status_code == 409
+
+    def test_delete_by_identity_forwards_to_service(self, client, contribution_service):
+        contribution_service.delete_one.return_value = BulkDeleteSummary(num_deleted=1, num_children_deleted=0)
+        r = client.delete("/api/v1/contributions/item?project=p&chemical_system_id=Fe-O&material_id=mp-1&formula=Fe2O3")
+        assert r.status_code == 200
+        identifiers = contribution_service.delete_one.await_args.args[0]
+        assert identifiers["project"] == "p"
+        assert "id" not in identifiers
+
+    def test_patch_by_identity_forwards_to_service(self, client, contribution_service):
+        contribution_service.update_one.return_value = SAMPLE_OUT
+        r = client.patch(
+            "/api/v1/contributions/item?project=p&chemical_system_id=Fe-O&material_id=mp-1&formula=Fe2O3",
+            json={"is_public": True},
+        )
+        assert r.status_code == 200
+        identifiers = contribution_service.update_one.await_args.args[0]
+        assert identifiers["project"] == "p"
+        assert "id" not in identifiers
+
+    # A material_id without a formula violates the identifier hierarchy. ``ContributionIdentity``'s
+    # model_validator rejects it at parse time (422) for every verb, so the request never reaches the
+    # service — DELETE in particular must not silently fall through to a 0-count delete.
+    def test_get_by_identity_bad_hierarchy_returns_422(self, client, contribution_service):
+        r = client.get("/api/v1/contributions/item?project=p&chemical_system_id=Fe-O&material_id=mp-1")
+        assert r.status_code == 422
+        contribution_service.read_one.assert_not_called()
+
+    def test_delete_by_identity_bad_hierarchy_returns_422(self, client, contribution_service):
+        r = client.delete("/api/v1/contributions/item?project=p&chemical_system_id=Fe-O&material_id=mp-1")
+        assert r.status_code == 422
+        contribution_service.delete_one.assert_not_called()
+
+    def test_patch_by_identity_bad_hierarchy_returns_422(self, client, contribution_service):
+        r = client.patch(
+            "/api/v1/contributions/item?project=p&chemical_system_id=Fe-O&material_id=mp-1",
+            json={"is_public": True},
+        )
+        assert r.status_code == 422
+        contribution_service.update_one.assert_not_called()
+
+
 # ===========================================================================
 # Single-resource behavior (independent of the routing bug, via current paths)
 # ===========================================================================
diff --git a/mpcontribs-api/tests/unit/domains/test_contribution_service.py b/mpcontribs-api/tests/unit/domains/test_contribution_service.py
index 026473144..9137e20b8 100644
--- a/mpcontribs-api/tests/unit/domains/test_contribution_service.py
+++ b/mpcontribs-api/tests/unit/domains/test_contribution_service.py
@@ -505,7 +505,7 @@ async def test_only_new_documents_count_against_cap(self, monkeypatch):
         assert len(summary.succeeded) == 2
         assert [f.index for f in summary.failed] == [2]
         assert summary.failed[0].error_code == "permission_denied"
-        upserted = {c.args[1].material_id for c in contrib_repo.upsert_one.call_args_list}
+        upserted = {c.args[0].material_id for c in contrib_repo.upsert_one.call_args_list}
         assert upserted == {_mp_id_for("a"), _mp_id_for("b")}
 
     async def test_pure_updates_are_never_capped(self, monkeypatch):
@@ -553,12 +553,12 @@ async def test_update_existing_allowed_even_over_cap(self, monkeypatch):
         contrib_repo = AsyncMock()
         contrib_repo.read_one.return_value = MagicMock(spec=Contribution)  # id exists -> update
         contrib_repo.count_matching.return_value = 99
-        contrib_repo.upsert_one.return_value = MagicMock(spec=Contribution)
+        contrib_repo.upsert_by_id.return_value = MagicMock(spec=Contribution)
         svc, *_ = _make_service(contributions=contrib_repo, projects=_unapproved_projects_repo())
 
-        await svc.upsert_one("someid", _contrib_in())
+        await svc.upsert_one({"id": "someid"}, _contrib_in())
 
-        contrib_repo.upsert_one.assert_called_once()
+        contrib_repo.upsert_by_id.assert_called_once()
         contrib_repo.count_matching.assert_not_called()
 
     async def test_new_insert_over_cap_rejected(self, monkeypatch):
@@ -569,21 +569,21 @@ async def test_new_insert_over_cap_rejected(self, monkeypatch):
         svc, *_ = _make_service(contributions=contrib_repo, projects=_unapproved_projects_repo())
 
         with pytest.raises(PermissionError):
-            await svc.upsert_one("someid", _contrib_in())
+            await svc.upsert_one({"id": "someid"}, _contrib_in())
 
-        contrib_repo.upsert_one.assert_not_called()
+        contrib_repo.upsert_by_id.assert_not_called()
 
     async def test_new_insert_under_cap_allowed(self, monkeypatch):
         monkeypatch.setattr(get_settings().consumer, "max_unapproved_contributions_per_project", 5)
         contrib_repo = AsyncMock()
         contrib_repo.read_one.return_value = None
         contrib_repo.count_matching.return_value = 1
-        contrib_repo.upsert_one.return_value = MagicMock(spec=Contribution)
+        contrib_repo.upsert_by_id.return_value = MagicMock(spec=Contribution)
         svc, *_ = _make_service(contributions=contrib_repo, projects=_unapproved_projects_repo())
 
-        await svc.upsert_one("someid", _contrib_in())
+        await svc.upsert_one({"id": "someid"}, _contrib_in())
 
-        contrib_repo.upsert_one.assert_called_once()
+        contrib_repo.upsert_by_id.assert_called_once()
 
     async def test_new_insert_at_exactly_cap_rejected(self, monkeypatch):
         # stored == cap: the project is full, so a brand-new document is rejected (no cap+1 slack).
@@ -594,20 +594,20 @@ async def test_new_insert_at_exactly_cap_rejected(self, monkeypatch):
         svc, *_ = _make_service(contributions=contrib_repo, projects=_unapproved_projects_repo())
 
         with pytest.raises(PermissionError):
-            await svc.upsert_one("someid", _contrib_in())
+            await svc.upsert_one({"id": "someid"}, _contrib_in())
 
-        contrib_repo.upsert_one.assert_not_called()
+        contrib_repo.upsert_by_id.assert_not_called()
 
     async def test_new_insert_approved_project_unlimited(self, monkeypatch):
         monkeypatch.setattr(get_settings().consumer, "max_unapproved_contributions_per_project", 1)
         contrib_repo = AsyncMock()
         contrib_repo.read_one.return_value = None
-        contrib_repo.upsert_one.return_value = MagicMock(spec=Contribution)
+        contrib_repo.upsert_by_id.return_value = MagicMock(spec=Contribution)
         svc, *_ = _make_service(contributions=contrib_repo, projects=_approved_projects_repo())
 
-        await svc.upsert_one("someid", _contrib_in())
+        await svc.upsert_one({"id": "someid"}, _contrib_in())
 
-        contrib_repo.upsert_one.assert_called_once()
+        contrib_repo.upsert_by_id.assert_called_once()
         contrib_repo.count_matching.assert_not_called()
 
 
@@ -960,8 +960,9 @@ async def test_upsert_passes_resolved_unique_value_in_identifiers(self):
 
         await svc.upsert_many([_contrib_in(data={"sample_id": "A"})])
 
-        identifiers = contrib_repo.upsert_one.call_args.args[0]
-        assert identifiers["unique_value"] == "A"
+        # The batch path builds the document and stamps the server-resolved unique_value onto it.
+        doc = contrib_repo.upsert_one.call_args.args[0]
+        assert doc.unique_value == "A"
 
     async def test_upsert_missing_unique_column_value_is_validation_failure(self):
         svc, contrib_repo, *_ = _make_service(unique_column="sample_id")
@@ -1046,15 +1047,16 @@ async def test_calls_atomic_repo_method_once_per_item(self):
         # The atomic upsert path is used, not the bulk insert path.
         contrib_repo.insert_one.assert_not_called()
 
-    async def test_passes_identifiers_dict_and_input_to_repo(self):
+    async def test_passes_identity_stamped_document_to_repo(self):
         svc, contrib_repo, *_ = _make_service()
         contrib_repo.upsert_one.return_value = MagicMock(spec=Contribution, project="proj")
 
         contrib = _contrib_in(project="my-proj", material_id="mp-99")
         await svc.upsert_many([contrib])
 
-        call = contrib_repo.upsert_one.call_args
-        assert call.args[0] == {
+        # The repo is handed a fully-built document; its identity carries the write's natural key.
+        doc = contrib_repo.upsert_one.call_args.args[0]
+        assert doc.identity().as_dict() == {
             "project": "my-proj",
             "material_id": "mp-99",
             "chemical_system_id": "Fe-O",
@@ -1062,7 +1064,6 @@ async def test_passes_identifiers_dict_and_input_to_repo(self):
             "unique_value": None,
             "condition_key": "",
         }
-        assert call.args[1] is contrib
 
     async def test_returns_repo_results_in_input_order(self):
         svc, contrib_repo, *_ = _make_service()
@@ -1071,9 +1072,9 @@ async def test_returns_repo_results_in_input_order(self):
             doc.project = "proj"  # real project so update_project can aggregate the affected set
         returned = {}
 
-        async def _upsert(identifiers, contrib):
-            doc = docs[int(contrib.material_id.split("-")[1])]
-            returned[contrib.material_id] = doc
+        async def _upsert(document):
+            doc = docs[int(document.material_id.split("-")[1])]
+            returned[document.material_id] = doc
             return doc
 
         contrib_repo.upsert_one.side_effect = _upsert
@@ -1111,8 +1112,8 @@ async def test_same_key_concurrent_upserts_both_go_through_atomic_call(self):
     async def test_one_failure_is_reported_not_raised(self):
         svc, contrib_repo, *_ = _make_service()
 
-        async def _upsert(identifiers, contrib):
-            if contrib.material_id == "mp-1":
+        async def _upsert(document):
+            if document.material_id == "mp-1":
                 raise ConflictError("boom")
             return MagicMock(spec=Contribution, project="proj")
 
@@ -1248,24 +1249,24 @@ async def test_upsert_authorized_member_proceeds(self):
         # not gated by the quota, so the write goes through.
         contrib_repo = AsyncMock()
         contrib_repo.read_one.return_value = MagicMock(spec=Contribution)  # exists -> update
-        contrib_repo.upsert_one.return_value = MagicMock(spec=Contribution)
+        contrib_repo.upsert_by_id.return_value = MagicMock(spec=Contribution)
         svc, *_ = _make_service(
             contributions=contrib_repo, projects=_unapproved_projects_repo(), user=_member_user("allowed")
         )
 
-        await svc.upsert_one("someid", _contrib_in(project="allowed"))
+        await svc.upsert_one({"id": "someid"}, _contrib_in(project="allowed"))
 
-        contrib_repo.upsert_one.assert_called_once()
+        contrib_repo.upsert_by_id.assert_called_once()
 
     async def test_upsert_admin_bypasses_authorization(self):
         contrib_repo = AsyncMock()
         contrib_repo.read_one.return_value = MagicMock(spec=Contribution)
-        contrib_repo.upsert_one.return_value = MagicMock(spec=Contribution)
+        contrib_repo.upsert_by_id.return_value = MagicMock(spec=Contribution)
         svc, *_ = _make_service(contributions=contrib_repo, projects=_approved_projects_repo())  # admin default
 
-        await svc.upsert_one("someid", _contrib_in(project="anything"))
+        await svc.upsert_one({"id": "someid"}, _contrib_in(project="anything"))
 
-        contrib_repo.upsert_one.assert_called_once()
+        contrib_repo.upsert_by_id.assert_called_once()
 
 
 # ---------------------------------------------------------------------------
@@ -1521,7 +1522,7 @@ async def test_patch_material_id_onto_doc_without_formula_raises(self):
         contrib_repo.read_one.return_value = existing
 
         with pytest.raises(ValidationError, match="formula is required when material_id"):
-            await svc.update_one(str(existing.id), ContributionPatch(material_id="mp-1"))
+            await svc.update_one({"id": str(existing.id)}, ContributionPatch(material_id="mp-1"))
 
         # Rejected before any write.
         contrib_repo.update_one.assert_not_called()
@@ -1532,7 +1533,7 @@ async def test_patch_material_id_when_existing_has_formula_ok(self):
         contrib_repo.read_one.return_value = existing
         contrib_repo.update_one.return_value = MagicMock(spec=Contribution)
 
-        await svc.update_one(str(existing.id), ContributionPatch(material_id="mp-1"))
+        await svc.update_one({"id": str(existing.id)}, ContributionPatch(material_id="mp-1"))
 
         contrib_repo.update_one.assert_called_once()
 
@@ -1540,7 +1541,7 @@ async def test_metadata_only_patch_skips_existing_read(self):
         svc, contrib_repo, *_ = _make_service()
         contrib_repo.update_one.return_value = MagicMock(spec=Contribution)
 
-        await svc.update_one("some-id", ContributionPatch(is_public=True))
+        await svc.update_one({"id": "some-id"}, ContributionPatch(is_public=True))
 
         # No identity/unique inputs touched -> no re-read, straight to the plain patch.
         contrib_repo.read_one.assert_not_called()
@@ -1560,7 +1561,7 @@ async def test_data_patch_defaults_to_merge_and_forwards_replace_false(self):
         contrib_repo.read_one.return_value = existing
         contrib_repo.update_one.return_value = MagicMock(spec=Contribution)
 
-        await svc.update_one(str(existing.id), ContributionPatch(data={"y": 9.0}))
+        await svc.update_one({"id": str(existing.id)}, ContributionPatch(data={"y": 9.0}))
 
         # The repo performs the actual dotted-$set merge; the service just forwards replace_data=False.
         assert contrib_repo.update_one.call_args.kwargs["replace_data"] is False
@@ -1573,7 +1574,7 @@ async def test_replace_data_flag_forwarded_to_repo(self):
         contrib_repo.update_one.return_value = MagicMock(spec=Contribution)
 
         await svc.update_one(
-            str(existing.id), ContributionPatch(data={"y": 9.0}), replace_data=True
+            {"id": str(existing.id)}, ContributionPatch(data={"y": 9.0}), replace_data=True
         )
 
         assert contrib_repo.update_one.call_args.kwargs["replace_data"] is True
@@ -1587,7 +1588,7 @@ async def test_merge_resolves_unique_value_from_merged_state(self):
         contrib_repo.read_one.return_value = existing
         contrib_repo.update_one.return_value = MagicMock(spec=Contribution)
 
-        await svc.update_one(str(existing.id), ContributionPatch(data={"y": 9.0}))
+        await svc.update_one({"id": str(existing.id)}, ContributionPatch(data={"y": 9.0}))
 
         # Resolved from {sample_id:42, x:1, y:9}, so the untouched unique_value survives the merge.
         assert contrib_repo.update_one.call_args.kwargs["unique_value"] == 42
@@ -1602,6 +1603,6 @@ async def test_replace_resolves_unique_value_from_patch_data_only(self):
 
         with pytest.raises(ValidationError, match="unique_column"):
             await svc.update_one(
-                str(existing.id), ContributionPatch(data={"y": 9.0}), replace_data=True
+                {"id": str(existing.id)}, ContributionPatch(data={"y": 9.0}), replace_data=True
             )
         contrib_repo.update_one.assert_not_called()
diff --git a/mpcontribs-api/tests/unit/domains/test_contributions_models.py b/mpcontribs-api/tests/unit/domains/test_contributions_models.py
index 360f2673a..028635bea 100644
--- a/mpcontribs-api/tests/unit/domains/test_contributions_models.py
+++ b/mpcontribs-api/tests/unit/domains/test_contributions_models.py
@@ -607,9 +607,14 @@ def test_hierarchy_fields_are_the_identifier_triple(self):
 class TestContributionIdentityIndexModel:
     def test_default_index_is_named_and_unique(self):
         index = ContributionIdentity.index_model()
-        assert index.document["name"] == "project_identity"
+        # Neutral default; each collection passes an explicit name matching its deployed index.
+        assert index.document["name"] == "identity"
         assert index.document["unique"] is True
 
+    def test_explicit_name_is_used(self):
+        # Contributions keeps its deployed index name by passing it explicitly.
+        assert ContributionIdentity.index_model(name="project_identity").document["name"] == "project_identity"
+
     def test_unique_can_be_disabled(self):
         assert ContributionIdentity.index_model(unique=False).document["unique"] is False
 
diff --git a/mpcontribs-api/tests/unit/domains/test_project_service.py b/mpcontribs-api/tests/unit/domains/test_project_service.py
index 4af888e3d..cde81ba5e 100644
--- a/mpcontribs-api/tests/unit/domains/test_project_service.py
+++ b/mpcontribs-api/tests/unit/domains/test_project_service.py
@@ -64,7 +64,8 @@ def _service(user: User, *, existing=None, scoped=None, count: int = 0, limits:
     projects.find_by_id_unscoped.return_value = existing
     projects.read_one.return_value = scoped
     projects.count_matching.return_value = count
-    projects.upsert_one.side_effect = lambda doc, **kw: doc
+    # PUT does a full-replace-by-id (repo.replace_one(id, doc)); return the doc it was handed.
+    projects.replace_one.side_effect = lambda id, doc, **kw: doc
     projects.update_one.return_value = _project()
     projects.delete_one.return_value = DeleteResponse(num_deleted=1)
     initiatives = AsyncMock()
@@ -111,13 +112,13 @@ async def test_anonymous_raises_permission(self):
         svc, projects, _ = _service(ANON)
         with pytest.raises(AppPermissionError):
             await svc.upsert_one({"id": "proj-1"}, _project_in("p1"))
-        projects.upsert_one.assert_not_called()
+        projects.replace_one.assert_not_called()
 
     async def test_existing_non_owner_raises_permission(self):
         svc, projects, _ = _service(BOB, existing=_project(owner=ALICE_EMAIL))
         with pytest.raises(AppPermissionError):
             await svc.upsert_one({"id": "proj-1"}, _project_in("p1"))
-        projects.upsert_one.assert_not_called()
+        projects.replace_one.assert_not_called()
 
     async def test_update_preserves_owner_and_server_fields(self):
         existing = _project(
@@ -130,7 +131,7 @@ async def test_update_preserves_owner_and_server_fields(self):
         svc, projects, _ = _service(ALICE, existing=existing)
         # Body tries to reassign owner and drop publication; both must be ignored/preserved.
         await svc.upsert_one({"id": "proj-1"}, _project_in("p1", owner=BOB_EMAIL, is_public=False))
-        saved = projects.upsert_one.call_args.args[0]
+        saved = projects.replace_one.call_args.args[1]
         assert saved.owner == ALICE_EMAIL
         assert saved.is_public is True
         assert saved.is_approved is True
@@ -140,7 +141,7 @@ async def test_update_preserves_owner_and_server_fields(self):
     async def test_new_forces_owner_and_unapproves(self):
         svc, projects, _ = _service(BOB, existing=None, count=0)
         await svc.upsert_one({"id": "proj-1"}, _project_in("p1", owner=ALICE_EMAIL, is_approved=True))
-        saved = projects.upsert_one.call_args.args[0]
+        saved = projects.replace_one.call_args.args[1]
         assert saved.owner == BOB_EMAIL
         assert saved.is_approved is False
 
@@ -148,14 +149,14 @@ async def test_new_over_cap_raises_permission(self):
         svc, projects, _ = _service(ALICE, existing=None, count=5, limits=ConsumerSettings(max_projects=2))
         with pytest.raises(AppPermissionError):
             await svc.upsert_one({"id": "proj-1"}, _project_in("p1", owner=ALICE_EMAIL))
-        projects.upsert_one.assert_not_called()
+        projects.replace_one.assert_not_called()
 
     async def test_public_unapproved_raises_validation(self):
         # Admin new project: approval is not forced off, so a public+unapproved body trips the invariant.
         svc, projects, _ = _service(ADMIN, existing=None, count=0)
         with pytest.raises(ValidationError):
             await svc.upsert_one({"id": "proj-1"}, _project_in("p1", is_public=True, is_approved=False))
-        projects.upsert_one.assert_not_called()
+        projects.replace_one.assert_not_called()
 
 
 # ---------------------------------------------------------------------------
diff --git a/mpcontribs-api/tests/unit/domains/test_shared_models.py b/mpcontribs-api/tests/unit/domains/test_shared_models.py
index 2ba8c3c9d..a162c120c 100644
--- a/mpcontribs-api/tests/unit/domains/test_shared_models.py
+++ b/mpcontribs-api/tests/unit/domains/test_shared_models.py
@@ -4,10 +4,18 @@
 
 from mpcontribs_api.domains._shared.models import (
     BaseDocumentWithInput,
+    ComponentIdentity,
     DeleteResponse,
     DocumentOut,
 )
-from mpcontribs_api.domains.attachments.models import Attachment, AttachmentIn
+from mpcontribs_api.domains.attachments.models import Attachment, ComponentIdentity, AttachmentIn
+from mpcontribs_api.domains.consumers.models import Consumer, ConsumerIdentity
+from mpcontribs_api.domains.contributions.models import Contribution, ContributionIdentity
+from mpcontribs_api.domains.initiatives.models import Initiative, InitiativeIdentity
+from mpcontribs_api.domains.project_groups.models import ProjectGroup, ProjectGroupIdentity
+from mpcontribs_api.domains.projects.models import Project, ProjectIdentity, ProjectIn
+from mpcontribs_api.domains.structures.models import Structure, ComponentIdentity
+from mpcontribs_api.domains.tables.models import Table, ComponentIdentity
 from mpcontribs_api.pagination import encode_cursor
 
 # ---------------------------------------------------------------------------
@@ -93,6 +101,74 @@ def test_serializes_under_id_not_underscore_id(self):
         assert "_id" not in dumped
 
 
+# ---------------------------------------------------------------------------
+# Identity abstraction: every domain declares its natural key once, via ``identity_model``.
+# The repository reads ``identity_model.model_fields`` directly (and ``identity()`` derives from it)
+# so methods stay agnostic to how a given domain is identified (``_id`` vs a compound business key).
+# ---------------------------------------------------------------------------
+
+
+# (Document class, its Identity class, expected natural-key field set)
+_DOMAIN_IDENTITIES = [
+    (Project, ProjectIdentity, {"id"}),
+    (Initiative, InitiativeIdentity, {"slug"}),
+    (ProjectGroup, ProjectGroupIdentity, {"name", "owner"}),
+    (Consumer, ConsumerIdentity, {"consumer_id"}),
+    (Structure, ComponentIdentity, {"md5"}),
+    (Table, ComponentIdentity, {"md5"}),
+    (Attachment, ComponentIdentity, {"md5"}),
+    (
+        Contribution,
+        ContributionIdentity,
+        {"project", "material_id", "chemical_system_id", "formula", "unique_value", "condition_key"},
+    ),
+]
+
+
+class TestIdentityContract:
+    @pytest.mark.parametrize(("document", "identity", "fields"), _DOMAIN_IDENTITIES)
+    def test_identity_model_is_bound(self, document, identity, fields):
+        assert document.identity_model is identity
+
+    @pytest.mark.parametrize(("document", "identity", "fields"), _DOMAIN_IDENTITIES)
+    def test_natural_key_derives_from_identity_model(self, document, identity, fields):
+        # Single source of truth: the natural key comes from the Identity class's fields directly.
+        assert identity.model_fields.keys() == fields
+        assert document.identity_model.model_fields.keys() == fields
+
+
+class TestDocumentIdentityRoundTrips:
+    def test_id_keyed_document_identity_is_its_id(self):
+        # A project's identity IS its id (the human-chosen slug), so ``identity()`` reads it off ``id``.
+        project = Project.from_input_model(
+            ProjectIn(title="my-project", authors="a", description="d", owner="google:a@example.com"),
+            id="my-proj",
+        )
+        assert project.identity() == ProjectIdentity(id="my-proj")
+        assert project.identity().as_dict() == {"id": "my-proj"}
+
+    def test_content_addressed_component_identity_is_its_md5(self):
+        # A component's identity is its server-computed content md5.
+        doc = Attachment.from_input(_attachment_in())
+        assert doc.identity() == ComponentIdentity(md5=doc.md5)
+        assert doc.identity().as_dict() == {"md5": doc.md5}
+
+    def test_compound_identity_reads_all_fields_and_tolerates_nulls(self):
+        # A chem-system-only contribution stores null material_id/formula; identity() falls back to the
+        # dataclass defaults (keep_nulls=False parity) rather than raising.
+        contribution = Contribution.model_validate(
+            {"_id": PydanticObjectId(), "project": "p", "chemical_system_id": "Fe-O", "data": {}}
+        )
+        assert contribution.identity() == ContributionIdentity(
+            project="p",
+            material_id=None,
+            chemical_system_id="Fe-O",
+            formula=None,
+            unique_value=None,
+            condition_key="",
+        )
+
+
 # ---------------------------------------------------------------------------
 # DeleteResponse.from_delete_result
 # ---------------------------------------------------------------------------