Module adcp.decisioning.pg

PostgreSQL-backed implementations for the decisioning module.

Ships durable backends behind the [pg] optional extra so the base adcp.decisioning import path stays free of SQL dependencies for adopters who only need the in-memory primitives.

Available when adcp[pg] is installed:

• :class:PgBuyerAgentRegistry — durable Tier 2 commercial-identity layer for v3 sellers. The framework calls the registry on every request to gate dispatch on the seller's commercial relationship with the buyer agent (allowlist + onboarding state + billing capabilities).
• :class:PgTaskRegistry — durable :class:~adcp.decisioning.TaskRegistry for HITL task state. Survives process restarts and is safe for multi-worker deployments sharing a single Postgres database. Drop-in replacement for :class:~adcp.decisioning.InMemoryTaskRegistry that satisfies the production-mode durability gate. (PgTaskRegistry is the pre-4.4 name and remains as a deprecated alias through 4.4.x.)
• :class:PgTaskWebhookOutbox — atomic terminal task-webhook publication coupled to PgTaskRegistry, with lease-based workers and 1–7 day retry retention.

The schema DDL ships alongside the Python code (e.g. adcp/decisioning/pg/buyer_agent_registry.sql, adcp/decisioning/pg/decisioning_tasks.sql) so adopters can run it through whatever migration tool they use (Alembic, Flyway, psql).

Sub-modules

adcp.decisioning.pg.buyer_agent_registry

PostgreSQL-backed :class:~adcp.decisioning.BuyerAgentRegistry …

adcp.decisioning.pg.lazy

Resolve-once wrappers for caller-owned PostgreSQL decisioning stores …

adcp.decisioning.pg.proposal_store

PostgreSQL-backed :class:~adcp.decisioning.ProposalStore implementation …

adcp.decisioning.pg.task_registry

PostgreSQL-backed :class:~adcp.decisioning.TaskRegistry implementation …

adcp.decisioning.pg.task_webhook_outbox

Crash-durable PostgreSQL outbox for terminal protocol-task webhooks …

Classes

class LazyProposalStore (factory: LazyProposalStoreFactory)

Expand source code

class LazyProposalStore(_LazyStore[PgProposalStore]):
    """Deferred PgProposalStore; durable before and after resolution."""

    is_durable: ClassVar[bool] = True

    def __init__(self, factory: LazyProposalStoreFactory) -> None:
        super().__init__(factory, PgProposalStore)

    def _validate(self, store: PgProposalStore) -> None:
        if store.is_durable is not True:
            raise ValueError("LazyProposalStore requires a durable PostgreSQL store")

    @classmethod
    def migration_sql(cls, table_name: str = DEFAULT_TABLE_NAME) -> dict[str, str]:
        """Return the concrete store's migration SQL without opening infrastructure."""
        return _migration_sql(table_name)

    async def create_schema(self) -> None:
        await (await self.resolve()).create_schema()

    async def put_draft(
        self,
        *,
        proposal_id: str,
        account_id: str,
        recipes: Mapping[str, Recipe],
        proposal_payload: Mapping[str, Any],
    ) -> None:
        await (await self.resolve()).put_draft(
            proposal_id=proposal_id,
            account_id=account_id,
            recipes=recipes,
            proposal_payload=proposal_payload,
        )

    async def get(self, proposal_id: str, *, expected_account_id: str) -> ProposalRecord | None:
        return await (await self.resolve()).get(
            proposal_id, expected_account_id=expected_account_id
        )

    async def commit(
        self,
        proposal_id: str,
        *,
        expires_at: datetime,
        proposal_payload: Mapping[str, Any],
        expected_account_id: str,
    ) -> None:
        await (await self.resolve()).commit(
            proposal_id,
            expires_at=expires_at,
            proposal_payload=proposal_payload,
            expected_account_id=expected_account_id,
        )

    async def try_reserve_consumption(
        self, proposal_id: str, *, expected_account_id: str
    ) -> ProposalRecord:
        return await (await self.resolve()).try_reserve_consumption(
            proposal_id, expected_account_id=expected_account_id
        )

    async def finalize_consumption(
        self, proposal_id: str, *, media_buy_id: str, expected_account_id: str
    ) -> None:
        await (await self.resolve()).finalize_consumption(
            proposal_id, media_buy_id=media_buy_id, expected_account_id=expected_account_id
        )

    async def release_consumption(self, proposal_id: str, *, expected_account_id: str) -> None:
        await (await self.resolve()).release_consumption(
            proposal_id, expected_account_id=expected_account_id
        )

    async def mark_consumed(
        self, proposal_id: str, *, media_buy_id: str, expected_account_id: str
    ) -> None:
        await (await self.resolve()).mark_consumed(
            proposal_id, media_buy_id=media_buy_id, expected_account_id=expected_account_id
        )

    async def discard(self, proposal_id: str, *, expected_account_id: str) -> None:
        await (await self.resolve()).discard(proposal_id, expected_account_id=expected_account_id)

    async def get_by_media_buy_id(
        self, media_buy_id: str, *, expected_account_id: str
    ) -> ProposalRecord | None:
        return await (await self.resolve()).get_by_media_buy_id(
            media_buy_id, expected_account_id=expected_account_id
        )

Deferred PgProposalStore; durable before and after resolution.

Ancestors

• adcp.decisioning.pg.lazy._LazyStore
• typing.Generic

Class variables

var is_durable : ClassVar[bool]

Static methods

def migration_sql(table_name: str = 'adcp_proposal_drafts') ‑> dict[str, str]

Return the concrete store's migration SQL without opening infrastructure.

Methods

async def commit(self,
proposal_id: str,
*,
expires_at: datetime,
proposal_payload: Mapping[str, Any],
expected_account_id: str) ‑> None

Expand source code

async def commit(
    self,
    proposal_id: str,
    *,
    expires_at: datetime,
    proposal_payload: Mapping[str, Any],
    expected_account_id: str,
) -> None:
    await (await self.resolve()).commit(
        proposal_id,
        expires_at=expires_at,
        proposal_payload=proposal_payload,
        expected_account_id=expected_account_id,
    )

async def create_schema(self) ‑> None

Expand source code

async def create_schema(self) -> None:
    await (await self.resolve()).create_schema()

async def discard(self, proposal_id: str, *, expected_account_id: str) ‑> None

Expand source code

async def discard(self, proposal_id: str, *, expected_account_id: str) -> None:
    await (await self.resolve()).discard(proposal_id, expected_account_id=expected_account_id)

async def finalize_consumption(self, proposal_id: str, *, media_buy_id: str, expected_account_id: str) ‑> None

Expand source code

async def finalize_consumption(
    self, proposal_id: str, *, media_buy_id: str, expected_account_id: str
) -> None:
    await (await self.resolve()).finalize_consumption(
        proposal_id, media_buy_id=media_buy_id, expected_account_id=expected_account_id
    )

async def get(self, proposal_id: str, *, expected_account_id: str) ‑> ProposalRecord | None

Expand source code

async def get(self, proposal_id: str, *, expected_account_id: str) -> ProposalRecord | None:
    return await (await self.resolve()).get(
        proposal_id, expected_account_id=expected_account_id
    )

async def get_by_media_buy_id(self, media_buy_id: str, *, expected_account_id: str) ‑> ProposalRecord | None

Expand source code

async def get_by_media_buy_id(
    self, media_buy_id: str, *, expected_account_id: str
) -> ProposalRecord | None:
    return await (await self.resolve()).get_by_media_buy_id(
        media_buy_id, expected_account_id=expected_account_id
    )

async def mark_consumed(self, proposal_id: str, *, media_buy_id: str, expected_account_id: str) ‑> None

Expand source code

async def mark_consumed(
    self, proposal_id: str, *, media_buy_id: str, expected_account_id: str
) -> None:
    await (await self.resolve()).mark_consumed(
        proposal_id, media_buy_id=media_buy_id, expected_account_id=expected_account_id
    )

async def put_draft(self,
*,
proposal_id: str,
account_id: str,
recipes: Mapping[str, Recipe],
proposal_payload: Mapping[str, Any]) ‑> None

Expand source code

async def put_draft(
    self,
    *,
    proposal_id: str,
    account_id: str,
    recipes: Mapping[str, Recipe],
    proposal_payload: Mapping[str, Any],
) -> None:
    await (await self.resolve()).put_draft(
        proposal_id=proposal_id,
        account_id=account_id,
        recipes=recipes,
        proposal_payload=proposal_payload,
    )

async def release_consumption(self, proposal_id: str, *, expected_account_id: str) ‑> None

Expand source code

async def release_consumption(self, proposal_id: str, *, expected_account_id: str) -> None:
    await (await self.resolve()).release_consumption(
        proposal_id, expected_account_id=expected_account_id
    )

async def try_reserve_consumption(self, proposal_id: str, *, expected_account_id: str) ‑> ProposalRecord

Expand source code

async def try_reserve_consumption(
    self, proposal_id: str, *, expected_account_id: str
) -> ProposalRecord:
    return await (await self.resolve()).try_reserve_consumption(
        proposal_id, expected_account_id=expected_account_id
    )

class LazyTaskRegistry (factory: LazyTaskRegistryFactory)

Expand source code

class LazyTaskRegistry(_LazyStore[PgTaskRegistry], _TaskLifecycleObservers):
    """Deferred PgTaskRegistry, including its listing and metrics contracts.

    A factory can return a registry, or an original (registry, outbox) pair. Build
    them together using one caller-owned pool. Pair and signing-scope validation
    are retained. Register observers before first use; they receive the first
    submitted transition too. No arbitrary delegate attributes are forwarded.

    For a server advertising SDK task-webhook signing, await resolve() before
    constructing the server: its synchronous boot validator needs the actual
    outbox/sender/horizon configuration. Polling-only servers can resolve on the
    first task operation.
    """

    is_durable: ClassVar[bool] = True

    def __init__(self, factory: LazyTaskRegistryFactory) -> None:
        self._init_lifecycle_observers()

        async def resolve_registry() -> PgTaskRegistry:
            result = factory()
            stores = await result if isinstance(result, Awaitable) else result
            if isinstance(stores, tuple):
                registry, outbox = stores
                if not isinstance(registry, PgTaskRegistry) or not isinstance(
                    outbox, PgTaskWebhookOutbox
                ):
                    raise TypeError("Task factory must return a concrete registry/outbox pair")
                if registry.task_webhook_outbox is not outbox or registry._pool is not outbox._pool:
                    raise ValueError(
                        "Task registry/outbox pair must share one pool and registration"
                    )
                return registry
            return stores

        super().__init__(resolve_registry, PgTaskRegistry)

    def _validate(self, store: PgTaskRegistry) -> None:
        if store.is_durable is not True:
            raise ValueError("LazyTaskRegistry requires a durable PostgreSQL registry")
        if not callable(store.list):
            raise TypeError("LazyTaskRegistry requires PostgreSQL listing support")
        outbox = store.task_webhook_outbox
        if outbox is not None and outbox._pool is not store._pool:
            raise ValueError("Task registry and outbox must share one pool")
        store.add_lifecycle_observer(self._relay_transition)

    def _relay_transition(
        self,
        event: TaskTransition,
        *,
        task_id: str,
        account_id: str,
        task_type: str,
        created_at: float,
        updated_at: float,
    ) -> None:
        self._notify_lifecycle_observers(
            event,
            {
                "task_id": task_id,
                "account_id": account_id,
                "task_type": task_type,
                "created_at": created_at,
                "updated_at": updated_at,
            },
        )

    @property
    def task_webhook_outbox(self) -> PgTaskWebhookOutbox | None:
        """Original coupled outbox, available after resolution; no eager opening."""
        store = self.resolved
        return store.task_webhook_outbox if store is not None else None

    @property
    def atomic_task_webhook_outbox(self) -> bool:
        store = self.resolved
        return store.atomic_task_webhook_outbox if store is not None else False

    async def create_schema(self) -> None:
        await (await self.resolve()).create_schema()

    async def issue(
        self,
        *,
        account_id: str,
        task_type: str,
        request_context: dict[str, Any] | None = None,
        webhook_url: str | None = None,
        webhook_operation_id: str | None = None,
        webhook_token: str | None = None,
        webhook_authentication: TaskWebhookAuthentication | None = None,
        webhook_signing_scope_id: str | None = None,
        **_extra: Any,
    ) -> str:
        return await (await self.resolve()).issue(
            account_id=account_id,
            task_type=task_type,
            request_context=request_context,
            webhook_url=webhook_url,
            webhook_operation_id=webhook_operation_id,
            webhook_token=webhook_token,
            webhook_authentication=webhook_authentication,
            webhook_signing_scope_id=webhook_signing_scope_id,
            **_extra,
        )

    async def resolve_webhook_signing_scope(self, context: RequestContext[Any]) -> str | None:
        return await (await self.resolve()).resolve_webhook_signing_scope(context)

    async def update_progress(self, task_id: str, progress: dict[str, Any]) -> None:
        await (await self.resolve()).update_progress(task_id, progress)

    async def complete(self, task_id: str, result: dict[str, Any]) -> None:
        await (await self.resolve()).complete(task_id, result)

    async def fail(self, task_id: str, error: dict[str, Any]) -> None:
        await (await self.resolve()).fail(task_id, error)

    async def get(
        self, task_id: str, *, expected_account_id: str | None = None
    ) -> dict[str, Any] | None:
        return await (await self.resolve()).get(task_id, expected_account_id=expected_account_id)

    async def list(
        self,
        *,
        account_id: str,
        filters: dict[str, Any] | None = None,
        sort: dict[str, Any] | None = None,
        pagination: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        return await (await self.resolve()).list(
            account_id=account_id, filters=filters, sort=sort, pagination=pagination
        )

    async def discard(self, task_id: str) -> None:
        await (await self.resolve()).discard(task_id)

Deferred PgTaskRegistry, including its listing and metrics contracts.

A factory can return a registry, or an original (registry, outbox) pair. Build them together using one caller-owned pool. Pair and signing-scope validation are retained. Register observers before first use; they receive the first submitted transition too. No arbitrary delegate attributes are forwarded.

For a server advertising SDK task-webhook signing, await resolve() before constructing the server: its synchronous boot validator needs the actual outbox/sender/horizon configuration. Polling-only servers can resolve on the first task operation.

Ancestors

• adcp.decisioning.pg.lazy._LazyStore
• typing.Generic
• adcp.decisioning.task_registry._TaskLifecycleObservers

Class variables

var is_durable : ClassVar[bool]

Instance variables

prop atomic_task_webhook_outbox : bool

Expand source code

@property
def atomic_task_webhook_outbox(self) -> bool:
    store = self.resolved
    return store.atomic_task_webhook_outbox if store is not None else False

prop task_webhook_outbox : PgTaskWebhookOutbox | None

Expand source code

@property
def task_webhook_outbox(self) -> PgTaskWebhookOutbox | None:
    """Original coupled outbox, available after resolution; no eager opening."""
    store = self.resolved
    return store.task_webhook_outbox if store is not None else None

Original coupled outbox, available after resolution; no eager opening.

Methods

async def complete(self, task_id: str, result: dict[str, Any]) ‑> None

Expand source code

async def complete(self, task_id: str, result: dict[str, Any]) -> None:
    await (await self.resolve()).complete(task_id, result)

async def create_schema(self) ‑> None

Expand source code

async def create_schema(self) -> None:
    await (await self.resolve()).create_schema()

async def discard(self, task_id: str) ‑> None

Expand source code

async def discard(self, task_id: str) -> None:
    await (await self.resolve()).discard(task_id)

async def fail(self, task_id: str, error: dict[str, Any]) ‑> None

Expand source code

async def fail(self, task_id: str, error: dict[str, Any]) -> None:
    await (await self.resolve()).fail(task_id, error)

async def get(self, task_id: str, *, expected_account_id: str | None = None) ‑> dict[str, typing.Any] | None

Expand source code

async def get(
    self, task_id: str, *, expected_account_id: str | None = None
) -> dict[str, Any] | None:
    return await (await self.resolve()).get(task_id, expected_account_id=expected_account_id)

async def issue(self,
*,
account_id: str,
task_type: str,
request_context: dict[str, Any] | None = None,
webhook_url: str | None = None,
webhook_operation_id: str | None = None,
webhook_token: str | None = None,
webhook_authentication: TaskWebhookAuthentication | None = None,
webhook_signing_scope_id: str | None = None,
**_extra: Any) ‑> str

Expand source code

async def issue(
    self,
    *,
    account_id: str,
    task_type: str,
    request_context: dict[str, Any] | None = None,
    webhook_url: str | None = None,
    webhook_operation_id: str | None = None,
    webhook_token: str | None = None,
    webhook_authentication: TaskWebhookAuthentication | None = None,
    webhook_signing_scope_id: str | None = None,
    **_extra: Any,
) -> str:
    return await (await self.resolve()).issue(
        account_id=account_id,
        task_type=task_type,
        request_context=request_context,
        webhook_url=webhook_url,
        webhook_operation_id=webhook_operation_id,
        webhook_token=webhook_token,
        webhook_authentication=webhook_authentication,
        webhook_signing_scope_id=webhook_signing_scope_id,
        **_extra,
    )

async def list(self,
*,
account_id: str,
filters: dict[str, Any] | None = None,
sort: dict[str, Any] | None = None,
pagination: dict[str, Any] | None = None) ‑> dict[str, typing.Any]

Expand source code

async def list(
    self,
    *,
    account_id: str,
    filters: dict[str, Any] | None = None,
    sort: dict[str, Any] | None = None,
    pagination: dict[str, Any] | None = None,
) -> dict[str, Any]:
    return await (await self.resolve()).list(
        account_id=account_id, filters=filters, sort=sort, pagination=pagination
    )

async def resolve_webhook_signing_scope(self, context: RequestContext[Any]) ‑> str | None

Expand source code

async def resolve_webhook_signing_scope(self, context: RequestContext[Any]) -> str | None:
    return await (await self.resolve()).resolve_webhook_signing_scope(context)

async def update_progress(self, task_id: str, progress: dict[str, Any]) ‑> None

Expand source code

async def update_progress(self, task_id: str, progress: dict[str, Any]) -> None:
    await (await self.resolve()).update_progress(task_id, progress)

class LazyTaskWebhookOutbox (factory: LazyTaskWebhookOutboxFactory)

Expand source code

class LazyTaskWebhookOutbox(_LazyStore[PgTaskWebhookOutbox]):
    """Deferred concrete outbox. Synchronous crypto helpers require resolve().

    Use from_registry() to share a registry factory's original outbox. Pass the
    original objects, constructed inside that factory, to PgTaskRegistry; do not
    substitute this worker facade into the concrete constructor's pool checks.
    """

    delivery_state_is_durable: ClassVar[bool] = True
    supports_atomic_task_outbox: ClassVar[bool] = True

    def __init__(self, factory: LazyTaskWebhookOutboxFactory) -> None:
        super().__init__(factory, PgTaskWebhookOutbox)

    def _validate(self, store: PgTaskWebhookOutbox) -> None:
        if (
            store.delivery_state_is_durable is not True
            or store.supports_atomic_task_outbox is not True
        ):
            raise ValueError("LazyTaskWebhookOutbox requires a durable atomic PostgreSQL outbox")

    @classmethod
    def from_registry(cls, registry: LazyTaskRegistry) -> LazyTaskWebhookOutbox:
        """Create a worker facade resolving the same original registry/outbox pair."""

        async def resolve_outbox() -> PgTaskWebhookOutbox:
            store = await registry.resolve()
            if store.task_webhook_outbox is None:
                raise ValueError("The registry factory did not configure a task webhook outbox")
            return store.task_webhook_outbox

        return cls(resolve_outbox)

    @property
    def delivery_retry_horizon_seconds(self) -> int:
        return self._require_resolved().delivery_retry_horizon_seconds

    @property
    def legacy_hmac_fallback(self) -> bool:
        return self._require_resolved().legacy_hmac_fallback

    async def create_schema(self) -> None:
        await (await self.resolve()).create_schema()

    async def enqueue_terminal(
        self,
        conn: Any,
        *,
        task_id: str,
        account_id: str,
        task_type: str,
        status: str,
        result: dict[str, Any],
        url: str,
        operation_id: str,
        token: str | None,
        authentication: TaskWebhookAuthentication | None = None,
        signing_scope_id: str | None = None,
    ) -> int:
        return await (await self.resolve()).enqueue_terminal(
            conn,
            task_id=task_id,
            account_id=account_id,
            task_type=task_type,
            status=status,
            result=result,
            url=url,
            operation_id=operation_id,
            token=token,
            authentication=authentication,
            signing_scope_id=signing_scope_id,
        )

    def validate_registration(
        self, url: str, authentication: TaskWebhookAuthentication | None = None
    ) -> None:
        (self._require_resolved()).validate_registration(url, authentication)

    def protect_registration(
        self,
        *,
        account_id: str,
        task_id: str,
        task_type: str,
        url: str,
        operation_id: str,
        token: str | None,
        authentication: TaskWebhookAuthentication | None = None,
        signing_scope_id: str | None = None,
    ) -> tuple[bytes, bytes]:
        return (self._require_resolved()).protect_registration(
            account_id=account_id,
            task_id=task_id,
            task_type=task_type,
            url=url,
            operation_id=operation_id,
            token=token,
            authentication=authentication,
            signing_scope_id=signing_scope_id,
        )

    def open_registration(
        self,
        *,
        account_id: str,
        task_id: str,
        task_type: str,
        encrypted_registration: bytes,
        nonce: bytes,
    ) -> tuple[str, str, str | None]:
        return (self._require_resolved()).open_registration(
            account_id=account_id,
            task_id=task_id,
            task_type=task_type,
            encrypted_registration=encrypted_registration,
            nonce=nonce,
        )

    async def run_worker(
        self, *, poll_interval: float = 1.0, purge_interval: float = 300.0
    ) -> None:
        await (await self.resolve()).run_worker(
            poll_interval=poll_interval, purge_interval=purge_interval
        )

    async def process_one(self) -> bool:
        return await (await self.resolve()).process_one()

    async def purge_expired(self) -> None:
        await (await self.resolve()).purge_expired()

Deferred concrete outbox. Synchronous crypto helpers require resolve().

Use from_registry() to share a registry factory's original outbox. Pass the original objects, constructed inside that factory, to PgTaskRegistry; do not substitute this worker facade into the concrete constructor's pool checks.

Ancestors

• adcp.decisioning.pg.lazy._LazyStore
• typing.Generic

Class variables

var delivery_state_is_durable : ClassVar[bool]

var supports_atomic_task_outbox : ClassVar[bool]

Static methods

def from_registry(registry: LazyTaskRegistry) ‑> LazyTaskWebhookOutbox

Create a worker facade resolving the same original registry/outbox pair.

Instance variables

prop delivery_retry_horizon_seconds : int

Expand source code

@property
def delivery_retry_horizon_seconds(self) -> int:
    return self._require_resolved().delivery_retry_horizon_seconds

prop legacy_hmac_fallback : bool

Expand source code

@property
def legacy_hmac_fallback(self) -> bool:
    return self._require_resolved().legacy_hmac_fallback

Methods

async def create_schema(self) ‑> None

Expand source code

async def create_schema(self) -> None:
    await (await self.resolve()).create_schema()

async def enqueue_terminal(self,
conn: Any,
*,
task_id: str,
account_id: str,
task_type: str,
status: str,
result: dict[str, Any],
url: str,
operation_id: str,
token: str | None,
authentication: TaskWebhookAuthentication | None = None,
signing_scope_id: str | None = None) ‑> int

Expand source code

async def enqueue_terminal(
    self,
    conn: Any,
    *,
    task_id: str,
    account_id: str,
    task_type: str,
    status: str,
    result: dict[str, Any],
    url: str,
    operation_id: str,
    token: str | None,
    authentication: TaskWebhookAuthentication | None = None,
    signing_scope_id: str | None = None,
) -> int:
    return await (await self.resolve()).enqueue_terminal(
        conn,
        task_id=task_id,
        account_id=account_id,
        task_type=task_type,
        status=status,
        result=result,
        url=url,
        operation_id=operation_id,
        token=token,
        authentication=authentication,
        signing_scope_id=signing_scope_id,
    )

def open_registration(self,
*,
account_id: str,
task_id: str,
task_type: str,
encrypted_registration: bytes,
nonce: bytes) ‑> tuple[str, str, str | None]

Expand source code

def open_registration(
    self,
    *,
    account_id: str,
    task_id: str,
    task_type: str,
    encrypted_registration: bytes,
    nonce: bytes,
) -> tuple[str, str, str | None]:
    return (self._require_resolved()).open_registration(
        account_id=account_id,
        task_id=task_id,
        task_type=task_type,
        encrypted_registration=encrypted_registration,
        nonce=nonce,
    )

async def process_one(self) ‑> bool

Expand source code

async def process_one(self) -> bool:
    return await (await self.resolve()).process_one()

def protect_registration(self,
*,
account_id: str,
task_id: str,
task_type: str,
url: str,
operation_id: str,
token: str | None,
authentication: TaskWebhookAuthentication | None = None,
signing_scope_id: str | None = None) ‑> tuple[bytes, bytes]

Expand source code

def protect_registration(
    self,
    *,
    account_id: str,
    task_id: str,
    task_type: str,
    url: str,
    operation_id: str,
    token: str | None,
    authentication: TaskWebhookAuthentication | None = None,
    signing_scope_id: str | None = None,
) -> tuple[bytes, bytes]:
    return (self._require_resolved()).protect_registration(
        account_id=account_id,
        task_id=task_id,
        task_type=task_type,
        url=url,
        operation_id=operation_id,
        token=token,
        authentication=authentication,
        signing_scope_id=signing_scope_id,
    )

async def purge_expired(self) ‑> None

Expand source code

async def purge_expired(self) -> None:
    await (await self.resolve()).purge_expired()

async def run_worker(self, *, poll_interval: float = 1.0, purge_interval: float = 300.0) ‑> None

Expand source code

async def run_worker(
    self, *, poll_interval: float = 1.0, purge_interval: float = 300.0
) -> None:
    await (await self.resolve()).run_worker(
        poll_interval=poll_interval, purge_interval=purge_interval
    )

def validate_registration(self, url: str, authentication: TaskWebhookAuthentication | None = None) ‑> None

Expand source code

def validate_registration(
    self, url: str, authentication: TaskWebhookAuthentication | None = None
) -> None:
    (self._require_resolved()).validate_registration(url, authentication)

class PgBuyerAgentRegistry (*, pool: ConnectionPool, table_name: str = 'adcp_buyer_agents')

Expand source code

class PgBuyerAgentRegistry:
    """PostgreSQL-backed :class:`~adcp.decisioning.BuyerAgentRegistry`.

    Parameters
    ----------
    pool:
        A :class:`psycopg_pool.ConnectionPool` owned by the caller.
        Each operation acquires a short-lived connection, runs a
        single statement, and returns the connection.
    table_name:
        Override the default ``adcp_buyer_agents`` table when two
        tenants share a database and need separate registries. Must
        be an ASCII-byte-clean identifier — the constructor validates.

    Concurrency
    -----------

    Safe to share across threads and processes. The
    :meth:`resolve_by_agent_url` / :meth:`resolve_by_credential`
    methods bridge the async Protocol to the sync pool via
    :func:`asyncio.to_thread`; concurrent dispatches each get their
    own thread + connection.
    """

    def __init__(
        self,
        *,
        pool: ConnectionPool,
        table_name: str = DEFAULT_TABLE_NAME,
    ) -> None:
        if not PG_AVAILABLE:
            raise ImportError(_INSTALL_HINT)
        if not _is_safe_identifier(table_name):
            raise ValueError(
                f"table_name must match [a-z_][a-z0-9_]* (ASCII only), got {table_name!r}"
            )
        self._pool = pool
        self._table = table_name
        self._mutation_observers: list[MutationObserver] = []
        self._mutation_observers_lock = threading.Lock()

        # Pre-format queries so the hot path doesn't f-string per call.
        # All identifier substitutions are validated at __init__; row
        # values flow through psycopg's parameter binding.
        cols = (
            "agent_url, display_name, status, billing_capabilities, "
            "api_key_id, default_terms, allowed_brands, ext"
        )
        self._sql_select_by_agent_url = (
            f"SELECT {cols} FROM {self._table} "  # noqa: S608 — table name validated
            f"WHERE agent_url = %s"
        )
        self._sql_select_by_api_key_id = (
            f"SELECT {cols} FROM {self._table} "  # noqa: S608
            f"WHERE api_key_id = %s LIMIT 2"
        )
        self._sql_upsert = (
            f"INSERT INTO {self._table} ("  # noqa: S608
            f"  agent_url, display_name, status, billing_capabilities, "
            f"  api_key_id, default_terms, allowed_brands, ext, updated_at"
            f") VALUES (%s, %s, %s, %s::jsonb, %s, %s::jsonb, %s::jsonb, "
            f"          %s::jsonb, now()) "
            f"ON CONFLICT (agent_url) DO UPDATE SET "
            f"  display_name = EXCLUDED.display_name, "
            f"  status = EXCLUDED.status, "
            f"  billing_capabilities = EXCLUDED.billing_capabilities, "
            f"  api_key_id = EXCLUDED.api_key_id, "
            f"  default_terms = EXCLUDED.default_terms, "
            f"  allowed_brands = EXCLUDED.allowed_brands, "
            f"  ext = EXCLUDED.ext, "
            f"  updated_at = now()"
        )
        self._sql_set_status = (
            f"UPDATE {self._table} "  # noqa: S608
            f"SET status = %s, updated_at = now() "
            f"WHERE agent_url = %s"
        )
        self._sql_delete = f"DELETE FROM {self._table} WHERE agent_url = %s"  # noqa: S608

    # ----- schema bootstrap ---------------------------------------------

    def create_schema(self) -> None:
        """Create the registry table + indexes for this store's
        ``table_name``. Idempotent once credential identifiers are unique.

        Existing deployments with duplicated ``api_key_id`` values fail
        before indexes change, with an actionable rotation/removal message.
        The legacy non-unique credential index is dropped only after the
        replacement unique index exists.

        The equivalent raw DDL ships at
        :file:`src/adcp/decisioning/pg/buyer_agent_registry.sql` for
        adopters using a migration tool (Alembic, Flyway, psql) —
        that file uses the canonical ``adcp_buyer_agents`` name.
        """
        table = self._table  # already validated at __init__
        table_ddl = (
            f"CREATE TABLE IF NOT EXISTS {table} ("  # noqa: S608 — validated
            f'    agent_url             TEXT        COLLATE "C" PRIMARY KEY,'
            f"    display_name          TEXT        NOT NULL,"
            f"    status                TEXT        NOT NULL DEFAULT 'active'"
            f"        CHECK (status IN ('active', 'suspended', 'blocked')),"
            f"    billing_capabilities  JSONB       NOT NULL DEFAULT '[\"operator\"]'::jsonb,"
            f'    api_key_id            TEXT        COLLATE "C",'
            f"    default_terms         JSONB,"
            f"    allowed_brands        JSONB,"
            f"    ext                   JSONB       NOT NULL DEFAULT '{{}}'::jsonb,"
            f"    created_at            TIMESTAMPTZ NOT NULL DEFAULT now(),"
            f"    updated_at            TIMESTAMPTZ NOT NULL DEFAULT now()"
            f");"
        )
        duplicate_preflight = (  # noqa: S608 — validated table name
            f"SELECT api_key_id, COUNT(*) FROM {table} "
            f"WHERE api_key_id IS NOT NULL GROUP BY api_key_id "
            f"HAVING COUNT(*) > 1 LIMIT 1"
        )
        unique_index_ddl = (  # noqa: S608
            f"CREATE UNIQUE INDEX IF NOT EXISTS {table}_api_key_id_uidx "
            f"ON {table} (api_key_id) WHERE api_key_id IS NOT NULL"
        )
        drop_legacy_index_ddl = f"DROP INDEX IF EXISTS {table}_api_key_id_idx"  # noqa: S608
        status_index_ddl = (  # noqa: S608
            f"CREATE INDEX IF NOT EXISTS {table}_status_idx "
            f"ON {table} (status) WHERE status <> 'active'"
        )
        with self._pool.connection() as conn, conn.cursor() as cur:
            cur.execute(table_ddl)
            cur.execute(duplicate_preflight)
            duplicate = cur.fetchall()
            if duplicate:
                raise RuntimeError(
                    f"Cannot enforce {table}.api_key_id uniqueness: duplicated credential "
                    "identifiers exist. Rotate or remove duplicate bearer credentials, then "
                    "rerun create_schema()."
                )
            cur.execute(unique_index_ddl)
            cur.execute(drop_legacy_index_ddl)
            cur.execute(status_index_ddl)

    # ----- BuyerAgentRegistry Protocol --------------------------------

    async def resolve_by_agent_url(self, agent_url: str) -> BuyerAgent | None:
        """Resolve a verified ``agent_url`` against the allowlist.

        The framework has already validated the RFC 9421 signature
        before this point — the registry's only job is the commercial
        lookup. Returns ``None`` when the agent isn't recognized;
        the framework converts that to ``PERMISSION_DENIED`` (with
        ``details`` omitted so the unrecognized-agent path is
        wire-indistinguishable from recognized-but-denied).
        """
        return await asyncio.to_thread(self._sync_lookup_by_agent_url, agent_url)

    async def resolve_by_credential(
        self,
        credential: Credential,
    ) -> BuyerAgent | None:
        """Resolve a bearer / API-key / OAuth credential.

        Looks up against the ``api_key_id`` column. For
        :class:`OAuthCredential`, the ``client_id`` is used as the
        lookup key — adopters with separate OAuth-client tables fork
        this registry impl and split the column. The MVP shape
        treats both bearer and OAuth as the same column for the
        common case (one identifier per agent).
        """
        if isinstance(credential, ApiKeyCredential):
            key = credential.key_id
        elif isinstance(credential, OAuthCredential):
            key = credential.client_id
        else:  # defensive: future Credential variants the registry can't dispatch
            return None
        return await asyncio.to_thread(self._sync_lookup_by_api_key_id, key)

    # ----- admin CRUD --------------------------------------------------

    def upsert(self, agent: BuyerAgent, *, api_key_id: str | None = None) -> None:
        """Insert or update a :class:`BuyerAgent` row.

        ``api_key_id`` is separate from the :class:`BuyerAgent` shape
        because the framework's typed model doesn't carry the
        bearer-table FK. Adopters running bearer auth populate this;
        signing-only adopters leave it ``None``.
        """
        if agent.status not in _VALID_STATUSES:
            raise ValueError(
                f"BuyerAgent.status must be one of {sorted(_VALID_STATUSES)!r}, "
                f"got {agent.status!r}"
            )
        terms_json = (
            json.dumps(_terms_to_dict(agent.default_account_terms))
            if agent.default_account_terms is not None
            else None
        )
        allowed_brands_json = (
            json.dumps(sorted(agent.allowed_brands)) if agent.allowed_brands is not None else None
        )
        params = (
            agent.agent_url,
            agent.display_name,
            agent.status,
            json.dumps(sorted(agent.billing_capabilities)),
            api_key_id,
            terms_json,
            allowed_brands_json,
            json.dumps(dict(agent.ext)),
        )
        with self._pool.connection() as conn, conn.cursor() as cur:
            cur.execute(self._sql_upsert, params)
        self._notify_mutation("upsert", agent.agent_url)

    def set_status(self, agent_url: str, status: BuyerAgentStatus) -> None:
        """Update an agent's lifecycle status. Use to suspend / block
        / reactivate without rewriting the full row."""
        if status not in _VALID_STATUSES:
            raise ValueError(f"status must be one of {sorted(_VALID_STATUSES)!r}, got {status!r}")
        with self._pool.connection() as conn, conn.cursor() as cur:
            cur.execute(self._sql_set_status, (status, agent_url))
        self._notify_mutation("set_status", agent_url)

    def delete(self, agent_url: str) -> None:
        """Remove an agent from the registry.

        Hard delete — no row history. Adopters needing audit retention
        keep the row and set ``status='blocked'`` instead.
        """
        with self._pool.connection() as conn, conn.cursor() as cur:
            cur.execute(self._sql_delete, (agent_url,))
        self._notify_mutation("delete", agent_url)

    # ----- mutation observability -------------------------------------

    def add_mutation_observer(self, observer: MutationObserver) -> None:
        """Register a callback fired after every successful mutation.

        Observers receive ``(operation, agent_url)`` where
        ``operation`` is one of ``"upsert"`` / ``"set_status"`` /
        ``"delete"``. They run synchronously on the calling thread
        AFTER the DB commit, so a failed commit does not invoke
        observers. Exceptions raised by an observer are logged and
        swallowed — they never block the mutation from succeeding or
        prevent later observers from running.

        Typical use is wiring a cache-invalidation hook so admin
        mutations propagate to read-side caches without manual
        :meth:`CachingBuyerAgentRegistry.invalidate` calls. See
        :meth:`with_caching` for the bundled pre-wired path.

        Observer registration is thread-safe. Mutations notify a
        snapshot of the current observer list; observers added or
        removed while a notification is in flight apply to the next
        mutation.
        """
        with self._mutation_observers_lock:
            self._mutation_observers.append(observer)

    def remove_mutation_observer(self, observer: MutationObserver) -> bool:
        """Unregister a mutation observer.

        Returns ``True`` when ``observer`` was registered and removed,
        ``False`` when it was not present. If the same callback was
        registered multiple times, one registration is removed per call.

        Removal is thread-safe. Mutations notify a snapshot of the
        observer list, so removing an observer while a notification is
        already in flight only affects subsequent mutations.
        """
        with self._mutation_observers_lock:
            try:
                self._mutation_observers.remove(observer)
            except ValueError:
                return False
        return True

    def with_caching(
        self,
        **cache_kwargs: Any,
    ) -> CachingBuyerAgentRegistry:
        """Return a :class:`CachingBuyerAgentRegistry` wrapping this
        registry, pre-wired so mutations through this instance
        automatically invalidate the cache.

        Forwards ``**cache_kwargs`` to
        :class:`CachingBuyerAgentRegistry` (``ttl_seconds``,
        ``max_entries``, ``hit_callback``, ``audit_sink``,
        ``sink_timeout_seconds``, ``time_source``).

        Example::

            pg = PgBuyerAgentRegistry(pool=pool)
            registry = pg.with_caching(ttl_seconds=60, audit_sink=sink)
            serve(buyer_agent_registry=registry, ...)

            # Admin mutations go through `pg` and invalidate the cache:
            pg.upsert(BuyerAgent(agent_url=..., status="suspended"))
            # Next resolve() through `registry` hits DB, sees suspended.

        Adopters with external admin paths (a separate process
        writing to the same DB) still need :meth:`CachingBuyerAgentRegistry.invalidate`
        or :meth:`clear_sync` — the observer hook fires on
        mutations through *this* :class:`PgBuyerAgentRegistry`
        instance only.
        """
        from adcp.decisioning.registry_cache import CachingBuyerAgentRegistry

        cache = CachingBuyerAgentRegistry(self, **cache_kwargs)
        self.add_mutation_observer(lambda _op, _agent_url: cache.clear_sync())
        return cache

    def with_full_stack(
        self,
        *,
        ttl_seconds: float = 60.0,
        max_entries: int = 4096,
        hit_callback: Callable[[str], None] | None = None,
        rps_per_tenant: float = 100.0,
        burst: float | None = None,
        audit_sink: AuditSink | None = None,
        sink_timeout_seconds: float = 5.0,
        time_source: Callable[[], float] = time.monotonic,
    ) -> CachingBuyerAgentRegistry:
        """Return the canonical production registry wrapper stack.

        Builds and returns ``Caching(RateLimited(Auditing(self)))``:

        * cache is outermost so cached hits skip rate-limit accounting
          and DB work;
        * rate limiting applies only to cache misses that need inner
          resolution;
        * auditing wraps the SQL-backed store so DB ``resolved`` /
          ``miss`` outcomes are recorded.

        ``audit_sink`` and ``sink_timeout_seconds`` are threaded through
        all three layers, so cache hits/misses, rate-limit rejects, and
        terminal DB outcomes can all land in the same audit trail.
        ``time_source`` is shared by the cache and rate limiter for
        deterministic tests.

        Mutations through this :class:`PgBuyerAgentRegistry` instance
        clear the returned cache via the same observer wiring as
        :meth:`with_caching`. Adopters needing a different layer order
        should compose :class:`CachingBuyerAgentRegistry`,
        :class:`RateLimitedBuyerAgentRegistry`, and
        :class:`AuditingBuyerAgentRegistry` manually.
        """
        from adcp.decisioning.registry_cache import (
            AuditingBuyerAgentRegistry,
            CachingBuyerAgentRegistry,
            RateLimitedBuyerAgentRegistry,
        )

        audited = AuditingBuyerAgentRegistry(
            self,
            audit_sink=audit_sink,
            sink_timeout_seconds=sink_timeout_seconds,
        )
        rate_limited = RateLimitedBuyerAgentRegistry(
            audited,
            rps_per_tenant=rps_per_tenant,
            burst=burst,
            audit_sink=audit_sink,
            sink_timeout_seconds=sink_timeout_seconds,
            time_source=time_source,
        )
        cache = CachingBuyerAgentRegistry(
            rate_limited,
            ttl_seconds=ttl_seconds,
            max_entries=max_entries,
            hit_callback=hit_callback,
            audit_sink=audit_sink,
            sink_timeout_seconds=sink_timeout_seconds,
            time_source=time_source,
        )
        self.add_mutation_observer(lambda _op, _agent_url: cache.clear_sync())
        return cache

    def _notify_mutation(self, op: str, agent_url: str) -> None:
        """Fire registered observers; log and swallow exceptions."""
        with self._mutation_observers_lock:
            observers = tuple(self._mutation_observers)
        for observer in observers:
            try:
                observer(op, agent_url)
            except Exception:  # noqa: BLE001 — observers must not break mutations
                logger.warning(
                    "[adcp.buyer_agent_registry] mutation observer raised for op=%s agent_url=%s",
                    op,
                    agent_url,
                    exc_info=True,
                )

    # ----- sync helpers (called via asyncio.to_thread) ----------------

    def _sync_lookup_by_agent_url(self, agent_url: str) -> BuyerAgent | None:
        with self._pool.connection() as conn, conn.cursor() as cur:
            cur.execute(self._sql_select_by_agent_url, (agent_url,))
            row = cur.fetchone()
            return _row_to_agent(row) if row else None

    def _sync_lookup_by_api_key_id(self, key: str) -> BuyerAgent | None:
        with self._pool.connection() as conn, conn.cursor() as cur:
            cur.execute(self._sql_select_by_api_key_id, (key,))
            rows = cur.fetchall()
            if len(rows) > 1:
                # Defense in depth for deployments that have not yet applied
                # the unique-index migration. Never select an arbitrary
                # commercial identity for an ambiguous credential.
                logger.error(
                    "PgBuyerAgentRegistry rejected an ambiguous credential mapping; "
                    "apply the unique api_key_id index migration"
                )
                return None
            return _row_to_agent(rows[0]) if rows else None

PostgreSQL-backed :class:~adcp.decisioning.BuyerAgentRegistry.

Parameters

pool: A :class:psycopg_pool.ConnectionPool owned by the caller. Each operation acquires a short-lived connection, runs a single statement, and returns the connection. table_name: Override the default adcp_buyer_agents table when two tenants share a database and need separate registries. Must be an ASCII-byte-clean identifier — the constructor validates.

Concurrency

Safe to share across threads and processes. The :meth:resolve_by_agent_url / :meth:resolve_by_credential methods bridge the async Protocol to the sync pool via :func:asyncio.to_thread; concurrent dispatches each get their own thread + connection.

Methods

def add_mutation_observer(self, observer: MutationObserver) ‑> None

Expand source code

def add_mutation_observer(self, observer: MutationObserver) -> None:
    """Register a callback fired after every successful mutation.

    Observers receive ``(operation, agent_url)`` where
    ``operation`` is one of ``"upsert"`` / ``"set_status"`` /
    ``"delete"``. They run synchronously on the calling thread
    AFTER the DB commit, so a failed commit does not invoke
    observers. Exceptions raised by an observer are logged and
    swallowed — they never block the mutation from succeeding or
    prevent later observers from running.

    Typical use is wiring a cache-invalidation hook so admin
    mutations propagate to read-side caches without manual
    :meth:`CachingBuyerAgentRegistry.invalidate` calls. See
    :meth:`with_caching` for the bundled pre-wired path.

    Observer registration is thread-safe. Mutations notify a
    snapshot of the current observer list; observers added or
    removed while a notification is in flight apply to the next
    mutation.
    """
    with self._mutation_observers_lock:
        self._mutation_observers.append(observer)

Register a callback fired after every successful mutation.

Observers receive (operation, agent_url) where operation is one of "upsert" / "set_status" / "delete". They run synchronously on the calling thread AFTER the DB commit, so a failed commit does not invoke observers. Exceptions raised by an observer are logged and swallowed — they never block the mutation from succeeding or prevent later observers from running.

Typical use is wiring a cache-invalidation hook so admin mutations propagate to read-side caches without manual :meth:CachingBuyerAgentRegistry.invalidate calls. See :meth:with_caching for the bundled pre-wired path.

Observer registration is thread-safe. Mutations notify a snapshot of the current observer list; observers added or removed while a notification is in flight apply to the next mutation.

def create_schema(self) ‑> None

Expand source code

def create_schema(self) -> None:
    """Create the registry table + indexes for this store's
    ``table_name``. Idempotent once credential identifiers are unique.

    Existing deployments with duplicated ``api_key_id`` values fail
    before indexes change, with an actionable rotation/removal message.
    The legacy non-unique credential index is dropped only after the
    replacement unique index exists.

    The equivalent raw DDL ships at
    :file:`src/adcp/decisioning/pg/buyer_agent_registry.sql` for
    adopters using a migration tool (Alembic, Flyway, psql) —
    that file uses the canonical ``adcp_buyer_agents`` name.
    """
    table = self._table  # already validated at __init__
    table_ddl = (
        f"CREATE TABLE IF NOT EXISTS {table} ("  # noqa: S608 — validated
        f'    agent_url             TEXT        COLLATE "C" PRIMARY KEY,'
        f"    display_name          TEXT        NOT NULL,"
        f"    status                TEXT        NOT NULL DEFAULT 'active'"
        f"        CHECK (status IN ('active', 'suspended', 'blocked')),"
        f"    billing_capabilities  JSONB       NOT NULL DEFAULT '[\"operator\"]'::jsonb,"
        f'    api_key_id            TEXT        COLLATE "C",'
        f"    default_terms         JSONB,"
        f"    allowed_brands        JSONB,"
        f"    ext                   JSONB       NOT NULL DEFAULT '{{}}'::jsonb,"
        f"    created_at            TIMESTAMPTZ NOT NULL DEFAULT now(),"
        f"    updated_at            TIMESTAMPTZ NOT NULL DEFAULT now()"
        f");"
    )
    duplicate_preflight = (  # noqa: S608 — validated table name
        f"SELECT api_key_id, COUNT(*) FROM {table} "
        f"WHERE api_key_id IS NOT NULL GROUP BY api_key_id "
        f"HAVING COUNT(*) > 1 LIMIT 1"
    )
    unique_index_ddl = (  # noqa: S608
        f"CREATE UNIQUE INDEX IF NOT EXISTS {table}_api_key_id_uidx "
        f"ON {table} (api_key_id) WHERE api_key_id IS NOT NULL"
    )
    drop_legacy_index_ddl = f"DROP INDEX IF EXISTS {table}_api_key_id_idx"  # noqa: S608
    status_index_ddl = (  # noqa: S608
        f"CREATE INDEX IF NOT EXISTS {table}_status_idx "
        f"ON {table} (status) WHERE status <> 'active'"
    )
    with self._pool.connection() as conn, conn.cursor() as cur:
        cur.execute(table_ddl)
        cur.execute(duplicate_preflight)
        duplicate = cur.fetchall()
        if duplicate:
            raise RuntimeError(
                f"Cannot enforce {table}.api_key_id uniqueness: duplicated credential "
                "identifiers exist. Rotate or remove duplicate bearer credentials, then "
                "rerun create_schema()."
            )
        cur.execute(unique_index_ddl)
        cur.execute(drop_legacy_index_ddl)
        cur.execute(status_index_ddl)

Create the registry table + indexes for this store's table_name. Idempotent once credential identifiers are unique.

Existing deployments with duplicated api_key_id values fail before indexes change, with an actionable rotation/removal message. The legacy non-unique credential index is dropped only after the replacement unique index exists.

The equivalent raw DDL ships at :file:src/adcp/decisioning/pg/buyer_agent_registry.sql for adopters using a migration tool (Alembic, Flyway, psql) — that file uses the canonical adcp_buyer_agents name.

def delete(self, agent_url: str) ‑> None

Expand source code

def delete(self, agent_url: str) -> None:
    """Remove an agent from the registry.

    Hard delete — no row history. Adopters needing audit retention
    keep the row and set ``status='blocked'`` instead.
    """
    with self._pool.connection() as conn, conn.cursor() as cur:
        cur.execute(self._sql_delete, (agent_url,))
    self._notify_mutation("delete", agent_url)

Remove an agent from the registry.

Hard delete — no row history. Adopters needing audit retention keep the row and set status='blocked' instead.

def remove_mutation_observer(self, observer: MutationObserver) ‑> bool

Expand source code

def remove_mutation_observer(self, observer: MutationObserver) -> bool:
    """Unregister a mutation observer.

    Returns ``True`` when ``observer`` was registered and removed,
    ``False`` when it was not present. If the same callback was
    registered multiple times, one registration is removed per call.

    Removal is thread-safe. Mutations notify a snapshot of the
    observer list, so removing an observer while a notification is
    already in flight only affects subsequent mutations.
    """
    with self._mutation_observers_lock:
        try:
            self._mutation_observers.remove(observer)
        except ValueError:
            return False
    return True

Unregister a mutation observer.

Returns True when observer was registered and removed, False when it was not present. If the same callback was registered multiple times, one registration is removed per call.

Removal is thread-safe. Mutations notify a snapshot of the observer list, so removing an observer while a notification is already in flight only affects subsequent mutations.

async def resolve_by_agent_url(self, agent_url: str) ‑> BuyerAgent | None

Expand source code

async def resolve_by_agent_url(self, agent_url: str) -> BuyerAgent | None:
    """Resolve a verified ``agent_url`` against the allowlist.

    The framework has already validated the RFC 9421 signature
    before this point — the registry's only job is the commercial
    lookup. Returns ``None`` when the agent isn't recognized;
    the framework converts that to ``PERMISSION_DENIED`` (with
    ``details`` omitted so the unrecognized-agent path is
    wire-indistinguishable from recognized-but-denied).
    """
    return await asyncio.to_thread(self._sync_lookup_by_agent_url, agent_url)

Resolve a verified agent_url against the allowlist.

The framework has already validated the RFC 9421 signature before this point — the registry's only job is the commercial lookup. Returns None when the agent isn't recognized; the framework converts that to PERMISSION_DENIED (with details omitted so the unrecognized-agent path is wire-indistinguishable from recognized-but-denied).

async def resolve_by_credential(self, credential: Credential) ‑> BuyerAgent | None

Expand source code

async def resolve_by_credential(
    self,
    credential: Credential,
) -> BuyerAgent | None:
    """Resolve a bearer / API-key / OAuth credential.

    Looks up against the ``api_key_id`` column. For
    :class:`OAuthCredential`, the ``client_id`` is used as the
    lookup key — adopters with separate OAuth-client tables fork
    this registry impl and split the column. The MVP shape
    treats both bearer and OAuth as the same column for the
    common case (one identifier per agent).
    """
    if isinstance(credential, ApiKeyCredential):
        key = credential.key_id
    elif isinstance(credential, OAuthCredential):
        key = credential.client_id
    else:  # defensive: future Credential variants the registry can't dispatch
        return None
    return await asyncio.to_thread(self._sync_lookup_by_api_key_id, key)

Resolve a bearer / API-key / OAuth credential.

Looks up against the api_key_id column. For :class:OAuthCredential, the client_id is used as the lookup key — adopters with separate OAuth-client tables fork this registry impl and split the column. The MVP shape treats both bearer and OAuth as the same column for the common case (one identifier per agent).

def set_status(self, agent_url: str, status: BuyerAgentStatus) ‑> None

Expand source code

def set_status(self, agent_url: str, status: BuyerAgentStatus) -> None:
    """Update an agent's lifecycle status. Use to suspend / block
    / reactivate without rewriting the full row."""
    if status not in _VALID_STATUSES:
        raise ValueError(f"status must be one of {sorted(_VALID_STATUSES)!r}, got {status!r}")
    with self._pool.connection() as conn, conn.cursor() as cur:
        cur.execute(self._sql_set_status, (status, agent_url))
    self._notify_mutation("set_status", agent_url)

Update an agent's lifecycle status. Use to suspend / block / reactivate without rewriting the full row.

def upsert(self, agent: BuyerAgent, *, api_key_id: str | None = None) ‑> None

Expand source code

def upsert(self, agent: BuyerAgent, *, api_key_id: str | None = None) -> None:
    """Insert or update a :class:`BuyerAgent` row.

    ``api_key_id`` is separate from the :class:`BuyerAgent` shape
    because the framework's typed model doesn't carry the
    bearer-table FK. Adopters running bearer auth populate this;
    signing-only adopters leave it ``None``.
    """
    if agent.status not in _VALID_STATUSES:
        raise ValueError(
            f"BuyerAgent.status must be one of {sorted(_VALID_STATUSES)!r}, "
            f"got {agent.status!r}"
        )
    terms_json = (
        json.dumps(_terms_to_dict(agent.default_account_terms))
        if agent.default_account_terms is not None
        else None
    )
    allowed_brands_json = (
        json.dumps(sorted(agent.allowed_brands)) if agent.allowed_brands is not None else None
    )
    params = (
        agent.agent_url,
        agent.display_name,
        agent.status,
        json.dumps(sorted(agent.billing_capabilities)),
        api_key_id,
        terms_json,
        allowed_brands_json,
        json.dumps(dict(agent.ext)),
    )
    with self._pool.connection() as conn, conn.cursor() as cur:
        cur.execute(self._sql_upsert, params)
    self._notify_mutation("upsert", agent.agent_url)

Insert or update a :class:BuyerAgent row.

api_key_id is separate from the :class:BuyerAgent shape because the framework's typed model doesn't carry the bearer-table FK. Adopters running bearer auth populate this; signing-only adopters leave it None.

def with_caching(self, **cache_kwargs: Any) ‑> CachingBuyerAgentRegistry

Expand source code

def with_caching(
    self,
    **cache_kwargs: Any,
) -> CachingBuyerAgentRegistry:
    """Return a :class:`CachingBuyerAgentRegistry` wrapping this
    registry, pre-wired so mutations through this instance
    automatically invalidate the cache.

    Forwards ``**cache_kwargs`` to
    :class:`CachingBuyerAgentRegistry` (``ttl_seconds``,
    ``max_entries``, ``hit_callback``, ``audit_sink``,
    ``sink_timeout_seconds``, ``time_source``).

    Example::

        pg = PgBuyerAgentRegistry(pool=pool)
        registry = pg.with_caching(ttl_seconds=60, audit_sink=sink)
        serve(buyer_agent_registry=registry, ...)

        # Admin mutations go through `pg` and invalidate the cache:
        pg.upsert(BuyerAgent(agent_url=..., status="suspended"))
        # Next resolve() through `registry` hits DB, sees suspended.

    Adopters with external admin paths (a separate process
    writing to the same DB) still need :meth:`CachingBuyerAgentRegistry.invalidate`
    or :meth:`clear_sync` — the observer hook fires on
    mutations through *this* :class:`PgBuyerAgentRegistry`
    instance only.
    """
    from adcp.decisioning.registry_cache import CachingBuyerAgentRegistry

    cache = CachingBuyerAgentRegistry(self, **cache_kwargs)
    self.add_mutation_observer(lambda _op, _agent_url: cache.clear_sync())
    return cache

Return a :class:CachingBuyerAgentRegistry wrapping this registry, pre-wired so mutations through this instance automatically invalidate the cache.

Forwards **cache_kwargs to :class:CachingBuyerAgentRegistry (ttl_seconds, max_entries, hit_callback, audit_sink, sink_timeout_seconds, time_source).

Example::

pg = PgBuyerAgentRegistry(pool=pool)
registry = pg.with_caching(ttl_seconds=60, audit_sink=sink)
serve(buyer_agent_registry=registry, ...)

# Admin mutations go through <code>pg</code> and invalidate the cache:
pg.upsert(BuyerAgent(agent_url=..., status="suspended"))
# Next resolve() through <code>registry</code> hits DB, sees suspended.

Adopters with external admin paths (a separate process writing to the same DB) still need :meth:CachingBuyerAgentRegistry.invalidate or :meth:clear_sync — the observer hook fires on mutations through this :class:PgBuyerAgentRegistry instance only.

def with_full_stack(self,
*,
ttl_seconds: float = 60.0,
max_entries: int = 4096,
hit_callback: Callable[[str], None] | None = None,
rps_per_tenant: float = 100.0,
burst: float | None = None,
audit_sink: AuditSink | None = None,
sink_timeout_seconds: float = 5.0,
time_source: Callable[[], float] = <built-in function monotonic>) ‑> CachingBuyerAgentRegistry

Expand source code

def with_full_stack(
    self,
    *,
    ttl_seconds: float = 60.0,
    max_entries: int = 4096,
    hit_callback: Callable[[str], None] | None = None,
    rps_per_tenant: float = 100.0,
    burst: float | None = None,
    audit_sink: AuditSink | None = None,
    sink_timeout_seconds: float = 5.0,
    time_source: Callable[[], float] = time.monotonic,
) -> CachingBuyerAgentRegistry:
    """Return the canonical production registry wrapper stack.

    Builds and returns ``Caching(RateLimited(Auditing(self)))``:

    * cache is outermost so cached hits skip rate-limit accounting
      and DB work;
    * rate limiting applies only to cache misses that need inner
      resolution;
    * auditing wraps the SQL-backed store so DB ``resolved`` /
      ``miss`` outcomes are recorded.

    ``audit_sink`` and ``sink_timeout_seconds`` are threaded through
    all three layers, so cache hits/misses, rate-limit rejects, and
    terminal DB outcomes can all land in the same audit trail.
    ``time_source`` is shared by the cache and rate limiter for
    deterministic tests.

    Mutations through this :class:`PgBuyerAgentRegistry` instance
    clear the returned cache via the same observer wiring as
    :meth:`with_caching`. Adopters needing a different layer order
    should compose :class:`CachingBuyerAgentRegistry`,
    :class:`RateLimitedBuyerAgentRegistry`, and
    :class:`AuditingBuyerAgentRegistry` manually.
    """
    from adcp.decisioning.registry_cache import (
        AuditingBuyerAgentRegistry,
        CachingBuyerAgentRegistry,
        RateLimitedBuyerAgentRegistry,
    )

    audited = AuditingBuyerAgentRegistry(
        self,
        audit_sink=audit_sink,
        sink_timeout_seconds=sink_timeout_seconds,
    )
    rate_limited = RateLimitedBuyerAgentRegistry(
        audited,
        rps_per_tenant=rps_per_tenant,
        burst=burst,
        audit_sink=audit_sink,
        sink_timeout_seconds=sink_timeout_seconds,
        time_source=time_source,
    )
    cache = CachingBuyerAgentRegistry(
        rate_limited,
        ttl_seconds=ttl_seconds,
        max_entries=max_entries,
        hit_callback=hit_callback,
        audit_sink=audit_sink,
        sink_timeout_seconds=sink_timeout_seconds,
        time_source=time_source,
    )
    self.add_mutation_observer(lambda _op, _agent_url: cache.clear_sync())
    return cache

Return the canonical production registry wrapper stack.

Builds and returns Caching(RateLimited(Auditing(self))):

• cache is outermost so cached hits skip rate-limit accounting and DB work;
• rate limiting applies only to cache misses that need inner resolution;
• auditing wraps the SQL-backed store so DB resolved / miss outcomes are recorded.

audit_sink and sink_timeout_seconds are threaded through all three layers, so cache hits/misses, rate-limit rejects, and terminal DB outcomes can all land in the same audit trail. time_source is shared by the cache and rate limiter for deterministic tests.

Mutations through this :class:PgBuyerAgentRegistry instance clear the returned cache via the same observer wiring as :meth:with_caching. Adopters needing a different layer order should compose :class:CachingBuyerAgentRegistry, :class:RateLimitedBuyerAgentRegistry, and :class:AuditingBuyerAgentRegistry manually.

class PgProposalStore (*,
pool: AsyncConnectionPool,
table_name: str = 'adcp_proposal_drafts',
recipe_decoder: Callable[[Mapping[str, Any]], Recipe] | None = None)

Expand source code

class PgProposalStore:
    """PostgreSQL-backed :class:`~adcp.decisioning.ProposalStore`.

    Durable counterpart to :class:`~adcp.decisioning.InMemoryProposalStore`.
    Set ``is_durable = True`` so production-mode gates accept it without
    requiring the dev-mode bypass.

    :param pool: ``psycopg_pool.AsyncConnectionPool`` owned by the
        caller. Each operation acquires a short-lived connection;
        :meth:`try_reserve_consumption` holds one for the duration of
        its CAS transaction.
    :param table_name: Override the default table name. Useful for
        adopters with one Postgres serving multiple AdCP instances or
        whose ``proposal_drafts`` table is already taken. Defaults to
        ``adcp_proposal_drafts``.
    :param recipe_decoder: Callable ``(payload: dict) -> Recipe`` used
        to rehydrate stored recipe payloads back to typed
        :class:`Recipe` instances. Adopters with subclasses
        (``GAMRecipe``, ``KevelRecipe``, etc.) MUST supply a decoder
        that branches on ``recipe_kind``. Defaults to
        :meth:`Recipe.model_validate` which only works for the base
        ``Recipe`` shape.

    :raises ImportError: when psycopg/psycopg-pool are not installed.
    :raises ValueError: when ``table_name`` is not a safe ASCII
        identifier (``[a-z_][a-z0-9_]{0,62}``).
    """

    is_durable: ClassVar[bool] = True

    def __init__(
        self,
        *,
        pool: AsyncConnectionPool,
        table_name: str = DEFAULT_TABLE_NAME,
        recipe_decoder: Callable[[Mapping[str, Any]], Recipe] | None = None,
    ) -> None:
        if not PG_AVAILABLE:
            raise ImportError(_INSTALL_HINT)
        if not _SAFE_IDENTIFIER_RE.fullmatch(table_name):
            raise ValueError(
                f"table_name must match [a-z_][a-z0-9_]{{0,62}} (ASCII only), "
                f"got {table_name!r}"
            )
        self._pool = pool
        self._table = table_name
        self._recipe_decoder = recipe_decoder or _default_recipe_decoder

        t = self._table
        # put_draft: insert a fresh DRAFT row, or rewrite an existing DRAFT
        # row in place. ON CONFLICT DO UPDATE is gated on state='draft' so
        # a buyer probing put_draft against a COMMITTED/CONSUMED record
        # falls through to a zero-row UPDATE — we then SELECT to surface
        # the INTERNAL_ERROR with the actual current state.
        self._sql_put_draft = (  # noqa: S608 — table name whitelisted
            f"INSERT INTO {t} "
            f"(account_id, proposal_id, state, recipes, proposal_payload, "
            f" recipe_schema_version, created_at, updated_at) "
            f"VALUES (%s, %s, 'draft', %s::jsonb, %s::jsonb, %s, now(), now()) "
            f"ON CONFLICT (account_id, proposal_id) DO UPDATE SET "
            f"  recipes          = EXCLUDED.recipes, "
            f"  proposal_payload = EXCLUDED.proposal_payload, "
            f"  recipe_schema_version = EXCLUDED.recipe_schema_version, "
            f"  updated_at       = now() "
            f"WHERE {t}.state = 'draft' "
            f"RETURNING xmax = 0 AS inserted"
        )
        self._sql_get_state = (  # noqa: S608
            f"SELECT state, expires_at, proposal_payload FROM {t} "
            f"WHERE account_id = %s AND proposal_id = %s"
        )
        # Tenant-scoped SELECT FOR UPDATE used by commit() to lock the
        # row before the state-machine check + UPDATE. Mirrors the
        # try_reserve_consumption pattern.
        self._sql_select_state_for_update = (  # noqa: S608
            f"SELECT state, expires_at, proposal_payload FROM {t} "
            f"WHERE account_id = %s AND proposal_id = %s FOR UPDATE"
        )
        self._sql_commit = (  # noqa: S608
            f"UPDATE {t} SET "
            f"  state            = 'committed', "
            f"  expires_at       = %s, "
            f"  proposal_payload = %s::jsonb, "
            f"  updated_at       = now() "
            f"WHERE account_id = %s AND proposal_id = %s AND state = 'draft' "
            f"RETURNING proposal_id"
        )
        # try_reserve_consumption uses SELECT ... FOR UPDATE inside a tx
        # so two parallel callers serialize on the row lock. The CAS
        # check (state='committed') happens after the lock is held; the
        # loser sees CONSUMING/CONSUMED and raises PROPOSAL_NOT_COMMITTED.
        self._sql_select_for_update = (  # noqa: S608
            f"SELECT state, recipes, proposal_payload, expires_at, "
            f"       media_buy_id, recipe_schema_version "
            f"FROM {t} WHERE account_id = %s AND proposal_id = %s FOR UPDATE"
        )
        self._sql_reserve = (  # noqa: S608
            f"UPDATE {t} SET state = 'consuming', updated_at = now() "
            f"WHERE account_id = %s AND proposal_id = %s AND state = 'committed'"
        )
        self._sql_finalize = (  # noqa: S608
            f"UPDATE {t} SET "
            f"  state        = 'consumed', "
            f"  media_buy_id = %s, "
            f"  updated_at   = now() "
            f"WHERE account_id = %s AND proposal_id = %s AND state = 'consuming' "
            f"RETURNING proposal_id"
        )
        self._sql_release = (  # noqa: S608
            f"UPDATE {t} SET state = 'committed', updated_at = now() "
            f"WHERE account_id = %s AND proposal_id = %s AND state = 'consuming' "
            f"RETURNING proposal_id"
        )
        self._sql_mark_consumed = (  # noqa: S608
            f"UPDATE {t} SET "
            f"  state        = 'consumed', "
            f"  media_buy_id = %s, "
            f"  updated_at   = now() "
            f"WHERE account_id = %s AND proposal_id = %s AND state = 'committed' "
            f"RETURNING proposal_id"
        )
        self._sql_discard = (  # noqa: S608
            f"DELETE FROM {t} WHERE account_id = %s AND proposal_id = %s"
        )
        self._sql_get_by_media_buy_id = (  # noqa: S608
            f"SELECT proposal_id, account_id, state, recipes, proposal_payload, "
            f"       expires_at, media_buy_id, recipe_schema_version "
            f"FROM {t} WHERE account_id = %s AND media_buy_id = %s"
        )

    # -- schema bootstrap -----------------------------------------------

    async def create_schema(self) -> None:
        """Create the proposal store table + supporting indexes.

        Honors the ``table_name`` kwarg the store was constructed with.
        Idempotent via ``CREATE TABLE IF NOT EXISTS`` — safe to call on
        every application boot. The equivalent raw DDL ships at
        :file:`adcp/decisioning/pg/proposal_store.sql` in the installed
        package for adopters using a migration tool.
        """
        t = self._table
        statements = [
            f"""CREATE TABLE IF NOT EXISTS {t} (
                account_id              TEXT        COLLATE "C" NOT NULL,
                proposal_id             TEXT        COLLATE "C" NOT NULL,
                state                   TEXT        NOT NULL
                    CHECK (state IN ('draft', 'committed', 'consuming', 'consumed')),
                recipes                 JSONB       NOT NULL DEFAULT '{{}}'::jsonb,
                proposal_payload        JSONB       NOT NULL,
                expires_at              TIMESTAMPTZ,
                media_buy_id            TEXT        COLLATE "C",
                recipe_schema_version   INTEGER     NOT NULL DEFAULT 1,
                created_at              TIMESTAMPTZ NOT NULL DEFAULT now(),
                updated_at              TIMESTAMPTZ NOT NULL DEFAULT now(),
                PRIMARY KEY (account_id, proposal_id)
            )""",
            f"""CREATE UNIQUE INDEX IF NOT EXISTS {t}_media_buy_idx
                ON {t} (account_id, media_buy_id)
                WHERE media_buy_id IS NOT NULL""",
            f"""CREATE INDEX IF NOT EXISTS {t}_expires_idx
                ON {t} (expires_at)
                WHERE expires_at IS NOT NULL""",
        ]
        async with self._pool.connection() as conn:
            for stmt in statements:
                await conn.execute(stmt)

    # -- ProposalStore Protocol -----------------------------------------

    async def put_draft(
        self,
        *,
        proposal_id: str,
        account_id: str,
        recipes: Mapping[str, Recipe],
        proposal_payload: Mapping[str, Any],
    ) -> None:
        recipes_json = _encode_recipes(recipes)
        payload_json = json.dumps(dict(proposal_payload))
        async with self._pool.connection() as conn:
            cur = await conn.execute(
                self._sql_put_draft,
                (account_id, proposal_id, recipes_json, payload_json, 1),
            )
            row = await cur.fetchone()
            if row is not None:
                # Either inserted fresh (xmax=0) or rewrote a DRAFT.
                return
            # ON CONFLICT DO UPDATE matched zero rows — the record is in
            # a non-DRAFT state. Re-fetch to surface the current state in
            # the error message.
            cur2 = await conn.execute(self._sql_get_state, (account_id, proposal_id))
            existing = await cur2.fetchone()
            if existing is None:
                # Race: row vanished between the failed INSERT and the
                # follow-up SELECT (concurrent discard from another
                # worker). Surface as INTERNAL_ERROR so the framework's
                # outer dispatcher can decide whether to retry; we don't
                # transparently retry here because put_draft is meant to
                # be one-shot per dispatch.
                raise AdcpError(
                    "INTERNAL_ERROR",
                    message=(
                        f"PgProposalStore.put_draft: proposal {proposal_id!r} "
                        "vanished between conflict and refetch. Concurrent "
                        "discard suspected."
                    ),
                    recovery="terminal",
                )
            state_str = existing[0]
            raise AdcpError(
                "INTERNAL_ERROR",
                message=(
                    f"Cannot put_draft on proposal {proposal_id!r} in "
                    f"state {state_str!r}; refine iterations are only "
                    "valid on draft proposals. Once committed or "
                    "consumed, a proposal_id is immutable."
                ),
                recovery="terminal",
            )

    async def get(
        self,
        proposal_id: str,
        *,
        expected_account_id: str,
    ) -> ProposalRecord | None:
        sql = (  # noqa: S608 — table name pre-validated at construction
            f"SELECT proposal_id, account_id, state, recipes, "
            f"proposal_payload, expires_at, media_buy_id, "
            f"recipe_schema_version FROM {self._table} "
            f"WHERE account_id = %s AND proposal_id = %s"
        )
        params: tuple[Any, ...] = (expected_account_id, proposal_id)
        async with self._pool.connection() as conn:
            cur = await conn.execute(sql, params)
            row = await cur.fetchone()
            if row is None:
                return None
            return self._row_to_record(row)

    async def commit(
        self,
        proposal_id: str,
        *,
        expires_at: datetime,
        proposal_payload: Mapping[str, Any],
        expected_account_id: str,
    ) -> None:
        payload_dict = dict(proposal_payload)
        payload_json = json.dumps(payload_dict)
        # Atomic: SELECT FOR UPDATE → state-machine check → UPDATE,
        # all inside one transaction. The SELECT predicate is keyed on
        # (account_id, proposal_id) so a cross-tenant probe collapses
        # to "not in store" without touching another tenant's row.
        # The row lock prevents a concurrent put_draft / commit from
        # racing the validate-then-update sequence.
        async with self._pool.connection() as conn:
            async with conn.transaction():
                cur = await conn.execute(
                    self._sql_select_state_for_update,
                    (expected_account_id, proposal_id),
                )
                existing = await cur.fetchone()
                if existing is None:
                    raise AdcpError(
                        "INTERNAL_ERROR",
                        message=(
                            f"Cannot commit proposal {proposal_id!r}: not "
                            "in store for the expected tenant. The "
                            "framework's finalize dispatch must put_draft "
                            "before commit."
                        ),
                        recovery="terminal",
                    )
                current_state, current_expires_at, current_payload = existing
                if current_state == "committed":
                    # Idempotent only when the second commit matches the
                    # first.
                    same_deadline = _ensure_utc(current_expires_at) == expires_at
                    cur_payload_dict = (
                        current_payload
                        if isinstance(current_payload, dict)
                        else json.loads(current_payload) if current_payload is not None else {}
                    )
                    same_payload = cur_payload_dict == payload_dict
                    if same_deadline and same_payload:
                        return
                    raise AdcpError(
                        "INTERNAL_ERROR",
                        message=(
                            f"Proposal {proposal_id!r} already committed "
                            "with a different expires_at or payload — "
                            "re-commit with different values is a developer "
                            "bug."
                        ),
                        recovery="terminal",
                    )
                if current_state != "draft":
                    raise AdcpError(
                        "INTERNAL_ERROR",
                        message=(
                            f"Cannot commit proposal {proposal_id!r} from "
                            f"state {current_state!r}; commit requires "
                            "DRAFT."
                        ),
                        recovery="terminal",
                    )
                update_cur = await conn.execute(
                    self._sql_commit,
                    (expires_at, payload_json, expected_account_id, proposal_id),
                )
                if await update_cur.fetchone() is None:
                    # Should not happen under the row lock, but fail loud
                    # if it does — silent zero-row UPDATE would let the
                    # caller believe the transition landed.
                    raise AdcpError(
                        "INTERNAL_ERROR",
                        message=(
                            f"PgProposalStore.commit: UPDATE returned zero "
                            f"rows for proposal {proposal_id!r} despite "
                            "passing the FOR UPDATE state check. Schema "
                            "drift suspected."
                        ),
                        recovery="terminal",
                    )

    async def try_reserve_consumption(
        self,
        proposal_id: str,
        *,
        expected_account_id: str,
    ) -> ProposalRecord:
        # Single-connection transaction so SELECT FOR UPDATE + UPDATE
        # serialize across parallel callers.
        async with self._pool.connection() as conn:
            async with conn.transaction():
                cur = await conn.execute(
                    self._sql_select_for_update,
                    (expected_account_id, proposal_id),
                )
                row = await cur.fetchone()
                if row is None:
                    raise AdcpError(
                        "PROPOSAL_NOT_FOUND",
                        message=(f"Proposal {proposal_id!r} not found."),
                        recovery="correctable",
                        field="proposal_id",
                    )
                (
                    state_str,
                    recipes_raw,
                    payload_raw,
                    expires_at_raw,
                    media_buy_id,
                    recipe_schema_version,
                ) = row
                if state_str != "committed":
                    raise AdcpError(
                        "PROPOSAL_NOT_COMMITTED",
                        message=(
                            f"Proposal {proposal_id!r} is in state "
                            f"{state_str!r}; create_media_buy requires a "
                            "committed proposal that hasn't been accepted "
                            "or reserved by another request."
                        ),
                        recovery="correctable",
                        field="proposal_id",
                    )
                await conn.execute(
                    self._sql_reserve,
                    (expected_account_id, proposal_id),
                )
                # Build the in-memory record reflecting the transition.
                return ProposalRecord(
                    proposal_id=proposal_id,
                    account_id=expected_account_id,
                    state=ProposalState.CONSUMING,
                    recipes=_decode_recipes(recipes_raw, self._recipe_decoder),
                    proposal_payload=_decode_payload(payload_raw),
                    expires_at=_ensure_utc(expires_at_raw),
                    media_buy_id=media_buy_id,
                    recipe_schema_version=int(recipe_schema_version or 1),
                )

    async def finalize_consumption(
        self,
        proposal_id: str,
        *,
        media_buy_id: str,
        expected_account_id: str,
    ) -> None:
        async with self._pool.connection() as conn:
            cur = await conn.execute(
                self._sql_finalize,
                (media_buy_id, expected_account_id, proposal_id),
            )
            if await cur.fetchone() is not None:
                return
            # Zero rows updated. Determine why.
            cur2 = await conn.execute(self._sql_get_state, (expected_account_id, proposal_id))
            existing = await cur2.fetchone()
            if existing is None:
                raise AdcpError(
                    "INTERNAL_ERROR",
                    message=(
                        f"finalize_consumption: proposal {proposal_id!r} "
                        "not found for the expected tenant."
                    ),
                    recovery="terminal",
                )
            state_str = existing[0]
            if state_str == "consumed":
                # Idempotent on re-call with the same media_buy_id.
                cur3 = await conn.execute(
                    f"SELECT media_buy_id FROM {self._table} "  # noqa: S608
                    f"WHERE account_id = %s AND proposal_id = %s",
                    (expected_account_id, proposal_id),
                )
                row = await cur3.fetchone()
                existing_media_buy_id = row[0] if row else None
                if existing_media_buy_id == media_buy_id:
                    return
                raise AdcpError(
                    "INTERNAL_ERROR",
                    message=(
                        f"Proposal {proposal_id!r} already consumed by "
                        f"media_buy_id={existing_media_buy_id!r}; cannot "
                        f"re-consume as {media_buy_id!r}."
                    ),
                    recovery="terminal",
                )
            raise AdcpError(
                "INTERNAL_ERROR",
                message=(
                    f"finalize_consumption requires CONSUMING; "
                    f"proposal {proposal_id!r} is in {state_str!r}. "
                    "Framework must call try_reserve_consumption first."
                ),
                recovery="terminal",
            )

    async def release_consumption(
        self,
        proposal_id: str,
        *,
        expected_account_id: str,
    ) -> None:
        async with self._pool.connection() as conn:
            cur = await conn.execute(
                self._sql_release,
                (expected_account_id, proposal_id),
            )
            if await cur.fetchone() is not None:
                return
            # Zero rows updated. Idempotent: no-op on unknown id /
            # cross-tenant probe, no-op on already-COMMITTED.
            cur2 = await conn.execute(self._sql_get_state, (expected_account_id, proposal_id))
            existing = await cur2.fetchone()
            if existing is None:
                # Unknown id or cross-tenant — idempotent no-op so the
                # adapter-failure rollback path can be unconditional.
                return
            state_str = existing[0]
            if state_str == "committed":
                # Already rolled back.
                return
            raise AdcpError(
                "INTERNAL_ERROR",
                message=(
                    f"release_consumption requires CONSUMING; "
                    f"proposal {proposal_id!r} is in {state_str!r}."
                ),
                recovery="terminal",
            )

    async def mark_consumed(
        self,
        proposal_id: str,
        *,
        media_buy_id: str,
        expected_account_id: str,
    ) -> None:
        # Tenant-scoped SELECT FOR UPDATE → state-machine check →
        # UPDATE. Cross-tenant probes collapse to "not in store".
        async with self._pool.connection() as conn:
            async with conn.transaction():
                cur = await conn.execute(
                    f"SELECT state, media_buy_id FROM {self._table} "  # noqa: S608
                    f"WHERE account_id = %s AND proposal_id = %s FOR UPDATE",
                    (expected_account_id, proposal_id),
                )
                row = await cur.fetchone()
                if row is None:
                    raise AdcpError(
                        "INTERNAL_ERROR",
                        message=(
                            f"Cannot mark_consumed proposal {proposal_id!r}: "
                            "not in store for the expected tenant."
                        ),
                        recovery="terminal",
                    )
                state_str, existing_media_buy_id = row
                if state_str == "consumed":
                    if existing_media_buy_id == media_buy_id:
                        return
                    raise AdcpError(
                        "INTERNAL_ERROR",
                        message=(
                            f"Proposal {proposal_id!r} already consumed by "
                            f"media_buy_id={existing_media_buy_id!r}; cannot "
                            f"re-consume as {media_buy_id!r}."
                        ),
                        recovery="terminal",
                    )
                if state_str != "committed":
                    raise AdcpError(
                        "INTERNAL_ERROR",
                        message=(
                            f"Cannot mark_consumed proposal {proposal_id!r} "
                            f"from state {state_str!r}; mark_consumed "
                            "requires COMMITTED."
                        ),
                        recovery="terminal",
                    )
                await conn.execute(
                    self._sql_mark_consumed,
                    (media_buy_id, expected_account_id, proposal_id),
                )

    async def discard(
        self,
        proposal_id: str,
        *,
        expected_account_id: str,
    ) -> None:
        async with self._pool.connection() as conn:
            await conn.execute(self._sql_discard, (expected_account_id, proposal_id))

    async def get_by_media_buy_id(
        self,
        media_buy_id: str,
        *,
        expected_account_id: str,
    ) -> ProposalRecord | None:
        async with self._pool.connection() as conn:
            cur = await conn.execute(
                self._sql_get_by_media_buy_id,
                (expected_account_id, media_buy_id),
            )
            row = await cur.fetchone()
            if row is None:
                return None
            return self._row_to_record(row)

    # -- helpers --------------------------------------------------------

    def _row_to_record(self, row: tuple[Any, ...]) -> ProposalRecord:
        """Project a SELECT row tuple to a typed :class:`ProposalRecord`."""
        (
            proposal_id,
            account_id,
            state_str,
            recipes_raw,
            payload_raw,
            expires_at_raw,
            media_buy_id,
            recipe_schema_version,
        ) = row
        return ProposalRecord(
            proposal_id=proposal_id,
            account_id=account_id,
            state=ProposalState(state_str),
            recipes=_decode_recipes(recipes_raw, self._recipe_decoder),
            proposal_payload=_decode_payload(payload_raw),
            expires_at=_ensure_utc(expires_at_raw),
            media_buy_id=media_buy_id,
            recipe_schema_version=int(recipe_schema_version or 1),
        )

PostgreSQL-backed :class:~adcp.decisioning.ProposalStore.

Durable counterpart to :class:~adcp.decisioning.InMemoryProposalStore. Set is_durable = True so production-mode gates accept it without requiring the dev-mode bypass.

:param pool: psycopg_pool.AsyncConnectionPool owned by the caller. Each operation acquires a short-lived connection; :meth:try_reserve_consumption holds one for the duration of its CAS transaction. :param table_name: Override the default table name. Useful for adopters with one Postgres serving multiple AdCP instances or whose proposal_drafts table is already taken. Defaults to adcp_proposal_drafts. :param recipe_decoder: Callable (payload: dict) -> Recipe used to rehydrate stored recipe payloads back to typed :class:Recipe instances. Adopters with subclasses (GAMRecipe, KevelRecipe, etc.) MUST supply a decoder that branches on recipe_kind. Defaults to :meth:Recipe.model_validate which only works for the base Recipe shape.

:raises ImportError: when psycopg/psycopg-pool are not installed. :raises ValueError: when table_name is not a safe ASCII identifier ([a-z_][a-z0-9_]{0,62}).

Class variables

var is_durable : ClassVar[bool]

Static methods

def migration_sql(table_name='adcp_proposal_drafts')

Methods

async def commit(self,
proposal_id: str,
*,
expires_at: datetime,
proposal_payload: Mapping[str, Any],
expected_account_id: str) ‑> None

Expand source code

async def commit(
    self,
    proposal_id: str,
    *,
    expires_at: datetime,
    proposal_payload: Mapping[str, Any],
    expected_account_id: str,
) -> None:
    payload_dict = dict(proposal_payload)
    payload_json = json.dumps(payload_dict)
    # Atomic: SELECT FOR UPDATE → state-machine check → UPDATE,
    # all inside one transaction. The SELECT predicate is keyed on
    # (account_id, proposal_id) so a cross-tenant probe collapses
    # to "not in store" without touching another tenant's row.
    # The row lock prevents a concurrent put_draft / commit from
    # racing the validate-then-update sequence.
    async with self._pool.connection() as conn:
        async with conn.transaction():
            cur = await conn.execute(
                self._sql_select_state_for_update,
                (expected_account_id, proposal_id),
            )
            existing = await cur.fetchone()
            if existing is None:
                raise AdcpError(
                    "INTERNAL_ERROR",
                    message=(
                        f"Cannot commit proposal {proposal_id!r}: not "
                        "in store for the expected tenant. The "
                        "framework's finalize dispatch must put_draft "
                        "before commit."
                    ),
                    recovery="terminal",
                )
            current_state, current_expires_at, current_payload = existing
            if current_state == "committed":
                # Idempotent only when the second commit matches the
                # first.
                same_deadline = _ensure_utc(current_expires_at) == expires_at
                cur_payload_dict = (
                    current_payload
                    if isinstance(current_payload, dict)
                    else json.loads(current_payload) if current_payload is not None else {}
                )
                same_payload = cur_payload_dict == payload_dict
                if same_deadline and same_payload:
                    return
                raise AdcpError(
                    "INTERNAL_ERROR",
                    message=(
                        f"Proposal {proposal_id!r} already committed "
                        "with a different expires_at or payload — "
                        "re-commit with different values is a developer "
                        "bug."
                    ),
                    recovery="terminal",
                )
            if current_state != "draft":
                raise AdcpError(
                    "INTERNAL_ERROR",
                    message=(
                        f"Cannot commit proposal {proposal_id!r} from "
                        f"state {current_state!r}; commit requires "
                        "DRAFT."
                    ),
                    recovery="terminal",
                )
            update_cur = await conn.execute(
                self._sql_commit,
                (expires_at, payload_json, expected_account_id, proposal_id),
            )
            if await update_cur.fetchone() is None:
                # Should not happen under the row lock, but fail loud
                # if it does — silent zero-row UPDATE would let the
                # caller believe the transition landed.
                raise AdcpError(
                    "INTERNAL_ERROR",
                    message=(
                        f"PgProposalStore.commit: UPDATE returned zero "
                        f"rows for proposal {proposal_id!r} despite "
                        "passing the FOR UPDATE state check. Schema "
                        "drift suspected."
                    ),
                    recovery="terminal",
                )

async def create_schema(self) ‑> None

Expand source code

async def create_schema(self) -> None:
    """Create the proposal store table + supporting indexes.

    Honors the ``table_name`` kwarg the store was constructed with.
    Idempotent via ``CREATE TABLE IF NOT EXISTS`` — safe to call on
    every application boot. The equivalent raw DDL ships at
    :file:`adcp/decisioning/pg/proposal_store.sql` in the installed
    package for adopters using a migration tool.
    """
    t = self._table
    statements = [
        f"""CREATE TABLE IF NOT EXISTS {t} (
            account_id              TEXT        COLLATE "C" NOT NULL,
            proposal_id             TEXT        COLLATE "C" NOT NULL,
            state                   TEXT        NOT NULL
                CHECK (state IN ('draft', 'committed', 'consuming', 'consumed')),
            recipes                 JSONB       NOT NULL DEFAULT '{{}}'::jsonb,
            proposal_payload        JSONB       NOT NULL,
            expires_at              TIMESTAMPTZ,
            media_buy_id            TEXT        COLLATE "C",
            recipe_schema_version   INTEGER     NOT NULL DEFAULT 1,
            created_at              TIMESTAMPTZ NOT NULL DEFAULT now(),
            updated_at              TIMESTAMPTZ NOT NULL DEFAULT now(),
            PRIMARY KEY (account_id, proposal_id)
        )""",
        f"""CREATE UNIQUE INDEX IF NOT EXISTS {t}_media_buy_idx
            ON {t} (account_id, media_buy_id)
            WHERE media_buy_id IS NOT NULL""",
        f"""CREATE INDEX IF NOT EXISTS {t}_expires_idx
            ON {t} (expires_at)
            WHERE expires_at IS NOT NULL""",
    ]
    async with self._pool.connection() as conn:
        for stmt in statements:
            await conn.execute(stmt)

Create the proposal store table + supporting indexes.

Honors the table_name kwarg the store was constructed with. Idempotent via CREATE TABLE IF NOT EXISTS — safe to call on every application boot. The equivalent raw DDL ships at :file:adcp/decisioning/pg/proposal_store.sql in the installed package for adopters using a migration tool.

async def discard(self, proposal_id: str, *, expected_account_id: str) ‑> None

Expand source code

async def discard(
    self,
    proposal_id: str,
    *,
    expected_account_id: str,
) -> None:
    async with self._pool.connection() as conn:
        await conn.execute(self._sql_discard, (expected_account_id, proposal_id))

async def finalize_consumption(self, proposal_id: str, *, media_buy_id: str, expected_account_id: str) ‑> None

Expand source code

async def finalize_consumption(
    self,
    proposal_id: str,
    *,
    media_buy_id: str,
    expected_account_id: str,
) -> None:
    async with self._pool.connection() as conn:
        cur = await conn.execute(
            self._sql_finalize,
            (media_buy_id, expected_account_id, proposal_id),
        )
        if await cur.fetchone() is not None:
            return
        # Zero rows updated. Determine why.
        cur2 = await conn.execute(self._sql_get_state, (expected_account_id, proposal_id))
        existing = await cur2.fetchone()
        if existing is None:
            raise AdcpError(
                "INTERNAL_ERROR",
                message=(
                    f"finalize_consumption: proposal {proposal_id!r} "
                    "not found for the expected tenant."
                ),
                recovery="terminal",
            )
        state_str = existing[0]
        if state_str == "consumed":
            # Idempotent on re-call with the same media_buy_id.
            cur3 = await conn.execute(
                f"SELECT media_buy_id FROM {self._table} "  # noqa: S608
                f"WHERE account_id = %s AND proposal_id = %s",
                (expected_account_id, proposal_id),
            )
            row = await cur3.fetchone()
            existing_media_buy_id = row[0] if row else None
            if existing_media_buy_id == media_buy_id:
                return
            raise AdcpError(
                "INTERNAL_ERROR",
                message=(
                    f"Proposal {proposal_id!r} already consumed by "
                    f"media_buy_id={existing_media_buy_id!r}; cannot "
                    f"re-consume as {media_buy_id!r}."
                ),
                recovery="terminal",
            )
        raise AdcpError(
            "INTERNAL_ERROR",
            message=(
                f"finalize_consumption requires CONSUMING; "
                f"proposal {proposal_id!r} is in {state_str!r}. "
                "Framework must call try_reserve_consumption first."
            ),
            recovery="terminal",
        )

async def get(self, proposal_id: str, *, expected_account_id: str) ‑> ProposalRecord | None

Expand source code

async def get(
    self,
    proposal_id: str,
    *,
    expected_account_id: str,
) -> ProposalRecord | None:
    sql = (  # noqa: S608 — table name pre-validated at construction
        f"SELECT proposal_id, account_id, state, recipes, "
        f"proposal_payload, expires_at, media_buy_id, "
        f"recipe_schema_version FROM {self._table} "
        f"WHERE account_id = %s AND proposal_id = %s"
    )
    params: tuple[Any, ...] = (expected_account_id, proposal_id)
    async with self._pool.connection() as conn:
        cur = await conn.execute(sql, params)
        row = await cur.fetchone()
        if row is None:
            return None
        return self._row_to_record(row)

async def get_by_media_buy_id(self, media_buy_id: str, *, expected_account_id: str) ‑> ProposalRecord | None

Expand source code

async def get_by_media_buy_id(
    self,
    media_buy_id: str,
    *,
    expected_account_id: str,
) -> ProposalRecord | None:
    async with self._pool.connection() as conn:
        cur = await conn.execute(
            self._sql_get_by_media_buy_id,
            (expected_account_id, media_buy_id),
        )
        row = await cur.fetchone()
        if row is None:
            return None
        return self._row_to_record(row)

async def mark_consumed(self, proposal_id: str, *, media_buy_id: str, expected_account_id: str) ‑> None

Expand source code

async def mark_consumed(
    self,
    proposal_id: str,
    *,
    media_buy_id: str,
    expected_account_id: str,
) -> None:
    # Tenant-scoped SELECT FOR UPDATE → state-machine check →
    # UPDATE. Cross-tenant probes collapse to "not in store".
    async with self._pool.connection() as conn:
        async with conn.transaction():
            cur = await conn.execute(
                f"SELECT state, media_buy_id FROM {self._table} "  # noqa: S608
                f"WHERE account_id = %s AND proposal_id = %s FOR UPDATE",
                (expected_account_id, proposal_id),
            )
            row = await cur.fetchone()
            if row is None:
                raise AdcpError(
                    "INTERNAL_ERROR",
                    message=(
                        f"Cannot mark_consumed proposal {proposal_id!r}: "
                        "not in store for the expected tenant."
                    ),
                    recovery="terminal",
                )
            state_str, existing_media_buy_id = row
            if state_str == "consumed":
                if existing_media_buy_id == media_buy_id:
                    return
                raise AdcpError(
                    "INTERNAL_ERROR",
                    message=(
                        f"Proposal {proposal_id!r} already consumed by "
                        f"media_buy_id={existing_media_buy_id!r}; cannot "
                        f"re-consume as {media_buy_id!r}."
                    ),
                    recovery="terminal",
                )
            if state_str != "committed":
                raise AdcpError(
                    "INTERNAL_ERROR",
                    message=(
                        f"Cannot mark_consumed proposal {proposal_id!r} "
                        f"from state {state_str!r}; mark_consumed "
                        "requires COMMITTED."
                    ),
                    recovery="terminal",
                )
            await conn.execute(
                self._sql_mark_consumed,
                (media_buy_id, expected_account_id, proposal_id),
            )

async def put_draft(self,
*,
proposal_id: str,
account_id: str,
recipes: Mapping[str, Recipe],
proposal_payload: Mapping[str, Any]) ‑> None

Expand source code

async def put_draft(
    self,
    *,
    proposal_id: str,
    account_id: str,
    recipes: Mapping[str, Recipe],
    proposal_payload: Mapping[str, Any],
) -> None:
    recipes_json = _encode_recipes(recipes)
    payload_json = json.dumps(dict(proposal_payload))
    async with self._pool.connection() as conn:
        cur = await conn.execute(
            self._sql_put_draft,
            (account_id, proposal_id, recipes_json, payload_json, 1),
        )
        row = await cur.fetchone()
        if row is not None:
            # Either inserted fresh (xmax=0) or rewrote a DRAFT.
            return
        # ON CONFLICT DO UPDATE matched zero rows — the record is in
        # a non-DRAFT state. Re-fetch to surface the current state in
        # the error message.
        cur2 = await conn.execute(self._sql_get_state, (account_id, proposal_id))
        existing = await cur2.fetchone()
        if existing is None:
            # Race: row vanished between the failed INSERT and the
            # follow-up SELECT (concurrent discard from another
            # worker). Surface as INTERNAL_ERROR so the framework's
            # outer dispatcher can decide whether to retry; we don't
            # transparently retry here because put_draft is meant to
            # be one-shot per dispatch.
            raise AdcpError(
                "INTERNAL_ERROR",
                message=(
                    f"PgProposalStore.put_draft: proposal {proposal_id!r} "
                    "vanished between conflict and refetch. Concurrent "
                    "discard suspected."
                ),
                recovery="terminal",
            )
        state_str = existing[0]
        raise AdcpError(
            "INTERNAL_ERROR",
            message=(
                f"Cannot put_draft on proposal {proposal_id!r} in "
                f"state {state_str!r}; refine iterations are only "
                "valid on draft proposals. Once committed or "
                "consumed, a proposal_id is immutable."
            ),
            recovery="terminal",
        )

async def release_consumption(self, proposal_id: str, *, expected_account_id: str) ‑> None

Expand source code

async def release_consumption(
    self,
    proposal_id: str,
    *,
    expected_account_id: str,
) -> None:
    async with self._pool.connection() as conn:
        cur = await conn.execute(
            self._sql_release,
            (expected_account_id, proposal_id),
        )
        if await cur.fetchone() is not None:
            return
        # Zero rows updated. Idempotent: no-op on unknown id /
        # cross-tenant probe, no-op on already-COMMITTED.
        cur2 = await conn.execute(self._sql_get_state, (expected_account_id, proposal_id))
        existing = await cur2.fetchone()
        if existing is None:
            # Unknown id or cross-tenant — idempotent no-op so the
            # adapter-failure rollback path can be unconditional.
            return
        state_str = existing[0]
        if state_str == "committed":
            # Already rolled back.
            return
        raise AdcpError(
            "INTERNAL_ERROR",
            message=(
                f"release_consumption requires CONSUMING; "
                f"proposal {proposal_id!r} is in {state_str!r}."
            ),
            recovery="terminal",
        )

async def try_reserve_consumption(self, proposal_id: str, *, expected_account_id: str) ‑> ProposalRecord

Expand source code

async def try_reserve_consumption(
    self,
    proposal_id: str,
    *,
    expected_account_id: str,
) -> ProposalRecord:
    # Single-connection transaction so SELECT FOR UPDATE + UPDATE
    # serialize across parallel callers.
    async with self._pool.connection() as conn:
        async with conn.transaction():
            cur = await conn.execute(
                self._sql_select_for_update,
                (expected_account_id, proposal_id),
            )
            row = await cur.fetchone()
            if row is None:
                raise AdcpError(
                    "PROPOSAL_NOT_FOUND",
                    message=(f"Proposal {proposal_id!r} not found."),
                    recovery="correctable",
                    field="proposal_id",
                )
            (
                state_str,
                recipes_raw,
                payload_raw,
                expires_at_raw,
                media_buy_id,
                recipe_schema_version,
            ) = row
            if state_str != "committed":
                raise AdcpError(
                    "PROPOSAL_NOT_COMMITTED",
                    message=(
                        f"Proposal {proposal_id!r} is in state "
                        f"{state_str!r}; create_media_buy requires a "
                        "committed proposal that hasn't been accepted "
                        "or reserved by another request."
                    ),
                    recovery="correctable",
                    field="proposal_id",
                )
            await conn.execute(
                self._sql_reserve,
                (expected_account_id, proposal_id),
            )
            # Build the in-memory record reflecting the transition.
            return ProposalRecord(
                proposal_id=proposal_id,
                account_id=expected_account_id,
                state=ProposalState.CONSUMING,
                recipes=_decode_recipes(recipes_raw, self._recipe_decoder),
                proposal_payload=_decode_payload(payload_raw),
                expires_at=_ensure_utc(expires_at_raw),
                media_buy_id=media_buy_id,
                recipe_schema_version=int(recipe_schema_version or 1),
            )

class PgTaskRegistry (*,
pool: AsyncConnectionPool,
task_webhook_outbox: PgTaskWebhookOutbox | None = None,
webhook_signing_scope_resolver: WebhookSigningScopeResolver | None = None)

Expand source code

class PgTaskRegistry(_TaskLifecycleObservers):
    """PostgreSQL-backed :class:`~adcp.decisioning.TaskRegistry` — v6.1.

    Durable counterpart to :class:`~adcp.decisioning.InMemoryTaskRegistry`.
    Set ``is_durable = True`` so the production-mode gate in
    :func:`adcp.decisioning.serve.create_adcp_server_from_platform` accepts it
    without requiring ``ADCP_DECISIONING_ALLOW_INMEMORY_TASKS=1``.

    Parameters
    ----------
    pool:
        An :class:`psycopg_pool.AsyncConnectionPool` owned by the caller.
        Each registry operation acquires a short-lived connection from the
        pool and returns it immediately after the query. No long-lived
        transactions, no cross-operation state.
    task_webhook_outbox:
        Optional :class:`PgTaskWebhookOutbox` sharing this pool. When a task
        carries callback registration, ``complete`` / ``fail`` enqueue its
        immutable terminal webhook on the same transaction connection.

    Notes
    -----
    Unlike :class:`~adcp.signing.PgReplayStore`, this class uses a fixed
    ``decisioning_tasks`` table name. Multi-tenant table-name isolation is not
    supported in this release — callers requiring strict schema separation
    should use separate databases or schemas.
    """

    is_durable: ClassVar[bool] = True

    def __init__(
        self,
        *,
        pool: AsyncConnectionPool,
        task_webhook_outbox: PgTaskWebhookOutbox | None = None,
        webhook_signing_scope_resolver: WebhookSigningScopeResolver | None = None,
        _table: str = _DEFAULT_TABLE,
    ) -> None:
        if not PG_AVAILABLE:
            raise ImportError(_INSTALL_HINT)
        if not _SAFE_IDENTIFIER_RE.fullmatch(_table):
            raise ValueError(f"_table must match [a-z_][a-z0-9_]* (ASCII only), got {_table!r}")
        if task_webhook_outbox is not None and task_webhook_outbox._pool is not pool:
            raise ValueError(
                "PgTaskRegistry and PgTaskWebhookOutbox must use the same connection pool"
            )
        uses_sender_resolver = (
            task_webhook_outbox is not None and task_webhook_outbox._sender_resolver is not None
        )
        if uses_sender_resolver != (webhook_signing_scope_resolver is not None):
            raise ValueError(
                "webhook_signing_scope_resolver is required exactly when the "
                "PgTaskWebhookOutbox uses sender_resolver"
            )
        self._init_lifecycle_observers()
        self._pool = pool
        self._table = _table
        self.task_webhook_outbox = task_webhook_outbox
        self._webhook_signing_scope_resolver = webhook_signing_scope_resolver
        self.atomic_task_webhook_outbox = task_webhook_outbox is not None

        # Pre-format queries at construction so the hot path avoids f-strings per call.
        # _table is whitelisted by _SAFE_IDENTIFIER_RE above.
        self._sql_insert = (  # noqa: S608 — table name is whitelisted
            f"INSERT INTO {self._table}"  # nosec B608 — table identifier is constructor-validated
            f" (task_id, account_id, state, task_type, request_context,"
            f"  webhook_registration, webhook_registration_nonce, created_at, updated_at)"
            f" VALUES (%s, %s, 'submitted', %s, %s::jsonb, %s, %s, %s, %s)"
        )
        self._sql_update_progress = (  # noqa: S608
            f"WITH previous AS (SELECT task_id, state FROM {self._table}"  # nosec B608 — table identifier is constructor-validated
            f" WHERE task_id = %s AND state NOT IN ('completed', 'failed') FOR UPDATE)"
            f" UPDATE {self._table} AS task"
            f" SET state = CASE task.state WHEN 'submitted' THEN 'working' ELSE task.state END,"
            f"     progress = %s::jsonb, updated_at = %s"
            f" FROM previous WHERE task.task_id = previous.task_id"
            f" RETURNING previous.state, task.task_id, task.account_id, task.task_type,"
            f" task.created_at, task.updated_at"
        )
        self._sql_complete = (  # noqa: S608
            f"UPDATE {self._table}"  # nosec B608 — table identifier is constructor-validated
            f" SET state = 'completed', result = %s::jsonb, updated_at = %s"
            f" WHERE task_id = %s AND state NOT IN ('completed', 'failed')"
            f" RETURNING task_id, account_id, task_type, webhook_registration,"
            f" webhook_registration_nonce, created_at, updated_at"
        )
        self._sql_fail = (  # noqa: S608
            f"UPDATE {self._table}"  # nosec B608 — table identifier is constructor-validated
            f" SET state = 'failed', error = %s::jsonb, updated_at = %s"
            f" WHERE task_id = %s AND state NOT IN ('completed', 'failed')"
            f" RETURNING task_id, account_id, task_type, webhook_registration,"
            f" webhook_registration_nonce, created_at, updated_at"
        )
        self._sql_clear_webhook_registration = (  # noqa: S608
            f"UPDATE {self._table} SET webhook_registration = NULL,"  # nosec B608 — table identifier is constructor-validated
            f" webhook_registration_nonce = NULL WHERE task_id = %s"
        )
        # Explicit ``::text`` cast on the optional account-filter
        # parameter so psycopg's bind-param type inference doesn't
        # fail with ``IndeterminateDatatype: could not determine
        # data type of parameter $2``. Without the cast, the
        # ``%s IS NULL`` predicate gives psycopg no type context
        # for the parameter and the query fails at prepare time.
        self._sql_get = (  # noqa: S608
            f"SELECT task_id, account_id, state, task_type,"  # nosec B608 — table identifier is constructor-validated
            f"       progress, result, error, request_context, created_at, updated_at"
            f" FROM {self._table}"
            f" WHERE task_id = %s AND (%s::text IS NULL OR account_id = %s)"
        )
        self._sql_get_state_result = (  # noqa: S608
            f"SELECT state, result, task_type FROM {self._table}"  # nosec B608 — validated identifier
            " WHERE task_id = %s"
        )
        self._sql_get_task_type = f"SELECT task_type FROM {self._table} WHERE task_id = %s"  # noqa: S608  # nosec B608 — table identifier is constructor-validated
        self._sql_get_state_error = f"SELECT state, error FROM {self._table} WHERE task_id = %s"  # noqa: S608  # nosec B608 — table identifier is constructor-validated
        self._sql_discard = (  # noqa: S608
            f"DELETE FROM {self._table} WHERE task_id = %s"  # nosec B608 — table identifier is constructor-validated
            f" RETURNING task_id, account_id, task_type, created_at, updated_at"
        )
        self._sql_ddl = (  # noqa: S608
            f"CREATE TABLE IF NOT EXISTS {self._table} ("
            f'    task_id     TEXT             COLLATE "C" NOT NULL PRIMARY KEY,'
            f'    account_id  TEXT             COLLATE "C" NOT NULL,'
            f"    state       TEXT             NOT NULL DEFAULT 'submitted',"
            f"    task_type   TEXT             NOT NULL,"
            f"    progress    JSONB,"
            f"    result      JSONB,"
            f"    error       JSONB,"
            f"    request_context JSONB,"
            f"    webhook_registration BYTEA,"
            f"    webhook_registration_nonce BYTEA,"
            f"    created_at  DOUBLE PRECISION NOT NULL,"
            f"    updated_at  DOUBLE PRECISION NOT NULL"
            f");"
            f"ALTER TABLE {self._table} ADD COLUMN IF NOT EXISTS request_context JSONB;"
            f"ALTER TABLE {self._table} ADD COLUMN IF NOT EXISTS webhook_registration BYTEA;"
            f"ALTER TABLE {self._table} ADD COLUMN IF NOT EXISTS webhook_registration_nonce BYTEA;"
            f"CREATE INDEX IF NOT EXISTS {self._table}_account_idx"  # noqa: S608
            f"    ON {self._table} (account_id);"
        )

    # -- schema bootstrap -----------------------------------------------

    async def create_schema(self) -> None:
        """Create the task registry table and supporting index.

        Honors the ``_table`` kwarg the store was constructed with.
        Idempotent via ``CREATE TABLE IF NOT EXISTS`` — safe to call on every
        application boot. The equivalent raw DDL ships at
        ``adcp/decisioning/pg/decisioning_tasks.sql`` in the installed package
        for adopters using a migration tool (Alembic, Flyway, psql).
        """
        async with self._pool.connection() as conn:
            await conn.execute(self._sql_ddl)

    # -- TaskRegistry Protocol ------------------------------------------

    async def issue(
        self,
        *,
        account_id: str,
        task_type: str,
        request_context: dict[str, Any] | None = None,
        webhook_url: str | None = None,
        webhook_operation_id: str | None = None,
        webhook_token: str | None = None,
        webhook_authentication: TaskWebhookAuthentication | None = None,
        webhook_signing_scope_id: str | None = None,
        **_extra: Any,
    ) -> str:
        """Allocate a task_id, persist a ``submitted`` row, return the id.

        Mirrors :meth:`~adcp.decisioning.InMemoryTaskRegistry.issue` including
        the account_id validation guard — empty or sentinel account_ids would
        allow cross-tenant task-id probing via the ``WHERE account_id = %s``
        predicate collapsing multiple tenants into one slot.
        """
        if not account_id or not account_id.strip() or account_id == "<unset>":
            raise ValueError(
                f"account_id must be a non-empty, non-default string; "
                f"got {account_id!r}. AccountStore.resolve must always "
                "return Account(id=<non-empty>) so cross-tenant cache "
                "scoping works correctly."
            )
        if (webhook_url is None) != (webhook_operation_id is None):
            raise ValueError("webhook_url and webhook_operation_id must be supplied together")
        if webhook_url is not None and not webhook_url:
            raise ValueError("webhook_url must be non-empty when supplied")
        if webhook_operation_id is not None and not webhook_operation_id:
            raise ValueError("webhook_operation_id must be non-empty when supplied")
        if webhook_url is None and webhook_signing_scope_id is not None:
            raise ValueError("webhook_signing_scope_id requires webhook_url")
        if webhook_url is None and webhook_authentication is not None:
            raise ValueError("webhook_authentication requires webhook_url")
        if webhook_authentication is not None and webhook_signing_scope_id is not None:
            raise ValueError(
                "legacy webhook_authentication must not carry an RFC 9421 signing scope"
            )
        outbox = self.task_webhook_outbox
        if webhook_url is not None:
            if outbox is None:
                raise ValueError("webhook registration requires the registry's task_webhook_outbox")
            outbox.validate_registration(webhook_url, webhook_authentication)
        task_id = f"task_{uuid.uuid4().hex[:16]}"
        encrypted_registration: bytes | None = None
        registration_nonce: bytes | None = None
        if webhook_url is not None:
            if webhook_operation_id is None or outbox is None:
                raise RuntimeError("validated webhook registration became incomplete")
            encrypted_registration, registration_nonce = outbox.protect_registration(
                account_id=account_id,
                task_id=task_id,
                task_type=task_type,
                url=webhook_url,
                operation_id=webhook_operation_id,
                token=webhook_token,
                authentication=webhook_authentication,
                signing_scope_id=webhook_signing_scope_id,
            )
        now = time.time()
        async with self._pool.connection() as conn:
            await conn.execute(
                self._sql_insert,
                (
                    task_id,
                    account_id,
                    task_type,
                    json.dumps(request_context) if request_context is not None else None,
                    encrypted_registration,
                    registration_nonce,
                    now,
                    now,
                ),
            )
        self._notify_lifecycle_observers(
            "submitted",
            {
                "task_id": task_id,
                "account_id": account_id,
                "task_type": task_type,
                "created_at": now,
                "updated_at": now,
            },
        )
        return task_id

    async def resolve_webhook_signing_scope(
        self,
        context: RequestContext[Any],
    ) -> str | None:
        """Derive an opaque signing scope from trusted framework context.

        The callback is operator wiring and receives the hydrated
        :class:`RequestContext`. It must use internal tenant/platform metadata,
        never buyer request ``context``, ``push_notification_config``, or an
        unqualified buyer account id.
        """
        resolver = self._webhook_signing_scope_resolver
        if resolver is None:
            return None
        from adcp.decisioning.types import AdcpError

        try:
            value: object = resolver(context)
            if inspect.isawaitable(value):
                value = await value
        except Exception:
            raise AdcpError(
                "INTERNAL_ERROR",
                message="Webhook signing scope resolution failed",
                recovery="terminal",
            ) from None
        if not isinstance(value, str):
            raise AdcpError(
                "INTERNAL_ERROR",
                message="Webhook signing scope resolver returned an invalid value",
                recovery="terminal",
            )
        # Reuse the outbox's bounded opaque-ID validation before any task row
        # is issued. This is trusted server state, but it still crosses a DB
        # and authenticated-envelope boundary.
        if self.task_webhook_outbox is None:
            raise RuntimeError("signing scope resolver requires a task webhook outbox")
        self.task_webhook_outbox._validate_signing_scope_id(value)
        return value

    async def update_progress(
        self,
        task_id: str,
        progress: dict[str, Any],
    ) -> None:
        """Write a progress payload; transition ``submitted`` → ``working``.

        Silently no-ops when the task is already in a terminal state or
        unknown — the dispatch wrapper expects this method never to raise on
        transient conditions (see :class:`~adcp.decisioning.TaskRegistry`
        docstring).

        The ``state NOT IN ('completed', 'failed')`` predicate is evaluated
        server-side so a concurrent terminal write cannot be overwritten by a
        straggler progress event.
        """
        async with self._pool.connection() as conn:
            cur = await conn.execute(
                self._sql_update_progress,
                (task_id, json.dumps(progress), time.time()),
            )
            row = await cur.fetchone()
        if row is not None and row[0] == "submitted":
            self._notify_lifecycle_observers("working", self._lifecycle_record(row[1:]))

    @staticmethod
    def _lifecycle_record(row: tuple[Any, ...]) -> dict[str, Any]:
        return dict(zip(("task_id", "account_id", "task_type", "created_at", "updated_at"), row))

    async def complete(self, task_id: str, result: dict[str, Any]) -> None:
        """Complete once; notify metrics only after the transaction commits."""
        row = await self._complete(task_id, result)
        if row is not None:
            self._notify_lifecycle_observers("completed", self._lifecycle_record(row[:3] + row[5:]))

    async def _complete(
        self,
        task_id: str,
        result: dict[str, Any],
    ) -> tuple[Any, ...] | None:
        """Mark the task ``completed`` with ``result`` as the terminal artifact.

        Idempotent on repeated calls with an equal ``result``; raises
        :class:`ValueError` on conflicting re-completion.

        Uses an atomic ``UPDATE ... RETURNING`` so concurrent workers cannot
        race each other into double-completion without detection.
        """
        async with self._pool.connection() as conn:
            # Do not rely on the pool connection's implicit transaction.
            # Explicitly bind the terminal transition, outbox insert, and
            # callback-registration clear even when adopters configure the
            # pool with autocommit=True.
            async with conn.transaction():
                type_cursor = await conn.execute(self._sql_get_task_type, (task_id,))
                type_row = await type_cursor.fetchone()
                safe_result = (
                    strip_credentials_from_wire_result(type_row[0], result)
                    if type_row is not None
                    else result
                )
                cur = await conn.execute(
                    self._sql_complete,
                    (json.dumps(safe_result), time.time(), task_id),
                )
                row: tuple[Any, ...] | None = await cur.fetchone()
                if row is not None:
                    await self._enqueue_terminal_if_registered(
                        conn,
                        row=row,
                        status="completed",
                        payload=safe_result,
                    )
                    return row  # connection and transaction exit before the caller notifies

                # Zero rows in RETURNING — task is unknown or already terminal.
                cur2 = await conn.execute(self._sql_get_state_result, (task_id,))
                row = await cur2.fetchone()
                if row is None:
                    raise ValueError(f"Task {task_id!r} not found")
                state, existing_result, task_type = row
                safe_result = strip_credentials_from_wire_result(task_type, result)
                if state == "completed":
                    if existing_result == safe_result:
                        return None  # idempotent
                    raise ValueError(f"Task {task_id!r} already completed with a different result")
                raise ValueError(f"Task {task_id!r} already in terminal state {state!r}")

    async def fail(self, task_id: str, error: dict[str, Any]) -> None:
        """Fail once; notify metrics only after the transaction commits."""
        row = await self._fail(task_id, error)
        if row is not None:
            self._notify_lifecycle_observers("failed", self._lifecycle_record(row[:3] + row[5:]))

    async def _fail(
        self,
        task_id: str,
        error: dict[str, Any],
    ) -> tuple[Any, ...] | None:
        """Mark the task ``failed`` with ``error`` as the terminal payload.

        Idempotent on repeated calls with an equal ``error``; raises
        :class:`ValueError` on conflicting re-failure.
        """
        async with self._pool.connection() as conn:
            async with conn.transaction():
                cur = await conn.execute(
                    self._sql_fail,
                    (json.dumps(error), time.time(), task_id),
                )
                row: tuple[Any, ...] | None = await cur.fetchone()
                if row is not None:
                    await self._enqueue_terminal_if_registered(
                        conn,
                        row=row,
                        status="failed",
                        payload=error,
                    )
                    return row  # notify only after connection/transaction exit

                # Zero rows in RETURNING — task is unknown or already terminal.
                cur2 = await conn.execute(self._sql_get_state_error, (task_id,))
                row = await cur2.fetchone()
                if row is None:
                    raise ValueError(f"Task {task_id!r} not found")
                state, existing_error = row
                if state == "failed":
                    if existing_error == error:
                        return None  # idempotent
                    raise ValueError(f"Task {task_id!r} already failed with a different error")
                raise ValueError(f"Task {task_id!r} already in terminal state {state!r}")

    async def get(
        self,
        task_id: str,
        *,
        expected_account_id: str | None = None,
    ) -> dict[str, Any] | None:
        """Look up a task record; cross-tenant probes return ``None``.

        The ``expected_account_id`` predicate is enforced at the SQL level
        (``WHERE account_id = %s``), not as a Python-level filter after fetch.
        This guarantees the row is never materialized for a mismatched probe,
        eliminating the fetch-then-filter anti-pattern.
        """
        async with self._pool.connection() as conn:
            cur = await conn.execute(
                self._sql_get, (task_id, expected_account_id, expected_account_id)
            )
            row = await cur.fetchone()
            if row is None:
                return None
            return {
                "task_id": row[0],
                "account_id": row[1],
                "state": row[2],
                "task_type": row[3],
                "progress": row[4],
                "result": row[5],
                "error": row[6],
                "created_at": row[8],
                "updated_at": row[9],
                **({"context": row[7]} if row[7] is not None else {}),
            }

    async def list(
        self,
        *,
        account_id: str,
        filters: dict[str, Any] | None = None,
        sort: dict[str, Any] | None = None,
        pagination: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        from adcp.decisioning.task_queries import list_task_records

        # Account isolation is a SQL predicate, before rows are materialized.
        async with self._pool.connection() as conn:
            cur = await conn.execute(
                f"SELECT task_id, account_id, state, task_type, progress, result, error,"  # nosec B608 — table identifier is constructor-validated
                f" request_context, created_at, updated_at,"
                f" (webhook_registration IS NOT NULL) FROM {self._table} WHERE account_id = %s",
                (account_id,),
            )
            rows = await cur.fetchall()
        records = [
            dict(
                zip(
                    (
                        "task_id",
                        "account_id",
                        "state",
                        "task_type",
                        "progress",
                        "result",
                        "error",
                        "context",
                        "created_at",
                        "updated_at",
                        "has_webhook",
                    ),
                    row,
                )
            )
            for row in rows
        ]
        return list_task_records(
            records, account_id=account_id, filters=filters, sort=sort, pagination=pagination
        )

    async def _enqueue_terminal_if_registered(
        self,
        conn: Any,
        *,
        row: tuple[Any, ...],
        status: str,
        payload: dict[str, Any],
    ) -> None:
        """Enqueue on ``conn`` so task state and webhook commit atomically."""
        task_id, account_id, task_type, encrypted_registration, registration_nonce = row[:5]
        if encrypted_registration is None:
            return
        if self.task_webhook_outbox is None:
            raise RuntimeError(
                "Task carries push_notification_config but PgTaskRegistry has no "
                "task_webhook_outbox; refusing a non-atomic terminal transition"
            )
        if registration_nonce is None:
            raise RuntimeError(f"Task {task_id!r} has incomplete webhook registration")
        url, operation_id, token, authentication, signing_scope_id = (
            self.task_webhook_outbox._open_registration_with_scope(
                account_id=account_id,
                task_id=task_id,
                task_type=task_type,
                encrypted_registration=bytes(encrypted_registration),
                nonce=bytes(registration_nonce),
            )
        )
        await self.task_webhook_outbox.enqueue_terminal(
            conn,
            task_id=task_id,
            account_id=account_id,
            task_type=task_type,
            status=status,
            result=payload,
            url=url,
            operation_id=operation_id,
            token=token,
            authentication=authentication,
            signing_scope_id=signing_scope_id,
        )
        # The encrypted outbox envelope now owns the callback registration.
        # Clear the task-row copy in this same transaction.
        await conn.execute(self._sql_clear_webhook_registration, (task_id,))

    async def discard(self, task_id: str) -> None:
        """Remove a task_id from the registry — rollback path.

        Idempotent: discarding an unknown task_id is a no-op (no raise),
        matching the :class:`~adcp.decisioning.InMemoryTaskRegistry` contract.
        """
        async with self._pool.connection() as conn:
            cur = await conn.execute(self._sql_discard, (task_id,))
            row = await cur.fetchone()
        if row is not None:
            event_record = self._lifecycle_record(row)
            event_record["updated_at"] = time.time()
            self._notify_lifecycle_observers("discarded", event_record)

PostgreSQL-backed :class:~adcp.decisioning.TaskRegistry — v6.1.

Durable counterpart to :class:~adcp.decisioning.InMemoryTaskRegistry. Set is_durable = True so the production-mode gate in :func:adcp.decisioning.serve.create_adcp_server_from_platform accepts it without requiring ADCP_DECISIONING_ALLOW_INMEMORY_TASKS=1.

Parameters

pool: An :class:psycopg_pool.AsyncConnectionPool owned by the caller. Each registry operation acquires a short-lived connection from the pool and returns it immediately after the query. No long-lived transactions, no cross-operation state. task_webhook_outbox: Optional :class:PgTaskWebhookOutbox sharing this pool. When a task carries callback registration, complete / fail enqueue its immutable terminal webhook on the same transaction connection.

Notes

Unlike :class:~adcp.signing.PgReplayStore, this class uses a fixed decisioning_tasks table name. Multi-tenant table-name isolation is not supported in this release — callers requiring strict schema separation should use separate databases or schemas.

Ancestors

• adcp.decisioning.task_registry._TaskLifecycleObservers

Class variables

var is_durable : ClassVar[bool]

Methods

async def complete(self, task_id: str, result: dict[str, Any]) ‑> None

Expand source code

async def complete(self, task_id: str, result: dict[str, Any]) -> None:
    """Complete once; notify metrics only after the transaction commits."""
    row = await self._complete(task_id, result)
    if row is not None:
        self._notify_lifecycle_observers("completed", self._lifecycle_record(row[:3] + row[5:]))

Complete once; notify metrics only after the transaction commits.

async def create_schema(self) ‑> None

Expand source code

async def create_schema(self) -> None:
    """Create the task registry table and supporting index.

    Honors the ``_table`` kwarg the store was constructed with.
    Idempotent via ``CREATE TABLE IF NOT EXISTS`` — safe to call on every
    application boot. The equivalent raw DDL ships at
    ``adcp/decisioning/pg/decisioning_tasks.sql`` in the installed package
    for adopters using a migration tool (Alembic, Flyway, psql).
    """
    async with self._pool.connection() as conn:
        await conn.execute(self._sql_ddl)

Create the task registry table and supporting index.

Honors the _table kwarg the store was constructed with. Idempotent via CREATE TABLE IF NOT EXISTS — safe to call on every application boot. The equivalent raw DDL ships at adcp/decisioning/pg/decisioning_tasks.sql in the installed package for adopters using a migration tool (Alembic, Flyway, psql).

async def discard(self, task_id: str) ‑> None

Expand source code

async def discard(self, task_id: str) -> None:
    """Remove a task_id from the registry — rollback path.

    Idempotent: discarding an unknown task_id is a no-op (no raise),
    matching the :class:`~adcp.decisioning.InMemoryTaskRegistry` contract.
    """
    async with self._pool.connection() as conn:
        cur = await conn.execute(self._sql_discard, (task_id,))
        row = await cur.fetchone()
    if row is not None:
        event_record = self._lifecycle_record(row)
        event_record["updated_at"] = time.time()
        self._notify_lifecycle_observers("discarded", event_record)

Remove a task_id from the registry — rollback path.

Idempotent: discarding an unknown task_id is a no-op (no raise), matching the :class:~adcp.decisioning.InMemoryTaskRegistry contract.

async def fail(self, task_id: str, error: dict[str, Any]) ‑> None

Expand source code

async def fail(self, task_id: str, error: dict[str, Any]) -> None:
    """Fail once; notify metrics only after the transaction commits."""
    row = await self._fail(task_id, error)
    if row is not None:
        self._notify_lifecycle_observers("failed", self._lifecycle_record(row[:3] + row[5:]))

Fail once; notify metrics only after the transaction commits.

async def get(self, task_id: str, *, expected_account_id: str | None = None) ‑> dict[str, typing.Any] | None

Expand source code

async def get(
    self,
    task_id: str,
    *,
    expected_account_id: str | None = None,
) -> dict[str, Any] | None:
    """Look up a task record; cross-tenant probes return ``None``.

    The ``expected_account_id`` predicate is enforced at the SQL level
    (``WHERE account_id = %s``), not as a Python-level filter after fetch.
    This guarantees the row is never materialized for a mismatched probe,
    eliminating the fetch-then-filter anti-pattern.
    """
    async with self._pool.connection() as conn:
        cur = await conn.execute(
            self._sql_get, (task_id, expected_account_id, expected_account_id)
        )
        row = await cur.fetchone()
        if row is None:
            return None
        return {
            "task_id": row[0],
            "account_id": row[1],
            "state": row[2],
            "task_type": row[3],
            "progress": row[4],
            "result": row[5],
            "error": row[6],
            "created_at": row[8],
            "updated_at": row[9],
            **({"context": row[7]} if row[7] is not None else {}),
        }

Look up a task record; cross-tenant probes return None.

The expected_account_id predicate is enforced at the SQL level (WHERE account_id = %s), not as a Python-level filter after fetch. This guarantees the row is never materialized for a mismatched probe, eliminating the fetch-then-filter anti-pattern.

async def issue(self,
*,
account_id: str,
task_type: str,
request_context: dict[str, Any] | None = None,
webhook_url: str | None = None,
webhook_operation_id: str | None = None,
webhook_token: str | None = None,
webhook_authentication: TaskWebhookAuthentication | None = None,
webhook_signing_scope_id: str | None = None,
**_extra: Any) ‑> str

Expand source code

async def issue(
    self,
    *,
    account_id: str,
    task_type: str,
    request_context: dict[str, Any] | None = None,
    webhook_url: str | None = None,
    webhook_operation_id: str | None = None,
    webhook_token: str | None = None,
    webhook_authentication: TaskWebhookAuthentication | None = None,
    webhook_signing_scope_id: str | None = None,
    **_extra: Any,
) -> str:
    """Allocate a task_id, persist a ``submitted`` row, return the id.

    Mirrors :meth:`~adcp.decisioning.InMemoryTaskRegistry.issue` including
    the account_id validation guard — empty or sentinel account_ids would
    allow cross-tenant task-id probing via the ``WHERE account_id = %s``
    predicate collapsing multiple tenants into one slot.
    """
    if not account_id or not account_id.strip() or account_id == "<unset>":
        raise ValueError(
            f"account_id must be a non-empty, non-default string; "
            f"got {account_id!r}. AccountStore.resolve must always "
            "return Account(id=<non-empty>) so cross-tenant cache "
            "scoping works correctly."
        )
    if (webhook_url is None) != (webhook_operation_id is None):
        raise ValueError("webhook_url and webhook_operation_id must be supplied together")
    if webhook_url is not None and not webhook_url:
        raise ValueError("webhook_url must be non-empty when supplied")
    if webhook_operation_id is not None and not webhook_operation_id:
        raise ValueError("webhook_operation_id must be non-empty when supplied")
    if webhook_url is None and webhook_signing_scope_id is not None:
        raise ValueError("webhook_signing_scope_id requires webhook_url")
    if webhook_url is None and webhook_authentication is not None:
        raise ValueError("webhook_authentication requires webhook_url")
    if webhook_authentication is not None and webhook_signing_scope_id is not None:
        raise ValueError(
            "legacy webhook_authentication must not carry an RFC 9421 signing scope"
        )
    outbox = self.task_webhook_outbox
    if webhook_url is not None:
        if outbox is None:
            raise ValueError("webhook registration requires the registry's task_webhook_outbox")
        outbox.validate_registration(webhook_url, webhook_authentication)
    task_id = f"task_{uuid.uuid4().hex[:16]}"
    encrypted_registration: bytes | None = None
    registration_nonce: bytes | None = None
    if webhook_url is not None:
        if webhook_operation_id is None or outbox is None:
            raise RuntimeError("validated webhook registration became incomplete")
        encrypted_registration, registration_nonce = outbox.protect_registration(
            account_id=account_id,
            task_id=task_id,
            task_type=task_type,
            url=webhook_url,
            operation_id=webhook_operation_id,
            token=webhook_token,
            authentication=webhook_authentication,
            signing_scope_id=webhook_signing_scope_id,
        )
    now = time.time()
    async with self._pool.connection() as conn:
        await conn.execute(
            self._sql_insert,
            (
                task_id,
                account_id,
                task_type,
                json.dumps(request_context) if request_context is not None else None,
                encrypted_registration,
                registration_nonce,
                now,
                now,
            ),
        )
    self._notify_lifecycle_observers(
        "submitted",
        {
            "task_id": task_id,
            "account_id": account_id,
            "task_type": task_type,
            "created_at": now,
            "updated_at": now,
        },
    )
    return task_id

Allocate a task_id, persist a submitted row, return the id.

Mirrors :meth:~adcp.decisioning.InMemoryTaskRegistry.issue including the account_id validation guard — empty or sentinel account_ids would allow cross-tenant task-id probing via the WHERE account_id = %s predicate collapsing multiple tenants into one slot.

async def list(self,
*,
account_id: str,
filters: dict[str, Any] | None = None,
sort: dict[str, Any] | None = None,
pagination: dict[str, Any] | None = None) ‑> dict[str, typing.Any]

Expand source code

async def list(
    self,
    *,
    account_id: str,
    filters: dict[str, Any] | None = None,
    sort: dict[str, Any] | None = None,
    pagination: dict[str, Any] | None = None,
) -> dict[str, Any]:
    from adcp.decisioning.task_queries import list_task_records

    # Account isolation is a SQL predicate, before rows are materialized.
    async with self._pool.connection() as conn:
        cur = await conn.execute(
            f"SELECT task_id, account_id, state, task_type, progress, result, error,"  # nosec B608 — table identifier is constructor-validated
            f" request_context, created_at, updated_at,"
            f" (webhook_registration IS NOT NULL) FROM {self._table} WHERE account_id = %s",
            (account_id,),
        )
        rows = await cur.fetchall()
    records = [
        dict(
            zip(
                (
                    "task_id",
                    "account_id",
                    "state",
                    "task_type",
                    "progress",
                    "result",
                    "error",
                    "context",
                    "created_at",
                    "updated_at",
                    "has_webhook",
                ),
                row,
            )
        )
        for row in rows
    ]
    return list_task_records(
        records, account_id=account_id, filters=filters, sort=sort, pagination=pagination
    )

async def resolve_webhook_signing_scope(self, context: RequestContext[Any]) ‑> str | None

Expand source code

async def resolve_webhook_signing_scope(
    self,
    context: RequestContext[Any],
) -> str | None:
    """Derive an opaque signing scope from trusted framework context.

    The callback is operator wiring and receives the hydrated
    :class:`RequestContext`. It must use internal tenant/platform metadata,
    never buyer request ``context``, ``push_notification_config``, or an
    unqualified buyer account id.
    """
    resolver = self._webhook_signing_scope_resolver
    if resolver is None:
        return None
    from adcp.decisioning.types import AdcpError

    try:
        value: object = resolver(context)
        if inspect.isawaitable(value):
            value = await value
    except Exception:
        raise AdcpError(
            "INTERNAL_ERROR",
            message="Webhook signing scope resolution failed",
            recovery="terminal",
        ) from None
    if not isinstance(value, str):
        raise AdcpError(
            "INTERNAL_ERROR",
            message="Webhook signing scope resolver returned an invalid value",
            recovery="terminal",
        )
    # Reuse the outbox's bounded opaque-ID validation before any task row
    # is issued. This is trusted server state, but it still crosses a DB
    # and authenticated-envelope boundary.
    if self.task_webhook_outbox is None:
        raise RuntimeError("signing scope resolver requires a task webhook outbox")
    self.task_webhook_outbox._validate_signing_scope_id(value)
    return value

Derive an opaque signing scope from trusted framework context.

The callback is operator wiring and receives the hydrated :class:RequestContext. It must use internal tenant/platform metadata, never buyer request context, push_notification_config, or an unqualified buyer account id.

async def update_progress(self, task_id: str, progress: dict[str, Any]) ‑> None

Expand source code

async def update_progress(
    self,
    task_id: str,
    progress: dict[str, Any],
) -> None:
    """Write a progress payload; transition ``submitted`` → ``working``.

    Silently no-ops when the task is already in a terminal state or
    unknown — the dispatch wrapper expects this method never to raise on
    transient conditions (see :class:`~adcp.decisioning.TaskRegistry`
    docstring).

    The ``state NOT IN ('completed', 'failed')`` predicate is evaluated
    server-side so a concurrent terminal write cannot be overwritten by a
    straggler progress event.
    """
    async with self._pool.connection() as conn:
        cur = await conn.execute(
            self._sql_update_progress,
            (task_id, json.dumps(progress), time.time()),
        )
        row = await cur.fetchone()
    if row is not None and row[0] == "submitted":
        self._notify_lifecycle_observers("working", self._lifecycle_record(row[1:]))

Write a progress payload; transition submitted → working.

Silently no-ops when the task is already in a terminal state or unknown — the dispatch wrapper expects this method never to raise on transient conditions (see :class:~adcp.decisioning.TaskRegistry docstring).

The state NOT IN ('completed', 'failed') predicate is evaluated server-side so a concurrent terminal write cannot be overwritten by a straggler progress event.

class PostgresTaskRegistry (*,
pool: AsyncConnectionPool,
task_webhook_outbox: PgTaskWebhookOutbox | None = None,
webhook_signing_scope_resolver: WebhookSigningScopeResolver | None = None)

Expand source code

class PgTaskRegistry(_TaskLifecycleObservers):
    """PostgreSQL-backed :class:`~adcp.decisioning.TaskRegistry` — v6.1.

    Durable counterpart to :class:`~adcp.decisioning.InMemoryTaskRegistry`.
    Set ``is_durable = True`` so the production-mode gate in
    :func:`adcp.decisioning.serve.create_adcp_server_from_platform` accepts it
    without requiring ``ADCP_DECISIONING_ALLOW_INMEMORY_TASKS=1``.

    Parameters
    ----------
    pool:
        An :class:`psycopg_pool.AsyncConnectionPool` owned by the caller.
        Each registry operation acquires a short-lived connection from the
        pool and returns it immediately after the query. No long-lived
        transactions, no cross-operation state.
    task_webhook_outbox:
        Optional :class:`PgTaskWebhookOutbox` sharing this pool. When a task
        carries callback registration, ``complete`` / ``fail`` enqueue its
        immutable terminal webhook on the same transaction connection.

    Notes
    -----
    Unlike :class:`~adcp.signing.PgReplayStore`, this class uses a fixed
    ``decisioning_tasks`` table name. Multi-tenant table-name isolation is not
    supported in this release — callers requiring strict schema separation
    should use separate databases or schemas.
    """

    is_durable: ClassVar[bool] = True

    def __init__(
        self,
        *,
        pool: AsyncConnectionPool,
        task_webhook_outbox: PgTaskWebhookOutbox | None = None,
        webhook_signing_scope_resolver: WebhookSigningScopeResolver | None = None,
        _table: str = _DEFAULT_TABLE,
    ) -> None:
        if not PG_AVAILABLE:
            raise ImportError(_INSTALL_HINT)
        if not _SAFE_IDENTIFIER_RE.fullmatch(_table):
            raise ValueError(f"_table must match [a-z_][a-z0-9_]* (ASCII only), got {_table!r}")
        if task_webhook_outbox is not None and task_webhook_outbox._pool is not pool:
            raise ValueError(
                "PgTaskRegistry and PgTaskWebhookOutbox must use the same connection pool"
            )
        uses_sender_resolver = (
            task_webhook_outbox is not None and task_webhook_outbox._sender_resolver is not None
        )
        if uses_sender_resolver != (webhook_signing_scope_resolver is not None):
            raise ValueError(
                "webhook_signing_scope_resolver is required exactly when the "
                "PgTaskWebhookOutbox uses sender_resolver"
            )
        self._init_lifecycle_observers()
        self._pool = pool
        self._table = _table
        self.task_webhook_outbox = task_webhook_outbox
        self._webhook_signing_scope_resolver = webhook_signing_scope_resolver
        self.atomic_task_webhook_outbox = task_webhook_outbox is not None

        # Pre-format queries at construction so the hot path avoids f-strings per call.
        # _table is whitelisted by _SAFE_IDENTIFIER_RE above.
        self._sql_insert = (  # noqa: S608 — table name is whitelisted
            f"INSERT INTO {self._table}"  # nosec B608 — table identifier is constructor-validated
            f" (task_id, account_id, state, task_type, request_context,"
            f"  webhook_registration, webhook_registration_nonce, created_at, updated_at)"
            f" VALUES (%s, %s, 'submitted', %s, %s::jsonb, %s, %s, %s, %s)"
        )
        self._sql_update_progress = (  # noqa: S608
            f"WITH previous AS (SELECT task_id, state FROM {self._table}"  # nosec B608 — table identifier is constructor-validated
            f" WHERE task_id = %s AND state NOT IN ('completed', 'failed') FOR UPDATE)"
            f" UPDATE {self._table} AS task"
            f" SET state = CASE task.state WHEN 'submitted' THEN 'working' ELSE task.state END,"
            f"     progress = %s::jsonb, updated_at = %s"
            f" FROM previous WHERE task.task_id = previous.task_id"
            f" RETURNING previous.state, task.task_id, task.account_id, task.task_type,"
            f" task.created_at, task.updated_at"
        )
        self._sql_complete = (  # noqa: S608
            f"UPDATE {self._table}"  # nosec B608 — table identifier is constructor-validated
            f" SET state = 'completed', result = %s::jsonb, updated_at = %s"
            f" WHERE task_id = %s AND state NOT IN ('completed', 'failed')"
            f" RETURNING task_id, account_id, task_type, webhook_registration,"
            f" webhook_registration_nonce, created_at, updated_at"
        )
        self._sql_fail = (  # noqa: S608
            f"UPDATE {self._table}"  # nosec B608 — table identifier is constructor-validated
            f" SET state = 'failed', error = %s::jsonb, updated_at = %s"
            f" WHERE task_id = %s AND state NOT IN ('completed', 'failed')"
            f" RETURNING task_id, account_id, task_type, webhook_registration,"
            f" webhook_registration_nonce, created_at, updated_at"
        )
        self._sql_clear_webhook_registration = (  # noqa: S608
            f"UPDATE {self._table} SET webhook_registration = NULL,"  # nosec B608 — table identifier is constructor-validated
            f" webhook_registration_nonce = NULL WHERE task_id = %s"
        )
        # Explicit ``::text`` cast on the optional account-filter
        # parameter so psycopg's bind-param type inference doesn't
        # fail with ``IndeterminateDatatype: could not determine
        # data type of parameter $2``. Without the cast, the
        # ``%s IS NULL`` predicate gives psycopg no type context
        # for the parameter and the query fails at prepare time.
        self._sql_get = (  # noqa: S608
            f"SELECT task_id, account_id, state, task_type,"  # nosec B608 — table identifier is constructor-validated
            f"       progress, result, error, request_context, created_at, updated_at"
            f" FROM {self._table}"
            f" WHERE task_id = %s AND (%s::text IS NULL OR account_id = %s)"
        )
        self._sql_get_state_result = (  # noqa: S608
            f"SELECT state, result, task_type FROM {self._table}"  # nosec B608 — validated identifier
            " WHERE task_id = %s"
        )
        self._sql_get_task_type = f"SELECT task_type FROM {self._table} WHERE task_id = %s"  # noqa: S608  # nosec B608 — table identifier is constructor-validated
        self._sql_get_state_error = f"SELECT state, error FROM {self._table} WHERE task_id = %s"  # noqa: S608  # nosec B608 — table identifier is constructor-validated
        self._sql_discard = (  # noqa: S608
            f"DELETE FROM {self._table} WHERE task_id = %s"  # nosec B608 — table identifier is constructor-validated
            f" RETURNING task_id, account_id, task_type, created_at, updated_at"
        )
        self._sql_ddl = (  # noqa: S608
            f"CREATE TABLE IF NOT EXISTS {self._table} ("
            f'    task_id     TEXT             COLLATE "C" NOT NULL PRIMARY KEY,'
            f'    account_id  TEXT             COLLATE "C" NOT NULL,'
            f"    state       TEXT             NOT NULL DEFAULT 'submitted',"
            f"    task_type   TEXT             NOT NULL,"
            f"    progress    JSONB,"
            f"    result      JSONB,"
            f"    error       JSONB,"
            f"    request_context JSONB,"
            f"    webhook_registration BYTEA,"
            f"    webhook_registration_nonce BYTEA,"
            f"    created_at  DOUBLE PRECISION NOT NULL,"
            f"    updated_at  DOUBLE PRECISION NOT NULL"
            f");"
            f"ALTER TABLE {self._table} ADD COLUMN IF NOT EXISTS request_context JSONB;"
            f"ALTER TABLE {self._table} ADD COLUMN IF NOT EXISTS webhook_registration BYTEA;"
            f"ALTER TABLE {self._table} ADD COLUMN IF NOT EXISTS webhook_registration_nonce BYTEA;"
            f"CREATE INDEX IF NOT EXISTS {self._table}_account_idx"  # noqa: S608
            f"    ON {self._table} (account_id);"
        )

    # -- schema bootstrap -----------------------------------------------

    async def create_schema(self) -> None:
        """Create the task registry table and supporting index.

        Honors the ``_table`` kwarg the store was constructed with.
        Idempotent via ``CREATE TABLE IF NOT EXISTS`` — safe to call on every
        application boot. The equivalent raw DDL ships at
        ``adcp/decisioning/pg/decisioning_tasks.sql`` in the installed package
        for adopters using a migration tool (Alembic, Flyway, psql).
        """
        async with self._pool.connection() as conn:
            await conn.execute(self._sql_ddl)

    # -- TaskRegistry Protocol ------------------------------------------

    async def issue(
        self,
        *,
        account_id: str,
        task_type: str,
        request_context: dict[str, Any] | None = None,
        webhook_url: str | None = None,
        webhook_operation_id: str | None = None,
        webhook_token: str | None = None,
        webhook_authentication: TaskWebhookAuthentication | None = None,
        webhook_signing_scope_id: str | None = None,
        **_extra: Any,
    ) -> str:
        """Allocate a task_id, persist a ``submitted`` row, return the id.

        Mirrors :meth:`~adcp.decisioning.InMemoryTaskRegistry.issue` including
        the account_id validation guard — empty or sentinel account_ids would
        allow cross-tenant task-id probing via the ``WHERE account_id = %s``
        predicate collapsing multiple tenants into one slot.
        """
        if not account_id or not account_id.strip() or account_id == "<unset>":
            raise ValueError(
                f"account_id must be a non-empty, non-default string; "
                f"got {account_id!r}. AccountStore.resolve must always "
                "return Account(id=<non-empty>) so cross-tenant cache "
                "scoping works correctly."
            )
        if (webhook_url is None) != (webhook_operation_id is None):
            raise ValueError("webhook_url and webhook_operation_id must be supplied together")
        if webhook_url is not None and not webhook_url:
            raise ValueError("webhook_url must be non-empty when supplied")
        if webhook_operation_id is not None and not webhook_operation_id:
            raise ValueError("webhook_operation_id must be non-empty when supplied")
        if webhook_url is None and webhook_signing_scope_id is not None:
            raise ValueError("webhook_signing_scope_id requires webhook_url")
        if webhook_url is None and webhook_authentication is not None:
            raise ValueError("webhook_authentication requires webhook_url")
        if webhook_authentication is not None and webhook_signing_scope_id is not None:
            raise ValueError(
                "legacy webhook_authentication must not carry an RFC 9421 signing scope"
            )
        outbox = self.task_webhook_outbox
        if webhook_url is not None:
            if outbox is None:
                raise ValueError("webhook registration requires the registry's task_webhook_outbox")
            outbox.validate_registration(webhook_url, webhook_authentication)
        task_id = f"task_{uuid.uuid4().hex[:16]}"
        encrypted_registration: bytes | None = None
        registration_nonce: bytes | None = None
        if webhook_url is not None:
            if webhook_operation_id is None or outbox is None:
                raise RuntimeError("validated webhook registration became incomplete")
            encrypted_registration, registration_nonce = outbox.protect_registration(
                account_id=account_id,
                task_id=task_id,
                task_type=task_type,
                url=webhook_url,
                operation_id=webhook_operation_id,
                token=webhook_token,
                authentication=webhook_authentication,
                signing_scope_id=webhook_signing_scope_id,
            )
        now = time.time()
        async with self._pool.connection() as conn:
            await conn.execute(
                self._sql_insert,
                (
                    task_id,
                    account_id,
                    task_type,
                    json.dumps(request_context) if request_context is not None else None,
                    encrypted_registration,
                    registration_nonce,
                    now,
                    now,
                ),
            )
        self._notify_lifecycle_observers(
            "submitted",
            {
                "task_id": task_id,
                "account_id": account_id,
                "task_type": task_type,
                "created_at": now,
                "updated_at": now,
            },
        )
        return task_id

    async def resolve_webhook_signing_scope(
        self,
        context: RequestContext[Any],
    ) -> str | None:
        """Derive an opaque signing scope from trusted framework context.

        The callback is operator wiring and receives the hydrated
        :class:`RequestContext`. It must use internal tenant/platform metadata,
        never buyer request ``context``, ``push_notification_config``, or an
        unqualified buyer account id.
        """
        resolver = self._webhook_signing_scope_resolver
        if resolver is None:
            return None
        from adcp.decisioning.types import AdcpError

        try:
            value: object = resolver(context)
            if inspect.isawaitable(value):
                value = await value
        except Exception:
            raise AdcpError(
                "INTERNAL_ERROR",
                message="Webhook signing scope resolution failed",
                recovery="terminal",
            ) from None
        if not isinstance(value, str):
            raise AdcpError(
                "INTERNAL_ERROR",
                message="Webhook signing scope resolver returned an invalid value",
                recovery="terminal",
            )
        # Reuse the outbox's bounded opaque-ID validation before any task row
        # is issued. This is trusted server state, but it still crosses a DB
        # and authenticated-envelope boundary.
        if self.task_webhook_outbox is None:
            raise RuntimeError("signing scope resolver requires a task webhook outbox")
        self.task_webhook_outbox._validate_signing_scope_id(value)
        return value

    async def update_progress(
        self,
        task_id: str,
        progress: dict[str, Any],
    ) -> None:
        """Write a progress payload; transition ``submitted`` → ``working``.

        Silently no-ops when the task is already in a terminal state or
        unknown — the dispatch wrapper expects this method never to raise on
        transient conditions (see :class:`~adcp.decisioning.TaskRegistry`
        docstring).

        The ``state NOT IN ('completed', 'failed')`` predicate is evaluated
        server-side so a concurrent terminal write cannot be overwritten by a
        straggler progress event.
        """
        async with self._pool.connection() as conn:
            cur = await conn.execute(
                self._sql_update_progress,
                (task_id, json.dumps(progress), time.time()),
            )
            row = await cur.fetchone()
        if row is not None and row[0] == "submitted":
            self._notify_lifecycle_observers("working", self._lifecycle_record(row[1:]))

    @staticmethod
    def _lifecycle_record(row: tuple[Any, ...]) -> dict[str, Any]:
        return dict(zip(("task_id", "account_id", "task_type", "created_at", "updated_at"), row))

    async def complete(self, task_id: str, result: dict[str, Any]) -> None:
        """Complete once; notify metrics only after the transaction commits."""
        row = await self._complete(task_id, result)
        if row is not None:
            self._notify_lifecycle_observers("completed", self._lifecycle_record(row[:3] + row[5:]))

    async def _complete(
        self,
        task_id: str,
        result: dict[str, Any],
    ) -> tuple[Any, ...] | None:
        """Mark the task ``completed`` with ``result`` as the terminal artifact.

        Idempotent on repeated calls with an equal ``result``; raises
        :class:`ValueError` on conflicting re-completion.

        Uses an atomic ``UPDATE ... RETURNING`` so concurrent workers cannot
        race each other into double-completion without detection.
        """
        async with self._pool.connection() as conn:
            # Do not rely on the pool connection's implicit transaction.
            # Explicitly bind the terminal transition, outbox insert, and
            # callback-registration clear even when adopters configure the
            # pool with autocommit=True.
            async with conn.transaction():
                type_cursor = await conn.execute(self._sql_get_task_type, (task_id,))
                type_row = await type_cursor.fetchone()
                safe_result = (
                    strip_credentials_from_wire_result(type_row[0], result)
                    if type_row is not None
                    else result
                )
                cur = await conn.execute(
                    self._sql_complete,
                    (json.dumps(safe_result), time.time(), task_id),
                )
                row: tuple[Any, ...] | None = await cur.fetchone()
                if row is not None:
                    await self._enqueue_terminal_if_registered(
                        conn,
                        row=row,
                        status="completed",
                        payload=safe_result,
                    )
                    return row  # connection and transaction exit before the caller notifies

                # Zero rows in RETURNING — task is unknown or already terminal.
                cur2 = await conn.execute(self._sql_get_state_result, (task_id,))
                row = await cur2.fetchone()
                if row is None:
                    raise ValueError(f"Task {task_id!r} not found")
                state, existing_result, task_type = row
                safe_result = strip_credentials_from_wire_result(task_type, result)
                if state == "completed":
                    if existing_result == safe_result:
                        return None  # idempotent
                    raise ValueError(f"Task {task_id!r} already completed with a different result")
                raise ValueError(f"Task {task_id!r} already in terminal state {state!r}")

    async def fail(self, task_id: str, error: dict[str, Any]) -> None:
        """Fail once; notify metrics only after the transaction commits."""
        row = await self._fail(task_id, error)
        if row is not None:
            self._notify_lifecycle_observers("failed", self._lifecycle_record(row[:3] + row[5:]))

    async def _fail(
        self,
        task_id: str,
        error: dict[str, Any],
    ) -> tuple[Any, ...] | None:
        """Mark the task ``failed`` with ``error`` as the terminal payload.

        Idempotent on repeated calls with an equal ``error``; raises
        :class:`ValueError` on conflicting re-failure.
        """
        async with self._pool.connection() as conn:
            async with conn.transaction():
                cur = await conn.execute(
                    self._sql_fail,
                    (json.dumps(error), time.time(), task_id),
                )
                row: tuple[Any, ...] | None = await cur.fetchone()
                if row is not None:
                    await self._enqueue_terminal_if_registered(
                        conn,
                        row=row,
                        status="failed",
                        payload=error,
                    )
                    return row  # notify only after connection/transaction exit

                # Zero rows in RETURNING — task is unknown or already terminal.
                cur2 = await conn.execute(self._sql_get_state_error, (task_id,))
                row = await cur2.fetchone()
                if row is None:
                    raise ValueError(f"Task {task_id!r} not found")
                state, existing_error = row
                if state == "failed":
                    if existing_error == error:
                        return None  # idempotent
                    raise ValueError(f"Task {task_id!r} already failed with a different error")
                raise ValueError(f"Task {task_id!r} already in terminal state {state!r}")

    async def get(
        self,
        task_id: str,
        *,
        expected_account_id: str | None = None,
    ) -> dict[str, Any] | None:
        """Look up a task record; cross-tenant probes return ``None``.

        The ``expected_account_id`` predicate is enforced at the SQL level
        (``WHERE account_id = %s``), not as a Python-level filter after fetch.
        This guarantees the row is never materialized for a mismatched probe,
        eliminating the fetch-then-filter anti-pattern.
        """
        async with self._pool.connection() as conn:
            cur = await conn.execute(
                self._sql_get, (task_id, expected_account_id, expected_account_id)
            )
            row = await cur.fetchone()
            if row is None:
                return None
            return {
                "task_id": row[0],
                "account_id": row[1],
                "state": row[2],
                "task_type": row[3],
                "progress": row[4],
                "result": row[5],
                "error": row[6],
                "created_at": row[8],
                "updated_at": row[9],
                **({"context": row[7]} if row[7] is not None else {}),
            }

    async def list(
        self,
        *,
        account_id: str,
        filters: dict[str, Any] | None = None,
        sort: dict[str, Any] | None = None,
        pagination: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        from adcp.decisioning.task_queries import list_task_records

        # Account isolation is a SQL predicate, before rows are materialized.
        async with self._pool.connection() as conn:
            cur = await conn.execute(
                f"SELECT task_id, account_id, state, task_type, progress, result, error,"  # nosec B608 — table identifier is constructor-validated
                f" request_context, created_at, updated_at,"
                f" (webhook_registration IS NOT NULL) FROM {self._table} WHERE account_id = %s",
                (account_id,),
            )
            rows = await cur.fetchall()
        records = [
            dict(
                zip(
                    (
                        "task_id",
                        "account_id",
                        "state",
                        "task_type",
                        "progress",
                        "result",
                        "error",
                        "context",
                        "created_at",
                        "updated_at",
                        "has_webhook",
                    ),
                    row,
                )
            )
            for row in rows
        ]
        return list_task_records(
            records, account_id=account_id, filters=filters, sort=sort, pagination=pagination
        )

    async def _enqueue_terminal_if_registered(
        self,
        conn: Any,
        *,
        row: tuple[Any, ...],
        status: str,
        payload: dict[str, Any],
    ) -> None:
        """Enqueue on ``conn`` so task state and webhook commit atomically."""
        task_id, account_id, task_type, encrypted_registration, registration_nonce = row[:5]
        if encrypted_registration is None:
            return
        if self.task_webhook_outbox is None:
            raise RuntimeError(
                "Task carries push_notification_config but PgTaskRegistry has no "
                "task_webhook_outbox; refusing a non-atomic terminal transition"
            )
        if registration_nonce is None:
            raise RuntimeError(f"Task {task_id!r} has incomplete webhook registration")
        url, operation_id, token, authentication, signing_scope_id = (
            self.task_webhook_outbox._open_registration_with_scope(
                account_id=account_id,
                task_id=task_id,
                task_type=task_type,
                encrypted_registration=bytes(encrypted_registration),
                nonce=bytes(registration_nonce),
            )
        )
        await self.task_webhook_outbox.enqueue_terminal(
            conn,
            task_id=task_id,
            account_id=account_id,
            task_type=task_type,
            status=status,
            result=payload,
            url=url,
            operation_id=operation_id,
            token=token,
            authentication=authentication,
            signing_scope_id=signing_scope_id,
        )
        # The encrypted outbox envelope now owns the callback registration.
        # Clear the task-row copy in this same transaction.
        await conn.execute(self._sql_clear_webhook_registration, (task_id,))

    async def discard(self, task_id: str) -> None:
        """Remove a task_id from the registry — rollback path.

        Idempotent: discarding an unknown task_id is a no-op (no raise),
        matching the :class:`~adcp.decisioning.InMemoryTaskRegistry` contract.
        """
        async with self._pool.connection() as conn:
            cur = await conn.execute(self._sql_discard, (task_id,))
            row = await cur.fetchone()
        if row is not None:
            event_record = self._lifecycle_record(row)
            event_record["updated_at"] = time.time()
            self._notify_lifecycle_observers("discarded", event_record)

PostgreSQL-backed :class:~adcp.decisioning.TaskRegistry — v6.1.

Durable counterpart to :class:~adcp.decisioning.InMemoryTaskRegistry. Set is_durable = True so the production-mode gate in :func:adcp.decisioning.serve.create_adcp_server_from_platform accepts it without requiring ADCP_DECISIONING_ALLOW_INMEMORY_TASKS=1.

Parameters

pool: An :class:psycopg_pool.AsyncConnectionPool owned by the caller. Each registry operation acquires a short-lived connection from the pool and returns it immediately after the query. No long-lived transactions, no cross-operation state. task_webhook_outbox: Optional :class:PgTaskWebhookOutbox sharing this pool. When a task carries callback registration, complete / fail enqueue its immutable terminal webhook on the same transaction connection.

Notes

Unlike :class:~adcp.signing.PgReplayStore, this class uses a fixed decisioning_tasks table name. Multi-tenant table-name isolation is not supported in this release — callers requiring strict schema separation should use separate databases or schemas.

Ancestors

• adcp.decisioning.task_registry._TaskLifecycleObservers

Class variables

var is_durable : ClassVar[bool]

Methods

async def complete(self, task_id: str, result: dict[str, Any]) ‑> None

Expand source code

async def complete(self, task_id: str, result: dict[str, Any]) -> None:
    """Complete once; notify metrics only after the transaction commits."""
    row = await self._complete(task_id, result)
    if row is not None:
        self._notify_lifecycle_observers("completed", self._lifecycle_record(row[:3] + row[5:]))

Complete once; notify metrics only after the transaction commits.

async def create_schema(self) ‑> None

Expand source code

async def create_schema(self) -> None:
    """Create the task registry table and supporting index.

    Honors the ``_table`` kwarg the store was constructed with.
    Idempotent via ``CREATE TABLE IF NOT EXISTS`` — safe to call on every
    application boot. The equivalent raw DDL ships at
    ``adcp/decisioning/pg/decisioning_tasks.sql`` in the installed package
    for adopters using a migration tool (Alembic, Flyway, psql).
    """
    async with self._pool.connection() as conn:
        await conn.execute(self._sql_ddl)

Create the task registry table and supporting index.

Honors the _table kwarg the store was constructed with. Idempotent via CREATE TABLE IF NOT EXISTS — safe to call on every application boot. The equivalent raw DDL ships at adcp/decisioning/pg/decisioning_tasks.sql in the installed package for adopters using a migration tool (Alembic, Flyway, psql).

async def discard(self, task_id: str) ‑> None

Expand source code

async def discard(self, task_id: str) -> None:
    """Remove a task_id from the registry — rollback path.

    Idempotent: discarding an unknown task_id is a no-op (no raise),
    matching the :class:`~adcp.decisioning.InMemoryTaskRegistry` contract.
    """
    async with self._pool.connection() as conn:
        cur = await conn.execute(self._sql_discard, (task_id,))
        row = await cur.fetchone()
    if row is not None:
        event_record = self._lifecycle_record(row)
        event_record["updated_at"] = time.time()
        self._notify_lifecycle_observers("discarded", event_record)

Remove a task_id from the registry — rollback path.

Idempotent: discarding an unknown task_id is a no-op (no raise), matching the :class:~adcp.decisioning.InMemoryTaskRegistry contract.

async def fail(self, task_id: str, error: dict[str, Any]) ‑> None

Expand source code

async def fail(self, task_id: str, error: dict[str, Any]) -> None:
    """Fail once; notify metrics only after the transaction commits."""
    row = await self._fail(task_id, error)
    if row is not None:
        self._notify_lifecycle_observers("failed", self._lifecycle_record(row[:3] + row[5:]))

Fail once; notify metrics only after the transaction commits.

async def get(self, task_id: str, *, expected_account_id: str | None = None) ‑> dict[str, typing.Any] | None

Expand source code

async def get(
    self,
    task_id: str,
    *,
    expected_account_id: str | None = None,
) -> dict[str, Any] | None:
    """Look up a task record; cross-tenant probes return ``None``.

    The ``expected_account_id`` predicate is enforced at the SQL level
    (``WHERE account_id = %s``), not as a Python-level filter after fetch.
    This guarantees the row is never materialized for a mismatched probe,
    eliminating the fetch-then-filter anti-pattern.
    """
    async with self._pool.connection() as conn:
        cur = await conn.execute(
            self._sql_get, (task_id, expected_account_id, expected_account_id)
        )
        row = await cur.fetchone()
        if row is None:
            return None
        return {
            "task_id": row[0],
            "account_id": row[1],
            "state": row[2],
            "task_type": row[3],
            "progress": row[4],
            "result": row[5],
            "error": row[6],
            "created_at": row[8],
            "updated_at": row[9],
            **({"context": row[7]} if row[7] is not None else {}),
        }

Look up a task record; cross-tenant probes return None.

The expected_account_id predicate is enforced at the SQL level (WHERE account_id = %s), not as a Python-level filter after fetch. This guarantees the row is never materialized for a mismatched probe, eliminating the fetch-then-filter anti-pattern.

async def issue(self,
*,
account_id: str,
task_type: str,
request_context: dict[str, Any] | None = None,
webhook_url: str | None = None,
webhook_operation_id: str | None = None,
webhook_token: str | None = None,
webhook_authentication: TaskWebhookAuthentication | None = None,
webhook_signing_scope_id: str | None = None,
**_extra: Any) ‑> str

Expand source code

async def issue(
    self,
    *,
    account_id: str,
    task_type: str,
    request_context: dict[str, Any] | None = None,
    webhook_url: str | None = None,
    webhook_operation_id: str | None = None,
    webhook_token: str | None = None,
    webhook_authentication: TaskWebhookAuthentication | None = None,
    webhook_signing_scope_id: str | None = None,
    **_extra: Any,
) -> str:
    """Allocate a task_id, persist a ``submitted`` row, return the id.

    Mirrors :meth:`~adcp.decisioning.InMemoryTaskRegistry.issue` including
    the account_id validation guard — empty or sentinel account_ids would
    allow cross-tenant task-id probing via the ``WHERE account_id = %s``
    predicate collapsing multiple tenants into one slot.
    """
    if not account_id or not account_id.strip() or account_id == "<unset>":
        raise ValueError(
            f"account_id must be a non-empty, non-default string; "
            f"got {account_id!r}. AccountStore.resolve must always "
            "return Account(id=<non-empty>) so cross-tenant cache "
            "scoping works correctly."
        )
    if (webhook_url is None) != (webhook_operation_id is None):
        raise ValueError("webhook_url and webhook_operation_id must be supplied together")
    if webhook_url is not None and not webhook_url:
        raise ValueError("webhook_url must be non-empty when supplied")
    if webhook_operation_id is not None and not webhook_operation_id:
        raise ValueError("webhook_operation_id must be non-empty when supplied")
    if webhook_url is None and webhook_signing_scope_id is not None:
        raise ValueError("webhook_signing_scope_id requires webhook_url")
    if webhook_url is None and webhook_authentication is not None:
        raise ValueError("webhook_authentication requires webhook_url")
    if webhook_authentication is not None and webhook_signing_scope_id is not None:
        raise ValueError(
            "legacy webhook_authentication must not carry an RFC 9421 signing scope"
        )
    outbox = self.task_webhook_outbox
    if webhook_url is not None:
        if outbox is None:
            raise ValueError("webhook registration requires the registry's task_webhook_outbox")
        outbox.validate_registration(webhook_url, webhook_authentication)
    task_id = f"task_{uuid.uuid4().hex[:16]}"
    encrypted_registration: bytes | None = None
    registration_nonce: bytes | None = None
    if webhook_url is not None:
        if webhook_operation_id is None or outbox is None:
            raise RuntimeError("validated webhook registration became incomplete")
        encrypted_registration, registration_nonce = outbox.protect_registration(
            account_id=account_id,
            task_id=task_id,
            task_type=task_type,
            url=webhook_url,
            operation_id=webhook_operation_id,
            token=webhook_token,
            authentication=webhook_authentication,
            signing_scope_id=webhook_signing_scope_id,
        )
    now = time.time()
    async with self._pool.connection() as conn:
        await conn.execute(
            self._sql_insert,
            (
                task_id,
                account_id,
                task_type,
                json.dumps(request_context) if request_context is not None else None,
                encrypted_registration,
                registration_nonce,
                now,
                now,
            ),
        )
    self._notify_lifecycle_observers(
        "submitted",
        {
            "task_id": task_id,
            "account_id": account_id,
            "task_type": task_type,
            "created_at": now,
            "updated_at": now,
        },
    )
    return task_id

Allocate a task_id, persist a submitted row, return the id.

Mirrors :meth:~adcp.decisioning.InMemoryTaskRegistry.issue including the account_id validation guard — empty or sentinel account_ids would allow cross-tenant task-id probing via the WHERE account_id = %s predicate collapsing multiple tenants into one slot.

async def list(self,
*,
account_id: str,
filters: dict[str, Any] | None = None,
sort: dict[str, Any] | None = None,
pagination: dict[str, Any] | None = None) ‑> dict[str, typing.Any]

Expand source code

async def list(
    self,
    *,
    account_id: str,
    filters: dict[str, Any] | None = None,
    sort: dict[str, Any] | None = None,
    pagination: dict[str, Any] | None = None,
) -> dict[str, Any]:
    from adcp.decisioning.task_queries import list_task_records

    # Account isolation is a SQL predicate, before rows are materialized.
    async with self._pool.connection() as conn:
        cur = await conn.execute(
            f"SELECT task_id, account_id, state, task_type, progress, result, error,"  # nosec B608 — table identifier is constructor-validated
            f" request_context, created_at, updated_at,"
            f" (webhook_registration IS NOT NULL) FROM {self._table} WHERE account_id = %s",
            (account_id,),
        )
        rows = await cur.fetchall()
    records = [
        dict(
            zip(
                (
                    "task_id",
                    "account_id",
                    "state",
                    "task_type",
                    "progress",
                    "result",
                    "error",
                    "context",
                    "created_at",
                    "updated_at",
                    "has_webhook",
                ),
                row,
            )
        )
        for row in rows
    ]
    return list_task_records(
        records, account_id=account_id, filters=filters, sort=sort, pagination=pagination
    )

async def resolve_webhook_signing_scope(self, context: RequestContext[Any]) ‑> str | None

Expand source code

async def resolve_webhook_signing_scope(
    self,
    context: RequestContext[Any],
) -> str | None:
    """Derive an opaque signing scope from trusted framework context.

    The callback is operator wiring and receives the hydrated
    :class:`RequestContext`. It must use internal tenant/platform metadata,
    never buyer request ``context``, ``push_notification_config``, or an
    unqualified buyer account id.
    """
    resolver = self._webhook_signing_scope_resolver
    if resolver is None:
        return None
    from adcp.decisioning.types import AdcpError

    try:
        value: object = resolver(context)
        if inspect.isawaitable(value):
            value = await value
    except Exception:
        raise AdcpError(
            "INTERNAL_ERROR",
            message="Webhook signing scope resolution failed",
            recovery="terminal",
        ) from None
    if not isinstance(value, str):
        raise AdcpError(
            "INTERNAL_ERROR",
            message="Webhook signing scope resolver returned an invalid value",
            recovery="terminal",
        )
    # Reuse the outbox's bounded opaque-ID validation before any task row
    # is issued. This is trusted server state, but it still crosses a DB
    # and authenticated-envelope boundary.
    if self.task_webhook_outbox is None:
        raise RuntimeError("signing scope resolver requires a task webhook outbox")
    self.task_webhook_outbox._validate_signing_scope_id(value)
    return value

Derive an opaque signing scope from trusted framework context.

The callback is operator wiring and receives the hydrated :class:RequestContext. It must use internal tenant/platform metadata, never buyer request context, push_notification_config, or an unqualified buyer account id.

async def update_progress(self, task_id: str, progress: dict[str, Any]) ‑> None

Expand source code

async def update_progress(
    self,
    task_id: str,
    progress: dict[str, Any],
) -> None:
    """Write a progress payload; transition ``submitted`` → ``working``.

    Silently no-ops when the task is already in a terminal state or
    unknown — the dispatch wrapper expects this method never to raise on
    transient conditions (see :class:`~adcp.decisioning.TaskRegistry`
    docstring).

    The ``state NOT IN ('completed', 'failed')`` predicate is evaluated
    server-side so a concurrent terminal write cannot be overwritten by a
    straggler progress event.
    """
    async with self._pool.connection() as conn:
        cur = await conn.execute(
            self._sql_update_progress,
            (task_id, json.dumps(progress), time.time()),
        )
        row = await cur.fetchone()
    if row is not None and row[0] == "submitted":
        self._notify_lifecycle_observers("working", self._lifecycle_record(row[1:]))

Write a progress payload; transition submitted → working.

Silently no-ops when the task is already in a terminal state or unknown — the dispatch wrapper expects this method never to raise on transient conditions (see :class:~adcp.decisioning.TaskRegistry docstring).

The state NOT IN ('completed', 'failed') predicate is evaluated server-side so a concurrent terminal write cannot be overwritten by a straggler progress event.

class PgTaskWebhookOutbox (*,
pool: AsyncConnectionPool,
sender: WebhookSender | None = None,
sender_resolver: WebhookSenderResolver | None = None,
encryption_key: bytes,
delivery_retry_horizon_seconds: int,
retry: RetryPolicy | None = None,
lease_seconds: int = 60,
legacy_hmac_fallback: bool = False,
legacy_allowed_destination_ports: frozenset[int] | None = None,
legacy_transport_hooks: tuple[TransportHook, ...] | None = None,
table: str = 'adcp_task_webhook_outbox')

Expand source code

class PgTaskWebhookOutbox:
    """Atomic task-webhook outbox and lease-based delivery worker.

    Construct this with the same pool as :class:`PgTaskRegistry`, then pass it
    to ``PgTaskRegistry(..., task_webhook_outbox=outbox)``. Call
    :meth:`create_schema` during migration/startup and run at least one
    :meth:`run_worker` loop in every deployment.
    """

    delivery_state_is_durable: ClassVar[bool] = True
    supports_atomic_task_outbox: ClassVar[bool] = True

    def __init__(
        self,
        *,
        pool: AsyncConnectionPool,
        sender: WebhookSender | None = None,
        sender_resolver: WebhookSenderResolver | None = None,
        encryption_key: bytes,
        delivery_retry_horizon_seconds: int,
        retry: RetryPolicy | None = None,
        lease_seconds: int = 60,
        legacy_hmac_fallback: bool = False,
        legacy_allowed_destination_ports: frozenset[int] | None = None,
        legacy_transport_hooks: tuple[TransportHook, ...] | None = None,
        table: str = DEFAULT_TABLE,
    ) -> None:
        if not PG_AVAILABLE:
            raise ImportError(_INSTALL_HINT)
        if (sender is None) == (sender_resolver is None):
            raise ValueError("pass exactly one of sender or sender_resolver")
        if sender_resolver is not None:
            resolver_method = getattr(sender_resolver, "resolve", None)
            if not callable(resolver_method):
                is_async_resolver = False
            else:
                try:
                    is_async_resolver = inspect.iscoroutinefunction(inspect.unwrap(resolver_method))
                except ValueError:
                    is_async_resolver = False
            if not is_async_resolver:
                raise ValueError("sender_resolver must define async resolve(signing_scope_id)")
        if len(encryption_key) != 32:
            raise ValueError("encryption_key must be exactly 32 bytes for AES-256-GCM")
        if sender is not None:
            self._validate_delivery_sender(sender)
        if type(delivery_retry_horizon_seconds) is not int or not (
            MIN_RETRY_HORIZON_SECONDS <= delivery_retry_horizon_seconds <= MAX_RETRY_HORIZON_SECONDS
        ):
            raise ValueError(
                "delivery_retry_horizon_seconds must be an integer from "
                f"{MIN_RETRY_HORIZON_SECONDS} through {MAX_RETRY_HORIZON_SECONDS}"
            )
        if type(lease_seconds) is not int or lease_seconds <= 1:
            raise ValueError("lease_seconds must be an integer greater than 1")
        if type(legacy_hmac_fallback) is not bool:
            raise ValueError("legacy_hmac_fallback must be a bool")
        sender_timeout = float(getattr(sender, "_timeout", 0.0)) if sender is not None else 0.0
        if sender is not None and lease_seconds < sender_timeout + 5:
            raise ValueError(
                "lease_seconds must exceed the sender HTTP timeout by at least 5 seconds"
            )
        resolved_retry = retry or RetryPolicy()
        if (
            resolved_retry.base_delay_seconds <= 0
            or resolved_retry.max_delay_seconds <= 0
            or resolved_retry.max_delay_seconds < resolved_retry.base_delay_seconds
        ):
            raise ValueError(
                "retry delays must be positive and max_delay_seconds must be at least "
                "base_delay_seconds"
            )
        if not _SAFE_IDENTIFIER_RE.fullmatch(table):
            raise ValueError(
                f"table must match [a-z_][a-z0-9_]{{0,44}} (ASCII only), got {table!r}"
            )

        self._pool = pool
        self._sender = sender
        self._sender_resolver = sender_resolver
        self._cipher = AESGCM(encryption_key)
        self.delivery_retry_horizon_seconds = delivery_retry_horizon_seconds
        self._retry = resolved_retry
        self._lease_seconds = lease_seconds
        self.legacy_hmac_fallback = legacy_hmac_fallback
        self._legacy_allowed_destination_ports = (
            legacy_allowed_destination_ports
            if legacy_allowed_destination_ports is not None
            else getattr(sender, "_allowed_destination_ports", None)
        )
        self._legacy_transport_hooks = (
            legacy_transport_hooks
            if legacy_transport_hooks is not None
            else tuple(getattr(sender, "_transport_hooks", ()))
        )
        self._table = table
        self._worker_started = False

        self._sql_insert = (  # noqa: S608
            f"INSERT INTO {table} ("
            "task_id, task_type, terminal_status, url, operation_id, "
            "idempotency_key, signing_scope_id, account_id, encrypted_body, envelope_nonce, "
            "retry_horizon_seconds"
            ") VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) RETURNING id"
        )
        self._sql_expire = (  # noqa: S608
            f"WITH expired AS (SELECT id FROM {table}"
            " WHERE state IN ('pending', 'in_flight') AND retry_until <= now()"
            " ORDER BY id FOR UPDATE SKIP LOCKED LIMIT 1000)"
            f" UPDATE {table} AS outbox SET state = 'expired', lease_token = NULL,"
            " lease_expires_at = NULL, updated_at = now()"
            " FROM expired WHERE outbox.id = expired.id"
        )
        self._sql_claim = (  # noqa: S608
            f"WITH candidate AS ("
            f" SELECT id FROM {table}"
            " WHERE (retry_until IS NULL OR retry_until > now()) AND ("
            "   (state = 'pending' AND available_at <= now()) OR"
            "   (state = 'in_flight' AND lease_expires_at <= now())"
            " ) ORDER BY available_at, id FOR UPDATE SKIP LOCKED LIMIT 1"
            f") UPDATE {table} AS outbox SET"
            " state = 'in_flight', lease_token = %s,"
            " lease_expires_at = now() + (%s * interval '1 second'),"
            " first_attempt_at = COALESCE(outbox.first_attempt_at, now()),"
            " retry_until = COALESCE("
            "   outbox.retry_until, now() + (outbox.retry_horizon_seconds * interval '1 second')"
            " ),"
            " attempt_count = outbox.attempt_count + 1, updated_at = now()"
            " FROM candidate WHERE outbox.id = candidate.id"
            " RETURNING outbox.id, outbox.account_id, outbox.task_id, outbox.task_type,"
            " outbox.terminal_status, outbox.url, outbox.operation_id,"
            " outbox.idempotency_key, outbox.signing_scope_id,"
            " outbox.encrypted_body, outbox.envelope_nonce,"
            " outbox.attempt_count"
        )
        self._sql_ack = (  # noqa: S608
            f"UPDATE {table} SET state = 'delivered', delivered_at = now(),"
            " lease_token = NULL, lease_expires_at = NULL,"
            " last_http_status = %s, last_error = NULL, updated_at = now()"
            " WHERE id = %s AND state = 'in_flight' AND lease_token = %s"
        )
        self._sql_release = (  # noqa: S608
            f"UPDATE {table} SET"
            " state = CASE WHEN retry_until <= now() THEN 'expired' ELSE 'pending' END,"
            " available_at = CASE WHEN retry_until <= now() THEN available_at"
            "                     ELSE now() + (%s * interval '1 second') END,"
            " lease_token = NULL, lease_expires_at = NULL,"
            " last_http_status = %s, last_error = %s, updated_at = now()"
            " WHERE id = %s AND state = 'in_flight' AND lease_token = %s"
        )
        self._sql_quarantine = (  # noqa: S608
            f"UPDATE {table} SET state = 'invalid', lease_token = NULL,"
            " lease_expires_at = NULL, last_error = %s, updated_at = now()"
            " WHERE id = %s AND state = 'in_flight' AND lease_token = %s"
        )
        self._sql_purge = (  # noqa: S608
            f"DELETE FROM {table} WHERE id IN ("
            f" SELECT id FROM {table} WHERE retry_until <= now()"
            " AND state IN ('delivered', 'expired', 'invalid')"
            " ORDER BY id LIMIT 1000"
            ")"
        )

    async def create_schema(self) -> None:
        """Create the outbox table and work index idempotently."""
        statements = [
            f"""CREATE TABLE IF NOT EXISTS {self._table} (
                id                 BIGSERIAL PRIMARY KEY,
                task_id            TEXT COLLATE "C" NOT NULL UNIQUE,
                account_id         TEXT COLLATE "C" NOT NULL,
                task_type          TEXT NOT NULL,
                terminal_status    TEXT NOT NULL,
                url                TEXT NOT NULL,
                operation_id       TEXT NOT NULL,
                idempotency_key    TEXT COLLATE "C" NOT NULL UNIQUE,
                signing_scope_id   TEXT COLLATE "C",
                encrypted_body     BYTEA NOT NULL,
                envelope_nonce     BYTEA NOT NULL,
                state              TEXT NOT NULL DEFAULT 'pending',
                attempt_count      INTEGER NOT NULL DEFAULT 0,
                available_at       TIMESTAMPTZ NOT NULL DEFAULT now(),
                lease_token        TEXT COLLATE "C",
                lease_expires_at   TIMESTAMPTZ,
                created_at         TIMESTAMPTZ NOT NULL DEFAULT now(),
                updated_at         TIMESTAMPTZ NOT NULL DEFAULT now(),
                first_attempt_at   TIMESTAMPTZ,
                retry_until        TIMESTAMPTZ,
                retry_horizon_seconds INTEGER NOT NULL,
                delivered_at       TIMESTAMPTZ,
                last_http_status   INTEGER,
                last_error         TEXT,
                CHECK (state IN ('pending', 'in_flight', 'delivered', 'expired', 'invalid')),
                CHECK (terminal_status IN ('completed', 'failed')),
                CHECK (attempt_count >= 0),
                CHECK (octet_length(envelope_nonce) = 12),
                CHECK (retry_horizon_seconds BETWEEN 86400 AND 604800),
                CHECK ((first_attempt_at IS NULL) = (retry_until IS NULL)),
                CHECK (retry_until IS NULL OR retry_until > first_attempt_at)
            )""",
            f'''ALTER TABLE {self._table}
                ADD COLUMN IF NOT EXISTS signing_scope_id TEXT COLLATE "C"''',
            f"""CREATE INDEX IF NOT EXISTS {self._table}_work_idx
                ON {self._table} (available_at, id)
                WHERE state IN ('pending', 'in_flight')""",
            f"""CREATE INDEX IF NOT EXISTS {self._table}_retry_until_idx
                ON {self._table} (retry_until)""",
        ]
        async with self._pool.connection() as conn:
            for statement in statements:
                await conn.execute(statement)

    async def enqueue_terminal(
        self,
        conn: Any,
        *,
        task_id: str,
        account_id: str,
        task_type: str,
        status: str,
        result: dict[str, Any],
        url: str,
        operation_id: str,
        token: str | None,
        authentication: TaskWebhookAuthentication | None = None,
        signing_scope_id: str | None = None,
    ) -> int:
        """Insert a terminal webhook using the caller's open transaction."""
        if status not in {"completed", "failed"}:
            raise ValueError(f"terminal webhook status must be completed or failed, got {status!r}")
        self._validate_authentication(authentication)
        self._validate_scope_for_mode(
            signing_scope_id,
            authentication=authentication,
            require_resolver_scope=False,
        )
        preparer = self._sender or WebhookSender
        prepared = preparer.prepare_mcp(
            url=url,
            task_id=task_id,
            task_type=task_type,
            status=status,
            result=result,
            operation_id=operation_id,
            token=token,
        )
        self._validate_callback_url(prepared.url)
        nonce = os.urandom(12)
        aad = self._envelope_aad(
            account_id=account_id,
            task_id=task_id,
            task_type=task_type,
            status=status,
            url=prepared.url,
            operation_id=operation_id,
            idempotency_key=prepared.idempotency_key,
            signing_scope_id=signing_scope_id,
        )
        protected_body = self._protect_delivery_body(prepared.body, authentication)
        encrypted_body = self._cipher.encrypt(nonce, protected_body, aad)
        cursor = await conn.execute(
            self._sql_insert,
            (
                task_id,
                task_type,
                status,
                prepared.url,
                operation_id,
                prepared.idempotency_key,
                signing_scope_id,
                account_id,
                encrypted_body,
                nonce,
                self.delivery_retry_horizon_seconds,
            ),
        )
        row = await cursor.fetchone()
        if row is None:
            raise RuntimeError("task webhook outbox insert returned no id")
        return int(row[0])

    def validate_registration(
        self,
        url: str,
        authentication: TaskWebhookAuthentication | None = None,
    ) -> None:
        """Validate callback syntax before a task is accepted as Submitted."""
        self._validate_callback_url(url)
        self._validate_authentication(authentication)

    def protect_registration(
        self,
        *,
        account_id: str,
        task_id: str,
        task_type: str,
        url: str,
        operation_id: str,
        token: str | None,
        authentication: TaskWebhookAuthentication | None = None,
        signing_scope_id: str | None = None,
    ) -> tuple[bytes, bytes]:
        """Encrypt and authenticate callback registration at task issue time."""
        self._validate_callback_url(url)
        self._validate_authentication(authentication)
        self._validate_scope_for_mode(
            signing_scope_id,
            authentication=authentication,
            require_resolver_scope=True,
        )
        nonce = os.urandom(12)
        plaintext = json.dumps(
            {
                "url": url,
                "operation_id": operation_id,
                "token": token,
                "authentication": (
                    {
                        "scheme": authentication.scheme,
                        "credentials": authentication.credentials,
                    }
                    if authentication is not None
                    else None
                ),
                "signing_scope_id": signing_scope_id,
            },
            ensure_ascii=False,
            separators=(",", ":"),
        ).encode("utf-8")
        return (
            self._cipher.encrypt(
                nonce,
                plaintext,
                self._registration_aad(
                    account_id=account_id,
                    task_id=task_id,
                    task_type=task_type,
                ),
            ),
            nonce,
        )

    def open_registration(
        self,
        *,
        account_id: str,
        task_id: str,
        task_type: str,
        encrypted_registration: bytes,
        nonce: bytes,
    ) -> tuple[str, str, str | None]:
        """Verify and decrypt a callback registration.

        The three-item return shape is retained for compatibility. Durable
        registry dispatch uses the private scope-aware decoder below.
        """
        url, operation_id, token, _authentication, _signing_scope_id = (
            self._open_registration_with_scope(
                account_id=account_id,
                task_id=task_id,
                task_type=task_type,
                encrypted_registration=encrypted_registration,
                nonce=nonce,
            )
        )
        return url, operation_id, token

    def _open_registration_with_scope(
        self,
        *,
        account_id: str,
        task_id: str,
        task_type: str,
        encrypted_registration: bytes,
        nonce: bytes,
    ) -> tuple[
        str,
        str,
        str | None,
        TaskWebhookAuthentication | None,
        str | None,
    ]:
        """Verify and decrypt callback registration with its trusted scope."""
        try:
            plaintext = self._cipher.decrypt(
                nonce,
                encrypted_registration,
                self._registration_aad(
                    account_id=account_id,
                    task_id=task_id,
                    task_type=task_type,
                ),
            )
            value = json.loads(plaintext)
        except (InvalidTag, json.JSONDecodeError, UnicodeDecodeError) as exc:
            raise ValueError("task webhook registration failed authenticated decryption") from exc
        if not isinstance(value, dict):
            raise ValueError("task webhook registration must decrypt to an object")
        url = value.get("url")
        operation_id = value.get("operation_id")
        token = value.get("token")
        authentication_value = value.get("authentication")
        signing_scope_id = value.get("signing_scope_id")
        if not isinstance(url, str) or not isinstance(operation_id, str):
            raise ValueError("task webhook registration has invalid URL or operation_id")
        if token is not None and not isinstance(token, str):
            raise ValueError("task webhook registration token must be a string or null")
        if signing_scope_id is not None and not isinstance(signing_scope_id, str):
            raise ValueError("task webhook registration signing scope must be a string or null")
        authentication: TaskWebhookAuthentication | None = None
        if authentication_value is not None:
            if not isinstance(authentication_value, dict):
                raise ValueError("task webhook registration authentication must be an object")
            scheme = authentication_value.get("scheme")
            credentials = authentication_value.get("credentials")
            if not isinstance(scheme, str) or not isinstance(credentials, str):
                raise ValueError("task webhook registration authentication is invalid")
            try:
                authentication = TaskWebhookAuthentication(
                    scheme=scheme,
                    credentials=credentials,
                )
            except (TypeError, ValueError) as exc:
                raise ValueError("task webhook registration authentication is invalid") from exc
        self._validate_callback_url(url)
        self._validate_authentication(authentication)
        self._validate_scope_for_mode(
            signing_scope_id,
            authentication=authentication,
            require_resolver_scope=False,
        )
        return url, operation_id, token, authentication, signing_scope_id

    async def run_worker(
        self,
        *,
        poll_interval: float = 1.0,
        purge_interval: float = 300.0,
    ) -> None:
        """Continuously publish eligible rows until the task is cancelled."""
        if poll_interval <= 0:
            raise ValueError("poll_interval must be positive")
        if purge_interval <= 0:
            raise ValueError("purge_interval must be positive")
        self._worker_started = True
        next_purge = 0.0
        try:
            while True:
                try:
                    now = time.monotonic()
                    if now >= next_purge:
                        await self.purge_expired()
                        next_purge = now + purge_interval
                    processed = await self.process_one()
                except asyncio.CancelledError:
                    raise
                except Exception:
                    logger.exception("[adcp.task_webhook_outbox] worker iteration failed; retrying")
                    await asyncio.sleep(poll_interval)
                    continue
                if not processed:
                    await asyncio.sleep(poll_interval)
        finally:
            self._worker_started = False

    async def process_one(self) -> bool:
        """Claim and attempt one delivery; return ``False`` when idle."""
        lease_token = uuid.uuid4().hex
        async with self._pool.connection() as conn:
            await conn.execute(self._sql_expire)
            cursor = await conn.execute(
                self._sql_claim,
                (lease_token, self._lease_seconds),
            )
            row = await cursor.fetchone()
        if row is None:
            return False

        (
            row_id,
            account_id,
            task_id,
            task_type,
            status,
            url,
            operation_id,
            idempotency_key,
            signing_scope_id,
            encrypted_body,
            nonce,
            attempt_count,
        ) = row
        aad = self._envelope_aad(
            account_id=str(account_id),
            task_id=str(task_id),
            task_type=str(task_type),
            status=str(status),
            url=str(url),
            operation_id=str(operation_id),
            idempotency_key=str(idempotency_key),
            signing_scope_id=(str(signing_scope_id) if signing_scope_id is not None else None),
        )
        try:
            protected_body = self._cipher.decrypt(bytes(nonce), bytes(encrypted_body), aad)
            body_bytes, authentication = self._open_delivery_body(protected_body)
            self._validate_stored_body(
                body_bytes,
                task_id=str(task_id),
                task_type=str(task_type),
                status=str(status),
                operation_id=str(operation_id),
                idempotency_key=str(idempotency_key),
            )
        except (InvalidTag, ValueError, json.JSONDecodeError, UnicodeDecodeError):
            error_message = (
                "stored webhook envelope failed authenticated binding verification; "
                "row quarantined without delivery"
            )
            async with self._pool.connection() as conn:
                await conn.execute(
                    self._sql_quarantine,
                    (error_message, row_id, lease_token),
                )
            logger.error(
                "[adcp.task_webhook_outbox] integrity failure for task %s; row quarantined",
                task_id,
            )
            return True
        prepared = PreparedWebhook(
            url=str(url),
            idempotency_key=str(idempotency_key),
            body=body_bytes,
        )
        delivery: WebhookDeliveryResult | None = None
        error: BaseException | None = None
        try:
            delivery = await asyncio.wait_for(
                self._deliver_prepared(
                    prepared,
                    str(signing_scope_id) if signing_scope_id is not None else None,
                    authentication,
                ),
                timeout=self._lease_seconds - 1,
            )
        except ScopePermanentlyUnknown:
            await self._quarantine_permanent_delivery_error(
                row_id=row_id,
                lease_token=lease_token,
                task_id=str(task_id),
                error=ValueError("webhook signing scope is permanently unavailable"),
            )
            return True
        except ScopeTransientlyUnavailable:
            error = RuntimeError("webhook signing scope is temporarily unavailable")
        except SSRFValidationError as exc:
            if exc.transient:
                error = exc
            else:
                await self._quarantine_permanent_delivery_error(
                    row_id=row_id,
                    lease_token=lease_token,
                    task_id=str(task_id),
                    error=exc,
                )
                return True
        except ValueError as exc:
            await self._quarantine_permanent_delivery_error(
                row_id=row_id,
                lease_token=lease_token,
                task_id=str(task_id),
                error=exc,
            )
            return True
        except Exception as exc:
            error = exc

        if delivery is not None and delivery.ok:
            async with self._pool.connection() as conn:
                await conn.execute(
                    self._sql_ack,
                    (delivery.status_code, row_id, lease_token),
                )
            return True

        if delivery is not None and not self._is_retryable_http_status(delivery.status_code):
            # Receiver-controlled bodies can echo Authorization/signature
            # material. Never persist them in the plaintext last_error column.
            error_message = f"permanent HTTP {delivery.status_code}"
            async with self._pool.connection() as conn:
                await conn.execute(
                    self._sql_quarantine,
                    (error_message[:1000], row_id, lease_token),
                )
            logger.error(
                "[adcp.task_webhook_outbox] permanent HTTP failure for task %s; row quarantined",
                task_id,
            )
            return True

        delay = self._retry_delay(int(attempt_count))
        http_status = delivery.status_code if delivery is not None else None
        if delivery is not None:
            error_message = f"HTTP {delivery.status_code}"
        elif error is not None:
            error_message = f"{type(error).__name__}: delivery failed"
        else:
            error_message = "delivery failed without a result"
        async with self._pool.connection() as conn:
            await conn.execute(
                self._sql_release,
                (delay, http_status, error_message, row_id, lease_token),
            )
        logger.warning(
            "[adcp.task_webhook_outbox] delivery failed for task %s; retry in %.1fs",
            task_id,
            delay,
        )
        return True

    async def _quarantine_permanent_delivery_error(
        self,
        *,
        row_id: int,
        lease_token: str,
        task_id: str,
        error: BaseException,
    ) -> None:
        # Validation errors can originate in adopter-provided transport hooks,
        # whose messages may contain request credentials. Persist only the
        # local exception discriminator in the plaintext last_error column.
        error_message = f"permanent delivery validation failure: {type(error).__name__}"
        async with self._pool.connection() as conn:
            await conn.execute(
                self._sql_quarantine,
                (error_message[:1000], row_id, lease_token),
            )
        logger.error(
            "[adcp.task_webhook_outbox] permanent delivery failure for task %s; row quarantined",
            task_id,
        )

    @staticmethod
    def _is_retryable_http_status(status_code: int) -> bool:
        return status_code >= 500 or status_code in {408, 425, 429}

    async def purge_expired(self) -> None:
        """Delete delivery proof only after the advertised horizon elapses."""
        async with self._pool.connection() as conn:
            await conn.execute(self._sql_expire)
            await conn.execute(self._sql_purge)

    def _retry_delay(self, attempt_count: int) -> float:
        exponent = max(0, min(attempt_count - 1, 30))
        delay: float = float(
            min(
                self._retry.base_delay_seconds * (2**exponent),
                self._retry.max_delay_seconds,
            )
        )
        if self._retry.jitter:
            delay *= 0.5 + random.random() * 0.5
        return delay

    @staticmethod
    def _validate_delivery_sender(sender: WebhookSender) -> None:
        if not callable(getattr(sender, "send_prepared", None)):
            raise ValueError("webhook sender resolver must return a WebhookSender")
        if not getattr(sender, "_owns_client", False) or getattr(
            sender, "_allow_private_destinations", False
        ):
            raise ValueError(
                "PgTaskWebhookOutbox requires a WebhookSender using the SDK-owned "
                "IP-pinned transport with private destinations disabled"
            )
        if getattr(sender, "signs_with_rfc9421", False) is not True:
            raise ValueError("PgTaskWebhookOutbox requires an RFC 9421 signing sender")

    @staticmethod
    def _validate_signing_scope_id(signing_scope_id: str | None) -> None:
        if signing_scope_id is None:
            return
        if not signing_scope_id or not signing_scope_id.isprintable():
            raise ValueError("signing_scope_id must be a non-empty printable string")
        if len(signing_scope_id.encode("utf-8")) > MAX_SIGNING_SCOPE_ID_BYTES:
            raise ValueError(
                f"signing_scope_id must not exceed {MAX_SIGNING_SCOPE_ID_BYTES} UTF-8 bytes"
            )

    def _validate_scope_for_mode(
        self,
        signing_scope_id: str | None,
        *,
        authentication: TaskWebhookAuthentication | None,
        require_resolver_scope: bool,
    ) -> None:
        self._validate_signing_scope_id(signing_scope_id)
        if authentication is not None:
            if signing_scope_id is not None:
                raise ValueError(
                    "legacy webhook authentication must not carry an RFC 9421 signing scope"
                )
            return
        if self._sender is not None and signing_scope_id is not None:
            raise ValueError("fixed-sender outboxes must not carry a signing_scope_id")
        if (
            require_resolver_scope
            and self._sender_resolver is not None
            and signing_scope_id is None
        ):
            raise ValueError("sender-resolver outboxes require a signing_scope_id")

    async def _resolve_delivery_sender(self, signing_scope_id: str | None) -> WebhookSender:
        """Resolve and revalidate the sender at every delivery attempt."""
        self._validate_signing_scope_id(signing_scope_id)
        if self._sender is not None:
            if signing_scope_id is not None:
                raise ScopePermanentlyUnknown
            return self._sender

        resolver = self._sender_resolver
        if resolver is None or signing_scope_id is None:
            raise ScopePermanentlyUnknown
        try:
            resolution = await resolver.resolve(signing_scope_id)
        except (ScopePermanentlyUnknown, ScopeTransientlyUnavailable):
            raise
        except Exception:
            # Resolver diagnostics may contain key-service details. Keep the
            # durable row and logs on a bounded local discriminator only.
            raise ScopeTransientlyUnavailable from None
        try:
            if not isinstance(resolution, WebhookSenderResolution):
                raise ValueError("resolver returned an invalid sender resolution")
            sender = resolution.sender
            self._validate_delivery_sender(sender)
            sender_algorithm = getattr(getattr(sender, "_auth", None), "alg", None)
            if sender_algorithm not in resolution.advertised_algorithms:
                raise ValueError("resolved sender algorithm was not advertised for its scope")
            sender_timeout = float(getattr(sender, "_timeout", 0.0))
            if self._lease_seconds < sender_timeout + 5:
                raise ValueError("resolved sender timeout exceeds the outbox lease budget")
        except Exception:
            # Treat malformed/adversarial resolver output as a permanent local
            # configuration error without persisting its diagnostics.
            raise ScopePermanentlyUnknown from None
        return sender

    async def _deliver_prepared(
        self,
        prepared: PreparedWebhook,
        signing_scope_id: str | None,
        authentication: TaskWebhookAuthentication | None = None,
    ) -> WebhookDeliveryResult:
        """Select the registered mode and send within one lease budget."""
        if authentication is None:
            sender = await self._resolve_delivery_sender(signing_scope_id)
            return await sender.send_prepared(prepared)
        if signing_scope_id is not None:
            raise ValueError("legacy webhook authentication cannot use an RFC 9421 scope")
        sender = self._legacy_sender(authentication)
        try:
            return await sender.send_prepared(prepared)
        finally:
            await sender.aclose()

    def _legacy_sender(self, authentication: TaskWebhookAuthentication) -> WebhookSender:
        """Build an SDK-owned, IP-pinned sender for encrypted legacy credentials."""
        self._validate_authentication(authentication)
        # Keep the HTTP attempt inside the outbox lease even for very short
        # adopter-configured leases. Private destinations remain disabled.
        timeout_seconds = max(0.1, min(10.0, self._lease_seconds - 5.0))
        common: dict[str, Any] = {
            "timeout_seconds": timeout_seconds,
            "allow_private_destinations": False,
            "allowed_destination_ports": self._legacy_allowed_destination_ports,
            "transport_hooks": self._legacy_transport_hooks,
        }
        if authentication.scheme == "Bearer":
            return WebhookSender.from_bearer_token(authentication.credentials, **common)
        return WebhookSender.from_adcp_legacy_hmac(
            authentication.credentials.encode("utf-8"),
            key_id="adcp-task-registration",
            **common,
        )

    def _validate_authentication(
        self,
        authentication: TaskWebhookAuthentication | None,
    ) -> None:
        if authentication is None:
            return
        if not isinstance(authentication, TaskWebhookAuthentication):
            raise ValueError("webhook authentication must be TaskWebhookAuthentication or None")
        if authentication.scheme not in _LEGACY_AUTH_SCHEMES:
            raise ValueError(
                f"unsupported task webhook authentication scheme {authentication.scheme!r}; "
                "supported legacy schemes are 'Bearer' and 'HMAC-SHA256'"
            )
        if authentication.scheme == "HMAC-SHA256" and not self.legacy_hmac_fallback:
            raise ValueError(
                "task webhook HMAC-SHA256 authentication requires "
                "legacy_hmac_fallback=True and a matching capability advertisement"
            )
        if any(char in authentication.credentials for char in ("\r", "\n", "\x00")):
            raise ValueError("webhook authentication credentials contain a control character")

    @staticmethod
    def _protect_delivery_body(
        body: bytes,
        authentication: TaskWebhookAuthentication | None,
    ) -> bytes:
        """Keep old RFC rows byte-compatible; wrap encrypted legacy secrets."""
        if authentication is None:
            return body
        return json.dumps(
            {
                "task_webhook_delivery_version": _ENCRYPTED_DELIVERY_VERSION,
                "body": base64.b64encode(body).decode("ascii"),
                "authentication": {
                    "scheme": authentication.scheme,
                    "credentials": authentication.credentials,
                },
            },
            ensure_ascii=False,
            separators=(",", ":"),
        ).encode("utf-8")

    def _open_delivery_body(
        self,
        protected_body: bytes,
    ) -> tuple[bytes, TaskWebhookAuthentication | None]:
        """Decode a legacy-auth envelope or accept a pre-feature RFC body."""
        try:
            value = json.loads(protected_body)
        except (json.JSONDecodeError, UnicodeDecodeError):
            return protected_body, None
        if not isinstance(value, dict) or "task_webhook_delivery_version" not in value:
            return protected_body, None
        if value.get("task_webhook_delivery_version") != _ENCRYPTED_DELIVERY_VERSION:
            raise ValueError("unsupported encrypted task webhook delivery version")
        authentication_value = value.get("authentication")
        if not isinstance(authentication_value, dict):
            raise ValueError("encrypted task webhook authentication is missing")
        scheme = authentication_value.get("scheme")
        credentials = authentication_value.get("credentials")
        encoded_body = value.get("body")
        if (
            not isinstance(scheme, str)
            or not isinstance(credentials, str)
            or not isinstance(encoded_body, str)
        ):
            raise ValueError("encrypted task webhook delivery envelope is invalid")
        try:
            authentication = TaskWebhookAuthentication(
                scheme=scheme,
                credentials=credentials,
            )
            body = base64.b64decode(encoded_body, validate=True)
        except (TypeError, ValueError) as exc:
            raise ValueError("encrypted task webhook delivery envelope is invalid") from exc
        self._validate_authentication(authentication)
        return body, authentication

    @staticmethod
    def _envelope_aad(
        *,
        account_id: str,
        task_id: str,
        task_type: str,
        status: str,
        url: str,
        operation_id: str,
        idempotency_key: str,
        signing_scope_id: str | None = None,
    ) -> bytes:
        """Canonical associated data binding every routing/security field."""
        fields: list[str | None] = [
            account_id,
            task_id,
            task_type,
            status,
            url,
            operation_id,
            idempotency_key,
        ]
        # Preserve the exact seven-field AAD for rows written before the
        # signing-scope migration. Scoped rows append the trusted scope and
        # therefore fail authenticated decryption if the DB column is swapped.
        if signing_scope_id is not None:
            fields.append(signing_scope_id)
        return json.dumps(
            fields,
            ensure_ascii=False,
            separators=(",", ":"),
        ).encode("utf-8")

    @staticmethod
    def _registration_aad(*, account_id: str, task_id: str, task_type: str) -> bytes:
        return json.dumps(
            ["task-webhook-registration-v1", account_id, task_id, task_type],
            ensure_ascii=False,
            separators=(",", ":"),
        ).encode("utf-8")

    @staticmethod
    def _validate_callback_url(url: str) -> None:
        if len(url) > 2048:
            raise ValueError("webhook URL must not exceed 2048 characters")
        parsed = httpx.URL(url)
        if parsed.scheme != "https" or not parsed.host:
            raise ValueError("webhook URL must be an absolute HTTPS URL")
        if parsed.username or parsed.password:
            raise ValueError("webhook URL must not contain userinfo")

    @staticmethod
    def _validate_stored_body(
        body: bytes,
        *,
        task_id: str,
        task_type: str,
        status: str,
        operation_id: str,
        idempotency_key: str,
    ) -> None:
        payload = json.loads(body)
        if not isinstance(payload, dict):
            raise ValueError("stored webhook body must be a JSON object")
        expected = {
            "task_id": task_id,
            "task_type": task_type,
            "status": status,
            "operation_id": operation_id,
            "idempotency_key": idempotency_key,
        }
        if any(payload.get(key) != value for key, value in expected.items()):
            raise ValueError("stored webhook body does not match its envelope metadata")

Atomic task-webhook outbox and lease-based delivery worker.

Construct this with the same pool as :class:PgTaskRegistry, then pass it to PgTaskRegistry(..., task_webhook_outbox=outbox). Call :meth:create_schema during migration/startup and run at least one :meth:run_worker loop in every deployment.

Class variables

var delivery_state_is_durable : ClassVar[bool]

var supports_atomic_task_outbox : ClassVar[bool]

Methods

async def create_schema(self) ‑> None

Expand source code

async def create_schema(self) -> None:
    """Create the outbox table and work index idempotently."""
    statements = [
        f"""CREATE TABLE IF NOT EXISTS {self._table} (
            id                 BIGSERIAL PRIMARY KEY,
            task_id            TEXT COLLATE "C" NOT NULL UNIQUE,
            account_id         TEXT COLLATE "C" NOT NULL,
            task_type          TEXT NOT NULL,
            terminal_status    TEXT NOT NULL,
            url                TEXT NOT NULL,
            operation_id       TEXT NOT NULL,
            idempotency_key    TEXT COLLATE "C" NOT NULL UNIQUE,
            signing_scope_id   TEXT COLLATE "C",
            encrypted_body     BYTEA NOT NULL,
            envelope_nonce     BYTEA NOT NULL,
            state              TEXT NOT NULL DEFAULT 'pending',
            attempt_count      INTEGER NOT NULL DEFAULT 0,
            available_at       TIMESTAMPTZ NOT NULL DEFAULT now(),
            lease_token        TEXT COLLATE "C",
            lease_expires_at   TIMESTAMPTZ,
            created_at         TIMESTAMPTZ NOT NULL DEFAULT now(),
            updated_at         TIMESTAMPTZ NOT NULL DEFAULT now(),
            first_attempt_at   TIMESTAMPTZ,
            retry_until        TIMESTAMPTZ,
            retry_horizon_seconds INTEGER NOT NULL,
            delivered_at       TIMESTAMPTZ,
            last_http_status   INTEGER,
            last_error         TEXT,
            CHECK (state IN ('pending', 'in_flight', 'delivered', 'expired', 'invalid')),
            CHECK (terminal_status IN ('completed', 'failed')),
            CHECK (attempt_count >= 0),
            CHECK (octet_length(envelope_nonce) = 12),
            CHECK (retry_horizon_seconds BETWEEN 86400 AND 604800),
            CHECK ((first_attempt_at IS NULL) = (retry_until IS NULL)),
            CHECK (retry_until IS NULL OR retry_until > first_attempt_at)
        )""",
        f'''ALTER TABLE {self._table}
            ADD COLUMN IF NOT EXISTS signing_scope_id TEXT COLLATE "C"''',
        f"""CREATE INDEX IF NOT EXISTS {self._table}_work_idx
            ON {self._table} (available_at, id)
            WHERE state IN ('pending', 'in_flight')""",
        f"""CREATE INDEX IF NOT EXISTS {self._table}_retry_until_idx
            ON {self._table} (retry_until)""",
    ]
    async with self._pool.connection() as conn:
        for statement in statements:
            await conn.execute(statement)

Create the outbox table and work index idempotently.

async def enqueue_terminal(self,
conn: Any,
*,
task_id: str,
account_id: str,
task_type: str,
status: str,
result: dict[str, Any],
url: str,
operation_id: str,
token: str | None,
authentication: TaskWebhookAuthentication | None = None,
signing_scope_id: str | None = None) ‑> int

Expand source code

async def enqueue_terminal(
    self,
    conn: Any,
    *,
    task_id: str,
    account_id: str,
    task_type: str,
    status: str,
    result: dict[str, Any],
    url: str,
    operation_id: str,
    token: str | None,
    authentication: TaskWebhookAuthentication | None = None,
    signing_scope_id: str | None = None,
) -> int:
    """Insert a terminal webhook using the caller's open transaction."""
    if status not in {"completed", "failed"}:
        raise ValueError(f"terminal webhook status must be completed or failed, got {status!r}")
    self._validate_authentication(authentication)
    self._validate_scope_for_mode(
        signing_scope_id,
        authentication=authentication,
        require_resolver_scope=False,
    )
    preparer = self._sender or WebhookSender
    prepared = preparer.prepare_mcp(
        url=url,
        task_id=task_id,
        task_type=task_type,
        status=status,
        result=result,
        operation_id=operation_id,
        token=token,
    )
    self._validate_callback_url(prepared.url)
    nonce = os.urandom(12)
    aad = self._envelope_aad(
        account_id=account_id,
        task_id=task_id,
        task_type=task_type,
        status=status,
        url=prepared.url,
        operation_id=operation_id,
        idempotency_key=prepared.idempotency_key,
        signing_scope_id=signing_scope_id,
    )
    protected_body = self._protect_delivery_body(prepared.body, authentication)
    encrypted_body = self._cipher.encrypt(nonce, protected_body, aad)
    cursor = await conn.execute(
        self._sql_insert,
        (
            task_id,
            task_type,
            status,
            prepared.url,
            operation_id,
            prepared.idempotency_key,
            signing_scope_id,
            account_id,
            encrypted_body,
            nonce,
            self.delivery_retry_horizon_seconds,
        ),
    )
    row = await cursor.fetchone()
    if row is None:
        raise RuntimeError("task webhook outbox insert returned no id")
    return int(row[0])

Insert a terminal webhook using the caller's open transaction.

def open_registration(self,
*,
account_id: str,
task_id: str,
task_type: str,
encrypted_registration: bytes,
nonce: bytes) ‑> tuple[str, str, str | None]

Expand source code

def open_registration(
    self,
    *,
    account_id: str,
    task_id: str,
    task_type: str,
    encrypted_registration: bytes,
    nonce: bytes,
) -> tuple[str, str, str | None]:
    """Verify and decrypt a callback registration.

    The three-item return shape is retained for compatibility. Durable
    registry dispatch uses the private scope-aware decoder below.
    """
    url, operation_id, token, _authentication, _signing_scope_id = (
        self._open_registration_with_scope(
            account_id=account_id,
            task_id=task_id,
            task_type=task_type,
            encrypted_registration=encrypted_registration,
            nonce=nonce,
        )
    )
    return url, operation_id, token

Verify and decrypt a callback registration.

The three-item return shape is retained for compatibility. Durable registry dispatch uses the private scope-aware decoder below.

async def process_one(self) ‑> bool

Expand source code

async def process_one(self) -> bool:
    """Claim and attempt one delivery; return ``False`` when idle."""
    lease_token = uuid.uuid4().hex
    async with self._pool.connection() as conn:
        await conn.execute(self._sql_expire)
        cursor = await conn.execute(
            self._sql_claim,
            (lease_token, self._lease_seconds),
        )
        row = await cursor.fetchone()
    if row is None:
        return False

    (
        row_id,
        account_id,
        task_id,
        task_type,
        status,
        url,
        operation_id,
        idempotency_key,
        signing_scope_id,
        encrypted_body,
        nonce,
        attempt_count,
    ) = row
    aad = self._envelope_aad(
        account_id=str(account_id),
        task_id=str(task_id),
        task_type=str(task_type),
        status=str(status),
        url=str(url),
        operation_id=str(operation_id),
        idempotency_key=str(idempotency_key),
        signing_scope_id=(str(signing_scope_id) if signing_scope_id is not None else None),
    )
    try:
        protected_body = self._cipher.decrypt(bytes(nonce), bytes(encrypted_body), aad)
        body_bytes, authentication = self._open_delivery_body(protected_body)
        self._validate_stored_body(
            body_bytes,
            task_id=str(task_id),
            task_type=str(task_type),
            status=str(status),
            operation_id=str(operation_id),
            idempotency_key=str(idempotency_key),
        )
    except (InvalidTag, ValueError, json.JSONDecodeError, UnicodeDecodeError):
        error_message = (
            "stored webhook envelope failed authenticated binding verification; "
            "row quarantined without delivery"
        )
        async with self._pool.connection() as conn:
            await conn.execute(
                self._sql_quarantine,
                (error_message, row_id, lease_token),
            )
        logger.error(
            "[adcp.task_webhook_outbox] integrity failure for task %s; row quarantined",
            task_id,
        )
        return True
    prepared = PreparedWebhook(
        url=str(url),
        idempotency_key=str(idempotency_key),
        body=body_bytes,
    )
    delivery: WebhookDeliveryResult | None = None
    error: BaseException | None = None
    try:
        delivery = await asyncio.wait_for(
            self._deliver_prepared(
                prepared,
                str(signing_scope_id) if signing_scope_id is not None else None,
                authentication,
            ),
            timeout=self._lease_seconds - 1,
        )
    except ScopePermanentlyUnknown:
        await self._quarantine_permanent_delivery_error(
            row_id=row_id,
            lease_token=lease_token,
            task_id=str(task_id),
            error=ValueError("webhook signing scope is permanently unavailable"),
        )
        return True
    except ScopeTransientlyUnavailable:
        error = RuntimeError("webhook signing scope is temporarily unavailable")
    except SSRFValidationError as exc:
        if exc.transient:
            error = exc
        else:
            await self._quarantine_permanent_delivery_error(
                row_id=row_id,
                lease_token=lease_token,
                task_id=str(task_id),
                error=exc,
            )
            return True
    except ValueError as exc:
        await self._quarantine_permanent_delivery_error(
            row_id=row_id,
            lease_token=lease_token,
            task_id=str(task_id),
            error=exc,
        )
        return True
    except Exception as exc:
        error = exc

    if delivery is not None and delivery.ok:
        async with self._pool.connection() as conn:
            await conn.execute(
                self._sql_ack,
                (delivery.status_code, row_id, lease_token),
            )
        return True

    if delivery is not None and not self._is_retryable_http_status(delivery.status_code):
        # Receiver-controlled bodies can echo Authorization/signature
        # material. Never persist them in the plaintext last_error column.
        error_message = f"permanent HTTP {delivery.status_code}"
        async with self._pool.connection() as conn:
            await conn.execute(
                self._sql_quarantine,
                (error_message[:1000], row_id, lease_token),
            )
        logger.error(
            "[adcp.task_webhook_outbox] permanent HTTP failure for task %s; row quarantined",
            task_id,
        )
        return True

    delay = self._retry_delay(int(attempt_count))
    http_status = delivery.status_code if delivery is not None else None
    if delivery is not None:
        error_message = f"HTTP {delivery.status_code}"
    elif error is not None:
        error_message = f"{type(error).__name__}: delivery failed"
    else:
        error_message = "delivery failed without a result"
    async with self._pool.connection() as conn:
        await conn.execute(
            self._sql_release,
            (delay, http_status, error_message, row_id, lease_token),
        )
    logger.warning(
        "[adcp.task_webhook_outbox] delivery failed for task %s; retry in %.1fs",
        task_id,
        delay,
    )
    return True

Claim and attempt one delivery; return False when idle.

def protect_registration(self,
*,
account_id: str,
task_id: str,
task_type: str,
url: str,
operation_id: str,
token: str | None,
authentication: TaskWebhookAuthentication | None = None,
signing_scope_id: str | None = None) ‑> tuple[bytes, bytes]

Expand source code

def protect_registration(
    self,
    *,
    account_id: str,
    task_id: str,
    task_type: str,
    url: str,
    operation_id: str,
    token: str | None,
    authentication: TaskWebhookAuthentication | None = None,
    signing_scope_id: str | None = None,
) -> tuple[bytes, bytes]:
    """Encrypt and authenticate callback registration at task issue time."""
    self._validate_callback_url(url)
    self._validate_authentication(authentication)
    self._validate_scope_for_mode(
        signing_scope_id,
        authentication=authentication,
        require_resolver_scope=True,
    )
    nonce = os.urandom(12)
    plaintext = json.dumps(
        {
            "url": url,
            "operation_id": operation_id,
            "token": token,
            "authentication": (
                {
                    "scheme": authentication.scheme,
                    "credentials": authentication.credentials,
                }
                if authentication is not None
                else None
            ),
            "signing_scope_id": signing_scope_id,
        },
        ensure_ascii=False,
        separators=(",", ":"),
    ).encode("utf-8")
    return (
        self._cipher.encrypt(
            nonce,
            plaintext,
            self._registration_aad(
                account_id=account_id,
                task_id=task_id,
                task_type=task_type,
            ),
        ),
        nonce,
    )

Encrypt and authenticate callback registration at task issue time.

async def purge_expired(self) ‑> None

Expand source code

async def purge_expired(self) -> None:
    """Delete delivery proof only after the advertised horizon elapses."""
    async with self._pool.connection() as conn:
        await conn.execute(self._sql_expire)
        await conn.execute(self._sql_purge)

Delete delivery proof only after the advertised horizon elapses.

async def run_worker(self, *, poll_interval: float = 1.0, purge_interval: float = 300.0) ‑> None

Expand source code

async def run_worker(
    self,
    *,
    poll_interval: float = 1.0,
    purge_interval: float = 300.0,
) -> None:
    """Continuously publish eligible rows until the task is cancelled."""
    if poll_interval <= 0:
        raise ValueError("poll_interval must be positive")
    if purge_interval <= 0:
        raise ValueError("purge_interval must be positive")
    self._worker_started = True
    next_purge = 0.0
    try:
        while True:
            try:
                now = time.monotonic()
                if now >= next_purge:
                    await self.purge_expired()
                    next_purge = now + purge_interval
                processed = await self.process_one()
            except asyncio.CancelledError:
                raise
            except Exception:
                logger.exception("[adcp.task_webhook_outbox] worker iteration failed; retrying")
                await asyncio.sleep(poll_interval)
                continue
            if not processed:
                await asyncio.sleep(poll_interval)
    finally:
        self._worker_started = False

Continuously publish eligible rows until the task is cancelled.

def validate_registration(self, url: str, authentication: TaskWebhookAuthentication | None = None) ‑> None

Expand source code

def validate_registration(
    self,
    url: str,
    authentication: TaskWebhookAuthentication | None = None,
) -> None:
    """Validate callback syntax before a task is accepted as Submitted."""
    self._validate_callback_url(url)
    self._validate_authentication(authentication)

Validate callback syntax before a task is accepted as Submitted.