feat(sdk): add otel support (#177)

## Summary

Added configurable observability sink selection for the Python SDK and
server so control events can be routed to the default backend,
registered custom sinks, or named sink factories.

Added a built-in OpenTelemetry sink for the SDK, including OTLP
configuration via settings/environment variables.

Updated docs and exports so custom sink registration and OTEL usage are
available as public integration points.

## Scope

**User-facing/API changes:**
- Added `observability_sink_name` and `observability_sink_config` to SDK
initialization/configuration
- Exposed sink registration helpers and OTEL conversion utilities from
the Python SDK public API
- Added documented support for the SDK `otel` extra and OTEL-related
environment variables
- Added server observability sink selection/config support for named
backends

**Internal changes:**
- Introduced shared sink-selection models/registry helpers in
`telemetry/`
- Refactored SDK observability to resolve active sinks dynamically and
support named/custom sinks
- Refactored server startup/shutdown to resolve observability backends
through sink selection
- Added test coverage for sink selection, lifecycle handling, OTEL sink
behavior, and re-init/policy refresh interactions

**Out of scope:**
- No changes to core control evaluation semantics
- No UI changes
- No new non-OTEL external sink implementation included beyond
registration hooks

## Risk and Rollout

**Risk level:** medium

**Rollback plan:** Revert the sink-selection/OTEL changeset to restore
the previous default SDK-to-server observability path only. If needed,
disable custom routing by keeping `observability_sink_name=default` and
not installing/configuring the `otel` extra.

## Testing

- [x] Added or updated automated tests
- [ ] Ran `make check` — typecheck was attempted but full local `make
check` was not run because `uv` was unavailable in this environment
- [x] Manually verified behavior

## Checklist

- [ ] Linked issue/spec (if applicable)
- [x] Updated docs/examples for user-facing changes
- [ ] Included any required follow-up tasks

Author: namrataghadi-galileo

README.md

@@ -185,6 +185,24 @@ events the SDK emits through its normal event-construction flow. The default
 SDK sink remains the OSS path to the Agent Control server. To use registered
 or named custom sinks, set `observability_sink_name` explicitly.
 
+The SDK also includes a built-in OpenTelemetry sink. Install the OTEL extra,
+select the `otel` sink, and configure the OTLP exporter through Agent Control
+settings or environment variables:
+
+```bash
+uv pip install "agent-control-sdk[otel]"
+export AGENT_CONTROL_OBSERVABILITY_SINK_NAME=otel
+export AGENT_CONTROL_OTEL_ENABLED=true
+export AGENT_CONTROL_OTEL_ENDPOINT=http://localhost:4318/v1/traces
+export AGENT_CONTROL_OTEL_HEADERS='{"authorization":"Bearer demo-token"}'
+export AGENT_CONTROL_OTEL_SERVICE_NAME=awesome-bot
+```
+
+If the `otel` sink is selected without an OTLP endpoint/exporter configured,
+the OTEL path stays inert and the default OSS SDK-to-server behavior still
+remains unchanged unless `observability_sink_name` is explicitly switched away
+from `default`.
+
 Next, create a control in Step 4, then run the setup and agent scripts in
 order to see blocking in action.
 

sdks/python/pyproject.toml

@@ -38,6 +38,11 @@ Repository = "https://github.com/yourusername/agent-control"
 [project.optional-dependencies]
 strands-agents = ["strands-agents>=1.26.0"]
 google-adk = ["google-adk>=1.0.0"]
+otel = [
+    "opentelemetry-api>=1.24.0",
+    "opentelemetry-sdk>=1.24.0",
+    "opentelemetry-exporter-otlp-proto-http>=1.24.0",
+]
 galileo = ["agent-control-evaluator-galileo>=7.5.0"]
 
 [dependency-groups]

sdks/python/src/agent_control/__init__.py

@@ -112,6 +112,7 @@ async def handle_input(user_message: str) -> str:
     unregister_control_event_sink_factory,
     write_events,
 )
+from .otel_sink import control_event_to_otel_span
 from .tracing import (
     get_current_span_id,
     get_current_trace_id,
@@ -1421,6 +1422,7 @@ async def main():
     "write_events",
     "shutdown_observability",
     "is_observability_enabled",
+    "control_event_to_otel_span",
     "get_event_batcher",
     "get_event_sink",
     "get_registered_control_event_sink_factory_names",

sdks/python/src/agent_control/control_decorators.py

@@ -36,6 +36,7 @@ async def chat(message: str) -> str:
 from typing import Any, TypeVar
 
 from agent_control_models import Step, normalize_action
+from agent_control_telemetry import get_trace_context_from_provider
 
 from agent_control import AgentControlClient
 from agent_control.evaluation import check_evaluation_with_local
@@ -53,6 +54,25 @@ async def chat(message: str) -> str:
 F = TypeVar("F", bound=Callable[..., Any])
 
 
+def _resolve_control_trace_context() -> tuple[str, str]:
+    """Resolve trace/span IDs for a decorated control site.
+
+    External providers, such as the Galileo bridge, are authoritative because
+    they may reserve the concrete span ID that the eventual LLM/tool call will
+    use. Without a provider, keep the existing behavior: share an active trace
+    but create a fresh function span for this decorated call.
+    """
+    provider_context = get_trace_context_from_provider()
+    if provider_context is not None:
+        return provider_context["trace_id"], provider_context["span_id"]
+
+    existing_trace_id = get_current_trace_id()
+    if existing_trace_id:
+        return existing_trace_id, _generate_span_id()
+
+    return get_trace_and_span_ids()
+
+
 @dataclass
 class ControlContext:
     """
@@ -697,14 +717,7 @@ async def _execute_with_control(
     # Get cached controls for local evaluation support
     controls = _get_server_controls()
 
-    # Get trace context: inherit trace_id if set, always generate new span_id
-    # This allows multiple @control() calls to share the same trace but have unique spans
-    existing_trace_id = get_current_trace_id()
-    if existing_trace_id:
-        trace_id = existing_trace_id
-        span_id = _generate_span_id()  # New span for this function
-    else:
-        trace_id, span_id = get_trace_and_span_ids()  # New trace and span
+    trace_id, span_id = _resolve_control_trace_context()
 
     ctx = ControlContext(
         agent_name=agent.agent_name,

sdks/python/src/agent_control/observability.py

@@ -28,6 +28,12 @@
 Configuration (Environment Variables):
     # Observability (event batching)
     AGENT_CONTROL_OBSERVABILITY_ENABLED: Enable observability (default: true)
+    AGENT_CONTROL_OBSERVABILITY_SINK_NAME: Selected control-event sink (default: default)
+    AGENT_CONTROL_OBSERVABILITY_SINK_CONFIG: JSON config for the selected sink
+    AGENT_CONTROL_OTEL_ENABLED: Enable the built-in OTEL sink (default: false)
+    AGENT_CONTROL_OTEL_ENDPOINT: OTLP HTTP endpoint for exported control-event spans
+    AGENT_CONTROL_OTEL_HEADERS: JSON object of OTLP exporter headers
+    AGENT_CONTROL_OTEL_SERVICE_NAME: OTEL service.name for emitted spans
     AGENT_CONTROL_BATCH_SIZE: Max events per batch (default: 100)
     AGENT_CONTROL_FLUSH_INTERVAL: Seconds between flushes (default: 5.0)
     AGENT_CONTROL_SHUTDOWN_JOIN_TIMEOUT: Seconds to wait for worker shutdown (default: 5.0)
@@ -75,6 +81,11 @@
 if TYPE_CHECKING:
     from agent_control_models import ControlExecutionEvent
 
+from .otel_sink import (
+    OTEL_CONTROL_EVENT_SINK_NAME,
+    create_otel_control_event_sink,
+)
+
 # =============================================================================
 # Logger Setup - Standard Library Pattern
 # =============================================================================
@@ -816,6 +827,19 @@ def get_stats(self) -> dict:
 _configured_named_event_sink: ControlEventSink | None = None
 _configured_named_event_sink_selection: ControlEventSinkSelection | None = None
 _configured_named_event_sink_lock = threading.Lock()
+_used_custom_event_sinks: list[ControlEventSink] = []
+_used_custom_event_sinks_lock = threading.Lock()
+
+
+def _register_builtin_control_event_sink_factories() -> None:
+    """Ensure built-in named sink factories are available."""
+    _named_event_sink_factories.register(
+        OTEL_CONTROL_EVENT_SINK_NAME,
+        create_otel_control_event_sink,
+    )
+
+
+_register_builtin_control_event_sink_factories()
 
 
 class _BatcherControlEventSink(BaseControlEventSink):
@@ -907,12 +931,37 @@ def get_registered_control_event_sink_factory_names() -> tuple[str, ...]:
     return _named_event_sink_factories.registered_names()
 
 
+def _remember_custom_control_event_sinks(sinks: Sequence[ControlEventSink]) -> None:
+    """Track custom sink instances that should be cleaned up on shutdown."""
+    with _used_custom_event_sinks_lock:
+        for sink in sinks:
+            if not any(remembered_sink is sink for remembered_sink in _used_custom_event_sinks):
+                _used_custom_event_sinks.append(sink)
+
+
+def _sink_is_active(sink: ControlEventSink) -> bool:
+    """Return whether a sink instance is currently able to deliver events."""
+    is_active = getattr(sink, "is_active", None)
+    if callable(is_active):
+        return bool(is_active())
+    return True
+
+
 def _get_sink_selection() -> ControlEventSinkSelection:
     """Build the current sink-selection model from SDK settings."""
     settings = get_settings()
+    config: JSONObject = dict(settings.observability_sink_config or {})
+    if settings.observability_sink_name == OTEL_CONTROL_EVENT_SINK_NAME:
+        # Materialize OTEL-specific settings into the selection so that
+        # changes to otel_endpoint / otel_headers / otel_service_name /
+        # otel_enabled invalidate the cached sink instance.
+        config.setdefault("enabled", settings.otel_enabled)
+        config.setdefault("endpoint", settings.otel_endpoint)
+        config.setdefault("headers", dict(settings.otel_headers))
+        config.setdefault("service_name", settings.otel_service_name)
     return ControlEventSinkSelection(
         name=settings.observability_sink_name,
-        config=settings.observability_sink_config,
+        config=config,
     )
 
 
@@ -967,13 +1016,19 @@ def _get_active_control_event_sinks() -> tuple[ControlEventSink, ...]:
 
     selection = _get_sink_selection()
     if selection.name == DEFAULT_CONTROL_EVENT_SINK_NAME:
-        return (_event_sink,) if _event_sink is not None else ()
+        if _event_sink is None or not _sink_is_active(_event_sink):
+            return ()
+        return (_event_sink,)
     if selection.name == REGISTERED_CONTROL_EVENT_SINK_NAME:
-        return get_registered_control_event_sinks()
+        sinks = get_registered_control_event_sinks()
+        sinks = tuple(sink for sink in sinks if _sink_is_active(sink))
+        _remember_custom_control_event_sinks(sinks)
+        return sinks
 
     named_sink = _get_or_create_named_control_event_sink(selection)
-    if named_sink is None:
+    if named_sink is None or not _sink_is_active(named_sink):
         return ()
+    _remember_custom_control_event_sinks((named_sink,))
     return (named_sink,)
 
 
@@ -987,6 +1042,20 @@ def _shutdown_built_in_event_sink() -> None:
     _event_sink = None
 
 
+def _shutdown_configured_named_event_sink() -> None:
+    """Stop and clear the cached configured named sink if it is active."""
+    global _configured_named_event_sink, _configured_named_event_sink_selection
+
+    configured_named_sink: ControlEventSink | None = None
+    with _configured_named_event_sink_lock:
+        configured_named_sink = _configured_named_event_sink
+        _configured_named_event_sink = None
+        _configured_named_event_sink_selection = None
+
+    if configured_named_sink is not None:
+        _shutdown_custom_control_event_sink(configured_named_sink)
+
+
 def _shutdown_custom_control_event_sink(sink: ControlEventSink) -> None:
     """Flush and close a custom sink when it exposes lifecycle hooks."""
     flush = getattr(sink, "flush", None)
@@ -1020,6 +1089,12 @@ async def _run_awaitable_during_shutdown(result: Awaitable[Any]) -> None:
     await result
 
 
+def _get_custom_control_event_sinks_to_shutdown() -> tuple[ControlEventSink, ...]:
+    """Collect custom sink instances that should be cleaned up on shutdown."""
+    with _used_custom_event_sinks_lock:
+        return tuple(_used_custom_event_sinks)
+
+
 def init_observability(
     server_url: str | None = None,
     api_key: str | None = None,
@@ -1046,6 +1121,8 @@ def init_observability(
 
     settings_updates: dict[str, object] = {}
     current_settings = get_settings()
+    if enabled is not None:
+        settings_updates["observability_enabled"] = enabled
     if sink_name is not None:
         settings_updates["observability_sink_name"] = sink_name
         if (
@@ -1058,11 +1135,12 @@ def init_observability(
     if settings_updates:
         configure_settings(**settings_updates)
 
-    is_enabled = enabled if enabled is not None else get_settings().observability_enabled
+    is_enabled = get_settings().observability_enabled
 
     if not is_enabled:
         logger.debug("Observability disabled")
         _shutdown_built_in_event_sink()
+        _shutdown_configured_named_event_sink()
         return None
 
     selection = _get_sink_selection()
@@ -1137,15 +1215,8 @@ def write_events(events: Sequence[ControlExecutionEvent]) -> SinkResult:
 
 def sync_shutdown_observability() -> None:
     """Synchronously shut down observability and flush remaining events."""
-    global _configured_named_event_sink, _configured_named_event_sink_selection
     _shutdown_built_in_event_sink()
-    configured_named_sink: ControlEventSink | None = None
-    with _configured_named_event_sink_lock:
-        configured_named_sink = _configured_named_event_sink
-        _configured_named_event_sink = None
-        _configured_named_event_sink_selection = None
-    if configured_named_sink is not None:
-        _shutdown_custom_control_event_sink(configured_named_sink)
+    _shutdown_configured_named_event_sink()
 
 
 async def shutdown_observability() -> None: