fix: A2A top-level failures return failed Task with AdCP envelope, not JSON-RPC InternalError

.duplication-baseline

@@ -1,5 +1,5 @@
 {
   "src": 36,
-  "tests": 89,
+  "tests": 88,
   "scripts": 0
 }

docs/test-obligations/bdd-traceability.yaml

@@ -858,6 +858,12 @@ mappings:
       upstream_refs: ["BR-UC-002-ext-d"]
       business_rules: []
       status: new
+    - adcp_scenario_id: "T-UC-002-ext-nl-unsupported"
+      adcp_feature: "BR-UC-002-create-media-buy.feature"
+      obligation_id: null
+      upstream_refs: ["transport-errors.mdx#Layer-Separation"]
+      business_rules: []
+      status: new
     - adcp_scenario_id: "T-UC-002-ext-e"
       adcp_feature: "BR-UC-002-create-media-buy.feature"
       obligation_id: null

src/a2a_server/adcp_a2a_server.py

@@ -147,23 +147,25 @@ def _dict_to_struct(d: dict) -> struct_pb2.Struct:
 
 
 def _internal_error_for(operation: str, exc: Exception) -> InternalError:
-    """Canonical InternalError shape for non-skill A2A boundary failures.
+    """Canonical InternalError shape for non-Task A2A boundary failures.
 
     Skill handlers raise typed ``AdCPError`` (or untyped exceptions that the
     dispatcher normalizes), and ``_handle_explicit_skill`` → ``on_message_send``
-    surface those as a two-layer envelope on a failed Task's DataPart. Non-skill
-    paths (``on_message_send`` fallthrough, NL handlers) historically picked their
-    own prefixes (``"Message processing failed: "``, ``"Error in ..."``)
-    for semantically identical untyped failures — divergence on the buyer-
-    facing wire message for the same condition.
+    surface those as a two-layer envelope on a failed Task's DataPart —
+    including the top-level ``on_message_send`` fallthrough, which per AdCP
+    3.1.x transport rules (building/operating/transport-errors.mdx "Layer
+    Separation") RETURNS a failed Task carrying the envelope artifact rather
+    than raising this helper. JSON-RPC errors are reserved for genuine
+    transport faults (``A2AError``) and methods with no Task to carry the
+    envelope.
 
-    Use this helper at every non-skill ``InternalError(...)`` raise site that
-    is NOT a deliberate protocol-level convention (see push-notif handlers
-    below). The canonical prefix is ``"{operation} failed: {exc}"`` so
-    storyboard runners can parse the failure uniformly.
+    Use this helper only at ``InternalError(...)`` raise sites that have no
+    async Task to attach an envelope artifact to. The canonical prefix is
+    ``"{operation} failed: {exc}"`` so storyboard runners can parse the
+    failure uniformly.
 
     The four ``on_*_task_push_notification_config`` JSON-RPC protocol methods use
-    this helper too — they have no async Task to carry a DataPart, so the two-layer
+    this helper — they have no async Task to carry a DataPart, so the two-layer
     envelope rides in the error's ``data`` field (``error.data["errors"][0]["code"]``
     / ``error.data["adcp_error"]``). ``InternalError`` stays an ``A2AError`` so the
     SDK's ``JsonRpcDispatcher`` serializes it as a structured JSON-RPC error; raising
@@ -220,6 +222,11 @@ def _build_failed_skill_result(skill_name: str, exc: Exception) -> dict[str, Any
             "success": False,
         }
 
+    async def _fail_task_with_webhook(self, task: Task, error: str) -> None:
+        """Mark a task failed and notify protocol webhooks with a required reason."""
+        task.status.CopyFrom(TaskStatus(state=TaskState.TASK_STATE_FAILED))
+        await self._send_protocol_webhook(task, status="failed", error=error)
+
     def _get_auth_token(self, context: ServerCallContext | None = None) -> str | None:
         """Extract Bearer token from ServerCallContext.
 
@@ -582,6 +589,11 @@ async def on_message_send(
             self._task_push_configs[task_id] = push_notification_config
         self.tasks[task_id] = task
 
+        # Initialized before the try so the outer error handler can always read
+        # it — a failure during auth-token extraction (before resolution) must
+        # not turn into a NameError inside the except block.
+        identity: ResolvedIdentity | None = None
+
         try:
             # Get authentication token
             auth_token = self._get_auth_token(context)
@@ -605,7 +617,6 @@ async def on_message_send(
             # ── Transport boundary: resolve identity ONCE ──
             # Like REST's _resolve_auth(), identity is resolved here and passed
             # to all downstream handlers. No handler should call resolve_identity().
-            identity: ResolvedIdentity | None = None
             if auth_token:
                 identity = self._resolve_a2a_identity(auth_token, require_valid_token=requires_auth, context=context)
             elif not requires_auth:
@@ -730,14 +741,10 @@ async def on_message_send(
                 successful_skills = [res["skill"] for res in results if res["success"]]
 
                 if failed_skills and not successful_skills:
-                    # All skills failed - mark task as failed
-                    task.status.CopyFrom(TaskStatus(state=TaskState.TASK_STATE_FAILED))
-
-                    # Send protocol-level webhook notification for failure
                     error_messages = [
                         res["error_envelope"]["errors"][0]["message"] for res in results if not res["success"]
                     ]
-                    await self._send_protocol_webhook(task, status="failed", error="; ".join(error_messages))
+                    await self._fail_task_with_webhook(task, "; ".join(error_messages))
 
                     return task
                 elif successful_skills:
@@ -864,9 +871,10 @@ async def on_message_send(
                 # ``_create_media_buy`` is an NL stub that always raises
                 # ``AdCPCapabilityNotSupportedError`` — the explicit-skill
                 # path is the spec contract for media buy creation. The
-                # outer error handler at on_message_send catches the raise
-                # and attaches a spec-compliant two-layer envelope to the
-                # failed Task artifact.
+                # outer error handler at on_message_send catches the raise,
+                # attaches a spec-compliant two-layer envelope to the failed
+                # Task artifact, and returns that failed Task (never a
+                # JSON-RPC error).
                 await self._create_media_buy(combined_text, identity)
             else:
                 # General help response
@@ -957,12 +965,13 @@ async def on_message_send(
                 principal_id=err_principal_id,
             )
 
-            # Send protocol-level webhook notification for failure if configured
-            task.status.CopyFrom(TaskStatus(state=TaskState.TASK_STATE_FAILED))
             # Attach error to task artifacts as a spec-compliant two-layer
             # envelope (same shape as failed-skill DataParts) so storyboard
             # runners can ``JSON.parse`` the artifact uniformly regardless of
-            # which failure path produced it.
+            # which failure path produced it. ``_build_error_envelope``
+            # normalizes untyped exceptions to base AdCPError (wire code
+            # SERVICE_UNAVAILABLE via ERROR_CODE_MAPPING) and passes
+            # through a typed AdCPError's own wire code.
             del task.artifacts[:]
             task.artifacts.append(
                 Artifact(
@@ -972,10 +981,18 @@ async def on_message_send(
                 )
             )
 
-            await self._send_protocol_webhook(task, status="failed")
-
-            # Raise A2A error instead of creating failed task
-            raise _internal_error_for("message processing", e)
+            await self._fail_task_with_webhook(task, str(e))
+
+            # Per AdCP 3.1.x transport rules (spec prose:
+            # building/operating/transport-errors.mdx "Layer Separation" and the
+            # two-layer error-handling model), application/task-execution
+            # failures MUST be returned in the task response body as a failed
+            # Task carrying the envelope artifact. JSON-RPC errors are reserved
+            # for genuine transport faults (A2AError, re-raised above). So we
+            # fall through to the shared store-and-return below — never raise
+            # InternalError here. Cross-transport execution of the same BDD
+            # scenario is tracked in #1574 once the transport-aware harness
+            # from #1430 is available.
 
         self.tasks[task_id] = task
         return task

tests/a2a_helpers.py

@@ -2,10 +2,15 @@
 
 Provides make_a2a_context() to build a ServerCallContext the same way
 AdCPCallContextBuilder.build() does in production, but without needing
-a Starlette request object.
+a Starlette request object, and extract_processing_error_envelope() to
+read the two-layer AdCP error envelope off a failed Task returned by
+on_message_send's outer error handler.
 """
 
+import json
+
 from a2a.server.context import ServerCallContext
+from google.protobuf import json_format
 
 from src.core.auth_context import AUTH_CONTEXT_STATE_KEY, AuthContext
 
@@ -28,3 +33,41 @@ def make_a2a_context(
     """
     auth_ctx = AuthContext(auth_token=auth_token, headers=headers or {})
     return ServerCallContext(state={AUTH_CONTEXT_STATE_KEY: auth_ctx})
+
+
+def extract_processing_error_envelope(task) -> dict:
+    """Read the two-layer AdCP envelope from a failed Task's processing_error artifact.
+
+    ``on_message_send``'s outer error handler attaches the envelope built by
+    ``AdCPRequestHandler._build_error_envelope`` to the failed Task as a
+    single ``processing_error`` artifact with one DataPart (AdCP 3.1.x
+    transport-errors.mdx "Layer Separation": application failures ride in
+    the task body, not JSON-RPC errors).
+    """
+    assert task.artifacts, "failed Task must carry the error envelope artifact"
+    artifact = task.artifacts[0]
+    assert artifact.name == "processing_error", f"expected processing_error artifact, got {artifact.name!r}"
+    part = artifact.parts[0]
+    assert part.HasField("data"), "envelope artifact part must be a DataPart"
+    return json.loads(json_format.MessageToJson(part.data))
+
+
+def make_mock_a2a_identity():
+    """Standard mock ResolvedIdentity for A2A handler unit tests."""
+    from tests.factories import PrincipalFactory
+
+    return PrincipalFactory.make_identity(
+        principal_id="test-principal",
+        tenant_id="test-tenant",
+        tenant={"tenant_id": "test-tenant"},
+        protocol="a2a",
+    )
+
+
+def make_nl_send_message_request(text: str):
+    """Build a minimal A2A SendMessageRequest carrying NL text (no skills)."""
+    from a2a.types import SendMessageRequest
+
+    from tests.utils.a2a_helpers import create_a2a_text_message
+
+    return SendMessageRequest(message=create_a2a_text_message(text))

tests/bdd/conftest.py

@@ -2584,6 +2584,9 @@ def pytest_collection_modifyitems(items: list[pytest.Item]) -> None:
     "T-UC-002-v31-idempotency-replay",
     "T-UC-002-v31-idempotency-missing",
 }
+_UC002_MEDIA_BUY_CREATE_WIRED: set[str] = _UC002_IDEMPOTENCY_WIRED | {
+    "transport-layer-separation",
+}
 
 # Admin scenarios have their own transport (Flask test_client / requests.Session).
 # They must NOT be parametrized across MCP/A2A/REST/IMPL API transports.
@@ -2843,13 +2846,10 @@ def _harness_env(request: pytest.FixtureRequest, ctx: dict) -> Generator[None, N
             with MediaBuyAccountEnv(e2e_config=ctx.get("e2e_config")) as env:
                 ctx["env"] = env
                 yield
-        elif marker_names & _UC002_IDEMPOTENCY_WIRED:
-            # v3.1 idempotency replay/missing scenarios — MediaBuyCreateEnv runs a
-            # real create_media_buy through every transport (the replay scenario
-            # creates once, then sends the same key again to exercise the
-            # production replay path). Only the two wired tags go live here; the
-            # remaining @idempotency-key scenarios (in-flight, expired, conflict,
-            # pattern, canonical) stay blanket-xfailed below until their
+        elif marker_names & _UC002_MEDIA_BUY_CREATE_WIRED:
+            # Wired create-media-buy scenarios — MediaBuyCreateEnv seeds the
+            # tenant/principal/product chain and drives the real transport path.
+            # Most UC-002 scenarios stay blanket-xfailed below until their
             # production gaps + steps are wired.
             request.getfixturevalue("integration_db")
             from tests.harness.media_buy_create import MediaBuyCreateEnv

tests/bdd/features/BR-UC-002-create-media-buy.feature

@@ -211,6 +211,17 @@ Feature: BR-UC-002 Create Media Buy
     # POST-F1: System state is unchanged on failure
     # POST-F2: Buyer knows what failed
     # POST-F3: Buyer knows how to fix the issue
+
+  @T-UC-002-ext-nl-unsupported @extension @error @transport-layer-separation @a2a
+  Scenario: NL media-buy request is unsupported -- returned as a failed task, not a transport error
+    Given an authenticated buyer
+    When the buyer sends a natural-language "create a media buy" request
+    Then the error code should be "UNSUPPORTED_FEATURE"
+    And the error recovery should be "correctable"
+    And the wire error envelope should carry code "UNSUPPORTED_FEATURE" with recovery "correctable"
+    # grounded: AdCP 3.1.0-beta.3 building/operating/transport-errors.mdx "Layer Separation"
+    # storyboard: ungraded, pending upstream obligation (#1574)
+    # drives the top-level on_message_send except -> failed Task envelope, not JSON-RPC
     # --- ext-e: Duplicate Products ---
 
   @T-UC-002-ext-e @extension @ext-e @error @post-f1 @post-f2 @post-f3

tests/bdd/steps/domain/uc002_create_media_buy.py

@@ -10,6 +10,9 @@
 
 from __future__ import annotations
 
+import asyncio
+from unittest.mock import MagicMock
+
 from pytest_bdd import given, parsers, then, when
 
 from tests.bdd.steps._harness_db import db_session as _db_session
@@ -61,6 +64,12 @@ def given_valid_request(ctx: dict) -> None:
     ctx.setdefault("account_ref", None)
 
 
+@given("an authenticated buyer")
+def given_authenticated_buyer(ctx: dict) -> None:
+    """Record that this scenario uses the harness-created buyer identity."""
+    ctx["has_auth"] = True
+
+
 @given(parsers.parse('a valid create_media_buy request with account "{account_id}"'))
 def given_request_with_account(ctx: dict, account_id: str) -> None:
     """Set up a create_media_buy request with account (short form)."""
@@ -187,6 +196,47 @@ def given_account_active(ctx: dict) -> None:
     AgentAccountAccessFactory(tenant_id=tenant.tenant_id, principal=principal, account=account)
 
 
+@when(parsers.parse('the buyer sends a natural-language "{request_text}" request'))
+def when_buyer_sends_nl_a2a_request(ctx: dict, request_text: str) -> None:
+    """Drive real A2A ``on_message_send`` with a natural-language text part."""
+    from a2a.server.routes.common import ServerCallContext
+    from a2a.types import Task, TaskState
+
+    from src.a2a_server.adcp_a2a_server import AdCPRequestHandler
+    from src.core.config_loader import set_current_tenant
+    from src.core.exceptions import AdCPError
+    from tests.a2a_helpers import make_nl_send_message_request
+    from tests.harness._base import _envelope_to_adcp_error
+    from tests.harness.transport import Transport
+    from tests.utils.a2a_helpers import extract_data_from_artifact
+
+    env = ctx["env"]
+    identity = env.identity_for(Transport.A2A)
+    set_current_tenant(identity.tenant)
+
+    handler = AdCPRequestHandler()
+    handler._get_auth_token = MagicMock(return_value=identity.auth_token)
+    handler._resolve_a2a_identity = MagicMock(return_value=identity)
+
+    async def _call() -> Task:
+        return await handler.on_message_send(make_nl_send_message_request(request_text), ServerCallContext())
+
+    try:
+        result = asyncio.run(_call())
+    except Exception as exc:
+        ctx["error"] = exc
+        return
+
+    ctx["response"] = result
+    if result.status.state == TaskState.TASK_STATE_FAILED:
+        if not result.artifacts:
+            ctx["error"] = AdCPError(f"A2A task failed without artifacts: {result.status}")
+            return
+        envelope = extract_data_from_artifact(result.artifacts[0])
+        ctx["wire_error_envelope"] = envelope
+        ctx["error"] = _envelope_to_adcp_error(envelope, fallback_message="A2A natural-language request failed")
+
+
 @given(parsers.parse("a create_media_buy request with account configuration {partition}"))
 def given_request_with_partition(ctx: dict, partition: str) -> None:
     """Set up request based on partition name (for Scenario Outline tables)."""

tests/bdd/steps/generic/then_error.py

@@ -335,6 +335,25 @@ def then_error_recovery(ctx: dict, recovery: str) -> None:
         raise AssertionError(f"Cannot check recovery on non-AdCPError: {type(error).__name__}")
 
 
+@then(parsers.parse('the wire error envelope should carry code "{code}" with recovery "{recovery}"'))
+def then_wire_error_envelope_shape(ctx: dict, code: str, recovery: str) -> None:
+    """Assert the two-layer AdCP envelope shape on the captured wire envelope.
+
+    Wire-envelope variant of the error-code/recovery steps above (tests/CLAUDE.md
+    § Error Verification Policy): asserts on ``ctx["wire_error_envelope"]``, the
+    envelope captured off the wire by the dispatching When step, rather than the
+    lossy reconstructed exception in ``ctx["error"]``.
+    """
+    from tests.helpers import assert_envelope_shape
+
+    envelope = ctx.get("wire_error_envelope")
+    assert envelope is not None, (
+        f"No wire error envelope captured in ctx — the operation succeeded or the "
+        f"When step does not populate ctx['wire_error_envelope']. response={ctx.get('response')!r}"
+    )
+    assert_envelope_shape(envelope, code, recovery=recovery)
+
+
 @then('the error should include a "suggestion" field')
 @then('the error should include "suggestion" field')
 def then_error_has_suggestion(ctx: dict) -> None: