@@ -95,12 +95,8 @@ async def handle_input(user_message: str) -> str:
     sync_shutdown_observability,
 )
 from .telemetry import (
-    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 (
@@ -400,6 +396,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:
     """
@@ -430,6 +427,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:
@@ -477,7 +477,6 @@ 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:
@@ -501,6 +500,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.
@@ -648,7 +648,7 @@ def _reset_state() -> None:
         state.server_controls = None
         state.server_url = None
         state.api_key = None
-        clear_control_event_sink()
+        state.merge_events = False
 
 
 async def ashutdown() -> None:
@@ -1313,10 +1313,6 @@ async def main():
     "set_trace_context_provider",
     "get_trace_context_from_provider",
     "clear_trace_context_provider",
-    "set_control_event_sink",
-    "has_control_event_sink",
-    "emit_control_events",
-    "clear_control_event_sink",
     # Observability
     "init_observability",
     "add_event",
 
@@ -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
 
@@ -19,7 +19,6 @@
 from .client import AgentControlClient
 from .evaluation_events import build_control_execution_events, enqueue_observability_events
 from .observability import is_observability_enabled
-from .telemetry import emit_control_events, has_control_event_sink
 from .tracing import get_trace_and_span_ids
 from .validation import ensure_agent_name
 
@@ -117,29 +116,40 @@ def _has_applicable_prefiltered_server_controls(
     )
 
 
-def _is_merged_event_mode_enabled(agent_name: str) -> bool:
-    """Return whether SDK-side merged event emission is safe for this request.
+def _is_merged_event_mode_enabled(
+    agent_name: str,
+    client: AgentControlClient | None = None,
+) -> bool:
+    """Return whether SDK-side merged event creation is safe for this request.
 
-    Merged reconstruction depends on initialized SDK session state:
-    a registered sink, an initialized agent, and a cached server-control
-    snapshot for the same agent. If any of those prerequisites are missing,
-    evaluation falls back to the default behavior where the server remains the
-    final emitter for server-side events.
+    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.
 
     Args:
         agent_name: Normalized agent name for the current request.
+        client: Optional client used for the current request. When provided,
+            merged mode is allowed only if it targets the active initialized
+            session server.
 
     Returns:
         ``True`` when the current SDK session has enough state to reconstruct
-        and emit merged events safely.
+        and enqueue merged events safely.
     """
-    if not has_control_event_sink():
+    if not state.merge_events or not is_observability_enabled():
         return False
 
     current_agent = state.current_agent
     if current_agent is None or current_agent.agent_name != agent_name:
         return False
 
+    if client is not None and state.server_url is not None:
+        normalized_state_server_url = state.server_url.rstrip("/")
+        normalized_client_server_url = client.base_url.rstrip("/")
+        if normalized_client_server_url != normalized_state_server_url:
+            return False
+
     return state.server_controls is not None
 
 
@@ -202,11 +212,12 @@ async def check_evaluation(
 ) -> EvaluationResult:
     """Check if agent interaction is safe through the public SDK helper.
 
-    This helper preserves the default server-only evaluation path, but it also
-    participates in the optional merged-event flow when a control event sink is
-    registered. In that mode, the SDK asks the server to skip final event
-    ingestion, reconstructs server events from the lightweight response, and
-    emits them through the registered sink before returning the parsed result.
+    This helper preserves the default server-only evaluation path, but it can
+    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.
 
     Args:
         client: Configured AgentControl client.
@@ -218,7 +229,7 @@ async def check_evaluation(
         The parsed evaluation result returned by the server.
     """
     normalized_name = ensure_agent_name(agent_name)
-    merged_emission_enabled = _is_merged_event_mode_enabled(normalized_name)
+    merged_emission_enabled = _is_merged_event_mode_enabled(normalized_name, client)
     trace_id = None
     span_id = None
     headers: dict[str, str] | None = None
@@ -257,7 +268,7 @@ async def check_evaluation(
             span_id,
             normalized_name,
         )
-        emit_control_events(server_events)
+        enqueue_observability_events(server_events)
 
     return cast(EvaluationResult, EvaluationResult.from_dict(evaluation_response.model_dump()))
 
@@ -275,11 +286,11 @@ async def check_evaluation_with_local(
     """Evaluate controls with local-first execution and configurable event flow.
 
     This is the main decision boundary between the two supported event
-    ingestion styles:
+    creation styles:
     - 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 emitted once through a registered sink
+      SDK and enqueued once through the built-in observability path
 
     In both cases, the evaluation result itself stays lightweight and event
     reconstruction happens after evaluation completes.
@@ -377,8 +388,8 @@ def _with_parse_errors(result: EvaluationResult) -> EvaluationResult:
         combined_errors = (result.errors or []) + parse_errors
         return result.model_copy(update={"errors": combined_errors})
 
-    merged_emission_enabled = _is_merged_event_mode_enabled(normalized_name)
-    should_reconstruct_local_events = merged_emission_enabled or is_observability_enabled()
+    merged_emission_enabled = _is_merged_event_mode_enabled(normalized_name, client)
+    should_reconstruct_local_events = is_observability_enabled()
 
     local_result: EvaluationResponse | None = None
     local_events = []
@@ -409,7 +420,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:
-                emit_control_events(local_events)
+                enqueue_observability_events(local_events)
             return result
 
     if _has_applicable_prefiltered_server_controls(server_control_payloads, request):
@@ -444,18 +455,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:
-                emit_control_events(local_events + server_events)
+                enqueue_observability_events(local_events + server_events)
             return result
 
         result = _with_parse_errors(EvaluationResult.model_validate(server_result.model_dump()))
         if merged_emission_enabled:
-            emit_control_events(server_events)
+            enqueue_observability_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:
-            emit_control_events(local_events)
+            enqueue_observability_events(local_events)
         return result
 
     return _with_parse_errors(EvaluationResult(is_safe=True, confidence=1.0))
 
@@ -127,12 +127,12 @@ def build_control_execution_events(
 ) -> list[ControlExecutionEvent]:
     """Reconstruct control execution events from an evaluation response.
 
-    This is the shared reconstruction step used by both supported ingestion
-    styles:
+    This is the shared reconstruction step used by both supported event
+    creation styles:
     - the default SDK observability path, where reconstructed local events are
       queued into the existing SDK batcher
     - the merged-event path, where local and server events are reconstructed in
-      the SDK and emitted together through a registered sink
+      the SDK and queued together through the existing SDK batcher
 
     Args:
         response: Evaluation response containing matches, errors, and
@@ -194,8 +194,8 @@ def build_control_execution_events(
 def enqueue_observability_events(events: list[ControlExecutionEvent]) -> None:
     """Enqueue reconstructed events through the existing SDK observability path.
 
-    This preserves the default SDK behavior of forwarding local events through
-    the existing observability batcher rather than a custom merged-event sink.
+    This preserves the built-in SDK behavior of forwarding events through the
+    existing observability batcher.
 
     Args:
         events: Reconstructed control execution events to enqueue.
 
@@ -1,12 +1,4 @@
-"""Telemetry interfaces for provider-agnostic tracing and event emission."""
-
-from .event_sink import (
-    ControlEventSink,
-    clear_control_event_sink,
-    emit_control_events,
-    has_control_event_sink,
-    set_control_event_sink,
-)
+"""Telemetry interfaces for provider-agnostic tracing."""
 from .trace_context import (
     TraceContext,
     TraceContextProvider,
@@ -16,14 +8,9 @@
 )
 
 __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",
 ]