fix(agentex): run SSE status recheck before the read so read failures can't bypass it

The recheck ran after the read cycle, so persistent read errors jumped straight
to the retry/backoff handler and never reached it — a terminal task whose event
publish was lost could stay open while Redis reads kept failing. Move the
interval-gated status recheck to the top of the loop so it runs every iteration,
including after a backoff, independent of read health.

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

@@ -133,6 +133,29 @@ async def stream_task_events(
             # Application-level control loop
             while True:
                 try:
+                    # Authoritative status recheck on an interval. Runs at the
+                    # TOP of every iteration — even after a read failure/backoff —
+                    # so a terminal task ends even if its event publish was lost
+                    # or Redis reads keep erroring.
+                    current_time = asyncio.get_running_loop().time()
+                    if current_time - last_status_check >= ping_interval:
+                        last_status_check = current_time
+                        try:
+                            task = await self.task_service.get_task(id=task_id)
+                        except ItemDoesNotExist:
+                            # Row permanently gone (e.g. retention) — end, don't retry.
+                            logger.info(
+                                f"Ending SSE stream for task {task_id}: "
+                                "task no longer exists"
+                            )
+                            return
+                        if task.status in TERMINAL_TASK_STATUSES:
+                            logger.info(
+                                f"Ending SSE stream for task {task_id}: "
+                                "terminal on status recheck"
+                            )
+                            return
+
                     # Process yielded messages one by one
                     message_generator = self.read_messages(
                         topic=stream_topic, last_id=last_id
@@ -163,27 +186,6 @@ async def stream_task_events(
                     # healthy again, so reset the backoff/error counter.
                     consecutive_errors = 0
 
-                    current_time = asyncio.get_running_loop().time()
-                    # Authoritative status recheck on an interval, busy or idle,
-                    # in case a terminal event's publish was lost.
-                    if current_time - last_status_check >= ping_interval:
-                        last_status_check = current_time
-                        try:
-                            task = await self.task_service.get_task(id=task_id)
-                        except ItemDoesNotExist:
-                            # Row permanently gone (e.g. retention) — end, don't retry.
-                            logger.info(
-                                f"Ending SSE stream for task {task_id}: "
-                                "task no longer exists"
-                            )
-                            return
-                        if task.status in TERMINAL_TASK_STATUSES:
-                            logger.info(
-                                f"Ending SSE stream for task {task_id}: "
-                                "terminal on status recheck"
-                            )
-                            return
-
                     # Idle: send keepalive ping so proxies don't reap us.
                     if message_count == 0:
                         if current_time - last_message_time >= ping_interval: