fix: Append metadata and context id when processing TaskStatusUpdateE… (#238)

…vent, and add more error logs

1. Fix the bug that during TaskManager.save_task_event, metadata from
the updated task event is lost
2. During EventQueue processing, if any of the processing asyncio task
failed with an unexpected exception, this exception is silently caught
but never exposed to user or logged in the console. Add more error logs
to make errors visible during debugging.

# Description

Thank you for opening a Pull Request!
Before submitting your PR, there are a few things you can do to make
sure it goes smoothly:

- [x] Follow the [`CONTRIBUTING`
Guide](https://github.com/a2aproject/a2a-python/blob/main/CONTRIBUTING.md).
- [x] Make your Pull Request title in the
<https://www.conventionalcommits.org/> specification.
- Important Prefixes for
[release-please](https://github.com/googleapis/release-please):
- `fix:` which represents bug fixes, and correlates to a
[SemVer](https://semver.org/) patch.
- `feat:` represents a new feature, and correlates to a SemVer minor.
- `feat!:`, or `fix!:`, `refactor!:`, etc., which represent a breaking
change (indicated by the `!`) and will result in a SemVer major.
- [ ] Ensure the tests and linter pass (Run `nox -s format` from the
repository root to format)
- [ ] Appropriate docs were updated (if necessary)

Fixes #<issue_number_goes_here> 🦕

Author: pwwpche

src/a2a/server/events/event_consumer.py

@@ -138,6 +138,10 @@ async def consume_all(self) -> AsyncGenerator[Event]:
                 # python 3.12 and get a queue empty error on an open queue
                 if self.queue.is_closed():
                     break
+            except Exception as e:
+                logger.error(f'Stopping event consumption due to exception: {e}')
+                self._exception = e
+                continue
 
     def agent_task_callback(self, agent_task: asyncio.Task[None]) -> None:
         """Callback to handle exceptions from the agent's execution task.

src/a2a/server/request_handlers/default_request_handler.py

@@ -248,7 +248,9 @@ async def on_message_send(
                 raise ServerError(
                     InternalError(message='Task ID mismatch in agent response')
                 )
-
+        except Exception as e:
+            logger.error(f'Agent execution failed. Error: {e}')
+            raise
         finally:
             if interrupted:
                 # TODO: Track this disconnected cleanup task.

src/a2a/server/tasks/task_manager.py

@@ -130,7 +130,10 @@ async def save_task_event(
                     task.history = [task.status.message]
                 else:
                     task.history.append(task.status.message)
-
+            if event.metadata:
+                if not task.metadata:
+                    task.metadata = {}
+                task.metadata.update(event.metadata)
             task.status = event.status
         else:
             logger.debug('Appending artifact to task %s', task.id)

tests/server/tasks/test_task_manager.py

@@ -125,6 +125,27 @@ async def test_save_task_event_artifact_update(
     updated_task.artifacts = [new_artifact]
     mock_task_store.save.assert_called_once_with(updated_task)
 
+@pytest.mark.asyncio
+async def test_save_task_event_metadata_update(
+    task_manager: TaskManager, mock_task_store: AsyncMock
+) -> None:
+    """Test saving an updated metadata for an existing task."""
+    initial_task = Task(**MINIMAL_TASK)
+    mock_task_store.get.return_value = initial_task
+    new_metadata = {"meta_key_test": "meta_value_test"}
+    
+    event = TaskStatusUpdateEvent(
+        taskId=MINIMAL_TASK['id'],
+        contextId=MINIMAL_TASK['contextId'],
+        metadata=new_metadata,
+        status=TaskStatus(state=TaskState.working),
+        final=False,
+    )
+    await task_manager.save_task_event(event)
+
+    updated_task = mock_task_store.save.call_args.args[0]
+    assert updated_task.metadata == new_metadata
+
 
 @pytest.mark.asyncio
 async def test_ensure_task_existing(