cadrumo.application.operations.persistence.journal module

Application ports for durable operation state, events, leases, and operands.

class OperationPersistedSnapshot(**data)[source]

Bases: BaseModel

Credential-free state and event batch written for one journal transition.

The runtime OperationSnapshot retains its concrete registered request payload. This record instead retains only the content digest that addresses that confidential operand in secure storage.

Parameters:

data (Any)

schema_version: Literal[6]
identity: OperationIdentity
definition_contract_digest: ContentDigest
request_storage: OperationRequestStoragePolicy
request_reference: ContentDigest
credential_free_request_json: str | None
secret_requirement: OperationSecretRequirement | None
executor_entered_at: datetime | None
revision: OperationRevision
lifecycle: OperationLifecycle
terminal_condition: OperationTerminalCondition | None
effect: OperationEffect
phase_code: OperationEventCode | None
started_at: datetime
updated_at: datetime
execution_deadline: datetime | None
cleanup_deadline: datetime | None
cancellation_requested_at: datetime | None
cancellation_acknowledged_at: datetime | None
cancellation_deferred: bool
event_cursor: OperationEventCursor
terminal_receipt: OperationTerminalReceipt | None
events: tuple[OperationEvent, ...]
idempotency_claim: OperationIdempotencyClaim | None
pending_interaction: OperationPendingInteraction | None
consumed_interactions: tuple[OperationConsumedInteraction, ...]
property operation_id: OperationId

Return the filename identity consumed by the journal substrate.

class OperationProgressFoldCheckpoint(**data)[source]

Bases: BaseModel

Compaction-safe progress state through one authoritative event cursor.

Parameters:

data (Any)

identity: OperationIdentity
through_cursor: OperationEventCursor
phase_code: OperationEventCode | None
progress_event: OperationProgressEvent | None
class OperationProgressFoldInput(**data)[source]

Bases: BaseModel

Exact checkpoint suffix needed to fold current progress through an anchor.

Parameters:

data (Any)

checkpoint: OperationProgressFoldCheckpoint | None
events: tuple[OperationEvent, ...]
class OperationObservationMaterialization(**data)[source]

Bases: BaseModel

Snapshot, replay page, and progress input from one atomic journal read.

Parameters:

data (Any)

snapshot: OperationPersistedSnapshot
anchor_cursor: OperationEventCursor
replay: OperationReplayPage
progress_fold: OperationProgressFoldInput
exception OperationObservationUnknownOperationError(operation_id)[source]

Bases: CadrumoError

The locked observation read found no journal for the requested operation.

Parameters:

operation_id (OperationId)

Return type:

None

code
exception OperationObservationCursorAheadError(*, requested_cursor, anchor_cursor)[source]

Bases: CadrumoError

The caller cursor is beyond the authoritative anchor read under the journal lock.

Parameters:

  • requested_cursor (OperationEventCursor)

  • anchor_cursor (OperationEventCursor)

Return type:

None

code
class OperationJournal(*args, **kwargs)[source]

Bases: Protocol

Atomic snapshot-plus-event persistence with optimistic revision checks.

async load(operation_id)[source]

Load the latest persisted snapshot for one operation.

Return type:

OperationPersistedSnapshot

Parameters:

operation_id (OperationId)

async resolve_idempotency(claim)[source]

Resolve a claim only when its matching operation journal exists.

Return type:

TypeAliasType | None

Parameters:

claim (OperationIdempotencyClaim)

async create(snapshot, *, lease)[source]

Atomically create one initial snapshot and its optional retry claim.

Return type:

TypeAliasType

Parameters:
async commit(snapshot, *, expected_revision, lease)[source]

Atomically compare revision and append the snapshot and ordered events.

Return type:

None

Parameters:
class OperationEventStream(*args, **kwargs)[source]

Bases: Protocol

Replay ordered operation events from an exclusive cursor.

async read_after(operation_id, cursor, *, limit)[source]

Replay the bounded event interval after one exclusive cursor.

Return type:

OperationReplayPage

Parameters:

  • operation_id (OperationId)

  • cursor (OperationEventCursor)

  • limit (OperationReplayLimit)

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

Bases: Protocol

Read one internally consistent operation observation from persistence.

A missing operation raises OperationObservationUnknownOperationError. A cursor beyond the read anchor raises OperationObservationCursorAheadError.

async read_observation(operation_id, after_cursor, *, limit)[source]

Read one atomic snapshot and its bounded event page.

Return type:

OperationObservationMaterialization

Parameters:

  • operation_id (OperationId)

  • after_cursor (OperationEventCursor)

  • limit (OperationReplayLimit)

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

Bases: Protocol

Observe and transition one durable owner lease against explicit evidence time.

async inspect(scope_ref, operation_id, *, observed_at)[source]

Observe the current lease state for one conflict scope.

Return type:

OperationLeaseObservation

Parameters:

  • scope_ref (OperationConflictScopeReference)

  • operation_id (OperationId)

  • observed_at (datetime)

async acquire(candidate, *, observed_at)[source]

Acquire a candidate owner lease against explicit evidence time.

Return type:

OperationLeaseResult

Parameters:
async compare_and_swap(predecessor, successor, *, observed_at)[source]

Replace one exact live lease with its verified successor.

Return type:

OperationLeaseResult

Parameters:
async release(predecessor, *, observed_at)[source]

Release one exact current lease against explicit evidence time.

Return type:

OperationLeaseResult

Parameters:
class OperationSecureReferenceStore(*args, **kwargs)[source]

Bases: Protocol

Store and resolve confidential operands outside credential-free journals.

async put(operand, *, written_at)[source]

Store one typed confidential operand and return its reference.

Return type:

TypeAliasType

Parameters:

  • operand (BaseModel)

  • written_at (datetime)

async resolve(reference, operand_type)[source]

Resolve one typed confidential operand by its content reference.

Return type:

TypeVar(OperandT, bound= BaseModel)

Parameters: