feat(accounts)!: ensure returns an AccountOutcome and routes to request_account

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,20 +57,59 @@ 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(
     GetProductsRequest(buying_mode="wholesale", account=key), account_policy="strict",
 )
 ```
 
-`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.

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,

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