feat(streaming): cleanup task Redis stream on terminal status transitions

The SDK defines RedisStreamRepository.cleanup_stream(topic) but never
calls it. As a result, every task:* stream that an agent writes to
during its lifetime stays in Redis forever, even after the task
completes. For long-lived clusters this leaks unbounded memory.

This change wires cleanup_stream into all terminal status transitions
on TasksService: cancel, complete, fail, terminate, timeout, and
delete. After the corresponding agentex API call returns successfully,
the task's Redis stream (task:{task_id}) is deleted best-effort -- any
failure is logged and swallowed so cleanup issues never break the
lifecycle call itself.

TasksService now accepts an optional stream_repository in __init__.
The two existing construction sites pass one in:
- get_all_activities() reuses the same RedisStreamRepository already
  constructed for StreamingService
- TasksModule's lazy default constructs its own when no service is
  provided

Note: this only fires when an application actually calls a terminal
transition method (e.g. adk.tasks.complete). Workflows that exit
without calling these still leak streams; that is a separate concern
for the application layer.

Author: x

src/agentex/lib/adk/_modules/tasks.py

@@ -7,6 +7,7 @@
 
 from agentex import AsyncAgentex  # noqa: F401
 from agentex.lib.adk.utils._modules.client import create_async_agentex_client
+from agentex.lib.core.adapters.streams.adapter_redis import RedisStreamRepository
 from agentex.lib.core.services.adk.tasks import TasksService
 from agentex.lib.core.temporal.activities.activity_helpers import ActivityHelpers
 from agentex.lib.core.temporal.activities.adk.tasks_activities import (
@@ -42,8 +43,11 @@ def __init__(
         if tasks_service is None:
             agentex_client = create_async_agentex_client()
             tracer = AsyncTracer(agentex_client)
+            stream_repository = RedisStreamRepository()
             self._tasks_service = TasksService(
-                agentex_client=agentex_client, tracer=tracer
+                agentex_client=agentex_client,
+                tracer=tracer,
+                stream_repository=stream_repository,
             )
         else:
             self._tasks_service = tasks_service

src/agentex/lib/core/services/adk/tasks.py

@@ -7,20 +7,41 @@
 from agentex.lib.utils.temporal import heartbeat_if_in_workflow
 from agentex.lib.core.tracing.tracer import AsyncTracer
 from agentex.types.task_retrieve_response import TaskRetrieveResponse
+from agentex.lib.core.adapters.streams.port import StreamRepository
 from agentex.types.task_query_workflow_response import TaskQueryWorkflowResponse
 from agentex.types.task_retrieve_by_name_response import TaskRetrieveByNameResponse
 
 logger = make_logger(__name__)
 
 
+def _stream_topic(task_id: str) -> str:
+    return f"task:{task_id}"
+
+
 class TasksService:
     def __init__(
         self,
         agentex_client: AsyncAgentex,
         tracer: AsyncTracer,
+        stream_repository: StreamRepository | None = None,
     ):
         self._agentex_client = agentex_client
         self._tracer = tracer
+        self._stream_repository = stream_repository
+
+    async def _cleanup_task_stream(self, task_id: str) -> None:
+        """Best-effort delete of the task's Redis stream.
+
+        Called after a terminal status transition (cancel/complete/fail/
+        terminate/timeout/delete). Failures are logged and swallowed so
+        cleanup issues never break the lifecycle call.
+        """
+        if self._stream_repository is None:
+            return
+        try:
+            await self._stream_repository.cleanup_stream(_stream_topic(task_id))
+        except Exception as e:
+            logger.warning(f"Failed to cleanup task stream {task_id}: {e}")
 
     async def get_task(
         self,
@@ -75,6 +96,8 @@ async def delete_task(
                 task_model = await self._agentex_client.tasks.delete_by_name(task_name=task_name)
             else:
                 raise ValueError("Either task_id or task_name must be provided.")
+            if task_model.id:
+                await self._cleanup_task_stream(task_model.id)
             if span:
                 span.output = task_model.model_dump()
             return task_model
@@ -94,6 +117,7 @@ async def cancel_task(
         ) as span:
             heartbeat_if_in_workflow("cancel task")
             task_model = await self._agentex_client.tasks.cancel(task_id=task_id, reason=reason)
+            await self._cleanup_task_stream(task_id)
             if span:
                 span.output = task_model.model_dump()
             return task_model
@@ -113,6 +137,7 @@ async def complete_task(
         ) as span:
             heartbeat_if_in_workflow("complete task")
             task_model = await self._agentex_client.tasks.complete(task_id=task_id, reason=reason)
+            await self._cleanup_task_stream(task_id)
             if span:
                 span.output = task_model.model_dump()
             return task_model
@@ -132,6 +157,7 @@ async def fail_task(
         ) as span:
             heartbeat_if_in_workflow("fail task")
             task_model = await self._agentex_client.tasks.fail(task_id=task_id, reason=reason)
+            await self._cleanup_task_stream(task_id)
             if span:
                 span.output = task_model.model_dump()
             return task_model
@@ -151,6 +177,7 @@ async def terminate_task(
         ) as span:
             heartbeat_if_in_workflow("terminate task")
             task_model = await self._agentex_client.tasks.terminate(task_id=task_id, reason=reason)
+            await self._cleanup_task_stream(task_id)
             if span:
                 span.output = task_model.model_dump()
             return task_model
@@ -170,6 +197,7 @@ async def timeout_task(
         ) as span:
             heartbeat_if_in_workflow("timeout task")
             task_model = await self._agentex_client.tasks.timeout(task_id=task_id, reason=reason)
+            await self._cleanup_task_stream(task_id)
             if span:
                 span.output = task_model.model_dump()
             return task_model

src/agentex/lib/core/temporal/activities/__init__.py

@@ -92,6 +92,7 @@ def get_all_activities(sgp_client=None):
     tasks_service = TasksService(
         agentex_client=agentex_client,
         tracer=tracer,
+        stream_repository=stream_repository,
     )
     tracing_service = TracingService(
         tracer=tracer,