[Bug] set_dynamic_signal_handler delivers buffered signals grouped by name instead of in arrival order #1996

import asyncio
import uuid
from collections.abc import Sequence
from datetime import timedelta

from temporalio import activity, workflow
from temporalio.common import RawValue
from temporalio.testing import WorkflowEnvironment
from temporalio.worker import Worker


@activity.defn
async def load_cart(cart_id: str) -> list[str]:
    # Stands in for a database read
    await asyncio.sleep(1)
    return []


@workflow.defn
class Cart:
    def __init__(self) -> None:
        self.items: list[str] = []
        self.handled: list[str] = []
        self.checked_out = False

    @workflow.run
    async def run(self, cart_id: str) -> tuple[list[str], list[str]]:
        self.items = await workflow.execute_activity(
            load_cart, cart_id, start_to_close_timeout=timedelta(seconds=10)
        )

        def on_event(name: str, args: Sequence[RawValue]) -> None:
            item = workflow.payload_converter().from_payload(args[0].payload, str)
            self.handled.append(f"{name} {item}".strip())
            if name == "add":
                self.items.append(item)
            elif name == "clear":
                self.items.clear()
            elif name == "checkout":
                self.checked_out = True

        workflow.set_dynamic_signal_handler(on_event)
        await workflow.wait_condition(lambda: self.checked_out)
        return self.handled, self.items


async def main() -> None:
    async with await WorkflowEnvironment.start_local() as env:
        async with Worker(
            env.client, task_queue="tq", workflows=[Cart], activities=[load_cart]
        ):
            for run in range(1, 6):
                cart = await env.client.start_workflow(
                    Cart.run, "cart-1", id=f"cart-{uuid.uuid4()}", task_queue="tq"
                )
                # The shopper acts while the cart is still loading
                sent = [("add", "apple"), ("clear", ""), ("add", "pear"), ("checkout", "")]
                for name, item in sent:
                    await cart.signal(name, item)
                handled, items = await cart.result()
                print(f"run {run}: handled {handled} -> cart {items}")


if __name__ == "__main__":
    asyncio.run(main())