cadrumo.application.workflow.persistence module¶

Encrypted persistence for workflow state and workflow runs.

Workflow state is stored as an Envelope-wrapped record in the secure-object backend. The load path deserialises the envelope and validates it; callers receive a typed WorkflowState or a diagnostic error class rather than a raw payload.

The envelope carries the SensitivityClass this store writes under, so the classification travels with the record rather than being re-decided at each write site.

See also

WorkflowState: Typed encrypted state payload persisted by WorkflowStateRepository.
WorkflowStateResetFingerprint: Row-level, plaintext-free reset audit summary emitted before deletion.
application.workflow.events.emit_workflow_state_reset(): Writes the append-only workflow_state.reset bucket event before the state row is removed.
BucketEventHistoryRepository: Stores the emitted reset event in the bucket event history.
WorkflowResult: Terminal workflow run record persisted separately by WorkflowRunRepository.

class WorkflowEnvelopeReasonClass(*values)[source]¶

Bases: StrEnum

Classification of the workflow-state envelope’s readability.

Carried in reason_class to distinguish a healthy envelope (READABLE), a row that cannot be decrypted (UNREADABLE), and an absent row (ABSENT).

READABLE¶

UNREADABLE¶

ABSENT¶

class WorkflowSecureObjectRecordPort(*args, **kwargs)[source]¶

Bases: Protocol

Decrypted secure-object fields required by workflow persistence.

property object_key: bytes¶: Opaque stored digest of the natural object key.

property payload: bytes¶: Decrypted inner-envelope bytes.

property revision_id: str¶: Current compare-and-swap revision token.

class WorkflowSecureObjectMetadataPort(*args, **kwargs)[source]¶

Bases: Protocol

Decryption-free state-row metadata used by reset fingerprints.

property schema_version: int¶: Persisted outer schema version.

property written_at: datetime¶: UTC instant the row was written.

property byte_length: int¶: Stored encrypted payload length.

class WorkflowSecureObjectStorePort(*args, **kwargs)[source]¶

Bases: Protocol

Encrypted-object operations consumed by the concrete workflow adapter.

load(namespace, object_key, *, expected_class, max_supported_version)[source]¶

Load one decrypted row under its registered contract.

Core types: SensitivityClass.

Return type:

WorkflowSecureObjectRecordPort | None

Parameters:

namespace (str)
object_key (str)
expected_class (SensitivityClass)
max_supported_version (int)

save_many(writes)[source]¶

Commit prepared writes in one storage transaction.

Return type:: None
Parameters:: writes (tuple[SecureObjectWrite, ...])

peek_metadata(namespace, object_key)[source]¶

Read row metadata without decrypting the payload.

Return type:

WorkflowSecureObjectMetadataPort | None

Parameters:

namespace (str)
object_key (str)

delete(namespace, object_key)[source]¶

Delete one addressed row when present.

Return type:

bool

Parameters:

namespace (str)
object_key (str)

save(*, namespace, object_key, classification, schema_version, written_at, payload)[source]¶

Persist one encrypted payload.

Core types: SensitivityClass.

Return type:

None

Parameters:

namespace (str)
object_key (str)
classification (SensitivityClass)
schema_version (int)
written_at (datetime)
payload (bytes)

list_records(namespace, *, expected_class, max_supported_version)[source]¶

Iterate every readable row under one namespace.

Core types: SensitivityClass.

Return type:

Iterator[WorkflowSecureObjectRecordPort]

Parameters:

namespace (str)
expected_class (SensitivityClass)
max_supported_version (int)

class WorkflowStateStorageProbe(metadata, state, envelope_readable)[source]¶

Bases: object

Storage-owned reset observation without adapter-specific DTOs or errors.

Parameters:

metadata (WorkflowSecureObjectMetadataPort | None)
state (WorkflowState | None)
envelope_readable (bool)

metadata: WorkflowSecureObjectMetadataPort | None¶

state: WorkflowState | None¶

envelope_readable: bool¶

class WorkflowPersistencePort(*args, **kwargs)[source]¶

Bases: Protocol

Concrete encrypted persistence capability composed by an executable host.

active_store()[source]¶

Return a secure-object store bound to the active bucket session.

Return type:: WorkflowSecureObjectStorePort

cold_bootstrap_store()[source]¶

Return the recovery-only store used when no bucket is active.

Return type:: WorkflowSecureObjectStorePort

load_state(store)[source]¶

Decode state and return the exact storage revision read.

Return type:: tuple[WorkflowState, str]
Parameters:: store (WorkflowSecureObjectStorePort)

prepare_state_write(state, *, expected_revision_id)[source]¶

Encode one state write without committing it.

Return type:

SecureObjectWrite

Parameters:

state (WorkflowState)
expected_revision_id (str | None)

save_writes(store, writes)[source]¶

Commit state and sibling writes atomically.

Return type:

None

Parameters:

store (WorkflowSecureObjectStorePort)
writes (tuple[SecureObjectWrite, ...])

probe_state(store)[source]¶

Return reset metadata plus any readable state payload.

Return type:: WorkflowStateStorageProbe
Parameters:: store (WorkflowSecureObjectStorePort)

delete_state(store)[source]¶

Delete only the workflow-state row.

Return type:: None
Parameters:: store (WorkflowSecureObjectStorePort)

is_revision_conflict(error)[source]¶

Recognise the concrete optimistic-concurrency refusal.

Return type:: bool
Parameters:: error (BaseException)

prepare_bucket_event_write(store, events)[source]¶

Prepare the history-catalogue write for events.

Return type:

SecureObjectWrite

Parameters:

store (WorkflowSecureObjectStorePort)
events (tuple[BucketEvent, ...])

save_run(store, result)[source]¶

Encode and persist one workflow run.

Return type:

None

Parameters:

store (WorkflowSecureObjectStorePort)
result (WorkflowResult)

load_run(store, run_id)[source]¶

Load one run and verify its stored identity.

Return type:

WorkflowResult

Parameters:

store (WorkflowSecureObjectStorePort)
run_id (str)

list_runs(store)[source]¶

Load every run after verifying stored identities.

Return type:: tuple[WorkflowResult, ...]
Parameters:: store (WorkflowSecureObjectStorePort)

bind_workflow_persistence_port(port)[source]¶

Bind one outward-composed workflow persistence adapter for this host lifetime.

Return type:: Generator[WorkflowPersistencePort]
Parameters:: port (WorkflowPersistencePort)

workflow_persistence_port()[source]¶

Resolve the explicitly composed workflow persistence adapter.

Return type:: WorkflowPersistencePort

class WorkflowStateRepository(*, objects=None, persistence=None, emit_reset=<function emit_workflow_state_reset>)[source]¶

Bases: object

Encrypted secure-object repository for WorkflowState.

Parameters:

objects (WorkflowSecureObjectStorePort | None)
persistence (WorkflowPersistencePort | None)
emit_reset (Callable[..., object])

load()[source]¶

Load state or return an empty payload when absent.

Returns the persisted WorkflowState, or an empty default when no state has been saved yet.

Return type:: WorkflowState

save(state)[source]¶

Persist state in the encrypted database object store.

Return type:: None
Parameters:: state (WorkflowState)

to_secure_object_write(state, *, expected_revision_id=None)[source]¶

Return the secure-object upsert for state without committing it.

Lets callers co-transactionally persist the workflow state and a sibling secure-object payload (typically an updated bucket-event-history catalogue) via a single save_many() call.

Return type:

SecureObjectWrite

Parameters:

state (WorkflowState)
expected_revision_id (str | None)

fingerprint_state(*, reason_class=None)[source]¶

Return a WorkflowStateResetFingerprint of the persisted state envelope.

Reads row-level metadata only; never decrypts the payload for the fingerprint fields. The state envelope is loaded once to derive recovered_bucket_id and to classify the envelope’s readability — a healthy, decryptable envelope is reported with reason_class="readable", an absent envelope with "absent", and an envelope row that cannot be decoded with "unreadable". A freshly-created storage root that has only just persisted a healthy state must therefore report readable, never unreadable.

reason_class may be supplied to override the derived classification when the caller already knows the trigger that forced the reset (e.g. a downstream handler that caught the concrete failure). When None the classification is derived from the envelope itself.

The repair reset-progress recovery verb is bootstrap-exempt and may run on a cold root where cadrumo_database_url does not resolve (no active profile). In that case there is no state envelope to reset; the fingerprint records empty metadata rather than crashing on the absent database.

Return type:: WorkflowStateResetFingerprint
Parameters:: reason_class (str | None)

reset_workflow_state(*, actor='cadrumo.application.workflow', source='aeat config repair reset-progress', reason_class=None)[source]¶

Delete the workflow-state envelope and emit a reset event.

The mutation is scoped to namespace cadrumo.workflow / key state; no other namespace or row is touched. The workflow_state.reset bucket event is appended BEFORE the secure-object row is deleted so the worst-case failure mode leaves an audit entry with the data still present (an idempotent recoverable state) rather than the data discarded without a trail. The fingerprint never carries plaintext envelope content.

Returns a WorkflowStateResetFingerprint with a hash of the deleted state for audit traceability.

Return type:

WorkflowStateResetFingerprint

Parameters:

actor (str)
source (str)
reason_class (str | None)

update(fn)[source]¶

Revision-safely load, transform, save, and return workflow state.

Return type:: WorkflowState
Parameters:: fn (Callable[[WorkflowState], WorkflowState])

update_with_writes(fn)[source]¶

CAS-update workflow state with optional sibling secure-object writes.

The callback is re-run from a fresh state after a revision conflict, so it must only derive values and prepare writes. All returned writes are committed with the workflow-state write in one SQL unit of work.

Return type:: WorkflowState
Parameters:: fn (Callable[[WorkflowState], tuple[WorkflowState, tuple[SecureObjectWrite, ...]]])

update_with_bucket_events(fn)[source]¶

CAS-update workflow state and append bucket events atomically.

Return type:: WorkflowState
Parameters:: fn (Callable[[WorkflowState], tuple[WorkflowState, tuple[BucketEvent, ...]]])

class WorkflowRunRepository(*, objects=None, persistence=None)[source]¶

Bases: object

Encrypted secure-object repository for WorkflowResult runs.

Parameters:

objects (WorkflowSecureObjectStorePort | None)
persistence (WorkflowPersistencePort | None)

save(result, *, runs_dir=None)[source]¶

Persist one workflow result in the secure object backend.

Return type:

Path

Parameters:

result (WorkflowResult)
runs_dir (Path | None)

load(run_id)[source]¶

Load one persisted WorkflowResult from the secure backend.

save files each run under its own run_id, so the secure-object key IS the run’s durable identity – nothing else in the row asserts it. The decrypted payload’s run_id is therefore compared with the requested key: a valid run B re-encrypted under A’s key would otherwise be returned here, and resume reads it to decide what to continue.

Returns the WorkflowResult for run_id.

Raises:: WorkflowError – When no run is stored under run_id, or when the stored payload names a different run than the key it sits under.
Return type:: WorkflowResult
Parameters:: run_id (str)

list(*, since=None)[source]¶

List persisted workflow runs newest-first, optionally filtered by date.

Each element is a WorkflowResult.

Return type:: tuple[WorkflowResult, ...]
Parameters:: since (date | None)

workflow_state_repository()[source]¶

Return the WorkflowStateRepository bound to the active-bucket database.

When an active profile bucket is present, the repository is backed by the bucket’s own encrypted database resolved through secure_object_repository_for_active_bucket() so the URL is derived from the live bucket path rather than the settings-override snapshot captured at test-fixture construction time. A cold root with no active bucket pointer is the bootstrap exception: it receives an explicit bare SecureObjectRepository so bootstrap-exempt recovery reads can still observe an absent state.

Return type:: WorkflowStateRepository

reset_workflow_state(*, actor='cadrumo.application.workflow', source='aeat config repair reset-progress', reason_class=None)[source]¶

Module-level helper around WorkflowStateRepository.reset_workflow_state().

Returns a WorkflowStateResetFingerprint with a hash of the deleted state for audit traceability.

Return type:

WorkflowStateResetFingerprint

Parameters:

actor (str)
source (str)
reason_class (str | None)

fingerprint_workflow_state(*, reason_class=None)[source]¶

Return a WorkflowStateResetFingerprint via WorkflowStateRepository.fingerprint_state().

Return type:: WorkflowStateResetFingerprint
Parameters:: reason_class (str | None)

validate_run_id(run_id)[source]¶

Return type:: str
Parameters:: run_id (str)

save_run(result, *, runs_dir=None)[source]¶

Persist one workflow result in the secure object backend.

runs_dir remains part of the API as a logical marker path for callers and tests, but no plaintext run file is written there.

Return type:

Path

Parameters:

result (WorkflowResult)
runs_dir (Path | None)

load_run(run_id)[source]¶

Load and return one WorkflowResult from the secure backend.

Return type:: WorkflowResult
Parameters:: run_id (str)

list_runs(*, since=None)[source]¶

List persisted WorkflowResult runs newest-first, optionally filtered by date.

Return type:: tuple[WorkflowResult, ...]
Parameters:: since (date | None)