cayu.storage.PostgresTaskStore

Part of cayu.storage.postgres. Read the Python API overview and install the required extra.

from cayu.storage import PostgresTaskStore

Postgres-backed task store for durable multi-tenant work items.

Task-admission wakeups use one dedicated LISTEN connection outside the connection pool. A store that owns its DSN listens on that DSN, or on task_admission_listener_conninfo when given (for example a direct server address behind a transaction-pooling proxy). A store built on a caller-owned pool listens only when task_admission_listener_conninfo is supplied; otherwise workers fall back to bounded claim polling.

Decorators: @model_store_surface('tasks').

Bases: PostgresVerifiedWorkMixin, _PostgresStoreBase, TaskStore.

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.DURABLE

publish_work_contract (method)

async PostgresTaskStore.publish_work_contract(contract: WorkContract) -> WorkContract

Publish one immutable version or replay its exact canonical content.

load_work_contract (method)

async PostgresTaskStore.load_work_contract(reference: WorkContractRef) -> WorkContract | None

Load the exact contract named by reference or reject a fingerprint conflict.

load_active_work_contract_task_for_session (method)

async PostgresTaskStore.load_active_work_contract_task_for_session(session_id: str) -> Task | None

Load a contracted task whose binding retains authority over a session.

A terminal task does not implicitly release pending session work into the ordinary runtime. Only work-attempt lifecycle settlement retires a binding (retired_contract_binding on its receipt): when the task completes through an accepted decision or is cancelled by group cancellation. Until then the durable session binding remains authoritative and callers must start a new ordinary session.

admit_ordinary_session_execution (method)

async PostgresTaskStore.admit_ordinary_session_execution(session_id: str) -> None

Atomically admit a session to the ordinary, non-verifier runtime.

Supporting stores must reject admission while a contracted task binding retains authority over the session and must durably prevent later contract attachment to an admitted session. Task terminalization alone is not a release; only the lifecycle settlement that retires the contract binding is. Repeated admission of the same session is idempotent.

hold_claimed_work_contract_task (method)

async PostgresTaskStore.hold_claimed_work_contract_task(task_id: str, *, worker_id: str, lease_expires_at: datetime | None=None, contract: WorkContractRef) -> Task

Claim-fence an unsupported contracted task into operator attention.

prepare_work_attempt_admission (method)

async PostgresTaskStore.prepare_work_attempt_admission(request: WorkAttemptAdmissionPrepare) -> WorkAttemptAdmission

Reserve one exact initial or continuation attempt before session mutation.

activate_work_attempt_admission (method)

async PostgresTaskStore.activate_work_attempt_admission(request: WorkAttemptAdmissionActivate) -> WorkAttemptAdmission

Publish a prepared attempt after exact session admission evidence exists.

load_work_attempt_admission (method)

async PostgresTaskStore.load_work_attempt_admission(admission_id: str) -> WorkAttemptAdmission | None

Load one durable admission intent or receipt by stable identity.

load_work_attempt_execution_claim (method)

async PostgresTaskStore.load_work_attempt_execution_claim(claim_id: str) -> WorkAttemptExecutionClaim | None

Load one immutable execution-claim generation by stable identity.

load_latest_work_attempt_admission (method)

async PostgresTaskStore.load_latest_work_attempt_admission(task_id: str) -> WorkAttemptAdmission | None

Read the admission with no successor, including pre-dispatch preparation.

This is discovery, not acquisition of execution authority. The caller must still use exact admission/claim operations before dispatch.

load_work_attempt_lifecycle_receipt (method)

async PostgresTaskStore.load_work_attempt_lifecycle_receipt(admission_id: str) -> WorkAttemptLifecycleReceipt | None

Read the original final outcome, including after acknowledgement loss.

enter_work_attempt_execution (method)

async PostgresTaskStore.enter_work_attempt_execution(request: WorkAttemptExecutionEntryRequest) -> WorkAttemptExecutionEntryResult

Elect one dispatch; an existing entry permits reconciliation only.

record_work_attempt_execution_stop (method)

async PostgresTaskStore.record_work_attempt_execution_stop(request: WorkAttemptExecutionStopRequest) -> WorkAttemptAdmission

Record immutable stop intent; exact replay never authorizes dispatch.

load_work_attempt_preparation_hold_receipt (method)

async PostgresTaskStore.load_work_attempt_preparation_hold_receipt(hold_id: str) -> WorkAttemptPreparationHoldReceipt | None

Read an advisory original result; exact replay must compare the request.

hold_work_attempt_preparation (method)

async PostgresTaskStore.hold_work_attempt_preparation(request: WorkAttemptPreparationHold) -> WorkAttemptPreparationHoldReceipt

Atomically park an exact unattached claim and publish its failure receipt.

list_unsettled_work_attempt_admissions (method)

async PostgresTaskStore.list_unsettled_work_attempt_admissions(*, task_filter: TaskAggregateFilter | None=None, limit: int=100, after: str | None=None) -> list[WorkAttemptAdmission]

Discover latest unfinished admissions in admission-ID keyset order.

Includes final decision application awaiting lifecycle retirement, even when the task is already completed. Results never acquire authority.

settle_work_attempt_lifecycle (method)

async PostgresTaskStore.settle_work_attempt_lifecycle(request: WorkAttemptLifecycleSettlement) -> WorkAttemptLifecycleReceipt

Atomically publish or exactly replay final task and release evidence.

renew_work_attempt_execution_claim (method)

async PostgresTaskStore.renew_work_attempt_execution_claim(request: WorkAttemptExecutionClaimRequest) -> WorkAttemptAdmission

Renew an exact live preparing, active, or recovering claim in place.

claim_work_attempt_recovery (method)

async PostgresTaskStore.claim_work_attempt_recovery(request: WorkAttemptExecutionClaimRequest) -> WorkAttemptAdmission

Claim recovery after the prior execution generation has expired.

activate_work_attempt_recovery (method)

async PostgresTaskStore.activate_work_attempt_recovery(request: WorkAttemptRecoveryActivate) -> WorkAttemptAdmission

Activate a replacement only after positive session-settlement evidence.

begin_work_attempt (method)

async PostgresTaskStore.begin_work_attempt(request: WorkAttemptCreate) -> WorkAttempt

Create or replay one bounded execution attempt under a task’s frozen contract.

load_work_attempt (method)

async PostgresTaskStore.load_work_attempt(attempt_id: str) -> WorkAttempt | None

Load one work attempt by stable identity.

submit_completion_proposal (method)

async PostgresTaskStore.submit_completion_proposal(request: CompletionProposalCreate) -> CompletionProposal

Persist a worker proposal without granting completion authority.

submit_admitted_completion_proposal (method)

async PostgresTaskStore.submit_admitted_completion_proposal(request: AdmittedCompletionProposalRequest) -> CompletionProposal

Publish one proposal under the exact current execution claim.

load_completion_proposal (method)

async PostgresTaskStore.load_completion_proposal(proposal_id: str) -> CompletionProposal | None

Load one completion proposal by stable identity.

load_completion_proposal_for_attempt (method)

async PostgresTaskStore.load_completion_proposal_for_attempt(attempt_id: str) -> CompletionProposal | None

Load the unique published proposal for an attempt, without guessing its ID.

prepare_completion_verifier_profile (method)

async PostgresTaskStore.prepare_completion_verifier_profile(request: CompletionVerifierProfilePreparationRequest) -> CompletionVerifierProfileRecord

Insert or exactly replay immutable verifier-profile authority.

Implementations must atomically bind an adoption idempotency key to at most one proposal within its task. A later proposal cannot reuse that task-scoped transition identity.

load_completion_verifier_profile (method)

async PostgresTaskStore.load_completion_verifier_profile(proposal_id: str) -> CompletionVerifierProfileRecord | None

Load the immutable verifier profile prepared for one proposal.

load_prior_completion_verifier_profile (method)

async PostgresTaskStore.load_prior_completion_verifier_profile(proposal_id: str) -> CompletionVerifierProfileRecord | None

Load the immediately preceding task attempt’s verifier profile.

claim_completion_verification (method)

async PostgresTaskStore.claim_completion_verification(request: CompletionVerificationClaimRequest) -> CompletionVerificationClaim

Claim bounded exclusive authority to verify one undecided proposal.

load_completion_verification_claim (method)

async PostgresTaskStore.load_completion_verification_claim(proposal_id: str) -> CompletionVerificationClaim | None

Load the latest verification claim, including an expired claim for recovery.

renew_completion_verification_claim (method)

async PostgresTaskStore.renew_completion_verification_claim(request: CompletionVerificationClaimRequest) -> CompletionVerificationClaim

Extend one exact live verification claim without changing its owner.

record_completion_decision (method)

async PostgresTaskStore.record_completion_decision(request: CompletionDecisionCreate) -> CompletionDecision

Persist the one authoritative verifier decision for a proposal.

load_completion_decision (method)

async PostgresTaskStore.load_completion_decision(decision_id: str) -> CompletionDecision | None

Load one completion decision by stable identity.

load_completion_decision_for_proposal (method)

async PostgresTaskStore.load_completion_decision_for_proposal(proposal_id: str) -> CompletionDecision | None

Load the one authoritative decision published for a proposal.

apply_completion_decision (method)

async PostgresTaskStore.apply_completion_decision(request: CompletionDecisionApplicationRequest) -> Task

Apply or exactly replay a decision-bound task transition.

load_completion_decision_application_receipt (method)

async PostgresTaskStore.load_completion_decision_application_receipt(task_id: str, idempotency_key: str) -> CompletionDecisionApplicationReceipt | None

Load exact durable evidence for decision application reconciliation.

init (method)

PostgresTaskStore.__init__(conninfo: str | None=None, *, pool: AsyncConnectionPool | None=None, min_size: int=1, max_size: int=8, clock: Callable[[], datetime] | None=None, schema_mode: schema.SchemaMode=schema.SchemaMode.VALIDATE, read_only: bool=False, task_admission_listener_conninfo: str | None=None) -> None

ensure_schema (method)

async PostgresTaskStore.ensure_schema() -> None

Open the pool and reconcile the schema now (per schema_mode).

Normally reconciliation happens lazily on first use; the cayu storage CLI calls this to run a migrate (or validate) as an explicit step.

close (method)

async PostgresTaskStore.close() -> None

list_task_group_reconciliation_candidates (method)

async PostgresTaskStore.list_task_group_reconciliation_candidates(*, after_group_id: str | None=None, limit: int=100) -> list[str]

reconcile_task_group (method)

async PostgresTaskStore.reconcile_task_group(group_id: str) -> TaskGroupSnapshot

resolve_task_group_quiescence (method)

async PostgresTaskStore.resolve_task_group_quiescence(request: TaskGroupQuiescenceResolution) -> TaskGroupSnapshot

create_task_group (method)

async PostgresTaskStore.create_task_group(request: TaskGroupCreate) -> TaskGroupCreationReceipt

load_task_group (method)

async PostgresTaskStore.load_task_group(group_id: str) -> TaskGroupSnapshot | None

list_task_group_events (method)

async PostgresTaskStore.list_task_group_events(group_id: str, *, after_sequence: int=0, limit: int=100) -> list[TaskGroupEvent]

create_task_graph (method)

async PostgresTaskStore.create_task_graph(request: TaskGraphCreate) -> TaskGraphCreationReceipt

load_task_graph (method)

async PostgresTaskStore.load_task_graph(graph_id: str) -> TaskGraphSnapshot | None

list_task_graph_events (method)

async PostgresTaskStore.list_task_graph_events(graph_id: str, *, after_sequence: int=0, limit: int=100) -> list[TaskGraphEvent]

prepare_local_execution_attempt (method)

async PostgresTaskStore.prepare_local_execution_attempt(authority: LocalExecutionAttemptAuthority) -> LocalExecutionAttemptRecord

start_local_execution_attempt (method)

async PostgresTaskStore.start_local_execution_attempt(start: LocalExecutionAttemptStart) -> LocalExecutionAttemptRecord

settle_local_execution_attempt (method)

async PostgresTaskStore.settle_local_execution_attempt(settlement: LocalExecutionAttemptSettlement) -> LocalExecutionAttemptRecord

load_local_execution_attempt (method)

async PostgresTaskStore.load_local_execution_attempt(attempt_id: str) -> LocalExecutionAttemptRecord | None

list_unsettled_local_execution_attempts (method)

async PostgresTaskStore.list_unsettled_local_execution_attempts(*, limit: int=100, after: LocalExecutionAttemptListCursor | None=None) -> tuple[LocalExecutionAttemptRecord, ...]

claim_local_execution_attempt_recovery (method)

async PostgresTaskStore.claim_local_execution_attempt_recovery(claim: LocalExecutionAttemptRecoveryClaim) -> LocalExecutionAttemptRecord

create_task (method)

async PostgresTaskStore.create_task(request: TaskCreate) -> Task

reschedule_task (method)

async PostgresTaskStore.reschedule_task(request: TaskRescheduleRequest) -> TaskScheduleReceipt

cancel_scheduled_task (method)

async PostgresTaskStore.cancel_scheduled_task(request: TaskScheduleCancelRequest) -> TaskScheduleReceipt

list_task_schedule_events (method)

async PostgresTaskStore.list_task_schedule_events(task_id: str, *, after_sequence: int=0, limit: int=100) -> list[TaskScheduleEvent]

next_task_schedule_wakeup (method)

async PostgresTaskStore.next_task_schedule_wakeup(query: TaskQuery | None=None) -> TaskScheduleWakeup

create_running_task (method)

async PostgresTaskStore.create_running_task(request: TaskCreate, *, session_invocation: SessionInvocationBinding) -> Task

load_task (method)

async PostgresTaskStore.load_task(task_id: str, *, _access_bounds=None) -> Task | None

load_active_attached_task_worker (method)

async PostgresTaskStore.load_active_attached_task_worker(task_id: str, worker_id: str, *, session_id: str, session_instance_id: str) -> Task

load_direct_attached_task_resume (method)

async PostgresTaskStore.load_direct_attached_task_resume(task_id: str, *, session_id: str, session_instance_id: str) -> Task

load_invocation_snapshot (method)

async PostgresTaskStore.load_invocation_snapshot(task_id: str) -> TaskInvocationSnapshot | None

list_tasks (method)

async PostgresTaskStore.list_tasks(query: TaskQuery | None=None, *, _access_bounds=None) -> list[Task]

load_session_closure_claim (method)

async PostgresTaskStore.load_session_closure_claim(session_id: str) -> TaskSessionClosureClaim | None

claim_session_closure (method)

async PostgresTaskStore.claim_session_closure(claim: TaskSessionClosureClaim) -> TaskSessionClosureClaim

delete_session_tasks (method)

async PostgresTaskStore.delete_session_tasks(session_id: str, *, task_ids: tuple[str, ...], policy: Any) -> None

Delete a bounded, quiescent session task graph transactionally.

aggregate_operational_snapshot (method)

async PostgresTaskStore.aggregate_operational_snapshot(filters: TaskAggregateFilter | None=None) -> TaskOperationalSnapshot

query_task_topology (method)

async PostgresTaskStore.query_task_topology(query: TaskTopologyQuery) -> TaskTopologyStoreResult

start_task (method)

async PostgresTaskStore.start_task(task_id: str, *, session_id: str | None=None, session_invocation: SessionInvocationBinding | None=None) -> Task

attach_task (method)

async PostgresTaskStore.attach_task(task_id: str, *, session_id: str, session_invocation: SessionInvocationBinding, worker_id: str, lease_expires_at: datetime | None=None) -> Task

complete_task (method)

async PostgresTaskStore.complete_task(task_id: str, result: dict[str, Any], *, worker_id: str | None=None, lease_expires_at: datetime | None=None, handoff_id: str | None=None) -> Task

fail_task (method)

async PostgresTaskStore.fail_task(task_id: str, error: dict[str, Any], *, worker_id: str | None=None, lease_expires_at: datetime | None=None, handoff_id: str | None=None) -> Task

terminalize_task (method)

async PostgresTaskStore.terminalize_task(request: TaskTerminalizationRequest) -> Task

load_task_terminalization_receipt (method)

async PostgresTaskStore.load_task_terminalization_receipt(task_id: str, idempotency_key: str) -> TaskTerminalizationReceipt | None

recover_attached_task_failure (method)

async PostgresTaskStore.recover_attached_task_failure(request: TaskTerminalizationRequest, *, session_id: str, session_instance_id: str) -> Task

release_interrupted_task_worker (method)

async PostgresTaskStore.release_interrupted_task_worker(request: TaskInterruptedHandoffRequest) -> TaskInterruptedHandoffReceipt

recover_interrupted_task_worker (method)

async PostgresTaskStore.recover_interrupted_task_worker(request: TaskInterruptedHandoffRequest) -> TaskInterruptedHandoffReceipt

load_interrupted_task_handoff_receipt (method)

async PostgresTaskStore.load_interrupted_task_handoff_receipt(task_id: str, handoff_id: str) -> TaskInterruptedHandoffReceipt | None

list_expired_interrupted_task_handoff_candidates (method)

async PostgresTaskStore.list_expired_interrupted_task_handoff_candidates(*, after: tuple[datetime, str] | None=None, limit: int=100) -> list[Task]

load_expired_interrupted_task_handoff_candidate (method)

async PostgresTaskStore.load_expired_interrupted_task_handoff_candidate(task_id: str) -> Task | None

claim_interrupted_task_continuation (method)

async PostgresTaskStore.claim_interrupted_task_continuation(worker_id: str, query: TaskQuery | None=None, *, handoff_id: str, task_id: str | None=None, lease_seconds: int=300, after: tuple[datetime, str] | None=None, scan_limit: int=_TASK_INTERRUPTED_HANDOFF_RECOVERY_MAX_PAGE_SIZE) -> InterruptedTaskContinuationClaimPage

reconcile_task_cancellation (method)

async PostgresTaskStore.reconcile_task_cancellation(request: TaskCancellationReconciliationRequest) -> TaskCancellationReconciliationResult

settle_task_retry_attempt (method)

async PostgresTaskStore.settle_task_retry_attempt(request: TaskRetrySettlementRequest) -> TaskRetrySettlementResult

load_task_retry_settlement (method)

async PostgresTaskStore.load_task_retry_settlement(task_id: str, idempotency_key: str) -> TaskRetrySettlementResult | None

reconcile_task_retry_cancellation (method)

async PostgresTaskStore.reconcile_task_retry_cancellation(request: TaskRetryCancellationReconciliationRequest) -> TaskRetrySettlementResult

enforce_task_retry_deadline (method)

async PostgresTaskStore.enforce_task_retry_deadline(task_id: str, worker_id: str, *, lease_expires_at: datetime, token_count: int=0, estimated_cost: Decimal=Decimal(0)) -> TaskRetrySettlementResult | None

task_retry_deadline_elapsed (method)

async PostgresTaskStore.task_retry_deadline_elapsed(task_id: str, worker_id: str, *, lease_expires_at: datetime) -> bool

cancel_task (method)

async PostgresTaskStore.cancel_task(task_id: str, error: dict[str, Any] | None=None) -> Task

request_claimed_task_cancellation (method)

async PostgresTaskStore.request_claimed_task_cancellation(task_id: str, worker_id: str, lease_expires_at: datetime, error: dict[str, Any] | None=None) -> Task

mark_claimed_task_execution_started (method)

async PostgresTaskStore.mark_claimed_task_execution_started(task_id: str, worker_id: str, lease_expires_at: datetime) -> Task

pause_task (method)

async PostgresTaskStore.pause_task(task_id: str, *, reason: str | None=None, payload: dict[str, Any] | None=None) -> Task

block_task (method)

async PostgresTaskStore.block_task(task_id: str, *, reason: str | None=None, payload: dict[str, Any] | None=None) -> Task

mark_task_needs_attention (method)

async PostgresTaskStore.mark_task_needs_attention(task_id: str, *, reason: str | None=None, payload: dict[str, Any] | None=None) -> Task

resume_task (method)

async PostgresTaskStore.resume_task(task_id: str) -> Task

claim_task (method)

async PostgresTaskStore.claim_task(worker_id: str, query: TaskQuery | None=None, *, lease_seconds: int=300) -> Task | None

heartbeat (method)

async PostgresTaskStore.heartbeat(task_id: str, worker_id: str, *, lease_expires_at: datetime, handoff_id: str | None=None, extend_seconds: int=300) -> Task

release_task (method)

async PostgresTaskStore.release_task(task_id: str, worker_id: str, *, lease_expires_at: datetime) -> Task

release_attached_task_worker (method)

async PostgresTaskStore.release_attached_task_worker(task_id: str, worker_id: str, *, lease_expires_at: datetime) -> Task

reclaim_expired (method)

async PostgresTaskStore.reclaim_expired(*, query: TaskQuery | None=None, max_reclaims: int=100) -> list[Task]

Release source · Runtime guarantees · Capability guides