Skip to content

Action definitions and ports

registry

Typed action definitions and heterogeneous registry boundary.

EffectKind module-attribute 

EffectKind = Literal['single', 'itemized']

SafeReference module-attribute 

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

CommandT module-attribute 

CommandT = TypeVar('CommandT', bound=BaseModel)

PrivateSnapshotT module-attribute 

PrivateSnapshotT = TypeVar('PrivateSnapshotT', bound=BaseModel)

PreviewT module-attribute 

PreviewT = TypeVar('PreviewT', bound=BaseModel)

ResultT module-attribute 

ResultT = TypeVar('ResultT', bound=BaseModel)

CommandContraT module-attribute 

CommandContraT = TypeVar('CommandContraT', bound=BaseModel, contravariant=True)

PrivateContraT module-attribute 

PrivateContraT = TypeVar('PrivateContraT', bound=BaseModel, contravariant=True)

JsonObject module-attribute 

JsonObject = dict[str, JsonValue]

_SAFE_REFERENCE_ADAPTER module-attribute 

_SAFE_REFERENCE_ADAPTER = TypeAdapter(SafeReference)

AuthorityBinding

Bases: ExperimentalModel

Security-relevant dimensions an authority record must match exactly.

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

proposal_commitment instance-attribute 

proposal_commitment: SafeReference

required_audience instance-attribute 

required_audience: SafeReference

required_channel_assurance instance-attribute 

required_channel_assurance: SafeReference

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

ActionType

Bases: ExperimentalModel

A host-defined action name with an explicit contract version.

namespace instance-attribute 

namespace: ActionNamespace

name instance-attribute 

name: ActionName

version instance-attribute 

version: Annotated[int, Field(ge=1)]

AuthoritativeTarget

Bases: ExperimentalModel

kind class-attribute instance-attribute 

kind: Literal['authoritative_target'] = 'authoritative_target'

reference instance-attribute 

reference: SafeReference

ConfirmingAuthority

Bases: ExperimentalModel

kind class-attribute instance-attribute 

kind: Literal['confirming_authority'] = 'confirming_authority'

reference instance-attribute 

reference: SafeReference

EvidenceConsumer

Bases: ExperimentalModel

kind class-attribute instance-attribute 

kind: Literal['evidence_consumer'] = 'evidence_consumer'

reference instance-attribute 

reference: SafeReference

ExperimentalModel

Bases: BaseModel

Strict, immutable base for the experimental public contract.

model_config class-attribute instance-attribute 

model_config = ConfigDict(extra='forbid', strict=True, frozen=True)

GovernedExecutor

Bases: ExperimentalModel

kind class-attribute instance-attribute 

kind: Literal['governed_executor'] = 'governed_executor'

reference instance-attribute 

reference: SafeReference

ProposingAgent

Bases: ExperimentalModel

kind class-attribute instance-attribute 

kind: Literal['proposing_agent'] = 'proposing_agent'

reference instance-attribute 

reference: SafeReference

RequestingPrincipal

Bases: ExperimentalModel

kind class-attribute instance-attribute 

kind: Literal['requesting_principal'] = 'requesting_principal'

reference instance-attribute 

reference: SafeReference

ExternalReference

Bases: ExperimentalModel

A minimized provider reference safe for an evidence projection.

system instance-attribute 

system: SafeReference

reference instance-attribute 

reference: SafeReference

ItemOutcome

Bases: ExperimentalModel

A minimized authoritative outcome for one declared itemized effect.

item_reference instance-attribute 

item_reference: SafeReference

status instance-attribute 

status: ItemOutcomeStatus

reason_code class-attribute instance-attribute 

reason_code: SafeReference | None = None

ItemOutcomeStatus

Bases: StrEnum

SUCCEEDED class-attribute instance-attribute 

SUCCEEDED = 'succeeded'

FAILED_KNOWN class-attribute instance-attribute 

FAILED_KNOWN = 'failed_known'

FAILED_UNKNOWN class-attribute instance-attribute 

FAILED_UNKNOWN = 'failed_unknown'

PreparationContext

Bases: ExperimentalModel

tenant_reference instance-attribute 

tenant_reference: SafeReference

requesting_principal instance-attribute 

requesting_principal: RequestingPrincipal

proposing_agent class-attribute instance-attribute 

proposing_agent: ProposingAgent | None = None

prepared_at instance-attribute 

prepared_at: AwareDatetime

DecisionContext

Bases: ExperimentalModel

tenant_reference instance-attribute 

tenant_reference: SafeReference

authority instance-attribute 

authority: ConfirmingAuthority

decided_at instance-attribute 

decided_at: AwareDatetime

ReadContext

Bases: ExperimentalModel

tenant_reference instance-attribute 

tenant_reference: SafeReference

consumer instance-attribute 

consumer: EvidenceConsumer

ExecutionContext

Bases: ExperimentalModel

tenant_reference instance-attribute 

tenant_reference: SafeReference

proposal_reference instance-attribute 

proposal_reference: SafeReference

semantic_effect_reference instance-attribute 

semantic_effect_reference: SafeReference

requesting_principal instance-attribute 

requesting_principal: RequestingPrincipal

authorities instance-attribute 

authorities: tuple[ConfirmingAuthority, ...]

observed_at instance-attribute 

observed_at: AwareDatetime

AuthorizationResult

Bases: ExperimentalModel

allowed instance-attribute 

allowed: bool

reason_code class-attribute instance-attribute 

reason_code: SafeReference | None = None

AuthorityEvaluation

Bases: ExperimentalModel

satisfied instance-attribute 

satisfied: bool

reason_code class-attribute instance-attribute 

reason_code: SafeReference | None = None

PreparedAction dataclass 

PreparedAction(private_snapshot: PrivateSnapshotT, display_preview: PreviewT, semantic_effect_reference: str)

Bases: Generic[PrivateSnapshotT, PreviewT]

private_snapshot instance-attribute 

private_snapshot: PrivateSnapshotT

display_preview instance-attribute 

display_preview: PreviewT

semantic_effect_reference instance-attribute 

semantic_effect_reference: str

__post_init__ 

__post_init__() -> None

ResolvedState dataclass 

ResolvedState(current_snapshot: PrivateSnapshotT, execution_precondition: str, materially_drifted: bool, replacement: PreparedAction[PrivateSnapshotT, PreviewT] | None = None)

Bases: Generic[PrivateSnapshotT, PreviewT]

current_snapshot instance-attribute 

current_snapshot: PrivateSnapshotT

execution_precondition instance-attribute 

execution_precondition: str

materially_drifted instance-attribute 

materially_drifted: bool

replacement class-attribute instance-attribute 

replacement: PreparedAction[PrivateSnapshotT, PreviewT] | None = None

__post_init__ 

__post_init__() -> None

ExecutionStatus

Bases: StrEnum

ACCEPTED class-attribute instance-attribute 

ACCEPTED = 'accepted'

STALE_NO_EFFECT class-attribute instance-attribute 

STALE_NO_EFFECT = 'stale_no_effect'

FAILED_KNOWN class-attribute instance-attribute 

FAILED_KNOWN = 'failed_known'

FAILED_UNKNOWN class-attribute instance-attribute 

FAILED_UNKNOWN = 'failed_unknown'

PARTIALLY_SUCCEEDED class-attribute instance-attribute 

PARTIALLY_SUCCEEDED = 'partially_succeeded'

ExecutionResult

Bases: ExperimentalModel, Generic[ResultT]

status instance-attribute 

status: ExecutionStatus

result class-attribute instance-attribute 

result: ResultT | None = None

item_outcomes class-attribute instance-attribute 

item_outcomes: tuple[ItemOutcome, ...] = ()

external_reference class-attribute instance-attribute 

external_reference: ExternalReference | None = None

reason_code class-attribute instance-attribute 

reason_code: SafeReference | None = None

partial_requires_items 

partial_requires_items() -> ExecutionResult[ResultT]

VerificationStatus

Bases: StrEnum

VERIFIED_COMPLETION class-attribute instance-attribute 

VERIFIED_COMPLETION = 'verified_completion'

VERIFIED_TERMINAL_FAILURE class-attribute instance-attribute 

VERIFIED_TERMINAL_FAILURE = 'verified_terminal_failure'

PROVISIONAL_ABSENCE class-attribute instance-attribute 

PROVISIONAL_ABSENCE = 'provisional_absence'

AUTHORITATIVE_FINAL_ABSENCE class-attribute instance-attribute 

AUTHORITATIVE_FINAL_ABSENCE = 'authoritative_final_absence'

TARGET_UNAVAILABLE class-attribute instance-attribute 

TARGET_UNAVAILABLE = 'target_unavailable'

VerificationResult

Bases: ExperimentalModel, Generic[ResultT]

status instance-attribute 

status: VerificationStatus

result class-attribute instance-attribute 

result: ResultT | None = None

item_outcomes class-attribute instance-attribute 

item_outcomes: tuple[ItemOutcome, ...] = ()

external_reference class-attribute instance-attribute 

external_reference: ExternalReference | None = None

reason_code class-attribute instance-attribute 

reason_code: SafeReference | None = None

settling_boundary_passed class-attribute instance-attribute 

settling_boundary_passed: bool = False

target_idempotency_guaranteed class-attribute instance-attribute 

target_idempotency_guaranteed: bool = False

absence_requires_consistent_evidence 

absence_requires_consistent_evidence() -> VerificationResult[ResultT]

PreparationPort

Bases: Protocol[CommandContraT, PrivateSnapshotT, PreviewT]

prepare async 

prepare(command: CommandContraT, *, context: PreparationContext) -> PreparedAction[PrivateSnapshotT, PreviewT]

AuthorizationPort

Bases: Protocol[CommandContraT, PrivateContraT]

can_prepare async 

can_prepare(command: CommandContraT, *, context: PreparationContext) -> AuthorizationResult

can_decide async 

can_decide(evidence: AuthorityEvidence, *, context: DecisionContext) -> AuthorizationResult

can_execute async 

can_execute(snapshot: PrivateContraT, *, context: ExecutionContext) -> AuthorizationResult

can_read async 

can_read(proposal_reference: str, *, context: ReadContext) -> bool

AuthorityEvaluatorPort

Bases: Protocol

evaluate async 

evaluate(*, binding: AuthorityBinding, evidence: tuple[AuthorityEvidence, ...]) -> AuthorityEvaluation

StateResolverPort

Bases: Protocol[PrivateSnapshotT, PreviewT]

resolve async 

resolve(snapshot: PrivateSnapshotT, *, context: ExecutionContext) -> ResolvedState[PrivateSnapshotT, PreviewT]

GovernedExecutorPort

Bases: Protocol[PrivateContraT, ResultT]

execute async 

execute(snapshot: PrivateContraT, *, context: ExecutionContext, execution_precondition: str) -> ExecutionResult[ResultT]

VerifierPort

Bases: Protocol[ResultT]

verify async 

verify(*, context: ExecutionContext) -> VerificationResult[ResultT]

RetentionPort

Bases: Protocol

authorize_erasure async 

authorize_erasure(proposal_reference: str, *, context: ReadContext) -> bool

ActionDefinition dataclass 

ActionDefinition(action_type: ActionType, command_model: type[CommandT], private_snapshot_model: type[PrivateSnapshotT], display_preview_model: type[PreviewT], result_model: type[ResultT], preparation: PreparationPort[CommandT, PrivateSnapshotT, PreviewT], authorization: AuthorizationPort[CommandT, PrivateSnapshotT], authority_evaluator: AuthorityEvaluatorPort, state_resolver: StateResolverPort[PrivateSnapshotT, PreviewT], executor: GovernedExecutorPort[PrivateSnapshotT, ResultT], verifier: VerifierPort[ResultT], commitment_provider: CommitmentProvider, protection_codec: ProtectionCodec, retention: RetentionPort, proposal_ttl: timedelta, executor_identity: GovernedExecutor, target_identity: AuthoritativeTarget, authority_audience: str, authority_channel_assurance: str, verification_delay: timedelta = timedelta(0), max_verification_attempts: int = 3, effect_kind: EffectKind = 'single', allow_resend_after_final_absence: bool = False, verification_lease_duration: timedelta = timedelta(minutes=1), semantic_idempotency_strategy: Literal['host_defined'] = 'host_defined')

Bases: Generic[CommandT, PrivateSnapshotT, PreviewT, ResultT]

action_type instance-attribute 

action_type: ActionType

command_model instance-attribute 

command_model: type[CommandT]

private_snapshot_model instance-attribute 

private_snapshot_model: type[PrivateSnapshotT]

display_preview_model instance-attribute 

display_preview_model: type[PreviewT]

result_model instance-attribute 

result_model: type[ResultT]

preparation instance-attribute 

preparation: PreparationPort[CommandT, PrivateSnapshotT, PreviewT]

authorization instance-attribute 

authorization: AuthorizationPort[CommandT, PrivateSnapshotT]

authority_evaluator instance-attribute 

authority_evaluator: AuthorityEvaluatorPort

state_resolver instance-attribute 

state_resolver: StateResolverPort[PrivateSnapshotT, PreviewT]

executor instance-attribute 

executor: GovernedExecutorPort[PrivateSnapshotT, ResultT]

verifier instance-attribute 

verifier: VerifierPort[ResultT]

commitment_provider instance-attribute 

commitment_provider: CommitmentProvider

protection_codec instance-attribute 

protection_codec: ProtectionCodec

retention instance-attribute 

retention: RetentionPort

proposal_ttl instance-attribute 

proposal_ttl: timedelta

executor_identity instance-attribute 

executor_identity: GovernedExecutor

target_identity instance-attribute 

target_identity: AuthoritativeTarget

authority_audience instance-attribute 

authority_audience: str

authority_channel_assurance instance-attribute 

authority_channel_assurance: str

verification_delay class-attribute instance-attribute 

verification_delay: timedelta = timedelta(0)

max_verification_attempts class-attribute instance-attribute 

max_verification_attempts: int = 3

effect_kind class-attribute instance-attribute 

effect_kind: EffectKind = 'single'

allow_resend_after_final_absence class-attribute instance-attribute 

allow_resend_after_final_absence: bool = False

verification_lease_duration class-attribute instance-attribute 

verification_lease_duration: timedelta = timedelta(minutes=1)

semantic_idempotency_strategy class-attribute instance-attribute 

semantic_idempotency_strategy: Literal['host_defined'] = 'host_defined'

__post_init__ 

__post_init__() -> None

DefinitionConformanceError

Bases: ValueError

Raised when declared boundary models cannot satisfy the runtime contract.

DuplicateActionError

Bases: RuntimeError

ActionNotRegisteredError

Bases: LookupError

DefinitionTypeMismatchError

Bases: TypeError

ActionRegistry 

ActionRegistry()

A heterogeneous registry with checked type recovery at its boundary.

register 

register(definition: ActionDefinition[CommandT, PrivateSnapshotT, PreviewT, ResultT]) -> None

get_typed 

get_typed(action_type: ActionType, *, command_model: type[CommandT], private_snapshot_model: type[PrivateSnapshotT], display_preview_model: type[PreviewT], result_model: type[ResultT]) -> ActionDefinition[CommandT, PrivateSnapshotT, PreviewT, ResultT]

assert_definition_conforms 

assert_definition_conforms(definition: ActionDefinition[CommandT, PrivateSnapshotT, PreviewT, ResultT]) -> None

Reject declared model shapes that cannot satisfy the runtime boundary.

_assert_boundary_model_config 

_assert_boundary_model_config(*, role: str, model: type[BaseModel]) -> None

_floating_point_field_paths 

_floating_point_field_paths(model: type[BaseModel]) -> tuple[str, ...]

_action_key 

_action_key(action_type: ActionType) -> tuple[str, str, int]