test(agentex): cover SSE fallback termination and reclaimed-key edge

Add two integration regression tests that had no coverage:
- test_stream_ends_on_late_connect_to_terminal_task: connecting after a task is
  already terminal never delivers a terminal event, so the keepalive-cadence
  fallback must end the stream.
- test_running_task_stream_survives_reclaimed_key: a still-RUNNING task whose
  idle stream key is reclaimed by the sliding TTL must NOT be closed (guards the
  review fix in the prior commit).

Both shrink SSE_KEEPALIVE_PING_INTERVAL so the fallback runs in ~1s.

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

Author: deepthi-rao-scale

agentex/tests/integration/test_task_stream.py

@@ -744,3 +744,110 @@ async def brief_reader():
         )
 
         print("✅ Shared stream survives a single subscriber disconnecting")
+
+    async def test_stream_ends_on_late_connect_to_terminal_task(
+        self, test_agent_and_task, tasks_use_case, streams_use_case
+    ):
+        """
+        Fallback termination path: a viewer that connects *after* the task has
+        already reached a terminal state never receives a terminal task_updated
+        event (it is behind the snapshot cursor), so the event-driven exit can't
+        fire. The keepalive-cadence fallback must still end the stream instead of
+        blocking on the topic forever.
+        """
+        # Shrink the keepalive interval so the fallback check runs within ~1s
+        # instead of the production 15s.
+        streams_use_case.environment_variables.SSE_KEEPALIVE_PING_INTERVAL = 1
+
+        _agent, task = test_agent_and_task
+
+        # Finish the task BEFORE anyone subscribes. The terminal task_updated is
+        # XADDed to the stream now, so a later subscriber snapshots past it.
+        completed = await tasks_use_case.complete_task(id=task.id)
+        assert completed.status == TaskStatus.COMPLETED
+
+        async def drain_until_end():
+            # Returns normally only if the generator terminates by itself.
+            async for _event_data in streams_use_case.stream_task_events(
+                task_id=task.id
+            ):
+                pass
+
+        reader_task = asyncio.create_task(drain_until_end())
+
+        # With no terminal event to react to, only the fallback can end it.
+        try:
+            await asyncio.wait_for(reader_task, timeout=10)
+        except TimeoutError as err:
+            reader_task.cancel()
+            try:
+                await reader_task
+            except asyncio.CancelledError:
+                pass
+            raise AssertionError(
+                "SSE stream to an already-terminal task did not self-close — the "
+                "fallback termination check is not firing, so late-connect "
+                "subscriptions leak as zombies."
+            ) from err
+
+        print("✅ Stream self-closes when connecting to an already-terminal task")
+
+    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.
+        """
+        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.
+        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.
+        await repo.send_data(stream_topic, {"type": "error", "message": "seed"})
+        assert await repo.stream_exists(stream_topic), "precondition: topic exists"
+
+        async def reader():
+            try:
+                async for _event_data in streams_use_case.stream_task_events(
+                    task_id=task.id
+                ):
+                    pass
+            except asyncio.CancelledError:
+                pass
+
+        reader_task = asyncio.create_task(reader())
+
+        # Let the reader connect and go idle, then simulate the sliding TTL
+        # 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"
+
+        # Give the fallback several cycles to (wrongly) close the stream.
+        await asyncio.sleep(4)
+
+        try:
+            assert not reader_task.done(), (
+                "SSE stream for a still-RUNNING task was closed after its idle key "
+                "was reclaimed — the fallback must not treat a reclaimed key as "
+                "terminal when the task is confirmed non-terminal."
+            )
+        finally:
+            reader_task.cancel()
+            try:
+                await reader_task
+            except asyncio.CancelledError:
+                pass
+
+        print("✅ Running task's stream stays open when its idle key is reclaimed")