fix(agentex): snapshot stream cursor before reading task status at connect

Close a connect-time race that could leave a zombie stream. A terminal
transition writes the DB before it XADDs its task_updated event, and get_task
reads the DB while get_stream_tail_id sets the cursor the live loop reads after.

Previously status was read before the tail snapshot: if the task went terminal
in that gap, the status read said RUNNING (stale) so the connect-time terminal
branch didn't fire, and the tail snapshot then landed on the just-published
terminal event, which the live loop reads past — blocking forever.

Snapshot the tail first, then read status. A non-terminal status read now
guarantees any racing terminal event lands after the cursor, so the live loop is
guaranteed to surface it. Pure reorder of two reads; no behavior change beyond
eliminating the race.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: deepthi-rao-scale

agentex/src/domain/use_cases/streams_use_case.py

@@ -103,18 +103,24 @@ async def stream_task_events(
             task_id = task.id
 
         stream_topic = get_task_event_stream_topic(task_id=task_id)
+        # Snapshot the read cursor BEFORE reading task status (and before yielding
+        # "connected"). Ordering is load-bearing for two reasons:
+        #   1. A terminal transition writes the DB *before* it XADDs its
+        #      task_updated event. Snapshotting the tail first means that if the
+        #      status read below still says non-terminal, the terminal event (if
+        #      one is racing in) is guaranteed to land *after* this cursor, so the
+        #      live loop will read it — no missed-terminal zombie.
+        #   2. "connected" is the client's cue to send its message, which makes
+        #      the agent start XADD-ing deltas; snapshotting after that yield
+        #      could let the first deltas advance the tail and be skipped.
+        # Resolves to "0-0" (from the beginning) when the stream is empty.
+        last_id = await self.stream_repository.get_stream_tail_id(stream_topic)
         # Authoritative connect-time check. If the task is ALREADY terminal, any
         # terminal event sits behind the live tail and the read loop below would
         # never surface it — so replay whatever is buffered and end here. This
         # handles the late-connect case deterministically (no polling needed);
         # the in-loop check below handles a task that finishes while streaming.
         task = await self.task_service.get_task(id=task_id)
-        # Snapshot the read cursor BEFORE yielding "connected". "connected" is the
-        # client's cue to send its message, which makes the agent start XADD-ing
-        # deltas; snapshotting after the yield could let those first deltas advance
-        # the tail and be skipped. Resolves to "0-0" (from the beginning) when the
-        # stream is empty. Only used by the live path below.
-        last_id = await self.stream_repository.get_stream_tail_id(stream_topic)
         # Send initial connection data
         yield f"data: {TaskStreamConnectedEventEntity(type='connected', taskId=task_id).model_dump_json()}\n\n"
         if task.status in TERMINAL_TASK_STATUSES: