""" Telemetry helpers (structured events + optional spans). This keeps OpenTelemetry optional so unit tests and lightweight installs don't need to pull in tracing dependencies. """ from __future__ import annotations from contextlib import contextmanager from typing import Any, Dict, Iterator, Optional def _try_get_tracer(): try: from opentelemetry import trace # type: ignore return trace.get_tracer("ylff") except Exception: # pragma: no cover return None @contextmanager def span(name: str, *, attributes: Optional[Dict[str, Any]] = None) -> Iterator[None]: """ Context manager that creates an OpenTelemetry span if available, otherwise a no-op. """ tracer = _try_get_tracer() if tracer is None: yield return with tracer.start_as_current_span(name) as s: # pragma: no cover if attributes: for k, v in attributes.items(): try: s.set_attribute(k, v) except Exception: # Avoid blowing up the business logic on bad attribute types. pass yield