Python: Add CreateConversationExecutor, fix input routing, remove unused handler layer (#4159)

* Fixed declarative deep research sample

* Small fix

* Resolved comment

* Add CreateConversationExecutor, fix input routing, remove unused handler layer

* Address Copilot feedback

* Fix System.ConversationId

---------

Co-authored-by: Dmytro Struk <13853051+dmytrostruk@users.noreply.github.com>

python/packages/declarative/tests/test_graph_coverage.py

@@ -740,6 +740,90 @@ class TestAgentExecutorsCoverage:
         name = executor._get_agent_name(state)
         assert name == "LegacyAgent"
 
+    async def test_agent_executor_get_agent_name_string_expression(self, mock_context, mock_state):
+        """Test agent name extraction from simple string expression."""
+        from unittest.mock import patch
+
+        from agent_framework_declarative._workflows._executors_agents import (
+            InvokeAzureAgentExecutor,
+        )
+
+        action_def = {
+            "kind": "InvokeAzureAgent",
+            "agent": "=Local.SelectedAgent",
+        }
+        executor = InvokeAzureAgentExecutor(action_def)
+
+        state = DeclarativeWorkflowState(mock_state)
+        state.initialize()
+
+        with patch.object(state, "eval_if_expression", return_value="DynamicAgent"):
+            name = executor._get_agent_name(state)
+        assert name == "DynamicAgent"
+
+    async def test_agent_executor_get_agent_name_dict_expression(self, mock_context, mock_state):
+        """Test agent name extraction from nested dict with expression."""
+        from unittest.mock import patch
+
+        from agent_framework_declarative._workflows._executors_agents import (
+            InvokeAzureAgentExecutor,
+        )
+
+        action_def = {
+            "kind": "InvokeAzureAgent",
+            "agent": {"name": "=Local.ManagerResult.next_speaker.answer"},
+        }
+        executor = InvokeAzureAgentExecutor(action_def)
+
+        state = DeclarativeWorkflowState(mock_state)
+        state.initialize()
+
+        with patch.object(state, "eval_if_expression", return_value="WeatherAgent"):
+            name = executor._get_agent_name(state)
+        assert name == "WeatherAgent"
+
+    async def test_agent_executor_get_agent_name_legacy_expression(self, mock_context, mock_state):
+        """Test agent name extraction from legacy agentName with expression."""
+        from unittest.mock import patch
+
+        from agent_framework_declarative._workflows._executors_agents import (
+            InvokeAzureAgentExecutor,
+        )
+
+        action_def = {
+            "kind": "InvokeAzureAgent",
+            "agentName": "=Local.NextAgent",
+        }
+        executor = InvokeAzureAgentExecutor(action_def)
+
+        state = DeclarativeWorkflowState(mock_state)
+        state.initialize()
+
+        with patch.object(state, "eval_if_expression", return_value="ResolvedAgent"):
+            name = executor._get_agent_name(state)
+        assert name == "ResolvedAgent"
+
+    async def test_agent_executor_get_agent_name_expression_returns_none(self, mock_context, mock_state):
+        """Test agent name returns None when expression evaluates to None."""
+        from unittest.mock import patch
+
+        from agent_framework_declarative._workflows._executors_agents import (
+            InvokeAzureAgentExecutor,
+        )
+
+        action_def = {
+            "kind": "InvokeAzureAgent",
+            "agent": {"name": "=Local.UndefinedVar"},
+        }
+        executor = InvokeAzureAgentExecutor(action_def)
+
+        state = DeclarativeWorkflowState(mock_state)
+        state.initialize()
+
+        with patch.object(state, "eval_if_expression", return_value=None):
+            name = executor._get_agent_name(state)
+        assert name is None
+
     async def test_agent_executor_get_input_config_simple(self, mock_context, mock_state):
         """Test input config parsing with simple non-dict input."""
         from agent_framework_declarative._workflows._executors_agents import (
@@ -2337,6 +2421,89 @@ class TestBuilderEdgeWiring:
         exit_exec = graph_builder._get_branch_exit(None)
         assert exit_exec is None
 
+    def test_get_branch_exit_returns_none_for_goto_terminator(self):
+        """Test that _get_branch_exit returns None when branch ends with GotoAction.
+
+        GotoAction is a terminator that handles its own control flow (jumping to
+        the target action). It should NOT be returned as a branch exit, because
+        that would cause the parent ConditionGroup to wire it to the next
+        sequential action, creating a dual-edge where both the goto target and
+        the next action receive messages.
+        """
+        from agent_framework_declarative._workflows._declarative_builder import DeclarativeWorkflowBuilder
+        from agent_framework_declarative._workflows._executors_control_flow import JoinExecutor
+
+        yaml_def = {"name": "test_workflow", "actions": []}
+        graph_builder = DeclarativeWorkflowBuilder(yaml_def)
+
+        # GotoAction executor is a JoinExecutor with a GotoAction action_def
+        goto_executor = JoinExecutor(
+            {"kind": "GotoAction", "id": "goto_summary", "actionId": "invoke_summary"},
+            id="goto_summary",
+        )
+
+        # Simulate a single-action branch chain
+        goto_executor._chain_executors = [goto_executor]  # type: ignore[attr-defined]
+
+        exit_exec = graph_builder._get_branch_exit(goto_executor)
+        assert exit_exec is None
+
+    def test_get_branch_exit_returns_none_for_end_workflow_terminator(self):
+        """Test that _get_branch_exit returns None when branch ends with EndWorkflow."""
+        from agent_framework_declarative._workflows._declarative_builder import DeclarativeWorkflowBuilder
+        from agent_framework_declarative._workflows._executors_control_flow import JoinExecutor
+
+        yaml_def = {"name": "test_workflow", "actions": []}
+        graph_builder = DeclarativeWorkflowBuilder(yaml_def)
+
+        end_executor = JoinExecutor(
+            {"kind": "EndWorkflow", "id": "end"},
+            id="end",
+        )
+        end_executor._chain_executors = [end_executor]  # type: ignore[attr-defined]
+
+        exit_exec = graph_builder._get_branch_exit(end_executor)
+        assert exit_exec is None
+
+    def test_get_branch_exit_returns_none_for_goto_in_chain(self):
+        """Test that _get_branch_exit returns None when chain ends with GotoAction.
+
+        Even when a branch has multiple actions before the GotoAction,
+        the branch exit should be None because the last action is a terminator.
+        """
+        from agent_framework_declarative._workflows._declarative_builder import DeclarativeWorkflowBuilder
+        from agent_framework_declarative._workflows._executors_basic import SendActivityExecutor
+        from agent_framework_declarative._workflows._executors_control_flow import JoinExecutor
+
+        yaml_def = {"name": "test_workflow", "actions": []}
+        graph_builder = DeclarativeWorkflowBuilder(yaml_def)
+
+        # A branch with: SendActivity -> GotoAction
+        activity = SendActivityExecutor({"kind": "SendActivity", "activity": {"text": "msg"}}, id="msg")
+        goto = JoinExecutor(
+            {"kind": "GotoAction", "id": "goto_target", "actionId": "some_target"},
+            id="goto_target",
+        )
+        activity._chain_executors = [activity, goto]  # type: ignore[attr-defined]
+
+        exit_exec = graph_builder._get_branch_exit(activity)
+        assert exit_exec is None
+
+    def test_get_branch_exit_returns_executor_for_non_terminator(self):
+        """Test that _get_branch_exit still returns the exit for non-terminator branches."""
+        from agent_framework_declarative._workflows._declarative_builder import DeclarativeWorkflowBuilder
+        from agent_framework_declarative._workflows._executors_basic import SendActivityExecutor
+
+        yaml_def = {"name": "test_workflow", "actions": []}
+        graph_builder = DeclarativeWorkflowBuilder(yaml_def)
+
+        exec1 = SendActivityExecutor({"kind": "SendActivity", "activity": {"text": "1"}}, id="e1")
+        exec2 = SendActivityExecutor({"kind": "SendActivity", "activity": {"text": "2"}}, id="e2")
+        exec1._chain_executors = [exec1, exec2]  # type: ignore[attr-defined]
+
+        exit_exec = graph_builder._get_branch_exit(exec1)
+        assert exit_exec == exec2
+
 
 # ---------------------------------------------------------------------------
 # Agent executor external loop response handler tests
@@ -2702,3 +2869,133 @@ class TestLongMessageTextHandling:
 
         result = state.eval('=!IsBlank(Find("CONGRATULATIONS", Upper(MessageText(Local.Messages))))')
         assert result is False
+
+
+class TestCreateConversationExecutor:
+    """Tests for CreateConversationExecutor."""
+
+    async def test_basic_creation(self, mock_context, mock_state):
+        """Test that a UUID is generated, stored at conversationId path, and conversation entry created."""
+        from agent_framework_declarative._workflows._executors_basic import (
+            CreateConversationExecutor,
+        )
+
+        state = DeclarativeWorkflowState(mock_state)
+        state.initialize()
+
+        action_def = {
+            "kind": "CreateConversation",
+            "conversationId": "Local.myConvId",
+        }
+        executor = CreateConversationExecutor(action_def)
+        await executor.handle_action(ActionTrigger(), mock_context)
+
+        # A UUID should be stored at the requested path
+        conv_id = state.get("Local.myConvId")
+        assert conv_id is not None
+        assert isinstance(conv_id, str)
+        assert len(conv_id) == 36  # UUID format
+
+        # Conversation entry should exist in System.conversations
+        conversations = state.get("System.conversations")
+        assert conversations is not None
+        assert conv_id in conversations
+        assert conversations[conv_id]["id"] == conv_id
+        assert conversations[conv_id]["messages"] == []
+
+    async def test_no_conversation_id_param(self, mock_context, mock_state):
+        """Test that conversation is still created even without a conversationId param."""
+        from agent_framework_declarative._workflows._executors_basic import (
+            CreateConversationExecutor,
+        )
+
+        state = DeclarativeWorkflowState(mock_state)
+        state.initialize()
+
+        action_def = {
+            "kind": "CreateConversation",
+        }
+        executor = CreateConversationExecutor(action_def)
+        await executor.handle_action(ActionTrigger(), mock_context)
+
+        # Conversation entry should still exist in System.conversations
+        # (initialize() seeds one default conversation, plus the one just created)
+        conversations = state.get("System.conversations")
+        assert conversations is not None
+        assert len(conversations) == 2
+
+    async def test_multiple_conversations(self, mock_context, mock_state):
+        """Test creating multiple conversations produces distinct IDs."""
+        from agent_framework_declarative._workflows._executors_basic import (
+            CreateConversationExecutor,
+        )
+
+        state = DeclarativeWorkflowState(mock_state)
+        state.initialize()
+
+        action_def1 = {
+            "kind": "CreateConversation",
+            "conversationId": "Local.conv1",
+        }
+        action_def2 = {
+            "kind": "CreateConversation",
+            "conversationId": "Local.conv2",
+        }
+
+        executor1 = CreateConversationExecutor(action_def1)
+        await executor1.handle_action(ActionTrigger(), mock_context)
+
+        executor2 = CreateConversationExecutor(action_def2)
+        await executor2.handle_action(ActionTrigger(), mock_context)
+
+        conv1 = state.get("Local.conv1")
+        conv2 = state.get("Local.conv2")
+
+        assert conv1 != conv2
+
+        # initialize() seeds one default conversation, plus the two just created
+        conversations = state.get("System.conversations")
+        assert len(conversations) == 3
+        assert conv1 in conversations
+        assert conv2 in conversations
+
+
+class TestDeclarativeWorkflowStateConversationIdInit:
+    """Tests that DeclarativeWorkflowState.initialize() generates a real UUID for ConversationId."""
+
+    async def test_conversation_id_is_not_default(self, mock_state):
+        """System.ConversationId should be a UUID, not 'default'."""
+        state = DeclarativeWorkflowState(mock_state)
+        state.initialize()
+
+        conv_id = state.get("System.ConversationId")
+        assert conv_id is not None
+        assert conv_id != "default"
+        # Validate it looks like a UUID
+        import uuid
+
+        uuid.UUID(conv_id)  # Raises ValueError if not a valid UUID
+
+    async def test_conversations_dict_initialized(self, mock_state):
+        """System.conversations should contain an entry matching ConversationId."""
+        state = DeclarativeWorkflowState(mock_state)
+        state.initialize()
+
+        conv_id = state.get("System.ConversationId")
+        conversations = state.get("System.conversations")
+        assert conversations is not None
+        assert conv_id in conversations
+        assert conversations[conv_id]["id"] == conv_id
+        assert conversations[conv_id]["messages"] == []
+
+    async def test_each_initialize_generates_unique_id(self, mock_state):
+        """Each call to initialize() should produce a different ConversationId."""
+        state = DeclarativeWorkflowState(mock_state)
+
+        state.initialize()
+        id1 = state.get("System.ConversationId")
+
+        state.initialize()
+        id2 = state.get("System.ConversationId")
+
+        assert id1 != id2