Add configurable event sinks

Author: namrataghadi-galileo

sdks/python/src/agent_control/__init__.py

@@ -95,8 +95,13 @@ async def handle_input(user_message: str) -> str:
     sync_shutdown_observability,
 )
 from .telemetry import (
+    ControlEventSink,
+    clear_control_event_sink,
     clear_trace_context_provider,
+    emit_control_events,
     get_trace_context_from_provider,
+    has_control_event_sink,
+    set_control_event_sink,
     set_trace_context_provider,
 )
 from .tracing import (
@@ -477,6 +482,7 @@ async def handle(message: str):
 
     # Re-init behavior: always stop existing loop before mutating shared agent/session globals.
     _stop_policy_refresh_loop()
+    clear_control_event_sink()
 
     # Configure logging if provided (do this early before any logging happens)
     if log_config:
@@ -641,6 +647,8 @@ def _reset_state() -> None:
     """Clear all global SDK state."""
     global _session_generation
 
+    clear_control_event_sink()
+
     with _session_lock:
         _session_generation += 1
         state.current_agent = None
@@ -1313,6 +1321,11 @@ async def main():
     "set_trace_context_provider",
     "get_trace_context_from_provider",
     "clear_trace_context_provider",
+    "ControlEventSink",
+    "set_control_event_sink",
+    "has_control_event_sink",
+    "emit_control_events",
+    "clear_control_event_sink",
     # Observability
     "init_observability",
     "add_event",

sdks/python/src/agent_control/evaluation.py

@@ -19,6 +19,7 @@
 from .client import AgentControlClient
 from .evaluation_events import build_control_execution_events, enqueue_observability_events
 from .observability import is_observability_enabled
+from .telemetry.event_sink import emit_control_events
 from .tracing import get_trace_and_span_ids
 from .validation import ensure_agent_name
 
@@ -116,6 +117,19 @@ def _has_applicable_prefiltered_server_controls(
     )
 
 
+def _has_merged_event_delivery() -> bool:
+    """Return whether merged event delivery has an active downstream consumer.
+
+    Args:
+        None.
+
+    Returns:
+        ``True`` when observability is enabled, which is the prerequisite for
+        merged event creation and final delivery.
+    """
+    return is_observability_enabled()
+
+
 def _is_merged_event_mode_enabled(
     agent_name: str,
     client: AgentControlClient | None = None,
@@ -124,8 +138,8 @@ def _is_merged_event_mode_enabled(
 
     Merged event creation is a session-scoped option that depends on
     initialized SDK state: an initialized agent, cached server controls for the
-    same agent, and observability enabled so the merged batch will actually be
-    consumed by the built-in queue/Postgres path.
+    same agent, observability enabled, and a valid initialized session so the
+    merged batch can be delivered safely through the configured ingestion path.
 
     Args:
         agent_name: Normalized agent name for the current request.
@@ -135,9 +149,9 @@ def _is_merged_event_mode_enabled(
 
     Returns:
         ``True`` when the current SDK session has enough state to reconstruct
-        and enqueue merged events safely.
+        and deliver merged events safely.
     """
-    if not state.merge_events or not is_observability_enabled():
+    if not state.merge_events or not _has_merged_event_delivery():
         return False
 
     current_agent = state.current_agent
@@ -216,8 +230,10 @@ async def check_evaluation(
     also participate in merged event creation when the initialized SDK session
     has ``merge_events`` enabled. In that mode, the SDK asks the server to skip
     final event ingestion, reconstructs server events from the lightweight
-    response, and enqueues them through the existing observability pipeline
-    before returning the parsed result.
+    response, and delivers them before returning the parsed result. Final
+    delivery uses the configured event ingestion path: the existing built-in
+    queue by default or a registered custom sink. Event creation and delivery
+    are both gated on observability being enabled.
 
     Args:
         client: Configured AgentControl client.
@@ -268,7 +284,7 @@ async def check_evaluation(
             span_id,
             normalized_name,
         )
-        enqueue_observability_events(server_events)
+        emit_control_events(server_events)
 
     return cast(EvaluationResult, EvaluationResult.from_dict(evaluation_response.model_dump()))
 
@@ -290,10 +306,11 @@ async def check_evaluation_with_local(
     - default behavior: local events are reconstructed and queued immediately in
       the SDK, while server-side events are still emitted by the server
     - merged-event behavior: local and server events are reconstructed in the
-      SDK and enqueued once through the built-in observability path
+      SDK and delivered once through the configured event ingestion path
 
     In both cases, the evaluation result itself stays lightweight and event
-    reconstruction happens after evaluation completes.
+    reconstruction happens after evaluation completes. Event creation and
+    delivery are both gated on observability being enabled.
 
     Args:
         client: Configured AgentControl client.
@@ -420,7 +437,7 @@ def _with_parse_errors(result: EvaluationResult) -> EvaluationResult:
         if not local_result.is_safe:
             result = _with_parse_errors(EvaluationResult.model_validate(local_result.model_dump()))
             if merged_emission_enabled:
-                enqueue_observability_events(local_events)
+                emit_control_events(local_events)
             return result
 
     if _has_applicable_prefiltered_server_controls(server_control_payloads, request):
@@ -455,18 +472,18 @@ def _with_parse_errors(result: EvaluationResult) -> EvaluationResult:
         if local_result is not None:
             result = _with_parse_errors(_merge_results(local_result, server_result))
             if merged_emission_enabled:
-                enqueue_observability_events(local_events + server_events)
+                emit_control_events(local_events + server_events)
             return result
 
         result = _with_parse_errors(EvaluationResult.model_validate(server_result.model_dump()))
         if merged_emission_enabled:
-            enqueue_observability_events(server_events)
+            emit_control_events(server_events)
         return result
 
     if local_result is not None:
         result = _with_parse_errors(EvaluationResult.model_validate(local_result.model_dump()))
         if merged_emission_enabled:
-            enqueue_observability_events(local_events)
+            emit_control_events(local_events)
         return result
 
     return _with_parse_errors(EvaluationResult(is_safe=True, confidence=1.0))

sdks/python/src/agent_control/telemetry/__init__.py

@@ -1,4 +1,11 @@
-"""Telemetry interfaces for provider-agnostic tracing."""
+"""Telemetry interfaces for provider-agnostic tracing and event delivery."""
+from .event_sink import (
+    ControlEventSink,
+    clear_control_event_sink,
+    emit_control_events,
+    has_control_event_sink,
+    set_control_event_sink,
+)
 from .trace_context import (
     TraceContext,
     TraceContextProvider,
@@ -8,9 +15,14 @@
 )
 
 __all__ = [
+    "ControlEventSink",
     "TraceContext",
     "TraceContextProvider",
+    "clear_control_event_sink",
     "clear_trace_context_provider",
+    "emit_control_events",
     "get_trace_context_from_provider",
+    "has_control_event_sink",
+    "set_control_event_sink",
     "set_trace_context_provider",
 ]

sdks/python/src/agent_control/telemetry/event_sink.py

@@ -0,0 +1,77 @@
+"""Control event delivery interfaces for SDK observability."""
+
+from collections.abc import Callable
+
+from agent_control_models import ControlExecutionEvent
+
+from ..evaluation_events import enqueue_observability_events
+from ..observability import get_logger
+
+ControlEventSink = Callable[[list[ControlExecutionEvent]], None]
+
+_control_event_sink: ControlEventSink | None = None
+_logger = get_logger(__name__)
+
+
+def set_control_event_sink(sink: ControlEventSink | None) -> None:
+    """Register a custom sink for control event delivery.
+
+    Args:
+        sink: Callable that consumes a batch of ``ControlExecutionEvent``
+            instances. Pass ``None`` to clear any existing sink.
+
+    Returns:
+        None.
+    """
+    global _control_event_sink
+    _control_event_sink = sink
+
+
+def clear_control_event_sink() -> None:
+    """Clear the currently registered control event sink.
+
+    Args:
+        None.
+
+    Returns:
+        None.
+    """
+    set_control_event_sink(None)
+
+
+def has_control_event_sink() -> bool:
+    """Return whether a custom control event sink is registered.
+
+    Args:
+        None.
+
+    Returns:
+        ``True`` when a custom sink is registered, otherwise ``False``.
+    """
+    return _control_event_sink is not None
+
+
+def emit_control_events(events: list[ControlExecutionEvent]) -> None:
+    """Deliver control events through the configured ingestion path.
+
+    By default, events continue through the existing SDK observability queue.
+    When a custom sink is registered, the same created events are delivered to
+    that sink instead. Event creation mode remains a separate concern.
+
+    Args:
+        events: Control execution events that have already been created.
+
+    Returns:
+        None.
+    """
+    if not events:
+        return
+
+    if _control_event_sink is not None:
+        try:
+            _control_event_sink(events)
+        except Exception:
+            _logger.exception("Custom control event sink failed while delivering events")
+        return
+
+    enqueue_observability_events(events)

sdks/python/tests/test_event_sink.py

@@ -0,0 +1,86 @@
+"""Tests for configurable control event sinks."""
+
+from unittest.mock import MagicMock, patch
+
+from agent_control.telemetry.event_sink import (
+    clear_control_event_sink,
+    emit_control_events,
+    has_control_event_sink,
+    set_control_event_sink,
+)
+from agent_control_models import ControlExecutionEvent
+
+
+def _make_event() -> ControlExecutionEvent:
+    return ControlExecutionEvent(
+        trace_id="a" * 32,
+        span_id="b" * 16,
+        agent_name="agent-000000000001",
+        control_id=1,
+        control_name="ctrl-1",
+        check_stage="pre",
+        applies_to="llm_call",
+        action="allow",
+        matched=True,
+        confidence=1.0,
+    )
+
+
+def teardown_function() -> None:
+    clear_control_event_sink()
+
+
+def test_emit_control_events_uses_default_enqueue_path_without_custom_sink() -> None:
+    events = [_make_event()]
+
+    with patch("agent_control.telemetry.event_sink.enqueue_observability_events") as mock_enqueue:
+        emit_control_events(events)
+
+    mock_enqueue.assert_called_once_with(events)
+
+
+def test_emit_control_events_uses_registered_custom_sink() -> None:
+    events = [_make_event()]
+    sink = MagicMock()
+    set_control_event_sink(sink)
+
+    with patch("agent_control.telemetry.event_sink.enqueue_observability_events") as mock_enqueue:
+        emit_control_events(events)
+
+    sink.assert_called_once_with(events)
+    mock_enqueue.assert_not_called()
+
+
+def test_emit_control_events_swallows_custom_sink_errors() -> None:
+    events = [_make_event()]
+    sink = MagicMock(side_effect=RuntimeError("boom"))
+    set_control_event_sink(sink)
+
+    with patch("agent_control.telemetry.event_sink._logger") as mock_logger, \
+         patch("agent_control.telemetry.event_sink.enqueue_observability_events") as mock_enqueue:
+        emit_control_events(events)
+
+    sink.assert_called_once_with(events)
+    mock_logger.exception.assert_called_once()
+    mock_enqueue.assert_not_called()
+
+
+def test_emit_control_events_noops_for_empty_event_batch() -> None:
+    sink = MagicMock()
+    set_control_event_sink(sink)
+
+    with patch("agent_control.telemetry.event_sink.enqueue_observability_events") as mock_enqueue:
+        emit_control_events([])
+
+    sink.assert_not_called()
+    mock_enqueue.assert_not_called()
+
+
+def test_has_control_event_sink_tracks_registration() -> None:
+    assert has_control_event_sink() is False
+
+    set_control_event_sink(lambda events: None)
+    assert has_control_event_sink() is True
+
+    clear_control_event_sink()
+    assert has_control_event_sink() is False