Python: feat(core): add max_duration_seconds bound and stop_reason signal to tool loop (#7587)

python/CHANGELOG.md

@@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+- **agent-framework-core**: Add `max_duration_seconds` to `FunctionInvocationConfiguration` to bound total wall-clock time of the function-invocation loop; when exceeded, tools are disabled and the model is forced to produce a final text response via the same graceful-degradation path as `max_function_calls`. Rejects `float("nan")` during configuration validation. Add `_agent_framework_stop_reason` to `ChatResponse.additional_properties` with machine-readable enum values (`"completed"`, `"max_iterations"`, `"max_duration_seconds"`, `"max_function_calls"`, `"max_consecutive_errors"`) so callers can detect how a run ended ([#7587](https://github.com/microsoft/agent-framework/issues/7587), [#7772](https://github.com/microsoft/agent-framework/pull/7772)). Includes a small, additive update to `ToolApprovalMiddleware` to persist budget state across human approval round-trips via session state.
+
 ## [1.14.0] - 2026-08-13
 
 ### Added

python/packages/core/agent_framework/_agents.py

@@ -1268,10 +1268,34 @@ def _suppress_response_id(update: AgentResponseUpdate) -> AgentResponseUpdate:
             return update
 
         def _finalizer(updates: Sequence[AgentResponseUpdate]) -> AgentResponse[Any]:
-            return self._finalize_response_updates(
+            agent_response = self._finalize_response_updates(
                 updates,
                 response_format=context["chat_options"].get("response_format"),
             )
+            # Propagate _agent_framework_stop_reason from the inner ChatResponse to the outer
+            # AgentResponse. _finalize_with_stop_reason sets it on the inner ChatResponse
+            # (a ResponseStream finalizer), but AgentResponse.from_updates only sees the
+            # AgentResponseUpdate sequence — none of which carries this key. The non-streaming
+            # path copies additional_properties directly via _build_agent_response_from_chat_response;
+            # for streaming we recover it here from the inner ChatResponse stored in the closure.
+            if inner_chat_responses:
+                inner = inner_chat_responses[0]
+                stop_reason = inner.additional_properties.get("_agent_framework_stop_reason")
+                if (
+                    stop_reason is not None
+                    and "_agent_framework_stop_reason" not in agent_response.additional_properties
+                ):
+                    agent_response.additional_properties["_agent_framework_stop_reason"] = stop_reason
+            return agent_response
+
+        # Mutable container for the inner ChatResponse; populated by the result hook below before
+        # _finalizer is called (inner finalizer runs before outer finalizer per ResponseStream.map contract).
+        inner_chat_responses: list[ChatResponse[Any]] = []
+
+        def capture_inner_chat_response(inner: ChatResponse[Any]) -> None:
+            inner_chat_responses.append(inner)
+
+        stream_response = stream_response.with_result_hook(capture_inner_chat_response)
 
         stream = stream_response.map(
             transform=partial(

python/packages/core/agent_framework/_harness/_tool_approval.py

@@ -385,7 +385,8 @@ async def process(self, context: AgentContext, call_next: Callable[[], Awaitable
             raise RuntimeError("ToolApprovalMiddleware requires an AgentSession.")
 
         state = _get_state(context.session, source_id=self.source_id)
-        context.client_kwargs.setdefault(_FUNCTION_INVOCATION_BUDGET_STATE_KEY, {})
+        session_budget = context.session.state.setdefault(_FUNCTION_INVOCATION_BUDGET_STATE_KEY, {})
+        context.client_kwargs.setdefault(_FUNCTION_INVOCATION_BUDGET_STATE_KEY, session_budget)
         context.messages = self._prepare_inbound_messages(context.messages, state, context.session)
         await self._drain_auto_approvable_queue(state)
         if next_queued := self._pop_next_queued_request(state):