refactor(agentex): single canonical terminal/non-terminal task-status sets

Define NON_TERMINAL_TASK_STATUSES ({RUNNING, INTERRUPTED}) and its complement
TERMINAL_TASK_STATUSES once on the TaskStatus entity, and reference them from
both call sites instead of duplicating the membership:
- streams_use_case: imports TERMINAL_TASK_STATUSES (was a local frozenset).
- tasks_use_case: _TERMINAL_TRANSITION_SOURCES now = NON_TERMINAL_TASK_STATUSES
  (was an inline (RUNNING, INTERRUPTED) tuple).

A newly added status is terminal by default unless explicitly added to the
non-terminal set, keeping the two views consistent by construction.

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

Author: deepthi-rao-scale

agentex/src/domain/entities/tasks.py

@@ -29,6 +29,19 @@ class TaskStatus(str, Enum):
     DELETED = "DELETED"
 
 
+# Canonical status partition (single source of truth for both the task-status
+# state machine and SSE stream termination).
+#
+# Non-terminal: a turn is in flight (RUNNING) or the task is paused and
+# resumable (INTERRUPTED). A task may only transition to a terminal status from
+# one of these. Terminal is the complement: no further events are produced and
+# no new turns are accepted. Defining terminal as the complement means a newly
+# added status is treated as terminal unless it is explicitly listed as
+# non-terminal here.
+NON_TERMINAL_TASK_STATUSES = frozenset({TaskStatus.RUNNING, TaskStatus.INTERRUPTED})
+TERMINAL_TASK_STATUSES = frozenset(TaskStatus) - NON_TERMINAL_TASK_STATUSES
+
+
 class TaskEntity(BaseModel):
     id: str = Field(
         ...,

agentex/src/domain/use_cases/streams_use_case.py

@@ -15,25 +15,13 @@
     TaskStreamTaskUpdatedEventEntity,
     convert_task_stream_event_to_entity,
 )
-from src.domain.entities.tasks import TaskStatus
+from src.domain.entities.tasks import TERMINAL_TASK_STATUSES
 from src.domain.services.task_service import DAgentTaskService
 from src.utils.logging import make_logger
 from src.utils.stream_topics import get_task_event_stream_topic
 
 logger = make_logger(__name__)
 
-# Statuses after which no more events are produced; a subscription must end.
-_TERMINAL_TASK_STATUSES = frozenset(
-    {
-        TaskStatus.CANCELED,
-        TaskStatus.COMPLETED,
-        TaskStatus.FAILED,
-        TaskStatus.TERMINATED,
-        TaskStatus.TIMED_OUT,
-        TaskStatus.DELETED,
-    }
-)
-
 
 class StreamsUseCase:
     def __init__(
@@ -129,7 +117,7 @@ async def stream_task_events(
         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:
+        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)
@@ -169,7 +157,7 @@ async def stream_task_events(
                         if (
                             isinstance(data, TaskStreamTaskUpdatedEventEntity)
                             and data.task is not None
-                            and data.task.status in _TERMINAL_TASK_STATUSES
+                            and data.task.status in TERMINAL_TASK_STATUSES
                         ):
                             logger.info(
                                 f"Ending SSE stream for task {task_id}: received "

agentex/src/domain/use_cases/tasks_use_case.py

@@ -3,7 +3,12 @@
 from fastapi import Depends
 
 from src.adapters.crud_store.exceptions import ItemDoesNotExist
-from src.domain.entities.tasks import TaskEntity, TaskRelationships, TaskStatus
+from src.domain.entities.tasks import (
+    NON_TERMINAL_TASK_STATUSES,
+    TaskEntity,
+    TaskRelationships,
+    TaskStatus,
+)
 from src.domain.exceptions import ClientError
 from src.domain.services.task_service import DAgentTaskService
 from src.utils.logging import make_logger
@@ -132,10 +137,11 @@ async def update_mutable_fields_on_task(
 
         return task_entity
 
-    # Non-terminal statuses a task can be transitioned to a terminal status from.
-    # RUNNING is the normal case; INTERRUPTED is also valid so an interrupted
-    # (paused, still-continuable) task can still be canceled/completed/etc later.
-    _TERMINAL_TRANSITION_SOURCES = (TaskStatus.RUNNING, TaskStatus.INTERRUPTED)
+    # Statuses a task can be transitioned to a terminal status from: RUNNING is
+    # the normal case; INTERRUPTED (paused, still-continuable) can also still be
+    # canceled/completed/etc later. This is exactly the canonical non-terminal
+    # set — its complement is what SSE streaming treats as terminal.
+    _TERMINAL_TRANSITION_SOURCES = NON_TERMINAL_TASK_STATUSES
 
     async def _transition_to_terminal(
         self,