feat(equipment): Assembly aggregate (5th equipment aggregate) + rename catchups

apps/api/src/cora/equipment/_projections.py

@@ -8,6 +8,7 @@
 """
 
 from cora.equipment.projections import (
+    AssemblySummaryProjection,
     AssetFamilyMembershipProjection,
     AssetLocationProjection,
     AssetSummaryProjection,
@@ -41,6 +42,7 @@ def register_equipment_projections(
     registry.register(MountSlotCodeProjection())
     registry.register(MountChildrenProjection())
     registry.register(AssetLocationProjection())
+    registry.register(AssemblySummaryProjection())
 
 
 __all__ = ["register_equipment_projections"]

apps/api/src/cora/equipment/_template_slot_body.py

@@ -0,0 +1,87 @@
+"""Shared Pydantic wire-format mirror of the `TemplateSlot` value object.
+
+`TemplateSlot` is a frozen dataclass at the domain layer
+(`cora.equipment.aggregates.assembly.state`). This body is the wire
+shape REST and MCP parse, with `to_domain()` converting to the
+domain VO (which may raise `InvalidSlotNameError`,
+`InvalidSlotCardinalityError`, `InvalidTemplateSlotError`, or
+`InvalidPlacementError` during construction).
+
+Pydantic enforces field-shape rules at the API boundary; domain
+invariants (non-empty `required_family_ids`, closed-enum
+`cardinality`, NaN-rejecting `default_placement`) surface from the
+domain VO constructors and map to HTTP 400 through the BC's
+exception handler.
+
+Hoisted alongside `PlacementBody` / `DrawingBody` since both the
+REST route and the MCP tool for `define_assembly` (and the future
+`version_assembly` slice) parse the same shape.
+"""
+
+from typing import Any
+from uuid import UUID
+
+from pydantic import BaseModel, Field
+
+from cora.equipment._placement_body import PlacementBody
+from cora.equipment.aggregates.assembly.state import (
+    SLOT_NAME_MAX_LENGTH,
+    SlotCardinality,
+    SlotName,
+    TemplateSlot,
+)
+
+
+class TemplateSlotBody(BaseModel):
+    """Wire format for a TemplateSlot value object."""
+
+    slot_name: str = Field(
+        ...,
+        min_length=1,
+        max_length=SLOT_NAME_MAX_LENGTH,
+        description=(
+            "Canonical name of this slot within the Assembly "
+            "(e.g., 'camera', 'rotary', 'trigger_source')."
+        ),
+    )
+    required_family_ids: frozenset[UUID] = Field(
+        ...,
+        min_length=1,
+        description=(
+            "Set of FamilyIds an instantiated Asset must include at "
+            "least one of. MUST be non-empty."
+        ),
+    )
+    cardinality: SlotCardinality = Field(
+        ...,
+        description=(
+            "How many Assets can fill this slot at instantiation: "
+            "Exactly1, ZeroOrOne, OneOrMore, ZeroOrMore."
+        ),
+    )
+    default_settings: dict[str, Any] | None = Field(
+        None,
+        description=(
+            "Optional template defaults applied at instantiation "
+            "unless overridden by the instantiator."
+        ),
+    )
+    default_placement: PlacementBody | None = Field(
+        None,
+        description="Optional template-default placement for the slot.",
+    )
+
+    def to_domain(self) -> TemplateSlot:
+        """Convert this wire body to the domain TemplateSlot VO."""
+        return TemplateSlot(
+            slot_name=SlotName(self.slot_name),
+            required_family_ids=self.required_family_ids,
+            cardinality=self.cardinality,
+            default_settings=self.default_settings,
+            default_placement=(
+                self.default_placement.to_domain() if self.default_placement is not None else None
+            ),
+        )
+
+
+__all__ = ["TemplateSlotBody"]

apps/api/src/cora/equipment/_template_wire_body.py

@@ -0,0 +1,57 @@
+"""Shared Pydantic wire-format mirror of the `TemplateWire` value object.
+
+`TemplateWire` is a frozen dataclass at the domain layer
+(`cora.equipment.aggregates.assembly.state`). The 4-tuple
+`(source_slot_name, source_port_name, target_slot_name,
+target_port_name)` IS the identity; the frozenset deduplicates on
+the tuple.
+
+Pydantic enforces field length at the API boundary; the degenerate
+full-self-loop case (same slot AND same port on both endpoints)
+surfaces from `TemplateWire.__post_init__` as
+`InvalidWireSpecError` -> HTTP 400.
+"""
+
+from pydantic import BaseModel, Field
+
+from cora.equipment.aggregates.assembly.state import (
+    WIRE_PORT_NAME_MAX_LENGTH,
+    TemplateWire,
+)
+
+
+class TemplateWireBody(BaseModel):
+    """Wire format for a TemplateWire value object."""
+
+    source_slot_name: str = Field(
+        ...,
+        min_length=1,
+        max_length=WIRE_PORT_NAME_MAX_LENGTH,
+    )
+    source_port_name: str = Field(
+        ...,
+        min_length=1,
+        max_length=WIRE_PORT_NAME_MAX_LENGTH,
+    )
+    target_slot_name: str = Field(
+        ...,
+        min_length=1,
+        max_length=WIRE_PORT_NAME_MAX_LENGTH,
+    )
+    target_port_name: str = Field(
+        ...,
+        min_length=1,
+        max_length=WIRE_PORT_NAME_MAX_LENGTH,
+    )
+
+    def to_domain(self) -> TemplateWire:
+        """Convert this wire body to the domain TemplateWire VO."""
+        return TemplateWire(
+            source_slot_name=self.source_slot_name,
+            source_port_name=self.source_port_name,
+            target_slot_name=self.target_slot_name,
+            target_port_name=self.target_port_name,
+        )
+
+
+__all__ = ["TemplateWireBody"]

apps/api/src/cora/equipment/aggregates/_assembly_content_hash.py

@@ -0,0 +1,103 @@
+"""Content-hash helper for the Assembly aggregate.
+
+Wraps the shared `cora.infrastructure.content_hash.compute_content_hash`
+pipeline with the Assembly-specific payload_type and the canonical
+subset materialized by `canonical_assembly_subset`.
+
+Used at two sites: `define_assembly` and `version_assembly`. Both
+fire the same hash for the same structural content; the
+payload_type is aggregate-level
+(`application/vnd.cora.assembly+json`), NOT event-level, so the
+hash is stable across "define" and "version" snapshots of the same
+canonical subset.
+
+Per `project_content_addressed_identity_design`: SHA-256 hex of
+DSSE PAE-wrapped canonical JSON. Per
+`project_canonicalization_research`: stdlib json sort-keys + NFC +
+set-to-sorted-list, all delegated to the shared
+`canonical_body_bytes`.
+
+Lives at BC root (`equipment/aggregates/_assembly_content_hash.py`)
+alongside `_drawing.py` and `_placement.py`, following the
+shared-helper convention.
+"""
+
+from uuid import UUID
+
+from cora.equipment.aggregates.assembly.state import (
+    Assembly,
+    AssemblyName,
+    TemplateSlot,
+    TemplateWire,
+    canonical_assembly_subset,
+)
+from cora.infrastructure.content_hash import compute_content_hash
+from cora.infrastructure.signing import event_type_to_payload_type
+
+# Aggregate-level payload_type: stable across define / version events
+# so two snapshots of the same canonical subset produce the same hash.
+ASSEMBLY_PAYLOAD_TYPE = "application/vnd.cora.assembly+json"
+
+# Per-event payload_types for signing the events themselves (not the
+# canonical-subset hash). Kept here to avoid scattering the strings
+# across slice deciders that emit these events.
+ASSEMBLY_DEFINED_PAYLOAD_TYPE = event_type_to_payload_type("AssemblyDefined")
+ASSEMBLY_VERSIONED_PAYLOAD_TYPE = event_type_to_payload_type("AssemblyVersioned")
+ASSEMBLY_DEPRECATED_PAYLOAD_TYPE = event_type_to_payload_type("AssemblyDeprecated")
+
+
+def compute_assembly_content_hash(
+    name: AssemblyName | str,
+    presents_as_family_id: UUID,
+    required_slots: frozenset[TemplateSlot],
+    required_wires: frozenset[TemplateWire],
+    parameter_overrides_schema: dict[str, object] | None,
+) -> str:
+    """Compute SHA-256 hex over an Assembly's canonical content subset.
+
+    Accepts either an AssemblyName VO or a raw string for `name` so
+    callers building from operator-supplied input do not need to
+    construct the VO twice (the AssemblyDefined decider validates
+    once via the VO, then passes the VO here for hashing).
+
+    Round-trip equivalence with
+    `compute_assembly_content_hash_from_state(Assembly(...))` is
+    pinned in tests; both paths funnel through
+    `canonical_assembly_subset` so structural drift between them is
+    impossible.
+    """
+    body = canonical_assembly_subset(
+        name=name,
+        presents_as_family_id=presents_as_family_id,
+        required_slots=required_slots,
+        required_wires=required_wires,
+        parameter_overrides_schema=parameter_overrides_schema,
+    )
+    return compute_content_hash(ASSEMBLY_PAYLOAD_TYPE, body)
+
+
+def compute_assembly_content_hash_from_state(state: Assembly) -> str:
+    """Compute the content_hash for a fully-constructed Assembly state.
+
+    Convenience wrapper: extracts the canonical subset via
+    `state.content_subset()` (which itself delegates to
+    `canonical_assembly_subset`) and feeds it to the shared
+    `compute_content_hash` pipeline under `ASSEMBLY_PAYLOAD_TYPE`.
+
+    Used in tests to verify that the state-method path and the
+    explicit-args path produce identical hashes, and may be useful
+    in future re-hash flows (e.g., a one-shot fitness that recomputes
+    all live Assembly hashes and asserts they match the stored
+    content_hash).
+    """
+    return compute_content_hash(ASSEMBLY_PAYLOAD_TYPE, state.content_subset())
+
+
+__all__ = [
+    "ASSEMBLY_DEFINED_PAYLOAD_TYPE",
+    "ASSEMBLY_DEPRECATED_PAYLOAD_TYPE",
+    "ASSEMBLY_PAYLOAD_TYPE",
+    "ASSEMBLY_VERSIONED_PAYLOAD_TYPE",
+    "compute_assembly_content_hash",
+    "compute_assembly_content_hash_from_state",
+]

apps/api/src/cora/equipment/aggregates/_drawing.py

@@ -157,9 +157,40 @@ def __post_init__(self) -> None:
             object.__setattr__(self, "revision", trimmed_revision)
 
 
+def drawing_to_payload(drawing: Drawing) -> dict[str, object]:
+    """Serialize a Drawing VO to a JSON-friendly dict for jsonb storage.
+
+    Canonical codec shared by every aggregate event that carries a
+    Drawing (Mount, Asset, Assembly). Sibling rebuilder is
+    `drawing_from_payload`. Hoisted to the VO module to close the
+    duplication anti-hook flagged in `project_mount_frame_design`
+    Watch items.
+    """
+    return {
+        "system": drawing.system.value,
+        "number": drawing.number,
+        "revision": drawing.revision,
+    }
+
+
+def drawing_from_payload(payload: dict[str, object]) -> Drawing:
+    """Reconstruct a Drawing VO from its JSON payload.
+
+    Sibling of `drawing_to_payload`. Round-trips losslessly with it.
+    Re-runs Drawing.__post_init__ validation.
+    """
+    return Drawing(
+        system=DrawingSystem(str(payload["system"])),
+        number=str(payload["number"]),
+        revision=(str(payload["revision"]) if payload.get("revision") is not None else None),
+    )
+
+
 __all__ = [
     "DRAWING_NUMBER_MAX_LENGTH",
     "DRAWING_REVISION_MAX_LENGTH",
     "Drawing",
     "DrawingSystem",
+    "drawing_from_payload",
+    "drawing_to_payload",
 ]

apps/api/src/cora/equipment/aggregates/_placement.py

@@ -201,8 +201,64 @@ def __post_init__(self) -> None:
                 )
 
 
+def placement_to_payload(placement: Placement) -> dict[str, object]:
+    """Serialize a Placement VO to a JSON-friendly dict for jsonb storage.
+
+    Canonical codec shared by every aggregate event that carries a
+    Placement (Mount, Frame, Assembly). Sibling rebuilder is
+    `placement_from_payload`. Hoisted to the VO module to close the
+    duplication anti-hook flagged in `project_mount_frame_design`
+    Watch items.
+    """
+    return {
+        "x": placement.x,
+        "y": placement.y,
+        "z": placement.z,
+        "rx": placement.rx,
+        "ry": placement.ry,
+        "rz": placement.rz,
+        "parent_frame_id": str(placement.parent_frame_id),
+        "reference_surface": placement.reference_surface.value,
+        "tol_x": placement.tol_x,
+        "tol_y": placement.tol_y,
+        "tol_z": placement.tol_z,
+        "tol_rx": placement.tol_rx,
+        "tol_ry": placement.tol_ry,
+        "tol_rz": placement.tol_rz,
+        "units": placement.units.value,
+    }
+
+
+def placement_from_payload(payload: dict[str, object]) -> Placement:
+    """Reconstruct a Placement VO from its JSON payload.
+
+    Sibling of `placement_to_payload`. Round-trips losslessly with it.
+    Re-runs Placement.__post_init__ validation so on-disk corruption
+    (NaN, Inf, negative tolerance) surfaces at load time.
+    """
+    return Placement(
+        x=payload["x"],  # type: ignore[arg-type]
+        y=payload["y"],  # type: ignore[arg-type]
+        z=payload["z"],  # type: ignore[arg-type]
+        rx=payload["rx"],  # type: ignore[arg-type]
+        ry=payload["ry"],  # type: ignore[arg-type]
+        rz=payload["rz"],  # type: ignore[arg-type]
+        parent_frame_id=UUID(str(payload["parent_frame_id"])),
+        reference_surface=ReferenceSurface(str(payload["reference_surface"])),
+        tol_x=payload["tol_x"],  # type: ignore[arg-type]
+        tol_y=payload["tol_y"],  # type: ignore[arg-type]
+        tol_z=payload["tol_z"],  # type: ignore[arg-type]
+        tol_rx=payload["tol_rx"],  # type: ignore[arg-type]
+        tol_ry=payload["tol_ry"],  # type: ignore[arg-type]
+        tol_rz=payload["tol_rz"],  # type: ignore[arg-type]
+        units=UnitSystem(str(payload["units"])),
+    )
+
+
 __all__ = [
     "Placement",
     "ReferenceSurface",
     "UnitSystem",
+    "placement_from_payload",
+    "placement_to_payload",
 ]