cadrumo.application.operations.registry module

Immutable registry for application-owned operation definitions.

class OperationSchemaIdentityV1(**data)[source]

Bases: BaseModel

Stable public identity of one exact strict Pydantic JSON schema.

Parameters:

  • schema_id (OperationPublicSchemaId)

  • schema_version (Annotated[int, FieldInfo(annotation=NoneType, required=True, metadata=[Ge(ge=1)])])

  • schema_fingerprint (ContentDigest)

schema_id: OperationPublicSchemaId
schema_version: Annotated[int, Field(ge=1)]
schema_fingerprint: ContentDigest
classmethod from_model(*, schema_id, schema_version, model_type)[source]

Derive the identity from the canonical closed schema of model_type.

Return type:

OperationSchemaIdentityV1

Parameters:

  • schema_id (OperationPublicSchemaId)

  • schema_version (int)

  • model_type (type[BaseModel])

class OperationPublicDefinitionContractV1(**data)[source]

Bases: BaseModel

Renderer-neutral public manifest row for one operation definition.

Parameters:
manifest_version: Literal[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
class OperationPublicContractSetV1(**data)[source]

Bases: BaseModel

Canonical fixed-point inventory of all public operation contracts.

Parameters:

data (Any)

contract_set_version: Literal[1]
definitions: tuple[OperationPublicDefinitionContractV1, ...]
contract_set_digest: ContentDigest
classmethod build(definitions)[source]

Build the canonical sorted set and its deterministic digest.

Return type:

OperationPublicContractSetV1

Parameters:

definitions (tuple[OperationPublicDefinitionContractV1, ...])

class OperationReconciliationPolicy(*values)[source]

Bases: StrEnum

Closed owner-loss behavior declared by an operation definition.

INTERRUPT
RESUME_FROM_CHECKPOINT
class OperationFrontendProjection(*values)[source]

Bases: StrEnum

Product-owned identities of permitted operation projections.

CLI
MCP
TUI
class OperationExecutorFactory(**data)[source]

Bases: BaseModel

Non-effectful descriptor binding an executor class to its request type.

Parameters:

data (Any)

request_type: type[BaseModel]
executor_type: type[object]
build: Callable[[], object]
create()[source]

Construct and validate the declared executor without running it.

Return type:

OperationExecutor[BaseModel]

class OperationDefinition(**data)[source]

Bases: BaseModel

Complete generic contract registered for one operation type.

Parameters:

data (Any)

definition_id: OperationDefinitionId
request_type: type[BaseModel]
result_type: type[BaseModel] | None
executor_factory: OperationExecutorFactory
phase_codes: tuple[OperationEventCode, ...]
interaction_kinds: frozenset[OperationInteractionKind]
capabilities: OperationCapabilities
reconciliation_policy: OperationReconciliationPolicy
permitted_frontends: frozenset[OperationFrontendProjection]
action_reference: ActionReference | None
ephemeral_secret: OperationEphemeralSecretDeclaration | None
transient_financial_operands: tuple[OperationTransientFinancialOperandDeclaration, ...]
class OperationEffectReceipt(**data)[source]

Bases: 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.

Parameters:

data (Any)

definition_id: OperationDefinitionId
effect: OperationEffect
interrupted: bool
narrowed_from: OperationEffect | None
class OperationSchemaBindingV1(**data)[source]

Bases: BaseModel

Runtime-only binding from a public schema identity to its exact model.

Parameters:
identity: OperationSchemaIdentityV1
model_type: type[BaseModel]
classmethod bind(*, schema_id, schema_version, model_type)[source]

Bind one stable identity to the model that produces its fingerprint.

Return type:

OperationSchemaBindingV1

Parameters:

  • schema_id (OperationPublicSchemaId)

  • schema_version (int)

  • model_type (type[BaseModel])

class OperationReviewProjector(*args, **kwargs)[source]

Bases: Protocol

Domain-owned, side-effect-free safe REVIEW projection contract.

class OperationWorkspaceRefreshAdapter(*args, **kwargs)[source]

Bases: Protocol

Domain-owned adapter from safe terminal facts to a refresh target.

class OperationResultProjector(*args, **kwargs)[source]

Bases: Protocol

Domain-owned, side-effect-free safe settled-result projection contract.

Symmetric with 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.

class OperationPublicDefinitionRegistrationV1(**data)[source]

Bases: BaseModel

Live models and adapters bound to one serializable public contract.

Parameters:
contract: OperationPublicDefinitionContractV1
schema_bindings: tuple[OperationSchemaBindingV1, ...]
reviewed_operand_type: type[BaseModel] | None
review_projector: OperationReviewProjector | None
workspace_refresh_adapter: OperationWorkspaceRefreshAdapter | None
result_projector: OperationResultProjector | None
classmethod compose_request_only(*, definition, request_schema_id, request_schema_version=1)[source]

Bind the common operation shape with no public result or projection.

Return type:

OperationPublicDefinitionRegistrationV1

Parameters:
classmethod compose(*, definition, request_schema, result_schema=None, review_projection_schema=None, interaction_response_schema=None, workspace_refresh_target_schema=None, reviewed_operand_type=None, review_projector=None, workspace_refresh_adapter=None, result_projector=None)[source]

Compose a manifest and its runtime-only bindings from one definition.

Return type:

OperationPublicDefinitionRegistrationV1

Parameters:
class OperationRegistry(**data)[source]

Bases: BaseModel

Deterministic definition registry with fail-closed immutable lookup.

Parameters:

data (Any)

definitions: tuple[OperationDefinition, ...]
public_registrations: tuple[OperationPublicDefinitionRegistrationV1, ...]
property public_contract_set: OperationPublicContractSetV1[source]

Return the validated public set; refuse an uncomposed internal registry.

lookup(definition_id)[source]

Return the exact registered definition or fail closed.

Return type:

OperationDefinition

Parameters:

definition_id (str)

lookup_public_contract(definition_id)[source]

Return the exact live public contract or refuse incomplete composition.

Return type:

OperationPublicDefinitionContractV1

Parameters:

definition_id (str)

lookup_public_registration(definition_id)[source]

Return the sole runtime binding for one public operation definition.

Return type:

OperationPublicDefinitionRegistrationV1

Parameters:

definition_id (str)

lookup_public_schema_binding(identity)[source]

Resolve one exact schema identity without accepting an ID-only match.

Return type:

OperationSchemaBindingV1

Parameters:

identity (OperationSchemaIdentityV1)

lookup_action(action)[source]

Resolve an optional canonical action join without owning its catalogue.

Return type:

OperationDefinition

Parameters:

action (ActionReference)

resolve_credential_free_payload(definition_id, raw)[source]

Hydrate a journal-safe payload only for its exact registered definition.

Return type:

BaseModel

Parameters:
operation_public_schema_reference(identity)[source]

Return the canonical internal reference for one registered public schema.

Return type:

str

Parameters:

identity (OperationSchemaIdentityV1)