@@ -19,10 +19,11 @@
 from __future__ import annotations
 
 import contextlib
+import traceback
 import urllib.parse
 from typing import TYPE_CHECKING, Any, Callable, Sequence
 
-from google.api_core import _feature_gating_helpers
+from google.api_core import _feature_gating_helpers, exceptions
 from google.api_core.client_options import ClientOptions
 
 try:
@@ -41,6 +42,22 @@
     from google.api_core.grpc_helpers import ClientInterceptor  # noqa: F401
 
 _TRACER_PROVIDER = "tracer_provider"
+_HAS_OPENTELEMETRY: bool | None = None
+
+
+def has_opentelemetry() -> bool:
+    """Checks once per process if opentelemetry-api is installed."""
+    global _HAS_OPENTELEMETRY
+    if _HAS_OPENTELEMETRY is None:
+        try:
+            # Ignored by mypy: Optional dependency only checked at runtime
+            # flake8: Imported only to verify package installation
+            import opentelemetry.trace  # type: ignore[import-not-found] # noqa: F401
+
+            _HAS_OPENTELEMETRY = True
+        except ImportError:
+            _HAS_OPENTELEMETRY = False
+    return _HAS_OPENTELEMETRY
 
 
 def is_otel_capabilities_enabled(
@@ -74,6 +91,218 @@ def is_otel_capabilities_enabled(
     return False
 
 
+def extract_trace_context(span: Any = None) -> dict[str, Any]:
+    """Extracts trace_id, span_id, and Cloud Logging trace correlation fields.
+
+    If `span` is None, falls back to `opentelemetry.trace.get_current_span()`
+    when OpenTelemetry is installed. Returns an empty dict if no valid recording
+    span context exists or if OpenTelemetry is not installed.
+
+    Args:
+        span (Optional[Any]): The active OpenTelemetry span, if any.
+
+    Returns:
+        dict[str, Any]: Trace and span correlation attributes.
+    """
+    if span is None and has_opentelemetry():
+        try:
+            # Ignored by mypy: Optional dependency only loaded after
+            # has_opentelemetry() check
+            from opentelemetry import trace  # type: ignore[import-not-found]
+
+            span = trace.get_current_span()
+        # Defensive fail-open: telemetry correlation must never crash an RPC call
+        # if a custom or broken OpenTelemetry TracerProvider raises at runtime.
+        except Exception:
+            span = None
+
+    if span is None or not hasattr(span, "get_span_context"):
+        return {}
+
+    try:
+        ctx = span.get_span_context()
+    # Defensive fail-open: guard against broken third-party or mock span objects.
+    except Exception:
+        return {}
+
+    if ctx is None or not getattr(ctx, "is_valid", False):
+        return {}
+
+    try:
+        trace_id_hex = f"{ctx.trace_id:032x}"
+        span_id_hex = f"{ctx.span_id:016x}"
+        trace_flags = getattr(ctx, "trace_flags", None)
+        if trace_flags is not None and hasattr(trace_flags, "sampled"):
+            sampled = bool(trace_flags.sampled)
+        elif isinstance(trace_flags, int):
+            sampled = bool(trace_flags & 0x01)
+        else:
+            sampled = False
+    except (TypeError, ValueError, AttributeError):
+        return {}
+
+    return {
+        "trace_id": trace_id_hex,
+        "span_id": span_id_hex,
+        "logging.googleapis.com/trace": trace_id_hex,
+        "logging.googleapis.com/spanId": span_id_hex,
+        "logging.googleapis.com/trace_sampled": sampled,
+    }
+
+
+def _extract_error_info_fields(target_exc: Any) -> tuple[Any, Any, Any]:
+    """Resolves (domain, reason, metadata) from an exception, ErrorInfo, or dict."""
+    # Prefer native GoogleAPICallError properties (.domain, .reason, .metadata)
+    # when present; fall back to .error_info or raw gRPC trailing_metadata.
+    error_info = getattr(target_exc, "error_info", None)
+    if error_info is None and not (
+        getattr(target_exc, "reason", None)
+        or getattr(target_exc, "domain", None)
+        or getattr(target_exc, "metadata", None)
+    ):
+        rpc_call = (
+            target_exc
+            if hasattr(target_exc, "trailing_metadata")
+            else getattr(target_exc, "response", None)
+        )
+        if rpc_call is not None and hasattr(rpc_call, "trailing_metadata"):
+            try:
+                _, error_info = exceptions._parse_grpc_error_details(rpc_call)
+            # Defensive fail-open: malformed trailing metadata must not abort
+            # error attribute extraction.
+            except Exception:
+                pass
+
+    source = error_info if error_info is not None else target_exc
+    if isinstance(source, dict):
+        return (
+            source.get("domain"),
+            source.get("reason"),
+            source.get("metadata"),
+        )
+    return (
+        getattr(source, "domain", None),
+        getattr(source, "reason", None),
+        getattr(source, "metadata", None),
+    )
+
+
+def _extract_error_message(
+    target_exc: Any, http_response: Any = None, fallback: str | None = None
+) -> str | None:
+    """Resolves the human-readable error message across HTTP, gRPC, and Python errors."""
+    http_msg = (
+        _TraceContext._extract_http_error_message(http_response)
+        if http_response is not None
+        else None
+    )
+    details = getattr(target_exc, "details", None)
+    details_msg = details() if callable(details) else details
+    exc_msg = str(target_exc) if isinstance(target_exc, BaseException) else None
+    raw_message = (
+        http_msg
+        or getattr(target_exc, "message", None)
+        or details_msg
+        or fallback
+        or exc_msg
+    )
+    return str(raw_message) if raw_message else None
+
+
+def build_telemetry_error_attributes(
+    exc: BaseException | None,
+    status_code: str | None = None,
+    status_message: str | None = None,
+    include_exception: bool = False,
+) -> dict[str, Any]:
+    """Extracts rpc.response.status_code, gcp.errors.*, error.type, status.message, and exception.* fields.
+
+    Error details and ErrorInfo structures are resolved by inspecting:
+    * Chained exceptions: Unwraps RetryError or __cause__ to the root error.
+    * Canonical status code: Resolves rpc.response.status_code via
+      exceptions._extract_status_code.
+    * GoogleAPICallError & ErrorInfo attributes: Reads .domain, .reason, and
+      .metadata directly (or from .error_info / gRPC trailing_metadata).
+    * Unified attribute extraction: Populates gcp.errors.domain, error.type,
+      gcp.errors.metadata.*, status.message, and optional exception.* fields.
+
+    Args:
+        exc (Optional[BaseException]): An exception (e.g. GoogleAPICallError
+            or grpc.RpcError) or ErrorInfo object.
+        status_code (Optional[str]): Optional pre-extracted or HTTP status code
+            fallback when exc is None or lacks a status code.
+        status_message (Optional[str]): Optional status message fallback when
+            exc is None or lacks a message.
+        include_exception (bool): When True, also populates exception.type,
+            exception.message, and exception.stacktrace.
+
+    Returns:
+        dict[str, Any]: Extracted error and exception attributes.
+    """
+    if exc is None:
+        if status_code is None:
+            return {}
+        code_str = str(status_code)
+        return {
+            "rpc.response.status_code": code_str,
+            "error.type": code_str,
+            "status.message": status_message or f"HTTP {code_str}",
+        }
+
+    # Unwrap chained exceptions once and resolve the canonical status code
+    target_exc = getattr(exc, "cause", None) or getattr(exc, "__cause__", None) or exc
+    resolved_code = exceptions._extract_status_code(target_exc, default=status_code)
+    domain, reason, metadata = _extract_error_info_fields(target_exc)
+
+    # Identify HTTP transport responses (excluding gRPC Call objects)
+    http_response = getattr(target_exc, "response", None)
+    http_status = (
+        getattr(http_response, "status_code", getattr(http_response, "status", None))
+        if http_response is not None and not hasattr(http_response, "trailing_metadata")
+        else None
+    )
+
+    # Populate OpenTelemetry semantic error attributes and ErrorInfo metadata
+    attrs: dict[str, Any] = {"rpc.response.status_code": resolved_code}
+    if domain:
+        attrs["gcp.errors.domain"] = domain
+    if reason:
+        attrs["error.type"] = reason
+    elif (
+        http_status is not None
+        and not isinstance(http_status, bool)
+        and str(http_status).isdigit()
+    ):
+        attrs["error.type"] = str(http_status)
+    else:
+        # Every failed span or log records a low-cardinality error.type; reuse
+        # the resolved canonical status code name or exception class name.
+        attrs["error.type"] = resolved_code
+
+    if metadata and hasattr(metadata, "items"):
+        for k, v in metadata.items():
+            attrs[f"gcp.errors.metadata.{k}"] = str(v)
+
+    # Extract human-readable error description for cross-language parity
+    message = _extract_error_message(
+        target_exc,
+        http_response if http_status is not None else None,
+        status_message,
+    )
+    if message:
+        attrs["status.message"] = message
+
+    # Populate exception.* diagnostic fields for structured logs when requested
+    if include_exception and isinstance(exc, BaseException):
+        attrs["exception.type"] = exc.__class__.__name__
+        attrs["exception.message"] = str(exc)
+        attrs["exception.stacktrace"] = "".join(
+            traceback.format_exception(type(exc), exc, exc.__traceback__)
+        )
+
+    return attrs
+
+
 def _extract_endpoint_attributes(
     client_options: ClientOptions | dict[str, Any] | None = None,
     default_host: str | None = None,
 
@@ -21,9 +21,9 @@
 import contextlib
 import enum
 import functools
-from typing import Any, List, Optional, Tuple
+from typing import List, Tuple
 
-from google.api_core import _observability, exceptions, grpc_helpers
+from google.api_core import _observability, grpc_helpers
 from google.api_core.gapic_v1 import client_info
 from google.api_core.gapic_v1.client_info import METRICS_METADATA_KEY
 from google.api_core.timeout import TimeToDeadlineTimeout
@@ -126,98 +126,6 @@ def _extract_rpc_identity(
     return method_str, service, method
 
 
-def _extract_error_attributes(exc: Optional[BaseException]) -> dict[str, Any]:
-    """Extract gcp.errors.* and error.type attributes from an exception.
-
-    Error details and ErrorInfo structures are resolved by inspecting the following locations:
-    * Chained exceptions: Unwraps RetryError or __cause__ to the root exception.
-    * GoogleAPICallError attributes: Reads ErrorInfo from ._error_info or .error_info.
-    * Native gRPC trailing metadata: Parses google.rpc.Status binary details from trailing_metadata.
-    * Unified attribute extraction: Extracts domain, reason, and metadata from ErrorInfo or exception attributes.
-
-    Args:
-        exc (Optional[BaseException]): An exception (such as GoogleAPICallError or grpc.RpcError) or ErrorInfo object.
-
-    Returns:
-        dict[str, Any]: Extracted error attributes (e.g. gcp.errors.domain, error.type, gcp.errors.metadata.*).
-    """
-    attrs: dict[str, Any] = {}
-    if exc is None:
-        return attrs
-
-    # 1. Unwrap chained exceptions: unwrap RetryError or __cause__ to the root failure
-    target_exc = getattr(exc, "cause", None) or getattr(exc, "__cause__", None) or exc
-
-    # 2. Check GoogleAPICallError ErrorInfo attributes
-    error_info = getattr(target_exc, "_error_info", None) or getattr(
-        target_exc, "error_info", None
-    )
-
-    # 3. Check native gRPC trailing metadata for binary google.rpc.Status details
-    if error_info is None:
-        rpc_call = (
-            target_exc
-            if hasattr(target_exc, "trailing_metadata")
-            else getattr(target_exc, "response", None)
-        )
-        if rpc_call is not None and hasattr(rpc_call, "trailing_metadata"):
-            try:
-                _, error_info = exceptions._parse_grpc_error_details(rpc_call)
-            except Exception:
-                pass
-
-    # 4. Unified attribute extraction: extract domain, reason, and metadata from ErrorInfo or exception attributes
-    http_response = getattr(target_exc, "response", None)
-    http_status = (
-        getattr(http_response, "status_code", getattr(http_response, "status", None))
-        if http_response is not None and not hasattr(http_response, "trailing_metadata")
-        else None
-    )
-
-    source = error_info or target_exc
-    if isinstance(source, dict):
-        domain = source.get("domain")
-        reason = source.get("reason")
-        metadata = source.get("metadata")
-    else:
-        domain = getattr(source, "domain", None)
-        reason = getattr(source, "reason", None)
-        metadata = getattr(source, "metadata", None)
-
-    if domain:
-        attrs["gcp.errors.domain"] = domain
-    if reason:
-        attrs["error.type"] = reason
-    elif isinstance(http_status, int):
-        attrs["error.type"] = str(http_status)
-    else:
-        # Fallback per OpenTelemetry Semantic Conventions: every failed span should record
-        # a low-cardinality error.type. Use canonical status code name or exception class name.
-        status_code = exceptions._extract_status_code(target_exc)
-        attrs["error.type"] = status_code or target_exc.__class__.__name__
-    if metadata:
-        for k, v in metadata.items():
-            attrs[f"gcp.errors.metadata.{k}"] = str(v)
-
-    # 5. Extract human-readable error description for cross-language PRD parity
-    message = (
-        _observability._TraceContext._extract_http_error_message(http_response)
-        if http_status is not None
-        else None
-    )
-    if not message:
-        message = getattr(target_exc, "message", None)
-    if not message and hasattr(target_exc, "details"):
-        details = target_exc.details
-        message = details() if callable(details) else details
-    if not message and isinstance(target_exc, BaseException):
-        message = str(target_exc)
-    if message:
-        attrs["status.message"] = str(message)
-
-    return attrs
-
-
 class _GapicCallable(object):
     """Callable that applies retry, timeout, and metadata logic.
 
@@ -381,10 +289,9 @@ def _trace_span(self):
                 raise
             except BaseException as exc:
                 if span is not None and hasattr(span, "set_attribute"):
-                    span.set_attribute(
-                        "rpc.response.status_code", exceptions._extract_status_code(exc)
-                    )
-                    for k, v in _extract_error_attributes(exc).items():
+                    for k, v in _observability.build_telemetry_error_attributes(
+                        exc
+                    ).items():
                         span.set_attribute(k, v)
                 raise
 
 
@@ -43,6 +43,7 @@ def mock_otel(monkeypatch):
 
     mock_trace = mock.Mock()
     mock_trace.get_tracer.return_value = mock_tracer
+    mock_trace.get_current_span.return_value = mock_span
     mock_trace.SpanKind.CLIENT = "CLIENT"
     mock_trace.SpanKind.INTERNAL = "INTERNAL"
     mock_trace.StatusCode.ERROR = "ERROR"