#protocols.src.orxtra.protocols._contracts

#protocols.src.orxtra.protocols._contracts

#EventSink

Receives typed events. Async because some sinks need I/O (PG writes).

#on_event

async def on_event(self, event: T_event_contra) -> None

#ActionExecutor

Injected executor for WorkflowAction dispatch.

The dispatch module cannot start workflows directly (that would create a downward dependency to the scheduler). Callers inject an executor that bridges the gap.

#execute_workflow

async def execute_workflow(self, workflow_path: str, config: dict[str, object], events: list[dict[str, object]]) -> None

#Renderer

Converts a typed result into a text string for the LLM.

#render

def render(self, data: T_contra) -> str

#SessionProtocol

Structural protocol for the Session, used to break the dependency from protocols to session.

#tools

def tools(self) -> list[Tool]

#update_tools

def update_tools(self, tools: list[Tool]) -> None

#send

def send(self, message: str) -> AsyncIterator[Any]

#CheckExecutor

#run_consult

async def run_consult(self, agent: str, question: str, variable_values: dict[str, str] | None=None) -> str

#run_workflow_check

async def run_workflow_check(self, execution: Execution) -> CheckResult

#OverseerProtocol

Protocol for the Overseer, used by the scheduler to avoid a direct dependency on the intelligence layer.

#prepare_event

def prepare_event(self, event: OverseerEvent) -> str

#HealthMonitorProtocol

Protocol for health monitoring, used by the scheduler to avoid a direct dependency on the intelligence layer.

#is_degraded

def is_degraded(self, event_type: str) -> bool

#record_event

def record_event(self, event_type: str, *, success: bool, is_repetition: bool=False) -> None

#EventDelivery

Protocol for event delivery (fire-and-wait).

Used by the scheduler so callers can inject alternative implementations (e.g. PG-backed LISTEN/NOTIFY).

#fire

async def fire(self, event_name: str, payload: dict[str, object] | None=None, *, source: str | None=None) -> None

#wait_for

async def wait_for(self, event_name: str, *, deadline_seconds: float) -> dict[str, object] | None

#FlushScheduler

Schedules deferred flush callbacks with a deadline.

Used by the write-safety module to schedule and cancel flush operations without depending on a concrete scheduler.

#schedule_flush

def schedule_flush(self, deadline: float, callback: Callable[[], Awaitable[None]]) -> object

#cancel_flush

def cancel_flush(self, handle: object) -> None

#SourceStorage

#create_source

async def create_source(self, source: Source) -> UUID

#get_source

async def get_source(self, source_id: UUID) -> Source | None

#get_source_by_slug

async def get_source_by_slug(self, slug: str) -> Source | None

#list_sources

async def list_sources(self) -> list[Source]

#delete_source

async def delete_source(self, source_id: UUID) -> None

#SubscriptionStorage

#create_subscription

async def create_subscription(self, subscription: Subscription) -> UUID

#get_subscription

async def get_subscription(self, sub_id: UUID) -> Subscription | None

#list_subscriptions

async def list_subscriptions(self, *, enabled_only: bool=True, principal_id: UUID | None=None) -> list[Subscription]

#update_subscription

async def update_subscription(self, sub_id: UUID, *, enabled: bool) -> None

#delete_subscription

async def delete_subscription(self, sub_id: UUID) -> None

#ActionStorage

#create_action

async def create_action(self, action: SubscriptionAction) -> UUID

#list_actions

async def list_actions(self, sub_id: UUID) -> list[SubscriptionAction]

#delete_actions

async def delete_actions(self, sub_id: UUID) -> None

#AccumulatorStorage

#buffer_event

async def buffer_event(self, entry: AccumulatorEntry) -> UUID

#claim_batch

async def claim_batch(self, action_id: UUID, limit: int=100) -> list[AccumulatorEntry]

#confirm_batch

async def confirm_batch(self, entry_ids: list[UUID]) -> None

#pending_count

async def pending_count(self, action_id: UUID) -> int

#DispatchBackend

class DispatchBackend(SourceStorage, SubscriptionStorage, ActionStorage, AccumulatorStorage, Protocol):

#EventBus

Event notification (replaces LISTEN/NOTIFY).

#subscribe

async def subscribe(self, channel: str, callback: Callable[[str], Awaitable[None]]) -> None

#unsubscribe

async def unsubscribe(self, channel: str, callback: Callable[[str], Awaitable[None]]) -> None

#publish

async def publish(self, channel: str, payload: str) -> None

#SurfaceGenerator

Generates A2UI surface specs from model types.

#generate

def generate(self, model_type: type) -> SurfaceSpec

#CardContributor

Contributes a fragment to an A2A Agent Card.

#card_fragment

def card_fragment(self) -> dict[str, Any]

#KeyedMacProvider

Non-exportable keyed MAC verification.

Modeled on KMS: the only operation is verify(). There is no get-value or resolve method -- key export is impossible by construction. Multiple concurrently-valid key versions enable rotation; verdicts report the matched version.

#verify

async def verify(self, key_ref: str, message: bytes, signature: str, algorithm: str) -> MacVerdict

#CredentialVerifier

Per-credential-type verification strategy.

Hash verifiers (bearer/api_key) need zero secret capability. HMAC verifiers are constructed with a KeyedMacProvider.

#credential_type

def credential_type(self) -> str

#verify

async def verify(self, credential_record: object, presented_credential: str) -> AuthContext

#AuthStorage

Storage protocol for auth data, replacing the concrete backend union.

#create_consumer

async def create_consumer(self, name: str, trust_tier: TrustTier, scope_grants: list[str], *, consumer_id: UUID, principal_id: UUID) -> UUID

#get_consumer

async def get_consumer(self, consumer_id: UUID) -> ConsumerRecord | None

#disable_consumer

async def disable_consumer(self, consumer_id: UUID) -> None

#create_credential

async def create_credential(self, consumer_id: UUID, credential_type: str, raw_value: str, *, secret_ref: str | None=None) -> UUID

#get_credential_by_id

async def get_credential_by_id(self, credential_id: UUID) -> CredentialRecord | None

#get_credential_by_hash

async def get_credential_by_hash(self, credential_hash: str) -> CredentialRecord | None

#get_credentials_by_consumer

async def get_credentials_by_consumer(self, consumer_id: UUID, *, credential_type: str | None=None) -> list[CredentialRecord]

#PrincipalStorage

Storage protocol for persisted identity (principals).

A Principal is the durable identity row for an actor; other tables FK to it for attribution and ownership. The identity module owns the concrete backend and the domain exceptions referenced below.

#mint_principal

async def mint_principal(self, kind: str, external_ref: UUID, display_name: str | None) -> Principal

Idempotent upsert on (kind, external_ref).

Creates the principal if absent; returns the existing row otherwise. Never errors on duplicates. This is the crash-safe eager-minting primitive: callers can mint unconditionally on every actor appearance, and a retry after a partial failure converges to the same single row.

#get_principal

async def get_principal(self, principal_id: UUID) -> Principal | None

#get_principal_by_ref

async def get_principal_by_ref(self, kind: str, external_ref: UUID) -> Principal | None

#list_principals

async def list_principals(self, kind: str | None=None) -> list[Principal]

#update_display_name

async def update_display_name(self, principal_id: UUID, display_name: str) -> None

Set the display name of an existing principal.

Hard error if the principal is absent -- this is not an upsert.

#delete_principal

async def delete_principal(self, principal_id: UUID) -> None

Delete a principal.

Cascades owned subscriptions. Hard domain error if the principal has history: events/runs/inbox/sources/consumers FKs are RESTRICT, so a principal with any such references cannot be deleted. The domain exception is defined by the identity module.

#sweep_orphaned_run_principals

async def sweep_orphaned_run_principals(self, older_than: timedelta) -> int

Delete kind=run principals with no matching runs row.

Orphaned run principals are created by a mint-first that crashed before create_run inserted the matching runs row. They accumulate silently; the RESTRICT FKs independently guarantee only zero-history rows can be deleted.

The older_than guard closes a race with concurrent start_run: a principal minted moments ago may not yet have its runs row (the INSERT is the next statement). Principals younger than now - older_than are skipped, making the race structurally impossible with a conservative window.

Returns the count of deleted principals.

#NotificationPort

Delivers notifications to principals.

A notification is a one-way message from any source to a target principal, carrying a source_ref (opaque origin label) and a free-form JSON payload. Notifications are created, listed, and acknowledged; the port knows nothing about routing policy -- that lives in the notification module itself.

#create_delivery

async def create_delivery(self, target_principal_id: UUID, source_ref: str, payload: dict[str, Any]) -> UUID

Create a notification delivery for the given principal.

Returns the delivery ID.

#list_for_principal

async def list_for_principal(self, principal_id: UUID, *, unacknowledged_only: bool=True, cursor: UUID | None=None, limit: int=50) -> list[NotificationDelivery]

List notifications for a principal.

#acknowledge

async def acknowledge(self, delivery_id: UUID) -> None

Mark a notification delivery as acknowledged.