Address Copilot review batch: tests + Workflow.reset escape hatch

* Add Workflow.reset() public method as recovery escape hatch when an
  in-flight run aborted (e.g. WorkflowConvergenceException) and the
  workflow is not checkpointed. Update the in-flight messages guard's
  error message to point callers at it.

* Add test_workflow_run_inflight_messages_guard exercising both the
  guard (sync + streaming) and the reset() recovery path.
* Add test_workflow_reset_rejects_concurrent_runs to lock down the
  in-progress guard on reset.

* Add test_as_agent_continuation_preserves_prior_state covering the
  is_continuation branch in _ensure_state_initialized: stamps a marker
  between calls and asserts it survives, while Inputs.input and
  System.LastMessageText refresh to the new turn.

* Add test_powerfx_safe.py regression tests for the Enum branch in
  _make_powerfx_safe (str-subclass, int-subclass, plain Enum, and
  Enums nested in dict/list).

* Drop redundant @pytest.mark.asyncio on
  test_as_agent_round_trip_with_last_message_text (asyncio_mode='auto').

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

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

@@ -618,7 +618,8 @@ class Workflow(DictConvertible):
             raise RuntimeError(
                 "Cannot start a new run with 'message' while in-flight executor "
                 "messages remain from a prior run. Either resume from a checkpoint "
-                "(checkpoint_id=...) or wait for the prior run to complete."
+                "(checkpoint_id=...), wait for the prior run to complete, or call "
+                "'await workflow.reset()' to drop the pending messages."
             )
 
         initial_executor_fn = self._resolve_execution_mode(
@@ -651,6 +652,30 @@ class Workflow(DictConvertible):
             self._runner.context.clear_runtime_checkpoint_storage()
         self._reset_running_flag()
 
+    async def reset(self) -> None:
+        """Drop all in-flight executor messages and per-run accounting.
+
+        Workflows preserve shared state and pending executor messages
+        across :meth:`run` calls so that multi-turn callers (e.g.
+        :class:`WorkflowAgent`) can deliver follow-up turns to the same
+        instance without losing context. If a prior run aborted (e.g. the
+        runner raised :class:`WorkflowConvergenceException`) and the
+        workflow is not checkpointed, those pending messages remain in
+        the runner context and every future ``run(message=...)`` call
+        fails with ``RuntimeError`` because of the in-flight-messages
+        guard. Callers that have no checkpoint to resume from can use
+        ``await workflow.reset()`` as an explicit escape hatch to clear
+        pending messages and start fresh.
+
+        Note: this does NOT clear the workflow ``State`` (use
+        :meth:`Workflow.run` with a ``checkpoint_id`` for state replay)
+        and is a no-op while another run is in progress on this instance.
+        """
+        if self._is_running:
+            raise RuntimeError("Cannot reset a workflow while a run is in progress.")
+        self._runner.context.reset_for_new_run()
+        self._runner.reset_iteration_count()
+
     @staticmethod
     def _finalize_events(
         events: Sequence[WorkflowEvent],

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

@@ -953,6 +953,50 @@ async def test_agent_streaming_vs_non_streaming() -> None:
     assert accumulated_text == "Hello World", f"Expected 'Hello World', got '{accumulated_text}'"
 
 
+async def test_workflow_run_inflight_messages_guard(simple_executor: Executor) -> None:
+    """``run(message=...)`` must reject in-flight executor messages from a prior run.
+
+    Workflows preserve state and pending messages across :meth:`Workflow.run`
+    calls. If a prior run aborted before the runner drained those pending
+    messages (e.g. it raised :class:`WorkflowConvergenceException`), the next
+    fresh-message call should fail loudly instead of silently mixing the
+    leftover messages with the new turn. Callers can recover via
+    :meth:`Workflow.reset`.
+    """
+    workflow = WorkflowBuilder(start_executor=simple_executor).add_edge(simple_executor, simple_executor).build()
+    test_message = WorkflowMessage(data="test", source_id="test", target_id=None)
+
+    # Simulate an aborted prior run by leaving a message in the runner context.
+    workflow._runner.context._messages["test"] = [test_message]
+    assert await workflow._runner.context.has_messages()
+
+    with pytest.raises(RuntimeError, match="in-flight executor messages"):
+        await workflow.run(test_message)
+
+    with pytest.raises(RuntimeError, match="in-flight executor messages"):
+        async for _ in workflow.run(test_message, stream=True):
+            pass
+
+    # ``Workflow.reset`` is the documented escape hatch.
+    await workflow.reset()
+    assert not await workflow._runner.context.has_messages()
+
+    # After reset, a new run is accepted again.
+    result = await workflow.run(test_message)
+    assert result.get_final_state() == WorkflowRunState.IDLE
+
+
+async def test_workflow_reset_rejects_concurrent_runs(simple_executor: Executor) -> None:
+    """``Workflow.reset`` must not stomp on an in-progress run."""
+    workflow = WorkflowBuilder(start_executor=simple_executor).add_edge(simple_executor, simple_executor).build()
+    workflow._is_running = True
+    try:
+        with pytest.raises(RuntimeError, match="run is in progress"):
+            await workflow.reset()
+    finally:
+        workflow._is_running = False
+
+
 async def test_workflow_run_parameter_validation(simple_executor: Executor) -> None:
     """Test that stream properly validate parameter combinations."""
     workflow = WorkflowBuilder(start_executor=simple_executor).add_edge(simple_executor, simple_executor).build()

python/packages/declarative/tests/test_powerfx_safe.py

@@ -0,0 +1,59 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Regression tests for ``_make_powerfx_safe``.
+
+PowerFx (via pythonnet) only accepts plain primitives, dicts, and lists.
+``Enum`` instances - especially ``str``- and ``int``-subclass enums like
+MAF's ``MessageRole`` - silently pass ``isinstance(v, str)`` /
+``isinstance(v, int)`` checks but blow up later inside pythonnet with
+``'<EnumName>' value cannot be converted to System.<X>``. These tests
+pin down the Enum coercion branch so we don't regress that interop fix.
+"""
+
+from enum import Enum, IntEnum
+
+from agent_framework_declarative._workflows._declarative_base import _make_powerfx_safe
+
+
+class _StrRole(str, Enum):
+    USER = "user"
+    SYSTEM = "system"
+
+
+class _IntCode(IntEnum):
+    ONE = 1
+    TWO = 2
+
+
+class _PlainEnum(Enum):
+    X = "x"
+    Y = 42
+
+
+def test_str_subclass_enum_reduces_to_str():
+    assert _make_powerfx_safe(_StrRole.USER) == "user"
+    assert type(_make_powerfx_safe(_StrRole.USER)) is str
+
+
+def test_int_subclass_enum_reduces_to_int():
+    assert _make_powerfx_safe(_IntCode.ONE) == 1
+    assert type(_make_powerfx_safe(_IntCode.ONE)) is int
+
+
+def test_plain_enum_reduces_to_underlying_value():
+    assert _make_powerfx_safe(_PlainEnum.X) == "x"
+    assert _make_powerfx_safe(_PlainEnum.Y) == 42
+
+
+def test_enum_inside_dict_is_coerced():
+    safe = _make_powerfx_safe({"role": _StrRole.USER, "code": _IntCode.TWO})
+    assert safe == {"role": "user", "code": 2}
+    assert type(safe["role"]) is str
+    assert type(safe["code"]) is int
+
+
+def test_enum_inside_list_is_coerced():
+    safe = _make_powerfx_safe([_StrRole.USER, _IntCode.ONE])
+    assert safe == ["user", 1]
+    assert type(safe[0]) is str
+    assert type(safe[1]) is int

python/packages/declarative/tests/test_workflow_factory.py

@@ -228,7 +228,6 @@ 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().
@@ -256,6 +255,71 @@ actions:
             f"Expected 'Hello there' in agent response text but got: {response.text!r}"
         )
 
+    async def test_as_agent_continuation_preserves_prior_state(self):
+        """Regression test for the ``is_continuation`` branch in
+        ``DeclarativeWorkflowExecutor._ensure_state_initialized``.
+
+        Verifies, end-to-end via ``Workflow.as_agent()``:
+          * Turn 1 initializes the declarative state via ``state.initialize``.
+          * Turn 2 takes the *continuation* branch (skips ``state.initialize``),
+            so any non-Inputs/non-System state stamped on turn 1 survives.
+          * Turn 2 still refreshes ``Inputs.input`` and
+            ``System.LastMessage*`` to the new user message.
+
+        Without state preservation, ``Workflow.run`` would clear shared state
+        on entry and ``state.initialize`` would re-run on every turn,
+        wiping the marker we stamped between calls.
+        """
+        from agent_framework_declarative._workflows._declarative_base import DECLARATIVE_STATE_KEY
+
+        factory = WorkflowFactory()
+        workflow = factory.create_workflow_from_yaml("""
+name: as-agent-continuation-test
+actions:
+  - kind: SendActivity
+    activity:
+      text: =System.LastMessageText
+""")
+
+        agent = workflow.as_agent(name="continuation-agent")
+
+        first = await agent.run("turn-1-msg")
+        assert first.text == "turn-1-msg", (
+            f"Expected turn-1 echo 'turn-1-msg', got: {first.text!r}"
+        )
+
+        # Stamp a marker into the declarative state between turns. The
+        # continuation branch must preserve it; a state-clearing run would
+        # wipe ``DECLARATIVE_STATE_KEY`` and force re-initialization.
+        state_data = workflow._state.get(DECLARATIVE_STATE_KEY)
+        assert isinstance(state_data, dict), (
+            "Expected declarative state to be initialized after turn 1"
+        )
+        state_data["Local"] = {"persisted_marker": "kept-from-turn-1"}
+        workflow._state.set(DECLARATIVE_STATE_KEY, state_data)
+        workflow._state.commit()
+
+        second = await agent.run("turn-2-msg")
+        assert second.text == "turn-2-msg", (
+            f"Expected System.LastMessageText to refresh to 'turn-2-msg', got: {second.text!r}"
+        )
+
+        # The continuation branch in ``_ensure_state_initialized`` must:
+        # 1. preserve the cross-turn marker we stamped above
+        # 2. refresh Inputs.input and System.LastMessage* to the new turn
+        post_state = workflow._state.get(DECLARATIVE_STATE_KEY)
+        assert isinstance(post_state, dict), "declarative state vanished between turns"
+        local = post_state.get("Local", {})
+        assert local.get("persisted_marker") == "kept-from-turn-1", (
+            f"Cross-turn marker was wiped (state was reset). post_state Local={local!r}"
+        )
+        assert post_state.get("Inputs", {}).get("input") == "turn-2-msg", (
+            f"Inputs.input not refreshed on turn 2: {post_state.get('Inputs')!r}"
+        )
+        assert post_state.get("System", {}).get("LastMessageText") == "turn-2-msg", (
+            f"System.LastMessageText not refreshed on turn 2: {post_state.get('System')!r}"
+        )
+
 
 class TestWorkflowFactoryAgentRegistration:
     """Tests for agent registration."""