Python: fix FIDES session isolation and runtime integration gaps

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

@@ -386,6 +386,14 @@ that manually replay messages own the equivalent rule: do not resend an approval
 - Approval-time `UserInputRequiredException` and `MiddlewareTermination` return immediately without another model
   call.
 
+### Middleware termination and policy refusals
+
+- A normal `MiddlewareTermination` stops the function loop after returning its correlated result.
+- A correlated `function_result` explicitly marked with `blocked_violation=True` is a policy refusal, not a pause:
+  it closes the original call and is sent back to the model so it can explain the refusal or choose another action.
+- Policy approval requests remain terminal until the user responds.
+- Streaming updates, streaming finalization, and non-streaming output follow the same continuation behavior.
+
 ### Approval control content
 
 - `function_approval_request` and `function_approval_response` are control-plane contents, not durable model
@@ -524,6 +532,7 @@ that manually replay messages own the equivalent rule: do not resend an approval
 | Consecutive error cap | Error threshold stops repeated failures, submits collected results, and makes only the required final no-tool model call. | `test_function_invocation_config_max_consecutive_errors`, `test_streaming_function_invocation_config_max_consecutive_errors`, `test_approval_resume_error_limit_forces_final_no_tool_response` |
 | Unknown call handling | Configured false returns an error result; configured true raises. | `test_function_invocation_config_terminate_on_unknown_calls_false`, `test_function_invocation_config_terminate_on_unknown_calls_true`, streaming equivalents |
 | Middleware termination | Normal non-approval loop stops without a second model call. | `test_terminate_loop_single_function_call`, `test_terminate_loop_multiple_function_calls_one_terminates`, `test_terminate_loop_streaming_single_function_call` |
+| Blocked policy result | A correlated result marked `blocked_violation=True` closes the call and continues to a second model turn in streaming and non-streaming modes; the blocked tool does not execute. | `test_blocked_policy_result_continues_function_loop` |
 | Middleware failure (fatal) | `MiddlewareFailure` from function middleware or a tool body propagates to the caller without becoming a tool result; the tool does not execute (pre-invocation) or its result never feeds another model call (post-invocation); the cause chain is preserved; ordinary exceptions still become tool-error results and the loop continues. | `packages/core/tests/core/test_middleware_with_agent.py::TestMiddlewareFailure::test_failure_before_tool_aborts_run`, `test_failure_after_tool_aborts_run_before_next_model_turn`, `test_failure_cause_chain_reaches_caller`, `test_failure_from_tool_escapes_without_middleware`, `test_failure_streaming_reaches_stream_consumer`, `test_ordinary_exception_still_becomes_tool_error` |
 | Middleware failure batch cancellation | A fatal signal fails the whole parallel batch: in-flight sibling tool invocations are cancelled and awaited before the failure propagates. Cancellation is cooperative — an async sibling stops at its next suspension point; a synchronous tool body already executing in a worker thread cannot be interrupted and may complete its side effects, but its result is discarded and never reaches the transcript, the model, or history, and failure propagation is not delayed behind it. | `TestMiddlewareFailure::test_failure_cancels_concurrent_sibling_tool`, `test_failure_with_sync_sibling_discards_late_result` |
 | Middleware failure on a service-managed conversation | The continuation state is already persisted when the batch fails, so before propagating, the loop settles the hosted thread: one error `function_result` per dangling call, sent with `tool_choice="none"` in one extra request; the persisted continuation advances to the settlement response (required for response-ID continuations, a no-op for conversation-object ids) and the settlement response is otherwise discarded; a settlement failure never masks the abort. Without a service-managed conversation no extra request is made. | `TestMiddlewareFailure::test_failure_settles_dangling_calls_on_service_conversation`, `test_failure_settles_service_conversation_streaming`, `test_failure_settlement_advances_response_id_continuation`, `test_failure_without_service_conversation_makes_no_settlement_request` |

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

@@ -295,9 +295,24 @@ def _function_call_from_request(request: Content) -> Content | None:
     function_call = request.function_call
     if function_call is None or function_call.type != "function_call" or function_call.name is None:
         return None
+    request_props = request.additional_properties or {}
+    if request_props:
+        function_call = copy.copy(function_call)
+        function_call.additional_properties = {
+            **(function_call.additional_properties or {}),
+            **request_props,
+        }
     return function_call
 
 
+def _has_policy_violation(request: Content) -> bool:
+    """Return whether an approval request represents a FIDES policy violation."""
+    properties = request.additional_properties or {}
+    return bool(
+        properties.get("policy_violation") or properties.get("blocked_violation") or properties.get("_fides_violations")
+    )
+
+
 def _arguments_match(rule_arguments: Mapping[str, str], function_call: Content) -> bool:
     call_arguments = _serialize_arguments(function_call) or {}
     if len(rule_arguments) != len(call_arguments):
@@ -579,7 +594,9 @@ def _inject_collected_responses(self, messages: Sequence[Message], state: ToolAp
     async def _drain_auto_approvable_queue(self, state: ToolApprovalState) -> None:
         remaining: list[Content] = []
         for request in state.queued_approval_requests:
-            if _matches_rule(request, state.rules) or await self._matches_auto_rule(request):
+            if not _has_policy_violation(request) and (
+                _matches_rule(request, state.rules) or await self._matches_auto_rule(request)
+            ):
                 state.collected_approval_responses.append(request.to_function_approval_response(approved=True))
                 continue
             remaining.append(request)
@@ -603,7 +620,9 @@ async def _process_outbound_messages(self, messages: list[Message], state: ToolA
         auto_approved: set[int] = set()
         unresolved: list[Content] = []
         for request in approval_requests:
-            if _matches_rule(request, state.rules) or await self._matches_auto_rule(request):
+            if not _has_policy_violation(request) and (
+                _matches_rule(request, state.rules) or await self._matches_auto_rule(request)
+            ):
                 state.collected_approval_responses.append(request.to_function_approval_response(approved=True))
                 auto_approved.add(id(request))
             else:

python/packages/core/agent_framework/_middleware.py

@@ -76,10 +76,18 @@ class MiddlewareTermination(MiddlewareException):
     """Control-flow exception to terminate middleware execution early."""
 
     result: Any = None  # Optional result to return when terminating
+    blocked_policy: bool = False  # Whether this termination represents a FIDES policy block
 
-    def __init__(self, message: str = "Middleware terminated execution.", *, result: Any = None) -> None:
+    def __init__(
+        self,
+        message: str = "Middleware terminated execution.",
+        *,
+        result: Any = None,
+        blocked_policy: bool = False,
+    ) -> None:
         super().__init__(message, log_level=None)
         self.result = result
+        self.blocked_policy = blocked_policy
 
 
 class MiddlewareFailure(MiddlewareException):

python/packages/core/agent_framework/_serialization.py

@@ -10,6 +10,7 @@
 from collections.abc import Mapping, MutableMapping
 from dataclasses import asdict, is_dataclass
 from datetime import date, datetime
+from enum import Enum
 from functools import lru_cache
 from typing import Any, ClassVar, Protocol, TypeGuard, TypeVar, cast, runtime_checkable
 
@@ -650,8 +651,8 @@ def make_json_safe(obj: Any) -> Any:
     """Recursively convert an object to a JSON-serializable form.
 
     Handles dataclasses, Pydantic models, objects with ``to_dict``/``dict``/``__dict__``,
-    datetimes, bytes (base64), lists, dicts, and primitives.  Falls back to ``str()`` for
-    any remaining non-serializable value so that ``json.dumps`` never raises a
+    datetimes, bytes/bytearray (base64), enums, sets, lists, dicts, and primitives. Falls back to
+    ``str()`` for any remaining non-serializable value so that ``json.dumps`` never raises a
     ``TypeError``.
 
     Args:
@@ -662,6 +663,8 @@ def make_json_safe(obj: Any) -> Any:
     """
     if isinstance(obj, _JSON_SCALAR_TYPES):
         return obj
+    if isinstance(obj, Enum):
+        return make_json_safe(obj.value)
     if isinstance(obj, (datetime, date)):
         return obj.isoformat()
     if isinstance(obj, (bytes, bytearray)):
@@ -691,6 +694,8 @@ def make_json_safe(obj: Any) -> Any:
         return {str(key): make_json_safe(value) for key, value in obj.items()}  # type: ignore[misc]
     if isinstance(obj, (list, tuple)):
         return [make_json_safe(item) for item in obj]  # type: ignore[misc]
+    if isinstance(obj, (set, frozenset)):
+        return [make_json_safe(item) for item in obj]  # type: ignore[misc]
     if hasattr(obj, "__dict__"):
         return {key: make_json_safe(value) for key, value in vars(obj).items()}
     return str(obj)

python/packages/core/agent_framework/_tools.py

@@ -1616,19 +1616,28 @@ async def final_function_handler(context_obj: Any) -> Any:
         return Content.from_function_result(call_id=call_id, result=function_result)
     except MiddlewareTermination as term_exc:
         # Re-raise to signal loop termination, but first capture any result set by middleware
-        if middleware_context.result is not None:
+        middleware_result = middleware_context.result
+        if middleware_result is not None:
+            blocked_result = cast(dict[str, Any], middleware_result) if isinstance(middleware_result, dict) else None
+            if blocked_result is not None and blocked_result.get("blocked_violation") is True:
+                blocked_properties = dict(function_call_content.additional_properties or {})
+                blocked_properties.update({key: value for key, value in blocked_result.items() if key != "error"})
+                blocked_error = str(blocked_result.get("error", "Tool blocked by security policy."))
+                term_exc.result = Content.from_function_result(
+                    call_id=call_id,
+                    result=blocked_error,
+                    exception=blocked_error,
+                    additional_properties=blocked_properties,
+                )
             # Pass through function_approval_request directly (e.g., from security policy middleware)
             # so the approval flow in _handle_function_call_results activates correctly.
-            if (
-                isinstance(middleware_context.result, Content)
-                and middleware_context.result.type == "function_approval_request"
-            ):
-                term_exc.result = middleware_context.result
+            elif isinstance(middleware_result, Content) and middleware_result.type == "function_approval_request":
+                term_exc.result = middleware_result
             else:
                 # Store result in exception for caller to extract
                 term_exc.result = Content.from_function_result(
                     call_id=call_id,
-                    result=middleware_context.result,
+                    result=middleware_result,
                     additional_properties=function_call_content.additional_properties,
                 )
         raise
@@ -1695,7 +1704,11 @@ async def _execute_single_function_call(
         return [result], False
     except MiddlewareTermination as exc:
         if isinstance(exc.result, Content):
-            return [exc.result], True
+            # A blocked FIDES call is a normal tool result: the model must receive
+            # the refusal and get a chance to explain or choose another action.
+            # Approval requests remain terminal and pause for user input.
+            is_blocked_policy = exc.blocked_policy
+            return [exc.result], not is_blocked_policy
         source_function_call = _underlying_function_call(function_call)
         return [
             Content.from_function_result(