Merge remote-tracking branch 'origin/main' into feat/support-agent-sandbox

api/libs/broadcast_channel/channel.py

@@ -125,7 +125,8 @@ class BroadcastChannel(Protocol):
     a specific topic, all subscription should receive the published message.
 
     There are no restriction for the persistence of messages. Once a subscription is created, it
-    should receive all subsequent messages published.
+    should receive all subsequent messages published. However, a subscription should not receive
+    any message published before the subscription is established.
 
     `BroadcastChannel` implementations must be thread-safe and support concurrent use by multiple threads.
     """

api/libs/broadcast_channel/redis/streams_channel.py

@@ -64,7 +64,10 @@ class _StreamsSubscription(Subscription):
         self._client = client
         self._key = key
         self._closed = threading.Event()
-        self._last_id = "0-0"
+        # Setting initial last id to `$` to signal redis that we only want new messages.
+        #
+        # ref: https://redis.io/docs/latest/commands/xread/#the-special--id
+        self._last_id = "$"
         self._queue: queue.Queue[object] = queue.Queue()
         self._start_lock = threading.Lock()
         self._listener: threading.Thread | None = None