Add suppress_invoke_agent_input option to core observability SDK (#111)

* Initial plan

* Add suppress_invoke_agent_input option to OpenAI trace processor

Co-authored-by: sergioescalera <8428450+sergioescalera@users.noreply.github.com>

* Refactor: Address code review feedback - reduce duplication and improve efficiency

Co-authored-by: sergioescalera <8428450+sergioescalera@users.noreply.github.com>

* Refactor: Use parent span operation name check instead of tracking dictionary

Co-authored-by: sergioescalera <8428450+sergioescalera@users.noreply.github.com>

* Fix: Pass both span and otel_span to _should_suppress_input as suggested in review

Co-authored-by: juliomenendez <9697+juliomenendez@users.noreply.github.com>

* Move suppress_invoke_agent_input to core SDK (exporter and span processor)

Co-authored-by: juliomenendez <9697+juliomenendez@users.noreply.github.com>

* Fix suppression logic: check current span instead of parent, remove unused span_processor param

Co-authored-by: nikhilNava <211831449+nikhilNava@users.noreply.github.com>

* Fix formatting: multi-line imports and remove trailing whitespace

Co-authored-by: nikhilNava <211831449+nikhilNava@users.noreply.github.com>

* Fix test: add suppress_invoke_agent_input parameter to exporter assertion

Co-authored-by: nikhilNava <211831449+nikhilNava@users.noreply.github.com>

---------

Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
Co-authored-by: sergioescalera <8428450+sergioescalera@users.noreply.github.com>
Co-authored-by: juliomenendez <9697+juliomenendez@users.noreply.github.com>
Co-authored-by: nikhilNava <211831449+nikhilNava@users.noreply.github.com>

Author: 4 people

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

@@ -53,6 +53,7 @@ def configure(
         token_resolver: Callable[[str, str], str | None] | None = None,
         cluster_category: str = "prod",
         exporter_options: Optional[Agent365ExporterOptions] = None,
+        suppress_invoke_agent_input: bool = False,
         **kwargs: Any,
     ) -> bool:
         """
@@ -67,6 +68,7 @@ def configure(
             Use exporter_options instead.
         :param exporter_options: Agent365ExporterOptions instance for configuring the exporter.
             If provided, exporter_options takes precedence. If exporter_options is None, the token_resolver and cluster_category parameters are used as fallback/legacy support to construct a default Agent365ExporterOptions instance.
+        :param suppress_invoke_agent_input: If True, suppress input messages for spans that are children of InvokeAgent spans.
         :return: True if configuration succeeded, False otherwise.
         """
         try:
@@ -78,6 +80,7 @@ def configure(
                     token_resolver,
                     cluster_category,
                     exporter_options,
+                    suppress_invoke_agent_input,
                     **kwargs,
                 )
         except Exception as e:
@@ -92,6 +95,7 @@ def _configure_internal(
         token_resolver: Callable[[str, str], str | None] | None = None,
         cluster_category: str = "prod",
         exporter_options: Optional[Agent365ExporterOptions] = None,
+        suppress_invoke_agent_input: bool = False,
         **kwargs: Any,
     ) -> bool:
         """Internal configuration method - not thread-safe, must be called with lock."""
@@ -151,6 +155,7 @@ def _configure_internal(
                 token_resolver=exporter_options.token_resolver,
                 cluster_category=exporter_options.cluster_category,
                 use_s2s_endpoint=exporter_options.use_s2s_endpoint,
+                suppress_invoke_agent_input=suppress_invoke_agent_input,
             )
         else:
             exporter = ConsoleSpanExporter()

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

@@ -18,6 +18,11 @@
 from opentelemetry.sdk.trace.export import SpanExporter, SpanExportResult
 from opentelemetry.trace import StatusCode
 
+from ..constants import (
+    GEN_AI_INPUT_MESSAGES_KEY,
+    GEN_AI_OPERATION_NAME_KEY,
+    INVOKE_AGENT_OPERATION_NAME,
+)
 from .utils import (
     get_validated_domain_override,
     hex_span_id,
@@ -53,6 +58,7 @@ def __init__(
         token_resolver: Callable[[str, str], str | None],
         cluster_category: str = "prod",
         use_s2s_endpoint: bool = False,
+        suppress_invoke_agent_input: bool = False,
     ):
         if token_resolver is None:
             raise ValueError("token_resolver must be provided.")
@@ -62,6 +68,7 @@ def __init__(
         self._token_resolver = token_resolver
         self._cluster_category = cluster_category
         self._use_s2s_endpoint = use_s2s_endpoint
+        self._suppress_invoke_agent_input = suppress_invoke_agent_input
         # Read domain override once at initialization
         self._domain_override = get_validated_domain_override()
 
@@ -279,6 +286,20 @@ def _map_span(self, sp: ReadableSpan) -> dict[str, Any]:
 
         # attributes
         attrs = dict(sp.attributes or {})
+
+        # Suppress input messages if configured and current span is an InvokeAgent span
+        if self._suppress_invoke_agent_input:
+            # Check if current span is an InvokeAgent span by:
+            # 1. Span name starts with "invoke_agent"
+            # 2. Has attribute gen_ai.operation.name set to INVOKE_AGENT_OPERATION_NAME
+            operation_name = attrs.get(GEN_AI_OPERATION_NAME_KEY)
+            if (
+                sp.name.startswith(INVOKE_AGENT_OPERATION_NAME)
+                and operation_name == INVOKE_AGENT_OPERATION_NAME
+            ):
+                # Remove input messages attribute
+                attrs.pop(GEN_AI_INPUT_MESSAGES_KEY, None)
+
         # events
         events = []
         for ev in sp.events:

tests/observability/core/test_agent365.py

@@ -119,6 +119,7 @@ def test_batch_span_processor_and_exporter_called_with_correct_values(
             token_resolver=self.mock_token_resolver,
             cluster_category="staging",
             use_s2s_endpoint=True,
+            suppress_invoke_agent_input=False,
         )
 
         # Verify BatchSpanProcessor was called with correct parameters from exporter_options

tests/observability/extensions/openai/test_prompt_suppression.py

@@ -0,0 +1,59 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import unittest
+
+from microsoft_agents_a365.observability.core.exporters.agent365_exporter import _Agent365Exporter
+
+
+class TestPromptSuppressionConfiguration(unittest.TestCase):
+    """Unit tests for prompt suppression configuration in the core SDK."""
+
+    def test_exporter_default_suppression_is_false(self):
+        """Test that the default value for suppress_invoke_agent_input is False in exporter."""
+        exporter = _Agent365Exporter(token_resolver=lambda x, y: "test")
+
+        self.assertFalse(
+            exporter._suppress_invoke_agent_input,
+            "Default value for suppress_invoke_agent_input should be False",
+        )
+
+    def test_exporter_can_enable_suppression(self):
+        """Test that suppression can be enabled via exporter constructor."""
+        exporter = _Agent365Exporter(
+            token_resolver=lambda x, y: "test", suppress_invoke_agent_input=True
+        )
+
+        self.assertTrue(
+            exporter._suppress_invoke_agent_input,
+            "suppress_invoke_agent_input should be True when explicitly set",
+        )
+
+
+def run_tests():
+    """Run all prompt suppression configuration tests."""
+    print("🧪 Running prompt suppression configuration tests...")
+    print("=" * 80)
+
+    loader = unittest.TestLoader()
+    suite = loader.loadTestsFromTestCase(TestPromptSuppressionConfiguration)
+
+    runner = unittest.TextTestRunner(verbosity=2)
+    result = runner.run(suite)
+
+    print("\n" + "=" * 80)
+    print("🏁 Test Summary:")
+    print(f"Tests run: {result.testsRun}")
+    print(f"Failures: {len(result.failures)}")
+    print(f"Errors: {len(result.errors)}")
+
+    if result.wasSuccessful():
+        print("🎉 All tests passed!")
+        return True
+    else:
+        print("🔧 Some tests failed. Check output above.")
+        return False
+
+
+if __name__ == "__main__":
+    success = run_tests()
+    exit(0 if success else 1)