Fix declarative Workflow.as_agent() by accepting list[Message] in start executor

The declarative start executor (JoinExecutor) only advertised dict and str
in its input_types, so WorkflowAgent.__init__ rejected it with
'Workflow's start executor cannot handle list[Message]'.

Add list[Message] to the JoinExecutor handler annotation and add a
matching branch in DeclarativeActionExecutor._ensure_state_initialized
that extracts the last user-message text and falls through to the
string-input initialization path, so =System.LastMessageText works
end-to-end via as_agent().

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

python/packages/declarative/agent_framework_declarative/_workflows/_declarative_base.py

@@ -36,6 +36,7 @@ from typing import Any, Literal, cast
 
 from agent_framework import (
     Executor,
+    Message,
     WorkflowContext,
 )
 from agent_framework._workflows._state import State
@@ -873,6 +874,9 @@ class DeclarativeActionExecutor(Executor):
         Follows .NET's DefaultTransform pattern - accepts any input type:
         - dict/Mapping: Used directly as workflow.inputs
         - str: Converted to {"input": value}
+        - list[Message]: Joined to a string from the last user message text
+          (or last message text if no user message). Falls through to the
+          string-input path so System.LastMessage.Text is populated.
         - DeclarativeMessage: Internal message, no initialization needed
         - Any other type: Converted via str() to {"input": str(value)}
 
@@ -888,6 +892,23 @@ class DeclarativeActionExecutor(Executor):
         if isinstance(trigger, dict):
             # Structured inputs - use directly
             state.initialize(trigger)  # type: ignore
+        elif isinstance(trigger, list) and all(isinstance(m, Message) for m in trigger):
+            # list[Message] (e.g. from WorkflowAgent / as_agent()) - extract the
+            # last user message text and treat it as the string input. Fall
+            # through to the same state initialization as the str case so
+            # =System.LastMessage.Text / =System.LastMessageText keep working.
+            messages_list = cast(list[Message], trigger)
+            user_text = ""
+            for msg in reversed(messages_list):
+                if str(msg.role).lower() == "user" and msg.text:
+                    user_text = msg.text
+                    break
+            if not user_text:
+                # Fallback: concatenate any text from the last message.
+                user_text = messages_list[-1].text if messages_list else ""
+            state.initialize({"input": user_text})
+            state.set("System.LastMessage", {"Text": user_text, "Id": ""})
+            state.set("System.LastMessageText", user_text)
         elif isinstance(trigger, str):
             # String input - wrap in dict and populate System.LastMessage.Text
             # so YAML expressions like =System.LastMessage.Text see the user input

python/packages/declarative/agent_framework_declarative/_workflows/_executors_control_flow.py

@@ -17,6 +17,7 @@ The key insight is that control flow becomes GRAPH STRUCTURE, not executor logic
 from typing import Any, cast
 
 from agent_framework import (
+    Message,
     WorkflowContext,
     handler,
 )
@@ -492,7 +493,13 @@ class JoinExecutor(DeclarativeActionExecutor):
     @handler
     async def handle_action(
         self,
-        trigger: dict[str, Any] | str | ActionTrigger | ActionComplete | ConditionResult | LoopIterationResult,
+        trigger: dict[str, Any]
+        | str
+        | list[Message]
+        | ActionTrigger
+        | ActionComplete
+        | ConditionResult
+        | LoopIterationResult,
         ctx: WorkflowContext[ActionComplete],
     ) -> None:
         """Simply pass through to continue the workflow."""

python/packages/declarative/tests/test_workflow_factory.py

@@ -228,6 +228,34 @@ actions:
         outputs = result.get_outputs()
         assert any("hello-world" in str(o) for o in outputs), f"Expected 'hello-world' in outputs but got: {outputs}"
 
+    @pytest.mark.asyncio
+    async def test_as_agent_round_trip_with_last_message_text(self):
+        """Regression test: a declarative workflow built via WorkflowFactory must be
+        consumable as an AIAgent via Workflow.as_agent().
+
+        Specifically, the declarative start executor must accept list[Message]
+        (the input passed by WorkflowAgent) and populate System.LastMessageText
+        so =System.LastMessageText is resolvable in the YAML.
+        """
+        factory = WorkflowFactory()
+        workflow = factory.create_workflow_from_yaml("""
+name: as-agent-roundtrip-test
+actions:
+  - kind: SetVariable
+    variable: Local.echo
+    value: =System.LastMessageText
+  - kind: SendActivity
+    activity:
+      text: =Local.echo
+""")
+
+        agent = workflow.as_agent(name="echo-agent")
+        response = await agent.run("Hello there")
+
+        assert "Hello there" in response.text, (
+            f"Expected 'Hello there' in agent response text but got: {response.text!r}"
+        )
+
 
 class TestWorkflowFactoryAgentRegistration:
     """Tests for agent registration."""