fix(telemetry): let trace_function alone decide a span's status (#1309)

# Description

`trace_function` opened its span with `start_as_current_span`'s defaults
(`record_exception=True`, `set_status_on_exception=True`). OTel's
`use_span` therefore recorded any `Exception` leaving the block and
marked the span `ERROR`, on top of what the wrapper had already done:

- `QueueShutDown` still ended as `ERROR` on every queue teardown, even
though #1075 treats it as a non-error. In practice that is one `ERROR`
`EventQueueSource.dequeue_event` span per streaming request.
- Every exception was recorded twice on the span, and OTel's `Type:
message` description replaced the wrapper's `str(e)`.

## Change

- `src/a2a/utils/telemetry.py`: pass `record_exception=False,
set_status_on_exception=False` to `start_as_current_span` in both the
async and the sync wrapper. The wrapper already records the exception
and sets the status itself.
- `tests/utils/test_telemetry.py`: add
`test_span_status_with_real_otel_sdk`. It goes through a real
`TracerProvider` + `InMemorySpanExporter`, because the existing mocked
tracer can't see what OTel does on exit. It is parametrized over:
  - `QueueShutDown`, expected not `ERROR`;
  - `ValueError`, expected `ERROR`.

Both cases check that the span has exactly one `exception` event. Both
fail on `main` and pass with this change.

## Before / after

The reproduction script from #1308 (a2a-sdk `main` at 6ff0a82,
opentelemetry-sdk 1.42.1):

```
# before
shutdown ERROR 'AsyncQueueShutDown: ' ['exception', 'exception']
boom ERROR 'ValueError: real failure' ['exception', 'exception']

# after
shutdown UNSET None ['exception']
boom ERROR 'real failure' ['exception']
```

## Checks

- `./scripts/lint.sh`: ruff check, ruff format and ty all pass.
- `uv run pytest --cov=src`: 2206 passed, 93 % coverage.
- **Pre-existing failure:**
`tests/integration/cross_version/client_server/test_client_server.py`
fails locally on `main` too, without this change (the 0.3 server/client
subprocess doesn't start on my machine). This PR doesn't touch it.

- [x] Follow the [`CONTRIBUTING`
Guide](https://github.com/a2aproject/a2a-python/blob/main/CONTRIBUTING.md).
- [x] Make your Pull Request title in the
<https://www.conventionalcommits.org/> specification.
- [x] Ensure the tests and linter pass (Run `bash scripts/format.sh`
from the repository root to format)
- [x] Appropriate docs were updated (if necessary)

Fixes #1308 🦕

Co-authored-by: Atenea Agent @svc_atenea_gitlab

Co-authored-by: mykytanetipa <mykytanetipa@google.com>

Author: ferponse

src/a2a/utils/telemetry.py

@@ -229,7 +229,15 @@ async def async_wrapper(*args, **kwargs) -> Any:
         tracer = trace.get_tracer(
             INSTRUMENTING_MODULE_NAME, INSTRUMENTING_MODULE_VERSION
         )
-        with tracer.start_as_current_span(actual_span_name, kind=kind) as span:
+        # The wrapper records exceptions and sets the span status itself, so
+        # OTel must not do it again: it would mark non-error exceptions
+        # (e.g. `QueueShutDown`) as ERROR and record every exception twice.
+        with tracer.start_as_current_span(
+            actual_span_name,
+            kind=kind,
+            record_exception=False,
+            set_status_on_exception=False,
+        ) as span:
             if attributes:
                 for k, v in attributes.items():
                     span.set_attribute(k, v)
@@ -272,7 +280,14 @@ async def async_wrapper(*args, **kwargs) -> Any:
     def sync_wrapper(*args, **kwargs) -> Any:
         """Sync Wrapper for the decorator."""
         tracer = trace.get_tracer(INSTRUMENTING_MODULE_NAME)
-        with tracer.start_as_current_span(actual_span_name, kind=kind) as span:
+        # The wrapper records exceptions and sets the span status itself;
+        # letting OTel do it too would record every exception twice.
+        with tracer.start_as_current_span(
+            actual_span_name,
+            kind=kind,
+            record_exception=False,
+            set_status_on_exception=False,
+        ) as span:
             if attributes:
                 for k, v in attributes.items():
                     span.set_attribute(k, v)

tests/utils/test_telemetry.py

@@ -294,3 +294,45 @@ async def non_error_exception() -> NoReturn:
     # the span was never marked as failed.
     for call in mock_span.set_status.call_args_list:
         assert 'description' not in call.kwargs
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ('exc_cls', 'is_error'), [(QueueShutDown, False), (ValueError, True)]
+)
+async def test_span_status_with_real_otel_sdk(
+    exc_cls: type[Exception], is_error: bool
+) -> None:
+    """Check the status a real OTel SDK span ends with.
+
+    The mocked-tracer tests only see the wrapper's own calls. By default,
+    `start_as_current_span` also records any exception leaving its block and
+    marks the span ERROR, which would turn a `QueueShutDown` into an error
+    span and record every exception twice.
+    """
+    from opentelemetry.sdk.trace import TracerProvider
+    from opentelemetry.sdk.trace.export import SimpleSpanProcessor
+    from opentelemetry.sdk.trace.export.in_memory_span_exporter import (
+        InMemorySpanExporter,
+    )
+    from opentelemetry.trace import StatusCode
+
+    exporter = InMemorySpanExporter()
+    provider = TracerProvider()
+    provider.add_span_processor(SimpleSpanProcessor(exporter))
+
+    with mock.patch(
+        'opentelemetry.trace.get_tracer', side_effect=provider.get_tracer
+    ):
+
+        @trace_function
+        async def fail() -> NoReturn:
+            await asyncio.sleep(0)
+            raise exc_cls
+
+        with pytest.raises(exc_cls):
+            await fail()
+
+    (span,) = exporter.get_finished_spans()
+    assert (span.status.status_code is StatusCode.ERROR) is is_error
+    assert [event.name for event in span.events] == ['exception']