diff --git a/python/packages/ag-ui/agent_framework_ag_ui/_event_converters.py b/python/packages/ag-ui/agent_framework_ag_ui/_event_converters.py
index 122e7d6ffbe..5a2c732006d 100644
--- a/python/packages/ag-ui/agent_framework_ag_ui/_event_converters.py
+++ b/python/packages/ag-ui/agent_framework_ag_ui/_event_converters.py
@@ -246,7 +246,6 @@ def _handle_run_error(self, event: dict[str, Any]) -> ChatResponseUpdate:
 
         return ChatResponseUpdate(
             role="assistant",
-            finish_reason="content_filter",
             contents=[
                 Content.from_error(
                     message=error_message,
diff --git a/python/packages/ag-ui/tests/ag_ui/test_event_converters.py b/python/packages/ag-ui/tests/ag_ui/test_event_converters.py
index fae715d97ef..625c83da95e 100644
--- a/python/packages/ag-ui/tests/ag_ui/test_event_converters.py
+++ b/python/packages/ag-ui/tests/ag_ui/test_event_converters.py
@@ -328,7 +328,7 @@ def test_run_error_event(self) -> None:
 
         assert update is not None
         assert update.role == "assistant"
-        assert update.finish_reason == "content_filter"
+        assert update.finish_reason is None
         assert len(update.contents) == 1
         assert update.contents[0].message == "Connection timeout"
         assert update.contents[0].error_code == "RUN_ERROR"
diff --git a/python/packages/bedrock/agent_framework_bedrock/_chat_client.py b/python/packages/bedrock/agent_framework_bedrock/_chat_client.py
index 0292b07e40f..cbe30683e28 100644
--- a/python/packages/bedrock/agent_framework_bedrock/_chat_client.py
+++ b/python/packages/bedrock/agent_framework_bedrock/_chat_client.py
@@ -207,6 +207,7 @@ class BedrockChatOptions(ChatOptions[ResponseModelT], Generic[ResponseModelT], t
     "max_tokens": "length",
     "length": "length",
     "content_filtered": "content_filter",
+    "guardrail_intervened": "content_filter",
     "tool_use": "tool_calls",
 }
 
@@ -760,10 +761,10 @@ def _parse_message_contents(self, content_blocks: Sequence[dict[str, Any]]) -> l
             logger.debug("Ignoring unsupported Bedrock content block: %s", block)
         return contents
 
-    def _map_finish_reason(self, reason: str | None) -> FinishReasonLiteral | None:
+    def _map_finish_reason(self, reason: str | None) -> str | None:
         if not reason:
             return None
-        return FINISH_REASON_MAP.get(reason.lower())
+        return FINISH_REASON_MAP.get(reason.lower(), reason)
 
     def _prepare_output_config(self, response_format: Any | None) -> dict[str, Any] | None:
         """Convert response_format into the AWS Bedrock outputConfig wire format.
diff --git a/python/packages/bedrock/tests/test_bedrock_settings.py b/python/packages/bedrock/tests/test_bedrock_settings.py
index c828c2dc940..a0cc3d1fb70 100644
--- a/python/packages/bedrock/tests/test_bedrock_settings.py
+++ b/python/packages/bedrock/tests/test_bedrock_settings.py
@@ -107,6 +107,12 @@ def test_process_response_parses_tool_use_and_result() -> None:
     assert chat_response.finish_reason == client._map_finish_reason("tool_use")
 
 
+def test_map_finish_reason_maps_guardrail_intervention() -> None:
+    client = _build_client()
+
+    assert client._map_finish_reason("guardrail_intervened") == "content_filter"
+
+
 def test_process_response_parses_tool_result() -> None:
     client = _build_client()
     response = {
diff --git a/python/packages/claude/agent_framework_claude/_agent.py b/python/packages/claude/agent_framework_claude/_agent.py
index c0d2bdb3b22..7db8eb7dbfd 100644
--- a/python/packages/claude/agent_framework_claude/_agent.py
+++ b/python/packages/claude/agent_framework_claude/_agent.py
@@ -68,6 +68,14 @@
 
 logger = logging.getLogger("agent_framework.claude")
 
+FINISH_REASON_MAP: dict[str, str] = {
+    "end_turn": "stop",
+    "stop_sequence": "stop",
+    "max_tokens": "length",
+    "tool_use": "tool_calls",
+    "refusal": "content_filter",
+}
+
 
 # Name of the in-process MCP server that hosts Agent Framework tools.
 # FunctionTool instances are converted to SDK MCP tools and served
@@ -843,7 +851,9 @@ async def _get_stream(
                     }.items()
                     if isinstance(value, int)
                 })
-                finish_reason = message.stop_reason
+                finish_reason = (
+                    FINISH_REASON_MAP.get(message.stop_reason, message.stop_reason) if message.stop_reason else None
+                )
                 if usage_details or finish_reason:
                     yield AgentResponseUpdate(
                         contents=[Content.from_usage(usage_details, raw_representation=message)]
diff --git a/python/packages/claude/tests/test_claude_agent.py b/python/packages/claude/tests/test_claude_agent.py
index 506dbf8611f..116d4114694 100644
--- a/python/packages/claude/tests/test_claude_agent.py
+++ b/python/packages/claude/tests/test_claude_agent.py
@@ -271,7 +271,13 @@ async def test_run_captures_session_id(self) -> None:
             await agent.run("Hello", session=session)
             assert session.service_session_id == "test-session-id"
 
-    async def test_run_captures_result_message_usage_and_finish_reason(self) -> None:
+    @pytest.mark.parametrize(
+        ("stop_reason", "expected_finish_reason"),
+        [("end_turn", "stop"), ("pause_turn", "pause_turn")],
+    )
+    async def test_run_captures_result_message_usage_and_finish_reason(
+        self, stop_reason: str, expected_finish_reason: str
+    ) -> None:
         """Test that ResultMessage metadata is propagated to the final AgentResponse."""
         from claude_agent_sdk import AssistantMessage, ResultMessage, TextBlock
         from claude_agent_sdk.types import StreamEvent
@@ -296,7 +302,7 @@ async def test_run_captures_result_message_usage_and_finish_reason(self) -> None
                 is_error=False,
                 num_turns=1,
                 session_id="test-session-id",
-                stop_reason="end_turn",
+                stop_reason=stop_reason,
                 usage={
                     "input_tokens": 42,
                     "output_tokens": 18,
@@ -311,7 +317,7 @@ async def test_run_captures_result_message_usage_and_finish_reason(self) -> None
             agent = ClaudeAgent()
             response = await agent.run("Hello")
 
-        assert response.finish_reason == "end_turn"
+        assert response.finish_reason == expected_finish_reason
         assert response.usage_details == {
             "input_token_count": 42,
             "output_token_count": 18,
@@ -465,7 +471,7 @@ async def test_run_stream_final_response_captures_usage_and_finish_reason(self)
                 pass
             response = await stream.get_final_response()
 
-        assert response.finish_reason == "max_tokens"
+        assert response.finish_reason == "length"
         assert response.usage_details == {
             "input_token_count": 7,
             "output_token_count": 9,
diff --git a/python/packages/core/agent_framework/observability.py b/python/packages/core/agent_framework/observability.py
index 192f8c585f7..92935c78e09 100644
--- a/python/packages/core/agent_framework/observability.py
+++ b/python/packages/core/agent_framework/observability.py
@@ -1582,11 +1582,15 @@ async def _finalize_stream() -> None:
                         and response.messages
                         and span.is_recording()
                     ):
+                        finish_reason = cast(
+                            "FinishReason | None",
+                            response.finish_reason if response.finish_reason in FINISH_REASON_MAP else None,
+                        )
                         _capture_messages(
                             span=span,
                             provider_name=provider_name,
                             messages=response.messages,
-                            finish_reason=response.finish_reason,  # type: ignore[arg-type]
+                            finish_reason=finish_reason,
                             output=True,
                         )
                 except Exception as exception:
diff --git a/python/packages/github_copilot/agent_framework_github_copilot/_agent.py b/python/packages/github_copilot/agent_framework_github_copilot/_agent.py
index d8ea5536a41..946fdce04c9 100644
--- a/python/packages/github_copilot/agent_framework_github_copilot/_agent.py
+++ b/python/packages/github_copilot/agent_framework_github_copilot/_agent.py
@@ -635,8 +635,11 @@ def usage_event_handler(event: SessionEvent) -> None:
                 parsed_usage_details = self._parse_usage_details_from_copilot(event.data)
                 if parsed_usage_details:
                     usage_details = add_usage_details(usage_details, parsed_usage_details)
-                if event.data.finish_reason:
-                    finish_reason = event.data.finish_reason
+                event_finish_reason = (
+                    "content_filter" if event.data.content_filter_triggered else event.data.finish_reason
+                )
+                if event_finish_reason:
+                    finish_reason = event_finish_reason
                 if event.data.model:
                     model = event.data.model
             else:
@@ -780,7 +783,7 @@ def event_handler(event: SessionEvent) -> None:
                     )
                     return
                 usage_details = self._parse_usage_details_from_copilot(event.data)
-                finish_reason = event.data.finish_reason or None
+                finish_reason = "content_filter" if event.data.content_filter_triggered else event.data.finish_reason
                 model = event.data.model or None
                 if usage_details or finish_reason or model:
                     update = AgentResponseUpdate(
diff --git a/python/packages/github_copilot/tests/test_github_copilot_agent.py b/python/packages/github_copilot/tests/test_github_copilot_agent.py
index 1ad6900b353..1f274101df0 100644
--- a/python/packages/github_copilot/tests/test_github_copilot_agent.py
+++ b/python/packages/github_copilot/tests/test_github_copilot_agent.py
@@ -488,11 +488,18 @@ async def test_run_captures_assistant_usage_event(
                 input_tokens=5,
                 output_tokens=2,
                 finish_reason="length",
+                content_filter_triggered=True,
             ),
             id=uuid4(),
             timestamp=datetime.now(timezone.utc),
             type=SessionEventType.ASSISTANT_USAGE,
         )
+        empty_usage_event = SessionEvent(
+            data=AssistantUsageData(model="gpt-5.1-mini"),
+            id=uuid4(),
+            timestamp=datetime.now(timezone.utc),
+            type=SessionEventType.ASSISTANT_USAGE,
+        )
         usage_handler: Any = None
 
         def mock_on(handler: Any) -> Any:
@@ -503,6 +510,7 @@ def mock_on(handler: Any) -> Any:
         async def mock_send_and_wait(*args: Any, **kwargs: Any) -> SessionEvent:
             usage_handler(usage_event)
             usage_handler(second_usage_event)
+            usage_handler(empty_usage_event)
             return assistant_message_event
 
         mock_session.on = mock_on
@@ -511,7 +519,7 @@ async def mock_send_and_wait(*args: Any, **kwargs: Any) -> SessionEvent:
         agent = GitHubCopilotAgent(client=mock_client)
         response = await agent.run("Hello")
 
-        assert response.finish_reason == "length"
+        assert response.finish_reason == "content_filter"
         assert response.usage_details == {
             "input_token_count": 125,
             "output_token_count": 42,
@@ -596,7 +604,7 @@ async def test_run_streaming_captures_assistant_usage_event(
             model="gpt-5.1-mini",
             input_tokens=10,
             output_tokens=4,
-            finish_reason="stop",
+            finish_reason="provider_specific_reason",
         )
         usage_event = SessionEvent(
             data=usage_data,
@@ -620,7 +628,7 @@ def mock_on(handler: Any) -> Any:
         response = await stream.get_final_response()
 
         assert response.text == "Hello"
-        assert response.finish_reason == "stop"
+        assert response.finish_reason == "provider_specific_reason"
         assert response.usage_details == {
             "input_token_count": 10,
             "output_token_count": 4,
diff --git a/python/packages/ollama/agent_framework_ollama/_chat_client.py b/python/packages/ollama/agent_framework_ollama/_chat_client.py
index f80163486e9..94e612f75db 100644
--- a/python/packages/ollama/agent_framework_ollama/_chat_client.py
+++ b/python/packages/ollama/agent_framework_ollama/_chat_client.py
@@ -24,6 +24,7 @@
     ChatResponse,
     ChatResponseUpdate,
     Content,
+    FinishReason,
     FunctionInvocationConfiguration,
     FunctionInvocationLayer,
     FunctionTool,
@@ -542,6 +543,11 @@ def _parse_contents_from_ollama(self, response: OllamaChatResponse) -> list[Cont
             contents.extend(tool_calls)
         return contents
 
+    def _get_finish_reason_from_ollama(self, response: OllamaChatResponse) -> FinishReason | None:
+        if response.message.tool_calls:
+            return FinishReason("tool_calls")
+        return FinishReason(response.done_reason) if response.done_reason else None
+
     def _parse_streaming_response_from_ollama(self, response: OllamaChatResponse) -> ChatResponseUpdate:
         contents = self._parse_contents_from_ollama(response)
         finish_reason = None
@@ -561,7 +567,7 @@ def _parse_streaming_response_from_ollama(self, response: OllamaChatResponse) ->
             )
             if usage_details:
                 contents.append(Content.from_usage(usage_details, raw_representation=response))
-            finish_reason = response.done_reason if response.done_reason in ("stop", "length") else None
+            finish_reason = self._get_finish_reason_from_ollama(response)
         return ChatResponseUpdate(
             contents=contents,
             role="assistant",
@@ -590,7 +596,7 @@ def _parse_response_from_ollama(
                 if isinstance(value, int)
             }
         )
-        finish_reason = response.done_reason if response.done_reason in ("stop", "length") else None
+        finish_reason = self._get_finish_reason_from_ollama(response)
 
         return ChatResponse(
             messages=[Message(role="assistant", contents=contents)],
diff --git a/python/packages/ollama/tests/test_ollama_chat_client.py b/python/packages/ollama/tests/test_ollama_chat_client.py
index d7704af4fe3..70f578cc37a 100644
--- a/python/packages/ollama/tests/test_ollama_chat_client.py
+++ b/python/packages/ollama/tests/test_ollama_chat_client.py
@@ -290,7 +290,7 @@ async def test_cmc_maps_done_reason_to_finish_reason(
 
 
 @patch.object(AsyncClient, "chat", new_callable=AsyncMock)
-async def test_cmc_leaves_unknown_done_reason_unset(
+async def test_cmc_preserves_unknown_done_reason(
     mock_chat: AsyncMock,
     ollama_unit_test_env: dict[str, str],
     chat_history: list[Message],
@@ -305,7 +305,7 @@ async def test_cmc_leaves_unknown_done_reason_unset(
     ollama_client = OllamaChatClient()
     result = await ollama_client.get_response(messages=chat_history)
 
-    assert result.finish_reason is None
+    assert result.finish_reason == "load"
 
 
 @patch.object(AsyncClient, "chat", new_callable=AsyncMock)
diff --git a/python/packages/openai/agent_framework_openai/_chat_client.py b/python/packages/openai/agent_framework_openai/_chat_client.py
index c9f2e6de300..de703509da9 100644
--- a/python/packages/openai/agent_framework_openai/_chat_client.py
+++ b/python/packages/openai/agent_framework_openai/_chat_client.py
@@ -49,6 +49,7 @@
     ChatResponseUpdate,
     Content,
     ContinuationToken,
+    FinishReason,
     Message,
     ResponseStream,
     Role,
@@ -2339,6 +2340,19 @@ def _parse_hosted_function_call_content(
         )
 
     # region Parse methods
+    def _get_finish_reason_from_openai_response(self, response: Any) -> FinishReason | None:
+        """Get the framework finish reason from a terminal Responses API response."""
+        incomplete_reason = getattr(getattr(response, "incomplete_details", None), "reason", None)
+        if incomplete_reason == "content_filter":
+            return FinishReason("content_filter")
+        if incomplete_reason == "max_output_tokens":
+            return FinishReason("length")
+        if getattr(response, "status", None) != "completed":
+            return None
+        if any(getattr(item, "type", None) == "function_call" for item in getattr(response, "output", ())):
+            return FinishReason("tool_calls")
+        return FinishReason("stop")
+
     def _parse_response_from_openai(
         self,
         response: OpenAIResponse | ParsedResponse[BaseModel],
@@ -2638,6 +2652,8 @@ def _parse_response_from_openai(
             args["value"] = structured_response
         elif response_format := options.get("response_format"):
             args["response_format"] = response_format
+        if finish_reason := self._get_finish_reason_from_openai_response(response):
+            args["finish_reason"] = finish_reason
         # Set continuation_token when background operation is still in progress
         if response.status and response.status in ("in_progress", "queued"):
             args["continuation_token"] = OpenAIContinuationToken(response_id=response.id)
@@ -2661,6 +2677,7 @@ def _parse_chunk_from_openai(
         response_id: str | None = None
         created_at: str | None = None
         continuation_token: OpenAIContinuationToken | None = None
+        finish_reason: FinishReason | None = None
         model = self.model
         match event.type:
             # types:
@@ -2837,13 +2854,14 @@ def _parse_chunk_from_openai(
                 response_id = event.response.id
                 conversation_id = self._get_conversation_id(event.response, options.get("store"))
                 continuation_token = OpenAIContinuationToken(response_id=event.response.id)
-            case "response.completed":
+            case "response.completed" | "response.incomplete" | "response.failed":
                 response_id = event.response.id
                 conversation_id = self._get_conversation_id(event.response, options.get("store"))
                 model = event.response.model
                 created_at = datetime.fromtimestamp(event.response.created_at, tz=timezone.utc).strftime(
                     "%Y-%m-%dT%H:%M:%S.%fZ"
                 )
+                finish_reason = self._get_finish_reason_from_openai_response(event.response)
                 if event.response.usage:
                     usage = self._parse_usage_from_openai(event.response.usage)
                     if usage:
@@ -3185,6 +3203,7 @@ def _get_ann_value(key: str) -> Any:
             model=model,
             created_at=created_at,
             continuation_token=continuation_token,
+            finish_reason=finish_reason,
             additional_properties=metadata,
             raw_representation=event,
         )
diff --git a/python/packages/openai/agent_framework_openai/_chat_completion_client.py b/python/packages/openai/agent_framework_openai/_chat_completion_client.py
index 37c2e5b52dd..57f74bfdc71 100644
--- a/python/packages/openai/agent_framework_openai/_chat_completion_client.py
+++ b/python/packages/openai/agent_framework_openai/_chat_completion_client.py
@@ -707,7 +707,7 @@ def _parse_response_from_openai(self, response: ChatCompletion, options: Mapping
         for choice in response.choices:
             response_metadata.update(self._get_metadata_from_chat_choice(choice))
             if choice.finish_reason:
-                finish_reason = choice.finish_reason  # type: ignore[assignment]
+                finish_reason = "tool_calls" if choice.finish_reason == "function_call" else choice.finish_reason  # type: ignore[assignment]
             contents: list[Content] = []
             if text_content := self._parse_text_from_openai(choice):
                 contents.append(text_content)
@@ -746,7 +746,7 @@ def _parse_response_update_from_openai(
         for choice in chunk.choices:
             chunk_metadata.update(self._get_metadata_from_chat_choice(choice))
             if choice.finish_reason:
-                finish_reason = choice.finish_reason  # type: ignore[assignment]
+                finish_reason = "tool_calls" if choice.finish_reason == "function_call" else choice.finish_reason  # type: ignore[assignment]
 
             # Some OpenAI-compatible providers (e.g. Azure) send `"delta": null`
             # on finish chunks instead of the spec-compliant `"delta": {}`.
diff --git a/python/packages/openai/tests/openai/test_openai_chat_client.py b/python/packages/openai/tests/openai/test_openai_chat_client.py
index 7849f4b67dc..8cc7fb737ab 100644
--- a/python/packages/openai/tests/openai/test_openai_chat_client.py
+++ b/python/packages/openai/tests/openai/test_openai_chat_client.py
@@ -6063,6 +6063,91 @@ def test_streaming_response_completed_sets_created_at() -> None:
     assert update.created_at == "2001-09-09T01:46:40.000000Z"
 
 
+@pytest.mark.parametrize(
+    ("status", "incomplete_reason", "output_type", "expected_finish_reason"),
+    [
+        ("completed", None, None, "stop"),
+        ("completed", None, "function_call", "tool_calls"),
+        ("incomplete", "max_output_tokens", None, "length"),
+        ("incomplete", "content_filter", None, "content_filter"),
+        ("failed", None, None, None),
+        ("incomplete", "other", None, None),
+    ],
+)
+def test_get_finish_reason_from_openai_response(
+    status: str,
+    incomplete_reason: str | None,
+    output_type: str | None,
+    expected_finish_reason: str | None,
+) -> None:
+    """Test mapping Responses API terminal states to framework finish reasons."""
+    client = OpenAIChatClient(model="test-model", api_key="test-key")
+    mock_response = MagicMock()
+    mock_response.status = status
+    mock_response.incomplete_details = MagicMock(reason=incomplete_reason) if incomplete_reason is not None else None
+    mock_response.output = [MagicMock(type=output_type)] if output_type is not None else []
+
+    finish_reason = client._get_finish_reason_from_openai_response(mock_response)
+
+    assert finish_reason == expected_finish_reason
+
+
+def test_parse_response_from_openai_sets_finish_reason() -> None:
+    """Test that non-streaming Responses API completions include a finish reason."""
+    client = OpenAIChatClient(model="test-model", api_key="test-key")
+    mock_response = MagicMock()
+    mock_response.output_parsed = None
+    mock_response.metadata = {}
+    mock_response.output = []
+    mock_response.id = "resp_done"
+    mock_response.model = "test-model"
+    mock_response.created_at = 1000000000
+    mock_response.usage = None
+    mock_response.status = "completed"
+    mock_response.incomplete_details = None
+
+    response = client._parse_response_from_openai(mock_response, options={})  # type: ignore[arg-type]
+
+    assert response.finish_reason == "stop"
+
+
+@pytest.mark.parametrize(
+    ("event_type", "status", "incomplete_reason", "output_type", "expected_finish_reason"),
+    [
+        ("response.completed", "completed", None, None, "stop"),
+        ("response.completed", "completed", None, "function_call", "tool_calls"),
+        ("response.incomplete", "incomplete", "max_output_tokens", None, "length"),
+        ("response.incomplete", "incomplete", "content_filter", None, "content_filter"),
+        ("response.failed", "failed", None, None, None),
+    ],
+)
+def test_streaming_terminal_response_sets_finish_reason(
+    event_type: str,
+    status: str,
+    incomplete_reason: str | None,
+    output_type: str | None,
+    expected_finish_reason: str | None,
+) -> None:
+    """Test that terminal Responses API events include the mapped finish reason."""
+    client = OpenAIChatClient(model="test-model", api_key="test-key")
+    mock_event = MagicMock()
+    mock_event.type = event_type
+    mock_event.response.id = "resp_done"
+    mock_event.response.conversation = None
+    mock_event.response.model = "test-model"
+    mock_event.response.created_at = 1000000000
+    mock_event.response.usage = None
+    mock_event.response.status = status
+    mock_event.response.incomplete_details = (
+        MagicMock(reason=incomplete_reason) if incomplete_reason is not None else None
+    )
+    mock_event.response.output = [MagicMock(type=output_type)] if output_type is not None else []
+
+    update = client._parse_chunk_from_openai(mock_event, options={}, function_call_ids={})
+
+    assert update.finish_reason == expected_finish_reason
+
+
 def test_map_chat_to_agent_update_preserves_continuation_token() -> None:
     """Test that map_chat_to_agent_update propagates continuation_token."""
     from agent_framework._types import map_chat_to_agent_update