fix(server): A2A auth middleware populates current_principal contextvars (closes #590)

src/adcp/server/auth.py

@@ -625,6 +625,13 @@ class A2ABearerAuthMiddleware:
     handlers see ``ServerCallContext.user.user_name`` populated with
     the principal's ``caller_identity`` without a custom builder.
 
+    Also populates :data:`current_principal`, :data:`current_tenant`,
+    and :data:`current_principal_metadata` for the duration of the
+    downstream call — symmetric with
+    :class:`BearerTokenAuthMiddleware`'s contract. Adopters reading
+    ``current_principal.get()`` from a platform method see identical
+    state on MCP and A2A.
+
     Composition order matters when ``transport="both"`` is in play:
     wrap the per-leg apps before any outer dispatcher closes over
     them. See ``serve.py:_build_mcp_and_a2a_app`` for the wiring.
@@ -671,7 +678,25 @@ async def __call__(self, scope: Any, receive: Any, send: Any) -> None:
             principal_metadata=dict(principal.metadata) if principal.metadata else None,
         )
         scope["auth"] = principal
-        await self._app(scope, receive, send)
+
+        # Populate the same ContextVars MCP's ``BearerTokenAuthMiddleware``
+        # sets, so adopters reading ``current_principal.get()`` (or the
+        # other two) from a platform method see identical state across
+        # transports. Without this, A2A handlers fall through to the
+        # ``None`` default while MCP handlers see the principal — a silent
+        # transport-coupled divergence that breaks tenant policies that
+        # require principal-bound calls. See issue #590.
+        principal_token = current_principal.set(principal.caller_identity)
+        tenant_token = current_tenant.set(principal.tenant_id)
+        metadata_token = current_principal_metadata.set(
+            dict(principal.metadata) if principal.metadata else None
+        )
+        try:
+            await self._app(scope, receive, send)
+        finally:
+            current_principal.reset(principal_token)
+            current_tenant.reset(tenant_token)
+            current_principal_metadata.reset(metadata_token)
 
     def _authenticate_scope(self, scope: Any) -> Principal | None:
         """Read + validate the bearer header off raw ASGI scope.

tests/test_serve_auth_both.py

@@ -33,6 +33,9 @@
     A2ABearerAuthMiddleware,
     BearerTokenAuth,
     Principal,
+    current_principal,
+    current_principal_metadata,
+    current_tenant,
     validator_from_token_map,
 )
 
@@ -94,6 +97,91 @@ async def inner(scope: Any, _receive: Any, _send: Any) -> None:
         assert "auth" in passed_scope
         assert passed_scope["auth"].caller_identity == "p-acme"
 
+    @pytest.mark.asyncio
+    async def test_valid_token_populates_contextvars(self):
+        """Issue #590: A2A path must populate ``current_principal``,
+        ``current_tenant``, and ``current_principal_metadata`` for the
+        duration of the downstream call — symmetric with MCP's
+        ``BearerTokenAuthMiddleware``. Adopters reading
+        ``current_principal.get()`` from a platform method previously
+        saw ``None`` on A2A while MCP worked, breaking tenant policies
+        that require principal-bound calls.
+        """
+        observed: dict[str, Any] = {}
+
+        async def inner(_scope: Any, _receive: Any, _send: Any) -> None:
+            observed["principal"] = current_principal.get()
+            observed["tenant"] = current_tenant.get()
+            observed["metadata"] = current_principal_metadata.get()
+
+        mw = A2ABearerAuthMiddleware(
+            inner,
+            BearerTokenAuth(
+                validate_token=validator_from_token_map(
+                    {
+                        "good-token": Principal(
+                            caller_identity="alice",
+                            tenant_id="t1",
+                            metadata={"role": "buyer"},
+                        )
+                    }
+                )
+            ),
+        )
+        await mw(
+            self._scope(headers=[(b"authorization", b"Bearer good-token")]),
+            lambda: None,
+            lambda _: None,
+        )
+
+        assert observed["principal"] == "alice"
+        assert observed["tenant"] == "t1"
+        assert observed["metadata"] == {"role": "buyer"}
+
+    @pytest.mark.asyncio
+    async def test_contextvars_reset_after_call(self):
+        """The middleware must reset the ContextVars after the downstream
+        call returns, so a later request on a shared task doesn't read
+        a stale principal. Mirrors ``BearerTokenAuthMiddleware``'s
+        unconditional-reset contract."""
+
+        async def inner(_scope: Any, _receive: Any, _send: Any) -> None:
+            # Inside the call the contextvars are populated.
+            assert current_principal.get() == "p-acme"
+
+        mw = A2ABearerAuthMiddleware(inner, _auth())
+        # Pre-call: the contextvars hold their default (None).
+        assert current_principal.get() is None
+        await mw(
+            self._scope(headers=[(b"authorization", b"Bearer good-token")]),
+            lambda: None,
+            lambda _: None,
+        )
+        # Post-call: reset back to default — no leakage.
+        assert current_principal.get() is None
+        assert current_tenant.get() is None
+        assert current_principal_metadata.get() is None
+
+    @pytest.mark.asyncio
+    async def test_contextvars_reset_even_on_inner_exception(self):
+        """If the downstream app raises, the ContextVars must still
+        reset — otherwise a buggy handler poisons the contextvars for
+        every subsequent request that shares this task."""
+
+        async def boom(_scope: Any, _receive: Any, _send: Any) -> None:
+            raise RuntimeError("inner blew up")
+
+        mw = A2ABearerAuthMiddleware(boom, _auth())
+        with pytest.raises(RuntimeError, match="inner blew up"):
+            await mw(
+                self._scope(headers=[(b"authorization", b"Bearer good-token")]),
+                lambda: None,
+                lambda _: None,
+            )
+        assert current_principal.get() is None
+        assert current_tenant.get() is None
+        assert current_principal_metadata.get() is None
+
     @pytest.mark.asyncio
     async def test_missing_header_returns_401(self):
         sent: list[dict] = []