feat(sdk): 59790 configurable event sinks

models/src/agent_control_models/evaluation.py

@@ -127,8 +127,6 @@ class EvaluationResponse(BaseModel):
         default=None,
         description="List of controls that were evaluated but did not match (if any)",
     )
-
-
 class EvaluationResult(EvaluationResponse):
     """
     Client-side result model for evaluation analysis.

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 (
@@ -396,6 +401,7 @@ def init(
     observability_enabled: bool | None = None,
     log_config: dict[str, Any] | None = None,
     policy_refresh_interval_seconds: int = 60,
+    merge_events: bool = False,
     **kwargs: object
 ) -> Agent:
     """
@@ -426,6 +432,9 @@ def init(
                {"enabled": True, "span_start": True, "span_end": True, "control_eval": True}
         policy_refresh_interval_seconds: Interval for background policy refresh loop.
             Defaults to 60 seconds. Set to 0 to disable background refresh.
+        merge_events: Whether to merge local and server event creation in the
+            SDK before enqueueing through the built-in observability path.
+            Defaults to False.
         **kwargs: Additional metadata to store with the agent
 
     Returns:
@@ -496,6 +505,7 @@ async def handle(message: str):
         state.current_agent = next_agent
         state.server_url = server_url or os.getenv('AGENT_CONTROL_URL') or 'http://localhost:8000'
         state.api_key = api_key
+        state.merge_events = merge_events
 
     # Merge auto-discovered steps from @control() decorators with explicit steps.
     # Explicit steps take precedence when (type, name) collides.
@@ -547,6 +557,7 @@ async def register() -> list[dict[str, Any]] | None:
                         state.current_agent,
                         steps=registration_steps,
                         conflict_mode=conflict_mode,
+                        merge_events=merge_events,
                     )
                     created = response.get('created', False)
                     controls: list[dict[str, Any]] = response.get('controls', [])
@@ -636,13 +647,16 @@ 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
         state.control_engine = None
         state.server_controls = None
         state.server_url = None
         state.api_key = None
+        state.merge_events = False
 
 
 async def ashutdown() -> None:
@@ -1307,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/_state.py

@@ -24,6 +24,7 @@ def __init__(self) -> None:
         self.server_controls: list[dict[str, Any]] | None = None
         self.server_url: str | None = None
         self.api_key: str | None = None
+        self.merge_events: bool = False
 
 
 # Singleton state instance

sdks/python/src/agent_control/agents.py

@@ -15,6 +15,7 @@ async def register_agent(
     agent: Agent,
     steps: list[dict[str, Any]] | None = None,
     conflict_mode: Literal["strict", "overwrite"] = "overwrite",
+    merge_events: bool = False,
 ) -> dict[str, Any]:
     """Register an agent with the server via /initAgent endpoint."""
     ensure_evaluators_discovered()
@@ -27,7 +28,12 @@ async def register_agent(
         "conflict_mode": conflict_mode,
     }
 
-    response = await client.http_client.post("/api/v1/agents/initAgent", json=payload)
+    headers = {"X-Agent-Control-Merge-Session": "true"} if merge_events else None
+    response = await client.http_client.post(
+        "/api/v1/agents/initAgent",
+        json=payload,
+        headers=headers,
+    )
     response.raise_for_status()
     return cast(dict[str, Any], response.json())