"""Immutable registry for application-owned operation definitions."""
from __future__ import annotations
from collections.abc import Callable
from enum import StrEnum
from functools import cached_property
from typing import Annotated, Literal, Protocol, cast, runtime_checkable
from pydantic import (
BaseModel,
ConfigDict,
Field,
field_validator,
model_validator,
)
from ...core.errors.hierarchy import InternalInvariantError, pydantic_validation_boundary
from ...core.hashing import content_hash_hex
from ...core.identity.digest import ContentDigest
from ...core.models import STRICT_FROZEN_CONFIG
from ...core.operations import (
OperationCancellation,
OperationClosePolicy,
OperationDeadline,
OperationDurability,
OperationEffect,
OperationInteractionKind,
)
from ..operator_actions.models import ActionReference
from ._registry_contracts import (
build_public_contract as _build_public_contract,
)
from ._registry_contracts import (
contract_set_digest as _contract_set_digest,
)
from ._registry_contracts import (
definition_contract_digest as _definition_contract_digest,
)
from ._registry_contracts import (
require_positional_callable_signature as _require_positional_callable_signature,
)
from ._registry_contracts import (
validate_public_registration as _validate_public_registration,
)
from .capabilities import (
OperationBaselinePolicy,
OperationCapabilities,
OperationConflictScope,
OperationOwnedResource,
OperationReplayPolicy,
OperationRequestStoragePolicy,
OperationSensitiveInputPolicy,
)
from .events import OperationEventCode
from .financial_operand import OperationTransientFinancialOperandDeclaration
from .interactions import OperationInteractionRequest
from .models import (
CredentialFreeOperationRequest,
OperationDefinitionId,
OperationTerminalReceipt,
)
from .owner import OperationExecutor, OperationResumableExecutor
from .registry_schema_validation import strict_model_json_schema, validate_credential_free_schema
from .secret_submission import OperationEphemeralSecretDeclaration
_STRICT_RUNTIME_BINDING_CONFIG = ConfigDict(
strict=True,
frozen=True,
extra="forbid",
arbitrary_types_allowed=True,
)
_STRICT_PUBLIC_MODEL_CONFIG = ConfigDict(
strict=True,
frozen=True,
extra="forbid",
validate_default=True,
)
type OperationPublicSchemaId = Annotated[
str,
Field(min_length=3, max_length=160, pattern=r"^[a-z][a-z0-9]*(?:[._-][a-z0-9]+)+$"),
]
[docs]
class OperationSchemaIdentityV1(BaseModel):
"""Stable public identity of one exact strict Pydantic JSON schema."""
model_config = _STRICT_PUBLIC_MODEL_CONFIG
schema_id: OperationPublicSchemaId
schema_version: Annotated[int, Field(ge=1)]
schema_fingerprint: ContentDigest
[docs]
@classmethod
def from_model(
cls,
*,
schema_id: OperationPublicSchemaId,
schema_version: int,
model_type: type[BaseModel],
) -> OperationSchemaIdentityV1:
"""Derive the identity from the canonical closed schema of ``model_type``."""
schema = strict_model_json_schema(model_type)
return cls(
schema_id=schema_id,
schema_version=schema_version,
schema_fingerprint=content_hash_hex(schema),
)
[docs]
class OperationPublicDefinitionContractV1(BaseModel):
"""Renderer-neutral public manifest row for one operation definition."""
model_config = _STRICT_PUBLIC_MODEL_CONFIG
manifest_version: Literal[1] = 1
definition_id: OperationDefinitionId
action_reference: ActionReference | None
request_schema: OperationSchemaIdentityV1
result_schema: OperationSchemaIdentityV1 | None
review_projection_schema: OperationSchemaIdentityV1 | None
interaction_response_schema: OperationSchemaIdentityV1 | None
workspace_refresh_target_schema: OperationSchemaIdentityV1 | None
interaction_kinds: frozenset[OperationInteractionKind]
request_storage: OperationRequestStoragePolicy
durability: OperationDurability
cancellation: OperationCancellation
deadline: OperationDeadline
replay: OperationReplayPolicy
baseline: OperationBaselinePolicy
sensitive_input: OperationSensitiveInputPolicy
conflict_scope: OperationConflictScope
owned_resources: frozenset[OperationOwnedResource]
permitted_effects: frozenset[OperationEffect]
close_policy: OperationClosePolicy
reconciliation_policy: OperationReconciliationPolicy
permitted_frontends: frozenset[OperationFrontendProjection]
ephemeral_secret_required: bool
definition_contract_digest: ContentDigest
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate_digest(self) -> OperationPublicDefinitionContractV1:
expected = _definition_contract_digest(self)
if self.definition_contract_digest != expected:
raise ValueError("operation definition contract digest does not reproduce")
return self
[docs]
class OperationPublicContractSetV1(BaseModel):
"""Canonical fixed-point inventory of all public operation contracts."""
model_config = _STRICT_PUBLIC_MODEL_CONFIG
contract_set_version: Literal[1] = 1
definitions: tuple[OperationPublicDefinitionContractV1, ...] = Field(min_length=1)
contract_set_digest: ContentDigest
@field_validator("definitions")
@classmethod
@pydantic_validation_boundary
def _canonical_contracts(
cls,
value: tuple[OperationPublicDefinitionContractV1, ...],
) -> tuple[OperationPublicDefinitionContractV1, ...]:
definition_ids = tuple(contract.definition_id for contract in value)
if len(set(definition_ids)) != len(definition_ids):
raise ValueError("public operation definition IDs must be unique")
if definition_ids != tuple(sorted(definition_ids)):
raise ValueError("public operation contracts must be sorted by definition ID")
return value
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate_digest(self) -> OperationPublicContractSetV1:
expected = _contract_set_digest(self.definitions)
if self.contract_set_digest != expected:
raise ValueError("operation public contract-set digest does not reproduce")
return self
[docs]
@classmethod
def build(
cls,
definitions: tuple[OperationPublicDefinitionContractV1, ...],
) -> OperationPublicContractSetV1:
"""Build the canonical sorted set and its deterministic digest."""
canonical = tuple(sorted(definitions, key=lambda item: item.definition_id))
return cls(definitions=canonical, contract_set_digest=_contract_set_digest(canonical))
[docs]
class OperationReconciliationPolicy(StrEnum):
"""Closed owner-loss behavior declared by an operation definition."""
INTERRUPT = "interrupt"
RESUME_FROM_CHECKPOINT = "resume_from_checkpoint"
[docs]
class OperationFrontendProjection(StrEnum):
"""Product-owned identities of permitted operation projections."""
CLI = "cli"
MCP = "mcp"
TUI = "tui"
[docs]
class OperationExecutorFactory(BaseModel):
"""Non-effectful descriptor binding an executor class to its request type."""
model_config = STRICT_FROZEN_CONFIG
request_type: type[BaseModel]
executor_type: type[object]
build: Callable[[], object]
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate_executor_type(self) -> OperationExecutorFactory:
if not issubclass(self.executor_type, OperationExecutor):
raise ValueError("operation executor type must structurally implement OperationExecutor")
return self
[docs]
def create(self) -> OperationExecutor[BaseModel]:
"""Construct and validate the declared executor without running it."""
executor = self.build()
if not isinstance(executor, self.executor_type) or not isinstance(executor, OperationExecutor):
raise TypeError("operation executor factory returned an undeclared or invalid executor")
return cast(OperationExecutor[BaseModel], executor)
[docs]
class OperationDefinition(BaseModel):
"""Complete generic contract registered for one operation type."""
model_config = STRICT_FROZEN_CONFIG
definition_id: OperationDefinitionId
request_type: type[BaseModel]
result_type: type[BaseModel] | None
executor_factory: OperationExecutorFactory
phase_codes: tuple[OperationEventCode, ...] = Field(min_length=1)
interaction_kinds: frozenset[OperationInteractionKind]
capabilities: OperationCapabilities
reconciliation_policy: OperationReconciliationPolicy
permitted_frontends: frozenset[OperationFrontendProjection] = Field(min_length=1)
action_reference: ActionReference | None = None
ephemeral_secret: OperationEphemeralSecretDeclaration | None = None
transient_financial_operands: tuple[OperationTransientFinancialOperandDeclaration, ...] = ()
@field_validator("phase_codes")
@classmethod
@pydantic_validation_boundary
def _canonical_phase_codes(cls, value: tuple[OperationEventCode, ...]) -> tuple[OperationEventCode, ...]:
if len(set(value)) != len(value):
raise ValueError("operation definition phase codes must be unique")
return tuple(sorted(value))
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate_factory_request_type(self) -> OperationDefinition:
if self.executor_factory.request_type is not self.request_type:
raise ValueError("operation executor factory request type must match the definition request type")
self._validate_request_storage()
self._validate_ephemeral_secret()
self._validate_transient_financial_operands()
if (
self.capabilities.durability is not OperationDurability.EPHEMERAL
and OperationEffect.UNKNOWN not in self.capabilities.permitted_effects
):
raise ValueError("operation definition must permit unknown effect for owner-loss reconciliation")
if self.reconciliation_policy is OperationReconciliationPolicy.RESUME_FROM_CHECKPOINT:
if self.capabilities.durability is not OperationDurability.RESUMABLE:
raise ValueError("checkpoint reconciliation requires resumable durability")
if not self.interaction_kinds:
raise ValueError("checkpoint reconciliation requires a declared interaction checkpoint")
if not issubclass(self.executor_factory.executor_type, OperationResumableExecutor):
raise ValueError("checkpoint reconciliation requires a resumable executor")
return self
def _validate_request_storage(self) -> None:
if self.capabilities.request_storage is not OperationRequestStoragePolicy.CREDENTIAL_FREE_JOURNAL:
return
if not issubclass(self.request_type, CredentialFreeOperationRequest):
raise ValueError(
"credential-free journal request type must explicitly inherit CredentialFreeOperationRequest"
)
schema = strict_model_json_schema(self.request_type)
validate_credential_free_schema(schema)
def _validate_ephemeral_secret(self) -> None:
if self.ephemeral_secret is None:
return
if self.capabilities.durability is not OperationDurability.RECORDED:
raise ValueError("ephemeral secret operations require recorded durability")
if self.reconciliation_policy is not OperationReconciliationPolicy.INTERRUPT:
raise ValueError("ephemeral secret operations cannot resume after owner loss")
if OperationEffect.NONE not in self.capabilities.permitted_effects:
raise ValueError("ephemeral secret operations must permit a pre-entry none effect")
def _validate_transient_financial_operands(self) -> None:
"""Refuse an operand declaration the runtime could not honour.
An operand lives only in the memory of the process that received it, so
a definition that expects to resume after owner loss is declaring
something custody cannot deliver: the restart would have to invent the
amount or the acknowledgement.
"""
if not self.transient_financial_operands:
return
kinds = [declaration.operand_kind for declaration in self.transient_financial_operands]
if len(set(kinds)) != len(kinds):
raise ValueError("operation definition cannot declare one financial operand kind twice")
if self.capabilities.durability is not OperationDurability.RECORDED:
raise ValueError("transient financial operand operations require recorded durability")
if self.reconciliation_policy is not OperationReconciliationPolicy.INTERRUPT:
raise ValueError("transient financial operand operations cannot resume after owner loss")
if OperationInteractionKind.INPUT not in self.interaction_kinds:
raise ValueError("transient financial operand operations must declare an input interaction")
if OperationEffect.UNKNOWN not in self.capabilities.permitted_effects:
raise ValueError("transient financial operand operations must permit an uncertain-delivery effect")
[docs]
class OperationEffectReceipt(BaseModel):
"""What committed evidence lets an operation claim about its own effect.
A claim is narrowed, never widened. An executor that reports it changed
nothing is believed; one that reports a mutation is held to the evidence
the application actually committed, because an operation interrupted
mid-flight cannot know on its own whether its write landed.
"""
model_config = STRICT_FROZEN_CONFIG
definition_id: OperationDefinitionId
effect: OperationEffect
interrupted: bool
narrowed_from: OperationEffect | None = None
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate_narrowing(self) -> OperationEffectReceipt:
if self.narrowed_from is not None and self.narrowed_from is self.effect:
raise ValueError("an effect receipt records a narrowing only when the claim actually changed")
return self
[docs]
class OperationSchemaBindingV1(BaseModel):
"""Runtime-only binding from a public schema identity to its exact model."""
model_config = _STRICT_RUNTIME_BINDING_CONFIG
identity: OperationSchemaIdentityV1
model_type: type[BaseModel]
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate_fingerprint(self) -> OperationSchemaBindingV1:
schema = strict_model_json_schema(self.model_type)
fingerprint = content_hash_hex(schema)
if fingerprint != self.identity.schema_fingerprint:
raise ValueError("registered operation schema fingerprint does not match its exact model")
return self
[docs]
@classmethod
def bind(
cls,
*,
schema_id: OperationPublicSchemaId,
schema_version: int,
model_type: type[BaseModel],
) -> OperationSchemaBindingV1:
"""Bind one stable identity to the model that produces its fingerprint."""
return cls(
identity=OperationSchemaIdentityV1.from_model(
schema_id=schema_id,
schema_version=schema_version,
model_type=model_type,
),
model_type=model_type,
)
[docs]
@runtime_checkable
class OperationReviewProjector(Protocol):
"""Domain-owned, side-effect-free safe REVIEW projection contract."""
def __call__(
self,
reviewed_operand: BaseModel,
_interaction_facts: OperationInteractionRequest,
/,
) -> BaseModel:
"""Project one resolved operand and its current interaction facts."""
...
[docs]
@runtime_checkable
class OperationWorkspaceRefreshAdapter(Protocol):
"""Domain-owned adapter from safe terminal facts to a refresh target."""
def __call__(self, terminal_receipt: OperationTerminalReceipt, /) -> BaseModel:
"""Return the typed target derived from one settled receipt."""
...
[docs]
@runtime_checkable
class OperationResultProjector(Protocol):
"""Domain-owned, side-effect-free safe settled-result projection contract.
Symmetric with :class:`OperationReviewProjector`: the resolver reloads the
private settled result behind the secure application port and hands it,
plus the safe terminal receipt, to this projector -- never the reverse.
Registered only when the public result schema is a distinct projection of
the definition's private result type; a result schema identical to that
private type declares no projector.
"""
def __call__(self, result: BaseModel, terminal_receipt: OperationTerminalReceipt, /) -> BaseModel:
"""Project one resolved settled result and its safe terminal receipt."""
...
[docs]
class OperationPublicDefinitionRegistrationV1(BaseModel):
"""Live models and adapters bound to one serializable public contract."""
model_config = _STRICT_RUNTIME_BINDING_CONFIG
contract: OperationPublicDefinitionContractV1
schema_bindings: tuple[OperationSchemaBindingV1, ...] = Field(min_length=1)
reviewed_operand_type: type[BaseModel] | None = None
review_projector: OperationReviewProjector | None = None
workspace_refresh_adapter: OperationWorkspaceRefreshAdapter | None = None
result_projector: OperationResultProjector | None = None
@field_validator("schema_bindings")
@classmethod
@pydantic_validation_boundary
def _unique_schema_bindings(
cls,
value: tuple[OperationSchemaBindingV1, ...],
) -> tuple[OperationSchemaBindingV1, ...]:
keys = tuple((binding.identity.schema_id, binding.identity.schema_version) for binding in value)
if len(set(keys)) != len(keys):
raise ValueError("registered operation schema identities must be unique")
return tuple(sorted(value, key=lambda item: (item.identity.schema_id, item.identity.schema_version)))
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate_adapter_signatures(self) -> OperationPublicDefinitionRegistrationV1:
if self.review_projector is not None:
_require_positional_callable_signature(self.review_projector, arity=2, label="REVIEW projector")
if self.workspace_refresh_adapter is not None:
_require_positional_callable_signature(
self.workspace_refresh_adapter,
arity=1,
label="Workspace refresh adapter",
)
return self
[docs]
@classmethod
def compose_request_only(
cls,
*,
definition: OperationDefinition,
request_schema_id: OperationPublicSchemaId,
request_schema_version: int = 1,
) -> OperationPublicDefinitionRegistrationV1:
"""Bind the common operation shape with no public result or projection."""
return cls.compose(
definition=definition,
request_schema=OperationSchemaBindingV1.bind(
schema_id=request_schema_id,
schema_version=request_schema_version,
model_type=definition.request_type,
),
)
[docs]
@classmethod
def compose(
cls,
*,
definition: OperationDefinition,
request_schema: OperationSchemaBindingV1,
result_schema: OperationSchemaBindingV1 | None = None,
review_projection_schema: OperationSchemaBindingV1 | None = None,
interaction_response_schema: OperationSchemaBindingV1 | None = None,
workspace_refresh_target_schema: OperationSchemaBindingV1 | None = None,
reviewed_operand_type: type[BaseModel] | None = None,
review_projector: OperationReviewProjector | None = None,
workspace_refresh_adapter: OperationWorkspaceRefreshAdapter | None = None,
result_projector: OperationResultProjector | None = None,
) -> OperationPublicDefinitionRegistrationV1:
"""Compose a manifest and its runtime-only bindings from one definition."""
bindings = tuple(
binding
for binding in (
request_schema,
result_schema,
review_projection_schema,
interaction_response_schema,
workspace_refresh_target_schema,
)
if binding is not None
)
contract = _public_contract_for_definition(
definition,
request_schema=request_schema.identity,
result_schema=result_schema.identity if result_schema is not None else None,
review_projection_schema=(
review_projection_schema.identity if review_projection_schema is not None else None
),
interaction_response_schema=(
interaction_response_schema.identity if interaction_response_schema is not None else None
),
workspace_refresh_target_schema=(
workspace_refresh_target_schema.identity if workspace_refresh_target_schema is not None else None
),
)
return cls(
contract=contract,
schema_bindings=bindings,
reviewed_operand_type=reviewed_operand_type,
review_projector=review_projector,
workspace_refresh_adapter=workspace_refresh_adapter,
result_projector=result_projector,
)
[docs]
class OperationRegistry(BaseModel):
"""Deterministic definition registry with fail-closed immutable lookup."""
model_config = STRICT_FROZEN_CONFIG
definitions: tuple[OperationDefinition, ...] = Field(min_length=1)
public_registrations: tuple[OperationPublicDefinitionRegistrationV1, ...] = ()
@field_validator("definitions")
@classmethod
@pydantic_validation_boundary
def _canonical_definitions(cls, value: tuple[OperationDefinition, ...]) -> tuple[OperationDefinition, ...]:
definition_ids = tuple(item.definition_id for item in value)
if len(set(definition_ids)) != len(definition_ids):
raise ValueError("operation definition IDs must be unique")
action_ids = tuple(item.action_reference.action_id for item in value if item.action_reference is not None)
if len(set(action_ids)) != len(action_ids):
raise ValueError("operator action references must map to at most one operation definition")
return tuple(sorted(value, key=lambda item: item.definition_id))
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate_public_fixed_point(self) -> OperationRegistry:
if not self.public_registrations:
return self
registrations = tuple(sorted(self.public_registrations, key=lambda item: item.contract.definition_id))
if registrations != self.public_registrations:
raise ValueError("public operation registrations must be sorted by definition ID")
definition_ids = tuple(definition.definition_id for definition in self.definitions)
registration_ids = tuple(registration.contract.definition_id for registration in registrations)
if registration_ids != definition_ids:
raise ValueError("public operation registrations must exactly cover the immutable registry")
seen_schema_identities: dict[tuple[str, int], tuple[ContentDigest, type[BaseModel]]] = {}
contracts: list[OperationPublicDefinitionContractV1] = []
for definition, registration in zip(self.definitions, registrations, strict=True):
self._validate_public_registration(definition, registration)
contracts.append(registration.contract)
for binding in registration.schema_bindings:
key = (binding.identity.schema_id, binding.identity.schema_version)
current = (binding.identity.schema_fingerprint, binding.model_type)
existing = seen_schema_identities.setdefault(key, current)
if existing != current:
raise ValueError("one operation schema identity must bind one exact model and fingerprint")
OperationPublicContractSetV1.build(tuple(contracts))
return self
@staticmethod
def _validate_public_registration(
definition: OperationDefinition,
registration: OperationPublicDefinitionRegistrationV1,
) -> None:
_validate_public_registration(
definition,
registration,
contract_builder=_public_contract_for_definition,
)
[docs]
@cached_property
def public_contract_set(self) -> OperationPublicContractSetV1:
"""Return the validated public set; refuse an uncomposed internal registry."""
if not self.public_registrations:
raise InternalInvariantError("operation registry has no public contract composition")
return OperationPublicContractSetV1.build(
tuple(registration.contract for registration in self.public_registrations),
)
[docs]
def lookup(self, definition_id: str) -> OperationDefinition:
"""Return the exact registered definition or fail closed."""
for definition in self.definitions:
if definition.definition_id == definition_id:
return definition
raise KeyError(f"unknown operation definition ID: {definition_id!r}")
[docs]
def lookup_public_contract(self, definition_id: str) -> OperationPublicDefinitionContractV1:
"""Return the exact live public contract or refuse incomplete composition."""
for registration in self.public_registrations:
if registration.contract.definition_id == definition_id:
return registration.contract
raise KeyError(f"operation definition has no public contract: {definition_id!r}")
[docs]
def lookup_public_registration(self, definition_id: str) -> OperationPublicDefinitionRegistrationV1:
"""Return the sole runtime binding for one public operation definition."""
for registration in self.public_registrations:
if registration.contract.definition_id == definition_id:
return registration
raise KeyError(f"operation definition has no public registration: {definition_id!r}")
[docs]
def lookup_public_schema_binding(self, identity: OperationSchemaIdentityV1) -> OperationSchemaBindingV1:
"""Resolve one exact schema identity without accepting an ID-only match."""
for registration in self.public_registrations:
for binding in registration.schema_bindings:
if binding.identity == identity:
return binding
raise KeyError(
"operation public schema identity is not registered: "
f"{identity.schema_id!r} version {identity.schema_version}"
)
[docs]
def lookup_action(self, action: ActionReference) -> OperationDefinition:
"""Resolve an optional canonical action join without owning its catalogue."""
for definition in self.definitions:
if definition.action_reference == action:
return definition
raise KeyError(f"operator action is not mapped to an operation definition: {action.action_id!r}")
[docs]
def resolve_credential_free_payload(self, definition_id: str, raw: str | bytes) -> BaseModel:
"""Hydrate a journal-safe payload only for its exact registered definition."""
definition = self.lookup(definition_id)
if definition.capabilities.request_storage is not OperationRequestStoragePolicy.CREDENTIAL_FREE_JOURNAL:
raise ValueError("operation definition does not use credential-free journal request storage")
return definition.request_type.model_validate_json(raw)
[docs]
def operation_public_schema_reference(identity: OperationSchemaIdentityV1) -> str:
"""Return the canonical internal reference for one registered public schema."""
return f"schema:{identity.schema_id}.v{identity.schema_version}"
def _public_contract_for_definition(
definition: OperationDefinition,
*,
request_schema: OperationSchemaIdentityV1,
result_schema: OperationSchemaIdentityV1 | None,
review_projection_schema: OperationSchemaIdentityV1 | None,
interaction_response_schema: OperationSchemaIdentityV1 | None,
workspace_refresh_target_schema: OperationSchemaIdentityV1 | None,
) -> OperationPublicDefinitionContractV1:
return cast(
OperationPublicDefinitionContractV1,
_build_public_contract(
contract_type=OperationPublicDefinitionContractV1,
definition=definition,
request_schema=request_schema,
result_schema=result_schema,
review_projection_schema=review_projection_schema,
interaction_response_schema=interaction_response_schema,
workspace_refresh_target_schema=workspace_refresh_target_schema,
),
)
OperationPublicDefinitionContractV1.model_rebuild()
__all__ = [
"OperationDefinition",
"OperationEffectReceipt",
"OperationExecutorFactory",
"OperationFrontendProjection",
"OperationPublicContractSetV1",
"OperationPublicDefinitionContractV1",
"OperationPublicDefinitionRegistrationV1",
"OperationPublicSchemaId",
"OperationReconciliationPolicy",
"OperationRegistry",
"OperationResultProjector",
"OperationReviewProjector",
"OperationSchemaBindingV1",
"OperationSchemaIdentityV1",
"OperationWorkspaceRefreshAdapter",
"operation_public_schema_reference",
]