test: add OTEL span content leakage integration tests (#485)

* test: add integration tests asserting no user content in OTEL spans

COE Finding: User prompts and agent responses must never leak into exported
OTEL span attributes, events, or resource attributes. These tests send
requests with unique sentinel strings through the full request lifecycle
and verify they are absent from all exported span data.

Covers: BedrockAgentCoreApp (sync + streaming), A2A, AG-UI.

* fix: resolve lint errors in OTEL test (unused var, undefined name)

* refactor: remove A2A and AG-UI tests from OTEL content leakage suite

* style: remove trailing blank lines

* style: move opentelemetry.trace import to top level

Author: jesseturner21

tests/integration/runtime/test_otel_no_user_content.py

@@ -0,0 +1,108 @@
+"""Integration tests: assert user content is NOT present in OTEL spans.
+
+COE Finding: User prompts and agent responses must never leak into exported
+OTEL span attributes, events, or resource attributes.
+"""
+
+import json
+
+import opentelemetry.trace
+import pytest
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import SimpleSpanProcessor
+from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
+from starlette.testclient import TestClient
+
+from bedrock_agentcore.runtime import BedrockAgentCoreApp
+from bedrock_agentcore.runtime import tracing as tracing_mod
+
+SENTINEL_PROMPT = "SENSITIVE_USER_PROMPT_a1b2c3d4e5"
+SENTINEL_RESPONSE = "SENSITIVE_AGENT_RESPONSE_f6g7h8i9j0"
+
+
+def _assert_no_sentinel_in_spans(spans, sentinels):
+    """Assert none of the sentinel strings appear anywhere in exported span data."""
+    for span in spans:
+        for sentinel in sentinels:
+            assert sentinel not in span.name, f"Sentinel found in span name: {span.name}"
+
+            for key, value in (span.attributes or {}).items():
+                serialized = json.dumps(value) if not isinstance(value, str) else value
+                assert sentinel not in str(key), f"Sentinel in attribute key: {key}"
+                assert sentinel not in serialized, f"Sentinel in attribute value for key={key}: {serialized}"
+
+            for event in span.events:
+                assert sentinel not in event.name, f"Sentinel in event name: {event.name}"
+                for key, value in (event.attributes or {}).items():
+                    serialized = json.dumps(value) if not isinstance(value, str) else value
+                    assert sentinel not in str(key), f"Sentinel in event attribute key: {key}"
+                    assert sentinel not in serialized, f"Sentinel in event attribute value for key={key}: {serialized}"
+
+            if hasattr(span, "resource") and span.resource:
+                for key, value in (span.resource.attributes or {}).items():
+                    serialized = json.dumps(value) if not isinstance(value, str) else value
+                    assert sentinel not in str(key), f"Sentinel in resource attribute key: {key}"
+                    assert sentinel not in serialized, (
+                        f"Sentinel in resource attribute value for key={key}: {serialized}"
+                    )
+
+
+@pytest.fixture()
+def otel_exporter():
+    """Set up TracerProvider with InMemorySpanExporter for span capture."""
+    exporter = InMemorySpanExporter()
+    provider = TracerProvider()
+    provider.add_span_processor(SimpleSpanProcessor(exporter))
+
+    original_provider = opentelemetry.trace.get_tracer_provider()
+    opentelemetry.trace.set_tracer_provider(provider)
+    tracing_mod._registered_on = None
+
+    yield exporter
+
+    provider.shutdown()
+    opentelemetry.trace.set_tracer_provider(original_provider)
+
+
+class TestBedrockAgentCoreAppNoUserContent:
+    """Verify BedrockAgentCoreApp does not leak user content into OTEL spans."""
+
+    def test_sync_invocation_no_user_content_in_spans(self, otel_exporter):
+        app = BedrockAgentCoreApp()
+
+        @app.entrypoint
+        def handler(payload):
+            return {"response": SENTINEL_RESPONSE, "echo": payload.get("prompt")}
+
+        client = TestClient(app)
+        response = client.post(
+            "/invocations",
+            json={"prompt": SENTINEL_PROMPT},
+            headers={"X-Amzn-Bedrock-AgentCore-Runtime-Session-Id": "test-session"},
+        )
+
+        assert response.status_code == 200
+        spans = otel_exporter.get_finished_spans()
+        _assert_no_sentinel_in_spans(spans, [SENTINEL_PROMPT, SENTINEL_RESPONSE])
+
+    def test_streaming_invocation_no_user_content_in_spans(self, otel_exporter):
+        app = BedrockAgentCoreApp()
+
+        @app.entrypoint
+        async def handler(payload):
+            yield f"chunk1: {SENTINEL_RESPONSE}"
+            yield f"chunk2: {payload.get('prompt')}"
+
+        client = TestClient(app)
+        response = client.post(
+            "/invocations",
+            json={"prompt": SENTINEL_PROMPT},
+            headers={
+                "X-Amzn-Bedrock-AgentCore-Runtime-Session-Id": "test-session",
+                "Accept": "text/event-stream",
+            },
+        )
+
+        assert response.status_code == 200
+        spans = otel_exporter.get_finished_spans()
+        _assert_no_sentinel_in_spans(spans, [SENTINEL_PROMPT, SENTINEL_RESPONSE])