diff --git a/python/packages/azurefunctions/agent_framework_azurefunctions/_context.py b/python/packages/azurefunctions/agent_framework_azurefunctions/_context.py
index 4912fe4cc92..6d76e179964 100644
--- a/python/packages/azurefunctions/agent_framework_azurefunctions/_context.py
+++ b/python/packages/azurefunctions/agent_framework_azurefunctions/_context.py
@@ -115,6 +115,18 @@ async def create_checkpoint(
         """Checkpointing not supported in activity context."""
         raise NotImplementedError("Checkpointing is not supported in Azure Functions activity context")
 
+    async def create_checkpoint_object(
+        self,
+        workflow_name: str,
+        graph_signature_hash: str,
+        state: State,
+        previous_checkpoint_id: str | None,
+        iteration_count: int,
+        metadata: dict[str, Any] | None = None,
+    ) -> WorkflowCheckpoint:
+        """Checkpointing not supported in activity context."""
+        raise NotImplementedError("Checkpointing is not supported in Azure Functions activity context")
+
     async def load_checkpoint(self, checkpoint_id: str) -> WorkflowCheckpoint | None:
         """Checkpointing not supported in activity context."""
         raise NotImplementedError("Checkpointing is not supported in Azure Functions activity context")
diff --git a/python/packages/core/agent_framework/_workflows/_functional.py b/python/packages/core/agent_framework/_workflows/_functional.py
index 098700bea55..2ffe99807c0 100644
--- a/python/packages/core/agent_framework/_workflows/_functional.py
+++ b/python/packages/core/agent_framework/_workflows/_functional.py
@@ -811,6 +811,24 @@ def run(
                 execution is not allowed).
         """
         self._validate_run_params(message, responses, checkpoint_id)
+        # Warn (but don't block) when a fresh message or a checkpoint restore begins while a prior
+        # run left request_info events pending. Mirrors Workflow.run. Delivering responses is the
+        # normal way to complete the pending cycle and is intentionally not warned.
+        if (message is not None or checkpoint_id is not None) and self._last_pending_request_ids:
+            logger.warning(
+                "Workflow %s received %s while %d request_info event(s) are still pending from an "
+                "unfinished request/response cycle; %s. Deliver responses (responses=...) to complete "
+                "the pending cycle before starting new input.",
+                self.name,
+                "a fresh message" if message is not None else "a checkpoint restore",
+                len(self._last_pending_request_ids),
+                (
+                    "those requests remain answerable, but this run advances workflow state, so a "
+                    "response that arrives later may apply to a workflow that has moved on"
+                    if message is not None
+                    else "those pending requests will be overwritten by the checkpoint's state"
+                ),
+            )
         if responses and checkpoint_id is None:
             # Require at least one response key to match a currently-pending
             # request; prevents silent replay against stale state while still
diff --git a/python/packages/core/agent_framework/_workflows/_runner.py b/python/packages/core/agent_framework/_workflows/_runner.py
index 543fe0d93d8..bf56bedfa06 100644
--- a/python/packages/core/agent_framework/_workflows/_runner.py
+++ b/python/packages/core/agent_framework/_workflows/_runner.py
@@ -245,7 +245,13 @@ async def _prepare_checkpoint_state(self) -> None:
         self._state.commit()
 
     async def create_checkpoint_if_enabled(self) -> None:
-        """Create a checkpoint if checkpointing is enabled and attach a label and metadata."""
+        """Create a checkpoint and save the checkpoint to the configured storage if one is configured.
+
+        Note:
+            1. This method has no effect if checkpointing is not enabled in the context.
+            2. Do not use this method with ``create_checkpoint_object`` as both methods advance the
+               ``previous_checkpoint_id`` and may result in unexpected checkpoint lineage.
+        """
         if not self._ctx.has_checkpointing():
             return
 
@@ -340,6 +346,63 @@ async def restore_from_checkpoint(
             logger.error(f"Failed to restore from checkpoint {checkpoint_id}: {e}")
             raise WorkflowCheckpointException(f"Failed to restore from checkpoint {checkpoint_id}") from e
 
+    async def create_checkpoint_object(self) -> WorkflowCheckpoint:
+        """Create a checkpoint object.
+
+        Note:
+            1. Do not use this method with ``create_checkpoint_if_enabled`` as both methods advance the
+               ``previous_checkpoint_id`` and may result in unexpected checkpoint lineage.
+
+        Returns:
+            A ``WorkflowCheckpoint``.
+        """
+        # Persist executor snapshots into committed shared state before exporting it.
+        await self._prepare_checkpoint_state()
+        checkpoint = await self._ctx.create_checkpoint_object(
+            self._workflow_name,
+            self._graph_signature_hash,
+            self._state,
+            self._previous_checkpoint_id,
+            self._iteration,
+        )
+        self._previous_checkpoint_id = checkpoint.checkpoint_id
+        return checkpoint
+
+    async def restore_from_checkpoint_object(self, checkpoint: WorkflowCheckpoint) -> None:
+        """Restore runner state from an in-memory ``WorkflowCheckpoint`` object.
+
+        Unlike :meth:`restore_from_checkpoint`, this does not load from a storage
+        backend; it applies a checkpoint the caller already holds - for example, a
+        child workflow checkpoint embedded in a parent ``WorkflowExecutor``'s state.
+
+        Restores shared state, executor snapshots, in-flight messages, and pending
+        request_info events, then marks the runner as resumed.
+
+        Args:
+            checkpoint: The checkpoint whose state should be restored.
+
+        Raises:
+            WorkflowCheckpointException: If the checkpoint's graph signature does not
+                match this runner's workflow, or if restoration otherwise fails.
+        """
+        if self._graph_signature_hash != checkpoint.graph_signature_hash:
+            raise WorkflowCheckpointException(
+                "Workflow graph has changed since the checkpoint was created. "
+                "Please rebuild the original workflow before resuming."
+            )
+
+        try:
+            # Clear first so import_state (which merges) does not leak stale keys from a
+            # prior run on this Workflow instance.
+            self._state.clear()
+            self._state.import_state(checkpoint.state)
+            await self._restore_executor_states()
+            await self._ctx.apply_checkpoint(checkpoint)
+            self._mark_resumed(checkpoint)
+        except Exception as e:
+            logger.error(f"Failed to restore from checkpoint {checkpoint.checkpoint_id}: {e}")
+            raise WorkflowCheckpointException(f"Failed to restore from checkpoint {checkpoint.checkpoint_id}") from e
+
     async def _save_executor_states(self) -> None:
         """Populate executor state by calling checkpoint hooks on executors."""
         for exec_id, executor in self._executors.items():
diff --git a/python/packages/core/agent_framework/_workflows/_runner_context.py b/python/packages/core/agent_framework/_workflows/_runner_context.py
index 6cbee0b039c..5a11afd8b60 100644
--- a/python/packages/core/agent_framework/_workflows/_runner_context.py
+++ b/python/packages/core/agent_framework/_workflows/_runner_context.py
@@ -195,6 +195,34 @@ def is_streaming(self) -> bool:
         """
         ...
 
+    async def create_checkpoint_object(
+        self,
+        workflow_name: str,
+        graph_signature_hash: str,
+        state: State,
+        previous_checkpoint_id: CheckpointID | None,
+        iteration_count: int,
+        metadata: dict[str, Any] | None = None,
+    ) -> WorkflowCheckpoint:
+        """Build an in-memory checkpoint of the current context state without persisting it.
+
+        Unlike :meth:`create_checkpoint`, this does not require checkpoint storage and does
+        not save anything; it returns the checkpoint object for the caller to own.
+
+        Args:
+            workflow_name: The name of the workflow for which the checkpoint is being created.
+            graph_signature_hash: Hash of the workflow graph topology to
+                validate checkpoint compatibility during restore.
+            state: The state to include in the checkpoint.
+            previous_checkpoint_id: The ID of the previous checkpoint, if any, to form a checkpoint chain.
+            iteration_count: The current iteration count of the workflow.
+            metadata: Optional metadata to associate with the checkpoint.
+
+        Returns:
+            A ``WorkflowCheckpoint`` snapshot of the current context state.
+        """
+        ...
+
     async def create_checkpoint(
         self,
         workflow_name: str,
@@ -204,7 +232,7 @@ async def create_checkpoint(
         iteration_count: int,
         metadata: dict[str, Any] | None = None,
     ) -> CheckpointID:
-        """Create a checkpoint of the current workflow state.
+        """Create and persist a checkpoint of the current workflow state.
 
         Args:
             workflow_name: The name of the workflow for which the checkpoint is being created.
@@ -381,7 +409,7 @@ def clear_runtime_checkpoint_storage(self) -> None:
     def has_checkpointing(self) -> bool:
         return self._get_effective_checkpoint_storage() is not None
 
-    async def create_checkpoint(
+    async def create_checkpoint_object(
         self,
         workflow_name: str,
         graph_signature_hash: str,
@@ -389,21 +417,40 @@ async def create_checkpoint(
         previous_checkpoint_id: CheckpointID | None,
         iteration_count: int,
         metadata: dict[str, Any] | None = None,
-    ) -> CheckpointID:
-        storage = self._get_effective_checkpoint_storage()
-        if not storage:
-            raise ValueError("Checkpoint storage not configured")
-
-        checkpoint = WorkflowCheckpoint(
+    ) -> WorkflowCheckpoint:
+        return WorkflowCheckpoint(
             workflow_name=workflow_name,
             graph_signature_hash=graph_signature_hash,
             previous_checkpoint_id=previous_checkpoint_id,
-            messages=dict(self._messages),
+            # Copy the per-source lists so the snapshot is isolated from later context mutations.
+            messages={source_id: list(messages) for source_id, messages in self._messages.items()},
             state=state.export_state(),
             pending_request_info_events=dict(self._pending_request_info_events),
             iteration_count=iteration_count,
             metadata=metadata or {},
         )
+
+    async def create_checkpoint(
+        self,
+        workflow_name: str,
+        graph_signature_hash: str,
+        state: State,
+        previous_checkpoint_id: CheckpointID | None,
+        iteration_count: int,
+        metadata: dict[str, Any] | None = None,
+    ) -> CheckpointID:
+        storage = self._get_effective_checkpoint_storage()
+        if not storage:
+            raise ValueError("Checkpoint storage not configured")
+
+        checkpoint = await self.create_checkpoint_object(
+            workflow_name,
+            graph_signature_hash,
+            state,
+            previous_checkpoint_id,
+            iteration_count,
+            metadata,
+        )
         checkpoint_id = await storage.save(checkpoint)
         logger.debug(f"Created checkpoint {checkpoint_id}")
         return checkpoint_id
diff --git a/python/packages/core/agent_framework/_workflows/_workflow.py b/python/packages/core/agent_framework/_workflows/_workflow.py
index 099e03df2ff..cef0f8b12c9 100644
--- a/python/packages/core/agent_framework/_workflows/_workflow.py
+++ b/python/packages/core/agent_framework/_workflows/_workflow.py
@@ -815,10 +815,9 @@ async def _run_core(
             # runner context has fully drained from any prior run. If it still
             # has in-flight executor messages, the prior run didn't complete -
             # the caller must either resume from a checkpoint or wait for the
-            # prior run to drain. (Pending request_info events are intentionally
-            # NOT blocked here: a follow-up run with message=... is the normal
-            # way to deliver a response to those pending requests, e.g. via
-            # WorkflowAgent._process_pending_requests.)
+            # prior run to drain. Pending request_info events are intentionally
+            # NOT blocked here (they are answered via a follow-up ``responses=...``
+            # run); the warning below surfaces the abandon/overwrite cases instead.
             # NOTE: _validate_run_params already enforces that ``message`` is
             # mutually exclusive with both ``checkpoint_id`` and ``responses``,
             # so we don't need to re-check those here.
@@ -831,6 +830,33 @@ async def _run_core(
                     "checkpointing; there is no in-process recovery path."
                 )
 
+            # Warn (but don't block) when a fresh message or a checkpoint restore begins while the
+            # workflow still has pending request_info events from an unfinished request/response
+            # cycle. A fresh ``message`` does NOT drop those pending requests - they remain pending and
+            # can still be answered later - but the new run advances executor and shared state, so when
+            # a response for an earlier request eventually arrives the workflow may have moved on,
+            # yielding inconsistent results. A ``checkpoint_id`` restore instead replaces the context's
+            # pending requests with the checkpoint's state. Delivering ``responses`` is the normal way to
+            # answer pending requests and is intentionally not warned. Mirrors the WorkflowExecutor
+            # warning for overlapping sub-workflow executions.
+            if message is not None or checkpoint_id is not None:
+                pending_request_info_events = await self._runner.context.get_pending_request_info_events()
+                if pending_request_info_events:
+                    logger.warning(
+                        "Workflow %s received %s while %d request_info event(s) are still pending from an "
+                        "unfinished request/response cycle; %s. Deliver responses (responses=...) to complete "
+                        "the pending cycle before starting new input.",
+                        self.id,
+                        "a fresh message" if message is not None else "a checkpoint restore",
+                        len(pending_request_info_events),
+                        (
+                            "those requests remain pending, but this run advances executor and shared state, "
+                            "so a response that arrives later may apply to a workflow that has moved on"
+                            if message is not None
+                            else "those pending requests will be overwritten by the checkpoint's state"
+                        ),
+                    )
+
             initial_executor_fn = self._resolve_execution_mode(message, responses, checkpoint_id, checkpoint_storage)
 
             async for event in self._run_workflow_with_tracing(
diff --git a/python/packages/core/agent_framework/_workflows/_workflow_executor.py b/python/packages/core/agent_framework/_workflows/_workflow_executor.py
index a77887ba053..972387c643f 100644
--- a/python/packages/core/agent_framework/_workflows/_workflow_executor.py
+++ b/python/packages/core/agent_framework/_workflows/_workflow_executor.py
@@ -4,14 +4,14 @@
 import logging
 import sys
 import types
-import uuid
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any
 
+from ..exceptions import WorkflowCheckpointException
+
 if TYPE_CHECKING:
     from ._workflow import Workflow
 
-from ._checkpoint_encoding import decode_checkpoint_value
 from ._const import GLOBAL_KWARGS_KEY, WORKFLOW_RUN_KWARGS_KEY
 from ._events import (
     WorkflowEvent,
@@ -36,7 +36,12 @@
 
 @dataclass
 class ExecutionContext:
-    """Context for tracking a single sub-workflow execution."""
+    """Legacy per-execution bookkeeping.
+
+    Retained only to decode checkpoints written before the sub-workflow's own checkpoint was
+    embedded (see ``WorkflowExecutor.on_checkpoint_restore``). It is no longer used at runtime -
+    the wrapped sub-workflow is the single source of truth for its pending requests.
+    """
 
     # The ID of the execution context
     execution_id: str
@@ -161,11 +166,9 @@ class WorkflowExecutor(Executor):
         # The response handler expects a SubWorkflowResponseMessage wrapping the response data.
 
     ### State Management
-    WorkflowExecutor maintains execution state across request/response cycles:
-    - Tracks pending requests by request_id
-    - Accumulates responses until all expected responses are received
-    - Resumes sub-workflow execution with complete response batch
-    - Handles concurrent executions and multiple pending requests
+    WorkflowExecutor keeps no request/response bookkeeping of its own. The wrapped sub-workflow
+    is the single source of truth for its pending requests; responses are forwarded to it and
+    validated against its own pending request_info events.
 
     ## Type System Integration
     WorkflowExecutor inherits its type signature from the wrapped workflow:
@@ -194,46 +197,21 @@ class WorkflowExecutor(Executor):
     - Converts to error event in parent context
     - Provides detailed error information including sub-workflow ID
 
-    ## Concurrent Execution Support
-    WorkflowExecutor fully supports multiple concurrent sub-workflow executions:
-
-    ### Per-Execution State Isolation
-    Each sub-workflow invocation creates an isolated ExecutionContext:
-
-    .. code-block:: python
-
-        # Multiple concurrent invocations are supported
-        workflow_executor = WorkflowExecutor(my_workflow, id="concurrent_executor")
-
-        # Each invocation gets its own execution context
-        # Execution 1: processes input_1 independently
-        # Execution 2: processes input_2 independently
-        # No state interference between executions
-
-    ### Request/Response Coordination
-    Responses are correctly routed to the originating execution:
-    - Each execution tracks its own pending requests and expected responses
-    - Request-to-execution mapping ensures responses reach the correct sub-workflow
-    - Response accumulation is isolated per execution
-    - Automatic cleanup when execution completes
-
-    ### Memory Management
-    - Unlimited concurrent executions supported
-    - Each execution has unique UUID-based identification
-    - Cleanup of completed execution contexts
-    - Thread-safe state management for concurrent access
-
-    ### Important Considerations
-    **Shared Workflow Instance**: All concurrent executions use the same underlying workflow instance.
-    For proper isolation, ensure that the wrapped workflow and its executors are stateless.
+    ## Overlapping Executions
+    A ``WorkflowExecutor`` wraps a single shared sub-workflow instance and keeps no per-execution
+    state. If a new input arrives while the sub-workflow still has pending request_info events from
+    an unfinished request/response cycle, the new input advances the shared sub-workflow state and
+    can interfere with that cycle - a response arriving later may apply to a sub-workflow that has
+    moved on. This is allowed but logs a warning, and is only safe when the wrapped workflow (and
+    its executors) are stateless.
 
     .. code-block:: python
 
-        # Avoid: Stateful executor with instance variables
+        # Avoid: stateful executor whose instance variables are shared across overlapping runs
         class StatefulExecutor(Executor):
             def __init__(self):
                 super().__init__(id="stateful")
-                self.data = []  # This will be shared across concurrent executions!
+                self.data = []  # Shared across overlapping sub-workflow executions!
 
     ## Integration with Parent Workflows
     Parent workflows can intercept sub-workflow requests:
@@ -255,12 +233,18 @@ async def handle_subworkflow_request(
                     # Forward to external handler
                     await ctx.request_info(request.source_event, response_type=request.source_event.response_type)
 
+    ## Checkpointing
+    The provided sub workflow may not have its own checkpoint storage. The sub workflow checkpointed states will
+    be managed by the parent workflow.
+
     ## Implementation Notes
-    - Sub-workflows run to completion before processing their results
-    - Event processing is atomic - all outputs are forwarded before requests
-    - Response accumulation ensures sub-workflows receive complete response batches
-    - Execution state is maintained for proper resumption after external requests
-    - Concurrent executions are fully isolated and do not interfere with each other
+    - Sub-workflows run to completion (or to idle-with-pending-requests) before their results are processed
+    - Event processing is ordered - outputs are forwarded before requests
+    - Responses are forwarded to the sub-workflow as they arrive; the sub-workflow tracks its own
+      pending requests and resumes when they are answered
+    - The WorkflowExecutor keeps no per-execution bookkeeping; the sub-workflow is the single source
+      of truth for its pending requests. Starting a new execution while the sub-workflow still has
+      pending requests logs a warning and is only safe when the wrapped workflow is stateless
     """
 
     def __init__(
@@ -274,19 +258,18 @@ def __init__(
         """Initialize the WorkflowExecutor.
 
         Args:
-            workflow: The workflow to execute as a sub-workflow.
+            workflow: The workflow to execute as a sub-workflow. This workflow instance (including
+                the executor instances within it) must be unique. If the same instances are shared
+                across multiple WorkflowExecutor instances, it may lead to incorrect behavior.
             id: Unique identifier for this executor.
-            allow_direct_output: Whether to allow direct output from the sub-workflow.
-                                 By default, outputs from the sub-workflow are sent to
-                                 other executors in the parent workflow as messages.
-                                 When this is set to true, the outputs are yielded
-                                 directly from the WorkflowExecutor to the parent
-                                 workflow's event stream.
-            propagate_request: Whether to propagate requests from the sub-workflow to the
-                               parent workflow. If set to true, requests from the sub-workflow
-                               will be propagated as the original WorkflowEvent to the parent
-                               workflow. Otherwise, they will be wrapped in a SubWorkflowRequestMessage,
-                               which should be handled by an executor in the parent workflow.
+            allow_direct_output: Whether to allow direct output from the sub-workflow. By default,
+                outputs from the sub-workflow are sent to other executors in the parent workflow as
+                messages. When this is set to true, the outputs are yielded directly from the
+                WorkflowExecutor to the parent workflow's event stream.
+            propagate_request: Whether to propagate requests from the sub-workflow to the parent
+                workflow. If set to true, requests from the sub-workflow will be propagated as the
+                original WorkflowEvent to the parent workflow. Otherwise, they will be wrapped in a
+                SubWorkflowRequestMessage, which should be handled by an executor in the parent workflow.
 
         Keyword Args:
             **kwargs: Additional keyword arguments passed to the parent constructor.
@@ -294,13 +277,14 @@ def __init__(
         super().__init__(id, **kwargs)
         self.workflow = workflow
         self.allow_direct_output = allow_direct_output
-
-        # Track execution contexts for concurrent sub-workflow executions
-        self._execution_contexts: dict[str, ExecutionContext] = {}  # execution_id -> ExecutionContext
-        # Map request_id to execution_id for response routing
-        self._request_to_execution: dict[str, str] = {}  # request_id -> execution_id
         self._propagate_request = propagate_request
 
+        if self.workflow._runner_context.has_checkpointing():  # type: ignore
+            raise WorkflowCheckpointException(
+                "The wrapped sub workflow must not have checkpointing enabled. "
+                "Sub workflow checkpointing is managed by the parent workflow."
+            )
+
     @property
     def input_types(self) -> list[type[Any] | types.UnionType]:
         """Get the input types based on the underlying workflow's input types plus WorkflowExecutor-specific types.
@@ -351,11 +335,10 @@ def can_handle(self, message: WorkflowMessage) -> bool:
             # Always handle SubWorkflowResponseMessage
             return True
 
-        if (
-            message.original_request_info_event is not None
-            and message.original_request_info_event.request_id in self._request_to_execution
-        ):
-            # Handle propagated responses for known requests
+        if message.original_request_info_event is not None:
+            # A propagated response is target-routed back to the executor that issued the request,
+            # so if one reaches this WorkflowExecutor it belongs to our sub-workflow. _handle_response
+            # validates it against the sub-workflow's pending requests and ignores anything unknown.
             return True
 
         # For other messages, only handle if the wrapped workflow can accept them as input
@@ -372,58 +355,52 @@ async def process_workflow(self, input_data: object, ctx: WorkflowContext[Any, A
             input_data: The input data to send to the sub-workflow.
             ctx: The workflow context from the parent.
         """
-        # Create execution context for this sub-workflow run
-        execution_id = str(uuid.uuid4())
-        execution_context = ExecutionContext(
-            execution_id=execution_id,
-            collected_responses={},
-            expected_response_count=0,
-            pending_requests={},
-        )
-        self._execution_contexts[execution_id] = execution_context
-
-        logger.debug(f"WorkflowExecutor {self.id} starting sub-workflow {self.workflow.id} execution {execution_id}")
-
-        try:
-            # Get kwargs from parent workflow's State to propagate to subworkflow
-            parent_kwargs: dict[str, Any] = ctx.get_state(WORKFLOW_RUN_KWARGS_KEY, {})
-
-            # Extract invocation kwargs recognised by Workflow.run()
-            # The state stores resolved format (with __global__ wrapper for global kwargs).
-            # Unwrap __global__ before passing to the subworkflow so it gets re-resolved
-            # against the subworkflow's own executor IDs.
-            fi_kwargs: dict[str, Any] | None = None
-            ci_kwargs: dict[str, Any] | None = None
-            for key in ("function_invocation_kwargs", "client_kwargs"):
-                resolved = parent_kwargs.get(key)
-                if isinstance(resolved, dict):
-                    # Unwrap global sentinel; pass per-executor dicts as-is
-                    unwrapped: dict[str, Any] = resolved.get(GLOBAL_KWARGS_KEY, resolved)  # type: ignore
-                    if key == "function_invocation_kwargs":
-                        fi_kwargs = unwrapped  # type: ignore
-                    else:
-                        ci_kwargs = unwrapped  # type: ignore
-
-            # Run the sub-workflow and collect all events, passing parent kwargs
-            result = await self.workflow.run(
-                input_data,
-                function_invocation_kwargs=fi_kwargs,  # type: ignore
-                client_kwargs=ci_kwargs,  # type: ignore
+        # The sub-workflow is a single shared instance. If it still has pending request_info events
+        # from an unfinished request/response cycle, a new input advances its shared state and can
+        # interfere with that cycle - a response arriving later may apply to a sub-workflow that has
+        # moved on. We allow it (the sub-workflow may be stateless) but warn so the risk is visible.
+        pending_requests = await self.workflow._runner_context.get_pending_request_info_events()  # pyright: ignore[reportPrivateUsage]
+        if pending_requests:
+            logger.warning(
+                f"WorkflowExecutor {self.id} received a new input message while its sub-workflow "
+                f"({self.workflow.id}) still has {len(pending_requests)} pending request(s) from an "
+                f"unfinished request/response cycle. The sub-workflow is a single shared instance, so the "
+                f"new input advances shared state and can interfere with the in-flight cycle. Ensure the "
+                f"sub-workflow is stateless, or complete the pending cycle before sending new input."
             )
 
-            logger.debug(
-                f"WorkflowExecutor {self.id} sub-workflow {self.workflow.id} "
-                f"execution {execution_id} completed with {len(result)} events"
-            )
+        logger.debug(f"WorkflowExecutor {self.id} starting sub-workflow {self.workflow.id}")
+
+        # Get kwargs from parent workflow's State to propagate to subworkflow
+        parent_kwargs: dict[str, Any] = ctx.get_state(WORKFLOW_RUN_KWARGS_KEY, {})
+
+        # Extract invocation kwargs recognised by Workflow.run()
+        # The state stores resolved format (with __global__ wrapper for global kwargs).
+        # Unwrap __global__ before passing to the subworkflow so it gets re-resolved
+        # against the subworkflow's own executor IDs.
+        fi_kwargs: dict[str, Any] | None = None
+        ci_kwargs: dict[str, Any] | None = None
+        for key in ("function_invocation_kwargs", "client_kwargs"):
+            resolved = parent_kwargs.get(key)
+            if isinstance(resolved, dict):
+                # Unwrap global sentinel; pass per-executor dicts as-is
+                unwrapped: dict[str, Any] = resolved.get(GLOBAL_KWARGS_KEY, resolved)  # type: ignore
+                if key == "function_invocation_kwargs":
+                    fi_kwargs = unwrapped  # type: ignore
+                else:
+                    ci_kwargs = unwrapped  # type: ignore
+
+        # Run the sub-workflow and collect all events, passing parent kwargs
+        result = await self.workflow.run(
+            input_data,
+            function_invocation_kwargs=fi_kwargs,  # type: ignore
+            client_kwargs=ci_kwargs,  # type: ignore
+        )
+
+        logger.debug(f"WorkflowExecutor {self.id} sub-workflow {self.workflow.id} completed with {len(result)} events")
 
-            # Process the workflow result using shared logic
-            await self._process_workflow_result(result, execution_context, ctx)
-        finally:
-            # Clean up execution context if it's completed (no pending requests)
-            if execution_id in self._execution_contexts:
-                exec_ctx = self._execution_contexts[execution_id]
-                if not exec_ctx.pending_requests:
-                    del self._execution_contexts[execution_id]
+        # Process the workflow result using shared logic
+        await self._process_workflow_result(result, ctx)
 
     @handler
     async def handle_message_wrapped_request_response(
@@ -433,8 +410,8 @@ async def handle_message_wrapped_request_response(
     ) -> None:
         """Handle response from parent for a forwarded request.
 
-        This handler accumulates responses and only resumes the sub-workflow
-        when all expected responses have been received for that execution.
+        Forwards the response to the sub-workflow, which resumes and validates it against its
+        own pending requests.
 
         Args:
             response: The response to a previous request.
@@ -474,61 +451,34 @@ async def handle_propagated_request_response(
     async def on_checkpoint_save(self) -> dict[str, Any]:
         """Get the current state of the WorkflowExecutor for checkpointing purposes."""
         return {
-            "execution_contexts": {
-                execution_id: execution_context for execution_id, execution_context in self._execution_contexts.items()
-            },
-            "request_to_execution": dict(self._request_to_execution),
+            # The sub-workflow's own checkpoint carries everything needed to resume: shared state,
+            # executor snapshots, in-flight messages, and pending request_info events. The
+            # WorkflowExecutor keeps no separate request/response bookkeeping of its own. The
+            # sub-workflow is quiescent here: it ran to idle within this parent superstep before
+            # the parent checkpoints.
+            "sub_workflow_checkpoint": await self.workflow._runner.create_checkpoint_object(),  # pyright: ignore[reportPrivateUsage]
         }
 
     @override
     async def on_checkpoint_restore(self, state: dict[str, Any]) -> None:
         """Restore the WorkflowExecutor state from a checkpoint snapshot."""
-        # Validate the state contains the right keys
-        if "execution_contexts" not in state:
-            raise KeyError("Missing 'execution_contexts' in WorkflowExecutor state.")
-        if "request_to_execution" not in state:
-            raise KeyError("Missing 'request_to_execution' in WorkflowExecutor state.")
-
-        # Validate the execution contexts stored in the state have the right keys and values
-        execution_contexts: dict[str, ExecutionContext] | None = None
-        try:
-            execution_contexts = {
-                key: decode_checkpoint_value(value) for key, value in state["execution_contexts"].items()
-            }
-        except Exception as ex:
-            raise RuntimeError("Failed to deserialize execution context.") from ex
-
-        if not all(
-            isinstance(key, str) and isinstance(value, ExecutionContext) for key, value in execution_contexts.items()
-        ):
-            raise ValueError("Execution contexts must have 'str' as key and 'ExecutionContext' as value.")
-        if not all(key == value.execution_id for key, value in execution_contexts.items()):
-            raise ValueError("Execution contexts must have matching keys and IDs.")
-
-        # Validate the request_to_execution map contain the right data
-        request_to_execution = state["request_to_execution"]
-        if not all(isinstance(key, str) and isinstance(value, str) for key, value in request_to_execution.items()):
-            raise ValueError("Request to execution map must have 'str' as key and 'str' as value.")
-        if not all(value in execution_contexts for value in request_to_execution.values()):
-            raise ValueError(
-                "'request_to_execution` contains unknown execution ID that is not part of the execution contexts."
-            )
+        # The storage backend fully materializes the checkpoint on load, checkpointed data arrives as live objects.
+        sub_workflow_checkpoint = state.get("sub_workflow_checkpoint")
+        if sub_workflow_checkpoint is not None:
+            await self.workflow._runner.restore_from_checkpoint_object(sub_workflow_checkpoint)  # pyright: ignore[reportPrivateUsage]
+            return
 
-        self._execution_contexts = execution_contexts
-        self._request_to_execution = request_to_execution
-
-        # Add the `request_info_event`s back to the sub workflow.
-        # This is only a temporary solution to rehydrate the sub workflow with the requests.
-        # The proper way would be to rehydrate the workflow from a checkpoint on a Workflow
-        # API instead of the '_runner_context' object that should be hidden. And the sub workflow
-        # should be rehydrated from a checkpoint object instead of from a subset of the state.
-        # TODO(@taochen): Issue #1614 - how to handle the case when the parent workflow has checkpointing
-        # set up but not the sub workflow?
-        request_info_events = [
-            request_info_event
-            for execution_context in self._execution_contexts.values()
-            for request_info_event in execution_context.pending_requests.values()
-        ]
+        # Backward-compatibility fallback for checkpoints written before the sub-workflow checkpoint
+        # was embedded. Those stored per-execution bookkeeping; recover only the pending
+        # request_info events so the sub-workflow re-emits its pending requests. The sub-workflow's
+        # deeper executor/shared state cannot be restored from these older checkpoints.
+        legacy_execution_contexts = state.get("execution_contexts")
+        if not legacy_execution_contexts:
+            return
+        request_info_events: list[WorkflowEvent[Any]] = []
+        for execution_context in legacy_execution_contexts.values():
+            if isinstance(execution_context, ExecutionContext):
+                request_info_events.extend(execution_context.pending_requests.values())
         await asyncio.gather(*[
             self.workflow._runner_context.add_request_info_event(event)  # pyright: ignore[reportPrivateUsage]
             for event in request_info_events
@@ -537,7 +487,6 @@ async def on_checkpoint_restore(self, state: dict[str, Any]) -> None:
     async def _process_workflow_result(
         self,
         result: WorkflowRunResult,
-        execution_context: ExecutionContext,
         ctx: WorkflowContext[Any],
     ) -> None:
         """Process the result from a workflow execution.
@@ -547,7 +496,6 @@ async def _process_workflow_result(
 
         Args:
             result: The workflow execution result.
-            execution_context: The execution context for this sub-workflow run.
             ctx: The workflow context.
         """
         # Collect all events from the workflow
@@ -586,10 +534,6 @@ async def _forward_intermediate_output(output: Any) -> None:
         for event in request_info_events:
             request_id = event.request_id
             response_type = event.response_type
-            # Track the pending request in execution context
-            execution_context.pending_requests[request_id] = event
-            # Map request to execution for response routing
-            self._request_to_execution[request_id] = execution_context.execution_id
             if self._propagate_request:
                 # In a workflow where the parent workflow does not handle the request, the request
                 # should be propagated via the `request_info` mechanism to an external source. And
@@ -600,9 +544,6 @@ async def _forward_intermediate_output(output: Any) -> None:
                 # request and handle it directly, a message should be sent.
                 await ctx.send_message(SubWorkflowRequestMessage(source_event=event, executor_id=self.id))
 
-        # Update expected response count for this execution
-        execution_context.expected_response_count = len(request_info_events)
-
         # Handle final state
         if workflow_run_state == WorkflowRunState.FAILED:
             # Find the failed event (type='failed').
@@ -621,26 +562,18 @@ async def _forward_intermediate_output(output: Any) -> None:
                 await ctx.add_event(error_event)
         elif workflow_run_state == WorkflowRunState.IDLE:
             # Sub-workflow is idle - nothing more to do now
-            logger.debug(
-                f"Sub-workflow {self.workflow.id} is idle with {len(self._execution_contexts)} active executions"
-            )
+            logger.debug(f"Sub-workflow {self.workflow.id} is idle")
         elif workflow_run_state == WorkflowRunState.CANCELLED:
             # Sub-workflow was cancelled - treat as completion
-            logger.debug(
-                f"Sub-workflow {self.workflow.id} was cancelled with {len(self._execution_contexts)} active executions"
-            )
+            logger.debug(f"Sub-workflow {self.workflow.id} was cancelled")
         elif workflow_run_state == WorkflowRunState.IN_PROGRESS_PENDING_REQUESTS:
             # Sub-workflow is still running with pending requests
             logger.debug(
-                f"Sub-workflow {self.workflow.id} is still in progress with {len(request_info_events)} "
-                f"pending requests with {len(self._execution_contexts)} active executions"
+                f"Sub-workflow {self.workflow.id} is still in progress with {len(request_info_events)} pending requests"
             )
         elif workflow_run_state == WorkflowRunState.IDLE_WITH_PENDING_REQUESTS:
             # Sub-workflow is idle but has pending requests
-            logger.debug(
-                f"Sub-workflow {self.workflow.id} is idle with pending requests: "
-                f"{len(request_info_events)} with {len(self._execution_contexts)} active executions"
-            )
+            logger.debug(f"Sub-workflow {self.workflow.id} is idle with pending requests: {len(request_info_events)}")
         else:
             raise RuntimeError(f"Unexpected workflow run state: {workflow_run_state}")
 
@@ -650,48 +583,17 @@ async def _handle_response(
         response: Any,
         ctx: WorkflowContext[Any],
     ) -> None:
-        execution_id = self._request_to_execution.get(request_id)
-        if not execution_id or execution_id not in self._execution_contexts:
-            logger.warning(
-                f"WorkflowExecutor {self.id} received response for unknown request_id: {request_id}. "
-                "This response will be ignored."
-            )
-            return
-
-        execution_context = self._execution_contexts[execution_id]
-
-        # Check if we have this pending request in the execution context
-        if request_id not in execution_context.pending_requests:
+        # The sub-workflow is the source of truth for what it is awaiting. Validate the response
+        # against its pending requests and ignore anything unknown or already handled.
+        pending_requests = await self.workflow._runner_context.get_pending_request_info_events()  # pyright: ignore[reportPrivateUsage]
+        if request_id not in pending_requests:
             logger.warning(
-                f"WorkflowExecutor {self.id} received response for unknown request_id: "
-                f"{request_id} in execution {execution_id}, ignoring"
+                f"WorkflowExecutor {self.id} received a response for an unknown or already-handled "
+                f"request_id: {request_id}. This response will be ignored."
             )
             return
 
-        # Remove the request from pending list and request mapping
-        execution_context.pending_requests.pop(request_id, None)
-        self._request_to_execution.pop(request_id, None)
-
-        # Accumulate the response in this execution's context
-        execution_context.collected_responses[request_id] = response
-        # Check if we have all expected responses for this execution
-        if len(execution_context.collected_responses) < execution_context.expected_response_count:
-            logger.debug(
-                f"WorkflowExecutor {self.id} execution {execution_id} waiting for more responses: "
-                f"{len(execution_context.collected_responses)}/{execution_context.expected_response_count} received"
-            )
-            return  # Wait for more responses
-
-        # Send all collected responses to the sub-workflow
-        responses_to_send = dict(execution_context.collected_responses)
-        execution_context.collected_responses.clear()  # Clear for next batch
-
-        try:
-            # Resume the sub-workflow with all collected responses
-            result = await self.workflow.run(responses=responses_to_send)
-            # Process the workflow result using shared logic
-            await self._process_workflow_result(result, execution_context, ctx)
-        finally:
-            # Clean up execution context if it's completed (no pending requests)
-            if not execution_context.pending_requests:
-                del self._execution_contexts[execution_id]
+        # Forward the response to the sub-workflow, which resumes and validates it against its own
+        # pending requests, then process whatever the sub-workflow produces.
+        result = await self.workflow.run(responses={request_id: response})
+        await self._process_workflow_result(result, ctx)
diff --git a/python/packages/core/tests/workflow/test_functional_workflow.py b/python/packages/core/tests/workflow/test_functional_workflow.py
index 4ae4069061c..c4313e4f4e9 100644
--- a/python/packages/core/tests/workflow/test_functional_workflow.py
+++ b/python/packages/core/tests/workflow/test_functional_workflow.py
@@ -257,6 +257,43 @@ async def review_wf(doc: str, ctx: RunContext) -> str:
         assert outputs == ["Final: Looks great!"]
         assert result2.get_final_state() == WorkflowRunState.IDLE
 
+    async def test_fresh_message_while_pending_requests_warns(self, caplog: pytest.LogCaptureFixture) -> None:
+        """A fresh message while request_info events are pending is allowed but logs a warning."""
+
+        @workflow
+        async def review_wf(doc: str, ctx: RunContext) -> str:
+            feedback = await ctx.request_info({"draft": doc}, response_type=str, request_id="req1")
+            return f"Final: {feedback}"
+
+        result1 = await review_wf.run("my doc")
+        assert result1.get_final_state() == WorkflowRunState.IDLE_WITH_PENDING_REQUESTS
+
+        # Starting fresh input while a request is pending does not abandon it, but advances
+        # workflow state so a later response may apply to a moved-on workflow -> warn (but proceed).
+        with caplog.at_level(logging.WARNING):
+            await review_wf.run("another doc")
+
+        assert "request_info event(s) are still pending" in caplog.text
+        assert "a fresh message" in caplog.text
+
+    async def test_responses_while_pending_requests_does_not_warn(self, caplog: pytest.LogCaptureFixture) -> None:
+        """Delivering responses is the normal completion path and must not warn."""
+
+        @workflow
+        async def review_wf(doc: str, ctx: RunContext) -> str:
+            feedback = await ctx.request_info({"draft": doc}, response_type=str, request_id="req1")
+            return f"Final: {feedback}"
+
+        result1 = await review_wf.run("my doc")
+        assert result1.get_final_state() == WorkflowRunState.IDLE_WITH_PENDING_REQUESTS
+
+        caplog.clear()
+        with caplog.at_level(logging.WARNING):
+            result2 = await review_wf.run(responses={"req1": "Looks great!"})
+
+        assert result2.get_final_state() == WorkflowRunState.IDLE
+        assert "still pending" not in caplog.text
+
     async def test_untyped_ctx_parameter(self):
         """ctx is injected by parameter name even without a RunContext annotation."""
 
diff --git a/python/packages/core/tests/workflow/test_runner.py b/python/packages/core/tests/workflow/test_runner.py
index 9c88febf8c4..8133c3080f5 100644
--- a/python/packages/core/tests/workflow/test_runner.py
+++ b/python/packages/core/tests/workflow/test_runner.py
@@ -512,6 +512,106 @@ async def test_runner_reset_iteration_count():
     assert runner._iteration == 0  # pyright: ignore[reportPrivateUsage]
 
 
+async def test_runner_capture_and_restore_checkpoint_object_roundtrip():
+    """create_checkpoint_object() then restore_from_checkpoint_object() must roundtrip.
+
+    Shared state and executor snapshots are captured into an in-memory ``WorkflowCheckpoint``
+    and restored from it without any storage backend (the path the parent WorkflowExecutor
+    uses to checkpoint a nested sub-workflow).
+    """
+
+    class CounterExecutor(Executor):
+        def __init__(self, id: str) -> None:
+            super().__init__(id=id)
+            self.count = 0
+
+        @handler
+        async def handle(self, message: MockMessage, ctx: WorkflowContext[Any, int]) -> None:
+            self.count += message.data
+
+        async def on_checkpoint_save(self) -> dict[str, Any]:
+            return {"count": self.count}
+
+        async def on_checkpoint_restore(self, state: dict[str, Any]) -> None:
+            self.count = int(state.get("count", 0))
+
+    executor = CounterExecutor(id="counter")
+    state = State()
+    ctx = InProcRunnerContext()
+    runner = Runner([], {executor.id: executor}, state, ctx, "test_name", graph_signature_hash="test_hash")
+
+    # Establish some state to capture.
+    executor.count = 7
+    state.set("shared_key", "shared_value")
+    state.commit()
+
+    checkpoint = await runner.create_checkpoint_object()
+    assert checkpoint.graph_signature_hash == "test_hash"
+
+    # Mutate after capture; restoring must roll back to the captured snapshot.
+    executor.count = 999
+    state.set("shared_key", "mutated")
+    state.commit()
+
+    await runner.restore_from_checkpoint_object(checkpoint)
+
+    assert executor.count == 7
+    assert state.get("shared_key") == "shared_value"
+    assert runner._resumed_from_checkpoint is True  # pyright: ignore[reportPrivateUsage]
+
+
+async def test_runner_create_checkpoint_object_includes_in_flight_messages():
+    """create_checkpoint_object() must snapshot in-flight messages non-destructively."""
+    executor = MockExecutor(id="executor_a")
+    state = State()
+    ctx = InProcRunnerContext()
+    runner = Runner([], {executor.id: executor}, state, ctx, "test_name", graph_signature_hash="test_hash")
+
+    await ctx.send_message(WorkflowMessage(data=MockMessage(data=1), source_id="START"))
+
+    checkpoint = await runner.create_checkpoint_object()
+
+    # The in-flight message is captured in the snapshot ...
+    assert list(checkpoint.messages.keys()) == ["START"]
+    assert len(checkpoint.messages["START"]) == 1
+    # ... without draining it from the runner (capture is non-destructive).
+    assert await ctx.has_messages() is True
+
+
+async def test_runner_create_checkpoint_object_advances_previous_checkpoint_id():
+    """create_checkpoint_object() must advance _previous_checkpoint_id so a later capture chains to it."""
+    executor = MockExecutor(id="executor_a")
+    state = State()
+    ctx = InProcRunnerContext()
+    runner = Runner([], {executor.id: executor}, state, ctx, "test_name", graph_signature_hash="test_hash")
+
+    # Pre-condition: nothing captured yet, so there is no parent to chain back to.
+    assert runner._previous_checkpoint_id is None  # pyright: ignore[reportPrivateUsage]
+
+    first = await runner.create_checkpoint_object()
+
+    # Capturing advances the tracked checkpoint id to the newly-created checkpoint ...
+    assert runner._previous_checkpoint_id == first.checkpoint_id  # pyright: ignore[reportPrivateUsage]
+    # ... and a fresh capture begins a new lineage with no parent.
+    assert first.previous_checkpoint_id is None
+
+    second = await runner.create_checkpoint_object()
+
+    # The tracked id advances again ...
+    assert runner._previous_checkpoint_id == second.checkpoint_id  # pyright: ignore[reportPrivateUsage]
+    # ... and the second checkpoint chains back to the first.
+    assert second.previous_checkpoint_id == first.checkpoint_id
+
+
+async def test_runner_restore_from_checkpoint_object_rejects_graph_mismatch():
+    """restore_from_checkpoint_object() must reject a checkpoint from a different graph."""
+    runner = Runner([], {}, State(), InProcRunnerContext(), "test_name", graph_signature_hash="hash-a")
+
+    foreign = WorkflowCheckpoint(workflow_name="test_name", graph_signature_hash="hash-b")
+    with pytest.raises(WorkflowCheckpointException, match="Workflow graph has changed"):
+        await runner.restore_from_checkpoint_object(foreign)
+
+
 class CheckpointingContext(InProcRunnerContext):
     """A context that supports checkpointing for testing."""
 
diff --git a/python/packages/core/tests/workflow/test_sub_workflow.py b/python/packages/core/tests/workflow/test_sub_workflow.py
index f5e2e8fa26f..974a1a4e91d 100644
--- a/python/packages/core/tests/workflow/test_sub_workflow.py
+++ b/python/packages/core/tests/workflow/test_sub_workflow.py
@@ -1,9 +1,11 @@
 # Copyright (c) Microsoft. All rights reserved.
 
+import logging
 from dataclasses import dataclass, field
 from typing import Any
 from uuid import uuid4
 
+import pytest
 from typing_extensions import Never
 
 from agent_framework import (
@@ -12,9 +14,11 @@
     SubWorkflowResponseMessage,
     Workflow,
     WorkflowBuilder,
+    WorkflowCheckpointException,
     WorkflowContext,
     WorkflowEvent,
     WorkflowExecutor,
+    WorkflowRunState,
     handler,
     response_handler,
 )
@@ -465,6 +469,62 @@ async def collect_result(self, result: ValidationResult, ctx: WorkflowContext) -
     # (This is implicitly tested by the fact that we got correct results for all emails)
 
 
+async def test_sub_workflow_warns_on_overlapping_execution(caplog: pytest.LogCaptureFixture) -> None:
+    """A new input while a prior sub-workflow execution is awaiting responses logs a warning.
+
+    Overlapping executions share one sub-workflow instance and its state, so WorkflowExecutor
+    allows the new execution but warns that it is only safe when the wrapped workflow is stateless.
+    """
+
+    class TwoInputParent(Executor):
+        def __init__(self) -> None:
+            super().__init__(id="two_input_parent")
+            self._pending: dict[str, SubWorkflowRequestMessage] = {}
+
+        @handler
+        async def start(self, emails: list[str], ctx: WorkflowContext[EmailValidationRequest]) -> None:
+            for email in emails:
+                await ctx.send_message(EmailValidationRequest(email=email))
+
+        @handler
+        async def handle_domain_request(
+            self,
+            sub_workflow_request: SubWorkflowRequestMessage,
+            ctx: WorkflowContext[SubWorkflowResponseMessage],
+        ) -> None:
+            domain_request = sub_workflow_request.source_event.data
+            assert isinstance(domain_request, DomainCheckRequest)
+            self._pending[domain_request.id] = sub_workflow_request
+            await ctx.request_info(domain_request, bool)
+
+        @handler
+        async def collect(self, result: ValidationResult, ctx: WorkflowContext) -> None: ...
+
+    parent = TwoInputParent()
+    workflow_executor = WorkflowExecutor(create_email_validation_workflow(), "email_workflow")
+    main_workflow = (
+        WorkflowBuilder(start_executor=parent)
+        .add_edge(parent, workflow_executor)
+        .add_edge(workflow_executor, parent)
+        .build()
+    )
+
+    with caplog.at_level(logging.WARNING, logger="agent_framework._workflows._workflow_executor"):
+        result = await main_workflow.run(["a@domain1.com", "b@domain2.com"])
+
+    # Two inputs are delivered to the same WorkflowExecutor in one superstep: the second execution
+    # starts while the sub-workflow still has a pending request, producing exactly one overlap
+    # warning from the WorkflowExecutor. (The substring is unique to the WorkflowExecutor warning so
+    # it is not confused with the core Workflow.run pending-request warning.)
+    assert len(result.get_request_info_events()) == 2
+    overlap_warnings = [
+        record
+        for record in caplog.records
+        if record.levelno == logging.WARNING and "new input message while its sub-workflow" in record.getMessage()
+    ]
+    assert len(overlap_warnings) == 1
+
+
 # region Checkpoint-related message types and executors for sub-workflow tests
 
 
@@ -571,6 +631,29 @@ def _build_checkpoint_test_workflow(storage: InMemoryCheckpointStorage) -> Workf
     )
 
 
+def test_workflow_executor_rejects_sub_workflow_with_checkpointing() -> None:
+    """WorkflowExecutor must reject a sub-workflow that already has checkpointing enabled.
+
+    Sub-workflow checkpointing is managed by the parent workflow, so wrapping a sub-workflow
+    that carries its own checkpoint storage is a configuration error and must fail fast at
+    construction time.
+    """
+
+    class _Passthrough(Executor):
+        @handler
+        async def run(self, value: str, ctx: WorkflowContext[Any, str]) -> None:
+            await ctx.yield_output(value)
+
+    storage = InMemoryCheckpointStorage()
+    sub_workflow = WorkflowBuilder(
+        start_executor=_Passthrough(id="passthrough"),
+        checkpoint_storage=storage,
+    ).build()
+
+    with pytest.raises(WorkflowCheckpointException, match="must not have checkpointing enabled"):
+        WorkflowExecutor(sub_workflow, id="sub_workflow_executor")
+
+
 async def test_sub_workflow_checkpoint_restore_no_duplicate_requests() -> None:
     """Test that resuming a sub-workflow from checkpoint does not emit duplicate requests.
 
@@ -619,6 +702,59 @@ async def test_sub_workflow_checkpoint_restore_no_duplicate_requests() -> None:
     assert request_events[0].data.prompt == "Second request"
 
 
+async def test_sub_workflow_checkpoint_restore_preserves_sub_workflow_state() -> None:
+    """Resuming a sub-workflow mid-progress must restore its internal executor state.
+
+    Regression guard for the issue where only the WorkflowExecutor's bookkeeping (pending
+    requests) was checkpointed, so a sub-workflow executor that accumulates state across
+    multiple request/response cycles (here ``TwoStepSubWorkflowExecutor._responses``) lost
+    that state on resume. With the sub-workflow's own checkpoint embedded in the parent
+    checkpoint, the second response now completes the two-step flow instead of triggering a
+    spurious third request.
+    """
+    storage = InMemoryCheckpointStorage()
+
+    # Step 1: run until the first request.
+    workflow1 = _build_checkpoint_test_workflow(storage)
+    first_request_id: str | None = None
+    async for event in workflow1.run("test_value", stream=True):
+        if event.type == "request_info":
+            first_request_id = event.request_id
+    assert first_request_id is not None
+
+    # Step 2: answer the first request so the sub-workflow accumulates internal state
+    # (``_responses == ["first_answer"]``) and emits the second request. This mid-progress
+    # point is what we checkpoint and resume from - the case the no-duplicate test (which
+    # checkpoints at the first request, before any state accrues) does not cover.
+    second_request_id: str | None = None
+    async for event in workflow1.run(stream=True, responses={first_request_id: "first_answer"}):
+        if event.type == "request_info":
+            second_request_id = event.request_id
+    assert second_request_id is not None
+
+    # Resume from the latest checkpoint (captured after the second request was made).
+    checkpoints = await storage.list_checkpoints(workflow_name=workflow1.name)
+    checkpoint_id = max(checkpoints, key=lambda cp: cp.iteration_count).checkpoint_id
+
+    workflow2 = _build_checkpoint_test_workflow(storage)
+    resumed_second_request_id: str | None = None
+    async for event in workflow2.run(checkpoint_id=checkpoint_id, stream=True):
+        if event.type == "request_info":
+            resumed_second_request_id = event.request_id
+    assert resumed_second_request_id is not None
+    assert resumed_second_request_id == second_request_id
+
+    # Step 3: answer the second request. With the sub-workflow's state restored, the two-step
+    # executor completes instead of emitting a spurious third request. If the internal state
+    # were lost, the second answer would be treated as a first answer and a third request
+    # ("Second request") would be emitted.
+    result = await workflow2.run(responses={resumed_second_request_id: "second_answer"})
+    assert result.get_request_info_events() == [], (
+        "Sub-workflow internal state was lost on resume: a spurious extra request was emitted"
+    )
+    assert result.get_final_state() == WorkflowRunState.IDLE
+
+
 async def test_sub_workflow_intermediate_outputs_propagate_to_parent() -> None:
     """A child workflow's intermediate emissions must bubble up through the parent.
 
diff --git a/python/packages/core/tests/workflow/test_workflow.py b/python/packages/core/tests/workflow/test_workflow.py
index c77688c03c2..1d7087b6fed 100644
--- a/python/packages/core/tests/workflow/test_workflow.py
+++ b/python/packages/core/tests/workflow/test_workflow.py
@@ -2,6 +2,7 @@
 
 import asyncio
 import gc
+import logging
 import tempfile
 from collections.abc import AsyncIterable, Awaitable, Sequence
 from dataclasses import dataclass, field
@@ -109,6 +110,38 @@ async def mock_handler_b(
             await ctx.send_message(NumberMessage(data=data))
 
 
+async def test_fresh_message_while_pending_advances_state_without_abandoning_requests(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """A fresh message while a request is pending is allowed but hazardous.
+
+    A fresh ``message`` does NOT abandon the pending request - it can still be answered
+    later - but the new run advances executor state, so a response for the earlier request
+    applies to a workflow that has moved on. The run is allowed and a warning is emitted.
+    """
+    executor = MockExecutorRequestApproval(id="approver")
+    workflow = WorkflowBuilder(start_executor=executor).build()
+
+    # Turn 1: request approval for data=1 -> workflow idles with a pending request.
+    result1 = await workflow.run(NumberMessage(data=1))
+    assert result1.get_final_state() == WorkflowRunState.IDLE_WITH_PENDING_REQUESTS
+    original_request_id = result1.get_request_info_events()[0].request_id
+
+    # Turn 2: a fresh message for data=2 while the first request is still pending. This is
+    # allowed but warns, and advances the executor's stored state from 1 to 2.
+    with caplog.at_level(logging.WARNING):
+        result2 = await workflow.run(NumberMessage(data=2))
+    assert "request_info event(s) are still pending" in caplog.text
+    assert "a fresh message" in caplog.text
+    assert result2.get_final_state() == WorkflowRunState.IDLE_WITH_PENDING_REQUESTS
+
+    # Turn 3: the ORIGINAL request is still answerable, proving the fresh message did not
+    # abandon it. But because the executor state moved on to 2, the response applies to the
+    # moved-on state and yields 2, not the original 1.
+    result3 = await workflow.run(responses={original_request_id: ApprovalMessage(approved=True)})
+    assert result3.get_outputs() == [2]
+
+
 async def test_workflow_run_streaming() -> None:
     """Test the workflow run stream."""
     executor_a = IncrementExecutor(id="executor_a")
diff --git a/python/packages/durabletask/agent_framework_durabletask/_workflows/runner_context.py b/python/packages/durabletask/agent_framework_durabletask/_workflows/runner_context.py
index 1851339bb37..d59d10bcce3 100644
--- a/python/packages/durabletask/agent_framework_durabletask/_workflows/runner_context.py
+++ b/python/packages/durabletask/agent_framework_durabletask/_workflows/runner_context.py
@@ -98,6 +98,17 @@ async def create_checkpoint(
     ) -> str:
         raise NotImplementedError("Checkpointing is not supported in activity context")
 
+    async def create_checkpoint_object(
+        self,
+        workflow_name: str,
+        graph_signature_hash: str,
+        state: State,
+        previous_checkpoint_id: str | None,
+        iteration_count: int,
+        metadata: dict[str, Any] | None = None,
+    ) -> WorkflowCheckpoint:
+        raise NotImplementedError("Checkpointing is not supported in activity context")
+
     async def load_checkpoint(self, checkpoint_id: str) -> WorkflowCheckpoint | None:
         raise NotImplementedError("Checkpointing is not supported in activity context")