Merge branch 'main' into docs/integrating-with-existing-opentelemetry

Author: juliomenendez

libraries/microsoft-agents-a365-notifications/microsoft_agents_a365/notifications/agent_notification.py

@@ -426,6 +426,121 @@ async def handle_user_deleted(context, state, notification):
         """
         return self.on_lifecycle_notification(AgentLifecycleEvent.USERDELETED, **kwargs)
 
+    def on_user_undeleted(
+        self, **kwargs: Any
+    ) -> Callable[[AgentHandler], Callable[[TurnContext, TurnState], Awaitable[None]]]:
+        """Register a handler for user un-deletion lifecycle events.
+
+        This is a convenience decorator that registers a handler specifically for
+        agentic user identity un-deletion events.
+
+        Args:
+            **kwargs: Additional keyword arguments passed to the app's add_route method.
+
+        Returns:
+            A decorator function that registers the handler with the application.
+
+        Example:
+            ```python
+            @notifications.on_user_undeleted()
+            async def handle_user_undeleted(context, state, notification):
+                print("Agentic user identity un-deleted")
+            ```
+        """
+        return self.on_lifecycle_notification(AgentLifecycleEvent.USERUNDELETED, **kwargs)
+
+    def on_user_updated(
+        self, **kwargs: Any
+    ) -> Callable[[AgentHandler], Callable[[TurnContext, TurnState], Awaitable[None]]]:
+        """Register a handler for user updated lifecycle events.
+
+        This is a convenience decorator that registers a handler specifically for
+        agentic user identity updated events.
+
+        Args:
+            **kwargs: Additional keyword arguments passed to the app's add_route method.
+
+        Returns:
+            A decorator function that registers the handler with the application.
+
+        Example:
+            ```python
+            @notifications.on_user_updated()
+            async def handle_user_updated(context, state, notification):
+                print("Agentic user identity updated")
+            ```
+        """
+        return self.on_lifecycle_notification(AgentLifecycleEvent.USERUPDATED, **kwargs)
+
+    def on_user_manager_updated(
+        self, **kwargs: Any
+    ) -> Callable[[AgentHandler], Callable[[TurnContext, TurnState], Awaitable[None]]]:
+        """Register a handler for user manager updated lifecycle events.
+
+        This is a convenience decorator that registers a handler specifically for
+        agentic user manager updated events.
+
+        Args:
+            **kwargs: Additional keyword arguments passed to the app's add_route method.
+
+        Returns:
+            A decorator function that registers the handler with the application.
+
+        Example:
+            ```python
+            @notifications.on_user_manager_updated()
+            async def handle_user_manager_updated(context, state, notification):
+                print("Agentic user manager updated")
+            ```
+        """
+        return self.on_lifecycle_notification(AgentLifecycleEvent.USERMANAGERUPDATED, **kwargs)
+
+    def on_user_enabled(
+        self, **kwargs: Any
+    ) -> Callable[[AgentHandler], Callable[[TurnContext, TurnState], Awaitable[None]]]:
+        """Register a handler for user enabled lifecycle events.
+
+        This is a convenience decorator that registers a handler specifically for
+        agentic user enabled events.
+
+        Args:
+            **kwargs: Additional keyword arguments passed to the app's add_route method.
+
+        Returns:
+            A decorator function that registers the handler with the application.
+
+        Example:
+            ```python
+            @notifications.on_user_enabled()
+            async def handle_user_enabled(context, state, notification):
+                print("Agentic user enabled")
+            ```
+        """
+        return self.on_lifecycle_notification(AgentLifecycleEvent.USERENABLED, **kwargs)
+
+    def on_user_disabled(
+        self, **kwargs: Any
+    ) -> Callable[[AgentHandler], Callable[[TurnContext, TurnState], Awaitable[None]]]:
+        """Register a handler for user disabled lifecycle events.
+
+        This is a convenience decorator that registers a handler specifically for
+        agentic user disabled events.
+
+        Args:
+            **kwargs: Additional keyword arguments passed to the app's add_route method.
+
+        Returns:
+            A decorator function that registers the handler with the application.
+
+        Example:
+            ```python
+            @notifications.on_user_disabled()
+            async def handle_user_disabled(context, state, notification):
+                print("Agentic user disabled")
+            ```
+        """
+        return self.on_lifecycle_notification(AgentLifecycleEvent.USERDISABLED, **kwargs)
+
     @staticmethod
     def _normalize_subchannel(value: str | AgentSubChannel | None) -> str:
         """Normalize a subchannel value to a lowercase string.

libraries/microsoft-agents-a365-notifications/microsoft_agents_a365/notifications/models/agent_lifecycle_event.py

@@ -15,8 +15,18 @@ class AgentLifecycleEvent(str, Enum):
         USERWORKLOADONBOARDINGUPDATED: Event triggered when a user's workload
             onboarding status is updated.
         USERDELETED: Event triggered when an agentic user identity is deleted.
+        USERUNDELETED: Event triggered when an agentic user identity is un-deleted.
+        USERUPDATED: Event triggered when an agentic user identity is updated.
+        USERMANAGERUPDATED: Event triggered when an agentic user's manager is updated.
+        USERENABLED: Event triggered when an agentic user is enabled.
+        USERDISABLED: Event triggered when an agentic user is disabled.
     """
 
     USERCREATED = "agenticuseridentitycreated"
     USERWORKLOADONBOARDINGUPDATED = "agenticuserworkloadonboardingupdated"
-    USERDELETED = "agenticuseridentitydeleted"
+    USERDELETED = "agenticuserdeleted"
+    USERUNDELETED = "agenticuserundeleted"
+    USERUPDATED = "agenticuseridentityupdated"
+    USERMANAGERUPDATED = "agenticusermanagerupdated"
+    USERENABLED = "agenticuserenabled"
+    USERDISABLED = "agenticuserdisabled"

libraries/microsoft-agents-a365-observability-core/microsoft_agents_a365/observability/core/config.py

@@ -180,6 +180,7 @@ def _configure_internal(
                 token_resolver=exporter_options.token_resolver,
                 cluster_category=exporter_options.cluster_category,
                 use_s2s_endpoint=exporter_options.use_s2s_endpoint,
+                max_payload_bytes=exporter_options.max_payload_bytes,
             )
 
         else:

libraries/microsoft-agents-a365-observability-core/microsoft_agents_a365/observability/core/exporters/agent365_exporter.py

@@ -18,13 +18,16 @@
 from opentelemetry.trace import StatusCode
 
 from .utils import (
+    DEFAULT_MAX_PAYLOAD_BYTES,
     build_export_url,
+    chunk_by_size,
+    estimate_span_bytes,
     get_validated_domain_override,
     hex_span_id,
     hex_trace_id,
     kind_name,
     parse_retry_after,
-    partition_by_identity,
+    filter_and_partition_by_identity,
     status_name,
     truncate_span,
 )
@@ -56,6 +59,7 @@ def __init__(
         token_resolver: Callable[[str, str], str | None],
         cluster_category: str = "prod",
         use_s2s_endpoint: bool = False,
+        max_payload_bytes: int = DEFAULT_MAX_PAYLOAD_BYTES,
     ):
         if token_resolver is None:
             raise ValueError("token_resolver must be provided.")
@@ -65,6 +69,7 @@ def __init__(
         self._token_resolver = token_resolver
         self._cluster_category = cluster_category
         self._use_s2s_endpoint = use_s2s_endpoint
+        self._max_payload_bytes = max_payload_bytes
         # Read domain override once at initialization
         self._domain_override = get_validated_domain_override()
 
@@ -75,10 +80,10 @@ def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
             return SpanExportResult.FAILURE
 
         try:
-            groups = partition_by_identity(spans)
+            groups = filter_and_partition_by_identity(spans)
             if not groups:
-                # No spans with identity; treat as success
-                logger.info("No spans with tenant/agent identity found; nothing exported.")
+                # No eligible genAI spans to export after filtering/partitioning; treat as success
+                logger.info("No eligible genAI spans to export; nothing exported.")
                 return SpanExportResult.SUCCESS
 
             # Log number of groups and total span count
@@ -89,8 +94,21 @@ def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
 
             any_failure = False
             for (tenant_id, agent_id), activities in groups.items():
-                payload = self._build_export_request(activities)
-                body = json.dumps(payload, separators=(",", ":"), ensure_ascii=False)
+                # Map and truncate spans first, then chunk by estimated byte size
+                mapped_spans = self._map_and_truncate_spans(activities)
+                resource_attrs = self._get_resource_attributes(activities)
+                chunks = chunk_by_size(
+                    mapped_spans,
+                    lambda ms: estimate_span_bytes(ms[0]),
+                    self._max_payload_bytes,
+                )
+
+                if len(chunks) > 1:
+                    # Logged at DEBUG to avoid leaking tenant/agent IDs in production logs.
+                    logger.debug(
+                        f"Split {len(activities)} spans into {len(chunks)} chunks "
+                        f"for tenantId: {tenant_id}, agentId: {agent_id}"
+                    )
 
                 # Resolve endpoint: domain override > default URL
                 if self._domain_override:
@@ -128,11 +146,40 @@ def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
                     any_failure = True
                     continue
 
-                # Basic retry loop
-                ok = self._post_with_retries(url, body, headers)
-
-                if not ok:
-                    any_failure = True
+                # Send each chunk (all-or-nothing: fail group on first chunk failure)
+                group_failed = False
+                for i, chunk in enumerate(chunks):
+                    payload = self._build_envelope(chunk, resource_attrs)
+                    body = json.dumps(payload, separators=(",", ":"), ensure_ascii=False)
+                    body_bytes = len(body.encode("utf-8"))
+                    logger.debug(
+                        f"Sending chunk {i + 1} of {len(chunks)} "
+                        f"({len(chunk)} spans, {body_bytes} bytes)"
+                    )
+                    # Defensive check: the estimator covers per-span content but not
+                    # envelope overhead (resource attributes, scope wrappers). Warn if
+                    # the assembled body exceeds the configured limit so operators can
+                    # observe estimator drift before the server starts rejecting requests.
+                    if body_bytes > self._max_payload_bytes:
+                        logger.warning(
+                            f"Chunk {i + 1} of {len(chunks)} body size ({body_bytes} bytes) "
+                            f"exceeds max_payload_bytes ({self._max_payload_bytes}); "
+                            "estimator may be under-counting envelope overhead. "
+                            f"Tenant: {tenant_id}, agent: {agent_id}, spans: {len(chunk)}."
+                        )
+
+                    ok = self._post_with_retries(url, body, headers)
+                    if not ok:
+                        logger.error(
+                            f"Chunk {i + 1} of {len(chunks)} failed for "
+                            f"tenant {tenant_id}, agent {agent_id}"
+                        )
+                        any_failure = True
+                        group_failed = True
+                        break
+
+                if group_failed:
+                    continue
 
             return SpanExportResult.FAILURE if any_failure else SpanExportResult.SUCCESS
 
@@ -231,32 +278,47 @@ def _post_with_retries(self, url: str, body: str, headers: dict[str, str]) -> bo
 
     # ------------- Payload mapping ------------------
 
-    def _build_export_request(self, spans: Sequence[ReadableSpan]) -> dict[str, Any]:
-        # Group by instrumentation scope (name, version)
-        scope_map: dict[tuple[str, str | None], list[dict[str, Any]]] = {}
+    def _map_and_truncate_spans(
+        self, spans: Sequence[ReadableSpan]
+    ) -> list[tuple[dict[str, Any], str, str | None]]:
+        """Map ReadableSpans to OTLP dicts and apply per-span truncation.
 
+        Returns a list of (mapped_span, scope_name, scope_version) tuples so
+        that envelope grouping by instrumentation scope can be performed
+        efficiently after byte-size chunking.
+        """
+        result: list[tuple[dict[str, Any], str, str | None]] = []
         for sp in spans:
             scope = sp.instrumentation_scope
-            scope_key = (scope.name, scope.version)
-            scope_map.setdefault(scope_key, []).append(self._map_span(sp))
-
-        scope_spans: list[dict[str, Any]] = []
-        for (name, version), mapped_spans in scope_map.items():
-            scope_spans.append(
-                {
-                    "scope": {
-                        "name": name,
-                        "version": version,
-                    },
-                    "spans": mapped_spans,
-                }
-            )
+            scope_name = scope.name if scope is not None else "unknown"
+            scope_version = scope.version if scope is not None else None
+            result.append((self._map_span(sp), scope_name, scope_version))
+        return result
 
-        # Resource attributes (from the first span – all spans in a batch usually share resource)
-        # If you need to merge across spans, adapt accordingly.
-        resource_attrs = {}
+    @staticmethod
+    def _get_resource_attributes(spans: Sequence[ReadableSpan]) -> dict[str, Any]:
+        """Extract resource attributes from the first span in the batch."""
         if spans:
-            resource_attrs = dict(getattr(spans[0].resource, "attributes", {}) or {})
+            return dict(getattr(spans[0].resource, "attributes", {}) or {})
+        return {}
+
+    def _build_envelope(
+        self,
+        mapped_spans: Sequence[tuple[dict[str, Any], str, str | None]],
+        resource_attrs: dict[str, Any],
+    ) -> dict[str, Any]:
+        """Build an OTLP export request envelope from pre-mapped spans."""
+        scope_map: dict[tuple[str, str | None], list[dict[str, Any]]] = {}
+        for mapped_span, scope_name, scope_version in mapped_spans:
+            scope_map.setdefault((scope_name, scope_version), []).append(mapped_span)
+
+        scope_spans: list[dict[str, Any]] = [
+            {
+                "scope": {"name": name, "version": version},
+                "spans": spans,
+            }
+            for (name, version), spans in scope_map.items()
+        ]
 
         return {
             "resourceSpans": [

libraries/microsoft-agents-a365-observability-core/microsoft_agents_a365/observability/core/exporters/agent365_exporter_options.py

@@ -3,6 +3,8 @@
 
 from typing import Awaitable, Callable, Optional
 
+from .utils import DEFAULT_MAX_PAYLOAD_BYTES
+
 
 class Agent365ExporterOptions:
     """
@@ -19,6 +21,7 @@ def __init__(
         scheduled_delay_ms: int = 5000,
         exporter_timeout_ms: int = 30000,
         max_export_batch_size: int = 512,
+        max_payload_bytes: int = DEFAULT_MAX_PAYLOAD_BYTES,
     ):
         """
         Args:
@@ -29,6 +32,10 @@ def __init__(
             scheduled_delay_ms: Delay between export batches (ms). Default is 5000.
             exporter_timeout_ms: Timeout for the export operation (ms). Default is 30000.
             max_export_batch_size: Maximum batch size for export operations. Default is 512.
+            max_payload_bytes: Upper bound on HTTP request body size in bytes. The exporter
+                splits per-identity batches into sub-batches whose estimated size stays under
+                this limit, providing headroom under the A365 1 MB server limit. Default is
+                900_000 (~100 KB headroom for estimator error and JSON envelope overhead).
         """
         self.cluster_category = cluster_category
         self.token_resolver = token_resolver
@@ -37,3 +44,4 @@ def __init__(
         self.scheduled_delay_ms = scheduled_delay_ms
         self.exporter_timeout_ms = exporter_timeout_ms
         self.max_export_batch_size = max_export_batch_size
+        self.max_payload_bytes = max_payload_bytes