diff --git a/python/packages/core/agent_framework/_workflows/_workflow.py b/python/packages/core/agent_framework/_workflows/_workflow.py index 0673132ade..b43c81b8d7 100644 --- a/python/packages/core/agent_framework/_workflows/_workflow.py +++ b/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], diff --git a/python/packages/core/tests/workflow/test_workflow.py b/python/packages/core/tests/workflow/test_workflow.py index 18d2f26997..e49e5f0e53 100644 --- a/python/packages/core/tests/workflow/test_workflow.py +++ b/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() diff --git a/python/packages/declarative/tests/test_powerfx_safe.py b/python/packages/declarative/tests/test_powerfx_safe.py new file mode 100644 index 0000000000..fccbd72b28 --- /dev/null +++ b/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 +``'' value cannot be converted to System.``. 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 diff --git a/python/packages/declarative/tests/test_workflow_factory.py b/python/packages/declarative/tests/test_workflow_factory.py index e9988ea97c..809747a037 100644 --- a/python/packages/declarative/tests/test_workflow_factory.py +++ b/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."""