cayu.sessions.SessionStore

Part of cayu.sessions.base. Read the Python API overview and install the required extra.

from cayu.sessions import SessionStore

Persistent store for sessions and append-only events.

Implementations own the run-epoch contract. A successful transition to RUNNING claims a new epoch for the current execution context. Runtime progress writes made by that context must reject a stale epoch with SessionRunFenced. Releasing the claim revokes the durable epoch before clearing task-local ownership so inherited child contexts cannot write late.

Implementations also own the clock used for inactivity, recovery-claim, epoch-takeover, and activity-stamp decisions. Callers pass durations. The store samples, compares, and stamps time inside one atomic boundary; a caller or worker clock is never durable ownership authority.

Bases: ABC.

Fields and attributes

Defaults below are source expressions; Field(...) includes schema/validation settings and default_factory creates a fresh value. Inherited declarations are included. Constructor behavior follows the model or dataclass shown in the bases/decorators.

Name	Type	Default
service_durability	RuntimeStoreDurability	RuntimeStoreDurability.UNVERIFIED
supports_session_execution	class attribute	False

read_session_creation_decision (method)

async SessionStore.read_session_creation_decision(target)

list_pending_session_creations (method)

async SessionStore.list_pending_session_creations(owner, *, cursor=None, limit=32)

create_participant_owned_session (method)

async SessionStore.create_participant_owned_session(creation_request: ParticipantSessionCreationRequest, *, resolved_request: RunRequest, identity: SessionIdentity, binding_factory: Callable[[Session], tuple[ParticipantSessionBinding, ParticipantSessionCreationReceipt]], recipient_provenance: object | None=None, recipient_selection: _ContextViewSelectionReceipt | None=None, creation_target=None, recipient_receipt_validator: Callable[[Session, ParticipantSessionCreationReceipt], object] | None=None) -> tuple[Session, ParticipantSessionCreationReceipt]

load_participant_session_binding (method)

async SessionStore.load_participant_session_binding(session_id: str) -> ParticipantSessionBinding | None

load_participant_session_creation_receipt (method)

async SessionStore.load_participant_session_creation_receipt(session_id: str) -> ParticipantSessionCreationReceipt | None

lookup_participant_session_creation (method)

async SessionStore.lookup_participant_session_creation(creation_request: ParticipantSessionCreationRequest) -> tuple[Session, ParticipantSessionCreationReceipt] | None

capture_context_view_publication_source (method)

async SessionStore.capture_context_view_publication_source(session_id: str) -> ContextViewPublicationSource

Capture a bounded completed-turn snapshot under one native read boundary.

capture_recipient_continuation (method)

async SessionStore.capture_recipient_continuation(session_id: str) -> RecipientContinuationSelection

Select exact identity without acquiring a writer or execution authority.

Wrappers must independently qualify this coherent native read. Inheriting an implementation is not evidence that wrapper-specific state is covered.

publish_context_view (method)

async SessionStore.publish_context_view(manifest: _ContextViewManifest, *, publication_key: str) -> _ContextViewManifest

lookup_context_view_selection (method)

async SessionStore.lookup_context_view_selection(selection_key: str) -> _ContextViewSelectionReceipt | None

lookup_context_view_publication (method)

async SessionStore.lookup_context_view_publication(publication_key: str) -> _ContextViewManifest | None

select_context_view (method)

async SessionStore.select_context_view(request: _ContextViewSelectionRequest) -> _ContextViewSelectionReceipt

read_context_view_selection_decision (method)

async SessionStore.read_context_view_selection_decision(request)

Exact receiving readback, not current disclosure or ownership authority.

read_context_view (method)

async SessionStore.read_context_view(view_id: str, *, source_session_id: str) -> _ContextViewReadback

transition_context_view_ownership (method)

async SessionStore.transition_context_view_ownership(request: _ContextViewOwnershipRequest) -> _ContextViewSelectionReceipt

validate_context_view_source_closure (method)

async SessionStore.validate_context_view_source_closure(session_id: str) -> None

Refuse source deletion while an active context-view pin needs it.

validate_context_view_compaction (method)

async SessionStore.validate_context_view_compaction(session_id: str, expected_transcript_cursor: int) -> None

Prove retained view material survives a source-session compaction.

read_context_view_lifecycle_events (method)

async SessionStore.read_context_view_lifecycle_events(view_id: str, *, limit: int=256, owner_participant: _ParticipantRef | None=None) -> tuple[_ContextViewLifecycleEvent, ...]

Read bounded evidence; apply an exact participant filter before limiting.

durable_state_paths (method)

SessionStore.durable_state_paths() -> tuple[Path, ...]

Return local files that own this store’s durable state.

In-memory and remote stores return no paths. Filesystem-confined features must fail closed when the store cannot provide this evidence; custom local stores can opt in by overriding this method.

public_authority_alias_codec (property)

SessionStore.public_authority_alias_codec() -> PublicAuthorityAliasCodec | None

Return the immutable codec bound to this store, when supported.

register_public_authority_alias (method)

async SessionStore.register_public_authority_alias(public_alias: str, *, field_name: str, private_value: str, scope_session_id: str | None=None) -> None

Persist one public-to-private authority mapping before exposure.

This optional capability is deliberately explicit. A runtime must not publish a one-way public alias unless its store can resolve that alias after restart and from another worker.

resolve_public_authority_alias (method)

async SessionStore.resolve_public_authority_alias(public_alias: str, *, field_name: str, scope_session_id: str | None=None) -> str | None

Resolve one exact indexed alias within its authority scope.

public_authority_private_value_exists (method)

async SessionStore.public_authority_private_value_exists(private_value: str, *, field_name: str, scope_session_id: str | None=None) -> bool

Return positive indexed evidence that private authority exists.

issue_targeted_tool_grants (method)

async SessionStore.issue_targeted_tool_grants(session_id: str, *, expected_run_epoch: int, records: tuple[TargetedToolGrantRecord, ...], events: tuple[Event, ...]) -> TargetedToolGrantIssueResult

Atomically issue or rejoin one exact interaction-scoped grant batch.

list_targeted_tool_grants (method)

async SessionStore.list_targeted_tool_grants(session_id: str, *, interaction_id: str | None=None, limit: int=TARGETED_TOOL_GRANT_INSPECTION_MAX_RECORDS) -> tuple[TargetedToolGrantRecord, ...]

Load a bounded session or interaction grant view for inspection.

load_targeted_tool_grant_state (method)

async SessionStore.load_targeted_tool_grant_state(session_id: str) -> TargetedToolGrantStateSnapshot

Load complete typed grant state for a trusted durable export.

bind_targeted_tool_grant_use (method)

async SessionStore.bind_targeted_tool_grant_use(request: TargetedToolUseRequest, *, observed_at: datetime) -> TargetedToolUseResult

Atomically consume one budget unit or rejoin its exact prior binding.

revoke_targeted_tool_grant (method)

async SessionStore.revoke_targeted_tool_grant(tool_ref: str, *, session_id: str, expected_run_epoch: int, reason: str, revoked_at: datetime) -> TargetedToolGrantRecord | None

Atomically revoke one exact session-scoped reference, idempotently.

reconstruct_targeted_tool_grants (method)

async SessionStore.reconstruct_targeted_tool_grants(session_id: str, *, expected_run_epoch: int, interaction_id: str, generation_id: str, agent_name: str, task_id: str | None, environment_name: str | None, principal: str | None, tenant: str | None, catalogue_revision: str, descriptors_by_id: Mapping[str, tuple[str, str, str]], capability_ceiling_names: frozenset[str], observed_at: datetime) -> TargetedToolGrantReconstructionResult

Validate durable grants against current bounded registration evidence.

create (method)

async SessionStore.create(request: RunRequest, *, identity: SessionIdentity, interaction_started_event: Event | None=None, interaction_source_messages: list[Message] | None=None, checkpoint_transform: CheckpointTransform | None=None, result_checkpoint_transform: CheckpointTransform | None=None, operation_initializer: SessionOperationInitializer | None=None, participant_binding_factory: Callable[[Session], tuple[ParticipantSessionBinding, ParticipantSessionCreationReceipt]] | None=None, participant_request_commitment: str | None=None, participant_provenance: object | None=None, recipient_selection: _ContextViewSelectionReceipt | None=None, creation_target=None, recipient_receipt_validator: Callable[[Session, ParticipantSessionCreationReceipt], object] | None=None) -> Session

Create a session, optionally admitting its first interaction atomically.

Implementations must mint the final session ID, load any requested parent, and call session_invocation_for_run_request inside the same create boundary. The returned immutable invocation record is persisted with the session and preserved by all later updates.

With admission, create the session directly in RUNNING, claim its first run epoch, and persist the start event plus deferred source batch in the same transaction. Both optional arguments are supplied together. checkpoint_transform projects initial authority inside the same transaction. Without admission, stores that advertise supports_pending_session_initial_checkpoint invoke it with a PENDING session and no current checkpoint. Stores that do not advertise that capability must reject such a request before creating the session.

operation_initializer receives the final detached session and synchronously returns bounded operation records that must commit atomically with it. The callback runs inside the creation boundary and must be deterministic and side-effect-free. Stores must reject the callback or any record before exposing a partial session.

create_fork (method)

async SessionStore.create_fork(*, source_session_id: str, fork: Session, source_statuses: set[SessionStatus], transcript_cursor: int | None, checkpoint_transform: CheckpointTransform | None, system_prompt_replacement: ForkSystemPromptReplacement | None=None, expected_source_run_epoch: int, operation_initializer: SessionOperationInitializer | None=None) -> Session

Create a forked session with copied transcript/checkpoint state.

A prompt replacement, when supplied by the runtime, atomically removes inherited system messages and installs its one authoritative system message while preserving non-system message attribution.

create_fork_with_transcript_validation (method)

async SessionStore.create_fork_with_transcript_validation(*, source_session_id: str, fork: Session, source_statuses: set[SessionStatus], transcript_cursor: int | None, checkpoint_transform: CheckpointTransform | None, system_prompt_replacement: ForkSystemPromptReplacement | None=None, expected_source_run_epoch: int, transcript_validator: ForkTranscriptValidator, operation_initializer: SessionOperationInitializer | None=None) -> Session

Create a fork after validating the final transcript inside the copy boundary.

Custom stores must opt into this safety capability explicitly. The conservative default prevents the runtime from silently falling back to a time-of-check/time-of-use transcript validation. When prompt replacement is requested, the validator receives the atomically transformed child transcript rather than the source transcript.

create_profiled_fork (method)

async SessionStore.create_profiled_fork(*, source_session_id: str, fork: Session, source_statuses: set[SessionStatus], transcript_cursor: int | None, checkpoint_transform: CheckpointTransform | None, system_prompt_replacement: ForkSystemPromptReplacement | None=None, expected_source_run_epoch: int, relationship: SessionForkProfileRelationship, events: list[Event], transcript_validator: ForkTranscriptValidator | None=None, checkpoint_authority_decoder: ForkCheckpointAuthorityDecoder | None=None, operation_initializer: SessionOperationInitializer | None=None) -> ProfiledSessionForkResult

Atomically create a child and its immutable profile evidence.

load (method)

async SessionStore.load(session_id: str) -> Session | None

Load a session by id.

inspect_session_execution (method)

async SessionStore.inspect_session_execution(session_id: str) -> SessionExecutionState

Read observational liveness; custom stores without leases return unknown.

load_state (method)

async SessionStore.load_state(session_id: str) -> SessionStateSnapshot | None

Load bounded mutable state without labels or unbounded metadata.

load_invocation_snapshot (method)

async SessionStore.load_invocation_snapshot(session_id: str) -> SessionInvocationSnapshot | None

Load bounded immutable provenance without labels or metadata.

inspect_identity (method)

async SessionStore.inspect_identity(session_id: str) -> SessionInspectionIdentity

Load bounded session identity without materializing session metadata.

update_status (method)

async SessionStore.update_status(session_id: str, status: SessionStatus) -> Session

Update session status and return the updated session.

transition_status (method)

async SessionStore.transition_status(session_id: str, *, from_statuses: set[SessionStatus], to_status: SessionStatus) -> Session

Atomically transition status, claiming a new epoch when entering RUNNING.

Entering RUNNING must reject an incomplete-recovery claim whose lease has expired at the store’s authoritative transaction time.

transition_status_and_checkpoint (method)

async SessionStore.transition_status_and_checkpoint(session_id: str, *, from_statuses: set[SessionStatus], to_status: SessionStatus, checkpoint_transform: CheckpointTransform | None=None, store_time_checkpoint_transform: StoreTimeCheckpointTransform | None=None, result_checkpoint_transform: CheckpointTransform | None=None, interaction_started_event: Event | None=None, interaction_source_messages: list[Message] | None=None, continued_interaction_id: str | None=None, defer_interaction_source: bool=False, model_transition: SessionModelTransition | None=None, execution_profile: ExecutionProfileIdentity | None=None, execution_profile_decision: ExecutionProfileDecision | None=None, adopted_runtime_identity: SessionRuntimeIdentity | None=None, tool_capability_ceiling: ToolCapabilityCeiling | None=None, expected_latest_interaction_event_id: str | None=None, require_no_active_model_completion_dispatch: bool=False, temporary_service_admission: TemporaryServiceAdmission | None=None) -> Session

Atomically persist status, checkpoint, interaction, target, and profile admission.

Entering RUNNING must reject an incomplete-recovery claim whose lease has expired at the store’s authoritative transaction time.

admit_execution_profile_resume (method)

async SessionStore.admit_execution_profile_resume(session_id: str, *, from_statuses: set[SessionStatus], to_status: SessionStatus, checkpoint_transform: CheckpointTransform | None=None, store_time_checkpoint_transform: StoreTimeCheckpointTransform | None=None, result_checkpoint_transform: CheckpointTransform | None=None, execution_profile: ExecutionProfileIdentity, tool_capability_ceiling: ToolCapabilityCeiling | None=None, execution_profile_decision: ExecutionProfileDecision | None=None, adopted_runtime_identity: SessionRuntimeIdentity | None=None, interaction_started_event: Event | None=None, interaction_source_messages: list[Message] | None=None, continued_interaction_id: str | None=None, defer_interaction_source: bool=False, model_transition: SessionModelTransition | None=None, temporary_service_admission: TemporaryServiceAdmission | None=None) -> Session

Atomically compare a profile through the normal resume transition.

admit_session_invocation (method)

async SessionStore.admit_session_invocation(session_id: str, *, admission: SessionInvocationAdmission) -> Session

Atomically claim one valid profiled invocation and its interaction.

apply_invocation_lifecycle_command (method)

async SessionStore.apply_invocation_lifecycle_command(command: object) -> object

Apply one command only after authenticating the store protocol.

settle_session_invocation (method)

async SessionStore.settle_session_invocation(command: Any) -> InteractionTransitionResult

Atomically settle an interaction under exact active authority.

release_session_invocation (method)

async SessionStore.release_session_invocation(command: Any) -> Any

Atomically revoke one exact active invocation epoch.

reject_execution_profile_resume (method)

async SessionStore.reject_execution_profile_resume(session_id: str, *, expected_session_instance_id: str | None=None, expected_statuses: set[SessionStatus], expected_run_epoch: int, expected_profile: ExecutionProfileIdentity, candidate_profile: ExecutionProfileIdentity, event: Event, decision: ExecutionProfileDecision | None=None, expected_active_invocation_profile_authority: CheckpointValueAuthority | None=None) -> ExecutionProfileRejectionResult

Atomically reject a changed resume profile without admitting work.

reject_active_invocation_execution_profile (method)

async SessionStore.reject_active_invocation_execution_profile(session_id: str, *, expected_session_instance_id: str | None=None, expected_statuses: set[SessionStatus], expected_run_epoch: int, expected_active_invocation_profile: ActiveInvocationExecutionProfile, candidate_profile: ExecutionProfileIdentity, event: Event, decision: ExecutionProfileDecision | None=None) -> ExecutionProfileRejectionResult

Atomically compare checkpoint authority and reject changed continuation.

transition_status_if_no_queued_messages (method)

async SessionStore.transition_status_if_no_queued_messages(session_id: str, *, from_statuses: set[SessionStatus], to_status: SessionStatus, checkpoint_mutation: dict[str, Any] | None=None) -> Session

Atomically terminalize only when no durable queued input remains.

A completion-finalization checkpoint mutation keeps RUNNING at the current epoch while installing its existing owned finalization marker. It does not admit a new invocation or rewrite an interaction receipt.

publish_interaction_transition (method)

async SessionStore.publish_interaction_transition(session_id: str, *, event: Event, from_statuses: set[SessionStatus], to_status: SessionStatus, only_if_no_queued_messages: bool=False, model_completion_stage_settlement: ModelCompletionStageSettlementRequest | None=None, checkpoint_mutation: dict[str, Any] | None=None, terminal_event: Event | None=None, terminal_decision: InvocationTerminalDecision | None=None, expected_session_instance_id: str | None=None, expected_active_invocation_profile: ActiveInvocationExecutionProfile | None=None, expected_invocation_authority_state: Literal['active', 'released']='active', expected_recovery_claim_id: str | None=None, terminalization_only: bool=False, terminalization_plan_ownership: Any=None) -> InteractionTransitionResult

Atomically publish one interaction state and its session transition.

A completion guarded by only_if_no_queued_messages always publishes the interaction event, but leaves the session status unchanged when queued input remains. Repeating the exact complete transition reconstructs the committed result without re-evaluating that queue boundary. Reusing its event identity with different transition data fails closed. Implementations validate a runtime-supplied recovery claim against store-owned time after acquiring the mutation lock and before first publication, and reject an unclaimed first publication while a store-time-live recovery claim exists. Exact receipt replay remains valid after the lease has settled. When terminal evidence is supplied, the interaction event, session terminal event, terminal status, and exact winning decision commit in one transaction.

load_interaction_transition_receipt (method)

async SessionStore.load_interaction_transition_receipt(session_id: str, *, transition: InteractionTransitionSpec, expected_recovery_claim_id: str | None=None) -> InteractionTransitionReceiptResult | None

Load immutable evidence for one exact interaction transition.

Stores that support acknowledgement-loss recovery must implement this read-only boundary. The default keeps existing custom stores source compatible while making unsupported cancellation reconciliation fail closed instead of issuing another transition publication. Claim-bound receipts require the exact original claim and reject a superseding durable claimant, while ordinary lease expiry does not invalidate exact acknowledgement-loss readback.

load_historical_interaction_settlement (method)

async SessionStore.load_historical_interaction_settlement(session_id: str, *, expected_session_instance_id: str, expected_event: Event, expected_profile: ExecutionProfileIdentity) -> InteractionTransitionReceiptResult

Read exact prior settlement evidence without granting run authority.

Unlike invocation release lookup, this permits a later current epoch. It cannot release a fence or authorize a historical invocation: callers must separately acquire their own execution or mutation authority.

load_invocation_settlement_transition (method)

async SessionStore.load_invocation_settlement_transition(session_id: str, *, expected_session_instance_id: str, expected_active_invocation_profile: ActiveInvocationExecutionProfile) -> InteractionTransitionSpec | None

Resolve exact terminal settlement for one active invocation.

Event lookup is bounded and supplies only the receipt identity. The backend authenticates the complete stored receipt and its invocation tuple before returning the transition. Release compares the same receipt again inside its mutation transaction.

fence_stalled_run (method)

async SessionStore.fence_stalled_run(session_id: str, *, statuses: set[SessionStatus], inactive_for_seconds: int) -> Session | None

Atomically evict a stale run and claim its newly incremented epoch.

Returns None when the session is no longer in one of statuses or has not been inactive for inactive_for_seconds at store time. An existing incomplete-recovery claim must be reconciled through the exact checkpoint-authorized takeover path instead.

reserve_stalled_run_recovery (method)

async SessionStore.reserve_stalled_run_recovery(session_id: str, *, statuses: set[SessionStatus], inactive_for_seconds: int | None, checkpoint_transform: StoreTimeCheckpointTransform) -> Session | None

Reserve recovery at store time without yet advancing the run epoch.

The reservation transform and eligibility check are atomic. The method intentionally leaves session activity unchanged so a subsequent typed invocation rebind can authenticate and fence the exact reserved state.

fence_run_and_transform_checkpoint (method)

async SessionStore.fence_run_and_transform_checkpoint(session_id: str, *, statuses: set[SessionStatus], checkpoint_transform: CheckpointTransform, result_checkpoint_transform: CheckpointTransform | None=None) -> Session

Atomically transform a checkpoint and claim a newly incremented epoch.

The transform must return the replacement checkpoint. An exception or None result aborts both the checkpoint update and the epoch fence. This operation is for ownership that must persist a checkpoint lease and fence the prior run as one transaction. Recovery that needs only an inactivity predicate should continue to use fence_stalled_run. An incomplete-recovery claim present in the source checkpoint must still be live at the store’s authoritative transaction time.

release_run_fence (method)

async SessionStore.release_run_fence(session_id: str) -> None

Revoke this task’s epoch after all trailing writes finish.

claim_budget_reservation_identity (method)

async SessionStore.claim_budget_reservation_identity(*, reservation_id: str, publication_session_id: str, publication_id: str) -> None

Durably bind one ledger-wide reservation id to an exact session event.

Claims are idempotent only for the same session/event pair and must not be removed when a session is deleted. append_events atomically marks the matching claim published when it stores budget.reserved. The claim is a session-scoped mutation: the publication session must exist and an active run fence must still own its current epoch.

append_event (method)

async SessionStore.append_event(session_id: str, event: Event) -> None

Append one event to a session.

append_events (method)

async SessionStore.append_events(session_id: str, events: list[Event]) -> None

Append events to a session in one durable batch.

append_workflow_step_started (method)

async SessionStore.append_workflow_step_started(session_id: str, event: Event, *, workflow_name: str, attempt_id: str) -> bool

Atomically append a step reservation only for the current attempt.

The attempt-fence check and event insert share one store transaction. False means either a newer attempt won or this exact reservation was already published.

append_tool_effect_conflict (method)

async SessionStore.append_tool_effect_conflict(request: object) -> Event

Append exact late-dispatch evidence without granting run authority.

Implementations compare the retained winner and session incarnation in the append transaction. This narrow operation accepts only an internal ToolEffectConflictAudit, constructs its own event, preserves liveness and all execution state, and returns original evidence on exact replay. A stale run context is permitted only for this evidence-only operation.

load_mcp_manifest_baselines (method)

async SessionStore.load_mcp_manifest_baselines(history_keys: tuple[str, ...]) -> McpManifestBaselineLoadResult

Load authoritative baselines for the requested manifest-history keys.

compare_and_publish_mcp_manifest_checks (method)

async SessionStore.compare_and_publish_mcp_manifest_checks(session_id: str, *, expected_generations: dict[str, int | None], baseline_updates: dict[str, McpManifestBaseline], events: list[Event]) -> McpManifestPublicationResult

Atomically publish decisions and return the exact resulting requested baselines.

Every accepted first_seen or changed event must install the matching authoritative baseline in the same publication. Every expected history key must have exactly one coherent outcome; unchanged must match its current baseline, and blocked outcomes must not update one.

claim_persisted_event_side_effect (method)

async SessionStore.claim_persisted_event_side_effect(*, session_id: str | None=None, event_id: str | None=None, lease_seconds: float=300.0) -> PersistedEventSideEffectClaim | None

Claim the oldest eligible persisted event side-effect handoff.

get_persisted_event_side_effect_delivery (method)

async SessionStore.get_persisted_event_side_effect_delivery(*, session_id: str, event_id: str) -> PersistedEventSideEffectDelivery | None

Load one persisted event side-effect handoff by event identity.

claim_first_persisted_event_side_effect (method)

async SessionStore.claim_first_persisted_event_side_effect(expected: PersistedEventSideEffectDelivery) -> PersistedEventSideEffectClaim | None

Claim only the exact unattempted snapshot under the ordinary claim lock.

Never take over a lease or retry a failed delivery. Return None when unsupported or when any expected field changed; callers must reread.

retire_failed_first_event_delivery (method)

async SessionStore.retire_failed_first_event_delivery(expected: PersistedEventSideEffectDelivery) -> PersistedEventSideEffectDelivery | None

Atomically dead-letter an exact failed first delivery, preserving its error.

Compare the complete snapshot under the claim lock. A competing retry wins or retirement wins, never both. Return None on conflict or when unsupported; that absence is not proof of callback settlement.

mark_persisted_event_side_effect_delivered (method)

async SessionStore.mark_persisted_event_side_effect_delivered(claim: PersistedEventSideEffectClaim) -> PersistedEventSideEffectDelivery

Mark one claimed event’s configured side effects delivered.

mark_persisted_event_side_effect_failed (method)

async SessionStore.mark_persisted_event_side_effect_failed(claim: PersistedEventSideEffectClaim, *, error: str, max_attempts: int, retry_delay_seconds: float) -> PersistedEventSideEffectDelivery

Release one failed claim for retry or dead-letter it.

list_persisted_event_side_effect_deliveries (method)

async SessionStore.list_persisted_event_side_effect_deliveries(*, statuses: set[PersistedEventSideEffectStatus] | None=None, claimable_only: bool=False, after_sequence: int | None=None, limit: int=100) -> list[PersistedEventSideEffectDelivery]

Inspect persisted side-effect delivery state in event order.

claimable_only excludes terminal deliveries and live leases.

get_persisted_event_side_effect_health (method)

async SessionStore.get_persisted_event_side_effect_health() -> PersistedEventSideEffectHealth

Read exact aggregate delivery health without loading source events.

query_persisted_event_side_effect_deliveries (method)

async SessionStore.query_persisted_event_side_effect_deliveries(query: PersistedEventSideEffectQuery) -> PersistedEventSideEffectPage

Read a bounded operational page; inspection never reserves a claim.

defer_persisted_event_side_effect (method)

async SessionStore.defer_persisted_event_side_effect(claim: PersistedEventSideEffectClaim) -> PersistedEventSideEffectDelivery

Release an exact claim without delivery or spending a failure attempt.

Used when an internal continuation cannot yet acquire execution ownership. The persisted event remains eligible for bounded recovery; its old claim must no longer be capable of acknowledging a replacement delivery.

renew_persisted_event_side_effect (method)

async SessionStore.renew_persisted_event_side_effect(claim: PersistedEventSideEffectClaim, *, lease_seconds: float=300.0) -> PersistedEventSideEffectDelivery

Extend an exact, still-live claim using the store’s ownership clock.

Expired ownership cannot be revived, even before another worker claims the event. Renewal preserves the claim identity and attempt count.

enqueue_session_message (method)

async SessionStore.enqueue_session_message(request: EnqueueSessionMessageRequest, *, expected_authorized_target_instance_id: str | None=None) -> EnqueueSessionMessageResult

Atomically accept queued input with its causal event, or replay it.

When supplied by the authorization owner, the target incarnation must match under the transaction lock before either replay or admission. This fence is not a persisted message freshness condition.

append_peer_content (method)

async SessionStore.append_peer_content(request: PeerContentAppendRequest, *, qualify_target: Callable[[Session], None] | None=None, pending_transcript_cursor: int | None=None) -> PeerContentReceipt

Atomically append or replay one authenticated peer occurrence.

The default is fail-closed: a custom store cannot accidentally expose peer content by inheriting a user-message implementation. Before queue insertion, native owners call the runtime’s synchronous, side-effect-free qualifier with a detached transaction-owned target. Missing qualification permits only pending/excluded decisions or replay. pending_transcript_cursor is a receiving-owner fence for servicing an already retained exact pending operation. It never changes caller intent.

read_peer_content (method)

async SessionStore.read_peer_content(append_key: PeerAppendKey) -> PeerContentReceipt | None

Return the bounded append receipt for an exact key, if present.

exclude_peer_content (method)

async SessionStore.exclude_peer_content(request: PeerContentAppendRequest, *, reason: str) -> PeerContentReceipt

Record a definitive exclusion without pretending it was appended.

read_peer_content_attempt (method)

async SessionStore.read_peer_content_attempt(request: PeerContentAppendRequest) -> PeerContentReceipt | None

Reconcile the complete expected operation, including excluded predecessors.

list_pending_peer_content (method)

async SessionStore.list_pending_peer_content(*, after_operation_key=None, limit=32)

Discover bounded peer responsibility independently of creation settlement.

retry_pending_peer_content (method)

async SessionStore.retry_pending_peer_content(session_id: str, *, expected_session_instance_id: str, expected_run_epoch: int, expected_transcript_cursor: int, admit=None) -> tuple[PeerContentReceipt, ...]

Retry retained peer appends only when the existing host reaches a boundary.

record_peer_content_exposure (method)

async SessionStore.record_peer_content_exposure(request: PeerContentExposureRequest) -> PeerContentExposureReceipt

begin_peer_content_exposure (method)

async SessionStore.begin_peer_content_exposure(request: PeerContentExposureRequest) -> PeerContentExposureReceipt

Durably reserve an exposure before provider serialization.

read_peer_content_exposure (method)

async SessionStore.read_peer_content_exposure(append_key: PeerAppendKey, exposure_id: str) -> PeerContentExposureReceipt | None

inspect_session_messages (method)

async SessionStore.inspect_session_messages(query: SessionMessageQuery, *, expected_authorized_session_instance_id: str | None=None) -> SessionMessageInspection

Read one bounded delivery-priority page, retaining unreadable row identities.

Check the optional authorization-owner incarnation before reading queue content, including on the first page without a cursor.

apply_session_message_action (method)

async SessionStore.apply_session_message_action(request: SessionMessageActionRequest) -> SessionMessageActionResult

Atomically withdraw/quarantine an exact undelivered row and publish its event.

snapshot_session_message_source (method)

async SessionStore.snapshot_session_message_source(session_id: str, *, include_transcript_digest: bool=False, include_checkpoint_digest: bool=False, expected_authorized_session_instance_id: str | None=None) -> SessionMessageSource

Capture source identity/cursor and requested digests under one store snapshot.

Check the optional authorization-owner incarnation before accessing transcript or checkpoint content, within that same snapshot.

deliver_queued_session_messages (method)

async SessionStore.deliver_queued_session_messages(session_id: str, *, include_on_idle: bool, delivery_id: str | None=None, eligible_through: int | None=None, limit: int=SESSION_MESSAGE_DELIVERY_BATCH_LIMIT, interaction_id: str | None=None, interaction_started_event: Event | None=None, profile_handoff: QueuedInteractionProfileHandoff | None=None, reject_only: bool=False) -> SessionMessageDeliveryBatch

Atomically append and mark one bounded queue batch delivered.

Runtime callers supply a stable delivery_id for acknowledgement-loss reconstruction. Omitting it creates a one-shot identity for direct store callers and returns that identity in the result.

reject_only retires stale/expired records without delivering valid input, for a runtime that has exhausted its model-step authority. Valid records remain pending; has_more then describes remaining rejection work inside the fixed eligibility boundary. The mode is part of the exact delivery identity and cannot change on replay.

repair_queued_interaction_profile_handoff (method)

async SessionStore.repair_queued_interaction_profile_handoff(session_id: str, *, interaction_started_event: Event, profile_handoff: QueuedInteractionProfileHandoff) -> ActiveInvocationExecutionProfile

Repair one historical split delivery from exact provider-stage evidence.

This compatibility seam is intentionally fail-closed for custom stores. New deliveries must bind the target profile in deliver_queued_session_messages instead.

publish_checkpoint_and_events (method)

async SessionStore.publish_checkpoint_and_events(session_id: str, *, checkpoint_transform: CheckpointTransform, events: list[Event], expected_statuses: set[SessionStatus] | None=None, expected_run_epoch: int | None=None, expected_transcript_cursor: int | None=None) -> Session

Atomically transform a checkpoint and append its causal event batch.

publish_checkpoint_and_events_with_store_time (method)

async SessionStore.publish_checkpoint_and_events_with_store_time(session_id: str, *, idempotency_key: str, checkpoint_transform: StoreTimeCheckpointTransform, commit_time_guard: Callable[[datetime], None], events: list[Event], expected_statuses: set[SessionStatus] | None=None, expected_run_epoch: int | None=None, expected_transcript_cursor: int | None=None) -> Session

Publish checkpoint state using the backend’s transactional clock.

load_session_operation (method)

async SessionStore.load_session_operation(session_id: str, idempotency_key: str, *, checkpoint_root_guard: CheckpointRootFieldGuard | None=None) -> dict[str, Any] | None

Load one terminal durable operation record by caller idempotency key.

A supplied root guard runs in the same read snapshot before the checkpoint-coupled operation record is interpreted.

publish_session_operation (method)

async SessionStore.publish_session_operation(session_id: str, *, idempotency_key: str, operation_transform: SessionOperationTransform, events: list[Event], expected_statuses: set[SessionStatus] | None=None, expected_run_epoch: int | None=None, expected_transcript_cursor: int | None=None) -> Session

Atomically publish a checkpoint, events, and terminal operation records.

supports_external_waits (method)

SessionStore.supports_external_waits() -> bool

publish_session_operation_guarded (method)

async SessionStore.publish_session_operation_guarded(session_id: str, *, idempotency_key: str, operation_transform: SessionOperationTransform, commit_guard: Callable[[], None], events: list[Event], expected_statuses: set[SessionStatus] | None=None, expected_run_epoch: int | None=None, expected_transcript_cursor: int | None=None) -> Session

Publish with a guard evaluated at the backend’s commit boundary.

publish_session_operation_guarded_with_store_time (method)

async SessionStore.publish_session_operation_guarded_with_store_time(session_id: str, *, idempotency_key: str, operation_transform: StoreTimeSessionOperationTransform, commit_guard: Callable[[], None], commit_time_guard: Callable[[datetime], None], events: list[Event], expected_statuses: set[SessionStatus] | None=None, expected_run_epoch: int | None=None, expected_transcript_cursor: int | None=None, context_view_compaction_cursor: int | None=None) -> Session

Publish an operation using time sampled at the commit transaction.

prepare_continuation_ticket (method)

async SessionStore.prepare_continuation_ticket(command: ContinuationPreparation) -> ContinuationRecord

Atomically retain one exact continuation ticket in ARMING state.

lookup_continuation_ticket (method)

async SessionStore.lookup_continuation_ticket(expected: ContinuationPreparation) -> ExactLookup[ContinuationRecord]

Return exact retained preparation evidence, never infer an outcome.

load_continuation_ticket (method)

async SessionStore.load_continuation_ticket(session_id: str, *, registration_key: str, session_instance_id: str) -> ContinuationRecord | None

Read one continuation aggregate without treating absence as proof.

mark_continuation_waiting (method)

async SessionStore.mark_continuation_waiting(ticket: ContinuationTicket) -> ContinuationRecord

Publish the ARMING-to-WAITING transition under the ticket’s CAS.

latch_continuation (method)

async SessionStore.latch_continuation(latch: ContinuationLatch) -> ContinuationRecord

Publish one immutable readiness latch, including an ARMING latch.

consume_continuation (method)

async SessionStore.consume_continuation(consumption: ContinuationConsumption) -> ContinuationRecord

Atomically elect one inline or queued continuation and replay it.

retire_continuation (method)

async SessionStore.retire_continuation(retirement: ContinuationRetirement) -> ContinuationRecord

Atomically retire a ticket, losing only to an already committed consume.

load_runtime_publication_receipt (method)

async SessionStore.load_runtime_publication_receipt(session_id: str, publication_id: str) -> RuntimePublicationReceipt | None

Load and verify one store-owned runtime publication receipt.

publish_runtime_publication (method)

async SessionStore.publish_runtime_publication(session_id: str, *, request: RuntimePublicationRequest, expected_statuses: set[SessionStatus] | None=None, expected_run_epoch: int | None=None, expected_transcript_cursor: int | None=None) -> RuntimePublicationResult

Atomically publish or exactly replay one logical runtime boundary.

load_model_completion_stage (method)

async SessionStore.load_model_completion_stage(session_id: str, stage_id: str) -> ModelCompletionStage | None

Load and verify append-only pre-dispatch and terminal completion evidence.

load_model_completion_stage_settlement (method)

async SessionStore.load_model_completion_stage_settlement(session_id: str, stage_id: str) -> ModelCompletionStageSettlement | None

Load verified terminal non-success evidence for one exact stage.

load_model_completion_stage_abandonment (method)

async SessionStore.load_model_completion_stage_abandonment(session_id: str, stage_id: str) -> ModelCompletionStageAbandonment | None

Read positive no-dispatch evidence after the preparation was removed.

This is a query, not permission to dispatch. Re-preparation compares the complete expected predecessor and this receipt in its own transaction. The shared operation-record lookup keeps all supported backends under their existing read owner; no second route cache is needed after restart.

load_active_model_completion_stage (method)

async SessionStore.load_active_model_completion_stage(session_id: str) -> ActiveModelCompletionStage | None

Load a verified snapshot of the dispatch currently eligible to publish.

load_model_completion_stage_dispatch (method)

async SessionStore.load_model_completion_stage_dispatch(session_id: str, stage_id: str) -> ModelCompletionStageDispatch | None

Load exact evidence that a prepared stage crossed the dispatch fence.

mark_model_completion_stage_dispatched (method)

async SessionStore.mark_model_completion_stage_dispatched(session_id: str, *, stage: ModelCompletionStage, consume_child_session_notifications: bool=True) -> ModelCompletionStageDispatch

Durably cross the last local fence before entering provider-controlled code.

prepare_model_completion_stage (method)

async SessionStore.prepare_model_completion_stage(session_id: str, *, request: ModelCompletionStageRequest, expected_statuses: set[SessionStatus], expected_run_epoch: int, expected_transcript_cursor: int) -> ModelCompletionStageResult

Insert or replay a stage; dispatch only when the result explicitly authorizes it.

complete_recovered_model_completion_stage (method)

async SessionStore.complete_recovered_model_completion_stage(session_id: str, *, stage_id: str, publication: RuntimePublicationRequest, recovery_fence: ModelCompletionStageRecoveryFence) -> ModelCompletionStageResult

Insert synthetic auxiliary terminal evidence under an atomic recovery fence.

Backends overriding the atomic hook must explicitly attest this protocol. Ordinary provider completion continues through complete_model_completion_stage.

complete_model_completion_stage (method)

async SessionStore.complete_model_completion_stage(session_id: str, *, stage_id: str, publication: RuntimePublicationRequest) -> ModelCompletionStageResult

Append or exactly replay immutable terminal provider-attempt material.

For auxiliary inference, terminal material records the explicit attempt outcome and accounting; it does not imply a successful model response.

abandon_model_completion_stage (method)

async SessionStore.abandon_model_completion_stage(session_id: str, *, stage_id: str, preparation_digest: str, expected_run_epoch: int, stage_source_run_epoch: int | None=None) -> ModelCompletionStageAbandonmentResult

Atomically abandon one exact active stage before provider dispatch.

An exact replay reconstructs the durable abandonment tombstone. A later preparation of the same request receives a new preparation digest, so a delayed retry of an older abandonment can never clear it.

promote_model_completion_stage (method)

async SessionStore.promote_model_completion_stage(session_id: str, *, stage_id: str, expected_run_epoch: int) -> RuntimePublicationResult

Atomically verify and publish a terminal stage without a load/publish race.

load_session_export_snapshot (method)

async SessionStore.load_session_export_snapshot(session_id: str, *, limits: SessionExportLimits | None=None) -> SessionExportSnapshot | None

Read all export-owned components at one bounded per-session snapshot.

Return None if the session has been deleted. No sequential-read fallback is safe for custom stores: implement one lock or database snapshot.

load_events (method)

async SessionStore.load_events(session_id: str) -> list[Event]

Load all events for a session.

load_user_input_supersession_events (method)

async SessionStore.load_user_input_supersession_events(session_id: str, input_id: str) -> list[Event]

Load exact interruption evidence for one user-input pause.

Built-in persistent stores override this with their fixed-size pending-action lookup indexes. The fallback pages complete durable history for correctness without retaining an unbounded result.

load_tool_round_lifecycle_events (method)

async SessionStore.load_tool_round_lifecycle_events(session_id: str, tool_call_ids: list[str] | tuple[str, ...]) -> list[Event]

Load exact started/terminal candidates for bounded tool-call identities.

Built-in stores override this with their fixed-size pending-action indexes. The paginated fallback preserves correctness for custom stores without silently truncating lifecycle evidence.

load_tool_round_lifecycle_events_for_round (method)

async SessionStore.load_tool_round_lifecycle_events_for_round(session_id: str, tool_call_ids: list[str] | tuple[str, ...], *, tool_round_identity: ToolRoundIdentity) -> list[Event]

Load bounded lifecycle evidence for one exact logical tool round.

Built-in stores provide indexed implementations. The default implementation pages over durable lifecycle events so reused provider call ids cannot hide current-round evidence behind the publication binding limit.

query_events (method)

async SessionStore.query_events(query: EventQuery | None=None) -> list[EventRecord]

Query stored events with durable sequence cursors.

read_usage_accounting (method)

async SessionStore.read_usage_accounting(query: EventQuery, *, by_session: bool=False, by_identity: bool=False) -> UsageAccountingSnapshot

Reduce exact usage in one stable store snapshot with 256-row pages.

The result retains totals, identity lists and optional per-session rows, never historical events. after_sequence is a single shared watermark across every usage-bearing event type. Unsupported stores fail closed.

read_cost_accounting (method)

async SessionStore.read_cost_accounting(query: EventQuery, pricing: PriceBook, *, currency: str='USD', details: bool=False, by_session: bool=False, additional_events: tuple[Event, ...]=(), max_detail_bytes: int | None=None, previous: CostAccountingSnapshot | None=None) -> CostAccountingSnapshot

Price a stable snapshot with bounded working memory plus requested output.

Built-ins group hosted evidence and completions by session and attempt; totals-only reads retain no historical line items. Unsupported stores fail closed.

event_exists (method)

async SessionStore.event_exists(query: EventQuery) -> bool

Test matching durable evidence without hydrating event payloads.

query_events_bounded (method)

async SessionStore.query_events_bounded(query: EventQuery, *, max_bytes: int) -> list[EventRecord]

Query events only when the backend can enforce bytes before hydration.

load_bounded (method)

async SessionStore.load_bounded(session_id: str, *, max_bytes: int) -> Session | None

Load a full session only after bounding transport bytes in one snapshot.

export_terminal_session_evidence (method)

async SessionStore.export_terminal_session_evidence(session_id: str, *, spool: EvidenceSpool) -> None

Export a stable terminal snapshot to bounded indexed backing.

Optional capability. Unsupported stores fail before reading payloads; callers must explicitly choose the separate bounded eager fallback.

load_terminal_session_evidence (method)

async SessionStore.load_terminal_session_evidence(session_id: str, *, limits: TerminalSessionEvidenceLimits | None=None) -> TerminalSessionEvidence

Read one exact bounded snapshot through the current terminal event.

Implementations must resolve terminal status, publication authority, the event cutoff, and the complete attributed transcript in one stable read. Unsupported, incomplete, contradictory, or oversized evidence fails with TerminalSessionEvidenceError; it is never truncated.

load_runner_owned_interrupted_evidence (method)

async SessionStore.load_runner_owned_interrupted_evidence(session_id: str, *, observed_events: tuple[RunnerObservedEventIdentity, ...] | None=None, expected_parent_session_id: str | None=None, limits: TerminalSessionEvidenceLimits | None=None) -> TerminalSessionEvidence

Read one exact interrupted snapshot from a runner-owned execution tree.

A root read must compare the complete durable event-log identity with observed_events. A descendant read instead proves its direct parent through expected_parent_session_id; the caller must have established that parent as part of the same freshly executed tree. Exactly one proof is required. Implementations must apply record/count/byte limits before hydrating evidence, all inside one stable store snapshot. This optional operation does not broaden ordinary historical terminal-evidence reads.

query_latest_interaction_events (method)

async SessionStore.query_latest_interaction_events(session_id: str, *, before_sequence: int | None=None, limit: int=100) -> list[EventRecord]

Return each interaction’s latest lifecycle event in descending order.

Built-in stores implement this as a native keyset query. The explicit capability error prevents endpoints from silently degrading to an unbounded event-log scan for out-of-tree stores.

summarize_events (method)

async SessionStore.summarize_events(session_id: str) -> EventSummary

Summarize stored events for one session without loading every event.

summarize_outcome (method)

async SessionStore.summarize_outcome(session_id: str) -> SessionOutcome

Derive the current session outcome from durable events.

inspect_summary (method)

async SessionStore.inspect_summary(session_id: str, *, checkpoint_root_guard: CheckpointRootFieldGuard | None=None) -> SessionInspectionSummary

Return one exact, content-free diagnostic summary.

The default implementation deliberately composes the public paginated query contracts so out-of-tree stores gain the same semantics without a backend-specific schema dependency. Implementations may replace it with an equivalent bounded native aggregate.

list_sessions (method)

async SessionStore.list_sessions(query: SessionQuery | None=None) -> SessionListResult

List sessions (filtered/sorted/paginated) with a keyset cursor and total count.

query_session_topology (method)

async SessionStore.query_session_topology(query: SessionTopologyQuery) -> SessionTopologyStoreResult

Load bounded ancestry and batched direct-child branches.

This optional store-native read prevents the control plane from issuing one list query per rendered node. Custom stores opt in explicitly through supports_session_topology.

query_session_lineage (method)

async SessionStore.query_session_lineage(query: SessionLineageQuery) -> SessionLineageResult

Load one minimal, byte-bounded direct-child lineage page.

This optional store-native read is the durable admission boundary for historical trajectory capture. Implementations must project only the bounded structural fields represented by SessionLineageNode and must not hydrate session display fields or event payloads.

query_child_session_lifecycle (method)

async SessionStore.query_child_session_lifecycle(query: ChildSessionLifecycleQuery) -> ChildSessionLifecyclePage

Read bounded canonical direct-child state and notification freshness.

Implementations advertising child_session_notification_version == 1 must resolve session state, its exact lifecycle occurrence, and any dispatch-fenced terminal consumption in one stable snapshot.

aggregate_operational_snapshot (method)

async SessionStore.aggregate_operational_snapshot(filters: SessionAggregateFilter | None=None) -> SessionOperationalSnapshot

Count current session states in one store-local read snapshot.

Default raises NotImplementedError so adding this control-plane read model does not make existing out-of-tree stores uninstantiable.

aggregate_usage (method)

async SessionStore.aggregate_usage(query: UsageRollupQuery) -> UsageRollupStoreResult

Aggregate bounded event-time activity and usage without loading histories.

Default raises NotImplementedError so adding this control-plane read model does not make existing out-of-tree stores uninstantiable.

list_sessions_with_pending_interruption_cascade (method)

async SessionStore.list_sessions_with_pending_interruption_cascade(query: SessionQuery | None=None) -> SessionListResult

List only sessions carrying a pending interruption cascade marker.

Implementations should filter at the storage layer. This is the restart-discovery path, so scanning all historical sessions and loading their checkpoints individually is not an acceptable implementation.

list_queued_dispatch_terminal_receipts (method)

async SessionStore.list_queued_dispatch_terminal_receipts(query: QueuedDispatchTerminalReceiptQuery | None=None) -> list[QueuedDispatchTerminalReceipt]

List bounded live queue/session handoffs for restart reconciliation.

query_pending_actions (method)

async SessionStore.query_pending_actions(query: PendingActionQuery | None=None, *, checkpoint_root_guard: CheckpointRootFieldGuard | None=None) -> PendingActionListResult

List current checkpoint-backed actions without scanning session histories.

Implementations must bound candidate and event reads, apply filtering at the storage layer where possible, and avoid per-session history queries. A supplied root guard runs on bounded opaque evidence in the same read snapshot before any pending state is interpreted.

validate_session_closure_admission (method)

async SessionStore.validate_session_closure_admission(session_id: str) -> None

Check every native active-operation/deletion guard without mutation.

Read the guards in one store snapshot. This preflight is not an ownership claim and does not replace final atomic deletion checks.

load_session_closure_records (method)

async SessionStore.load_session_closure_records(session_id: str, *, max_records: int, max_bytes: int) -> dict[str, Any]

Return a bounded native closure snapshot with per-class counts/bytes.

Read all included classes under one store snapshot. Exceeding a bound must return no partial records. Unsupported stores must fail explicitly.

delete_session (method)

async SessionStore.delete_session(session_id: str, *, closure_receipt: dict[str, Any] | None=None) -> None

Delete a session and cascade to its events, transcript, and checkpoint.

Raises ValueError if the session is in-flight (RUNNING or INTERRUPTING — interrupt it first), has incomplete current-run terminal evidence, still owns a durable subagent child, or if one of its accepted budget reservations has no fully delivered terminal settlement audit event. Durable subagent children must be settled and deleted before their parent so deletion cannot erase queue-admission authority while child work still exists. When closure_receipt is supplied, reject every remaining child edge in the deletion transaction, including non-subagent children. Closure callers must detach or delete those edges explicitly before finalization. Idempotent: deleting a session that does not exist is a no-op.

Default raises NotImplementedError so out-of-tree stores keep working.

detach_session_children (method)

async SessionStore.detach_session_children(parent_session_id: str, child_session_ids: tuple[str, ...], *, closure_receipt: dict[str, Any]) -> tuple[dict[str, Any], ...]

Atomically detach children and retain auditable lineage tombstones.

Replay binds the root/plan, original parent and exact unordered child set. A conflicting replay or invalid child must mutate no batch member.

load_session_closure_tombstones (method)

async SessionStore.load_session_closure_tombstones(root_session_id: str, plan_id: str) -> tuple[dict[str, Any], ...]

Load durable lineage tombstones for one closure operation.

load_session_closure_receipt (method)

async SessionStore.load_session_closure_receipt(session_id: str, plan_id: str) -> dict[str, Any] | None

Load an exact durable closure receipt, if this store supports them.

load_session_closure_progress (method)

async SessionStore.load_session_closure_progress(session_id: str, plan_id: str) -> dict[str, Any] | None

save_session_closure_progress (method)

async SessionStore.save_session_closure_progress(progress: dict[str, Any]) -> None

claim_session_closure_progress (method)

async SessionStore.claim_session_closure_progress(progress: dict[str, Any]) -> None

Atomically validate and reserve a recursive plan’s current lineage.

Verify the complete recursive edge set, not just the listed edges, including plans with no descendants. Reject unexplained additions before publishing ownership. Creation must serialize with this admission and reject destinations retired by retained closure claims or receipts. Validate every target’s native inventory against progress max_records and max_bytes in that transaction, before publishing any claim. Replay must preserve these admitted limits as part of the exact plan. Retain ownership across failures and restart. Detachment and competing deletion must reject claimed sessions; only exact closure deletion may consume them. No lease expiry may release a still-live cleanup owner.

update_labels (method)

async SessionStore.update_labels(session_id: str, labels: dict[str, str]) -> Session

Replace a session’s labels (full replacement, not a merge) and return it.

Annotation writes update updated_at but do not refresh the runtime recovery signal in last_activity_at.

Raises KeyError if the session does not exist. Default raises NotImplementedError so out-of-tree stores keep working.

update_metadata (method)

async SessionStore.update_metadata(session_id: str, metadata: dict[str, Any]) -> Session

Replace a session’s user metadata and return the updated session.

This is a full replacement of user-authored metadata, not a merge. Runtime-owned entries are retained atomically and cannot be supplied by callers. Annotation writes update updated_at but do not refresh the runtime recovery signal in last_activity_at.

Raises KeyError if the session does not exist. Default raises NotImplementedError so out-of-tree stores keep working.

append_transcript_messages (method)

async SessionStore.append_transcript_messages(session_id: str, messages: list[Message], *, interaction_id: InteractionAttribution=INHERIT_INTERACTION) -> None

Append provider-neutral transcript messages to a session.

Stored state must not alias caller-passed messages: implementations must detach nested JSON payloads (serialize, or use cayu.messages.detach_message) so a producer mutating a message after append cannot rewrite history.

replace_initial_transcript_messages (method)

async SessionStore.replace_initial_transcript_messages(session_id: str, expected_messages: list[Message], replacement_messages: list[Message], *, interaction_id: InteractionAttribution=INHERIT_INTERACTION, checkpoint_transform: CheckpointTransform | None=None, runtime_suffix_count: int=0) -> None

Atomically materialize deferred input as the ordered initial transcript.

The replacement must contain expected_messages after any leading bootstrap/system messages and before runtime_suffix_count runtime-owned messages. The bootstrap prefix is stored with null interaction attribution; the admitted source and runtime suffix are attributed to interaction_id. When deferred input retains a runtime-authenticated complete initial projection, the replacement must equal that projection exactly. No partial or externally visible pre-finalization transcript is permitted. When supplied, checkpoint_transform runs inside the same write transaction before the initial-transcript authority marker is read.

load_deferred_interaction_input (method)

async SessionStore.load_deferred_interaction_input(session_id: str) -> DeferredInteractionInput | None

Load the private admitted source batch, if one still exists.

materialize_deferred_interaction_input (method)

async SessionStore.materialize_deferred_interaction_input(session_id: str, *, interaction_id: InteractionAttribution=INHERIT_INTERACTION) -> bool

Materialize deferred source input without a bootstrap prefix.

Returns True when a pending admission was materialized and False when no matching deferred input exists. Recovery and setup-failure paths use this before making the owning interaction or session terminal.

append_transcript_messages_and_transform_checkpoint (method)

async SessionStore.append_transcript_messages_and_transform_checkpoint(session_id: str, messages: list[Message], checkpoint_transform: CheckpointTransform, *, interaction_id: InteractionAttribution=INHERIT_INTERACTION) -> None

Append transcript messages and atomically transform the current checkpoint.

The transform must be synchronous and thread-safe; a store may execute it on a worker thread while holding its transactional write boundary.

load_transcript (method)

async SessionStore.load_transcript(session_id: str) -> list[Message]

Load provider-neutral transcript messages for a session.

Returned messages must not alias stored state: callers own the result and may mutate nested payloads without corrupting the transcript.

load_transcript_snapshot (method)

async SessionStore.load_transcript_snapshot(session_id: str) -> TranscriptSnapshot

Load retained records and the permanent cursor in one read snapshot.

Unlike len(load_transcript(...)), cursor does not move backwards when a backend applies physical transcript retention.

load_transcript_cursor (method)

async SessionStore.load_transcript_cursor(session_id: str) -> int

Load the permanent append cursor without requiring retained rows.

load_transcript_window (method)

async SessionStore.load_transcript_window(session_id: str, *, start_index: int, limit: int) -> TranscriptSnapshot

Load a bounded absolute-index window and its permanent append cursor.

start_index is inclusive. If its message was removed, return the first surviving records at or after it; at or beyond the append cursor, return no records. The returned cursor is the full logical extent, not the end of this window. Continue from the last returned index plus one. An empty window does not prove that earlier payloads never existed.

Stores may override this method to avoid loading the complete retained transcript. The default keeps custom stores correct through their atomic load_transcript_snapshot implementation.

load_latest_transcript_message (method)

async SessionStore.load_latest_transcript_message(session_id: str, *, role: MessageRole) -> TranscriptRecord | None

Load one latest role-matched record without requiring full history.

Stores must override this method with a reverse-indexed read. There is deliberately no OFFSET-based fallback because its work grows with the transcript and would make callers’ bounded-read guarantees dishonest.

load_latest_transcript_text (method)

async SessionStore.load_latest_transcript_text(session_id: str, *, role: MessageRole, max_chars: int) -> tuple[str, bool] | None

Load a bounded text projection without hydrating the source message.

query_transcript (method)

async SessionStore.query_transcript(query: TranscriptQuery) -> TranscriptPage

Query provider-neutral transcript messages with stable message indexes.

Offset and total_records refer to retained role/interaction-matched rows, not logical extent. Thinking exclusion is applied after pagination; advance by the raw page span, not the number of projected records. Same result-ownership isolation contract as load_transcript.

search_transcript (method)

async SessionStore.search_transcript(query: TranscriptSearchQuery) -> TranscriptSearchResult

Search explicitly selected transcript narratives within hard bounds.

This optional capability fails closed for custom stores until they opt in. Implementations must enforce the supplied session set inside the query and must never interpret an empty or omitted set as global access.

create_recall_receipt (method)

async SessionStore.create_recall_receipt(receipt: RecallReceipt) -> RecallReceipt

Idempotently persist one immutable, session-scoped recall receipt.

load_recall_receipt (method)

async SessionStore.load_recall_receipt(session_id: str, receipt_id: str) -> RecallReceipt | None

Load one exact receipt without widening its session scope.

list_recall_receipts (method)

async SessionStore.list_recall_receipts(query: RecallEvidenceQuery) -> RecallReceiptPage

List a deterministic, byte-bounded page of scoped recall receipts.

create_context_exposure (method)

async SessionStore.create_context_exposure(exposure: ContextExposure, item_exposures: tuple[RecallItemExposure, ...]=()) -> ContextExposure

Atomically persist one planned composition and its exact recalled items.

load_context_exposure (method)

async SessionStore.load_context_exposure(session_id: str, exposure_id: str) -> ContextExposure | None

Load one exact provider-attempt lifecycle within its session scope.

load_recall_item_exposures (method)

async SessionStore.load_recall_item_exposures(session_id: str, exposure_id: str) -> tuple[RecallItemExposure, ...]

Load the complete bounded item set atomically linked to one exposure.

list_context_exposures (method)

async SessionStore.list_context_exposures(query: RecallEvidenceQuery) -> ContextExposurePage

List a deterministic, byte-bounded page of scoped exposure lifecycles.

transition_context_exposure (method)

async SessionStore.transition_context_exposure(session_id: str, exposure_id: str, request: ContextExposureTransitionRequest) -> ContextExposure

Apply one idempotent lifecycle transition behind an exact CAS fence.

checkpoint (method)

async SessionStore.checkpoint(session_id: str, state: dict[str, Any]) -> None

Persist a checkpoint for resume/replay.

transform_checkpoint (method)

async SessionStore.transform_checkpoint(session_id: str, checkpoint_transform: CheckpointTransform) -> None

Atomically transform the latest checkpoint for a session.

The transform runs while the store owns its session/checkpoint write boundary and must be synchronous and thread-safe because a store may execute it on a worker thread. Returning None leaves the checkpoint unchanged.

transform_checkpoint_with_store_time (method)

async SessionStore.transform_checkpoint_with_store_time(session_id: str, checkpoint_transform: StoreTimeCheckpointTransform) -> None

Transform a checkpoint with time sampled by the store transaction.

The same sampled value must govern any lease comparison made by the transform and the resulting checkpoint/activity timestamps.

load_checkpoint (method)

async SessionStore.load_checkpoint(session_id: str) -> dict[str, Any] | None

Load the latest checkpoint for a session.

load_interruption_cascade_marker (method)

async SessionStore.load_interruption_cascade_marker(session_id: str, *, checkpoint_root_guard: CheckpointRootFieldGuard | None=None) -> dict[str, Any] | None

Load a bounded structural projection of the durable cascade marker.

A supplied root guard runs on bounded opaque evidence in the same read snapshot before the marker is interpreted.

Release source · Runtime guarantees · Capability guides