fix(events): preserve subclass validation aliases

src/google/adk/events/event.py

@@ -20,6 +20,8 @@
 from google.adk.platform import time as platform_time
 from google.adk.platform import uuid as platform_uuid
 from pydantic import alias_generators
+from pydantic import AliasChoices
+from pydantic import AliasPath
 from pydantic import BaseModel
 from pydantic import ConfigDict
 from pydantic import Field
@@ -206,6 +208,19 @@ def _accept_convenience_kwargs(cls, data: Any) -> Any:
     for f in cls.model_fields.values():
       if f.alias:
         field_names.add(f.alias)
+      # A validation alias may consume an entire top-level input container.
+      validation_alias = f.validation_alias
+      if isinstance(validation_alias, str):
+        field_names.add(validation_alias)
+      elif isinstance(validation_alias, (AliasChoices, AliasPath)):
+        alias_paths = (
+            [validation_alias.path]
+            if isinstance(validation_alias, AliasPath)
+            else validation_alias.convert_to_aliases()
+        )
+        for alias_path in alias_paths:
+          if alias_path and isinstance(alias_path[0], str):
+            field_names.add(alias_path[0])
     message = None if 'message' in field_names else data.pop('message', None)
     state = None if 'state' in field_names else data.pop('state', None)
     route = None if 'route' in field_names else data.pop('route', None)

tests/unittests/events/test_event.py

@@ -598,6 +598,53 @@ class _AliasedEvent(Event):
     assert event.custom_route == 'custom_target'
     assert event.actions.route is None
 
+  @pytest.mark.parametrize(
+      'keyword', ['message', 'state', 'route', 'node_path']
+  )
+  @pytest.mark.parametrize(
+      'alias_kind', ['string', 'choices', 'path', 'choices_with_path']
+  )
+  def test_subclass_validation_alias_takes_the_convenience_kwarg(
+      self, keyword, alias_kind
+  ):
+    """Subclass input aliases receive their data without convenience routing."""
+    alias: str | pydantic.AliasChoices | pydantic.AliasPath = keyword
+    value: str | dict[str, str] = 'custom_value'
+    if alias_kind == 'choices':
+      alias = pydantic.AliasChoices('other', keyword)
+    elif alias_kind == 'path':
+      alias = pydantic.AliasPath(keyword, 'value')
+      value = {'value': value}
+    elif alias_kind == 'choices_with_path':
+      alias = pydantic.AliasChoices(
+          'other', pydantic.AliasPath(keyword, 'value')
+      )
+      value = {'value': value}
+
+    class _AliasedEvent(Event):
+      custom_value: str = pydantic.Field(validation_alias=alias)
+
+    event = _AliasedEvent(**{keyword: value})
+
+    assert event.custom_value == 'custom_value'
+    assert event.content is None
+    assert event.actions.state_delta == {}
+    assert event.actions.route is None
+    assert event.node_info.path == ''
+
+  def test_nested_validation_alias_does_not_take_top_level_route(self):
+    """A nested alias leaf does not shadow an unrelated convenience kwarg."""
+
+    class _AliasedEvent(Event):
+      custom_value: str = pydantic.Field(
+          validation_alias=pydantic.AliasPath('payload', 'route')
+      )
+
+    event = _AliasedEvent(payload={'route': 'custom_value'}, route='next')
+
+    assert event.custom_value == 'custom_value'
+    assert event.actions.route == 'next'
+
 
 @pytest.mark.parametrize(
     'path, name, run_id, parent_run_id',

tests/unittests/workflow/test_workflow.py

@@ -216,6 +216,26 @@ async def test_sequential_two_nodes():
   assert by_node.index(('NodeA', 'Hello')) < by_node.index(('NodeB', 'World'))
 
 
+async def test_event_validation_alias_does_not_route_workflow():
+  """A custom event's aliased data does not become a workflow route."""
+
+  class _DomainEvent(Event):
+    destination: str = Field(validation_alias='route')
+
+  async def emit_event():
+    yield _DomainEvent(route='domain_destination', output='ready')
+
+  def consume(node_input: str) -> str:
+    return f'{node_input} processed'
+
+  wf = Workflow(name='wf', edges=[(START, emit_event, consume)])
+
+  events, _, _ = await _run_workflow(wf)
+
+  assert ('consume', 'ready processed') in _output_by_node(events)
+  assert all(event.actions.route is None for event in events)
+
+
 @pytest.mark.asyncio
 async def test_finished_turn_frees_its_session_without_gc():
   """A turn's session copy is freed by reference counting once the turn ends."""