feat: add enduser.id OTel span attribute from request identity

src/bedrock_agentcore/memory/integrations/strands/session_manager.py

@@ -799,14 +799,22 @@ def list_messages(
             return []
 
     def _filter_restored_tool_context(self, messages: list[SessionMessage]) -> list[SessionMessage]:
-        """Strip historical toolUse/toolResult context from restored messages."""
+        """Strip historical toolUse/toolResult context from restored messages.
+
+        Extended-thinking (reasoningContent) blocks are coupled to the tool
+        calls that follow them.  Bedrock rejects an assistant message whose
+        reasoningContent blocks have been separated from their companion
+        toolUse blocks, so we must strip both together.
+        """
         filtered_messages: list[SessionMessage] = []
         for session_message in messages:
             message = session_message.to_message()
             filtered_content = [
                 content
                 for content in message.get("content", [])
-                if "toolUse" not in content and "toolResult" not in content
+                if "toolUse" not in content
+                and "toolResult" not in content
+                and "reasoningContent" not in content
             ]
 
             if not filtered_content:

src/bedrock_agentcore/runtime/a2a.py

@@ -21,10 +21,12 @@
     OAUTH2_CALLBACK_URL_HEADER,
     REQUEST_ID_HEADER,
     SESSION_HEADER,
+    USER_ID_HEADER,
     PingStatus,
     is_forwardable_header,
 )
 from .tracing import _ensure_baggage_processor_registered
+from .utils import extract_sub_from_bearer
 
 logger = logging.getLogger(__name__)
 
@@ -179,6 +181,11 @@ def build(self, request: Any) -> Any:
         session_id = headers.get(SESSION_HEADER)
         BedrockAgentCoreContext.set_request_context(request_id, session_id)
 
+        enduser_id = headers.get(USER_ID_HEADER) or extract_sub_from_bearer(
+            headers.get(AUTHORIZATION_HEADER) or headers.get(_AUTHORIZATION_HEADER_LOWER)
+        )
+        BedrockAgentCoreContext.set_enduser_id(enduser_id)
+
         workload_access_token = headers.get(ACCESS_TOKEN_HEADER)
         if workload_access_token:
             BedrockAgentCoreContext.set_workload_access_token(workload_access_token)

src/bedrock_agentcore/runtime/ag_ui.py

@@ -31,10 +31,12 @@
     OAUTH2_CALLBACK_URL_HEADER,
     REQUEST_ID_HEADER,
     SESSION_HEADER,
+    USER_ID_HEADER,
     PingStatus,
     is_forwardable_header,
 )
 from .tracing import _ensure_baggage_processor_registered
+from .utils import extract_sub_from_bearer
 
 logger = logging.getLogger(__name__)
 
@@ -163,6 +165,11 @@ def _build_request_context(self, request: Request | WebSocket) -> RequestContext
         session_id = headers.get(SESSION_HEADER)
         BedrockAgentCoreContext.set_request_context(request_id, session_id)
 
+        enduser_id = headers.get(USER_ID_HEADER) or extract_sub_from_bearer(
+            headers.get(AUTHORIZATION_HEADER) or headers.get(_AUTHORIZATION_HEADER_LOWER)
+        )
+        BedrockAgentCoreContext.set_enduser_id(enduser_id)
+
         workload_access_token = headers.get(ACCESS_TOKEN_HEADER)
         if workload_access_token:
             BedrockAgentCoreContext.set_workload_access_token(workload_access_token)

src/bedrock_agentcore/runtime/app.py

@@ -46,11 +46,12 @@
     TASK_ACTION_FORCE_HEALTHY,
     TASK_ACTION_JOB_STATUS,
     TASK_ACTION_PING_STATUS,
+    USER_ID_HEADER,
     PingStatus,
     is_forwardable_header,
 )
 from .tracing import _ensure_baggage_processor_registered
-from .utils import convert_complex_objects
+from .utils import convert_complex_objects, extract_sub_from_bearer
 
 # Sentinel so we only parse OTEL_RESOURCE_ATTRIBUTES once per process.
 _UNRESOLVED = object()
@@ -413,6 +414,11 @@ def _build_request_context(self, request) -> RequestContext:
             session_id = headers.get(SESSION_HEADER)
             BedrockAgentCoreContext.set_request_context(request_id, session_id)
 
+            enduser_id = headers.get(USER_ID_HEADER) or extract_sub_from_bearer(
+                headers.get(AUTHORIZATION_HEADER) or headers.get(_AUTHORIZATION_HEADER_LOWER)
+            )
+            BedrockAgentCoreContext.set_enduser_id(enduser_id)
+
             agent_identity_token = headers.get(IDENTITY_WAT_HEADER) or headers.get(ACCESS_TOKEN_HEADER)
             if agent_identity_token:
                 BedrockAgentCoreContext.set_workload_access_token(agent_identity_token)

src/bedrock_agentcore/runtime/context.py

@@ -34,6 +34,7 @@ class BedrockAgentCoreContext:
     _oauth2_callback_url: ContextVar[Optional[str]] = ContextVar("oauth2_callback_url")
     _request_id: ContextVar[Optional[str]] = ContextVar("request_id")
     _session_id: ContextVar[Optional[str]] = ContextVar("session_id")
+    _enduser_id: ContextVar[Optional[str]] = ContextVar("enduser_id", default=None)
     _request_headers: ContextVar[Optional[Dict[str, str]]] = ContextVar("request_headers")
     _routing_experiment_arn: ContextVar[Optional[str]] = ContextVar("routing_experiment_arn", default=None)
     _routing_experiment_variant: ContextVar[Optional[str]] = ContextVar("routing_experiment_variant", default=None)
@@ -93,6 +94,23 @@ def get_session_id(cls) -> Optional[str]:
         except LookupError:
             return None
 
+    @classmethod
+    def set_enduser_id(cls, enduser_id: Optional[str]) -> None:
+        """Set the end-user identity for the current request.
+
+        The value is stamped onto every OpenTelemetry span as the ``enduser.id``
+        attribute (OTel semantic convention).  It is extracted automatically from
+        the ``X-Amzn-Bedrock-AgentCore-Runtime-User-Id`` request header, or from
+        the ``sub`` claim of a Bearer JWT in the ``Authorization`` header.
+        Applications can also set it explicitly to override the inferred value.
+        """
+        cls._enduser_id.set(enduser_id)
+
+    @classmethod
+    def get_enduser_id(cls) -> Optional[str]:
+        """Return the end-user identity for the current request, or None."""
+        return cls._enduser_id.get()
+
     @classmethod
     def set_request_headers(cls, headers: Dict[str, str]):
         """Set request headers in the context."""

src/bedrock_agentcore/runtime/models.py

@@ -17,6 +17,7 @@ class PingStatus(str, Enum):
 SESSION_HEADER = "X-Amzn-Bedrock-AgentCore-Runtime-Session-Id"
 SHELL_ID_HEADER = "X-Amzn-Bedrock-AgentCore-Shell-Id"
 REQUEST_ID_HEADER = "X-Amzn-Bedrock-AgentCore-Runtime-Request-Id"
+USER_ID_HEADER = "X-Amzn-Bedrock-AgentCore-Runtime-User-Id"
 ACCESS_TOKEN_HEADER = "WorkloadAccessToken"  # nosec
 IDENTITY_WAT_HEADER = "X-Amz-Bedrock-AgentCore-Identity-WAT"  # nosec
 OAUTH2_CALLBACK_URL_HEADER = "OAuth2CallbackUrl"

src/bedrock_agentcore/runtime/tracing.py

@@ -103,7 +103,7 @@ class BaggageSpanProcessor(_get_base_class()):  # type: ignore[misc]
     """
 
     def on_start(self, span: object, parent_context: Optional[object] = None) -> None:
-        """Set routing experiment attributes on every new span.
+        """Set routing experiment and end-user identity attributes on every new span.
 
         Primary source: ContextVars set by ``_build_request_context`` — covers
         all spans created after request parsing (agent spans, tool spans, etc.).
@@ -134,6 +134,10 @@ def on_start(self, span: object, parent_context: Optional[object] = None) -> Non
         if variant is not None:
             span.set_attribute("aws.agentcore.gateway.routing_experiment_variant_name", variant)  # type: ignore[union-attr]
 
+        enduser_id = _context.get_enduser_id()
+        if enduser_id is not None:
+            span.set_attribute("enduser.id", enduser_id)  # type: ignore[union-attr]
+
     def on_end(self, span: object) -> None:
         """No-op."""
 

src/bedrock_agentcore/runtime/utils.py

@@ -1,7 +1,12 @@
 """Bedrock AgentCore runtime utilities for object conversion and serialization."""
 
+import base64
+import json
+import logging
 from dataclasses import asdict, is_dataclass
-from typing import Any
+from typing import Any, Optional
+
+logger = logging.getLogger(__name__)
 
 
 def convert_complex_objects(obj: Any, _depth: int = 0) -> Any:
@@ -38,3 +43,33 @@ def convert_complex_objects(obj: Any, _depth: int = 0) -> Any:
 def is_valid_partition(partition: str) -> bool:
     """Returns if parsed-arn partition is valid."""
     return partition in ("aws", "aws-us-gov")
+
+
+def extract_sub_from_bearer(authorization: Optional[str]) -> Optional[str]:
+    """Return the 'sub' claim from a Bearer JWT without signature validation.
+
+    Intended only for populating the OTel ``enduser.id`` span attribute.
+    The token is NOT validated — its signature, expiry, and issuer are not
+    checked.  Trust decisions must be made by the inbound auth layer before
+    the request reaches agent code.
+
+    Returns ``None`` when the header is absent, malformed, or has no 'sub'.
+    """
+    if not authorization:
+        return None
+    parts = authorization.split(" ", 1)
+    if len(parts) != 2 or parts[0].lower() != "bearer":
+        return None
+    segments = parts[1].strip().split(".")
+    if len(segments) < 2:
+        return None
+    payload = segments[1]
+    # JWT base64url uses no padding; add it back for Python's decoder.
+    padding = (4 - len(payload) % 4) % 4
+    try:
+        decoded = base64.urlsafe_b64decode(payload + "=" * padding)
+        sub = json.loads(decoded).get("sub")
+        return str(sub) if sub is not None else None
+    except Exception:
+        logger.debug("Could not decode JWT payload for enduser.id extraction", exc_info=True)
+        return None

tests/bedrock_agentcore/memory/integrations/strands/test_agentcore_memory_session_manager.py

@@ -3925,4 +3925,104 @@ def test_same_millisecond_different_microseconds_is_a_tie(self, session_manager)
         # through as a false non-tie.
         assert r1 == datetime(2024, 1, 1, 12, 0, 0, 0, tzinfo=timezone.utc)
         assert r2 == datetime(2024, 1, 1, 12, 0, 0, 1000, tzinfo=timezone.utc)
-        assert r2 > r1
+
+
+class TestFilterRestoredToolContext:
+    """Regression tests for _filter_restored_tool_context with extended thinking.
+
+    Regression for https://github.com/aws/bedrock-agentcore-sdk-python/issues/621.
+
+    When extended thinking is enabled, Bedrock rejects assistant messages whose
+    reasoningContent blocks have been stripped of their companion toolUse blocks
+    (or vice-versa). The filter must remove reasoningContent blocks alongside
+    toolUse/toolResult so that no partial assistant message reaches the API.
+    """
+
+    def _make_session_message(self, role: str, content: list) -> SessionMessage:
+        return SessionMessage.from_message({"role": role, "content": content}, 0)
+
+    def test_strips_reasoning_content_alongside_tool_use(self, session_manager):
+        """reasoningContent blocks paired with toolUse must both be removed."""
+        messages = [
+            self._make_session_message(
+                "assistant",
+                [
+                    {"reasoningContent": {"reasoningText": {"text": "I should call weather.", "signature": "sig1"}}},
+                    {"toolUse": {"toolUseId": "tu_1", "name": "get_weather", "input": {"city": "Paris"}}},
+                ],
+            ),
+            self._make_session_message("user", [{"toolResult": {"toolUseId": "tu_1", "content": [{"text": "22C"}]}}]),
+            self._make_session_message("assistant", [{"text": "It is 22°C in Paris."}]),
+        ]
+
+        result = session_manager._filter_restored_tool_context(messages)
+
+        # The tool-use assistant turn and its toolResult are dropped entirely.
+        # The plain-text assistant turn survives.
+        assert len(result) == 1
+        assert result[0].to_message()["content"] == [{"text": "It is 22°C in Paris."}]
+
+    def test_message_with_only_reasoning_and_tool_use_is_dropped_entirely(self, session_manager):
+        """An assistant message whose entire content is reasoningContent + toolUse
+        produces an empty filtered_content and must be excluded from the output."""
+        messages = [
+            self._make_session_message(
+                "assistant",
+                [
+                    {"reasoningContent": {"reasoningText": {"text": "Reasoning.", "signature": "s"}}},
+                    {"toolUse": {"toolUseId": "tu_x", "name": "calc", "input": {}}},
+                ],
+            ),
+        ]
+
+        result = session_manager._filter_restored_tool_context(messages)
+        assert result == []
+
+    def test_text_alongside_reasoning_and_tool_use_is_preserved(self, session_manager):
+        """If the assistant message has text content in addition to reasoningContent
+        and toolUse, the text survives after the other blocks are stripped."""
+        messages = [
+            self._make_session_message(
+                "assistant",
+                [
+                    {"reasoningContent": {"reasoningText": {"text": "Let me look this up.", "signature": "s"}}},
+                    {"text": "Checking now…"},
+                    {"toolUse": {"toolUseId": "tu_2", "name": "search", "input": {"q": "Paris weather"}}},
+                ],
+            ),
+        ]
+
+        result = session_manager._filter_restored_tool_context(messages)
+
+        assert len(result) == 1
+        content = result[0].to_message()["content"]
+        assert content == [{"text": "Checking now…"}]
+
+    def test_messages_without_tool_context_are_unchanged(self, session_manager):
+        """Plain user/assistant messages (no tool or reasoning blocks) pass through."""
+        messages = [
+            self._make_session_message("user", [{"text": "What is the weather?"}]),
+            self._make_session_message("assistant", [{"text": "I don't know."}]),
+        ]
+
+        result = session_manager._filter_restored_tool_context(messages)
+
+        assert len(result) == 2
+        assert result[0].to_message()["content"] == [{"text": "What is the weather?"}]
+        assert result[1].to_message()["content"] == [{"text": "I don't know."}]
+
+    def test_existing_tool_use_filtering_still_works(self, session_manager):
+        """Original toolUse/toolResult filtering behaviour is preserved."""
+        messages = [
+            self._make_session_message(
+                "assistant",
+                [{"toolUse": {"toolUseId": "tu_3", "name": "calc", "input": {"expr": "1+1"}}}],
+            ),
+            self._make_session_message("user", [{"toolResult": {"toolUseId": "tu_3", "content": [{"text": "2"}]}}]),
+            self._make_session_message("assistant", [{"text": "The answer is 2."}]),
+        ]
+
+        result = session_manager._filter_restored_tool_context(messages)
+
+        assert len(result) == 1
+        assert result[0].to_message()["content"] == [{"text": "The answer is 2."}]