fix(x402): preserve handled failures after waiter cancellation

src/inflowpay/x402/_payment.py

@@ -106,8 +106,10 @@ async def await_payload(
         completion = self._completion
         self._waiters += 1
         try:
-            # A cancelled waiter must not cancel another caller's wait for this same payment.
-            result = await asyncio.shield(completion)
+            # Keep cancellation local to this waiter without shield() logging late
+            # hook errors on Python 3.14; _observe_completion already observes them.
+            await asyncio.wait((completion,))
+            result = completion.result()
             return EncodedPayment(
                 result.encoded_payload,
                 result.payment_payload.model_copy(deep=True),

src/inflowpay/x402/buyer.py

@@ -131,7 +131,11 @@ async def get_supported(self, *, refresh: bool = False) -> SupportedResponse:
             if self._refresh is None:
                 self._refresh = asyncio.create_task(self._refresh_supported())
                 self._refresh.add_done_callback(self._finish_refresh)
-            await asyncio.shield(self._refresh)
+            # shield() logs late failures after cancellation on Python 3.14,
+            # even though _finish_refresh observes them. Keep the shared request independent.
+            task = self._refresh
+            await asyncio.wait((task,))
+            task.result()
         return self._supported.model_copy(deep=True)
 
     async def _refresh_supported(self) -> None:

src/inflowpay/x402/seller.py

@@ -33,7 +33,11 @@ async def get(self, refresh: bool = False) -> _T:
             if self.task is None:
                 self.task = asyncio.create_task(self._fetch())
                 self.task.add_done_callback(self._finish)
-            await asyncio.shield(self.task)
+            # shield() logs late failures after cancellation on Python 3.14,
+            # even though _finish observes them. Keep the shared request independent.
+            task = self.task
+            await asyncio.wait((task,))
+            task.result()
         assert self.value is not None
         return self.value.model_copy(deep=True)
 

tests/test_x402_buyer.py

@@ -796,6 +796,106 @@ async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
                     await task
 
 
+@pytest.mark.parametrize("remaining_waiter", [False, True])
+async def test_cancelled_refresh_waiter_does_not_log_late_failure(remaining_waiter: bool) -> None:
+    class BlockingPlatform(Platform):
+        async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
+            if request.url.path.endswith("x402-supported") and self.requests:
+                self.poll_started.set()
+                await self.release.wait()
+            return await super().handle_async_request(request)
+
+    platform = BlockingPlatform()
+    observed: list[dict[str, Any]] = []
+    loop = asyncio.get_running_loop()
+    previous = loop.get_exception_handler()
+    loop.set_exception_handler(lambda _, event: observed.append(event))
+    try:
+        async with await buyer(platform) as client:
+            first = asyncio.create_task(client.get_supported(refresh=True))
+            await platform.poll_started.wait()
+            shared = client._refresh
+            assert shared is not None
+            second = (
+                asyncio.create_task(client.get_supported(refresh=True))
+                if remaining_waiter
+                else None
+            )
+            await asyncio.sleep(0)
+            first.cancel()
+            with pytest.raises(asyncio.CancelledError):
+                await first
+            assert not shared.done()
+            platform.supported = httpx.Response(503)
+            platform.release.set()
+            await asyncio.gather(shared, return_exceptions=True)
+            if second is not None:
+                with pytest.raises(InflowApiError) as failure:
+                    await second
+                assert failure.value is shared.exception()
+            await asyncio.sleep(0)
+            assert observed == []
+            assert client._refresh is None
+            assert client.supports(REQUIREMENT)
+            assert len(platform.requests) == 2
+            platform.supported = SUPPORTED
+            await client.get_supported(refresh=True)
+            assert len(platform.requests) == 3
+    finally:
+        loop.set_exception_handler(previous)
+
+
+@pytest.mark.parametrize("remaining_waiter", [False, True])
+async def test_cancelled_payment_waiter_preserves_late_hook_error(remaining_waiter: bool) -> None:
+    platform = Platform()
+    started, release = asyncio.Event(), asyncio.Event()
+    failure = RuntimeError("application hook failed")
+    calls = 0
+    observed: list[dict[str, Any]] = []
+    loop = asyncio.get_running_loop()
+    previous = loop.get_exception_handler()
+    loop.set_exception_handler(lambda _, event: observed.append(event))
+    try:
+        async with await buyer(platform) as client:
+
+            async def after(_: PaymentCreatedContext) -> None:
+                nonlocal calls
+                calls += 1
+                started.set()
+                await release.wait()
+                raise failure
+
+            client.on_after_payment_creation(after)
+            payment = await client.prepare(REQUIREMENT, RESOURCE)
+            first = asyncio.create_task(payment.await_payload())
+            await started.wait()
+            completion = payment._completion
+            assert completion is not None
+            second = asyncio.create_task(payment.await_payload()) if remaining_waiter else None
+            await asyncio.sleep(0)
+            first.cancel()
+            with pytest.raises(asyncio.CancelledError):
+                await first
+            assert not completion.done()
+            release.set()
+            await asyncio.gather(completion, return_exceptions=True)
+            if second is not None:
+                with pytest.raises(RuntimeError) as caught:
+                    await second
+                assert caught.value is failure
+            await asyncio.sleep(0)
+            assert observed == []
+            for _ in range(2):
+                with pytest.raises(RuntimeError) as caught:
+                    await payment.await_payload()
+                assert caught.value is failure
+            assert calls == 1
+            assert payment._completion is completion
+            assert not any(r.url.path.endswith("/cancel") for r in platform.requests)
+    finally:
+        loop.set_exception_handler(previous)
+
+
 async def test_expired_capabilities_refresh() -> None:
     platform = Platform()
     async with await buyer(platform) as client:

tests/test_x402_seller.py

@@ -264,6 +264,48 @@ async def test_refresh_concurrency_cancellation_and_failure() -> None:
     assert platform.closed
 
 
+@pytest.mark.parametrize("operation", ["config", "supported"])
+@pytest.mark.parametrize("remaining_waiter", [False, True])
+async def test_cancelled_refresh_waiter_does_not_log_late_failure(
+    operation: str, remaining_waiter: bool
+) -> None:
+    platform = Platform()
+    observed: list[dict[str, Any]] = []
+    loop = asyncio.get_running_loop()
+    previous = loop.get_exception_handler()
+    loop.set_exception_handler(lambda _, event: observed.append(event))
+    try:
+        async with await Seller.create(platform.options()) as seller:
+            platform.block = operation
+            refresh = seller.config if operation == "config" else seller.get_supported
+            first = asyncio.create_task(refresh(refresh=True))
+            await platform.started.wait()
+            cache = seller._config if operation == "config" else seller._supported
+            shared = cache.task
+            assert shared is not None
+            second = asyncio.create_task(refresh(refresh=True)) if remaining_waiter else None
+            await asyncio.sleep(0)
+            first.cancel()
+            with pytest.raises(asyncio.CancelledError):
+                await first
+            assert not shared.done()
+            setattr(platform, operation, httpx.Response(503))
+            platform.release.set()
+            await asyncio.gather(shared, return_exceptions=True)
+            if second is not None:
+                with pytest.raises(InflowApiError) as failure:
+                    await second
+                assert failure.value is shared.exception()
+            await asyncio.sleep(0)
+            assert observed == []
+            assert cache.task is None
+            assert sum(r.url.path.endswith(operation) for r in platform.requests) == 2
+            await refresh()
+            assert len(platform.requests) == 3
+    finally:
+        loop.set_exception_handler(previous)
+
+
 async def test_fixed_facilitator_snapshot_and_recreation() -> None:
     platform = Platform()
     async with await Facilitator.create(platform.options()) as facilitator: