Python: Replay Anthropic redacted_thinking blocks in tool loops

docs/specs/004-python-function-calling-loop.md

@@ -616,7 +616,7 @@ that manually replay messages own the equivalent rule: do not resend an approval
 | Declaration-only mixed batch | Known local siblings are never converted into Host pauses. With an authoritative session they remain deferred until every Host pause occurrence is answered, survive session serialization with a process-portable logical-run budget (including a reclassified approval continuation), reclassify against the current tool policy, execute exactly once using the configured concurrent or sequential invocation mode, replay results in original model order, and remain inert on replay; without one they execute on the same turn under the same configured invocation mode. Streaming updates and finalization preserve one Host pause and one result per local sibling. | `test_mixed_declaration_only_batch_defers_siblings_until_all_host_results`, `test_mixed_declaration_only_deferred_call_keeps_ordinary_middleware_semantics`, `test_mixed_declaration_only_deferred_call_reclassifies_changed_approval_policy`, `test_mixed_declaration_only_budget_survives_reclassification_to_approval`, `test_mixed_declaration_only_deferred_call_does_not_execute_removed_tool`, `test_mixed_declaration_only_budget_is_process_portable_without_approval_middleware`, `test_sessionless_mixed_declaration_only_batch_executes_siblings_sequentially`, `test_sessionless_mixed_declaration_only_batch_executes_siblings_concurrently`, `test_streaming_sessionless_mixed_declaration_only_batch_returns_pause_and_result` |
 | Function invocation disabled | The client bypasses the invocation loop without losing invocation kwargs. | `test_function_invocation_config_enabled_false`, `test_function_invocation_config_enabled_false_preserves_invocation_kwargs`, `test_streaming_function_invocation_config_enabled_false` |
 | Runtime tool changes | Added tools become available on the next iteration and retain approval behavior. | `test_add_tools_available_next_iteration`, `test_add_tools_with_approval_required_tool` |
-| Anthropic pause-turn continuation | `pause_turn` replays the returned assistant content with the same tools until Anthropic completes the server-side turn in streaming and non-streaming modes when every native content block can be reconstructed faithfully. Intermediate messages and usage remain caller-visible and retain provider message boundaries. Unsupported parsed content, native blocks dropped during parsing (including redacted reasoning), streamed MCP calls whose argument deltas are not retained, and bounded-continuation exhaustion keep `finish_reason="pause_turn"` instead of sending a lossy transcript or reporting normal completion. | `packages/anthropic/tests/test_anthropic_client.py::test_agent_continues_anthropic_pause_turn`, `test_streaming_hosted_tool_pause_turn_continues`, `test_streaming_pause_turn_preserves_message_boundaries_for_structured_output`, `test_non_streaming_code_execution_pause_turn_is_not_auto_continued`, `test_streaming_code_execution_pause_turn_is_not_auto_continued`, `test_streaming_mcp_pause_turn_is_not_auto_continued`, `test_non_streaming_redacted_thinking_pause_turn_is_not_auto_continued`, `test_streaming_redacted_thinking_pause_turn_is_not_auto_continued`, `test_anthropic_pause_turn_limit_preserves_finish_reason` |
+| Anthropic pause-turn continuation | `pause_turn` replays the returned assistant content with the same tools until Anthropic completes the server-side turn in streaming and non-streaming modes when every native content block can be reconstructed faithfully. Intermediate messages and usage remain caller-visible and retain provider message boundaries. Unsupported parsed content, native blocks dropped during parsing, streamed MCP calls whose argument deltas are not retained, and bounded-continuation exhaustion keep `finish_reason="pause_turn"` instead of sending a lossy transcript or reporting normal completion. | `packages/anthropic/tests/test_anthropic_client.py::test_agent_continues_anthropic_pause_turn`, `test_streaming_hosted_tool_pause_turn_continues`, `test_streaming_pause_turn_preserves_message_boundaries_for_structured_output`, `test_non_streaming_code_execution_pause_turn_is_not_auto_continued`, `test_streaming_code_execution_pause_turn_is_not_auto_continued`, `test_streaming_mcp_pause_turn_is_not_auto_continued`, `test_non_streaming_redacted_thinking_pause_turn_continues`, `test_streaming_redacted_thinking_pause_turn_continues`, `test_anthropic_pause_turn_limit_preserves_finish_reason` |
 | Direct delegated session propagation | A custom loop must explicitly identify parent provider-owned state before sharing a non-empty session; missing ownership fails before the child runs, while declared parent keys remain isolated and application state still propagates. | `packages/core/tests/core/test_agents.py::test_chat_agent_as_tool_direct_propagation_requires_parent_provider_ownership`, `test_chat_agent_as_tool_direct_propagation_uses_explicit_parent_provider_ownership` |
 | In-run compaction summaries | Summaries inserted for tool groups excluded by in-run compaction are returned in the final non-streaming response transcript before the group they replace, so history loaded with `skip_excluded` keeps the summarized content; summaries of caller-owned input messages are not added. | `packages/core/tests/core/test_clients.py::test_function_loop_returns_compaction_summaries_in_final_response`, `test_function_loop_returns_compaction_summaries_when_iteration_budget_exhausted`, `test_function_loop_reconciles_nested_compaction_summaries`, `test_function_loop_returns_compacted_transcript_on_early_terminal_exit`, `packages/core/tests/core/test_agents.py::test_agent_run_returns_and_persists_compaction_summaries` |
 | Per-service-call provider context | When framework-managed history is loaded around each model call, tools and instructions contributed by `HistoryProvider.before_run` remain available to the model and function loop without duplication in streaming or non-streaming execution. | `packages/core/tests/core/test_agents.py::test_vector_history_search_tool_is_available_with_per_service_call_persistence` |

docs/specs/005-python-content-identity-and-protocol-metadata.md

@@ -179,6 +179,7 @@ Example:
 | OpenAI | OAuth consent request server | `server_label` | `openai["server_label"]` | Read indefinitely; dual-write through the writer package's next major version. |
 | Foundry | reasoning content | `__foundry_reasoning_replay_item__` | `foundry["reasoning_item"]` | Read indefinitely; dual-write through the Foundry package's next major version. |
 | Bedrock | redacted reasoning marker | `bedrock.redacted_reasoning` | `bedrock["redacted_reasoning"]` | Read indefinitely; dual-write through the Bedrock package's next major version. |
+| Anthropic | native reasoning block type | N/A | `anthropic["reasoning_block"]` | New metadata; values are the Anthropic block types `thinking` and `redacted_thinking`, and a redacted payload uses `Content.protected_data`. |
 | Mistral | response service tier | `service_tier` | `mistral["service_tier"]` | Read indefinitely; dual-write through the Mistral package's next major version. |
 | A2A | response/update/message metadata | `a2a_metadata` | `a2a` | Read indefinitely with a warning for legacy-only input; dual-write through the A2A package's next major version. |
 | all | data/URI/hosted-file display name | `filename` | `Content.name` | Read indefinitely where accepted; dual-write where older adapters require it. |
@@ -242,7 +243,7 @@ merge_additional_properties(existing, incoming, *, incoming_wins=True)
 | AG-UI stream merge modes, hosted/local ownership, and approval replay | `python/packages/ag-ui/tests/ag_ui/test_event_converters.py`, `test_run.py`, `test_workflow_run.py`, and `test_endpoint.py` |
 | A2A stream merge modes and namespaced metadata round trip | `python/packages/a2a/tests/test_a2a_agent.py` |
 | Hosting A2A message metadata conversion | `python/packages/hosting-a2a/tests/hosting_a2a/test_conversion.py` |
-| Anthropic streamed merge modes and provider-hosted tool routing | `python/packages/anthropic/tests/test_anthropic_client.py` |
+| Anthropic streamed merge modes, provider-hosted tool routing, and redacted-thinking replay metadata | `python/packages/anthropic/tests/test_anthropic_client.py` |
 | Bedrock streamed merge modes and redacted-reasoning replay metadata | `python/packages/bedrock/tests/test_bedrock_client.py` |
 | Gemini streamed merge modes and persisted provider-hosted tool replay | `python/packages/gemini/tests/test_gemini_client.py` |
 | Mistral streamed merge modes and response metadata | `python/packages/mistral/tests/mistral/test_mistral_chat_client.py` |

python/packages/anthropic/agent_framework_anthropic/_chat_client.py

@@ -9,6 +9,7 @@
 from collections.abc import AsyncIterable, Awaitable, Callable, Mapping, Sequence
 from dataclasses import dataclass, field
 from typing import Any, ClassVar, Final, Generic, Literal, TypeAlias, TypedDict, cast
+from uuid import uuid4
 
 from agent_framework import (
     Annotation,
@@ -106,6 +107,12 @@
     "web_search_tool_result",
 })
 _MAX_PAUSE_TURN_CONTINUATIONS: Final[int] = 10
+# Marks text_reasoning content parsed from a native Anthropic reasoning block; the value is the
+# Anthropic block type. Distinguishes native reasoning items from provider reasoning summaries
+# that carry a Content.id without a protected payload.
+_REASONING_BLOCK_KEY: Final[str] = "reasoning_block"
+_REASONING_BLOCK_THINKING: Final[str] = "thinking"
+_REASONING_BLOCK_REDACTED_THINKING: Final[str] = "redacted_thinking"
 _PAUSE_TURN_REPLAYABLE_CONTENT_TYPES: Final[frozenset[str]] = frozenset({
     "text",
     "text_reasoning",
@@ -835,6 +842,8 @@ def _append_pause_turn_messages_for_anthropic(
 
     def _is_pause_turn_content_replayable(self, content: Content) -> bool:
         """Return whether the Anthropic serializer preserves this paused content item."""
+        if content.type == "text_reasoning" and self._is_redacted_thinking_content(content):
+            return bool(content.protected_data)
         if content.type in _PAUSE_TURN_REPLAYABLE_CONTENT_TYPES:
             return True
         if content.type == "function_call":
@@ -1130,6 +1139,36 @@ def _message_has_tool_use(self, message: dict[str, Any]) -> bool:
 
         return False
 
+    @staticmethod
+    def _generate_reasoning_id() -> str:
+        """Return a fresh identity for one reasoning item.
+
+        Anthropic reasoning blocks have no provider identity of their own, and response
+        aggregation folds adjacent reasoning chunks without ids into one item. Giving each
+        parsed reasoning block its own id keeps consecutive ``thinking`` and
+        ``redacted_thinking`` blocks distinct when a streamed response is finalized.
+        """
+        return f"reasoning-{uuid4().hex}"
+
+    @staticmethod
+    def _native_reasoning_block_type(content: Content) -> Any:
+        """Return the Anthropic native block type recorded on a reasoning item, if any."""
+        return get_namespaced_value(
+            content.additional_properties,
+            _ANTHROPIC_ADDITIONAL_PROPS_NAMESPACE,
+            _REASONING_BLOCK_KEY,
+        )
+
+    @classmethod
+    def _is_redacted_thinking_content(cls, content: Content) -> bool:
+        """Return whether this reasoning item carries an Anthropic redacted-thinking payload."""
+        return cls._native_reasoning_block_type(content) == _REASONING_BLOCK_REDACTED_THINKING
+
+    @classmethod
+    def _is_thinking_content(cls, content: Content) -> bool:
+        """Return whether this reasoning item was parsed from an Anthropic ``thinking`` block."""
+        return cls._native_reasoning_block_type(content) == _REASONING_BLOCK_THINKING
+
     @staticmethod
     def _server_tool_block(content: Content) -> Mapping[str, Any] | None:
         """Return validated persisted Anthropic server-tool metadata."""
@@ -1363,6 +1402,11 @@ def _prepare_message_for_anthropic(self, message: Message) -> dict[str, Any]:
                         mcp_result["is_error"] = is_error
                     a_content.append(mcp_result)
                 case "text_reasoning":
+                    if self._is_redacted_thinking_content(content):
+                        if not content.protected_data:
+                            raise ContentError("Anthropic redacted thinking content is missing its opaque payload.")
+                        a_content.append({"type": "redacted_thinking", "data": content.protected_data})
+                        continue
                     if content.text is None:
                         if (
                             content.protected_data
@@ -1372,7 +1416,7 @@ def _prepare_message_for_anthropic(self, message: Message) -> dict[str, Any]:
                         ):
                             a_content[-1]["signature"] = content.protected_data
                         continue
-                    if content.id and not content.protected_data:
+                    if content.id and not content.protected_data and not self._is_thinking_content(content):
                         a_content.append({"type": "text", "text": content.text})
                         continue
                     thinking_block: dict[str, Any] = {"type": "thinking", "thinking": content.text}
@@ -2081,11 +2125,37 @@ def _parse_contents_from_anthropic(
                                 raw_representation=content_block,
                             )
                         )
+                case "redacted_thinking":
+                    # Redacted reasoning is opaque provider state. Anthropic requires it to be replayed
+                    # unchanged while tool use continues, so keep it as its own reasoning item.
+                    contents.append(
+                        Content.from_text_reasoning(
+                            id=self._generate_reasoning_id(),
+                            protected_data=content_block.data,
+                            additional_properties={
+                                _ANTHROPIC_ADDITIONAL_PROPS_NAMESPACE: {
+                                    _REASONING_BLOCK_KEY: _REASONING_BLOCK_REDACTED_THINKING
+                                }
+                            },
+                            merge_mode="replace" if streaming else None,
+                            raw_representation=content_block,
+                        )
+                    )
                 case "thinking" | "thinking_delta":
                     contents.append(
                         Content.from_text_reasoning(
+                            id=self._generate_reasoning_id() if content_block.type == "thinking" else None,
                             text=content_block.thinking,
                             protected_data=getattr(content_block, "signature", None),
+                            additional_properties=(
+                                {
+                                    _ANTHROPIC_ADDITIONAL_PROPS_NAMESPACE: {
+                                        _REASONING_BLOCK_KEY: _REASONING_BLOCK_THINKING
+                                    }
+                                }
+                                if content_block.type == "thinking"
+                                else None
+                            ),
                             merge_mode="append" if streaming else None,
                             raw_representation=content_block,
                         )