Feat(core): [1083] transformation system packer

diracx-core/src/diracx/core/exceptions.py

@@ -8,6 +8,7 @@
     "DocumentUpsertError",
     "IAMClientError",
     "IAMServerError",
+    "InputStatusConflictError",
     "InvalidCredentialsError",
     "InvalidQueryError",
     "NotReadyError",
@@ -19,6 +20,7 @@
     "SandboxAlreadyInsertedError",
     "SandboxNotFoundError",
     "TokenNotFoundError",
+    "TransformationNotFoundError",
 ]
 
 
@@ -185,3 +187,33 @@ class PilotAlreadyExistsError(DiracError):
 
 class PilotAlreadyAssociatedWithJobError(DiracError):
     """We can't associate a pilot with the same job twice."""
+
+
+class TransformationNotFoundError(DiracError):
+    """Raised when a transformation cannot be found.
+
+    Attributes:
+        transformation_id: Identifier of the missing transformation.
+    """
+
+    def __init__(self, transformation_id, detail: str = ""):
+        self.transformation_id = transformation_id
+        super().__init__(
+            f"Transformation {transformation_id} not found"
+            + (f" ({detail})" if detail else "")
+        )
+
+
+class InputStatusConflictError(DiracError):
+    """Raised when transformation inputs are not in the status a change expects.
+
+    Attributes:
+        transformation_id: Transformation the inputs belong to.
+    """
+
+    def __init__(self, transformation_id, detail: str = ""):
+        self.transformation_id = transformation_id
+        super().__init__(
+            f"Inputs of transformation {transformation_id} changed status"
+            + (f" ({detail})" if detail else "")
+        )

diracx-core/src/diracx/core/extensions.py

@@ -34,6 +34,7 @@ class DiracEntryPoint(StrEnum):
     RESOURCES = "diracx.resources"
     SERVICES = "diracx.services"
     LOCK_OBJECT_TYPES = "diracx.lock_object_types"
+    TRANSFORMATION_PACKERS = "diracx.transformations.packers"
 
 
 @cached(cache=LRUCache(maxsize=1))

diracx-core/src/diracx/core/models/__init__.py

@@ -14,8 +14,12 @@
     "WORKGRAPH_TRANSITIONS",
     "AccessTokenPayload",
     "ActionResult",
+    "BadInput",
     "ChecksumAlgorithm",
+    "ComputeParcel",
     "CounterEntity",
+    "DataParcel",
+    "DelayInput",
     "GrantType",
     "GroupInfo",
     "HeartbeatData",
@@ -33,6 +37,8 @@
     "JobStatusUpdate",
     "Metadata",
     "OpenIDConfiguration",
+    "PackerInput",
+    "PackerProposal",
     "ParcelStatus",
     "RefreshTokenPayload",
     "ReplicaMap",
@@ -48,6 +54,7 @@
     "SetJobStatusReturn",
     "SortDirection",
     "SortSpec",
+    "StorageDelta",
     "SummaryParams",
     "TokenPayload",
     "TokenResponse",
@@ -114,9 +121,16 @@
     WORKGRAPH_MEMBER_STATUSES,
     WORKGRAPH_TRANSITIONS,
     ActionResult,
+    BadInput,
+    ComputeParcel,
     CounterEntity,
+    DataParcel,
+    DelayInput,
     InputStatus,
+    PackerInput,
+    PackerProposal,
     ParcelStatus,
+    StorageDelta,
     TransformationKind,
     TransformationStatus,
     WorkgraphStatus,

diracx-core/src/diracx/core/models/transformation.py

@@ -1,4 +1,7 @@
-"""Enumerations of the Transformation System (DX-ADR-004, DX-ADR-005).
+"""Enumerations and plugin contracts of the Transformation System.
+
+The enumerations and state machines follow DX-ADR-004 and DX-ADR-005; the
+packer contract at the end of the module follows DX-ADR-006.
 
 The members of each status enum are written in state-machine order and new
 states must only ever be appended: the database stores them as native
@@ -17,6 +20,12 @@
 from __future__ import annotations
 
 from enum import StrEnum
+from typing import Any
+from uuid import UUID
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from .types import UTCDatetime
 
 
 class WorkgraphStatus(StrEnum):
@@ -290,3 +299,106 @@ class CounterEntity(StrEnum):
     ),
     WorkgraphStatus.CANCELLED: frozenset({TransformationStatus.CANCELLED}),
 }
+
+
+# --------------------------------------------------------------------------
+# Packer contract (DX-ADR-006)
+# --------------------------------------------------------------------------
+
+
+class PackerInput(BaseModel):
+    """An ``Unassigned`` input claimed by the core and offered to a packer.
+
+    Attributes:
+        input_id: Identifier of the input.
+        parent_input_id: Input this one was split from, if any.
+        lfn: Logical file name, if the input is a file.
+        lfn_size: Size of the whole file, whatever the descriptor selects.
+        descriptor: Everything else describing the input.
+        error_count: Failed parcels the input was part of since the last reset.
+    """
+
+    model_config = ConfigDict(frozen=True)
+
+    input_id: UUID
+    parent_input_id: UUID | None = None
+    lfn: str | None = None
+    lfn_size: int | None = None
+    descriptor: dict[str, Any] = {}
+    error_count: int = 0
+
+
+class ComputeParcel(BaseModel):
+    """A packer's proposal for a parcel that runs as a job.
+
+    Attributes:
+        inputs: Inputs the parcel is made of.
+        metadata: Free-form annotations stored on the parcel.
+        requirements: Additions merged over the transformation's
+            requirements template.
+        parameters: The CWL input object passed to the runner.
+    """
+
+    inputs: list[UUID] = Field(min_length=1)
+    metadata: dict[str, Any] = {}
+    requirements: dict[str, Any] | None = None
+    parameters: dict[str, Any] = {}
+
+
+class StorageDelta(BaseModel):
+    """What a data parcel's request costs one storage element.
+
+    Attributes:
+        se: Storage element affected.
+        files: Files added there if positive, freed if negative.
+        bytes: Bytes added there if positive, freed if negative.
+    """
+
+    se: str
+    files: int
+    bytes: int
+
+
+class DataParcel(BaseModel):
+    """A packer's proposal for a parcel that runs as a request.
+
+    Attributes:
+        inputs: Inputs the parcel is made of.
+        metadata: Free-form annotations stored on the parcel.
+        request: DIRAC RMS request body (DX-ADR-003).
+        deltas: What the request costs each storage element.
+    """
+
+    inputs: list[UUID] = Field(min_length=1)
+    metadata: dict[str, Any] = {}
+    request: dict[str, Any]
+    deltas: list[StorageDelta] = []
+
+
+class BadInput(BaseModel):
+    """A packer's verdict that an input cannot be used at all.
+
+    The core moves the input to ``Problematic`` with the reason.
+
+    Attributes:
+        input_id: The input rejected.
+        reason: Why the packer rejected it.
+    """
+
+    input_id: UUID
+    reason: str = "Unknown reason"
+
+
+class DelayInput(BaseModel):
+    """A packer's request to withhold inputs from packing for a while.
+
+    Attributes:
+        input_ids: The inputs to withhold.
+        delay_until: Time before which they are not offered again.
+    """
+
+    input_ids: list[UUID] = Field(min_length=1)
+    delay_until: UTCDatetime
+
+
+PackerProposal = ComputeParcel | DataParcel | BadInput | DelayInput

diracx-core/tests/test_transformation_models.py

@@ -1,6 +1,9 @@
 from __future__ import annotations
 
+from uuid import uuid4
+
 import pytest
+from pydantic import ValidationError
 
 from diracx.core.models.transformation import (
     INPUT_TRANSITIONS,
@@ -9,8 +12,11 @@
     WORKGRAPH_MEMBER_STATUSES,
     WORKGRAPH_TRANSITIONS,
     ActionResult,
+    ComputeParcel,
     CounterEntity,
+    DelayInput,
     InputStatus,
+    PackerInput,
     ParcelStatus,
     TransformationKind,
     TransformationStatus,
@@ -66,3 +72,17 @@ def test_enum_values_are_append_only(enum_type):
 )
 def test_state_tables_cover_every_state(transitions, enum_type):
     assert set(transitions) == set(enum_type)
+
+
+def test_packer_proposals_need_inputs():
+    with pytest.raises(ValidationError):
+        ComputeParcel(inputs=[])
+    with pytest.raises(ValidationError):
+        DelayInput(input_ids=[], delay_until="2026-01-01T00:00:00Z")
+
+
+def test_packer_input_is_frozen():
+    packer_input = PackerInput(input_id=uuid4(), lfn="/vo/file")
+    assert packer_input.descriptor == {}
+    with pytest.raises(ValidationError):
+        packer_input.lfn = "/vo/other"  # type: ignore[misc]