feat(sdk): add config driven sink selection (#176)

## Summary

Added config-driven observability sink selection across the SDK and
server.

Introduced shared telemetry sink-selection primitives so observability
can use the default backend, registered SDK sinks, or named custom sink
factories.

Expanded tests and README examples to cover the new sink-selection
behavior and lifecycle handling.

Fixed the SDK mypy issue in shutdown handling by wrapping sink lifecycle
awaitables in a concrete coroutine before passing them to
`asyncio.run()`.

## Scope

**User-facing/API changes:**
- SDK `agent_control.init()` now accepts `observability_sink_name` and
`observability_sink_config`
- SDK exports new sink registration helpers for external/custom
observability sinks
- Server observability config now supports sink selection and sink
config via environment/settings
- README includes a new example for registering and selecting an
external sink

**Internal changes:**
- Added shared sink-selection models/registry in `telemetry/`
- Refactored SDK observability to resolve active sinks from config
rather than always using the built-in batcher
- Refactored server startup to resolve an observability backend from
sink selection
- Added server/SDK/telemetry tests for sink selection, lifecycle, and
event delivery paths
- Adjusted SDK shutdown typing to satisfy mypy

**Out of scope:**
- No changes to core control evaluation logic
- No new built-in third-party sink implementation is included in this
branch
- No UI work or rollout automation is included

## Risk and Rollout

**Risk level:** medium

**Rollback plan:** Revert this branch to restore the previous
default-only observability path. As a partial mitigation, deployments
can keep `observability_sink_name=default` to stay on the legacy
behavior even with this code merged.

## Testing

- [x] Added or updated automated tests
- [ ] Ran `make check` — could not run in this environment because `uv`
was not available in the shell
- [x] Manually verified behavior

## Checklist

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

**Follow-up tasks:**
- Run `make check` in a full local/CI environment with `uv` available
- Confirm any downstream custom sink integrations against the new
SDK/server registration APIs

Author: namrataghadi-galileo

README.md

@@ -149,7 +149,7 @@ if __name__ == "__main__":
 Use `agent_control.shutdown()` or `await agent_control.ashutdown()` before process exit so short-lived scripts flush pending observability events cleanly.
 
 External integrations can register a sink for the same finalized
-control-event payloads:
+control-event payloads and then select it through observability config:
 
 ```python
 from agent_control import (
@@ -169,14 +169,21 @@ class MyControlEventSink(BaseControlEventSink):
 sink = MyControlEventSink()
 register_control_event_sink(sink)
 
+# Select registered sinks instead of the default SDK -> server sink.
+agent_control.init(
+    agent_name="awesome_bot_3000",
+    observability_enabled=True,
+    observability_sink_name="registered",
+)
+
 # Later, when tearing down the integration:
 unregister_control_event_sink(sink)
 ```
 
 Registered sinks receive the same local, server, and merged control-execution
-events the SDK emits through its normal event-construction flow. If no
-external sink is registered, the default OSS delivery path is unchanged. If one
-or more sinks are registered, they replace the default built-in delivery path.
+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.
 
 Next, create a control in Step 4, then run the setup and agent scripts in
 order to see blocking in action.

sdks/python/src/agent_control/__init__.py

@@ -65,6 +65,7 @@ async def handle_input(user_message: str) -> str:
     EvaluationResult,
     EvaluatorResult,
     EvaluatorSpec,
+    JSONObject,
     Step,
     StepSchema,
     TemplateControlInput,
@@ -98,14 +99,17 @@ async def handle_input(user_message: str) -> str:
     get_event_sink,
     get_log_config,
     get_logger,
+    get_registered_control_event_sink_factory_names,
     get_registered_control_event_sinks,
     init_observability,
     is_observability_enabled,
     log_control_evaluation,
     register_control_event_sink,
+    register_control_event_sink_factory,
     shutdown_observability,
     sync_shutdown_observability,
     unregister_control_event_sink,
+    unregister_control_event_sink_factory,
     write_events,
 )
 from .tracing import (
@@ -415,6 +419,8 @@ def init(
     steps: list[StepSchemaDict] | None = None,
     conflict_mode: Literal["strict", "overwrite"] = "overwrite",
     observability_enabled: bool | None = None,
+    observability_sink_name: str | None = None,
+    observability_sink_config: JSONObject | None = None,
     log_config: dict[str, Any] | None = None,
     policy_refresh_interval_seconds: int = 60,
     **kwargs: object
@@ -443,6 +449,9 @@ def init(
         conflict_mode: Conflict handling mode for initAgent registration.
             Defaults to "overwrite" in SDK flows.
         observability_enabled: Optional bool to enable/disable observability (defaults to env var)
+        observability_sink_name: Optional sink selection name. Use "default" to preserve
+            SDK -> server OSS delivery or "registered" / a named sink factory for custom sinks.
+        observability_sink_config: Optional JSON config payload for the selected sink.
         log_config: Optional logging configuration dict:
                {"enabled": True, "span_start": True, "span_end": True, "control_eval": True}
         policy_refresh_interval_seconds: Interval for background policy refresh loop.
@@ -636,6 +645,8 @@ def run_in_thread() -> None:
         server_url=state.server_url,
         api_key=state.api_key,
         enabled=observability_enabled,
+        sink_name=observability_sink_name,
+        sink_config=observability_sink_config,
     )
     if batcher:
         logger.info("Observability enabled")
@@ -1371,9 +1382,12 @@ async def main():
     "is_observability_enabled",
     "get_event_batcher",
     "get_event_sink",
+    "get_registered_control_event_sink_factory_names",
     "get_registered_control_event_sinks",
     "register_control_event_sink",
+    "register_control_event_sink_factory",
     "unregister_control_event_sink",
+    "unregister_control_event_sink_factory",
     "configure_logging",
     "get_log_config",
     "log_control_evaluation",