From 7ea4883d73ceed23abdaeeb6be34ddc4cca0c17d Mon Sep 17 00:00:00 2001 From: Namrata Ghadi Date: Fri, 21 Aug 2026 13:23:54 -0700 Subject: [PATCH 1/2] add optional traceprovider parameter --- sdks/python/README.md | 35 ++++ sdks/python/src/agent_control/__init__.py | 13 +- .../python/src/agent_control/observability.py | 63 ++++++- sdks/python/src/agent_control/otel_sink.py | 69 +++++-- sdks/python/tests/test_init_validation.py | 39 +++- sdks/python/tests/test_otel_sink.py | 168 +++++++++++++++++- 6 files changed, 365 insertions(+), 22 deletions(-) diff --git a/sdks/python/README.md b/sdks/python/README.md index e61f41d5..f9a608b0 100644 --- a/sdks/python/README.md +++ b/sdks/python/README.md @@ -24,3 +24,38 @@ async def search_database(query: str) -> str: ``` Controls are defined centrally and enforced automatically at runtime. See [Python SDK Documentation](https://docs.agentcontrol.dev/sdk/python-sdk) for complete reference. + +## Sharing an OpenTelemetry provider with Google ADK + +When Google ADK and Agent Control should export through the same OpenTelemetry +pipeline, configure one SDK `TracerProvider` and pass it to both components: + +```python +import agent_control +from opentelemetry import trace +from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter +from openinference.instrumentation.google_adk import GoogleADKInstrumentor +from opentelemetry.sdk.resources import Resource +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import BatchSpanProcessor + +resource = Resource.create({"service.name": "my-google-adk-agent"}) +provider = TracerProvider(resource=resource) +provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter())) +trace.set_tracer_provider(provider) + +GoogleADKInstrumentor().instrument(tracer_provider=provider) +agent_control.init( + agent_name="my-google-adk-agent", + observability_enabled=True, + observability_sink_name="otel", + observability_sink_config={"enabled": True}, + otel_tracer_provider=provider, +) +``` + +Agent Control reuses this provider without adding an exporter or span processor. +It force-flushes the provider during shutdown but leaves provider shutdown to the +application. If no provider is passed, the OTEL sink first checks for a globally +registered SDK provider, then falls back to its existing Agent Control-owned OTLP +pipeline. diff --git a/sdks/python/src/agent_control/__init__.py b/sdks/python/src/agent_control/__init__.py index bef9e4d1..6c6e117f 100644 --- a/sdks/python/src/agent_control/__init__.py +++ b/sdks/python/src/agent_control/__init__.py @@ -38,6 +38,8 @@ async def handle_input(user_message: str) -> str: ) """ +from __future__ import annotations + from importlib.metadata import PackageNotFoundError, version try: @@ -51,7 +53,7 @@ async def handle_input(user_message: str) -> str: from collections.abc import Callable, Coroutine from dataclasses import dataclass from datetime import UTC, datetime -from typing import Any, Literal, TypeVar +from typing import TYPE_CHECKING, Any, Literal, TypeVar import httpx from agent_control_models import ( @@ -123,6 +125,9 @@ async def handle_input(user_message: str) -> str: ) from .validation import ensure_agent_name +if TYPE_CHECKING: + from opentelemetry.sdk.trace import TracerProvider + # Module logger logger = get_logger(__name__) @@ -460,6 +465,7 @@ def init( target_type: str | None = None, target_id: str | None = None, runtime_token_header: str | None = None, + otel_tracer_provider: TracerProvider | None = None, **kwargs: object ) -> Agent: """ @@ -514,6 +520,10 @@ def init( X-Agent-Control-Runtime-Token) when the server runs behind a gateway that reserves Authorization for its own identity JWT. The server must be configured to read the same header. + otel_tracer_provider: Optional application-owned OpenTelemetry SDK provider. + Used only when observability_sink_name is ``"otel"``. If omitted, + the OTEL sink reuses a globally registered SDK provider when available, + otherwise it creates and owns a provider as before. **kwargs: Additional metadata to store with the agent Returns: @@ -738,6 +748,7 @@ def run_in_thread() -> None: enabled=observability_enabled, sink_name=observability_sink_name, sink_config=observability_sink_config, + otel_tracer_provider=otel_tracer_provider, ) if batcher: logger.info("Observability enabled") diff --git a/sdks/python/src/agent_control/observability.py b/sdks/python/src/agent_control/observability.py index 199017c1..39015bdb 100644 --- a/sdks/python/src/agent_control/observability.py +++ b/sdks/python/src/agent_control/observability.py @@ -83,6 +83,7 @@ if TYPE_CHECKING: from agent_control_models import ControlExecutionEvent + from opentelemetry.sdk.trace import TracerProvider from .otel_sink import ( OTEL_CONTROL_EVENT_SINK_NAME, @@ -833,6 +834,8 @@ def get_stats(self) -> dict: ) _configured_named_event_sink: ControlEventSink | None = None _configured_named_event_sink_selection: ControlEventSinkSelection | None = None +_configured_otel_tracer_provider: TracerProvider | None = None +_configured_named_event_sink_provider: TracerProvider | None = None _configured_named_event_sink_lock = threading.Lock() _used_custom_event_sinks: list[ControlEventSink] = [] _used_custom_event_sinks_lock = threading.Lock() @@ -842,7 +845,15 @@ 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, + _create_configured_otel_control_event_sink, + ) + + +def _create_configured_otel_control_event_sink(config: JSONObject) -> BaseControlEventSink: + """Create the OTEL sink with its runtime-only provider configuration.""" + return create_otel_control_event_sink( + config, + tracer_provider=_configured_otel_tracer_provider, ) @@ -976,14 +987,21 @@ def _get_or_create_named_control_event_sink( selection: ControlEventSinkSelection, ) -> ControlEventSink | None: """Resolve and cache a named configured sink.""" - global _configured_named_event_sink, _configured_named_event_sink_selection + global _configured_named_event_sink, _configured_named_event_sink_provider + global _configured_named_event_sink_selection previous_sink: ControlEventSink | None = None + selected_provider = ( + _configured_otel_tracer_provider + if selection.name == OTEL_CONTROL_EVENT_SINK_NAME + else None + ) with _configured_named_event_sink_lock: if ( _configured_named_event_sink is not None and _configured_named_event_sink_selection == selection + and _configured_named_event_sink_provider is selected_provider ): return _configured_named_event_sink @@ -1004,6 +1022,7 @@ def _get_or_create_named_control_event_sink( previous_sink = _configured_named_event_sink _configured_named_event_sink = sink _configured_named_event_sink_selection = selection.model_copy(deep=True) + _configured_named_event_sink_provider = selected_provider if previous_sink is not None: _shutdown_custom_control_event_sink(previous_sink) @@ -1051,18 +1070,41 @@ def _shutdown_built_in_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 + global _configured_named_event_sink, _configured_named_event_sink_provider + global _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 + _configured_named_event_sink_provider = None if configured_named_sink is not None: _shutdown_custom_control_event_sink(configured_named_sink) +def _invalidate_cached_otel_sink() -> None: + """Clear and shut down only the cached built-in OTEL sink, if present.""" + global _configured_named_event_sink, _configured_named_event_sink_provider + global _configured_named_event_sink_selection + + configured_otel_sink: ControlEventSink | None = None + with _configured_named_event_sink_lock: + if ( + _configured_named_event_sink_selection is None + or _configured_named_event_sink_selection.name != OTEL_CONTROL_EVENT_SINK_NAME + ): + return + configured_otel_sink = _configured_named_event_sink + _configured_named_event_sink = None + _configured_named_event_sink_selection = None + _configured_named_event_sink_provider = None + + if configured_otel_sink is not None: + _shutdown_custom_control_event_sink(configured_otel_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) @@ -1109,6 +1151,7 @@ def init_observability( enabled: bool | None = None, sink_name: str | None = None, sink_config: JSONObject | None = None, + otel_tracer_provider: TracerProvider | None = None, ) -> EventBatcher | None: """ Initialize observability system. @@ -1122,11 +1165,12 @@ def init_observability( enabled: Override AGENT_CONTROL_OBSERVABILITY_ENABLED sink_name: Override AGENT_CONTROL_OBSERVABILITY_SINK_NAME sink_config: Override AGENT_CONTROL_OBSERVABILITY_SINK_CONFIG + otel_tracer_provider: Runtime-only provider for the built-in OTEL sink Returns: EventBatcher instance if enabled, None otherwise """ - global _batcher, _event_sink + global _batcher, _configured_otel_tracer_provider, _event_sink settings_updates: dict[str, object] = {} current_settings = get_settings() @@ -1144,6 +1188,17 @@ def init_observability( if settings_updates: configure_settings(**settings_updates) + selected_sink_name = get_settings().observability_sink_name + selected_otel_tracer_provider = ( + otel_tracer_provider + if selected_sink_name == OTEL_CONTROL_EVENT_SINK_NAME + else None + ) + provider_changed = selected_otel_tracer_provider is not _configured_otel_tracer_provider + _configured_otel_tracer_provider = selected_otel_tracer_provider + if provider_changed: + _invalidate_cached_otel_sink() + is_enabled = get_settings().observability_enabled if not is_enabled: diff --git a/sdks/python/src/agent_control/otel_sink.py b/sdks/python/src/agent_control/otel_sink.py index e724f5af..aca0b4fe 100644 --- a/sdks/python/src/agent_control/otel_sink.py +++ b/sdks/python/src/agent_control/otel_sink.py @@ -8,7 +8,7 @@ from collections.abc import Sequence from dataclasses import dataclass from datetime import UTC, datetime -from typing import Any, cast +from typing import TYPE_CHECKING, Any, cast from agent_control_models import ControlExecutionEvent, JSONObject from agent_control_telemetry.sinks import BaseControlEventSink, SinkResult @@ -16,6 +16,9 @@ from .settings import get_settings from .tracing import _generate_span_id, validate_span_id, validate_trace_id +if TYPE_CHECKING: + from opentelemetry.sdk.trace import TracerProvider + logger = logging.getLogger(__name__) OTEL_CONTROL_EVENT_SINK_NAME = "otel" @@ -70,7 +73,7 @@ class OTELSDKModules: tracer_provider_cls: type[Any] resource_cls: type[Any] batch_span_processor_cls: type[Any] - otlp_span_exporter_cls: type[Any] + otlp_span_exporter_cls: type[Any] | None span_context_cls: type[Any] non_recording_span_cls: type[Any] trace_flags_cls: type[Any] @@ -79,6 +82,7 @@ class OTELSDKModules: status_code_cls: type[Any] span_kind: Any set_span_in_context: Any + get_tracer_provider: Any def _to_unix_nano(timestamp: datetime, /) -> int: @@ -171,10 +175,12 @@ def __init__( tracer_provider: Any, tracer: Any, sdk_modules: OTELSDKModules, + provider_owned: bool, ) -> None: self._tracer_provider: Any = tracer_provider self._tracer: Any = tracer self._sdk_modules = sdk_modules + self._provider_owned = provider_owned def write_events(self, events: Sequence[ControlExecutionEvent]) -> SinkResult: accepted = 0 @@ -196,6 +202,8 @@ def flush(self) -> None: force_flush() def close(self) -> None: + if not self._provider_owned: + return shutdown = getattr(self._tracer_provider, "shutdown", None) if callable(shutdown): shutdown() @@ -285,10 +293,7 @@ def _has_explicit_otel_exporter_configuration(config: OTELSinkConfig) -> bool: def _load_otel_sdk_modules() -> OTELSDKModules: - """Import OTEL SDK modules on demand so the sink remains optional.""" - from opentelemetry.exporter.otlp.proto.http.trace_exporter import ( # type: ignore[import-not-found] - OTLPSpanExporter, - ) + """Import core OTEL SDK modules on demand so the sink remains optional.""" from opentelemetry.sdk.resources import Resource # type: ignore[import-not-found] from opentelemetry.sdk.trace import TracerProvider # type: ignore[import-not-found] from opentelemetry.sdk.trace.export import BatchSpanProcessor # type: ignore[import-not-found] @@ -300,6 +305,7 @@ def _load_otel_sdk_modules() -> OTELSDKModules: StatusCode, TraceFlags, TraceState, + get_tracer_provider, set_span_in_context, ) @@ -307,7 +313,7 @@ def _load_otel_sdk_modules() -> OTELSDKModules: tracer_provider_cls=TracerProvider, resource_cls=Resource, batch_span_processor_cls=BatchSpanProcessor, - otlp_span_exporter_cls=OTLPSpanExporter, + otlp_span_exporter_cls=None, span_context_cls=SpanContext, non_recording_span_cls=NonRecordingSpan, trace_flags_cls=TraceFlags, @@ -316,11 +322,30 @@ def _load_otel_sdk_modules() -> OTELSDKModules: status_code_cls=StatusCode, span_kind=SpanKind, set_span_in_context=set_span_in_context, + get_tracer_provider=get_tracer_provider, + ) + + +def _load_otlp_span_exporter_cls() -> type[Any]: + """Import the OTLP exporter only for an Agent Control-owned pipeline.""" + from opentelemetry.exporter.otlp.proto.http.trace_exporter import ( # type: ignore[import-not-found] + OTLPSpanExporter, ) + return cast(type[Any], OTLPSpanExporter) + + +def create_otel_control_event_sink( + config: JSONObject, + *, + tracer_provider: TracerProvider | None = None, +) -> BaseControlEventSink: + """Create the built-in OTEL control-event sink. -def create_otel_control_event_sink(config: JSONObject) -> BaseControlEventSink: - """Create the built-in OTEL control-event sink.""" + An explicitly supplied or globally registered SDK provider is reused as an + application-owned provider. Only providers created here receive an Agent + Control exporter and processor and are shut down with the sink. + """ resolved_config = _resolve_otel_sink_config(config) if not resolved_config.enabled: return _NoOpControlEventSink() @@ -331,6 +356,29 @@ def create_otel_control_event_sink(config: JSONObject) -> BaseControlEventSink: logger.warning(_OTEL_NOOP_WARNING) return _NoOpControlEventSink() + external_provider: Any = tracer_provider + if external_provider is None: + global_provider = sdk_modules.get_tracer_provider() + if isinstance(global_provider, sdk_modules.tracer_provider_cls): + external_provider = global_provider + + if external_provider is not None: + tracer = external_provider.get_tracer(_OTEL_INSTRUMENTATION_SCOPE) + return OTELControlEventSink( + tracer_provider=external_provider, + tracer=tracer, + sdk_modules=sdk_modules, + provider_owned=False, + ) + + exporter_cls = sdk_modules.otlp_span_exporter_cls + if exporter_cls is None: + try: + exporter_cls = _load_otlp_span_exporter_cls() + except ImportError: + logger.warning(_OTEL_NOOP_WARNING) + return _NoOpControlEventSink() + resource = sdk_modules.resource_cls.create({"service.name": resolved_config.service_name}) tracer_provider = sdk_modules.tracer_provider_cls(resource=resource) @@ -346,7 +394,7 @@ def create_otel_control_event_sink(config: JSONObject) -> BaseControlEventSink: exporter_kwargs["endpoint"] = resolved_config.endpoint if resolved_config.headers: exporter_kwargs["headers"] = resolved_config.headers - exporter = sdk_modules.otlp_span_exporter_cls(**exporter_kwargs) + exporter = exporter_cls(**exporter_kwargs) tracer_provider.add_span_processor(sdk_modules.batch_span_processor_cls(exporter)) tracer = tracer_provider.get_tracer(_OTEL_INSTRUMENTATION_SCOPE) @@ -354,4 +402,5 @@ def create_otel_control_event_sink(config: JSONObject) -> BaseControlEventSink: tracer_provider=tracer_provider, tracer=tracer, sdk_modules=sdk_modules, + provider_owned=True, ) diff --git a/sdks/python/tests/test_init_validation.py b/sdks/python/tests/test_init_validation.py index 6defcf14..eec2764d 100644 --- a/sdks/python/tests/test_init_validation.py +++ b/sdks/python/tests/test_init_validation.py @@ -3,14 +3,13 @@ from unittest.mock import AsyncMock, patch from uuid import uuid4 +import agent_control import pytest +from agent_control._state import state from agent_control_models import ControlMatch as ModelControlMatch from agent_control_models import ControlScope as ModelControlScope from agent_control_models import EvaluatorResult as ModelEvaluatorResult -import agent_control -from agent_control._state import state - def test_init_rejects_invalid_agent_name() -> None: with pytest.raises(ValueError, match="at least 10 characters"): @@ -157,3 +156,37 @@ def test_init_preserves_positional_argument_order() -> None: assert state.runtime_token_header is None finally: agent_control._reset_state() + + +def test_init_passes_otel_provider_through_runtime_only_path() -> None: + # Given: an application provider and successful server initialization + provider = object() + health_check_mock = AsyncMock(return_value={"status": "healthy"}) + register_agent_mock = AsyncMock(return_value={"created": True, "controls": []}) + + try: + with patch( + "agent_control.__init__.AgentControlClient.health_check", + new=health_check_mock, + ), patch( + "agent_control.__init__.agents.register_agent", + new=register_agent_mock, + ), patch("agent_control.init_observability") as init_observability_mock: + # When: initializing the public SDK API with the provider + agent_control.init( + agent_name=f"agent-{uuid4().hex[:12]}", + observability_enabled=True, + observability_sink_name="otel", + observability_sink_config={"enabled": True}, + otel_tracer_provider=provider, # type: ignore[arg-type] + policy_refresh_interval_seconds=0, + ) + + # Then: the provider is separate from the JSON sink configuration + init_observability_mock.assert_called_once() + call_kwargs = init_observability_mock.call_args.kwargs + assert call_kwargs["otel_tracer_provider"] is provider + assert call_kwargs["sink_config"] == {"enabled": True} + assert "otel_tracer_provider" not in call_kwargs["sink_config"] + finally: + agent_control._reset_state() diff --git a/sdks/python/tests/test_otel_sink.py b/sdks/python/tests/test_otel_sink.py index 4d4aa451..d9a7ad4b 100644 --- a/sdks/python/tests/test_otel_sink.py +++ b/sdks/python/tests/test_otel_sink.py @@ -3,12 +3,11 @@ from __future__ import annotations import os +from dataclasses import replace from datetime import UTC, datetime from typing import cast from unittest.mock import patch -from agent_control_models import ControlExecutionEvent - from agent_control import add_event, init_observability, sync_shutdown_observability from agent_control.observability import is_observability_enabled from agent_control.otel_sink import ( @@ -19,6 +18,7 @@ create_otel_control_event_sink, ) from agent_control.settings import configure_settings, get_settings +from agent_control_models import ControlExecutionEvent def _make_event(**overrides: object) -> ControlExecutionEvent: @@ -107,7 +107,7 @@ def start_span( class FakeTracerProvider: - def __init__(self, *, resource: object) -> None: + def __init__(self, *, resource: object = None) -> None: self.resource = resource self.processors: list[object] = [] self.tracer = FakeTracer() @@ -193,7 +193,7 @@ def _fake_set_span_in_context(span: FakeNonRecordingSpan) -> dict[str, object]: return {"parent": span} -def _fake_otel_sdk_modules() -> OTELSDKModules: +def _fake_otel_sdk_modules(global_provider: object = None) -> OTELSDKModules: return OTELSDKModules( tracer_provider_cls=FakeTracerProvider, resource_cls=FakeResource, @@ -207,6 +207,7 @@ def _fake_otel_sdk_modules() -> OTELSDKModules: status_code_cls=FakeStatusCode, span_kind=FakeSpanKind, set_span_in_context=_fake_set_span_in_context, + get_tracer_provider=lambda: global_provider, ) @@ -331,6 +332,132 @@ def test_create_otel_control_event_sink_uses_exporter_config_and_emits_spans() - assert tracer_provider.shutdown_calls == 1 +def test_explicit_provider_takes_precedence_and_remains_application_owned() -> None: + # Given: distinct explicit and globally registered SDK providers + configure_settings(otel_enabled=True, otel_endpoint=None) + explicit_provider = FakeTracerProvider() + global_provider = FakeTracerProvider() + + # When: creating and using the OTEL sink with the explicit provider + with patch( + "agent_control.otel_sink._load_otel_sdk_modules", + return_value=_fake_otel_sdk_modules(global_provider), + ): + sink = create_otel_control_event_sink( + {}, + tracer_provider=explicit_provider, # type: ignore[arg-type] + ) + result = sink.write_events([_make_event()]) + sink.flush() + sink.close() + + # Then: only the explicit provider emits and flushes, with no SDK-added processor + assert isinstance(sink, OTELControlEventSink) + assert result.accepted == 1 + assert len(explicit_provider.tracer.calls) == 1 + assert len(global_provider.tracer.calls) == 0 + assert explicit_provider.processors == [] + assert explicit_provider.force_flush_calls == 1 + assert explicit_provider.shutdown_calls == 0 + + +def test_external_provider_does_not_require_agent_control_otlp_exporter() -> None: + # Given: a reusable provider with core OTEL installed but no OTLP/HTTP exporter + configure_settings(otel_enabled=True) + provider = FakeTracerProvider() + sdk_modules = replace(_fake_otel_sdk_modules(), otlp_span_exporter_cls=None) + + # When: creating the sink with the external provider + with patch( + "agent_control.otel_sink._load_otel_sdk_modules", + return_value=sdk_modules, + ), patch( + "agent_control.otel_sink._load_otlp_span_exporter_cls", + side_effect=AssertionError("external providers must not load an exporter"), + ): + sink = create_otel_control_event_sink( + {}, + tracer_provider=provider, # type: ignore[arg-type] + ) + + # Then: the external pipeline remains sufficient and unchanged + assert isinstance(sink, OTELControlEventSink) + assert sink._tracer_provider is provider + assert provider.processors == [] + + +def test_global_sdk_provider_is_reused_without_adding_a_processor() -> None: + # Given: a globally registered SDK provider and no explicit provider + configure_settings(otel_enabled=True, otel_endpoint=None) + global_provider = FakeTracerProvider() + + # When: creating the OTEL sink + with patch( + "agent_control.otel_sink._load_otel_sdk_modules", + return_value=_fake_otel_sdk_modules(global_provider), + ): + sink = create_otel_control_event_sink({}) + result = sink.write_events([_make_event()]) + sink.close() + + # Then: Agent Control reuses but does not modify or shut down the provider + assert isinstance(sink, OTELControlEventSink) + assert result.accepted == 1 + assert sink._tracer_provider is global_provider + assert global_provider.processors == [] + assert global_provider.shutdown_calls == 0 + + +def test_default_proxy_provider_is_rejected_and_owned_provider_is_created() -> None: + # Given: the OTEL API's proxy/no-op provider and valid Agent Control exporter config + configure_settings( + otel_enabled=True, + otel_endpoint="http://collector:4318/v1/traces", + ) + proxy_provider = object() + + # When: creating the OTEL sink without an explicit provider + with patch( + "agent_control.otel_sink._load_otel_sdk_modules", + return_value=_fake_otel_sdk_modules(proxy_provider), + ): + sink = create_otel_control_event_sink({}) + sink.close() + + # Then: a new configured and Agent Control-owned SDK provider is used + assert isinstance(sink, OTELControlEventSink) + assert isinstance(sink._tracer_provider, FakeTracerProvider) + assert sink._tracer_provider is not proxy_provider + assert len(sink._tracer_provider.processors) == 1 + assert sink._tracer_provider.shutdown_calls == 1 + + +def test_external_provider_preserves_trace_and_parent_correlation() -> None: + # Given: an application-owned provider and an event with supplied correlation IDs + configure_settings(otel_enabled=True) + provider = FakeTracerProvider() + event = _make_event() + + # When: emitting the control span + with patch( + "agent_control.otel_sink._load_otel_sdk_modules", + return_value=_fake_otel_sdk_modules(), + ): + sink = create_otel_control_event_sink( + {}, + tracer_provider=provider, # type: ignore[arg-type] + ) + sink.write_events([event]) + + # Then: the selected provider's tracer receives the supplied trace and parent span IDs + context = provider.tracer.calls[0]["context"] + assert isinstance(context, dict) + parent_span = context["parent"] + assert isinstance(parent_span, FakeNonRecordingSpan) + assert parent_span.span_context.trace_id == int(event.trace_id, 16) + assert parent_span.span_context.span_id == int(event.span_id, 16) + + def test_observability_uses_builtin_otel_sink_when_selected() -> None: configure_settings( observability_sink_name=OTEL_CONTROL_EVENT_SINK_NAME, @@ -419,3 +546,36 @@ def test_observability_rebuilds_otel_sink_when_effective_settings_change() -> No assert obs._configured_named_event_sink_selection.config["endpoint"] == ( "http://collector-2:4318/v1/traces" ) + + +def test_observability_rebuilds_otel_sink_when_explicit_provider_changes() -> None: + # Given: an initialized built-in OTEL sink using an external provider + import agent_control.observability as obs + + configure_settings( + observability_sink_name=OTEL_CONTROL_EVENT_SINK_NAME, + otel_enabled=True, + ) + first_provider = FakeTracerProvider() + second_provider = FakeTracerProvider() + + with patch( + "agent_control.otel_sink._load_otel_sdk_modules", + return_value=_fake_otel_sdk_modules(), + ): + init_observability(enabled=True, otel_tracer_provider=first_provider) # type: ignore[arg-type] + assert add_event(_make_event()) is True + first_sink = obs._configured_named_event_sink + + # When: reinitializing with a different explicit provider + init_observability(enabled=True, otel_tracer_provider=second_provider) # type: ignore[arg-type] + assert obs._configured_named_event_sink is None + assert add_event(_make_event(control_execution_id="ce-456")) is True + second_sink = obs._configured_named_event_sink + + # Then: the cached OTEL sink is replaced without shutting down either external provider + assert first_sink is not second_sink + assert len(first_provider.tracer.calls) == 1 + assert len(second_provider.tracer.calls) == 1 + assert first_provider.shutdown_calls == 0 + assert second_provider.shutdown_calls == 0 From e70b0788e48a59724b6975e73b47e3b07157792b Mon Sep 17 00:00:00 2001 From: Namrata Ghadi Date: Fri, 21 Aug 2026 13:46:14 -0700 Subject: [PATCH 2/2] code coverage --- sdks/python/tests/test_otel_sink.py | 151 ++++++++++++++++++++++++++++ 1 file changed, 151 insertions(+) diff --git a/sdks/python/tests/test_otel_sink.py b/sdks/python/tests/test_otel_sink.py index d9a7ad4b..913e2fa7 100644 --- a/sdks/python/tests/test_otel_sink.py +++ b/sdks/python/tests/test_otel_sink.py @@ -14,6 +14,7 @@ OTEL_CONTROL_EVENT_SINK_NAME, OTELControlEventSink, OTELSDKModules, + _normalize_attribute_value, control_event_to_otel_span, create_otel_control_event_sink, ) @@ -84,6 +85,7 @@ class FakeTracer: def __init__(self) -> None: self.calls: list[dict[str, object]] = [] self.spans: list[FakeSpan] = [] + self.raise_on_start = False def start_span( self, @@ -93,6 +95,8 @@ def start_span( kind: object = None, start_time: int | None = None, ) -> FakeSpan: + if self.raise_on_start: + raise RuntimeError("span creation failed") self.calls.append( { "name": name, @@ -252,6 +256,43 @@ def test_control_event_to_otel_span_maps_event_fields() -> None: assert span.end_time_unix_nano >= span.start_time_unix_nano +def test_control_event_to_otel_span_handles_optional_fields_and_attribute_types() -> None: + # Given: an event without optional timing or evaluator fields and typed metadata lists + event = _make_event( + execution_duration_ms=None, + evaluator_name=None, + selector_path=None, + metadata={ + "bools": [True, False], + "ints": [1, 2], + "floats": [1.5, 2.5], + }, + ) + + # When: normalizing it into an OTEL span + span = control_event_to_otel_span(event) + + # Then: timestamps collapse to an instant and list element types are retained + assert span.start_time_unix_nano == span.end_time_unix_nano + assert span.attributes["agent_control.metadata.bools"] == [True, False] + assert span.attributes["agent_control.metadata.ints"] == [1, 2] + assert span.attributes["agent_control.metadata.floats"] == [1.5, 2.5] + assert "agent_control.execution_duration_ms" not in span.attributes + assert "agent_control.evaluator_name" not in span.attributes + assert "agent_control.selector_path" not in span.attributes + + +def test_attribute_normalization_accepts_tuples() -> None: + # Given: a tuple from an internal event metadata boundary + value = ("security", "pii") + + # When: normalizing the attribute value directly to reach the tuple-only branch + normalized = _normalize_attribute_value(value) + + # Then: OTEL receives its supported homogeneous list representation + assert normalized == ["security", "pii"] + + def test_create_otel_control_event_sink_is_inert_when_disabled() -> None: configure_settings(otel_enabled=False) @@ -263,6 +304,44 @@ def test_create_otel_control_event_sink_is_inert_when_disabled() -> None: assert result.dropped == 0 +def test_create_otel_control_event_sink_is_inert_without_core_otel_sdk() -> None: + # Given: an enabled OTEL sink without the optional core SDK dependency + configure_settings(otel_enabled=True) + + # When: creating the sink + with patch( + "agent_control.otel_sink._load_otel_sdk_modules", + side_effect=ImportError("opentelemetry-sdk is unavailable"), + ): + sink = create_otel_control_event_sink({}) + + # Then: events are safely accepted by an inactive no-op sink + assert sink.is_active() is False + assert sink.write_events([_make_event()]).accepted == 1 + + +def test_create_otel_control_event_sink_is_inert_without_owned_pipeline_exporter() -> None: + # Given: core OTEL support but no reusable provider or OTLP/HTTP exporter + configure_settings( + otel_enabled=True, + otel_endpoint="http://collector:4318/v1/traces", + ) + sdk_modules = replace(_fake_otel_sdk_modules(), otlp_span_exporter_cls=None) + + # When: Agent Control attempts to create its owned pipeline + with patch( + "agent_control.otel_sink._load_otel_sdk_modules", + return_value=sdk_modules, + ), patch( + "agent_control.otel_sink._load_otlp_span_exporter_cls", + side_effect=ImportError("OTLP exporter is unavailable"), + ): + sink = create_otel_control_event_sink({}) + + # Then: it falls back to the inactive no-op sink + assert sink.is_active() is False + + def test_create_otel_control_event_sink_without_exporter_stays_inert() -> None: configure_settings(otel_enabled=True, otel_endpoint=None) @@ -458,6 +537,78 @@ def test_external_provider_preserves_trace_and_parent_correlation() -> None: assert parent_span.span_context.span_id == int(event.span_id, 16) +def test_external_provider_rejects_invalid_trace_id_context() -> None: + # Given: an event with an invalid trace ID + configure_settings(otel_enabled=True) + provider = FakeTracerProvider() + + # When: emitting it through an external provider + with patch( + "agent_control.otel_sink._load_otel_sdk_modules", + return_value=_fake_otel_sdk_modules(), + ): + sink = create_otel_control_event_sink( + {}, + tracer_provider=provider, # type: ignore[arg-type] + ) + result = sink.write_events([_make_event(trace_id="invalid")]) + + # Then: the span is emitted without attaching an invalid parent context + assert result.accepted == 1 + assert provider.tracer.calls[0]["context"] is None + + +def test_external_provider_replaces_invalid_parent_span_id() -> None: + # Given: a valid trace ID paired with an invalid parent span ID + configure_settings(otel_enabled=True) + provider = FakeTracerProvider() + replacement_span_id = "c" * 16 + + # When: emitting the event + with patch( + "agent_control.otel_sink._load_otel_sdk_modules", + return_value=_fake_otel_sdk_modules(), + ), patch( + "agent_control.otel_sink._generate_span_id", + return_value=replacement_span_id, + ): + sink = create_otel_control_event_sink( + {}, + tracer_provider=provider, # type: ignore[arg-type] + ) + result = sink.write_events([_make_event(span_id="invalid")]) + + # Then: correlation retains the trace and uses a valid generated parent span ID + assert result.accepted == 1 + context = provider.tracer.calls[0]["context"] + assert isinstance(context, dict) + parent_span = context["parent"] + assert isinstance(parent_span, FakeNonRecordingSpan) + assert parent_span.span_context.span_id == int(replacement_span_id, 16) + + +def test_otel_sink_drops_only_events_that_fail_to_emit() -> None: + # Given: an active external provider whose tracer fails during span creation + configure_settings(otel_enabled=True) + provider = FakeTracerProvider() + provider.tracer.raise_on_start = True + + # When: writing a control event + with patch( + "agent_control.otel_sink._load_otel_sdk_modules", + return_value=_fake_otel_sdk_modules(), + ): + sink = create_otel_control_event_sink( + {}, + tracer_provider=provider, # type: ignore[arg-type] + ) + result = sink.write_events([_make_event()]) + + # Then: the failure is contained and reported as one dropped event + assert result.accepted == 0 + assert result.dropped == 1 + + def test_observability_uses_builtin_otel_sink_when_selected() -> None: configure_settings( observability_sink_name=OTEL_CONTROL_EVENT_SINK_NAME,