Conformance helpers

conformance

Reusable conformance helpers for action stores, hosts, and safe projections.

SafeReference module-attribute

SafeReference = Annotated[str, StringConstraints(min_length=1, max_length=255), AfterValidator(_validate_safe_reference)]

ALLOWED_LIFECYCLE_TRANSITIONS module-attribute

ALLOWED_LIFECYCLE_TRANSITIONS: dict[LifecycleStatus, frozenset[LifecycleStatus]] = {LifecycleStatus.AWAITING_AUTHORITY: frozenset({LifecycleStatus.AUTHORIZED, LifecycleStatus.DENIED, LifecycleStatus.EXPIRED}), LifecycleStatus.AUTHORIZED: frozenset({LifecycleStatus.BLOCKED, LifecycleStatus.EXPIRED, LifecycleStatus.EXECUTING, LifecycleStatus.STALE}), LifecycleStatus.EXECUTING: frozenset({LifecycleStatus.STALE, LifecycleStatus.FAILED_KNOWN, LifecycleStatus.FAILED_UNKNOWN, LifecycleStatus.VERIFICATION_PENDING}), LifecycleStatus.FAILED_UNKNOWN: frozenset({LifecycleStatus.VERIFICATION_PENDING}), LifecycleStatus.VERIFICATION_PENDING: frozenset({LifecycleStatus.AUTHORIZED, LifecycleStatus.EXECUTING, LifecycleStatus.FAILED_KNOWN, LifecycleStatus.FAILED_UNKNOWN, LifecycleStatus.PARTIALLY_SUCCEEDED, LifecycleStatus.VERIFICATION_UNRESOLVED, LifecycleStatus.VERIFIED}), LifecycleStatus.STALE: frozenset({LifecycleStatus.SUPERSEDED}), LifecycleStatus.VERIFIED: frozenset(), LifecycleStatus.BLOCKED: frozenset(), LifecycleStatus.DENIED: frozenset(), LifecycleStatus.EXPIRED: frozenset(), LifecycleStatus.FAILED_KNOWN: frozenset(), LifecycleStatus.PARTIALLY_SUCCEEDED: frozenset(), LifecycleStatus.SUPERSEDED: frozenset(), LifecycleStatus.VERIFICATION_UNRESOLVED: frozenset()}

T module-attribute

T = TypeVar('T')

_SAFE_REFERENCE_ADAPTER module-attribute

_SAFE_REFERENCE_ADAPTER = TypeAdapter(SafeReference)

__all__ module-attribute

__all__ = ['BenchmarkResult', 'ConformanceError', 'LeakageFinding', 'PerformanceProfile', 'RuntimeConformanceDriver', 'StoreConformanceCase', 'assert_action_store_conforms', 'assert_no_sensitive_data', 'assert_performance_profile', 'assert_providers_conform', 'assert_runtime_conforms', 'benchmark_async', 'benchmark_async_concurrent', 'benchmark_sync', 'find_sensitive_data']

LifecycleStatus

Bases: StrEnum

Closed lifecycle vocabulary for proposal execution and verification.

AWAITING_AUTHORITY class-attribute instance-attribute

AWAITING_AUTHORITY = 'awaiting_authority'

DENIED class-attribute instance-attribute

DENIED = 'denied'

EXPIRED class-attribute instance-attribute

EXPIRED = 'expired'

AUTHORIZED class-attribute instance-attribute

AUTHORIZED = 'authorized'

BLOCKED class-attribute instance-attribute

BLOCKED = 'blocked'

STALE class-attribute instance-attribute

STALE = 'stale'

SUPERSEDED class-attribute instance-attribute

SUPERSEDED = 'superseded'

EXECUTING class-attribute instance-attribute

EXECUTING = 'executing'

FAILED_KNOWN class-attribute instance-attribute

FAILED_KNOWN = 'failed_known'

FAILED_UNKNOWN class-attribute instance-attribute

FAILED_UNKNOWN = 'failed_unknown'

VERIFICATION_PENDING class-attribute instance-attribute

VERIFICATION_PENDING = 'verification_pending'

VERIFICATION_UNRESOLVED class-attribute instance-attribute

VERIFICATION_UNRESOLVED = 'verification_unresolved'

PARTIALLY_SUCCEEDED class-attribute instance-attribute

PARTIALLY_SUCCEEDED = 'partially_succeeded'

VERIFIED class-attribute instance-attribute

VERIFIED = 'verified'

AuthorityReceipt

Bases: _ReceiptBase

receipt_type class-attribute instance-attribute

receipt_type: Literal['authority'] = 'authority'

status instance-attribute

status: AuthorityReceiptStatus

participant instance-attribute

participant: ConfirmingAuthority

AuthorityReceiptStatus

Bases: StrEnum

RECORDED class-attribute instance-attribute

RECORDED = 'recorded'

REJECTED class-attribute instance-attribute

REJECTED = 'rejected'

FAILED class-attribute instance-attribute

FAILED = 'failed'

MISSING class-attribute instance-attribute

MISSING = 'missing'

OperationOutcome

Bases: StrEnum

PREPARED class-attribute instance-attribute

PREPARED = 'prepared'

AUTHORITY_PENDING class-attribute instance-attribute

AUTHORITY_PENDING = 'authority_pending'

AUTHORIZED class-attribute instance-attribute

AUTHORIZED = 'authorized'

DENIED class-attribute instance-attribute

DENIED = 'denied'

BLOCKED class-attribute instance-attribute

BLOCKED = 'blocked'

STALE class-attribute instance-attribute

STALE = 'stale'

IN_PROGRESS class-attribute instance-attribute

IN_PROGRESS = 'in_progress'

EXPIRED class-attribute instance-attribute

EXPIRED = 'expired'

REPLAYED class-attribute instance-attribute

REPLAYED = 'replayed'

CONFLICT class-attribute instance-attribute

CONFLICT = 'conflict'

VERIFICATION_PENDING class-attribute instance-attribute

VERIFICATION_PENDING = 'verification_pending'

VERIFIED class-attribute instance-attribute

VERIFIED = 'verified'

PARTIALLY_SUCCEEDED class-attribute instance-attribute

PARTIALLY_SUCCEEDED = 'partially_succeeded'

FAILED_KNOWN class-attribute instance-attribute

FAILED_KNOWN = 'failed_known'

FAILED_UNKNOWN class-attribute instance-attribute

FAILED_UNKNOWN = 'failed_unknown'

VERIFICATION_UNRESOLVED class-attribute instance-attribute

VERIFICATION_UNRESOLVED = 'verification_unresolved'

RESEND_ALLOWED class-attribute instance-attribute

RESEND_ALLOWED = 'resend_allowed'

ERASED class-attribute instance-attribute

ERASED = 'erased'

EffectClaimResult

Bases: StrEnum

ACQUIRED class-attribute instance-attribute

ACQUIRED = 'acquired'

OWNED_BY_PROPOSAL class-attribute instance-attribute

OWNED_BY_PROPOSAL = 'owned_by_proposal'

CONFLICT class-attribute instance-attribute

CONFLICT = 'conflict'

PROPOSAL_NOT_FOUND class-attribute instance-attribute

PROPOSAL_NOT_FOUND = 'proposal_not_found'

PROPOSAL_NOT_AUTHORIZED class-attribute instance-attribute

PROPOSAL_NOT_AUTHORIZED = 'proposal_not_authorized'

StoreInvariantError

Bases: RuntimeError

AuthorityEvidence

Bases: ExperimentalModel

A bound authority decision; it is not authorization by itself.

kind class-attribute instance-attribute

kind: Literal['bound_decision'] = 'bound_decision'

domain class-attribute instance-attribute

domain: Literal['threvo.actions.authority-evidence'] = 'threvo.actions.authority-evidence'

schema_version class-attribute instance-attribute

schema_version: Literal['internal/v0'] = 'internal/v0'

tenant_reference instance-attribute

tenant_reference: SafeReference

action_type instance-attribute

action_type: ActionType

proposal_instance_reference instance-attribute

proposal_instance_reference: SafeReference

semantic_effect_reference instance-attribute

semantic_effect_reference: SafeReference

authority instance-attribute

authority: ConfirmingAuthority

audience class-attribute instance-attribute

audience: tuple[SafeReference, ...] = Field(min_length=1)

decision instance-attribute

decision: AuthorityDecision

proposal_commitment instance-attribute

proposal_commitment: SafeReference

channel_assurance instance-attribute

channel_assurance: SafeReference

issued_at instance-attribute

issued_at: AwareDatetime

expires_at instance-attribute

expires_at: AwareDatetime

expiry_follows_issue_time

expiry_follows_issue_time() -> AuthorityEvidence

CommitmentProvider

Bases: Protocol

Host-owned proposal-scoped keyed commitment boundary.

Destruction must be idempotent so an interrupted erasure can resume safely.

create async

create(*, proposal_reference: str, canonical_payload: bytes) -> KeyedCommitment

verify async

verify(*, proposal_reference: str, canonical_payload: bytes, commitment: KeyedCommitment) -> bool

destroy_commitment async

destroy_commitment(*, commitment: KeyedCommitment) -> None

ProtectionCodec

Bases: Protocol

Host-owned protection boundary for canonical private snapshots.

Destruction must be idempotent so an interrupted erasure can resume safely.

protect async

protect(*, proposal_reference: str, canonical_payload: bytes) -> ProtectedPayload

unprotect async

unprotect(*, payload: ProtectedPayload) -> bytes

destroy_payload async

destroy_payload(*, payload: ProtectedPayload) -> None

ActionOperationResult

Bases: ExperimentalModel

proposal_reference instance-attribute

proposal_reference: SafeReference

lifecycle_status instance-attribute

lifecycle_status: LifecycleStatus

outcome instance-attribute

outcome: OperationOutcome

revision instance-attribute

revision: int

display_preview class-attribute instance-attribute

display_preview: JsonObject = Field(default_factory=dict)

safe_result class-attribute instance-attribute

safe_result: JsonObject | None = None

fresh_proposal_reference class-attribute instance-attribute

fresh_proposal_reference: SafeReference | None = None

reason_code class-attribute instance-attribute

reason_code: SafeReference | None = None

is_terminal property

is_terminal: bool

Whether the proposal lifecycle has no valid transition left.

needs_reconciliation property

needs_reconciliation: bool

Whether authoritative reconciliation may advance this proposal.

ActionStore

Bases: Protocol

create async

create(proposal: StoredProposal) -> None

get async

get(tenant_reference: str, proposal_reference: str) -> StoredProposal | None

compare_and_set async

compare_and_set(*, tenant_reference: str, proposal_reference: str, expected_revision: int, expected_statuses: tuple[LifecycleStatus, ...], updated: StoredProposal) -> bool

admit_execution async

admit_execution(*, tenant_reference: str, proposal_reference: str, expected_revision: int, admitted_at: datetime, updated: StoredProposal) -> EffectClaimResult

get_effect_claim_owner async

get_effect_claim_owner(*, tenant_reference: str, action_type: ActionType, semantic_effect_reference: str) -> str | None

RetentionStore

Bases: Protocol

Privileged persistence operations kept outside the runtime DB role.

mark_erasure_pending async

mark_erasure_pending(*, tenant_reference: str, proposal_reference: str, expected_revision: int, pending_at: datetime) -> bool

complete_erasure async

complete_erasure(*, tenant_reference: str, proposal_reference: str, expected_revision: int, erased_at: datetime) -> bool

StoredProposal

Bases: ExperimentalModel

Persistence-neutral lifecycle record; private state is always protected.

tenant_reference instance-attribute

tenant_reference: SafeReference

proposal_reference instance-attribute

proposal_reference: SafeReference

action_type instance-attribute

action_type: ActionType

semantic_effect_reference instance-attribute

semantic_effect_reference: SafeReference

effect_kind instance-attribute

effect_kind: EffectKind

lifecycle_status instance-attribute

lifecycle_status: LifecycleStatus

revision class-attribute instance-attribute

revision: int = Field(ge=0)

protected_private_snapshot instance-attribute

protected_private_snapshot: ProtectedPayload | None

commitment instance-attribute

commitment: KeyedCommitment | None

display_preview class-attribute instance-attribute

display_preview: JsonObject = Field(default_factory=dict)

requesting_principal class-attribute instance-attribute

requesting_principal: RequestingPrincipal | None = None

proposing_agent class-attribute instance-attribute

proposing_agent: ProposingAgent | None = None

created_at instance-attribute

created_at: AwareDatetime

expires_at instance-attribute

expires_at: AwareDatetime

authority_evidence class-attribute instance-attribute

authority_evidence: tuple[AuthorityEvidence, ...] = ()

receipts class-attribute instance-attribute

receipts: tuple[Receipt, ...] = ()

verification_attempts class-attribute instance-attribute

verification_attempts: int = Field(default=0, ge=0)

max_verification_attempts class-attribute instance-attribute

max_verification_attempts: int = Field(ge=1)

next_verification_at class-attribute instance-attribute

next_verification_at: AwareDatetime | None = None

safe_result class-attribute instance-attribute

safe_result: JsonObject | None = None

execution_precondition class-attribute instance-attribute

execution_precondition: SafeReference | None = None

superseded_by class-attribute instance-attribute

superseded_by: SafeReference | None = None

erasure_pending_at class-attribute instance-attribute

erasure_pending_at: AwareDatetime | None = None

erased_at class-attribute instance-attribute

erased_at: AwareDatetime | None = None

ConformanceError

ConformanceError(code: str)

Bases: AssertionError

A stable, secret-free conformance failure.

code instance-attribute

code = code

LeakageFinding dataclass

LeakageFinding(path: str, label: str, kind: str)

Location and caller-supplied label for sensitive data, never its value.

path instance-attribute

path: str

label instance-attribute

label: str

kind instance-attribute

kind: str

StoreConformanceCase dataclass

StoreConformanceCase(store: ActionStore, retention_store: RetentionStore, original: StoredProposal, evidence: AuthorityEvidence, observed_at: datetime)

store instance-attribute

store: ActionStore

retention_store instance-attribute

retention_store: RetentionStore

original instance-attribute

original: StoredProposal

evidence instance-attribute

evidence: AuthorityEvidence

observed_at instance-attribute

observed_at: datetime

IndependentStoreConformanceCase dataclass

IndependentStoreConformanceCase(first_store: ActionStore, second_store: ActionStore, original: StoredProposal, evidence: AuthorityEvidence, observed_at: datetime, security_profile_identifier: str)

Two adapters backed by independently created connections to one store.

first_store instance-attribute

first_store: ActionStore

second_store instance-attribute

second_store: ActionStore

original instance-attribute

original: StoredProposal

evidence instance-attribute

evidence: AuthorityEvidence

observed_at instance-attribute

observed_at: datetime

security_profile_identifier instance-attribute

security_profile_identifier: str

__post_init__

__post_init__() -> None

IndependentStoreConformanceReport dataclass

IndependentStoreConformanceReport(security_profile_identifier: str, checks: tuple[str, ...])

Deterministic evidence emitted after all independent-store checks pass.

security_profile_identifier instance-attribute

security_profile_identifier: str

checks instance-attribute

checks: tuple[str, ...]

BenchmarkResult dataclass

BenchmarkResult(profile: str, iterations: int, p50_ms: float, p95_ms: float, p99_ms: float)

profile instance-attribute

profile: str

iterations instance-attribute

iterations: int

p50_ms instance-attribute

p50_ms: float

p95_ms instance-attribute

p95_ms: float

p99_ms instance-attribute

p99_ms: float

PerformanceProfile dataclass

PerformanceProfile(name: str, max_p99_ms: float, max_p95_ms: float | None = None, min_iterations: int = 100)

name instance-attribute

name: str

max_p99_ms instance-attribute

max_p99_ms: float

max_p95_ms class-attribute instance-attribute

max_p95_ms: float | None = None

min_iterations class-attribute instance-attribute

min_iterations: int = 100

__post_init__

__post_init__() -> None

RuntimeConformanceDriver

Bases: Protocol

Small host-owned driver used by the reusable runtime scenarios.

executor_calls property

executor_calls: int

prepare async

prepare() -> ActionOperationResult

record_approval async

record_approval(proposal_reference: str) -> ActionOperationResult

execute async

execute(proposal_reference: str) -> ActionOperationResult

reconcile async

reconcile(proposal_reference: str) -> ActionOperationResult

make_verification_due async

make_verification_due() -> None

revoke_execution_authorization async

revoke_execution_authorization() -> None

introduce_material_drift async

introduce_material_drift() -> None

_derived_safe_reference

_derived_safe_reference(source: str, *, purpose: str, discriminator: str) -> str

find_sensitive_data

find_sensitive_data(value: object, *, forbidden_literals: Mapping[str, str], forbidden_key_fragments: Collection[str] = ()) -> tuple[LeakageFinding, ...]

Recursively locate seeded secrets without echoing them in the report.

assert_no_sensitive_data

assert_no_sensitive_data(value: object, *, forbidden_literals: Mapping[str, str], forbidden_key_fragments: Collection[str] = ()) -> None

Fail with labels and structural paths while keeping seeded values secret.

assert_action_store_conforms async

assert_action_store_conforms(case: StoreConformanceCase) -> None

Exercise tenant isolation, guarded updates, and atomic effect admission.

assert_independent_store_connections_conform async

assert_independent_store_connections_conform(case: IndependentStoreConformanceCase) -> IndependentStoreConformanceReport

Prove shared visibility, guarded revisions, and effect admission across stores.

_assert_store_update_invariants async

_assert_store_update_invariants(case: StoreConformanceCase) -> None

_assert_lifecycle_transition_guards async

_assert_lifecycle_transition_guards(case: StoreConformanceCase) -> None

_proposal_at_status async

_proposal_at_status(case: StoreConformanceCase, status: LifecycleStatus) -> StoredProposal

_lifecycle_seed_paths

_lifecycle_seed_paths() -> dict[LifecycleStatus, tuple[LifecycleStatus, ...]]

_conformance_proposal

_conformance_proposal(case: StoreConformanceCase, suffix: str) -> StoredProposal

_conformance_evidence

_conformance_evidence(case: StoreConformanceCase, proposal: StoredProposal) -> AuthorityEvidence

_independent_proposal

_independent_proposal(original: StoredProposal, suffix: str) -> StoredProposal

_independent_evidence

_independent_evidence(original: AuthorityEvidence, proposal: StoredProposal) -> AuthorityEvidence

_require_update_rejected async

_require_update_rejected(case: StoreConformanceCase, *, current: StoredProposal, updated: StoredProposal, code: str) -> None

assert_providers_conform async

assert_providers_conform(*, commitment_provider: CommitmentProvider, protection_codec: ProtectionCodec, proposal_reference: str, canonical_payload: bytes, mutated_payload: bytes) -> None

Check commitment binding and protected-snapshot round trips.

assert_runtime_conforms async

assert_runtime_conforms(factory: Callable[[], RuntimeConformanceDriver]) -> None

Run framework-neutral happy, forged-resume, revocation, and drift scenarios.

benchmark_sync

benchmark_sync(operation: Callable[[], T], *, profile: PerformanceProfile, iterations: int | None = None, warmup: int = 10) -> BenchmarkResult

Measure framework-owned synchronous work independently from external I/O.

benchmark_async async

benchmark_async(operation: Callable[[], Awaitable[T]], *, profile: PerformanceProfile, iterations: int | None = None, warmup: int = 10) -> BenchmarkResult

Measure asynchronous orchestration supplied by an in-process test host.

benchmark_async_concurrent async

benchmark_async_concurrent(operation: Callable[[int], Awaitable[T]], *, profile: PerformanceProfile, concurrency: int = 100, batches: int = 10, warmup_batches: int = 1) -> BenchmarkResult

Measure per-operation latency while running fixed concurrent batches.

assert_performance_profile

assert_performance_profile(result: BenchmarkResult, profile: PerformanceProfile) -> None

_benchmark_result

_benchmark_result(samples: Sequence[float], *, profile: PerformanceProfile) -> BenchmarkResult

_gather_timings async

_gather_timings(operation: Callable[[int], Awaitable[float]], *, offset: int, concurrency: int) -> list[float]

_percentile

_percentile(ordered: Sequence[float], quantile: float) -> float

_validate_benchmark_counts

_validate_benchmark_counts(*, count: int, warmup: int, profile: PerformanceProfile) -> None

_require

_require(condition: object, code: str) -> None