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

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

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

adcp.decisioning.pg.task_registry

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

Classes

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(
                "table_name must match [a-z_][a-z0-9_]* (ASCII only), " f"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"
        )
        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 via ``CREATE ... IF NOT EXISTS``;
        safe to call on every app boot.

        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__
        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");"
            f"CREATE INDEX IF NOT EXISTS {table}_api_key_id_idx "  # noqa: S608
            f"    ON {table} (api_key_id) WHERE api_key_id IS NOT NULL;"
            f"CREATE INDEX IF NOT EXISTS {table}_status_idx "  # noqa: S608
            f"    ON {table} (status) WHERE status <> 'active';"
        )
        with self._pool.connection() as conn, conn.cursor() as cur:
            cur.execute(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,))
            row = cur.fetchone()
            return _row_to_agent(row) if row 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 via ``CREATE ... IF NOT EXISTS``;
    safe to call on every app boot.

    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__
    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");"
        f"CREATE INDEX IF NOT EXISTS {table}_api_key_id_idx "  # noqa: S608
        f"    ON {table} (api_key_id) WHERE api_key_id IS NOT NULL;"
        f"CREATE INDEX IF NOT EXISTS {table}_status_idx "  # noqa: S608
        f"    ON {table} (status) WHERE status <> 'active';"
    )
    with self._pool.connection() as conn, conn.cursor() as cur:
        cur.execute(ddl)

Create the registry table + indexes for this store's table_name. Idempotent via CREATE … IF NOT EXISTS; safe to call on every app boot.

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 | None = None,
    ) -> ProposalRecord | None:
        # The Protocol allows expected_account_id=None — historically a
        # convenience for diagnostic / admin callers. We still serve
        # that case but route it through a separate query without the
        # account predicate so the tenancy-aware fast path is purely
        # parameterised; a future code reader can't accidentally pass
        # None into a tenancy-required call.
        if expected_account_id is 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 proposal_id = %s"
            )
            params: tuple[Any, ...] = (proposal_id,)
        else:
            sql = (  # noqa: S608
                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 = (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 | None = None) ‑> ProposalRecord | None

Expand source code

async def get(
    self,
    proposal_id: str,
    *,
    expected_account_id: str | None = None,
) -> ProposalRecord | None:
    # The Protocol allows expected_account_id=None — historically a
    # convenience for diagnostic / admin callers. We still serve
    # that case but route it through a separate query without the
    # account predicate so the tenancy-aware fast path is purely
    # parameterised; a future code reader can't accidentally pass
    # None into a tenancy-required call.
    if expected_account_id is 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 proposal_id = %s"
        )
        params: tuple[Any, ...] = (proposal_id,)
    else:
        sql = (  # noqa: S608
            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 = (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)

Expand source code

class PgTaskRegistry:
    """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.

    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, _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}")
        self._pool = pool
        self._table = _table

        # 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}"
            f" (task_id, account_id, state, task_type, created_at, updated_at)"
            f" VALUES (%s, %s, 'submitted', %s, %s, %s)"
        )
        self._sql_update_progress = (  # noqa: S608
            f"UPDATE {self._table}"
            f" SET state = CASE state WHEN 'submitted' THEN 'working' ELSE state END,"
            f"     progress = %s::jsonb, updated_at = %s"
            f" WHERE task_id = %s AND state NOT IN ('completed', 'failed')"
        )
        self._sql_complete = (  # noqa: S608
            f"UPDATE {self._table}"
            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"
        )
        self._sql_fail = (  # noqa: S608
            f"UPDATE {self._table}"
            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"
        )
        # 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,"
            f"       progress, result, error, 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 FROM {self._table} WHERE task_id = %s"
        )
        self._sql_get_state_error = (  # noqa: S608
            f"SELECT state, error FROM {self._table} WHERE task_id = %s"
        )
        self._sql_discard = f"DELETE FROM {self._table} WHERE task_id = %s"  # noqa: S608
        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"    created_at  DOUBLE PRECISION NOT NULL,"
            f"    updated_at  DOUBLE PRECISION NOT NULL"
            f");"
            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,
    ) -> 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."
            )
        task_id = f"task_{uuid.uuid4().hex[:16]}"
        now = time.time()
        async with self._pool.connection() as conn:
            await conn.execute(self._sql_insert, (task_id, account_id, task_type, now, now))
        return task_id

    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:
            await conn.execute(
                self._sql_update_progress,
                (json.dumps(progress), time.time(), task_id),
            )
            # Zero rows updated means unknown task_id or terminal state — silent
            # no-op per Protocol contract. The InMemoryTaskRegistry logs a
            # WARNING on terminal-state drops; we omit the extra SELECT needed
            # to distinguish the two cases since the dispatch wrapper swallows
            # the result either way.

    async def complete(
        self,
        task_id: str,
        result: dict[str, 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:
            cur = await conn.execute(self._sql_complete, (json.dumps(result), time.time(), task_id))
            if await cur.fetchone() is not None:
                return  # updated successfully

            # 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 = row
            if state == "completed":
                if existing_result == result:
                    return  # 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:
        """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:
            cur = await conn.execute(self._sql_fail, (json.dumps(error), time.time(), task_id))
            if await cur.fetchone() is not None:
                return  # updated successfully

            # 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  # 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[7],
                "updated_at": row[8],
            }

    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:
            await conn.execute(self._sql_discard, (task_id,))

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.

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.

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:
    """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:
        cur = await conn.execute(self._sql_complete, (json.dumps(result), time.time(), task_id))
        if await cur.fetchone() is not None:
            return  # updated successfully

        # 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 = row
        if state == "completed":
            if existing_result == result:
                return  # 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}")

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 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:
        await conn.execute(self._sql_discard, (task_id,))

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:
    """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:
        cur = await conn.execute(self._sql_fail, (json.dumps(error), time.time(), task_id))
        if await cur.fetchone() is not None:
            return  # updated successfully

        # 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  # 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}")

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 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[7],
            "updated_at": row[8],
        }

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) ‑> str

Expand source code

async def issue(
    self,
    *,
    account_id: str,
    task_type: str,
) -> 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."
        )
    task_id = f"task_{uuid.uuid4().hex[:16]}"
    now = time.time()
    async with self._pool.connection() as conn:
        await conn.execute(self._sql_insert, (task_id, account_id, task_type, now, 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 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:
        await conn.execute(
            self._sql_update_progress,
            (json.dumps(progress), time.time(), task_id),
        )
        # Zero rows updated means unknown task_id or terminal state — silent
        # no-op per Protocol contract. The InMemoryTaskRegistry logs a
        # WARNING on terminal-state drops; we omit the extra SELECT needed
        # to distinguish the two cases since the dispatch wrapper swallows
        # the result either way.

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)

Expand source code

class PgTaskRegistry:
    """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.

    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, _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}")
        self._pool = pool
        self._table = _table

        # 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}"
            f" (task_id, account_id, state, task_type, created_at, updated_at)"
            f" VALUES (%s, %s, 'submitted', %s, %s, %s)"
        )
        self._sql_update_progress = (  # noqa: S608
            f"UPDATE {self._table}"
            f" SET state = CASE state WHEN 'submitted' THEN 'working' ELSE state END,"
            f"     progress = %s::jsonb, updated_at = %s"
            f" WHERE task_id = %s AND state NOT IN ('completed', 'failed')"
        )
        self._sql_complete = (  # noqa: S608
            f"UPDATE {self._table}"
            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"
        )
        self._sql_fail = (  # noqa: S608
            f"UPDATE {self._table}"
            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"
        )
        # 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,"
            f"       progress, result, error, 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 FROM {self._table} WHERE task_id = %s"
        )
        self._sql_get_state_error = (  # noqa: S608
            f"SELECT state, error FROM {self._table} WHERE task_id = %s"
        )
        self._sql_discard = f"DELETE FROM {self._table} WHERE task_id = %s"  # noqa: S608
        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"    created_at  DOUBLE PRECISION NOT NULL,"
            f"    updated_at  DOUBLE PRECISION NOT NULL"
            f");"
            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,
    ) -> 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."
            )
        task_id = f"task_{uuid.uuid4().hex[:16]}"
        now = time.time()
        async with self._pool.connection() as conn:
            await conn.execute(self._sql_insert, (task_id, account_id, task_type, now, now))
        return task_id

    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:
            await conn.execute(
                self._sql_update_progress,
                (json.dumps(progress), time.time(), task_id),
            )
            # Zero rows updated means unknown task_id or terminal state — silent
            # no-op per Protocol contract. The InMemoryTaskRegistry logs a
            # WARNING on terminal-state drops; we omit the extra SELECT needed
            # to distinguish the two cases since the dispatch wrapper swallows
            # the result either way.

    async def complete(
        self,
        task_id: str,
        result: dict[str, 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:
            cur = await conn.execute(self._sql_complete, (json.dumps(result), time.time(), task_id))
            if await cur.fetchone() is not None:
                return  # updated successfully

            # 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 = row
            if state == "completed":
                if existing_result == result:
                    return  # 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:
        """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:
            cur = await conn.execute(self._sql_fail, (json.dumps(error), time.time(), task_id))
            if await cur.fetchone() is not None:
                return  # updated successfully

            # 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  # 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[7],
                "updated_at": row[8],
            }

    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:
            await conn.execute(self._sql_discard, (task_id,))

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.

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.

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:
    """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:
        cur = await conn.execute(self._sql_complete, (json.dumps(result), time.time(), task_id))
        if await cur.fetchone() is not None:
            return  # updated successfully

        # 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 = row
        if state == "completed":
            if existing_result == result:
                return  # 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}")

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 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:
        await conn.execute(self._sql_discard, (task_id,))

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:
    """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:
        cur = await conn.execute(self._sql_fail, (json.dumps(error), time.time(), task_id))
        if await cur.fetchone() is not None:
            return  # updated successfully

        # 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  # 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}")

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 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[7],
            "updated_at": row[8],
        }

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) ‑> str

Expand source code

async def issue(
    self,
    *,
    account_id: str,
    task_type: str,
) -> 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."
        )
    task_id = f"task_{uuid.uuid4().hex[:16]}"
    now = time.time()
    async with self._pool.connection() as conn:
        await conn.execute(self._sql_insert, (task_id, account_id, task_type, now, 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 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:
        await conn.execute(
            self._sql_update_progress,
            (json.dumps(progress), time.time(), task_id),
        )
        # Zero rows updated means unknown task_id or terminal state — silent
        # no-op per Protocol contract. The InMemoryTaskRegistry logs a
        # WARNING on terminal-state drops; we omit the extra SELECT needed
        # to distinguish the two cases since the dispatch wrapper swallows
        # the result either way.

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.