refactor(agentex): terminate SSE via connect-time check + terminal event, drop the poll

Collapse stream termination into two deterministic checks and remove the
recurring keepalive fallback:
- Connect-time: if the task is already terminal, replay buffered events and
  end. Handles late connects deterministically (the fallback's main job).
- In-stream: end on a terminal task_updated event (unchanged).

Every real terminal transition funnels through transition_status, which emits a
task_updated, so the event path covers finishes while streaming and the
connect-time check covers finishes before subscribing. The idle branch is now
just the keepalive ping.

Removes _stream_is_finished, the StreamRepository.stream_exists port method and
its Redis impl, and the stream_ever_existed bookkeeping. This also eliminates
the 'expired key closes a live stream' edge (there is no vanished-key check
anymore): a still-running task's stream simply stays open until the client
disconnects, which is correct.

Tests updated: stream_exists assertions use redis.exists directly; the
reclaimed-key test now asserts a running stream is never closed by key
reclamation.

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

Author: deepthi-rao-scale

agentex/src/adapters/streams/adapter_redis.py

@@ -229,14 +229,6 @@ async def read_messages(
             logger.error(f"Error reading from Redis stream {topic}: {e}")
             raise
 
-    async def stream_exists(self, topic: str) -> bool:
-        """Whether the Redis stream key exists (EXISTS)."""
-        try:
-            return bool(await self.redis.exists(topic))
-        except Exception as e:
-            logger.error(f"Error checking existence of Redis stream {topic}: {e}")
-            raise
-
     async def cleanup_stream(self, topic: str) -> None:
         """
         Clean up a Redis stream.

agentex/src/adapters/streams/port.py

@@ -60,14 +60,6 @@ async def get_stream_tail_id(self, topic: str) -> str:
         """
         raise NotImplementedError
 
-    @abstractmethod
-    async def stream_exists(self, topic: str) -> bool:
-        """
-        Report whether a stream topic currently exists, to tell a reclaimed
-        topic ("gone") apart from one that is still live.
-        """
-        raise NotImplementedError
-
     @abstractmethod
     async def cleanup_stream(self, topic: str) -> None:
         """

agentex/src/domain/use_cases/streams_use_case.py

@@ -115,15 +115,29 @@ 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 yielding "connected". "connected" is
-        # the client's cue to send its message, which makes the agent start
-        # XADD-ing deltas. Snapshotting after the yield lets a congested relay
-        # fall behind far enough that those deltas land before the snapshot and
-        # are never read. Snapshotting first resolves to "0-0" (stream is empty
-        # until the client sends), so we read from the beginning.
+        # 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:
+            async for _id, data in self.read_messages(topic=stream_topic, last_id="0"):
+                yield f"data: {data.model_dump_json()}\n\n"
+                await asyncio.sleep(0.02)
+            logger.info(
+                f"Ending SSE stream for task {task_id}: already terminal at connect"
+            )
+            return
+
         last_message_time = asyncio.get_running_loop().time()
         ping_interval = float(
             self.environment_variables.SSE_KEEPALIVE_PING_INTERVAL
@@ -133,12 +147,6 @@ async def stream_task_events(
         # client's read fails on each cycle; without backoff this turns into a
         # log-ingestion firehose (one failure per client per cycle, ~once/sec).
         consecutive_errors = 0
-        # End the sub once no more events can arrive, else XREAD blocks forever,
-        # pinning a Redis connection ("zombie"). Ends via a terminal task_updated
-        # (primary) or the keepalive fallback. A non-"0-0" tail means data existed
-        # at connect; stream_ever_existed gates the fallback so a not-yet-created
-        # topic isn't mistaken for a reclaimed one.
-        stream_ever_existed = last_id != "0-0"
         try:
             # Application-level control loop
             while True:
@@ -152,9 +160,6 @@ async def stream_task_events(
                         # Update the last_id for the next iteration
                         last_id = new_id
                         message_count += 1
-                        # Topic exists; a later disappearance now means "gone",
-                        # not "not yet created".
-                        stream_ever_existed = True
                         # Send the data to the client
                         data_str = f"data: {data.model_dump_json()}\n\n"
                         yield data_str
@@ -177,21 +182,11 @@ async def stream_task_events(
                     # healthy again, so reset the backoff/error counter.
                     consecutive_errors = 0
 
-                    # No messages this cycle: on each keepalive interval, run the
-                    # fallback done-check (for finishes the event loop never
-                    # sees: late connect, silent producer death, a payload-less
-                    # task_updated) and, if still live, send a keepalive ping.
+                    # No messages this cycle: send a keepalive ping on the
+                    # configured interval so proxies don't reap an idle stream.
                     if message_count == 0:
                         current_time = asyncio.get_running_loop().time()
                         if current_time - last_message_time >= ping_interval:
-                            if await self._stream_is_finished(
-                                task_id, stream_topic, stream_ever_existed
-                            ):
-                                logger.info(
-                                    f"Ending SSE stream for task {task_id}: "
-                                    "fallback check found it finished"
-                                )
-                                break
                             yield ":ping\n\n"
                             last_message_time = current_time
                         await asyncio.sleep(0.1)
@@ -240,41 +235,5 @@ async def stream_task_events(
             # others. The sliding TTL reclaims it instead.
             logger.info(f"SSE stream for task {task_id} has ended")
 
-    async def _stream_is_finished(
-        self,
-        task_id: str,
-        stream_topic: str,
-        stream_ever_existed: bool,
-    ) -> bool:
-        """
-        Fallback terminal check. True if the task is terminal, or the task's
-        status could not be confirmed AND its topic was reclaimed after having
-        existed (covers a hard-deleted task or a failing lookup). A confirmed
-        non-terminal task is treated as live even if its idle key was
-        reclaimed, and a not-yet-written topic is not treated as finished.
-        """
-        try:
-            task = await self.task_service.get_task(id=task_id)
-        except Exception as e:
-            logger.warning(
-                f"Terminal-state lookup failed for task {task_id}; "
-                f"treating status as unknown: {e}"
-            )
-            task = None
-        # Normal finish: the DB says the task is done, so no more XADDs.
-        if task is not None and task.status in _TERMINAL_TASK_STATUSES:
-            return True
-        # Vanished topic, but only when the status is unknown (lookup failed /
-        # hard delete). A confirmed non-terminal task is still live even if its
-        # idle key was reclaimed by the sliding TTL — a later XADD recreates the
-        # key, so we must not close its stream on the expired key alone.
-        if (
-            task is None
-            and stream_ever_existed
-            and not await self.stream_repository.stream_exists(stream_topic)
-        ):
-            return True
-        return False
-
 
 DStreamsUseCase = Annotated[StreamsUseCase, Depends(StreamsUseCase)]

agentex/tests/integration/test_task_stream.py

@@ -716,7 +716,7 @@ async def test_cleanup_does_not_delete_shared_stream_on_disconnect(
 
         # Ensure the topic exists.
         await repo.send_data(stream_topic, {"type": "error", "message": "seed"})
-        assert await repo.stream_exists(stream_topic), "precondition: topic exists"
+        assert bool(await repo.redis.exists(stream_topic)), "precondition: topic exists"
 
         # Run one subscriber briefly, then disconnect it (simulating one viewer
         # of a task that other viewers are still watching).
@@ -738,7 +738,7 @@ async def brief_reader():
             pass
 
         # The shared topic must survive one subscriber leaving.
-        assert await repo.stream_exists(stream_topic), (
+        assert bool(await repo.redis.exists(stream_topic)), (
             "Topic was deleted when a single subscriber disconnected — other "
             "live viewers of this task would lose their stream."
         )
@@ -796,26 +796,24 @@ async def test_running_task_stream_survives_reclaimed_key(
         self, test_agent_and_task, streams_use_case
     ):
         """
-        Regression for the review finding "expired key ends live stream": a
-        still-RUNNING task whose idle stream key is reclaimed by the sliding TTL
-        must NOT be closed. A later XADD recreates the key, so the fallback must
-        only treat a vanished topic as terminal when the task status is unknown,
-        never when the task is confirmed non-terminal.
+        A still-RUNNING task whose idle stream key is reclaimed by the sliding
+        TTL must NOT have its stream closed — a later XADD recreates the key.
+        Termination is driven by the terminal task_updated event (and the
+        connect-time terminal check), never by a vanished key, so a live task's
+        stream stays open regardless of key reclamation.
         """
         from src.utils.stream_topics import get_task_event_stream_topic
 
-        # Fast fallback cadence so the vanished-key check gets a chance to (wrongly)
-        # fire within the test window if the regression is present.
+        # Short keepalive so several idle cycles elapse within the test window.
         streams_use_case.environment_variables.SSE_KEEPALIVE_PING_INTERVAL = 1
 
         _agent, task = test_agent_and_task  # created in RUNNING state
         stream_topic = get_task_event_stream_topic(task_id=task.id)
         repo = streams_use_case.stream_repository
 
-        # Seed the topic so it has existed (arms the vanished-topic fallback), then
-        # subscribe.
+        # Seed the topic, subscribe, then reclaim the key mid-stream.
         await repo.send_data(stream_topic, {"type": "error", "message": "seed"})
-        assert await repo.stream_exists(stream_topic), "precondition: topic exists"
+        assert bool(await repo.redis.exists(stream_topic)), "precondition: topic exists"
 
         async def reader():
             try:
@@ -832,7 +830,9 @@ async def reader():
         # reclaiming the key while the task is still RUNNING.
         await asyncio.sleep(0.5)
         await repo.cleanup_stream(stream_topic)
-        assert not await repo.stream_exists(stream_topic), "precondition: key reclaimed"
+        assert not bool(await repo.redis.exists(stream_topic)), (
+            "precondition: key reclaimed"
+        )
 
         # Give the fallback several cycles to (wrongly) close the stream.
         await asyncio.sleep(4)