fix(accounts): never serve destination_billing_entity on account reads

src/adcp/decisioning/account_projection.py

@@ -18,7 +18,7 @@
    every emit path the framework controls; adopters typically don't
    call them directly.
 
-The AdCP v3 spec marks two write-only paths that the framework MUST
+The AdCP v3 spec marks write-only paths that the framework MUST
 strip on every response:
 
 * :attr:`BusinessEntity.bank` — bank coordinates flow buyer→seller in
@@ -28,6 +28,11 @@
   credentials the seller persists for outbound ``check_governance``
   calls but MUST NOT echo to the buyer or land in the idempotency
   replay cache.
+* :attr:`Account.destination_billing_entity` — the staged billing
+  identity for an operator-domain handoff. The buyer sends it on a
+  ``sync_accounts`` request; it MUST NOT appear in any account read
+  model (``list_accounts`` rows, ``account_requests[]``,
+  ``request_account``, ``sync_accounts`` results, embedded accounts).
 
 The schemas describe both rules in docstrings; neither is structurally
 enforced by Pydantic. The helpers here ARE the structural enforcement.
@@ -80,18 +85,22 @@ def project_account_for_response(account: Account) -> Account:
     marks bank details as write-only. Adopters that persist the full
     :class:`BusinessEntity` (with bank populated for invoicing) MUST
     project through this helper before serializing on any response.
+    Also clears :attr:`Account.destination_billing_entity`, a
+    request-only staging field that MUST NOT appear in account read
+    models.
 
-    Returns the input unchanged when ``billing_entity`` is ``None``
-    or ``billing_entity.bank`` is already absent — defensive copy
-    via ``model_copy()`` so callers can mutate the returned object
-    freely without touching the caller's input.
+    Returns an unchanged copy when neither field needs stripping —
+    defensive copy via ``model_copy()`` so callers can mutate the
+    returned object freely without touching the caller's input.
 
     The original ``account`` object is not modified.
     """
-    if account.billing_entity is None or account.billing_entity.bank is None:
-        return account.model_copy()
-    safe_billing_entity = account.billing_entity.model_copy(update={"bank": None})
-    return account.model_copy(update={"billing_entity": safe_billing_entity})
+    update: dict[str, Any] = {}
+    if account.destination_billing_entity is not None:
+        update["destination_billing_entity"] = None
+    if account.billing_entity is not None and account.billing_entity.bank is not None:
+        update["billing_entity"] = account.billing_entity.model_copy(update={"bank": None})
+    return account.model_copy(update=update) if update else account.model_copy()
 
 
 def project_business_entity_for_response(entity: BusinessEntity) -> BusinessEntity:
@@ -385,13 +394,18 @@ def _scrub_dict(value: dict[str, Any]) -> dict[str, Any]:
     * ``billing_entity.bank`` — write-only bank coordinates.
     * ``notification_configs[i].authentication.credentials`` — legacy
       webhook bearer/HMAC secret.
+    * ``destination_billing_entity`` — request-only handoff staging
+      field; dropped wholesale (non-bank fields included) because it
+      MUST NOT appear in any account read model.
 
     Walks recursively into nested dicts and lists. Returns a NEW dict —
     the input is not mutated, so callers (idempotency replay cache,
     middleware) can rely on input stability.
     """
     out: dict[str, Any] = {}
     for key, sub in value.items():
+        if key == "destination_billing_entity":
+            continue
         if key == "governance_agents" and isinstance(sub, list):
             out[key] = [
                 (
@@ -483,8 +497,9 @@ def strip_credentials_from_wire_result(method_name: str, result: Any) -> Any:
 
     Defense-in-depth boundary called by the dispatcher on every
     response that may surface an :class:`Account` envelope. Removes
-    ``governance_agents[i].authentication`` and ``billing_entity.bank``
-    recursively — the same fields the typed projections
+    ``governance_agents[i].authentication``, ``billing_entity.bank`` and
+    the request-only ``destination_billing_entity`` recursively — the
+    same fields the typed projections
     (:func:`to_wire_account`, :func:`to_wire_sync_governance_row`)
     strip when the adopter returns the framework's typed dataclasses.
 

tests/test_account_projection.py

@@ -122,6 +122,21 @@ def test_project_account_does_not_mutate_input() -> None:
     assert account.billing_entity.bank is not None
 
 
+def test_project_account_clears_destination_billing_entity() -> None:
+    """Request-only handoff staging never rides an account read model."""
+    account = _account_fixture().model_copy(
+        update={"destination_billing_entity": {"legal_name": "Destination GmbH"}}
+    )
+
+    projected = project_account_for_response(account)
+
+    assert projected.destination_billing_entity is None
+    assert "destination_billing_entity" not in projected.model_dump(mode="json", exclude_none=True)
+    assert projected.billing_entity is not None
+    assert projected.billing_entity.bank is None
+    assert account.destination_billing_entity == {"legal_name": "Destination GmbH"}
+
+
 def test_project_account_when_billing_entity_is_none() -> None:
     """No billing_entity to project — return a defensive copy."""
     account = Account(account_id="acct_x", name="x", status="active")

tests/test_account_requests_seller.py

@@ -33,6 +33,7 @@
     ListAccountsResponse,
     RequestAccountRequest,
     RequestAccountResponse,
+    SyncAccountsRequest,
 )
 from adcp.types.core import TaskResult, TaskStatus
 
@@ -293,6 +294,7 @@ async def list_accounts(params: dict[str, Any]) -> TaskResult:
 _BANK = {"account_holder": "Acme Inc", "iban": "DE89370400440532013000"}
 _GOVERNANCE_SECRET = "gov-secret-token-0123456789abcdef"
 _WEBHOOK_SECRET = "webhook-secret-0123456789abcdefghij"
+_DESTINATION = {"legal_name": "Confidential Destination GmbH", "bank": dict(_BANK)}
 
 
 def _credentialed_account() -> dict[str, Any]:
@@ -314,14 +316,17 @@ def _credentialed_account() -> dict[str, Any]:
                 "authentication": {"schemes": ["Bearer"], "credentials": _WEBHOOK_SECRET},
             }
         ],
+        "destination_billing_entity": dict(_DESTINATION),
     }
 
 
 def _assert_scrubbed(account: dict[str, Any]) -> None:
     assert account["billing_entity"] == {"legal_name": "Acme Inc"}
     assert account["governance_agents"] == [{"url": "https://governance.example"}]
     assert "credentials" not in account["notification_configs"][0]["authentication"]
-    for secret in (_BANK["iban"], _GOVERNANCE_SECRET, _WEBHOOK_SECRET):
+    # Request-only staging field: never on an account read, not even its non-bank fields.
+    assert "destination_billing_entity" not in account
+    for secret in (_BANK["iban"], _GOVERNANCE_SECRET, _WEBHOOK_SECRET, _DESTINATION["legal_name"]):
         assert secret not in repr(account)
 
 
@@ -347,3 +352,40 @@ async def test_account_requests_are_scrubbed_like_the_accounts_beside_them() ->
     )
     assert answered["request"]["state"] == "resolved"
     _assert_scrubbed(answered["request"]["account"])
+
+
+@pytest.mark.asyncio
+async def test_list_accounts_rows_never_serve_destination_billing_entity() -> None:
+    class _LooseRowStore(_Store):
+        def list(self, filter: dict[str, Any] | None = None, ctx: Any = None) -> Any:
+            return [_credentialed_account()]
+
+    listed = await _handler(_LooseRowStore()).list_accounts(ListAccountsRequest(), _ctx())
+    (row,) = listed["accounts"]
+    _assert_scrubbed(row)
+
+
+@pytest.mark.asyncio
+async def test_sync_accounts_results_never_serve_destination_billing_entity() -> None:
+    class _EchoingStore(_Store):
+        def upsert(self, refs: list[Any], ctx: Any = None) -> list[dict[str, Any]]:
+            # The natural echo idiom: spread the stored account onto the result row.
+            return [{**_credentialed_account(), **KEY, "action": "updated"}]
+
+    synced = await _handler(_EchoingStore()).sync_accounts(
+        SyncAccountsRequest.model_validate(
+            {
+                "idempotency_key": "k" * 16,
+                "accounts": [
+                    {
+                        **KEY,
+                        "billing": "operator",
+                        "destination_billing_entity": {"legal_name": "Destination GmbH"},
+                    }
+                ],
+            }
+        ),
+        _ctx(),
+    )
+    (row,) = synced["accounts"]
+    _assert_scrubbed(row)