Python: fix(orchestration): add ensure_trailing_user_turn to SequentialBuilder for providers that return empty text on assistant-ending prompts

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

@@ -314,6 +314,7 @@ def __init__(
         id: str | None = None,
         context_mode: Literal["full", "last_agent", "custom"] | None = None,
         context_filter: Callable[[list[Message]], list[Message]] | None = None,
+        ensure_trailing_user_turn: bool = False,
     ):
         """Initialize the executor with a unique identifier.
 
@@ -332,7 +333,12 @@ def __init__(
             context_filter: A function that takes the full conversation (list of Messages) as input and returns
                 a filtered list of Messages to be used as context for the agent run. This is required
                 if context_mode is set to "custom".
+            ensure_trailing_user_turn: If True, appends a synthetic user continuation message
+                ("please continue.") when the cache ends on an assistant message. Some
+                chat-completions providers return empty text when the prompt ends on
+                ``assistant``. Defaults to False to preserve existing behavior.
         """
+        self._ensure_trailing_user_turn = ensure_trailing_user_turn
         # Prefer provided id; else use agent.name if present; else generate deterministic prefix
         exec_id = id or resolve_agent_id(agent)
         if not exec_id:
@@ -746,14 +752,21 @@ async def _run_agent_and_emit(
         containing incremental updates (streaming mode) or a single output event (type='output')
         containing the complete response (non-streaming mode).
         """
+        invocation_messages = self._build_invocation_messages()
         self._session.state[_WORKFLOW_DEFER_COMPUTER_FUNCTION_RESULTS_KEY] = True
         try:
             if ctx.is_streaming():
                 # Streaming mode: emit incremental updates
-                response = await self._run_agent_streaming(cast(WorkflowContext[Never, AgentResponseUpdate], ctx))
+                response = await self._run_agent_streaming(
+                    cast(WorkflowContext[Never, AgentResponseUpdate], ctx),
+                    messages=invocation_messages,
+                )
             else:
                 # Non-streaming mode: use run() and emit single event
-                response = await self._run_agent(cast(WorkflowContext[Never, AgentResponse], ctx))
+                response = await self._run_agent(
+                    cast(WorkflowContext[Never, AgentResponse], ctx),
+                    messages=invocation_messages,
+                )
         finally:
             self._session.state.pop(_WORKFLOW_DEFER_COMPUTER_FUNCTION_RESULTS_KEY, None)
 
@@ -771,11 +784,39 @@ async def _run_agent_and_emit(
         await ctx.send_message(agent_response)
         self._cache.clear()
 
-    async def _run_agent(self, ctx: WorkflowContext[Never, AgentResponse]) -> AgentResponse | None:
+    def _build_invocation_messages(self) -> list[Message]:
+        """Return an invocation-only copy of the cache with trailing user turn if needed.
+
+        The copy — not ``self._cache`` — is what gets passed to the agent's
+        ``run()`` call, so the synthetic "please continue." message never
+        persists in the cache or ``full_conversation``. Only the invocation
+        that needs the trailing user turn sees it; downstream agents in a
+        multi-agent chain are unaffected.
+        """
+        messages = list(self._cache)
+        if self._ensure_trailing_user_turn and messages and messages[-1].role == "assistant":
+            logger.debug(
+                "AgentExecutor %s: cache ends on an assistant message; appending a synthetic "
+                "continuation turn to the invocation-only copy to avoid empty completions from "
+                "providers that require a trailing user message.",
+                self.id,
+            )
+            messages.append(Message(role="user", contents=["please continue."]))
+        return messages
+
+    async def _run_agent(
+        self,
+        ctx: WorkflowContext[Never, AgentResponse],
+        *,
+        messages: list[Message] | None = None,
+    ) -> AgentResponse | None:
         """Execute the underlying agent in non-streaming mode.
 
         Args:
             ctx: The workflow context for emitting events.
+            messages: Invocation-only message list. If provided, used instead
+                of ``self._cache`` so the synthetic trailing user turn does
+                not persist in the cache.
 
         Returns:
             The complete AgentResponse, or None if waiting for user input.
@@ -785,7 +826,8 @@ async def _run_agent(self, ctx: WorkflowContext[Never, AgentResponse]) -> AgentR
         function_invocation_kwargs, client_kwargs = self._prepare_agent_run_args(raw_run_kwargs, resolved_run_kwargs)
         tools = ctx.get_runtime_tools()
 
-        if not self._cache:
+        cache = messages if messages is not None else self._cache
+        if not cache:
             logger.warning(
                 "AgentExecutor %s: Running agent with empty message cache. "
                 "This could lead to service error for some LLM providers.",
@@ -801,7 +843,7 @@ async def _run_agent(self, ctx: WorkflowContext[Never, AgentResponse]) -> AgentR
         }
         if tools is not None and self._accepts_runtime_tools:
             run_kwargs["tools"] = tools
-        response = await run_agent(self._cache, **run_kwargs)
+        response = await run_agent(cache, **run_kwargs)
 
         # Handle any user input requests
         if response.user_input_requests:
@@ -826,11 +868,19 @@ async def _run_agent(self, ctx: WorkflowContext[Never, AgentResponse]) -> AgentR
         await ctx.yield_output(response)
         return response
 
-    async def _run_agent_streaming(self, ctx: WorkflowContext[Never, AgentResponseUpdate]) -> AgentResponse | None:
+    async def _run_agent_streaming(
+        self,
+        ctx: WorkflowContext[Never, AgentResponseUpdate],
+        *,
+        messages: list[Message] | None = None,
+    ) -> AgentResponse | None:
         """Execute the underlying agent in streaming mode and collect the full response.
 
         Args:
             ctx: The workflow context for emitting events.
+            messages: Invocation-only message list. If provided, used instead
+                of ``self._cache`` so the synthetic trailing user turn does
+                not persist in the cache.
 
         Returns:
             The complete AgentResponse, or None if waiting for user input.
@@ -840,7 +890,8 @@ async def _run_agent_streaming(self, ctx: WorkflowContext[Never, AgentResponseUp
         function_invocation_kwargs, client_kwargs = self._prepare_agent_run_args(raw_run_kwargs, resolved_run_kwargs)
         tools = ctx.get_runtime_tools()
 
-        if not self._cache:
+        cache = messages if messages is not None else self._cache
+        if not cache:
             logger.warning(
                 "AgentExecutor %s: Running agent with empty message cache. "
                 "This could lead to service error for some LLM providers.",
@@ -859,7 +910,7 @@ async def _run_agent_streaming(self, ctx: WorkflowContext[Never, AgentResponseUp
         }
         if tools is not None and self._accepts_runtime_tools:
             run_kwargs["tools"] = tools
-        stream = run_agent_stream(self._cache, **run_kwargs)
+        stream = run_agent_stream(cache, **run_kwargs)
         async for update in stream:
             updates.append(update)
             if any(content.type == "computer_tool_call" and content.user_input_request for content in update.contents):

python/packages/core/tests/workflow/test_agent_executor.py

@@ -576,6 +576,130 @@ def only_user_messages(msgs: list[Message]) -> list[Message]:
     assert seen[0].role == "user" and "hello" in (seen[0].text or "")
 
 
+def test_ensure_trailing_user_turn_defaults_to_false() -> None:
+    """ensure_trailing_user_turn should default to False, preserving existing behavior."""
+    agent = _CountingAgent(id="a", name="A")
+    executor = AgentExecutor(agent)
+    assert executor._ensure_trailing_user_turn is False  # pyright: ignore[reportPrivateUsage]
+
+
+def test_build_invocation_messages_noop_when_flag_disabled() -> None:
+    """When the flag is off, the returned messages match the cache without modification."""
+    agent = _CountingAgent(id="a", name="A")
+    executor = AgentExecutor(agent, ensure_trailing_user_turn=False)
+    executor._cache = [
+        Message(role="user", contents=["go"]),
+        Message(role="assistant", contents=["W-OK"]),
+    ]
+    messages = executor._build_invocation_messages()
+    assert len(messages) == 2
+    assert messages[-1].role == "assistant"
+    assert len(executor._cache) == 2
+
+
+def test_build_invocation_messages_noop_when_cache_not_assistant_ending() -> None:
+    """When the flag is on but the cache already ends on user/tool, nothing is appended."""
+    agent = _CountingAgent(id="a", name="A")
+    executor = AgentExecutor(agent, ensure_trailing_user_turn=True)
+    executor._cache = [
+        Message(role="user", contents=["go"]),
+        Message(role="assistant", contents=["W-OK"]),
+        Message(role="user", contents=["continue"]),
+    ]
+    messages = executor._build_invocation_messages()
+    assert len(messages) == 3
+    assert messages[-1].role == "user"
+    assert len(executor._cache) == 3
+
+
+def test_build_invocation_messages_appends_synthetic_user_turn_when_enabled() -> None:
+    """When the flag is on and cache ends on assistant, a synthetic user turn is
+    appended to the returned copy — but the original cache is not mutated."""
+    agent = _CountingAgent(id="a", name="A")
+    executor = AgentExecutor(agent, ensure_trailing_user_turn=True)
+    executor._cache = [
+        Message(role="user", contents=["go"]),
+        Message(role="assistant", contents=["W-OK"]),
+    ]
+    messages = executor._build_invocation_messages()
+    assert len(messages) == 3
+    assert messages[-1].role == "user"
+    assert "please continue" in (messages[-1].text or "")
+    assert len(executor._cache) == 2
+    assert executor._cache[-1].role == "assistant"
+
+
+def test_build_invocation_messages_noop_when_cache_empty() -> None:
+    """An empty cache must not raise or append anything, even with the flag enabled."""
+    agent = _CountingAgent(id="a", name="A")
+    executor = AgentExecutor(agent, ensure_trailing_user_turn=True)
+    assert executor._cache == []
+    messages = executor._build_invocation_messages()
+    assert messages == []
+    assert executor._cache == []
+
+
+async def test_synthetic_turn_does_not_leak_to_third_agent() -> None:
+    """The synthetic 'please continue.' must not appear in the third agent's
+    context — it is invocation-only and must not persist in full_conversation."""
+    a1 = _MessageCapturingAgent(id="first", name="First", reply_text="first reply")
+    a2 = _MessageCapturingAgent(id="second", name="Second", reply_text="second reply")
+    a3 = _MessageCapturingAgent(id="third", name="Third", reply_text="third reply")
+    exec_a = AgentExecutor(a1, id="exec_a")
+    exec_b = AgentExecutor(a2, id="exec_b", context_mode="full", ensure_trailing_user_turn=True)
+    exec_c = AgentExecutor(a3, id="exec_c", context_mode="full")
+    wf = WorkflowBuilder(start_executor=exec_a).add_edge(exec_a, exec_b).add_edge(exec_b, exec_c).build()
+    async for ev in wf.run("hello", stream=True):
+        if ev.type == "status" and ev.state == WorkflowRunState.IDLE:
+            break
+    seen = a3.last_messages
+    assert len(seen) == 3
+    for msg in seen:
+        assert "please continue" not in (msg.text or "")
+
+
+async def test_from_response_ensure_trailing_user_turn_end_to_end_full_mode() -> None:
+    """chained full mode context should get a synthetic  trailing user turn before the second agent runs,
+    when the flag is enabled."""
+
+    first = _MessageCapturingAgent(id="first", name="First", reply_text="first reply")
+    second = _MessageCapturingAgent(id="second", name="Second", reply_text="second reply")
+
+    exec_a = AgentExecutor(first, id="exec_a")
+    exec_b = AgentExecutor(second, id="exec_b", context_mode="full", ensure_trailing_user_turn=True)
+
+    wf = WorkflowBuilder(start_executor=exec_a).add_edge(exec_a, exec_b).build()
+
+    async for ev in wf.run("hello", stream=True):
+        if ev.type == "status" and ev.state == WorkflowRunState.IDLE:
+            break
+
+    seen = second.last_messages
+    assert len(seen) == 3
+    assert seen[0].role == "user" and "hello" in (seen[0].text or "")
+    assert seen[1].role == "assistant" and "first reply" in (seen[1].text or "")
+    assert seen[2].role == "user"
+
+
+async def test_from_response_without_ensure_trailing_user_turn_still_ends_on_assistant() -> None:
+    """Default (flag off) behavior is unchanged: chained context still ends on assistant."""
+    first = _MessageCapturingAgent(id="first", name="First", reply_text="first reply")
+    second = _MessageCapturingAgent(id="second", name="Second", reply_text="second reply")
+
+    exec_a = AgentExecutor(first, id="exec_a")
+    exec_b = AgentExecutor(second, id="exec_b", context_mode="full")
+
+    wf = WorkflowBuilder(start_executor=exec_a).add_edge(exec_a, exec_b).build()
+
+    async for ev in wf.run("hello", stream=True):
+        if ev.type == "status" and ev.state == WorkflowRunState.IDLE:
+            break
+
+    seen = second.last_messages
+    assert len(seen) == 2
+    assert seen[-1].role == "assistant"
+
+
 async def test_checkpoint_save_does_not_include_context_mode() -> None:
     """on_checkpoint_save should not include context_mode in the saved state."""
     agent = _CountingAgent(id="a", name="A")

python/packages/orchestrations/agent_framework_orchestrations/_orchestration_request_info.py

@@ -199,6 +199,7 @@ def __init__(
         context_mode: Literal["full", "last_agent", "custom"] | None = None,
         *,
         allow_direct_output: bool = False,
+        ensure_trailing_user_turn: bool = False,
     ) -> None:
         """Initialize the AgentApprovalExecutor.
 
@@ -210,8 +211,13 @@ def __init__(
                 downstream participant). Set this when this executor is the workflow's
                 terminator — so the user-approved final response surfaces as a workflow
                 ``output`` event.
+            ensure_trailing_user_turn: If True, appends a synthetic user continuation
+                message when the cache ends on an assistant message. Some chat-completions
+                providers return empty text when the prompt ends on ``assistant``.
+                Defaults to False.
         """
         self._context_mode: Literal["full", "last_agent", "custom"] | None = context_mode
+        self._ensure_trailing_user_turn = ensure_trailing_user_turn
         self._description = agent.description
 
         super().__init__(
@@ -235,6 +241,7 @@ def _build_workflow(self, agent: SupportsAgentRun, *, terminal: bool) -> Workflo
         agent_executor = AgentExecutor(
             agent,
             context_mode=self._context_mode,
+            ensure_trailing_user_turn=self._ensure_trailing_user_turn,
         )
         request_info_cls = _TerminalAgentRequestInfoExecutor if terminal else AgentRequestInfoExecutor
         request_info_executor = request_info_cls(id="agent_request_info_executor")

python/packages/orchestrations/agent_framework_orchestrations/_sequential.py

@@ -105,6 +105,7 @@ def __init__(
         chain_only_agent_responses: bool = False,
         output_from: Sequence[_ParticipantOutputSpecifier] | Literal["all"] | None = cast(Any, UNSET),
         intermediate_output_from: _ParticipantIntermediateOutputSelection = None,
+        ensure_trailing_user_turn: bool = False,
     ) -> None:
         """Initialize the SequentialBuilder.
 
@@ -120,6 +121,11 @@ def __init__(
             intermediate_output_from: Optional participant names or instances whose ``yield_output`` calls
                 surface as workflow ``intermediate`` events. Pass ``"all_other"`` to select every participant
                 not selected by ``output_from``. Unlisted participant outputs are hidden.
+            ensure_trailing_user_turn: If True, whenever a participant would be run with a
+                conversation ending on an assistant message (e.g. chaining into the next agent),
+                a synthetic user continuation message is appended first. Some chat-completions
+                providers return empty text when the prompt ends on ``assistant``. Defaults to
+                False to preserve existing behavior.
         """
         self._name = name or DEFAULT_WORKFLOW_NAME
         self._participants: list[SupportsAgentRun | Executor] = []
@@ -129,6 +135,7 @@ def __init__(
         self._request_info_filter: set[str] | None = None
         self._output_from = _coalesce_output_from(output_from=output_from)
         self._intermediate_output_from = _coerce_intermediate_output_from(intermediate_output_from)
+        self._ensure_trailing_user_turn: bool = ensure_trailing_user_turn
 
         self._set_participants(participants)
 
@@ -223,10 +230,17 @@ def _resolve_participants(self) -> list[Executor]:
                             p,
                             context_mode=context_mode,
                             allow_direct_output=(idx == last_idx),
+                            ensure_trailing_user_turn=self._ensure_trailing_user_turn,
                         )
                     )
                 else:
-                    executors.append(AgentExecutor(p, context_mode=context_mode))
+                    executors.append(
+                        AgentExecutor(
+                            p,
+                            context_mode=context_mode,
+                            ensure_trailing_user_turn=self._ensure_trailing_user_turn,
+                        )
+                    )
             else:
                 raise TypeError(f"Participants must be SupportsAgentRun or Executor instances. Got {type(p).__name__}.")
 

python/packages/orchestrations/tests/test_orchestration_request_info.py

@@ -257,6 +257,38 @@ def test_propagate_request_enabled(self):
 
         assert executor._propagate_request is True  # type: ignore
 
+    def test_ensure_trailing_user_turn_defaults_to_false(self):
+        """ensure_trailing_user_turn should default to False, preserving existing behavior."""
+        agent = _TestAgent(id="test_id", name="test_agent", description="test description")
+        executor = AgentApprovalExecutor(cast(SupportsAgentRun, agent))
+        assert executor._ensure_trailing_user_turn is False  # type: ignore
+
+    def test_ensure_trailing_user_turn_stored_on_executor(self):
+        """ensure_trailing_user_turn=True should be stored on the AgentApprovalExecutor instance."""
+        agent = _TestAgent(id="test_id", name="test_name", description="test description")
+        executor = AgentApprovalExecutor(cast(SupportsAgentRun, agent), ensure_trailing_user_turn=True)
+        assert executor._ensure_trailing_user_turn is True  # type: ignore
+
+    def test_ensure_trailing_user_turn_propagated_to_inner_agent_executor(self):
+        """ensure_trailing_user_turn must be threaded through to the inner AgentExecutor
+        built by _build_workflow, since that inner executor is what actually runs the agent.
+
+        without this propagation, the flag would be accepted
+        by AgentApprovalExecutor but silently dropped, so request-info-wrapped agents
+        would never get the synthetic trailing user turn.
+        """
+        agent = _TestAgent(id="test_id", name="test_agent", description="Test description")
+        executor = AgentApprovalExecutor(cast(SupportsAgentRun, agent), ensure_trailing_user_turn=True)
+        inner_agent_executor = executor.workflow.get_start_executor()
+        assert inner_agent_executor._ensure_trailing_user_turn is True  # type: ignore
+
+    def test_ensure_trailing_user_turn_false_propagated_to_inner_agent_executor(self):
+        """When left at the default, the inner AgentExecutor must also default to False."""
+        agent = _TestAgent(id="test_id", name="test_agent", description="Test description")
+        executor = AgentApprovalExecutor(cast(SupportsAgentRun, agent))
+        inner_agent_executor = executor.workflow.get_start_executor()
+        assert inner_agent_executor._ensure_trailing_user_turn is False  # type: ignore
+
 
 def _build_with_request_info(builder_name: str, agents: list[str]) -> Any:
     from agent_framework.orchestrations import ConcurrentBuilder, GroupChatBuilder, SequentialBuilder