refactor(genai): share provider response ID resolution

Author: sipercai

instrumentation-loongsuite/loongsuite-instrumentation-hermes-agent/src/opentelemetry/instrumentation/hermes_agent/helpers.py

@@ -35,6 +35,7 @@
     InvokeAgentInvocation,
     ReactStepInvocation,
 )
+from opentelemetry.util.genai.response_id import resolve_response_id
 from opentelemetry.util.genai.types import (
     FunctionToolDefinition,
     GenericToolDefinition,
@@ -69,38 +70,6 @@ def obj_get(value: Any, field: str, default: Any = None) -> Any:
     return getattr(value, field, default)
 
 
-def response_identifier(
-    value: Any,
-    *,
-    fields: tuple[str, ...] = (
-        "id",
-        "request_id",
-        "response_id",
-        "_request_id",
-    ),
-) -> str | None:
-    """Return a normalized provider/framework response identifier.
-
-    OpenAI-compatible providers expose ``id`` while native DashScope-style
-    responses commonly call the same correlation value ``request_id``.  Keep
-    the extraction intentionally narrow so unrelated object identifiers are
-    never promoted to ``gen_ai.response.id``.
-    """
-
-    if isinstance(value, str):
-        normalized_value = value.strip()
-        return normalized_value or None
-
-    for field in fields:
-        candidate = obj_get(value, field)
-        if not isinstance(candidate, (str, int)):
-            continue
-        normalized = str(candidate).strip()
-        if normalized:
-            return normalized
-    return None
-
-
 def _normalize_platform(value: Any) -> str:
     platform = getattr(value, "value", value)
     return str(platform or "").strip().lower()
@@ -715,17 +684,10 @@ def update_llm_invocation_from_response(
     else:
         invocation.response_model_name = invocation.request_model
 
-    # Prefer the ID captured from the provider SDK stream. Hermes currently
-    # synthesizes ``stream-*`` IDs while aggregating OpenAI-compatible chunks,
-    # so the final framework response is only a fallback. Native-style
-    # ``request_id`` fields are checked before the framework ``id`` for the
-    # same reason.
-    response_id = response_identifier(
+    response_id = resolve_response_id(
         provider_response_id,
-        fields=("id",),
-    ) or response_identifier(
         response,
-        fields=("request_id", "id", "response_id", "_request_id"),
+        framework_fields=("request_id", "id", "response_id"),
     )
     if response_id:
         invocation.response_id = response_id

instrumentation-loongsuite/loongsuite-instrumentation-hermes-agent/src/opentelemetry/instrumentation/hermes_agent/wrappers.py

@@ -26,6 +26,7 @@
 
 from opentelemetry import trace as trace_api
 from opentelemetry.util.genai.extended_handler import ExtendedTelemetryHandler
+from opentelemetry.util.genai.response_id import extract_response_id
 from opentelemetry.util.genai.types import Error
 
 from .helpers import (
@@ -39,7 +40,6 @@
     push_state,
     reset_state,
     resolve_entry_platform,
-    response_identifier,
     start_step,
     state,
     step_finish_reason,
@@ -64,7 +64,8 @@
 )
 _PROVIDER_CREATE_WRAPPED = "_otel_hermes_response_id_capture_wrapped"
 _PROVIDER_ATTEMPT = threading.local()
-_MISSING_PROVIDER_ATTEMPT = object()
+_PROVIDER_CAPTURE = threading.local()
+_MISSING_THREAD_LOCAL = object()
 
 
 class _ProviderResponseAttempt:
@@ -80,7 +81,10 @@ def response_id(self) -> str | None:
             return self._response_id
 
     def record(self, value: Any) -> None:
-        response_id = response_identifier(value)
+        response_id = extract_response_id(
+            value,
+            fields=("request_id", "id", "response_id"),
+        )
         if response_id is None:
             return
         with self._lock:
@@ -99,6 +103,10 @@ def adopt_current_attempt(self) -> None:
         attempt = getattr(_PROVIDER_ATTEMPT, "value", None)
         if attempt is None:
             return
+        _PROVIDER_CAPTURE.value = self
+        self.adopt(attempt)
+
+    def adopt(self, attempt: _ProviderResponseAttempt) -> None:
         with self._lock:
             self._attempt = attempt
 
@@ -143,16 +151,21 @@ def __getattr__(self, name: str) -> Any:
 
 
 def _wrap_provider_create(resource: Any) -> None:
-    if resource is None or getattr(resource, _PROVIDER_CREATE_WRAPPED, False):
-        return
-    create = getattr(resource, "create", None)
-    if not callable(create):
+    create = (
+        getattr(resource, "create", None) if resource is not None else None
+    )
+    if not callable(create) or getattr(
+        create, _PROVIDER_CREATE_WRAPPED, False
+    ):
         return
 
     @functools.wraps(create)
     def _capturing_create(*args, **kwargs):
         attempt = _ProviderResponseAttempt()
         _PROVIDER_ATTEMPT.value = attempt
+        capture = getattr(_PROVIDER_CAPTURE, "value", None)
+        if capture is not None:
+            capture.adopt(attempt)
         response = create(*args, **kwargs)
         attempt.record(response)
 
@@ -164,9 +177,9 @@ def _capturing_create(*args, **kwargs):
             return _ProviderStreamProxy(response, attempt)
         return response
 
+    setattr(_capturing_create, _PROVIDER_CREATE_WRAPPED, True)
     try:
         setattr(resource, "create", _capturing_create)
-        setattr(resource, _PROVIDER_CREATE_WRAPPED, True)
     except (AttributeError, TypeError):
         # A provider resource may use slots/read-only descriptors. Telemetry
         # must never make the model call fail; the framework response remains
@@ -179,13 +192,9 @@ class ProviderClientWrapper:
 
     def __call__(self, wrapped, instance, args, kwargs):
         client = wrapped(*args, **kwargs)
-        resources = (
-            getattr(getattr(client, "chat", None), "completions", None),
-            getattr(client, "responses", None),
-            getattr(client, "completions", None),
+        _wrap_provider_create(
+            getattr(getattr(client, "chat", None), "completions", None)
         )
-        for resource in resources:
-            _wrap_provider_create(resource)
         return client
 
 
@@ -443,9 +452,15 @@ def __call__(self, wrapped, instance, args, kwargs):
         previous_provider_attempt = getattr(
             _PROVIDER_ATTEMPT,
             "value",
-            _MISSING_PROVIDER_ATTEMPT,
+            _MISSING_THREAD_LOCAL,
+        )
+        previous_provider_capture = getattr(
+            _PROVIDER_CAPTURE,
+            "value",
+            _MISSING_THREAD_LOCAL,
         )
         _PROVIDER_ATTEMPT.value = None
+        _PROVIDER_CAPTURE.value = provider_response_capture
 
         try:
             if self._streaming:
@@ -521,11 +536,16 @@ def _wrapped_first_delta():
             finish_step(instance, "error", exc=exc)
             raise
         finally:
-            if previous_provider_attempt is _MISSING_PROVIDER_ATTEMPT:
+            if previous_provider_attempt is _MISSING_THREAD_LOCAL:
                 with suppress(AttributeError):
                     del _PROVIDER_ATTEMPT.value
             else:
                 _PROVIDER_ATTEMPT.value = previous_provider_attempt
+            if previous_provider_capture is _MISSING_THREAD_LOCAL:
+                with suppress(AttributeError):
+                    del _PROVIDER_CAPTURE.value
+            else:
+                _PROVIDER_CAPTURE.value = previous_provider_capture
             current_state["active_llm_depth"] = max(
                 0, current_state["active_llm_depth"] - 1
             )

instrumentation-loongsuite/loongsuite-instrumentation-hermes-agent/tests/test_telemetry_spec.py

@@ -523,60 +523,20 @@ def test_streaming_request_id_from_usage_trailer_is_preserved(
     assert llm_span.attributes["gen_ai.response.id"] == "dashscope-request-456"
 
 
-def test_streaming_retry_uses_successful_provider_attempt_id(
-    instrumentation_module,
-    tracer_provider,
-    meter_provider,
-    span_exporter,
-):
-    runtime = _runtime(instrumentation_module, tracer_provider, meter_provider)
-    agent = _FakeAgent(session_id="session-provider-retry")
-    wrappers_module = importlib.import_module(
-        "opentelemetry.instrumentation.hermes_agent.wrappers"
-    )
-    provider_client_wrapper = wrappers_module.ProviderClientWrapper()
-    client = _provider_client(
-        [
-            iter([_stream_chunk("chatcmpl-failed-attempt")]),
-            iter([_stream_chunk("chatcmpl-successful-attempt")]),
-        ]
-    )
-
-    def streaming_call(_api_kwargs, *, on_first_delta):
-        request_client = provider_client_wrapper(
-            lambda: client,
-            agent,
-            (),
-            {},
-        )
-        for _chunk in request_client.chat.completions.create(stream=True):
-            on_first_delta()
-        for _chunk in request_client.chat.completions.create(stream=True):
-            on_first_delta()
-        return _response(
-            content="恢复成功",
-            response_id="stream-retry-framework-id",
-        )
-
-    runtime.streaming_llm_wrapper(
-        streaming_call,
-        agent,
-        ({"model": agent.model, "messages": []},),
-        {},
-    )
-
-    llm_span = _spans_by_kind(span_exporter, "LLM")[0]
-    assert (
-        llm_span.attributes["gen_ai.response.id"]
-        == "chatcmpl-successful-attempt"
-    )
-
-
-def test_streaming_retry_does_not_reuse_failed_provider_attempt_id(
+@pytest.mark.parametrize(
+    ("retry_response_id", "expected_response_id"),
+    [
+        ("chatcmpl-successful-attempt", "chatcmpl-successful-attempt"),
+        (None, "stream-retry-framework-fallback"),
+    ],
+)
+def test_streaming_retry_uses_only_final_provider_attempt(
     instrumentation_module,
     tracer_provider,
     meter_provider,
     span_exporter,
+    retry_response_id,
+    expected_response_id,
 ):
     runtime = _runtime(instrumentation_module, tracer_provider, meter_provider)
     agent = _FakeAgent(session_id="session-provider-retry-fallback")
@@ -587,21 +547,39 @@ def test_streaming_retry_does_not_reuse_failed_provider_attempt_id(
     client = _provider_client(
         [
             iter([_stream_chunk("chatcmpl-failed-attempt")]),
-            iter([_stream_chunk()]),
+            iter([_stream_chunk(retry_response_id)]),
         ]
     )
 
     def streaming_call(_api_kwargs, *, on_first_delta):
-        request_client = provider_client_wrapper(
-            lambda: client,
-            agent,
-            (),
-            {},
-        )
-        for _chunk in request_client.chat.completions.create(stream=True):
-            on_first_delta()
-        for _chunk in request_client.chat.completions.create(stream=True):
-            on_first_delta()
+        errors = []
+
+        def provider_worker():
+            try:
+                request_client = provider_client_wrapper(
+                    lambda: client,
+                    agent,
+                    (),
+                    {},
+                )
+                for _chunk in request_client.chat.completions.create(
+                    stream=True
+                ):
+                    # Hermes invokes this callback once for the whole logical
+                    # call, even when it starts another provider attempt.
+                    on_first_delta()
+                for _chunk in request_client.chat.completions.create(
+                    stream=True
+                ):
+                    pass
+            except BaseException as exc:  # pragma: no cover - defensive
+                errors.append(exc)
+
+        worker = threading.Thread(target=provider_worker)
+        worker.start()
+        worker.join(timeout=5)
+        assert not worker.is_alive()
+        assert errors == []
         return _response(
             content="恢复成功",
             response_id="stream-retry-framework-fallback",
@@ -615,10 +593,7 @@ def streaming_call(_api_kwargs, *, on_first_delta):
     )
 
     llm_span = _spans_by_kind(span_exporter, "LLM")[0]
-    assert (
-        llm_span.attributes["gen_ai.response.id"]
-        == "stream-retry-framework-fallback"
-    )
+    assert llm_span.attributes["gen_ai.response.id"] == expected_response_id
 
 
 def test_streaming_without_provider_id_falls_back_to_hermes_response_id(
@@ -692,6 +667,53 @@ def streaming_call(_api_kwargs, *, on_first_delta):
     )
 
 
+def test_provider_error_is_reported_without_masking_original_exception(
+    instrumentation_module,
+    tracer_provider,
+    meter_provider,
+    span_exporter,
+):
+    runtime = _runtime(instrumentation_module, tracer_provider, meter_provider)
+    agent = _FakeAgent(session_id="session-provider-error")
+    wrappers_module = importlib.import_module(
+        "opentelemetry.instrumentation.hermes_agent.wrappers"
+    )
+    provider_client_wrapper = wrappers_module.ProviderClientWrapper()
+
+    class ProviderError(RuntimeError):
+        pass
+
+    class FailingResource:
+        @staticmethod
+        def create(**_kwargs):
+            raise ProviderError("provider unavailable")
+
+    client = SimpleNamespace(
+        chat=SimpleNamespace(completions=FailingResource())
+    )
+
+    def streaming_call(_api_kwargs, *, on_first_delta):
+        del on_first_delta
+        request_client = provider_client_wrapper(
+            lambda: client,
+            agent,
+            (),
+            {},
+        )
+        return request_client.chat.completions.create(stream=True)
+
+    with pytest.raises(ProviderError, match="provider unavailable"):
+        runtime.streaming_llm_wrapper(
+            streaming_call,
+            agent,
+            ({"model": agent.model, "messages": []},),
+            {},
+        )
+
+    llm_span = _spans_by_kind(span_exporter, "LLM")[0]
+    assert llm_span.status.status_code == StatusCode.ERROR
+
+
 def test_provider_attempt_from_previous_call_does_not_leak(
     instrumentation_module,
     tracer_provider,