Python: keep MCP labels subordinate to local FIDES policy

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

@@ -412,6 +412,10 @@ that manually replay messages own the equivalent rule: do not resend an approval
 - If policy middleware detects that the exact resolved invocation changed after approval, the old response executes
   nothing and yields a caller-visible, session-persisted replacement request for the same occurrence; execution
   requires a second approval and happens exactly once.
+- If session-bound middleware no longer holds the reviewed authority because it expired or was evicted, the matched
+  response executes nothing and produces a replacement approval request with the same occurrence identity. The
+  replacement is caller-visible, becomes the authoritative pending session snapshot, and requires a second approval
+  before the tool can execute. Rejection or cancellation releases only the matching occurrence in the owning session.
 - Unmatched occurrence-aware responses leave the pending request intact for a corrected retry and produce an
   observable warning/log. A nested `call_id` is never accepted as an occurrence-identity alias.
 - Session-backed pending snapshots are trusted host state and require tenant-scoped, authorized storage. Consume-on-bind
@@ -508,6 +512,8 @@ that manually replay messages own the equivalent rule: do not resend an approval
 | Approval re-entry after iteration budget | Pending approved calls resolve once even when prior model calls consumed `max_iterations`. | `packages/core/tests/core/test_harness_tool_approval.py::test_auto_approval_resolves_after_iteration_budget_is_exhausted` |
 | Approval resume with reasoning | Model-bound resume history retains reasoning before the call and terminal result in both modes. | `packages/core/tests/core/test_harness_tool_approval.py::test_approval_resume_replays_reasoning_with_function_call_group` |
 | Changed resolved policy invocation | The stale decision executes nothing; a same-occurrence replacement request is visible and persisted in both modes, and the second approval executes exactly once. | `packages/core/tests/core/test_harness_tool_approval.py::test_changed_hidden_snapshot_requires_visible_second_approval` |
+| Expired or evicted policy authority | The old response executes nothing, surfaces and persists a same-occurrence replacement request in both modes, and executes exactly once only after the replacement is approved; model history remains balanced and stale replay is inert. | `packages/core/tests/core/test_harness_tool_approval.py::test_policy_reapproval_is_visible_persisted_and_executes_once` |
+| Session-bound policy cleanup | FIFO/TTL lifecycle and authenticated rejection/cancellation cleanup use occurrence identity within only the owning session. | `packages/core/tests/test_security.py::TestPolicyEnforcementMiddleware::test_pending_policy_approvals_are_fifo_bounded_by_occurrence`, `test_pending_policy_approval_ttl_is_deterministic_and_durable`, `test_non_grant_cleanup_is_authenticated_session_and_occurrence_bound` |
 | Session-bound substituted response | A response is rebound to the immutable recorded call and cannot replace its call id, tool name, or arguments. | `packages/core/tests/core/test_function_invocation_logic.py::test_session_approval_binding_rebinds_consumes_and_rejects_duplicates` |
 | Occurrence-aware local binding | New local requests use `function_call.id`; missing, mismatched, or stale occurrence ids do not execute or consume pending state, while the canonical occurrence id binds without an embedded call. | `test_occurrence_aware_approval_rejects_stale_reused_call_id_response`, `test_occurrence_aware_approval_mismatched_identity_does_not_consume_pending`, `test_occurrence_aware_approval_binds_without_embedded_function_call` |
 | Legacy stored approval | A serialized pending request without `function_call.id` retains exact request-id binding once and warns only when resumed. | `test_legacy_serialized_pending_approval_resumes_once_with_migration_warning`, `packages/core/tests/core/test_types.py::test_legacy_function_call_deserialization_does_not_generate_an_occurrence_id` |
@@ -525,6 +531,7 @@ that manually replay messages own the equivalent rule: do not resend an approval
 | Reused id after completion | A later round with the same id creates a second valid pair. | `test_replace_approval_contents_with_results_allows_reused_call_id_after_completion` |
 | Replayed approval wrapper | A duplicated wrapper does not restore another function call. | `test_replace_approval_contents_with_results_deduplicates_replayed_approval_request` |
 | Historical resolved response plus new round | The old response is removed from normalized input and is not converted into a rejection result. | `test_replace_approval_contents_with_results_ignores_already_resolved_response` |
+| Replacement request with reused occurrence id | A request after a stale response starts a new unanswered round rather than inheriting the old decision. | `test_collect_unanswered_approval_requests_tracks_replacement_request` |
 | Multiple reused-id rounds | Approved and rejected rounds retain separate call/result occurrences. | `test_replace_approval_contents_with_results_correlates_reused_call_id_occurrences` |
 | Multi-content result with reused id | Every content produced by one execution stays with that approval occurrence and cannot bleed into the next reused-id round. | `test_replace_approval_contents_with_results_keeps_multi_content_group_with_reused_call_id` |
 | Follow-up request closes one occurrence | A user-input follow-up consumes only the preceding approval authority and leaves a later reused-id response pending. | `test_collect_approval_responses_consumes_matching_follow_up_request_occurrence` |

python/packages/core/agent_framework/_middleware.py

@@ -19,6 +19,7 @@
     AgentRunInputs,
     ChatResponse,
     ChatResponseUpdate,
+    Content,
     Message,
     ResponseStream,
     normalize_messages,
@@ -755,6 +756,25 @@ async def process(
         """
         ...
 
+    def on_approval_responses(
+        self,
+        responses: Sequence[Content],
+        *,
+        session: AgentSession | None,
+    ) -> None:
+        """Observe authenticated approval responses that do not execute a function.
+
+        The function loop calls this only after binding responses to the active session's
+        authoritative pending snapshot. Stateful middleware can discard rejected or
+        cancelled authority here; the default implementation retains no state.
+
+        Args:
+            responses: Session-rebound approval responses.
+
+        Keyword Args:
+            session: The active invocation session, if any.
+        """
+
 
 class ChatMiddleware(ABC):
     """Abstract base class for chat middleware that can intercept chat client requests.
@@ -1218,6 +1238,17 @@ def matches(self, middleware: Sequence[FunctionMiddlewareTypes]) -> bool:
         """Return whether this pipeline was built from the provided middleware sequence."""
         return self._source_middleware == tuple(middleware)
 
+    def notify_approval_responses(
+        self,
+        responses: Sequence[Content],
+        *,
+        session: AgentSession | None,
+    ) -> None:
+        """Notify class-based middleware of authenticated non-executing decisions."""
+        for middleware in self._middleware:
+            if isinstance(middleware, FunctionMiddleware):
+                middleware.on_approval_responses(responses, session=session)
+
     def _register_middleware(self, middleware: FunctionMiddlewareTypes) -> None:
         """Register a function middleware item.
 

python/packages/core/agent_framework/_tools.py

@@ -3239,6 +3239,7 @@ async def _resolve_approval_responses(
     max_errors: int,
     execute_function_calls: _FunctionCallExecutor,
     invocation_session: AgentSession | None = None,
+    middleware_pipeline: FunctionMiddlewarePipeline | None = None,
     settle_dangling_calls: Callable[[Sequence[Content]], Awaitable[None]] | None = None,
 ) -> _FunctionProcessingResult:
     """Resolve inbound approval responses before the next model call.
@@ -3281,6 +3282,11 @@ async def _resolve_approval_responses(
     responses_to_execute = [
         response for response in pending_approval_responses.values() if _is_approval_granted(response.approved)
     ]
+    responses_not_granted = [
+        response for response in pending_approval_responses.values() if not _is_approval_granted(response.approved)
+    ]
+    if middleware_pipeline is not None and responses_not_granted:
+        middleware_pipeline.notify_approval_responses(responses_not_granted, session=invocation_session)
     execution_result_groups: list[list[Content]] = []
     should_terminate = False
     reached_error_limit = False
@@ -3577,6 +3583,7 @@ async def _get_response_with_function_invocation(
         invocation_session: AgentSession | None,
         budget_state: dict[str, Any],
         max_errors: int,
+        middleware_pipeline: FunctionMiddlewarePipeline | None = None,
     ) -> ChatResponse[Any]:
         """Run the non-streaming function invocation loop."""
         from ._middleware import MiddlewareFailure
@@ -3622,6 +3629,7 @@ async def settle_approval_replay_calls(function_calls: Sequence[Content]) -> Non
             max_errors=max_errors,
             execute_function_calls=execute_function_calls,
             invocation_session=invocation_session,
+            middleware_pipeline=middleware_pipeline,
             settle_dangling_calls=settle_approval_replay_calls,
         )
         function_call_messages.extend(approval_processing.response_messages)
@@ -3762,6 +3770,7 @@ async def _stream_response_with_function_invocation(
         invocation_session: AgentSession | None,
         budget_state: dict[str, Any],
         max_errors: int,
+        middleware_pipeline: FunctionMiddlewarePipeline | None = None,
     ) -> AsyncIterable[ChatResponseUpdate]:
         """Run the streaming function invocation loop."""
         from ._middleware import MiddlewareFailure
@@ -3804,6 +3813,7 @@ async def settle_approval_replay_calls(function_calls: Sequence[Content]) -> Non
             max_errors=max_errors,
             execute_function_calls=execute_function_calls,
             invocation_session=invocation_session,
+            middleware_pipeline=middleware_pipeline,
             settle_dangling_calls=settle_approval_replay_calls,
         )
         errors_in_a_row = approval_processing.errors_in_a_row
@@ -4175,6 +4185,7 @@ def get_response(
                 invocation_session=invocation_session,
                 budget_state=budget_state,
                 max_errors=max_errors,
+                middleware_pipeline=function_middleware_pipeline,
             )
 
         response_format = mutable_options.get("response_format")
@@ -4191,6 +4202,7 @@ def get_response(
                 invocation_session=invocation_session,
                 budget_state=budget_state,
                 max_errors=max_errors,
+                middleware_pipeline=function_middleware_pipeline,
             ),
             finalizer=partial(ChatResponse.from_updates, output_format_type=response_format),
         )