Python: Harden FIDES service history continuation

docs/features/FIDES_IMPLEMENTATION_SUMMARY.md

@@ -4,6 +4,10 @@
 
 **FIDES**  is a comprehensive deterministic prompt injection defense system for the agent framework. The implementation provides label-based security mechanisms to defend against prompt injection attacks by tracking integrity and confidentiality of content throughout agent execution.
 
+FIDES enforces tool-result, hidden-variable, caller-replayed, and history-provider flows. Other `ContextProvider`
+implementations remain trusted plugins; provider-injected messages require an explicit provider/application trust
+boundary until a first-class provider-message labeling contract is defined.
+
 **🚀 Key Features:**
 - **Context Provider Pattern** - `SecureAgentConfig` extends `ContextProvider`, injecting tools, instructions, and middleware automatically
 - **Automatic Variable Hiding** - UNTRUSTED content is automatically hidden without requiring manual intervention
@@ -363,4 +367,8 @@ cd python/packages/core && ../../.venv/bin/pytest tests/test_security.py -v
 - **Full traceability**: Message-level label tracking
 - **Complete auditability**: All security events logged
 
-The system ensures that untrusted content never directly reaches the LLM context and that all tool calls are policy-checked based on the cumulative security state before execution.
+The system hides FIDES-labeled untrusted tool output from the LLM context and policy-checks tool calls against the
+cumulative security state before execution. Newly supplied assistant/tool messages are conservatively labeled even
+when state was restored. History-provider messages reconstruct missing state, while restored state remains
+authoritative for unstamped stored history; trusted non-history context providers remain responsible for the data they
+inject.

python/packages/core/agent_framework/_agents.py

@@ -1564,8 +1564,11 @@ async def _prepare_run_context(
         )
         # Resolve conversation_id from the same combined view so an agent-level default is honored
         # when the runtime omits it (a live session id still takes precedence below).
-        effective_conversation_id = effective_options.get("conversation_id")
+        effective_conversation_id = cast(str | None, effective_options.get("conversation_id"))
         session_conversation_id = self._get_chat_conversation_id(session)
+        resolved_conversation_id = (
+            session_conversation_id if session_conversation_id is not None else effective_conversation_id
+        )
         # Auto-inject InMemoryHistoryProvider when session is provided, no context providers
         # registered, and no service-side storage indicators
         if (
@@ -1615,6 +1618,8 @@ async def _prepare_run_context(
             session=active_session,
             input_messages=input_messages,
             options=opts,
+            conversation_id=resolved_conversation_id,
+            service_stores_history=service_stores_history,
         )
         default_additional_args = chat_options.pop("additional_function_arguments", None)
         if isinstance(default_additional_args, Mapping):
@@ -1732,6 +1737,15 @@ async def _prepare_run_context(
                 providers=per_service_call_history_providers,
                 service_stores_history=service_stores_history,
             )
+        if per_service_call_history_middleware is not None:
+            existing_middleware = cast(
+                "MiddlewareTypes | Sequence[MiddlewareTypes] | None", effective_client_kwargs.get("middleware")
+            )
+            effective_client_kwargs["middleware"] = [
+                *_as_middleware_list(existing_middleware),
+                per_service_call_history_middleware,
+            ]
+
         provider_middleware = session_context.get_middleware()
         if provider_middleware:
             # Providers may only contribute chat/function middleware (enforced by
@@ -1751,15 +1765,6 @@ async def _prepare_run_context(
                     *provider_function_chat_middleware,
                 ]
 
-        if per_service_call_history_middleware is not None:
-            existing_middleware = cast(
-                "MiddlewareTypes | Sequence[MiddlewareTypes] | None", effective_client_kwargs.get("middleware")
-            )
-            effective_client_kwargs["middleware"] = [
-                *_as_middleware_list(existing_middleware),
-                per_service_call_history_middleware,
-            ]
-
         return {
             "session": active_session,
             "session_context": session_context,
@@ -1780,6 +1785,8 @@ async def _prepare_session_and_messages(
         session: AgentSession | None,
         input_messages: list[Message] | None = None,
         options: dict[str, Any] | None = None,
+        conversation_id: str | None = None,
+        service_stores_history: bool = False,
     ) -> tuple[SessionContext, dict[str, Any]]:
         """Prepare the session context and messages for agent execution.
 
@@ -1790,6 +1797,8 @@ async def _prepare_session_and_messages(
             session: The conversation session (None for stateless invocation).
             input_messages: Messages to process.
             options: Runtime options dict (already copied, safe to mutate).
+            conversation_id: Effective provider conversation identity after session/default/runtime precedence.
+            service_stores_history: Whether the service owns history for this run.
 
         Returns:
             A tuple containing:
@@ -1811,11 +1820,14 @@ async def _prepare_session_and_messages(
         if provider_session is None and self.context_providers:
             provider_session = AgentSession()
 
+        context_options = dict(options or {})
+        if conversation_id is not None:
+            context_options["conversation_id"] = conversation_id
         session_context = SessionContext(
             session_id=provider_session.session_id if provider_session else None,
             service_session_id=provider_session.service_session_id if provider_session else None,
             input_messages=input_messages or [],
-            options=options or {},
+            options=context_options,
         )
 
         # When per-service-call persistence is enabled, the per-service-call middleware owns
@@ -1841,6 +1853,43 @@ async def _prepare_session_and_messages(
                 state=provider_session.state.setdefault(provider.source_id, {}),
             )
 
+        approval_history_consumers = [
+            (provider, prepare)
+            for provider in self.context_providers
+            if callable(prepare := getattr(provider, "_prepare_approval_history", None))
+        ]
+        if approval_history_consumers:
+            if provider_session is None:
+                raise RuntimeError("Provider session must be available when approval history is prepared.")
+            approval_history_context = session_context
+            if per_service_call_history_required and not service_stores_history:
+                approval_history_context = SessionContext(
+                    session_id=provider_session.session_id,
+                    service_session_id=provider_session.service_session_id,
+                    input_messages=list(session_context.input_messages),
+                    options=dict(session_context.options),
+                    metadata=dict(session_context.metadata),
+                )
+                for provider in self.context_providers:
+                    if not isinstance(provider, HistoryProvider) or not provider.load_messages:
+                        continue
+                    await provider.before_run(
+                        agent=self,
+                        session=provider_session,
+                        context=approval_history_context,
+                        state=provider_session.state.setdefault(provider.source_id, {}),
+                    )
+
+            for provider, prepare in approval_history_consumers:
+                prepared = prepare(
+                    agent=self,
+                    session=provider_session,
+                    context=approval_history_context,
+                    state=provider_session.state.setdefault(provider.source_id, {}),
+                )
+                if isawaitable(prepared):
+                    _ = await prepared
+
         # Merge provider-contributed tools into chat_options
         if session_context.tools:
             if chat_options.get("tools") is not None:

python/packages/core/agent_framework/_middleware.py

@@ -1559,6 +1559,15 @@ def create_next_handler(index: int) -> Callable[[], Awaitable[None]]:
             if index >= len(self._middleware):
 
                 async def final_wrapper() -> None:
+                    from ._tools import (
+                        _SECURITY_ARGUMENT_MATERIALIZATION_CONTEXT,  # pyright: ignore[reportPrivateUsage]
+                    )
+
+                    security_materialization = _SECURITY_ARGUMENT_MATERIALIZATION_CONTEXT.get()
+                    if security_materialization is not None:
+                        active_context, materialize = security_materialization
+                        if context is active_context:
+                            materialize(context)
                     context.result = final_handler(context)
                     if inspect.isawaitable(context.result):
                         context.result = await context.result

python/packages/core/agent_framework/_tools.py

@@ -138,6 +138,9 @@ def _has_authoritative_approval_session(invocation_session: AgentSession | None)
 _SECURITY_ARGUMENTS_SNAPSHOT_CONTEXT_KEY: Final[str] = "_security_function_arguments"
 _PREPARED_ARGUMENTS_CONTEXT_KEY: Final[str] = "_prepared_function_arguments"
 _AUTO_ARGUMENT_PREPARATION_CONTEXT_KEY: Final[str] = "_auto_prepare_function_arguments"
+_SECURITY_ARGUMENT_MATERIALIZATION_CONTEXT: Final[contextvars.ContextVar[tuple[Any, Callable[[Any], None]] | None]] = (
+    contextvars.ContextVar("_security_argument_materialization_context", default=None)
+)
 _FUNCTION_INVOCATION_LIMIT_FALLBACK_TEXT: Final[str] = (
     "Function invocation limit reached before a final answer could be produced."
 )
@@ -1138,6 +1141,15 @@ async def invoke(
         from ._types import Content
         from .observability import OBSERVABILITY_SETTINGS
 
+        security_materialization = _SECURITY_ARGUMENT_MATERIALIZATION_CONTEXT.get()
+        if security_materialization is not None:
+            active_context, materialize = security_materialization
+            if active_context.function is self and (context is None or context is active_context):
+                active_context.metadata[_AUTO_ARGUMENT_PREPARATION_CONTEXT_KEY] = True
+                materialize(active_context)
+                arguments = active_context.arguments
+                context = active_context
+
         configured_parser = self.result_parser
         skip_parsing = skip_parsing or configured_parser is SKIP_PARSING