From 406a58dc062c996f370680192a086c6cb0588e61 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Fri, 30 Oct 2026 19:54:49 +0000 Subject: [PATCH 1/3] feat(accounts)!: ensure returns an AccountOutcome and routes to request_account client.accounts.ensure is reshaped, not added (DX plan slice P8). - It routes by capability: request_account when the seller advertises account.requests, sync_accounts otherwise (which needs billing=). - It returns an AccountOutcome: AccountReady, AccountPending (with a PendingAction), AccountAmbiguous (with choose(account_id)) or AccountRejected. A sync_accounts pending_approval row with setup maps to AccountPending, and a failed row to AccountRejected. Pending, ambiguous and rejected are values; only a failure to process the call raises AccountRequestError. - It stays idempotent per seller, buyer identity and natural key: a repeat call returns the account, or the open request's current state read back through list_accounts account_request_ids, and never opens a duplicate. currency= and timezone= join the key. - accounts.refresh() re-reads open requests and pending accounts and returns the outcomes whose state moved; accounts.pending() lists what is still waiting without a network call. Both see records across restarts through storage that implements items(seller) (ListableAccountStorage), as InMemoryAccountStorage does. - next_step(outcome) returns wait_for_seller, send_human_to_url, pick_candidate, contact_seller or none, so agents never switch on the extensible kind. format_pending_action(action) renders safe text: the seller message is untrusted, and the URL is for a person, never fetched. AccountRecord gains request_id, request_state and pending_action. BREAKING CHANGE: AccountRegistry.ensure returns an AccountOutcome instead of an AccountRecord, billing= is optional and required only on the sync_accounts path, and a failed or rejected row is returned as AccountRejected instead of raised. AccountReady keeps account_id and status. Refs #1695 Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/account-lifecycle.md | 58 +++- src/adcp/__init__.py | 34 ++ src/adcp/account_outcomes.py | 211 ++++++++++++ src/adcp/accounts.py | 431 +++++++++++++++++++++--- src/adcp/exceptions.py | 9 + tests/fixtures/public_api_snapshot.json | 10 + tests/test_account_lifecycle.py | 6 +- tests/test_account_outcomes.py | 255 ++++++++++++++ 8 files changed, 960 insertions(+), 54 deletions(-) create mode 100644 src/adcp/account_outcomes.py create mode 100644 tests/test_account_outcomes.py diff --git a/docs/account-lifecycle.md b/docs/account-lifecycle.md index a6f587694..a2b39cbb9 100644 --- a/docs/account-lifecycle.md +++ b/docs/account-lifecycle.md @@ -33,7 +33,14 @@ Start with the seller's capabilities and discover public inventory before choosing billing and payment terms: ```python -from adcp import ADCPClient +from adcp import ( + AccountAmbiguous, + AccountPending, + AccountReady, + AccountRejected, + ADCPClient, + format_pending_action, +) from adcp.types import GetProductsRequest client = ADCPClient(agent_config, account_policy="auto", raise_account_errors=True) @@ -50,10 +57,18 @@ key = { public = await client.get_products(GetProductsRequest(buying_mode="wholesale", account=key)) # Provision only when you have selected the commercial terms. -account = await client.accounts.ensure( +outcome = await client.accounts.ensure( key, billing="agent", payment_terms="net_30", billing_entity=billing_entity, ) -print(account.status, account.account_id) +match outcome: + case AccountReady(account_id=account_id, status=status): + print(status, account_id) + case AccountPending(action=action): + print(format_pending_action(action)) # show a person; never fetch action.url + case AccountAmbiguous(candidates=candidates): + outcome = await outcome.choose(candidates[0]["account_id"]) + case AccountRejected(errors=errors): + print(errors) # The provisioned natural key is retained on subsequent requests. priced = await client.get_products( @@ -61,9 +76,40 @@ priced = await client.get_products( ) ``` -`ensure` calls `sync_accounts` once for each complete key at each seller and -returns an `AccountRecord`. Successful pending-approval or payment-required -accounts are provisioned; they still need buyer action before committing spend. +`ensure` returns an `AccountOutcome` (AdCP 3.3, a breaking change from the +`AccountRecord` it returned before), one of: + +- `AccountReady(account_id, status, via, ...)`: the account exists. Its status may + still need buyer action before spend (`payment_required`, for example). +- `AccountPending(action, via, request_id, ...)`: waiting on the seller or a + buyer-side human. `action` is a `PendingAction(waiting_on, kind, message, url, + expires_at)`. `url` is for a person to open; it is not a bearer link and an + agent must never fetch it. Render it with `format_pending_action(action)`, + which treats `message` as untrusted text. +- `AccountAmbiguous(candidates, request_id)`: several of your existing accounts + match; `await outcome.choose(account_id)` settles it. +- `AccountRejected(errors, via, request_id)`: the seller declined, or a buyer + action expired. + +Pending, ambiguous and rejected are values, not errors; only a failure to +process the call raises `AccountRequestError`. Agents should switch on +`next_step(outcome)` (`none`, `wait_for_seller`, `send_human_to_url`, +`pick_candidate` or `contact_seller`), not on `action.kind`, which is extensible. + +`ensure` routes by capability. When the seller advertises `account.requests`, it +calls `request_account`. Otherwise it calls `sync_accounts` once for each +complete key at each seller, which needs `billing=`; a `pending_approval` row +with `setup` is `AccountPending`. Either way `ensure` is idempotent: a repeat +call returns the account, or the open request's current state read back through +`list_accounts`, and never opens a duplicate. `currency=` and `timezone=` join +the key when it does not carry them. + +`await client.accounts.refresh()` re-reads open requests and pending accounts +(call it from a cron job or an `account_request.changed` webhook) and returns +the outcomes whose state moved; `await client.accounts.pending()` lists what is +still waiting without a network call. Both find records across restarts when +the storage implements `items(seller)` (`ListableAccountStorage`), as +`InMemoryAccountStorage` does. Only active and payment-required accounts can browse account-priced products; other statuses use public discovery in `auto` mode when allowed, or raise in `strict` mode. diff --git a/src/adcp/__init__.py b/src/adcp/__init__.py index f17c8bb6b..10a190bef 100644 --- a/src/adcp/__init__.py +++ b/src/adcp/__init__.py @@ -113,6 +113,17 @@ def _resolve_version() -> str: "AccountRegistry", "AccountStorage", "InMemoryAccountStorage", + "ListableAccountStorage", + ), + "adcp.account_outcomes": ( + "AccountAmbiguous", + "AccountOutcome", + "AccountPending", + "AccountReady", + "AccountRejected", + "PendingAction", + "format_pending_action", + "next_step", ), "adcp.acceptance": ( "AcceptancePolicyAssessment", @@ -270,6 +281,7 @@ def _resolve_version() -> str: "AccountNotFoundError", "AccountSetupRequiredError", "AccountPaymentRequiredError", + "AccountRequestError", "ConfigurationError", "extract_adcp_error_info", "is_pre_dispatch_error", @@ -1035,6 +1047,15 @@ def get_adcp_version() -> str: "AccountRegistry", "AccountStorage", "InMemoryAccountStorage", + "ListableAccountStorage", + "AccountAmbiguous", + "AccountOutcome", + "AccountPending", + "AccountReady", + "AccountRejected", + "PendingAction", + "format_pending_action", + "next_step", "ADCPClient", "ADCPMultiAgentClient", "Checkpoint", @@ -1646,6 +1667,7 @@ def get_adcp_version() -> str: "AccountNotFoundError", "AccountSetupRequiredError", "AccountPaymentRequiredError", + "AccountRequestError", "ConfigurationError", "IdempotencyConflictError", "IdempotencyExpiredError", @@ -1819,12 +1841,23 @@ def get_adcp_version() -> str: AcceptancePolicyResolver, AcceptancePolicySurface, ) + from adcp.account_outcomes import ( + AccountAmbiguous, + AccountOutcome, + AccountPending, + AccountReady, + AccountRejected, + PendingAction, + format_pending_action, + next_step, + ) from adcp.accounts import ( AccountPolicy, AccountRecord, AccountRegistry, AccountStorage, InMemoryAccountStorage, + ListableAccountStorage, ) from adcp.adagents import ( AdagentsCacheEntry, @@ -1876,6 +1909,7 @@ def get_adcp_version() -> str: from adcp.exceptions import ( # noqa: F401 AccountNotFoundError, AccountPaymentRequiredError, + AccountRequestError, AccountSetupRequiredError, AdagentsAccessBlockedError, AdagentsHTTPError, diff --git a/src/adcp/account_outcomes.py b/src/adcp/account_outcomes.py new file mode 100644 index 000000000..12c313196 --- /dev/null +++ b/src/adcp/account_outcomes.py @@ -0,0 +1,211 @@ +"""What ``client.accounts.ensure`` returns: a discriminated account outcome (AdCP 3.3). + +Pending, ambiguous and rejected are values, not errors. Only a failure to +process the call raises (:class:`~adcp.exceptions.AccountRequestError`):: + + outcome = await client.accounts.ensure(key, billing="operator") + match outcome: + case AccountReady(account_id=account_id): use(account_id) + case AccountPending(action=action) if action.waiting_on == "buyer": + show_to_human(format_pending_action(action)) + case AccountAmbiguous(candidates=candidates): + outcome = await outcome.choose(pick(candidates)) + case AccountRejected(errors=errors): report(errors) + +Agents should switch on :func:`next_step`, not on ``PendingAction.kind``, +which is extensible. The TypeScript SDK spells these ``AccountOutcome``, +``nextStep`` and ``formatPendingAction``. +""" + +from __future__ import annotations + +import re +from collections.abc import Awaitable, Callable, Mapping, Sequence +from dataclasses import dataclass, field +from typing import Any, Literal + +__all__ = [ + "AccountAmbiguous", + "AccountOutcome", + "AccountPending", + "AccountReady", + "AccountRejected", + "NextStep", + "PendingAction", + "format_pending_action", + "next_step", +] + +#: The provisioning path an outcome came from. +AccountVia = Literal["sync_accounts", "request_account"] +NextStep = Literal[ + "wait_for_seller", "send_human_to_url", "pick_candidate", "contact_seller", "none" +] + + +@dataclass(frozen=True) +class PendingAction: + """Whose turn it is and what is needed (the wire ``pending_action``/``setup``). + + ``url`` is where a *human* completes the action. Show it to a person. It is + not a bearer link, and an agent must never fetch it. ``message`` is + untrusted seller text; render it through :func:`format_pending_action`. + ``kind`` is extensible: switch on :func:`next_step` instead. + """ + + waiting_on: str + kind: str + message: str + url: str | None = None + expires_at: str | None = None + + @classmethod + def from_wire(cls, value: Mapping[str, Any]) -> PendingAction: + url = value.get("url") + waiting_on = value.get("waiting_on") or ("buyer" if url else "seller") + return cls( + waiting_on=str(waiting_on), + kind=str(value.get("kind") or "review"), + message=str(value.get("message") or ""), + url=None if url is None else str(url), + expires_at=None if value.get("expires_at") is None else str(value["expires_at"]), + ) + + def to_wire(self) -> dict[str, Any]: + payload: dict[str, Any] = { + "waiting_on": self.waiting_on, + "kind": self.kind, + "message": self.message, + } + if self.url is not None: + payload["url"] = self.url + if self.expires_at is not None: + payload["expires_at"] = self.expires_at + return payload + + +@dataclass(frozen=True) +class AccountReady: + """The account exists: use ``account_id`` on account-scoped tasks. + + ``status`` is the account's status, which may still need buyer action + before spend (``payment_required``, for example). + """ + + account_id: str | None + status: str | None + via: AccountVia + request_id: str | None = None + account: Mapping[str, Any] | None = None + kind: Literal["ready"] = "ready" + + +@dataclass(frozen=True) +class AccountPending: + """Waiting on someone: :attr:`action` says who and what. + + ``request_id`` identifies an open ``request_account`` request (``None`` + on the ``sync_accounts`` path, where ``account_id`` may already exist). + The registry keeps it, so :meth:`~adcp.accounts.AccountRegistry.refresh` + reports when it moves and a repeat ``ensure`` never opens a duplicate. + """ + + action: PendingAction + via: AccountVia + request_id: str | None = None + account_id: str | None = None + status: str | None = None + kind: Literal["pending"] = "pending" + + +@dataclass(frozen=True) +class AccountAmbiguous: + """Several existing accounts match; pick one with :meth:`choose`. + + ``candidates`` are the wire ``account_candidate`` objects, limited to + accounts the caller can already reach. + """ + + candidates: tuple[Mapping[str, Any], ...] + request_id: str + via: AccountVia = "request_account" + kind: Literal["ambiguous"] = "ambiguous" + _choose: Callable[[str, str], Awaitable[AccountOutcome]] | None = field( + default=None, repr=False, compare=False + ) + + async def choose(self, account_id: str) -> AccountOutcome: + """Continue the request with the chosen candidate; returns the new outcome.""" + if self._choose is None: + raise RuntimeError("this outcome is not bound to an account registry") + if account_id not in {str(item.get("account_id")) for item in self.candidates}: + raise ValueError(f"{account_id!r} is not one of this request's candidates") + return await self._choose(self.request_id, account_id) + + +@dataclass(frozen=True) +class AccountRejected: + """The seller declined, or a buyer action expired. ``errors`` are wire errors.""" + + errors: tuple[Mapping[str, Any], ...] + via: AccountVia + request_id: str | None = None + kind: Literal["rejected"] = "rejected" + + +AccountOutcome = AccountReady | AccountPending | AccountAmbiguous | AccountRejected + + +def next_step(outcome: AccountOutcome) -> NextStep: + """What to do next, independent of the extensible ``kind`` vocabulary. + + * ``none``: the account is ready. + * ``wait_for_seller``: the seller is reviewing; call ``refresh()`` later + or wait for the ``account_request.changed`` webhook. + * ``send_human_to_url``: show ``action.url`` to a person (never fetch it). + * ``pick_candidate``: choose one of ``candidates``. + * ``contact_seller``: rejected, or a buyer action without a URL. + """ + if isinstance(outcome, AccountReady): + return "none" + if isinstance(outcome, AccountAmbiguous): + return "pick_candidate" + if isinstance(outcome, AccountRejected): + return "contact_seller" + if outcome.action.waiting_on == "buyer": + return "send_human_to_url" if outcome.action.url else "contact_seller" + return "wait_for_seller" + + +_CONTROL = re.compile(r"[\x00-\x08\x0b-\x1f\x7f-\x9f​-‏‪-‮⁦-⁩]") + + +def _untrusted(text: str, *, limit: int = 1000) -> str: + cleaned = " ".join(_CONTROL.sub("", text).split()) + if len(cleaned) > limit: + cleaned = cleaned[: limit - 1] + "…" + return cleaned.replace('"', "'") + + +def format_pending_action(action: PendingAction) -> str: + """Safe text for a CLI or a chat: who is waiting, what is needed, and where. + + The seller's ``message`` is untrusted: control and bidirectional + characters are removed, whitespace is collapsed, it is quoted and capped + at 1000 characters. The URL is shown for a person to open, never followed. + """ + who = "you" if action.waiting_on == "buyer" else "the seller" + lines = [f"Waiting on {who}: {_untrusted(action.kind, limit=64)}."] + if action.message: + lines.append(f'Seller says: "{_untrusted(action.message)}"') + if action.url: + lines.append(f"A person completes it at {_untrusted(action.url, limit=2048)}") + if action.expires_at: + lines.append(f"Expires {_untrusted(action.expires_at, limit=64)}.") + return "\n".join(lines) + + +def _errors(value: Any) -> tuple[Mapping[str, Any], ...]: + if not isinstance(value, Sequence) or isinstance(value, (str, bytes)): + return () + return tuple(dict(item) for item in value if isinstance(item, Mapping)) diff --git a/src/adcp/accounts.py b/src/adcp/accounts.py index 59e84b969..ca3d5143d 100644 --- a/src/adcp/accounts.py +++ b/src/adcp/accounts.py @@ -5,20 +5,30 @@ import asyncio import logging import uuid -from collections.abc import Mapping +from collections.abc import Awaitable, Callable, Mapping from copy import deepcopy -from dataclasses import dataclass +from dataclasses import dataclass, replace from typing import TYPE_CHECKING, Any, Literal, Protocol, cast from pydantic import TypeAdapter from typing_extensions import TypeForm from adcp.account_identity import account_key, account_key_payload +from adcp.account_outcomes import ( + AccountAmbiguous, + AccountOutcome, + AccountPending, + AccountReady, + AccountRejected, + AccountVia, + PendingAction, + _errors, +) from adcp.exceptions import ( AccountNotFoundError, AccountPaymentRequiredError, + AccountRequestError, AccountSetupRequiredError, - classify_task_error, ) from adcp.types import AccountReference, SyncAccountsRequest @@ -51,12 +61,21 @@ @dataclass(frozen=True) class AccountRecord: - """A successful provisioning result; pending setup still means provisioned.""" + """What the registry knows about one natural key at one seller. + + ``request_id``, ``request_state`` and ``pending_action`` track an + ``account.requests`` request (or a ``sync_accounts`` account pending + approval), so a repeat ``ensure`` returns its current state and never + opens a duplicate. + """ account_id: str | None status: str | None wire_key: dict[str, Any] | None = None wire_version: str | None = None + request_id: str | None = None + request_state: str | None = None + pending_action: dict[str, Any] | None = None class AccountStorage(Protocol): @@ -75,6 +94,17 @@ async def save(self, seller: str, key: str, record: AccountRecord) -> None: ... async def delete(self, seller: str, key: str) -> None: ... +class ListableAccountStorage(AccountStorage, Protocol): + """Optional: storage that can enumerate a seller's records. + + ``AccountRegistry.pending()`` and ``refresh()`` use it to find open + requests across restarts; without it they see the keys this process + touched. + """ + + async def items(self, seller: str) -> list[tuple[str, AccountRecord]]: ... + + class InMemoryAccountStorage: """Default storage; pass a durable adapter to survive process restarts.""" @@ -90,6 +120,9 @@ async def save(self, seller: str, key: str, record: AccountRecord) -> None: async def delete(self, seller: str, key: str) -> None: self._records.pop((seller, key), None) + async def items(self, seller: str) -> list[tuple[str, AccountRecord]]: + return [(key, record) for (owner, key), record in self._records.items() if owner == seller] + def _payload(value: Any) -> dict[str, Any]: if hasattr(value, "model_dump"): @@ -144,6 +177,64 @@ def _sync_key( return next(iter(matches.values())) if len(matches) == 1 else None +_OPEN_REQUEST_STATES = frozenset({"in_review", "ambiguous"}) +_SELLER_REVIEW = {"waiting_on": "seller", "kind": "review", "message": "The seller is reviewing"} + + +def _requests_supported(caps: Any) -> bool: + """Whether the seller advertises ``account.requests`` (request_account).""" + requests = getattr(getattr(caps, "account", None), "requests", None) + if requests is None: + return False + supported = getattr(requests, "supported", None) + if supported is None and hasattr(requests, "root"): + supported = getattr(requests.root, "supported", None) + if supported is None and isinstance(requests, Mapping): + supported = requests.get("supported") + return supported is True + + +def _key_payload(key: str) -> dict[str, Any]: + import json + + return cast(dict[str, Any], json.loads(key)) + + +def _request_outcome( + wire: Mapping[str, Any], *, choose: Callable[[str, str], Awaitable[AccountOutcome]] +) -> AccountOutcome: + """The outcome of one wire ``account_request`` record.""" + state = str(wire["state"]) + request_id = str(wire["request_id"]) + if state == "resolved": + account = wire.get("account") if isinstance(wire.get("account"), Mapping) else None + return AccountReady( + account_id=str(wire["account_id"]), + status=str((account or {}).get("status") or "active"), + via="request_account", + request_id=request_id, + account=account, + ) + if state == "in_review": + return AccountPending( + action=PendingAction.from_wire(wire.get("pending_action") or _SELLER_REVIEW), + via="request_account", + request_id=request_id, + ) + if state == "ambiguous": + return AccountAmbiguous( + candidates=tuple(dict(item) for item in wire.get("candidates") or ()), + request_id=request_id, + _choose=choose, + ) + return AccountRejected( + errors=_errors(wire.get("errors")) + or ({"code": "ACCOUNT_REQUEST_REJECTED", "message": "The seller declined the request"},), + via="request_account", + request_id=request_id, + ) + + def has_natural_reference(request: Any) -> bool: """Return True when ``request`` carries a buyer-declared natural account key.""" if "account" not in getattr(type(request), "model_fields", {}): @@ -160,6 +251,8 @@ def __init__(self, client: ADCPClient, storage: AccountStorage | None = None) -> self._storage = storage if storage is not None else InMemoryAccountStorage() self._seller = client.agent_config.agent_uri self._ensure_lock = asyncio.Lock() + #: Keys this process recorded, for storage that cannot enumerate. + self._touched: set[str] = set() async def get(self, key: AccountReference | Mapping[str, Any]) -> AccountRecord | None: return await self._storage.load(self._seller, account_key(key)) @@ -233,65 +326,301 @@ async def ensure( self, key: AccountReference | Mapping[str, Any], *, - billing: str, + billing: str | None = None, payment_terms: str | None = None, billing_entity: Any = None, - ) -> AccountRecord: - """Provision once through sync_accounts, retaining the ID and status. - - Billing and payment terms are supplied only on first provisioning; - use sync_accounts explicitly for subsequent settings changes. + currency: str | None = None, + timezone: str | None = None, + ) -> AccountOutcome: + """Get an account for a natural key at this seller; returns an :data:`AccountOutcome`. + + Routes by capability: ``request_account`` when the seller advertises + ``account.requests``, otherwise ``sync_accounts`` (which needs + ``billing``). Either way the outcome is one of + :class:`~adcp.account_outcomes.AccountReady`, + :class:`~adcp.account_outcomes.AccountPending`, + :class:`~adcp.account_outcomes.AccountAmbiguous` or + :class:`~adcp.account_outcomes.AccountRejected`; pending, ambiguous and + rejected are values, not errors. Only a failure to process the call + raises :class:`~adcp.exceptions.AccountRequestError`. + + Idempotent per (seller, buyer identity, natural key): a repeat call + returns the account, or the open request's current state, and never + opens a duplicate. ``currency`` and ``timezone`` join the key when it + does not carry them. Billing terms are sent only when an account is + first requested; change them later with ``request_account`` or + ``sync_accounts``. """ - ref = account_key_payload(TypeAdapter(AccountReference).validate_python(key)) + raw = _payload(TypeAdapter(AccountReference).validate_python(key)) + for name, value in (("currency", currency), ("timezone", timezone)): + if value is not None: + raw.setdefault(name, value) + ref = account_key_payload(TypeAdapter(AccountReference).validate_python(raw)) if "account_id" in ref: raise ValueError("ensure requires a buyer-declared natural key") + terms: dict[str, Any] = {} + if billing is not None: + terms["billing"] = billing + if payment_terms is not None: + terms["payment_terms"] = payment_terms + if billing_entity is not None: + terms["billing_entity"] = _payload(billing_entity) async with self._ensure_lock: + caps = await self._client.fetch_capabilities() + requests = _requests_supported(caps) existing = await self.get(ref) if existing is not None: - return existing - caps = await self._client.fetch_capabilities() + outcome = await self._known_outcome(ref, existing, requests=requests) + if outcome is not None: + return outcome + if requests: + return await self._request_account(ref, {**ref, **terms}) if getattr(caps.account, "require_operator_auth", False) is True: self._raise_setup("sync_accounts", "Discover an account_id through list_accounts") - entry = {**ref, "billing": billing} - if payment_terms is not None: - entry["payment_terms"] = payment_terms - if billing_entity is not None: - entry["billing_entity"] = _payload(billing_entity) - result = await self._client.sync_accounts( - SyncAccountsRequest.model_validate( + if billing is None: + raise ValueError( + "this seller provisions through sync_accounts, which needs billing=; " + "pass the invoiced party" + ) + return await self._sync_account(ref, {**ref, **terms}) + + async def _known_outcome( + self, ref: dict[str, Any], record: AccountRecord, *, requests: bool + ) -> AccountOutcome | None: + """The outcome a stored record already answers, or ``None`` to request again.""" + if record.request_id and record.request_state in _OPEN_REQUEST_STATES and requests: + refreshed = await self._read_requests([record.request_id]) + wire = refreshed.get(record.request_id) + if wire is not None: + return await self._record_request(ref, wire) + # The seller no longer knows the request (expired or dropped). + return None + if record.request_state == "rejected" and record.account_id is None: + return None + if record.account_id is None and record.status is None: + return None + via: AccountVia = "request_account" if record.request_id else "sync_accounts" + if record.status == "pending_approval": + return AccountPending( + action=PendingAction.from_wire(record.pending_action or _SELLER_REVIEW), + via=via, + request_id=record.request_id, + account_id=record.account_id, + status=record.status, + ) + return AccountReady( + account_id=record.account_id, + status=record.status, + via=via, + request_id=record.request_id, + ) + + def _failure(self, operation: str, result: Any, body: Mapping[str, Any]) -> AccountRequestError: + errors = body.get("errors") or [ + result.adcp_error + or {"code": "SERVICE_UNAVAILABLE", "message": result.error or f"{operation} failed"} + ] + return AccountRequestError(operation, list(errors), agent_id=self._client.agent_config.id) + + async def _request_account(self, ref: dict[str, Any], entry: dict[str, Any]) -> AccountOutcome: + from adcp.types import RequestAccountRequest + + result = await self._client.request_account( + RequestAccountRequest.model_validate({"idempotency_key": str(uuid.uuid4()), **entry}) + ) + body = _payload(result.data) + request = body.get("request") + if not result.success or body.get("adcp_error") or not isinstance(request, Mapping): + raise self._failure("request_account", result, body) + return await self._record_request(ref, request) + + async def _choose( + self, ref: dict[str, Any], request_id: str, account_id: str + ) -> AccountOutcome: + from adcp.types import RequestAccountRequest + + async with self._ensure_lock: + result = await self._client.request_account( + RequestAccountRequest.model_validate( { "idempotency_key": str(uuid.uuid4()), - "accounts": [entry], + "request_id": request_id, + "account_id": account_id, } ) ) body = _payload(result.data) + request = body.get("request") + if not result.success or body.get("adcp_error") or not isinstance(request, Mapping): + raise self._failure("request_account", result, body) + return await self._record_request(ref, request) + + async def _record_request(self, ref: dict[str, Any], wire: Mapping[str, Any]) -> AccountOutcome: + """Persist a request's state under its natural key and return its outcome.""" + outcome = _request_outcome( + wire, choose=lambda request_id, account_id: self._choose(ref, request_id, account_id) + ) + previous = await self.get(ref) + ready = outcome if isinstance(outcome, AccountReady) else None + await self._storage.save( + self._seller, + account_key(ref), + AccountRecord( + account_id=None if ready is None else ready.account_id, + status=None if ready is None else ready.status, + wire_key=previous.wire_key if previous is not None else None, + wire_version=previous.wire_version if previous is not None else None, + request_id=str(wire["request_id"]), + request_state=str(wire["state"]), + pending_action=( + outcome.action.to_wire() if isinstance(outcome, AccountPending) else None + ), + ), + ) + self._touched.add(account_key(ref)) + return outcome + + async def _sync_account(self, ref: dict[str, Any], entry: dict[str, Any]) -> AccountOutcome: + result = await self._client.sync_accounts( + SyncAccountsRequest.model_validate( + {"idempotency_key": str(uuid.uuid4()), "accounts": [entry]} + ) + ) + body = _payload(result.data) + if not result.success or body.get("errors"): + raise self._failure("sync_accounts", result, body) + rows = body.get("accounts", []) + if not rows: + raise self._failure("sync_accounts", result, body) + row = rows[0] + if row.get("action") == "failed" or row.get("status") == "rejected": + return AccountRejected( + errors=_errors(row.get("errors")) + or ({"code": "ACCOUNT_REJECTED", "message": "The seller declined the account"},), + via="sync_accounts", + ) + record = await self.get(ref) + if record is None: + self._raise_setup("sync_accounts", "Seller did not confirm account provisioning") + assert record is not None + self._touched.add(account_key(ref)) + if record.status == "pending_approval": + action = PendingAction.from_wire(row.get("setup") or _SELLER_REVIEW) + await self._storage.save( + self._seller, + account_key(ref), + replace(record, pending_action=action.to_wire()), + ) + return AccountPending( + action=action, + via="sync_accounts", + account_id=record.account_id, + status=record.status, + ) + return AccountReady(account_id=record.account_id, status=record.status, via="sync_accounts") + + async def _read_requests(self, request_ids: list[str]) -> dict[str, Mapping[str, Any]]: + """Read the caller's own requests back through ``list_accounts``.""" + from adcp.types import ListAccountsRequest + + found: dict[str, Mapping[str, Any]] = {} + cursor: str | None = None + for _page in range(100): + payload: dict[str, Any] = {"account_request_ids": request_ids} + if cursor: + payload["requests_pagination"] = {"cursor": cursor} + result = await self._client.list_accounts(ListAccountsRequest.model_validate(payload)) + body = _payload(result.data) if not result.success or body.get("errors"): - raise classify_task_error( - "sync_accounts", - body.get("errors") - or [ - result.adcp_error - or { - "code": "ACCOUNT_NOT_FOUND", - "message": result.error or "Provisioning failed", - } - ], - agent_id=self._client.agent_config.id, - ) - rows = body.get("accounts", []) - if not rows or rows[0].get("action") == "failed": - raise classify_task_error( - "sync_accounts", - (rows[0].get("errors") if rows else None) - or [{"code": "ACCOUNT_NOT_FOUND", "message": "Provisioning failed"}], - agent_id=self._client.agent_config.id, + raise self._failure("list_accounts", result, body) + await self.observe("list_accounts", payload, result, fetch_capabilities=False) + for item in body.get("account_requests") or (): + if isinstance(item, Mapping) and item.get("request_id"): + found[str(item["request_id"])] = item + pagination = body.get("account_requests_pagination") or {} + cursor = pagination.get("cursor") if pagination.get("has_more") else None + if not cursor: + break + return found + + async def _known(self) -> list[tuple[str, AccountRecord]]: + items = getattr(self._storage, "items", None) + if callable(items): + return list(await items(self._seller)) + known = [] + for key in sorted(self._touched): + record = await self._storage.load(self._seller, key) + if record is not None: + known.append((key, record)) + return known + + async def pending(self) -> list[AccountOutcome]: + """The outcomes still waiting: open requests and accounts pending approval. + + Read from storage; no network call. Use :meth:`refresh` to learn what moved. + """ + result: list[AccountOutcome] = [] + for _key, record in await self._known(): + if record.request_state == "ambiguous": + continue # its candidates live with the seller; refresh() reads them + if record.request_state == "in_review" or record.status == "pending_approval": + result.append( + AccountPending( + action=PendingAction.from_wire(record.pending_action or _SELLER_REVIEW), + via="request_account" if record.request_id else "sync_accounts", + request_id=record.request_id, + account_id=record.account_id, + status=record.status, + ) ) - record = await self.get(ref) - if record is None: - self._raise_setup("sync_accounts", "Seller did not confirm account provisioning") - assert record is not None - return record + return result + + async def refresh(self) -> list[AccountOutcome]: + """Re-read open requests and pending accounts; return the outcomes whose state moved. + + Call it from a cron job or an ``account_request.changed`` webhook. One + ``list_accounts`` read covers every open request (``account_request_ids``) + and the first page of accounts. + """ + known = await self._known() + open_requests = { + str(record.request_id): (key, record) + for key, record in known + if record.request_id and record.request_state in _OPEN_REQUEST_STATES + } + pending_accounts = { + key: record + for key, record in known + if not record.request_id and record.status == "pending_approval" + } + if not open_requests and not pending_accounts: + return [] + moved: list[AccountOutcome] = [] + if open_requests: + wire = await self._read_requests(sorted(open_requests)) + else: + from adcp.types import ListAccountsRequest + + listing = await self._client.list_accounts(ListAccountsRequest()) + await self.observe("list_accounts", {}, listing, fetch_capabilities=False) + wire = {} + for request_id, (key, record) in open_requests.items(): + current = wire.get(request_id) + if current is None: + continue + ref = _key_payload(key) + if str(current["state"]) == record.request_state and ( + current.get("pending_action") or None + ) == (record.pending_action or None): + continue + moved.append(await self._record_request(ref, current)) + for key, record in pending_accounts.items(): + latest = await self._storage.load(self._seller, key) + if latest is not None and latest.status != record.status: + outcome = await self._known_outcome(_key_payload(key), latest, requests=False) + if outcome is not None: + moved.append(outcome) + return moved async def observe( self, @@ -388,14 +717,24 @@ async def observe( if account_key(reduced) != account_key(key): wire_key = account_key_payload(reduced) wire_version = observed_version + status = row.get("status") await self._storage.save( self._seller, account_key(key), AccountRecord( account_id=row.get("account_id"), - status=row.get("status"), + status=status, wire_key=wire_key, wire_version=wire_version, + request_id=previous.request_id if previous is not None else None, + request_state=previous.request_state if previous is not None else None, + pending_action=( + (row.get("setup") or previous.pending_action) + if previous is not None and status == "pending_approval" + else row.get("setup") + if status == "pending_approval" + else None + ), ), ) diff --git a/src/adcp/exceptions.py b/src/adcp/exceptions.py index d27a1f7ee..451b38c08 100644 --- a/src/adcp/exceptions.py +++ b/src/adcp/exceptions.py @@ -953,6 +953,15 @@ class AccountPaymentRequiredError(_BuyerAccountError): """Account exists but needs payment before use.""" +class AccountRequestError(ADCPTaskError): + """``client.accounts.ensure`` could not be processed (AdCP 3.3). + + Raised only when the seller could not process the call; ``errors`` carry + the wire codes. A pending, ambiguous or rejected account is a value + (``AccountPending``, ``AccountAmbiguous``, ``AccountRejected``), never this. + """ + + ACCOUNT_ERROR_CODE_MAP: dict[str, type[ADCPTaskError]] = { "ACCOUNT_NOT_FOUND": AccountNotFoundError, "ACCOUNT_REQUIRED": AccountSetupRequiredError, diff --git a/tests/fixtures/public_api_snapshot.json b/tests/fixtures/public_api_snapshot.json index 2e06710e3..cfe5d7cf1 100644 --- a/tests/fixtures/public_api_snapshot.json +++ b/tests/fixtures/public_api_snapshot.json @@ -35,15 +35,21 @@ "AcceptancePolicyRule": "media_buy.acceptance_policy_rule", "AcceptancePolicySurface": "union[value:'account',value:'creative',value:'delivery',value:'format',value:'landing_page',value:'media_buy',value:'targeting']", "AcceptedLoss": "media_buy.legacy_purchase_continuation_input", + "AccountAmbiguous": "adcp.account_outcomes", "AccountAuthorization": "core.account_authorization", "AccountNotFoundError": "adcp.exceptions", + "AccountOutcome": "union[adcp.account_outcomes:AccountAmbiguous,adcp.account_outcomes:AccountPending,adcp.account_outcomes:AccountReady,adcp.account_outcomes:AccountRejected]", "AccountPaymentRequiredError": "adcp.exceptions", + "AccountPending": "adcp.account_outcomes", "AccountPolicy": "union[value:'auto',value:'off',value:'strict']", + "AccountReady": "adcp.account_outcomes", "AccountRecord": "adcp.accounts", "AccountReference": "union[core.account_ref:AccountReference1,core.account_ref:AccountReference2]", "AccountReferenceById": "core.account_ref:AccountReference1", "AccountReferenceByNaturalKey": "core.account_ref:AccountReference2", "AccountRegistry": "adcp.accounts", + "AccountRejected": "adcp.account_outcomes", + "AccountRequestError": "adcp.exceptions", "AccountScope": "enums.account_scope", "AccountSetupRequiredError": "adcp.exceptions", "AccountStorage": "adcp.accounts", @@ -365,6 +371,7 @@ "ListTasksResponse": "protocol.list_tasks_response", "ListTransformersRequest": "creative.list_transformers_request:ListTransformersRequestCreativeAgent", "ListTransformersResponse": "creative.list_transformers_response:ListTransformersResponseCreativeAgent", + "ListableAccountStorage": "adcp.accounts", "LogEventErrorResponse": "media_buy.log_event_response:LogEventResponse2", "LogEventRequest": "media_buy.log_event_request", "LogEventResponse": "union[media_buy.log_event_response:LogEventResponse1,media_buy.log_event_response:LogEventResponse2]", @@ -416,6 +423,7 @@ "PackageSignalTargetingGroup": "core.package_signal_targeting_group", "PackageSignalTargetingGroups": "core.package_signal_targeting_groups", "PaginationRequest": "core.pagination_request", + "PendingAction": "adcp.account_outcomes", "PendingOAuthAuthorization": "adcp.oauth", "PendingOAuthFlowStore": "adcp.oauth", "PgNotificationOutbox": "adcp.notification_outbox_pg", @@ -694,6 +702,7 @@ "filter_revoked_selectors": "adcp.adagents", "find_authorized_agent_entries": "adcp.adagents", "format_is_supported": "adcp.canonical_formats.compat_helpers", + "format_pending_action": "adcp.account_outcomes", "formats_are_equivalent": "adcp.canonical_formats.compat_helpers", "generate_webhook_challenge_value": "adcp.webhooks", "generate_webhook_idempotency_key": "adcp.webhooks", @@ -715,6 +724,7 @@ "inject_trace_headers": "adcp.observability", "is_tracing_available": "adcp.observability", "materialize_change_terms": "adcp.media_buy_actions", + "next_step": "adcp.account_outcomes", "normalize_assets_required": "adcp.utils.format_assets", "normalize_domain": "adcp.identifiers", "normalize_identifier": "adcp.identifiers", diff --git a/tests/test_account_lifecycle.py b/tests/test_account_lifecycle.py index 4cd5b47cc..faf0d40ac 100644 --- a/tests/test_account_lifecycle.py +++ b/tests/test_account_lifecycle.py @@ -11,6 +11,7 @@ from adcp import ( AccountNotFoundError, AccountPaymentRequiredError, + AccountReady, AccountRecord, AccountSetupRequiredError, ADCPClient, @@ -263,8 +264,9 @@ async def test_ensure_ignores_echoed_seller_assigned_timezone(): client = make_client(timezone={"mode": "seller_fixed", "fixed_timezone": "UTC"}) key = {name: value for name, value in KEY.items() if name != "timezone"} mock_sync(client) - record = await client.accounts.ensure(key, billing="agent") - assert record == AccountRecord("seller-handle", "active") + outcome = await client.accounts.ensure(key, billing="agent") + assert outcome == AccountReady("seller-handle", "active", via="sync_accounts") + assert await client.accounts.get(key) == AccountRecord("seller-handle", "active") @pytest.mark.asyncio diff --git a/tests/test_account_outcomes.py b/tests/test_account_outcomes.py new file mode 100644 index 000000000..6464e6e58 --- /dev/null +++ b/tests/test_account_outcomes.py @@ -0,0 +1,255 @@ +"""``client.accounts.ensure`` returns a discriminated AccountOutcome (AdCP 3.3). + +It routes to ``request_account`` when the seller advertises ``account.requests`` +and to ``sync_accounts`` otherwise. Pending, ambiguous and rejected are values; +only a failure to process the call raises ``AccountRequestError``. +""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import AsyncMock + +import pytest + +from adcp import ( + AccountAmbiguous, + AccountPending, + AccountReady, + AccountRejected, + AccountRequestError, + InMemoryAccountStorage, + PendingAction, + format_pending_action, + next_step, +) +from adcp.types.core import TaskResult, TaskStatus +from tests.test_account_lifecycle import KEY, make_client, mock_sync + +REQUESTS = {"supported": True, "request_task": "request_account", "read_task": "list_accounts"} +SIGN_UP = { + "waiting_on": "buyer", + "kind": "sign_up", + "message": "Sign the advertiser agreement", + "url": "https://seller.example/onboard/abc", + "expires_at": "2026-12-01T00:00:00Z", +} + + +def _record(state: str, **fields: Any) -> dict[str, Any]: + return { + "request_id": "req-1", + "state": state, + "requested": {"brand": KEY["brand"], "operator": KEY["operator"]}, + "created_at": "2026-11-01T00:00:00Z", + "updated_at": "2026-11-01T00:00:00Z", + **fields, + } + + +def _completed(data: dict[str, Any]) -> TaskResult: + return TaskResult(status=TaskStatus.COMPLETED, data=data) + + +def _requests_client(*records: dict[str, Any], storage=None): + client = make_client(requests=REQUESTS, storage=storage) + client.adapter.request_account = AsyncMock( + side_effect=[_completed({"status": "completed", "request": item}) for item in records] + ) + client.adapter.list_accounts = AsyncMock(return_value=_completed({"accounts": []})) + client.adapter.sync_accounts = AsyncMock() + client.adapter._parse_response = lambda raw, _type: raw + return client + + +def _listing(*records: dict[str, Any]) -> TaskResult: + return _completed({"accounts": [], "account_requests": list(records)}) + + +@pytest.mark.asyncio +async def test_requests_route_to_request_account_and_resolve_ready() -> None: + client = _requests_client( + _record( + "resolved", + account_id="acc-1", + resolution="created", + account={"account_id": "acc-1", "name": "Acme", "status": "active"}, + ) + ) + outcome = await client.accounts.ensure(KEY, billing="operator") + assert isinstance(outcome, AccountReady) + assert (outcome.account_id, outcome.status, outcome.via) == ( + "acc-1", + "active", + "request_account", + ) + assert next_step(outcome) == "none" + client.adapter.sync_accounts.assert_not_called() + sent = client.adapter.request_account.call_args.args[0] + assert sent["brand"]["domain"] == "acme.example" and sent["billing"] == "operator" + # Idempotent: the account is known, so nothing is sent again. + again = await client.accounts.ensure(KEY, billing="operator") + assert again.account_id == "acc-1" + assert client.adapter.request_account.call_count == 1 + + +@pytest.mark.asyncio +async def test_pending_is_a_value_and_a_repeat_ensure_never_opens_a_duplicate() -> None: + client = _requests_client(_record("in_review", pending_action=SIGN_UP)) + outcome = await client.accounts.ensure(KEY) + assert isinstance(outcome, AccountPending) + assert outcome.action == PendingAction.from_wire(SIGN_UP) and outcome.request_id == "req-1" + assert next_step(outcome) == "send_human_to_url" + client.adapter.list_accounts.return_value = _listing( + _record("in_review", pending_action=SIGN_UP) + ) + again = await client.accounts.ensure(KEY) + assert isinstance(again, AccountPending) + assert client.adapter.request_account.call_count == 1 + read = client.adapter.list_accounts.call_args.args[0] + assert read["account_request_ids"] == ["req-1"] + assert [item.request_id for item in await client.accounts.pending()] == ["req-1"] + + +@pytest.mark.asyncio +async def test_refresh_reports_requests_whose_state_moved() -> None: + storage = InMemoryAccountStorage() + client = _requests_client( + _record("in_review", pending_action={**SIGN_UP, "waiting_on": "seller", "kind": "review"}), + storage=storage, + ) + pending = await client.accounts.ensure(KEY) + assert next_step(pending) == "wait_for_seller" + client.adapter.list_accounts.return_value = _listing( + _record("in_review", pending_action={**SIGN_UP, "waiting_on": "seller", "kind": "review"}) + ) + assert await client.accounts.refresh() == [] + client.adapter.list_accounts.return_value = _listing( + _record("resolved", account_id="acc-9", resolution="matched") + ) + # A restarted process finds the open request through listable storage. + restarted = make_client(requests=REQUESTS, storage=storage) + restarted.adapter.list_accounts = client.adapter.list_accounts + restarted.adapter._parse_response = lambda raw, _type: raw + [moved] = await restarted.accounts.refresh() + assert isinstance(moved, AccountReady) and moved.account_id == "acc-9" + assert await restarted.accounts.pending() == [] + assert (await restarted.accounts.get(KEY)).account_id == "acc-9" + + +@pytest.mark.asyncio +async def test_ambiguous_requests_are_settled_by_choosing_a_candidate() -> None: + candidates = [ + {"account_id": "acc-a", "name": "Acme US"}, + {"account_id": "acc-b", "name": "Acme UK"}, + ] + client = _requests_client( + _record("ambiguous", candidates=candidates), + _record("resolved", account_id="acc-b", resolution="matched"), + ) + outcome = await client.accounts.ensure(KEY) + assert isinstance(outcome, AccountAmbiguous) and next_step(outcome) == "pick_candidate" + with pytest.raises(ValueError, match="candidates"): + await outcome.choose("acc-z") + chosen = await outcome.choose("acc-b") + assert isinstance(chosen, AccountReady) and chosen.account_id == "acc-b" + sent = client.adapter.request_account.call_args.args[0] + assert (sent["request_id"], sent["account_id"]) == ("req-1", "acc-b") + + +@pytest.mark.asyncio +async def test_a_rejected_request_is_a_value_and_a_later_ensure_asks_again() -> None: + client = _requests_client( + _record("rejected", errors=[{"code": "BRAND_NOT_AUTHORIZED", "message": "no"}]), + _record("in_review", pending_action=SIGN_UP), + ) + rejected = await client.accounts.ensure(KEY) + assert isinstance(rejected, AccountRejected) and next_step(rejected) == "contact_seller" + assert rejected.errors[0]["code"] == "BRAND_NOT_AUTHORIZED" + assert isinstance(await client.accounts.ensure(KEY), AccountPending) + assert client.adapter.request_account.call_count == 2 + + +@pytest.mark.asyncio +async def test_a_failure_to_process_raises_account_request_error() -> None: + client = _requests_client() + client.adapter.request_account = AsyncMock( + return_value=TaskResult( + status=TaskStatus.FAILED, + success=False, + adcp_error={"code": "SERVICE_UNAVAILABLE", "message": "down"}, + ) + ) + with pytest.raises(AccountRequestError) as failed: + await client.accounts.ensure(KEY) + assert failed.value.errors[0]["code"] == "SERVICE_UNAVAILABLE" + assert await client.accounts.get(KEY) is None + + +@pytest.mark.asyncio +async def test_sync_accounts_pending_approval_maps_to_pending() -> None: + client = make_client() + mock_sync(client, status="pending_approval", setup=SIGN_UP) + outcome = await client.accounts.ensure(KEY, billing="agent") + assert isinstance(outcome, AccountPending) + assert (outcome.via, outcome.account_id) == ("sync_accounts", "seller-handle") + assert outcome.action.url == SIGN_UP["url"] + # Memoized, with its action. + again = await client.accounts.ensure(KEY, billing="agent") + assert again == outcome + client.adapter.sync_accounts.assert_called_once() + + +@pytest.mark.asyncio +async def test_a_failed_sync_row_is_rejected_not_raised() -> None: + client = make_client() + mock_sync( + client, + action="failed", + status="rejected", + errors=[{"code": "BILLING_NOT_SUPPORTED", "message": "no agent billing"}], + ) + outcome = await client.accounts.ensure(KEY, billing="agent") + assert isinstance(outcome, AccountRejected) and outcome.via == "sync_accounts" + assert outcome.errors[0]["code"] == "BILLING_NOT_SUPPORTED" + + +@pytest.mark.asyncio +async def test_sync_accounts_needs_billing() -> None: + client = make_client() + mock_sync(client) + with pytest.raises(ValueError, match="billing="): + await client.accounts.ensure(KEY) + client.adapter.sync_accounts.assert_not_called() + + +@pytest.mark.asyncio +async def test_currency_and_timezone_join_the_key() -> None: + client = _requests_client(_record("resolved", account_id="acc-1", resolution="created")) + key = {name: value for name, value in KEY.items() if name not in {"currency", "timezone"}} + await client.accounts.ensure(key, currency="USD", timezone="UTC") + assert (await client.accounts.get(KEY)).account_id == "acc-1" + + +def test_format_pending_action_treats_the_message_as_untrusted() -> None: + action = PendingAction( + waiting_on="buyer", + kind="sign_up", + message='Click ‮here‬ "now"\x1b[31m\n\nplease', + url="https://seller.example/onboard/abc", + expires_at="2026-12-01T00:00:00Z", + ) + text = format_pending_action(action) + assert "‮" not in text and "\x1b" not in text + assert "Seller says: \"Click here 'now'[31m please\"" in text + assert "Waiting on you: sign_up." in text + assert "https://seller.example/onboard/abc" in text and "Expires 2026-12-01" in text + long = format_pending_action(PendingAction("seller", "review", "x" * 5000)) + assert len(long) < 1100 and long.startswith("Waiting on the seller") + + +def test_next_step_for_a_buyer_action_without_a_url() -> None: + pending = AccountPending( + action=PendingAction("buyer", "grant", "Grant access"), via="request_account" + ) + assert next_step(pending) == "contact_seller" From e779518ae362657b125eb60a8ec70a5fc1976411 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Fri, 30 Oct 2026 20:19:53 +0000 Subject: [PATCH 2/3] fix(accounts): read requests back by request_ids and dedupe across instances Review follow-ups on the ensure reshape: - Open requests are read back through list_accounts request_ids, the schema's field; account_request_ids was not one. - refresh() re-reads accounts pending approval with their own list_accounts call even when requests are open, because request reads page account_requests[] only. - request_account idempotency keys are deterministic per seller, request body and the request a new attempt follows. Registry instances sharing one buyer identity and durable storage send one key, so the seller replays one request instead of opening two. Refs #1695 Co-Authored-By: Claude Opus 5.5 (1M context) --- src/adcp/accounts.py | 43 ++++++++++++++++++++++------ tests/test_account_outcomes.py | 51 +++++++++++++++++++++++++++++++++- 2 files changed, 85 insertions(+), 9 deletions(-) diff --git a/src/adcp/accounts.py b/src/adcp/accounts.py index ca3d5143d..0de5cd7b1 100644 --- a/src/adcp/accounts.py +++ b/src/adcp/accounts.py @@ -194,6 +194,23 @@ def _requests_supported(caps: Any) -> bool: return supported is True +def _ensure_key(seller: str, body: Mapping[str, Any], previous: str | None = None) -> str: + """A deterministic ``request_account`` idempotency key. + + Registry instances sharing one buyer identity and durable storage send the + same key for the same request, so the seller replays one request instead + of opening two. ``previous`` (the request a new attempt follows, after a + rejection) starts a new key. + """ + import hashlib + import json + + digest = hashlib.sha256( + json.dumps([seller, body, previous], sort_keys=True, separators=(",", ":")).encode() + ).hexdigest() + return f"adcp-ensure-{digest[:48]}" + + def _key_payload(key: str) -> dict[str, Any]: import json @@ -425,8 +442,16 @@ def _failure(self, operation: str, result: Any, body: Mapping[str, Any]) -> Acco async def _request_account(self, ref: dict[str, Any], entry: dict[str, Any]) -> AccountOutcome: from adcp.types import RequestAccountRequest + previous = await self.get(ref) result = await self._client.request_account( - RequestAccountRequest.model_validate({"idempotency_key": str(uuid.uuid4()), **entry}) + RequestAccountRequest.model_validate( + { + "idempotency_key": _ensure_key( + self._seller, entry, None if previous is None else previous.request_id + ), + **entry, + } + ) ) body = _payload(result.data) request = body.get("request") @@ -443,7 +468,9 @@ async def _choose( result = await self._client.request_account( RequestAccountRequest.model_validate( { - "idempotency_key": str(uuid.uuid4()), + "idempotency_key": _ensure_key( + self._seller, {"request_id": request_id, "account_id": account_id} + ), "request_id": request_id, "account_id": account_id, } @@ -526,7 +553,7 @@ async def _read_requests(self, request_ids: list[str]) -> dict[str, Mapping[str, found: dict[str, Mapping[str, Any]] = {} cursor: str | None = None for _page in range(100): - payload: dict[str, Any] = {"account_request_ids": request_ids} + payload: dict[str, Any] = {"request_ids": request_ids} if cursor: payload["requests_pagination"] = {"cursor": cursor} result = await self._client.list_accounts(ListAccountsRequest.model_validate(payload)) @@ -579,7 +606,7 @@ async def refresh(self) -> list[AccountOutcome]: """Re-read open requests and pending accounts; return the outcomes whose state moved. Call it from a cron job or an ``account_request.changed`` webhook. One - ``list_accounts`` read covers every open request (``account_request_ids``) + ``list_accounts`` read covers every open request (``request_ids``) and the first page of accounts. """ known = await self._known() @@ -596,14 +623,14 @@ async def refresh(self) -> list[AccountOutcome]: if not open_requests and not pending_accounts: return [] moved: list[AccountOutcome] = [] - if open_requests: - wire = await self._read_requests(sorted(open_requests)) - else: + wire = await self._read_requests(sorted(open_requests)) if open_requests else {} + if pending_accounts: from adcp.types import ListAccountsRequest + # Request reads page account_requests[] only, so accounts pending + # approval are re-read on their own. listing = await self._client.list_accounts(ListAccountsRequest()) await self.observe("list_accounts", {}, listing, fetch_capabilities=False) - wire = {} for request_id, (key, record) in open_requests.items(): current = wire.get(request_id) if current is None: diff --git a/tests/test_account_outcomes.py b/tests/test_account_outcomes.py index 6464e6e58..14963a9bb 100644 --- a/tests/test_account_outcomes.py +++ b/tests/test_account_outcomes.py @@ -107,7 +107,7 @@ async def test_pending_is_a_value_and_a_repeat_ensure_never_opens_a_duplicate() assert isinstance(again, AccountPending) assert client.adapter.request_account.call_count == 1 read = client.adapter.list_accounts.call_args.args[0] - assert read["account_request_ids"] == ["req-1"] + assert read["request_ids"] == ["req-1"] assert [item.request_id for item in await client.accounts.pending()] == ["req-1"] @@ -253,3 +253,52 @@ def test_next_step_for_a_buyer_action_without_a_url() -> None: action=PendingAction("buyer", "grant", "Grant access"), via="request_account" ) assert next_step(pending) == "contact_seller" + + +@pytest.mark.asyncio +async def test_instances_sharing_a_buyer_send_one_idempotency_key() -> None: + first = _requests_client(_record("in_review", pending_action=SIGN_UP)) + second = _requests_client(_record("in_review", pending_action=SIGN_UP)) + await first.accounts.ensure(KEY, billing="operator") + await second.accounts.ensure(KEY, billing="operator") + keys = [ + client.adapter.request_account.call_args.args[0]["idempotency_key"] + for client in (first, second) + ] + assert keys[0] == keys[1] and keys[0].startswith("adcp-ensure-") + + +@pytest.mark.asyncio +async def test_refresh_rereads_pending_accounts_beside_open_requests() -> None: + from adcp.types import ListAccountsResponse + + timezone = { + "mode": "account_fixed", + "account_selection": "buyer_selected", + "supported_timezones": ["UTC"], + } + storage = InMemoryAccountStorage() + client = make_client(storage=storage, timezone=timezone) + mock_sync(client, status="pending_approval", setup=SIGN_UP) + await client.accounts.ensure(KEY, billing="agent") + requests = make_client(requests=REQUESTS, storage=storage, timezone=timezone) + requests.adapter._parse_response = lambda raw, _type: raw + requests.adapter.request_account = AsyncMock( + return_value=_completed( + {"status": "completed", "request": _record("in_review", pending_action=SIGN_UP)} + ) + ) + await requests.accounts.ensure({**KEY, "currency": "EUR"}) + active = ListAccountsResponse.model_validate( + {"accounts": [{**KEY, "account_id": "seller-handle", "name": "Acme", "status": "active"}]} + ) + requests.adapter.list_accounts = AsyncMock( + side_effect=[ + _listing(_record("in_review", pending_action=SIGN_UP)), + TaskResult(status=TaskStatus.COMPLETED, data=active), + ] + ) + moved = await requests.accounts.refresh() + assert [type(item).__name__ for item in moved] == ["AccountReady"] + assert moved[0].account_id == "seller-handle" + assert requests.adapter.list_accounts.call_count == 2 From 7873bac850877243ebe0aea6153c93c15394d4a2 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Fri, 30 Oct 2026 21:37:27 +0000 Subject: [PATCH 3/3] fix(accounts): compare normalized pending actions in refresh refresh() compared the raw wire pending_action with the normalized form the registry stores. A seller that adds a field to pending_action, or sends an in_review request without one, made an unchanged request read as moved on every poll. Both sides are now normalized before the comparison. Refs #1695 Co-Authored-By: Claude Opus 5.5 (1M context) --- src/adcp/accounts.py | 11 ++++++++--- tests/test_account_outcomes.py | 14 ++++++++++++++ 2 files changed, 22 insertions(+), 3 deletions(-) diff --git a/src/adcp/accounts.py b/src/adcp/accounts.py index 0de5cd7b1..060f61077 100644 --- a/src/adcp/accounts.py +++ b/src/adcp/accounts.py @@ -636,9 +636,14 @@ async def refresh(self) -> list[AccountOutcome]: if current is None: continue ref = _key_payload(key) - if str(current["state"]) == record.request_state and ( - current.get("pending_action") or None - ) == (record.pending_action or None): + # Compare the normalized action, as stored: unknown wire fields + # and an absent action must not read as a change on every poll. + action = ( + PendingAction.from_wire(current.get("pending_action") or _SELLER_REVIEW).to_wire() + if str(current["state"]) == "in_review" + else None + ) + if str(current["state"]) == record.request_state and action == record.pending_action: continue moved.append(await self._record_request(ref, current)) for key, record in pending_accounts.items(): diff --git a/tests/test_account_outcomes.py b/tests/test_account_outcomes.py index 14963a9bb..c9cdb3ef1 100644 --- a/tests/test_account_outcomes.py +++ b/tests/test_account_outcomes.py @@ -302,3 +302,17 @@ async def test_refresh_rereads_pending_accounts_beside_open_requests() -> None: assert [type(item).__name__ for item in moved] == ["AccountReady"] assert moved[0].account_id == "seller-handle" assert requests.adapter.list_accounts.call_count == 2 + + +@pytest.mark.asyncio +async def test_refresh_ignores_unknown_action_fields_and_an_absent_action() -> None: + client = _requests_client(_record("in_review")) + await client.accounts.ensure(KEY) # stored with the seller-review default + client.adapter.list_accounts.return_value = _listing(_record("in_review")) + assert await client.accounts.refresh() == [] + client2 = _requests_client(_record("in_review", pending_action=SIGN_UP)) + await client2.accounts.ensure(KEY) + client2.adapter.list_accounts.return_value = _listing( + _record("in_review", pending_action={**SIGN_UP, "x_vendor_hint": "ignored"}) + ) + assert await client2.accounts.refresh() == []