Python: consume local process inputs before awaiting step invocations

python/semantic_kernel/processes/local_runtime/local_step.py

@@ -149,6 +149,10 @@ async def handle_message(self, message: LocalMessage):
         if function is None:
             raise ProcessFunctionNotFoundException(f"Function {target_function} not found in plugin {self.name}")
 
+        # Consume the complete input set before invoking. Concurrent messages must
+        # populate a fresh buffer, which this invocation must not reset on completion.
+        self.inputs[target_function] = self.initial_inputs.get(target_function, {}).copy()
+
         invoke_result = None
         event_name: str = ""
         event_value = None
@@ -169,9 +173,6 @@ async def handle_message(self, message: LocalMessage):
         finally:
             await self.emit_event(KernelProcessEvent(id=event_name, data=event_value))
 
-            # Reset the inputs for the function that was just executed
-            self.inputs[target_function] = self.initial_inputs.get(target_function, {}).copy()
-
     async def invoke_function(self, function: "KernelFunction", kernel: "Kernel", arguments: dict[str, Any]):
         """Invokes the function."""
         return await kernel.invoke(function, **arguments)

python/tests/unit/processes/dapr_runtime/test_step_actor.py

@@ -1,5 +1,6 @@
 # Copyright (c) Microsoft. All rights reserved.
 
+import asyncio
 import json
 from typing import cast
 from unittest.mock import AsyncMock, MagicMock, patch
@@ -11,6 +12,8 @@
 from semantic_kernel.exceptions.process_exceptions import (
     ProcessFunctionNotFoundException,
 )
+from semantic_kernel.functions import kernel_function
+from semantic_kernel.kernel import Kernel
 from semantic_kernel.processes.dapr_runtime.actors.actor_state_key import ActorStateKeys
 from semantic_kernel.processes.dapr_runtime.actors.event_buffer_actor import EventBufferActor
 from semantic_kernel.processes.dapr_runtime.actors.step_actor import StepActor
@@ -21,6 +24,7 @@
     KernelProcessEventVisibility,
 )
 from semantic_kernel.processes.kernel_process.kernel_process_function_target import KernelProcessFunctionTarget
+from semantic_kernel.processes.kernel_process.kernel_process_step import KernelProcessStep
 from semantic_kernel.processes.kernel_process.kernel_process_step_state import KernelProcessStepState
 from semantic_kernel.processes.process_event import ProcessEvent
 from semantic_kernel.processes.process_message import ProcessMessage
@@ -122,6 +126,56 @@ async def test_process_incoming_messages(actor_context):
         mock_try_add_state.assert_any_call("incomingMessagesState", expected_messages)
 
 
+@pytest.mark.parametrize("defer_second_b", [False, True])
+@pytest.mark.parametrize("fail_first", [False, True])
+async def test_process_incoming_messages_keeps_awaiting_invocations_serial(actor_context, defer_second_b, fail_first):
+    pairs = []
+
+    class Join(KernelProcessStep):
+        @kernel_function
+        async def join(self, a: int, b: int) -> int:
+            await asyncio.sleep(0)
+            pairs.append((a, b))
+            if fail_first and a == 1:
+                raise RuntimeError("first pair failed")
+            return a + b
+
+    actor_context.kernel = Kernel()
+    actor_context.factories = {"Join": Join}
+    proxy = AsyncMock()
+    messages = [
+        ProcessMessage(source_id="source", destination_id="join", function_name="join", values={key: value})
+        for key, value in (("a", 1), ("b", 1), ("a", 2), ("b", 2))
+    ]
+    with (
+        patch("dapr.actor.ActorProxy.create", return_value=proxy),
+        patch.object(actor_context._state_manager, "try_add_state", new=AsyncMock()),
+        patch.object(actor_context._state_manager, "set_state", new=AsyncMock()),
+        patch.object(actor_context._state_manager, "save_state", new=AsyncMock()),
+    ):
+        await actor_context.initialize_step(
+            json.dumps({
+                "step_info": {
+                    "state": {"name": "Join", "version": "1.0", "id": "join"},
+                    "inner_step_python_type": "Join",
+                    "edges": {},
+                },
+            })
+        )
+        first_batch = messages[:-1] if defer_second_b else messages
+        proxy.dequeue_all.return_value = [message.model_dump_json() for message in first_batch]
+        assert await actor_context.prepare_incoming_messages() == len(first_batch)
+        await actor_context.process_incoming_messages()
+
+        if defer_second_b:
+            assert pairs == [(1, 1)]
+            proxy.dequeue_all.return_value = [messages[-1].model_dump_json()]
+            assert await actor_context.prepare_incoming_messages() == 1
+            await actor_context.process_incoming_messages()
+
+    assert pairs == [(1, 1), (2, 2)]
+
+
 async def test_activate_step_with_factory_creates_state(actor_context):
     fake_step_instance = FakeStep()
     fake_step_instance.activate = AsyncMock(side_effect=fake_step_instance.activate)

python/tests/unit/processes/local_runtime/test_local_process.py

@@ -8,6 +8,7 @@
 
 from semantic_kernel.exceptions.kernel_exceptions import KernelException
 from semantic_kernel.exceptions.process_exceptions import ProcessEventUndefinedException
+from semantic_kernel.functions import kernel_function
 from semantic_kernel.kernel import Kernel
 from semantic_kernel.processes.kernel_process.kernel_process import KernelProcess
 from semantic_kernel.processes.kernel_process.kernel_process_edge import KernelProcessEdge
@@ -16,10 +17,13 @@
     KernelProcessEventVisibility,
 )
 from semantic_kernel.processes.kernel_process.kernel_process_state import KernelProcessState
+from semantic_kernel.processes.kernel_process.kernel_process_step import KernelProcessStep
 from semantic_kernel.processes.kernel_process.kernel_process_step_info import KernelProcessStepInfo
+from semantic_kernel.processes.local_runtime.local_kernel_process import start
 from semantic_kernel.processes.local_runtime.local_message import LocalMessage
 from semantic_kernel.processes.local_runtime.local_process import LocalProcess
 from semantic_kernel.processes.local_runtime.local_step import LocalStep
+from semantic_kernel.processes.process_builder import ProcessBuilder
 
 
 @pytest.fixture
@@ -331,6 +335,50 @@ async def test_handle_message_with_valid_event_id(mock_process_with_output_edges
 END_PROCESS_ID = "END"
 
 
+@pytest.mark.parametrize("defer_second_b", [False, True])
+@pytest.mark.parametrize("fail_first", [False, True])
+async def test_messages_to_same_step_do_not_reuse_inputs_during_invocation(defer_second_b, fail_first):
+    from semantic_kernel.processes.kernel_process.kernel_process_step_context import KernelProcessStepContext
+
+    pairs = []
+
+    class Producer(KernelProcessStep):
+        @kernel_function
+        async def emit(self, context: KernelProcessStepContext) -> None:
+            for event, value in (("a1", 1), ("b1", 1), ("a2", 2), ("b2", 2)):
+                await context.emit_event(event, data=value)
+
+    class Join(KernelProcessStep):
+        @kernel_function
+        async def join(self, a: int, b: int) -> None:
+            await asyncio.sleep(0)
+            pairs.append((a, b))
+            if fail_first and a == 1:
+                raise RuntimeError("first pair failed")
+
+    class Relay(KernelProcessStep):
+        @kernel_function
+        async def relay(self, value: int) -> int:
+            return value
+
+    builder = ProcessBuilder(name="join")
+    producer = builder.add_step(Producer)
+    join = builder.add_step(Join)
+    builder.on_input_event("start").send_event_to(producer, function_name="emit")
+    for event in ("a1", "b1", "a2", "b2"):
+        if defer_second_b and event == "b2":
+            relay = builder.add_step(Relay)
+            producer.on_event(event).send_event_to(relay, function_name="relay", parameter_name="value")
+            relay.on_function_result("relay").send_event_to(join, function_name="join", parameter_name="b")
+        else:
+            producer.on_event(event).send_event_to(join, function_name="join", parameter_name=event[0])
+
+    async with await start(builder.build(), Kernel(), initial_event="start"):
+        pass
+
+    assert pairs == [(1, 1), (2, 2)]
+
+
 @pytest.fixture
 def mock_process_with_steps(mock_process, mock_kernel, build_model):
     """Fixture to create a LocalProcess with mock steps."""