Python: Isolate declarative MCP client sessions by workflow

docs/decisions/0046-python-declarative-request-scope-recovery.md

@@ -0,0 +1,74 @@
+---
+status: proposed
+contact: baywet
+date: 2026-10-15
+deciders: eavanvalkenburg, westey-m
+---
+
+# Preserve declarative request origins with opt-in legacy recovery
+
+## Context and Problem Statement
+
+A declarative workflow can retain requests from several fresh runs. Resuming a
+question, tool approval, or agent external-input request must restore its
+originating workflow scope rather than inherit the most recent run's MCP session.
+Older checkpoints do not retain that origin, so the framework cannot infer which
+session should be resumed.
+
+## Decision Drivers
+
+- All declarative pause types need the same origin and response-batch rules.
+- Existing checkpoints need an explicit recovery path without guessing an origin.
+- Repeated restoration must select the same migrated scope without sharing a newer run's session.
+
+## Considered Options
+
+- Use the current workflow scope for requests without an origin. This needs no
+  host changes and retains apparent continuity, but can attach an older request
+  to an unrelated fresh run.
+- Reject all unknown origins and require restarting the workflow. This avoids
+  guessing, but prevents hosts from recovering otherwise usable pending requests.
+- Automatically assign isolated scopes. This permits recovery without host
+  changes, but silently loses the previous MCP session's server-side state.
+- Require explicit host opt-in to assign stable, isolated scopes. This makes the
+  continuity trade-off visible and permits recovery, but requires host changes
+  and separate resumes for legacy requests.
+
+## Decision Outcome
+
+Choose explicit host opt-in through
+`WorkflowFactory(migrate_legacy_request_scopes=True)`, defaulting to `False`:
+
+- Questions/external input, function approvals, MCP approvals, and agent external
+  input retain their originating scope in the request. Response handlers restore
+  it before continuing. Agent requests also retain their own loop configuration.
+- Factory-built workflows validate all submitted request origins before consuming
+  responses. Mixed-scope batches are rejected without changing request metadata.
+- Checkpoint restores validate queued response origins before restoring shared or
+  executor state, including embedded child checkpoints. Submitted and already
+  queued responses share the same batch validation before pending requests are consumed.
+- Nested restoration preflights a detached copy of the complete embedded tree
+  with each child's own policy before applying live state or queues. Migration
+  metadata is confined to that candidate; a rejected child leaves earlier
+  siblings, parent executors, and the supplied checkpoint unchanged.
+- Without opt-in, a missing origin produces an actionable recovery error.
+- With opt-in, a missing origin becomes
+  `legacy-request:<framework request ID>`. The distinct namespace never selects
+  the active run's scope; preserving the request ID makes migration stable across
+  workers and repeated restoration. Legacy requests must be resumed separately.
+- Existing valid origins are unchanged. Invalid stored origins are not migrated.
+  Migration does not change header verification or authorize an unverifiable
+  MCP header context.
+- Newly emitted requests and subsequent checkpoints retain the migrated scope.
+  The old MCP session and its server-side history cannot be reconstructed.
+  Hosts requiring that continuity must restart rather than opt in.
+- Legacy agent loop configuration remains available from its old checkpoint
+  slot; new requests own their configuration instead of replacing shared state.
+
+Checkpoint contents and framework request IDs remain trusted host-owned data.
+Migration does not reconstruct state already overwritten in a legacy checkpoint.
+This is a checkpoint-compatibility change with an explicit recovery path, not
+transparent restoration of an old session.
+
+The proposed deciders are Python code owners; maintainer and
+engineering-management approval is still required.

python/packages/core/AGENTS.md

@@ -333,6 +333,9 @@ The vector store API is experimental under the shared `VECTOR_STORES` feature ID
   by that continuation, can atomically restore a supplied checkpoint before cancellation, and returns the resulting
   `WorkflowRunResult`. Cancelling a computer request also cancels the other pending requests in that agent's batch,
   retains already resolved sibling evidence in terminal output, and resets its session before new input.
+- **Nested checkpoint restoration** - Stored and embedded restores preflight a detached checkpoint tree before
+  replacing live state or queues. `WorkflowExecutor` delegates its private validation hook to its child's runner,
+  so each child uses its own response-origin policy and migration metadata stays in the candidate snapshot.
 - **`WorkflowBuilder`** - Fluent API for building workflows, including explicit
   `output_from` / `intermediate_output_from` selection for caller-facing emissions. `output_from`
   is an allow-list for **Workflow Output**; unselected executor payloads are hidden unless

python/packages/core/agent_framework/_workflows/_agent.py

@@ -332,6 +332,7 @@ async def _run_impl(
         output_events: list[WorkflowEvent[Any]] = []
         async for event in self._run_core(
             session_messages,
+            session.session_id if session is not None else None,
             checkpoint_id,
             checkpoint_storage,
             streaming=False,
@@ -418,6 +419,7 @@ async def _run_stream_impl(
         all_updates: list[AgentResponseUpdate] = []
         async for event in self._run_core(
             session_messages,
+            session.session_id if session is not None else None,
             checkpoint_id,
             checkpoint_storage,
             streaming=True,
@@ -439,6 +441,7 @@ async def _run_stream_impl(
     async def _run_core(
         self,
         input_messages: Sequence[Message],
+        agent_session_id: str | None,
         checkpoint_id: str | None,
         checkpoint_storage: CheckpointStorage | None,
         streaming: bool,
@@ -453,6 +456,7 @@ async def _run_core(
 
         Args:
             input_messages: Normalized input messages to process.
+            agent_session_id: Framework-local identity of the supplied agent session.
             checkpoint_id: ID of checkpoint to restore from.
             checkpoint_storage: Runtime checkpoint storage.
             streaming: Whether to use streaming workflow methods.
@@ -470,19 +474,23 @@ async def _run_core(
             logger.debug(f"Restoring workflow from checkpoint {checkpoint_id}")
             # Restore the workflow from checkpoint
             if streaming:
-                async for _ in self.workflow.run(
+                restore_stream = self.workflow.run(
                     stream=True,
                     checkpoint_id=checkpoint_id,
                     checkpoint_storage=checkpoint_storage,
                     tools=tools,
-                ):
+                )
+                self.workflow._set_agent_run_context(agent_session_id)  # pyright: ignore[reportPrivateUsage]
+                async for _ in restore_stream:
                     pass
             else:
-                _ = await self.workflow.run(
+                restore_result = self.workflow.run(
                     checkpoint_id=checkpoint_id,
                     checkpoint_storage=checkpoint_storage,
                     tools=tools,
                 )
+                self.workflow._set_agent_run_context(agent_session_id)  # pyright: ignore[reportPrivateUsage]
+                _ = await restore_result
             if not input_messages:
                 logger.info("No input messages provided; the workflow has been restored to the checkpoint state.")
                 return
@@ -500,43 +508,49 @@ async def _run_core(
             pending_requests = await self.workflow._runner_context.get_pending_request_info_events()  # pyright: ignore[reportPrivateUsage]
             function_responses = self._extract_function_responses(input_messages, pending_requests)
             if streaming:
-                async for event in self.workflow.run(
+                response_stream = self.workflow.run(
                     responses=function_responses,
                     stream=True,
                     checkpoint_storage=checkpoint_storage,
                     tools=tools,
                     function_invocation_kwargs=function_invocation_kwargs,
                     client_kwargs=client_kwargs,
-                ):
+                )
+                async for event in response_stream:
                     yield event
             else:
-                for event in await self.workflow.run(
+                response_result = self.workflow.run(
                     responses=function_responses,
                     checkpoint_storage=checkpoint_storage,
                     tools=tools,
                     function_invocation_kwargs=function_invocation_kwargs,
                     client_kwargs=client_kwargs,
-                ):
+                )
+                for event in await response_result:
                     yield event
         elif final_state == WorkflowRunState.IDLE:
             if streaming:
-                async for event in self.workflow.run(
+                response_stream = self.workflow.run(
                     message=input_messages,
                     stream=True,
                     checkpoint_storage=checkpoint_storage,
                     tools=tools,
                     function_invocation_kwargs=function_invocation_kwargs,
                     client_kwargs=client_kwargs,
-                ):
+                )
+                self.workflow._set_agent_run_context(agent_session_id)  # pyright: ignore[reportPrivateUsage]
+                async for event in response_stream:
                     yield event
             else:
-                for event in await self.workflow.run(
+                response_result = self.workflow.run(
                     message=input_messages,
                     checkpoint_storage=checkpoint_storage,
                     tools=tools,
                     function_invocation_kwargs=function_invocation_kwargs,
                     client_kwargs=client_kwargs,
-                ):
+                )
+                self.workflow._set_agent_run_context(agent_session_id)  # pyright: ignore[reportPrivateUsage]
+                for event in await response_result:
                     yield event
         else:
             raise AgentException(f"The underlying workflow is in an invalid state to restart: {final_state}.")

python/packages/core/agent_framework/_workflows/_const.py

@@ -17,6 +17,12 @@
 # to pass kwargs from workflow.run() through to agent.run() and @tool functions.
 WORKFLOW_RUN_KWARGS_KEY = "_workflow_run_kwargs"
 
+# State key for the AgentSession identity supplied through WorkflowAgent.
+WORKFLOW_AGENT_SESSION_ID_KEY = "_workflow_agent_session_id"
+
+# State key identifying WorkflowAgent turns, including turns without an AgentSession.
+WORKFLOW_AGENT_RUN_KEY = "_workflow_agent_run"
+
 # Key used to store collision-free, executor-aware workflow invocation kwargs.
 # WORKFLOW_RUN_KWARGS_KEY remains in the legacy dict format for compatibility with
 # older first-party packages and checkpoints.

python/packages/core/agent_framework/_workflows/_executor.py

@@ -7,7 +7,7 @@
 import inspect
 import logging
 import types
-from collections.abc import Awaitable, Callable
+from collections.abc import Awaitable, Callable, Sequence
 from typing import Any, TypeVar, overload
 
 from ..observability import create_processing_span
@@ -31,6 +31,8 @@
 
 logger = logging.getLogger(__name__)
 
+_ResponseBatchValidation = tuple[Callable[[Sequence[WorkflowEvent]], None], Sequence[WorkflowEvent]]
+
 
 # region Executor
 class Executor(RequestInfoMixin, DictConvertible):
@@ -533,6 +535,17 @@ async def _cancel_pending_request(self, request_id: str, ctx: WorkflowContext[An
         """Release executor-owned state for a cancelled workflow request."""
         del request_id, ctx
 
+    async def _get_response_validation_batches(
+        self, requests: Sequence[WorkflowEvent]
+    ) -> list[_ResponseBatchValidation]:
+        """Collect delegated response policies without changing pending requests."""
+        del requests
+        return []
+
+    def _validate_checkpoint_restore(self, state: dict[str, Any], requests: Sequence[WorkflowEvent] = ()) -> None:
+        """Validate a checkpoint snapshot without changing live executor state."""
+        del state, requests
+
     async def on_checkpoint_save(self) -> dict[str, Any]:
         """Hook called when the workflow is being saved to a checkpoint.