refactor: define terminal and interrupted task states once in a2a.utils.constants

The set of terminal task states was defined four times (active_task,
default_request_handler, the cluster VersionedDatabaseTaskStore and
TaskUpdater), and default_request_handler_v2 imported the task state sets
from the ActiveTask module. Define TERMINAL_TASK_STATES and
INTERRUPTED_TASK_STATES once, as frozensets, in a2a.utils.constants and use
them everywhere.

The previous public names (active_task.TERMINAL_TASK_STATES,
active_task.INTERRUPTED_TASK_STATES and
default_request_handler.TERMINAL_TASK_STATES) still resolve to the shared
sets, so existing imports keep working. The private copies in the cluster
store and TaskUpdater are removed. The legacy EventConsumer stop condition is
left as is, since it intentionally covers a different set of states.

a2a.utils.constants now imports a2a.types.a2a_pb2. a2a.types imports
a2a.utils.errors, so errors must not import constants.

Verification:
- Every a2a module imported first in a fresh interpreter: no new failures
  compared with upstream/main (the same 3 pre-existing failures on both).
- 27 first-import orders and ordered pairs of constants, errors, types and
  a2a.utils: all pass.
- ./scripts/lint.sh: ruff and ty pass (3 pre-existing ty warnings in
  unrelated files).
- pytest: 2487 passed, 203 skipped, 4 xfailed, 1 xpassed; the 4 cross-version
  integration errors also occur on upstream/main in this environment.
- scripts/test_install_smoke.sh: all 6 profiles pass.

Author: mykytanetipa

src/a2a/server/agent_execution/active_task.py

@@ -75,6 +75,10 @@
     TaskStatus,
     TaskStatusUpdateEvent,
 )
+from a2a.utils.constants import (
+    INTERRUPTED_TASK_STATES,
+    TERMINAL_TASK_STATES,
+)
 from a2a.utils.errors import (
     InternalError,
     InvalidAgentResponseError,
@@ -86,18 +90,6 @@
 logger = logging.getLogger(__name__)
 
 
-TERMINAL_TASK_STATES = {
-    TaskState.TASK_STATE_COMPLETED,
-    TaskState.TASK_STATE_CANCELED,
-    TaskState.TASK_STATE_FAILED,
-    TaskState.TASK_STATE_REJECTED,
-}
-INTERRUPTED_TASK_STATES = {
-    TaskState.TASK_STATE_AUTH_REQUIRED,
-    TaskState.TASK_STATE_INPUT_REQUIRED,
-}
-
-
 class _RequestStarted:
     def __init__(self, request_id: uuid.UUID, request_context: RequestContext):
         self.request_id = request_id

src/a2a/server/cluster/database_task_store.py

@@ -44,20 +44,12 @@
     Task,
     TaskState,
 )
+from a2a.utils.constants import TERMINAL_TASK_STATES
 from a2a.utils.proto_utils import to_stream_response
 
 
 logger = logging.getLogger(__name__)
 
-_TERMINAL_STATES = frozenset(
-    {
-        TaskState.TASK_STATE_COMPLETED,
-        TaskState.TASK_STATE_CANCELED,
-        TaskState.TASK_STATE_FAILED,
-        TaskState.TASK_STATE_REJECTED,
-    }
-)
-
 # DB contention retry
 _MAX_ATTEMPTS = 5
 _RETRY_DELAY_S = 0.02
@@ -195,7 +187,10 @@ async def _save_once(
             if task.status.state == TaskState.TASK_STATE_CANCELED:
                 stored = await self._lock_version(session, task.id, owner)
                 current = await self._current_task(session, task.id, owner)
-                if current is None or current.status.state in _TERMINAL_STATES:
+                if (
+                    current is None
+                    or current.status.state in TERMINAL_TASK_STATES
+                ):
                     raise ConcurrentTaskModificationError(task.id)
                 new_version = (stored or 0) + 1
                 if stored is None:

src/a2a/server/request_handlers/default_request_handler.py

@@ -53,6 +53,7 @@
     TaskPushNotificationConfig,
     TaskState,
 )
+from a2a.utils.constants import TERMINAL_TASK_STATES
 from a2a.utils.errors import (
     ExtendedAgentCardNotConfiguredError,
     InternalError,
@@ -74,13 +75,6 @@
 
 logger = logging.getLogger(__name__)
 
-TERMINAL_TASK_STATES = {
-    TaskState.TASK_STATE_COMPLETED,
-    TaskState.TASK_STATE_CANCELED,
-    TaskState.TASK_STATE_FAILED,
-    TaskState.TASK_STATE_REJECTED,
-}
-
 
 @trace_class(kind=SpanKind.SERVER)
 class LegacyRequestHandler(RequestHandler):

src/a2a/server/request_handlers/default_request_handler_v2.py

@@ -15,10 +15,6 @@
     RequestContextBuilder,
     SimpleRequestContextBuilder,
 )
-from a2a.server.agent_execution.active_task import (
-    INTERRUPTED_TASK_STATES,
-    TERMINAL_TASK_STATES,
-)
 from a2a.server.agent_execution.active_task_registry import ActiveTaskRegistry
 from a2a.server.agent_execution.plugin_agent_executor import (
     PluginAgentExecutor,
@@ -57,6 +53,10 @@
     TaskState,
     TaskStatusUpdateEvent,
 )
+from a2a.utils.constants import (
+    INTERRUPTED_TASK_STATES,
+    TERMINAL_TASK_STATES,
+)
 from a2a.utils.errors import (
     ExtendedAgentCardNotConfiguredError,
     InternalError,

src/a2a/server/tasks/task_updater.py

@@ -21,6 +21,7 @@
     TaskStatus,
     TaskStatusUpdateEvent,
 )
+from a2a.utils.constants import TERMINAL_TASK_STATES
 
 
 class TaskUpdater:
@@ -51,12 +52,6 @@ def __init__(
         self.context_id = context_id
         self._lock = asyncio.Lock()
         self._terminal_state_reached = False
-        self._terminal_states = {
-            TaskState.TASK_STATE_COMPLETED,
-            TaskState.TASK_STATE_CANCELED,
-            TaskState.TASK_STATE_FAILED,
-            TaskState.TASK_STATE_REJECTED,
-        }
         self._artifact_id_generator = (
             artifact_id_generator if artifact_id_generator else UUIDGenerator()
         )
@@ -105,7 +100,7 @@ async def update_status(
                 metadata=metadata,
                 status=status,
             )
-            if state in self._terminal_states:
+            if state in TERMINAL_TASK_STATES:
                 self._terminal_state_reached = True
             await self.event_queue.enqueue_event(event)
 

src/a2a/utils/constants.py

@@ -2,6 +2,8 @@
 
 from enum import Enum
 
+from a2a.types.a2a_pb2 import TaskState
+
 
 AGENT_CARD_WELL_KNOWN_PATH = '/.well-known/agent-card.json'
 DEFAULT_RPC_URL = '/'
@@ -29,3 +31,19 @@ class TransportProtocol(str, Enum):
 PROTOCOL_VERSION_1_0 = '1.0'
 PROTOCOL_VERSION_0_3 = '0.3'
 PROTOCOL_VERSION_CURRENT = PROTOCOL_VERSION_1_0
+
+TERMINAL_TASK_STATES: frozenset[TaskState.ValueType] = frozenset(
+    {
+        TaskState.TASK_STATE_COMPLETED,
+        TaskState.TASK_STATE_CANCELED,
+        TaskState.TASK_STATE_FAILED,
+        TaskState.TASK_STATE_REJECTED,
+    }
+)
+
+INTERRUPTED_TASK_STATES: frozenset[TaskState.ValueType] = frozenset(
+    {
+        TaskState.TASK_STATE_INPUT_REQUIRED,
+        TaskState.TASK_STATE_AUTH_REQUIRED,
+    }
+)

tests/utils/test_constants.py

@@ -1,5 +1,7 @@
 """Tests for a2a.utils.constants module."""
 
+from a2a.server.agent_execution import active_task
+from a2a.server.request_handlers import default_request_handler
 from a2a.utils import constants
 
 
@@ -24,3 +26,15 @@ def test_protocol_versions():
     """Test protocol version constants."""
     assert constants.PROTOCOL_VERSION_1_0 == '1.0'
     assert constants.PROTOCOL_VERSION_CURRENT == '1.0'
+
+
+def test_task_state_sets_keep_their_previous_import_paths():
+    """Test that the previous public names still resolve to the shared sets."""
+    assert active_task.TERMINAL_TASK_STATES is constants.TERMINAL_TASK_STATES
+    assert (
+        active_task.INTERRUPTED_TASK_STATES is constants.INTERRUPTED_TASK_STATES
+    )
+    assert (
+        default_request_handler.TERMINAL_TASK_STATES
+        is constants.TERMINAL_TASK_STATES
+    )