fix(a2a): mark terminal A2A tasks as final responses

src/google/adk/a2a/converters/event_converter.py

@@ -51,6 +51,24 @@
 # Logger
 logger = logging.getLogger("google_adk." + __name__)
 
+_TERMINAL_TASK_STATES = frozenset({
+    _compat.TS_COMPLETED,
+    _compat.TS_FAILED,
+    _compat.TS_CANCELED,
+})
+"""Task states that mean the peer's turn is over."""
+
+
+def _is_terminal_task(a2a_task: Task) -> bool:
+  """Returns whether the task reports a state that ends the peer's turn.
+
+  ``state`` is read defensively: 1.x tasks carry a protobuf ``TaskStatus``
+  whose fields are not always reachable on stand-in objects, and an
+  unreadable state simply means we cannot claim the turn is over.
+  """
+  status = getattr(a2a_task, "status", None)
+  return getattr(status, "state", None) in _TERMINAL_TASK_STATES
+
 
 AdkEventToA2AEventsConverter = Callable[
     [
@@ -256,6 +274,12 @@ def convert_a2a_task_to_event(
         event: Event = convert_a2a_message_to_event(
             message, author, invocation_context, part_converter=part_converter
         )
+        if _is_terminal_task(a2a_task):
+          # See the note in ``to_adk_event.convert_a2a_task_to_event``: a
+          # terminal task is the end of the peer's turn, but the event holds
+          # the peer's tool activity, so ``is_final_response()`` would report
+          # False without this.
+          event.actions.skip_summarization = True
         return event
       except Exception as e:
         logger.error("Failed to convert A2A task message to event: %s", e)

src/google/adk/a2a/converters/to_adk_event.py

@@ -290,6 +290,19 @@ def _extract_genai_metadata(
 """
 
 
+_TERMINAL_TASK_STATES = frozenset({
+    _compat.TS_COMPLETED,
+    _compat.TS_FAILED,
+    _compat.TS_CANCELED,
+})
+"""Task states that mean the peer's turn is over.
+
+``input_required`` and ``auth_required`` are deliberately excluded: those pause
+the turn to ask the caller for something, and they already reach
+``is_final_response()`` through the mock function call built for them.
+"""
+
+
 def _extract_event_actions(metadata: Any) -> EventActions:
   """Extracts ADK event actions from A2A metadata.
 
@@ -534,6 +547,17 @@ def convert_a2a_task_to_event(
         )
     )
 
+    if a2a_task.status.state in _TERMINAL_TASK_STATES:
+      # A terminal task is the whole of the peer's turn, so the event carries
+      # the peer's tool activity alongside its closing text. Without this,
+      # `Event.is_final_response()` sees those function calls/responses and
+      # reports False, and a caller that closes the turn on that helper never
+      # closes it. `skip_summarization` is the existing signal for "this event
+      # is final despite carrying tool activity"; it is also already trusted
+      # from the peer (`_PEER_SETTABLE_ACTION_FIELDS`), so setting it here does
+      # not widen what an A2A response can influence.
+      event_actions.skip_summarization = True
+
     return _create_event(
         output_parts,
         invocation_context,

tests/unittests/a2a/converters/test_event_converter.py

@@ -739,6 +739,85 @@ def test_convert_a2a_task_to_event_with_artifacts_priority(self):
       assert called_message.role == _compat.ROLE_AGENT
       assert called_message.parts == [artifact_part]
 
+  @pytest.mark.parametrize(
+      "terminal_state",
+      [_compat.TS_COMPLETED, _compat.TS_FAILED, _compat.TS_CANCELED],
+  )
+  def test_convert_a2a_task_to_event_terminal_task_is_final_response(
+      self, terminal_state
+  ):
+    """A terminal task carrying tool activity must read as a final response."""
+    task = Task(
+        id="task-1",
+        context_id="context-1",
+        status=_compat.make_task_status(
+            terminal_state, timestamp="2024-01-01T00:00:00Z"
+        ),
+        artifacts=[
+            _compat.make_artifact(
+                artifact_id="art-1",
+                artifact_type="message",
+                parts=[_compat.make_text_part("fr")],
+            )
+        ],
+    )
+
+    def part_converter(part):
+      return [
+          genai_types.Part(
+              function_response=genai_types.FunctionResponse(
+                  id="call-1", name="divide", response={"quotient": 5}
+              )
+          )
+      ]
+
+    result = convert_a2a_task_to_event(
+        task,
+        "test-author",
+        self.mock_invocation_context,
+        part_converter=part_converter,
+    )
+
+    assert len(result.get_function_responses()) == 1
+    assert result.actions.skip_summarization is True
+    assert result.is_final_response() is True
+
+  def test_convert_a2a_task_to_event_working_task_is_not_final(self):
+    """An in-flight task must keep reporting that the turn is still open."""
+    task = Task(
+        id="task-1",
+        context_id="context-1",
+        status=_compat.make_task_status(
+            _compat.TS_WORKING, timestamp="2024-01-01T00:00:00Z"
+        ),
+        artifacts=[
+            _compat.make_artifact(
+                artifact_id="art-1",
+                artifact_type="message",
+                parts=[_compat.make_text_part("fr")],
+            )
+        ],
+    )
+
+    def part_converter(part):
+      return [
+          genai_types.Part(
+              function_response=genai_types.FunctionResponse(
+                  id="call-1", name="divide", response={"quotient": 5}
+              )
+          )
+      ]
+
+    result = convert_a2a_task_to_event(
+        task,
+        "test-author",
+        self.mock_invocation_context,
+        part_converter=part_converter,
+    )
+
+    assert result.actions.skip_summarization is None
+    assert result.is_final_response() is False
+
   def test_convert_a2a_task_to_event_with_status_message(self):
     """Test convert_a2a_task_to_event with status message (no artifacts)."""
 

tests/unittests/a2a/converters/test_to_adk.py

@@ -62,6 +62,53 @@ def _make_a2a_part_for_test(metadata=None):
     return m
 
 
+def _tool_activity_part_converter(part):
+  """Turns the "fc"/"fr" marker parts below into real tool activity."""
+  text = _compat.part_text(part)
+  if text == "fc":
+    return [
+        genai_types.Part(
+            function_call=genai_types.FunctionCall(
+                id="call-1", name="divide", args={"a": 10, "b": 0}
+            )
+        )
+    ]
+  if text == "fr":
+    return [
+        genai_types.Part(
+            function_response=genai_types.FunctionResponse(
+                id="call-1",
+                name="divide",
+                response={"error": "cannot divide by zero"},
+            )
+        )
+    ]
+  return [genai_types.Part.from_text(text=text)]
+
+
+def _make_task_with_tool_activity(state):
+  """A non-streaming peer turn: narration, a tool call, its result, an answer.
+
+  This is what `message/send` returns for a peer whose card leaves
+  `capabilities.streaming` at False, which is the default for `to_a2a(...)`.
+  """
+  return Task(
+      id="task-1",
+      context_id="context-1",
+      status=_compat.make_task_status(state, timestamp="2024-01-01T00:00:00Z"),
+      artifacts=[
+          _compat.make_artifact(
+              artifact_id=f"art-{index}",
+              artifact_type="message",
+              parts=[_compat.make_text_part(text)],
+          )
+          for index, text in enumerate(
+              ["Let me divide that.", "fc", "fr", "You cannot divide by zero."]
+          )
+      ],
+  )
+
+
 class TestToAdk:
   """Test suite for to_adk functions."""
 
@@ -179,6 +226,71 @@ def test_convert_a2a_task_to_event_success(self):
     assert len(event.content.parts) == 1
     assert event.content.parts[0] == mock_genai_part
 
+  @pytest.mark.parametrize(
+      "terminal_state",
+      [_compat.TS_COMPLETED, _compat.TS_FAILED, _compat.TS_CANCELED],
+  )
+  def test_convert_a2a_task_to_event_terminal_task_is_final_response(
+      self, terminal_state
+  ):
+    """A terminal task carrying tool activity must read as a final response."""
+    task = _make_task_with_tool_activity(terminal_state)
+
+    event = convert_a2a_task_to_event(
+        task,
+        author="test-author",
+        invocation_context=self.mock_context,
+        part_converter=_tool_activity_part_converter,
+    )
+
+    # The peer's tool activity is still carried on the event ...
+    assert len(event.get_function_calls()) == 1
+    assert len(event.get_function_responses()) == 1
+    # ... but the turn is over, so the caller can detect the close.
+    assert event.actions.skip_summarization is True
+    assert event.is_final_response() is True
+
+  @pytest.mark.parametrize(
+      "non_terminal_state",
+      [_compat.TS_SUBMITTED, _compat.TS_WORKING],
+  )
+  def test_convert_a2a_task_to_event_non_terminal_task_is_not_final(
+      self, non_terminal_state
+  ):
+    """An in-flight task must keep reporting that the turn is still open."""
+    task = _make_task_with_tool_activity(non_terminal_state)
+
+    event = convert_a2a_task_to_event(
+        task,
+        author="test-author",
+        invocation_context=self.mock_context,
+        part_converter=_tool_activity_part_converter,
+    )
+
+    assert event.actions.skip_summarization is None
+    assert event.is_final_response() is False
+
+  @pytest.mark.parametrize(
+      "pause_state",
+      [_compat.TS_INPUT_REQUIRED, _compat.TS_AUTH_REQUIRED],
+  )
+  def test_convert_a2a_task_to_event_pause_state_left_untouched(
+      self, pause_state
+  ):
+    """input/auth-required already resolve via the mock call; leave them be."""
+    task = _make_task_with_tool_activity(pause_state)
+
+    event = convert_a2a_task_to_event(
+        task,
+        author="test-author",
+        invocation_context=self.mock_context,
+        part_converter=_tool_activity_part_converter,
+    )
+
+    assert event.actions.skip_summarization is None
+    assert event.long_running_tool_ids
+    assert event.is_final_response() is True
+
   def test_convert_a2a_task_to_event_returns_action_only_event(self):
     """Test A2A task conversion returns action-only events."""
     task = Task(