Python: Bug: local process messages reuse in-flight inputs and invoke an extra mixed-input call #14759

Describe the bug

The local process runtime dispatches all messages in a superstep concurrently, including messages to the same step. A step keeps its input buffer populated until the invoked function finishes. If the function awaits, the next message can reuse an input from the in-flight invocation and trigger an extra call with values from different input sets.

For input messages a=1, b=1, a=2, b=2, the join function executes three times: (1,1), (2,1), (2,2). The middle call was never intended by the process graph. This can duplicate side effects and combine unrelated records.

To Reproduce

No model, credentials, or external service is needed. On main at df1cb33d0d32699c8d6a2e5b3ea7e5f8990d86d5:

import asyncio
from semantic_kernel import Kernel
from semantic_kernel.functions import kernel_function
from semantic_kernel.processes.kernel_process.kernel_process_step import KernelProcessStep
from semantic_kernel.processes.kernel_process.kernel_process_step_context import KernelProcessStepContext
from semantic_kernel.processes.local_runtime.local_kernel_process import start
from semantic_kernel.processes.process_builder import ProcessBuilder

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)  # Any asynchronous work exposes the overlap.
        pairs.append((a, b))

async def main():
    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"):
        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
    print(pairs)

asyncio.run(main())

Actual output:

[(1, 1), (2, 1), (2, 2)]

Expected behavior

Once a complete input set triggers an invocation, those inputs should no longer be eligible for another invocation. Later messages should form a fresh input set, producing:

[(1, 1), (2, 2)]

Independent steps should still execute concurrently.

Screenshots

Not applicable.

Platform

• Language: Python 3.12.14
• Source: semantic-kernel 1.47.0, main at df1cb33d0d32699c8d6a2e5b3ea7e5f8990d86d5
• AI model: none
• IDE: not applicable
• OS: macOS, Apple Silicon

Additional context

LocalProcess.internal_execute gathers one coroutine per message. LocalStep.handle_message only resets its function inputs in finally, after awaiting the function. The example uses asyncio.sleep(0) solely to yield to other queued messages; it does not depend on wall-clock timing.

I intend to work on a focused fix for the input-buffer lifecycle while preserving the runtime's existing concurrency.

AI assistance: Codex prepared the reproduction and diagnosis; the script was executed locally.