fix: Add Input Validation for Task IDs in TaskManager (#310)

Previously, the `TaskManager` class in
`a2a/server/tasks/task_manager.py` did not perform any validation on the
`task_id` during initialization. This could lead to silent failures or
inconsistencies in task storage if an invalid ID (such as an empty
string) was provided.

### Impact

Adding this validation improves the robustness of the `TaskManager` and
prevents downstream errors in task management, especially for database
or in-memory stores.

### Fix

In `a2a/server/tasks/task_manager.py`, a check has been added to the
`__init__` method of the `TaskManager` to validate the `task_id`. It now
ensures that if a `task_id` is provided, it is a non-empty string. If
the validation fails, a `ValueError` is raised.

A corresponding unit test has been added to
`tests/server/tasks/test_task_manager.py` to verify that the validation
works as expected.

Author: pankaj-bind

src/a2a/server/tasks/task_manager.py

@@ -41,6 +41,9 @@ def __init__(
             initial_message: The `Message` that initiated the task, if any.
                              Used when creating a new task object.
         """
+        if task_id is not None and not (isinstance(task_id, str) and task_id):
+            raise ValueError('Task ID must be a non-empty string')
+
         self.task_id = task_id
         self.context_id = context_id
         self.task_store = task_store

tests/server/tasks/test_task_manager.py

@@ -45,6 +45,20 @@ def task_manager(mock_task_store: AsyncMock) -> TaskManager:
     )
 
 
+@pytest.mark.parametrize('invalid_task_id', ['', 123])
+def test_task_manager_invalid_task_id(
+    mock_task_store: AsyncMock, invalid_task_id: Any
+):
+    """Test that TaskManager raises ValueError for an invalid task_id."""
+    with pytest.raises(ValueError, match='Task ID must be a non-empty string'):
+        TaskManager(
+            task_id=invalid_task_id,
+            context_id='test_context',
+            task_store=mock_task_store,
+            initial_message=None,
+        )
+
+
 @pytest.mark.asyncio
 async def test_get_task_existing(
     task_manager: TaskManager, mock_task_store: AsyncMock