Python: Populate WorkflowAgent response for empty streams

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

@@ -424,8 +424,7 @@ async def _run_stream_impl(
 
         # Build the final response from collected updates so after_run providers
         # (e.g. InMemoryHistoryProvider) can persist the response messages.
-        if all_updates:
-            session_context._response = AgentResponse.from_updates(all_updates)  # type: ignore[assignment]
+        session_context._response = AgentResponse.from_updates(all_updates)  # type: ignore[assignment]
 
         await self._run_after_providers(session=provider_session, context=session_context)
 

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

@@ -14,13 +14,15 @@
     AgentResponseUpdate,
     AgentSession,
     Content,
+    ContextProvider,
     Executor,
     FinishReason,
     HistoryProvider,
     InMemoryHistoryProvider,
     Message,
     ResponseStream,
     ServiceSessionId,
+    SessionContext,
     SupportsAgentRun,
     UsageDetails,
     WorkflowAgent,
@@ -1220,6 +1222,44 @@ async def test_multi_turn_session_stores_responses_streaming(self) -> None:
         assert roles[1] == "assistant"
         assert roles[2] == "user"
 
+    async def test_zero_output_stream_populates_response_before_after_run(self) -> None:
+        """After-run providers receive a response even when a workflow emits no output."""
+
+        @executor
+        async def silent(messages: list[Message], ctx: WorkflowContext[Any, str]) -> None:
+            assert messages[0].text == "hello"
+            del ctx
+
+        captured: list[AgentResponse | None] = []
+
+        class CapturingProvider(ContextProvider):
+            async def after_run(
+                self,
+                *,
+                agent: SupportsAgentRun,
+                session: AgentSession,
+                context: SessionContext,
+                state: dict[str, Any],
+            ) -> None:
+                captured.append(context.response)
+
+        workflow = WorkflowBuilder(start_executor=silent, output_from=[silent]).build()
+        agent = workflow.as_agent(context_providers=[CapturingProvider("capture")])
+
+        non_stream_response = await agent.run("hello")
+        stream = agent.run("hello", stream=True)
+        updates = [update async for update in stream]
+        stream_response = await stream.get_final_response()
+
+        assert updates == []
+        assert non_stream_response.messages == []
+        assert stream_response.messages == []
+        assert len(captured) == 2
+        assert isinstance(captured[0], AgentResponse)
+        assert captured[0].messages == []
+        assert isinstance(captured[1], AgentResponse)
+        assert captured[1].messages == []
+
     async def test_multi_turn_session_roundtrip_serialization(self) -> None:
         """Test that session can be serialized/deserialized and multi-turn still works."""
         capturing_executor = ConversationHistoryCapturingExecutor(id="roundtrip_test", streaming=False)