fix: deduplicate STATE_DELTA logging and add log_state_changes toggle

- Remove inline STATE_DELTA logging from after_tool_callback to avoid
  duplicate rows (on_state_change_callback is the canonical path)
- Add event ID dedup guard in Runner._exec_with_plugin to prevent
  the same event from triggering on_state_change_callback twice
- Add log_state_changes config flag (default False) to
  BigQueryLoggerConfig for explicit opt-in to STATE_DELTA logging
- Update tests to verify dedup, toggle on, and toggle off behavior

Author: mportdata

src/google/adk/plugins/bigquery_agent_analytics_plugin.py

@@ -410,6 +410,8 @@ class BigQueryLoggerConfig:
   # Format: "location.connection_id" (e.g. "us.my-connection")
   connection_id: Optional[str] = None
 
+  # Toggle for state change (STATE_DELTA) logging via on_state_change_callback
+  log_state_changes: bool = False
   # Toggle for session metadata (e.g. gchat thread-id)
   log_session_metadata: bool = True
   # Static custom tags (e.g. {"agent_role": "sales"})
@@ -2183,6 +2185,8 @@ async def on_state_change_callback(
         state_delta: The change in state to log.
         **kwargs: Additional arguments.
     """
+    if not self.config.log_state_changes:
+      return
     await self._log_event(
         "STATE_DELTA",
         callback_context,
@@ -2510,13 +2514,6 @@ async def after_tool_callback(
         parent_span_id_override=parent_span_id,
     )
 
-    if tool_context.actions.state_delta:
-      await self._log_event(
-          "STATE_DELTA",
-          tool_context,
-          state_delta=tool_context.actions.state_delta,
-      )
-
   async def on_tool_error_callback(
       self,
       *,

src/google/adk/runners.py

@@ -774,6 +774,7 @@ async def _exec_with_plugin(
       # transcription event.
       buffered_events: list[Event] = []
       is_transcribing: bool = False
+      notified_state_change_event_ids: set[str] = set()
 
       async with Aclosing(execute_fn(invocation_context)) as agen:
         async for event in agen:
@@ -845,7 +846,11 @@ async def _exec_with_plugin(
           yield final_event
 
           # Step 3b: Notify plugins of state changes, if any.
-          if final_event.actions.state_delta:
+          if (
+              final_event.actions.state_delta
+              and final_event.id not in notified_state_change_event_ids
+          ):
+            notified_state_change_event_ids.add(final_event.id)
             from .agents.callback_context import CallbackContext
 
             await plugin_manager.run_on_state_change_callback(

tests/unittests/plugins/test_bigquery_agent_analytics_plugin.py

@@ -1560,9 +1560,14 @@ async def test_after_tool_callback_logs_correctly(
     assert content_dict["result"] == {"res": "success"}
 
   @pytest.mark.asyncio
-  async def test_after_tool_callback_state_delta_logging(
+  async def test_after_tool_callback_state_delta_not_logged_inline(
       self, bq_plugin_inst, mock_write_client, tool_context, dummy_arrow_schema
   ):
+    """STATE_DELTA is not logged inline by after_tool_callback.
+
+    State changes are logged exclusively via on_state_change_callback to
+    avoid duplicate STATE_DELTA rows.
+    """
     mock_tool = mock.create_autospec(
         base_tool_lib.BaseTool, instance=True, spec_set=True
     )
@@ -1581,31 +1586,12 @@ async def test_after_tool_callback_state_delta_logging(
     )
     await asyncio.sleep(0.01)
 
-    # We should have two events appended: TOOL_COMPLETED and STATE_DELTA
-    assert mock_write_client.append_rows.call_count >= 1
-
-    # Retrieve all flushed events
-    rows = await _get_captured_rows_async(mock_write_client, dummy_arrow_schema)
-    assert len(rows) == 2
-
-    # Sort by event_type to reliably access them
-    rows.sort(key=lambda x: x["event_type"])
-
-    state_delta_event = (
-        rows[0] if rows[0]["event_type"] == "STATE_DELTA" else rows[1]
-    )
-    tool_event = (
-        rows[1] if rows[1]["event_type"] == "TOOL_COMPLETED" else rows[0]
+    # Only TOOL_COMPLETED should be logged; STATE_DELTA is handled
+    # exclusively by on_state_change_callback.
+    log_entry = await _get_captured_event_dict_async(
+        mock_write_client, dummy_arrow_schema
     )
-
-    assert state_delta_event["event_type"] == "STATE_DELTA"
-    assert tool_event["event_type"] == "TOOL_COMPLETED"
-
-    # Verify STATE_DELTA payload
-    attributes = json.loads(state_delta_event["attributes"])
-    assert "state_delta" in attributes
-    assert attributes["state_delta"] == {"new_key": "new_value"}
-    assert state_delta_event["content"] is None
+    assert log_entry["event_type"] == "TOOL_COMPLETED"
 
   @pytest.mark.asyncio
   async def test_on_state_change_callback_logs_correctly(
@@ -1615,6 +1601,7 @@ async def test_on_state_change_callback_logs_correctly(
       callback_context,
       dummy_arrow_schema,
   ):
+    bq_plugin_inst.config.log_state_changes = True
     state_delta = {"key": "value", "new_key": 123}
     bigquery_agent_analytics_plugin.TraceManager.push_span(callback_context)
     await bq_plugin_inst.on_state_change_callback(
@@ -1632,6 +1619,23 @@ async def test_on_state_change_callback_logs_correctly(
     attributes = json.loads(log_entry["attributes"])
     assert attributes["state_delta"] == state_delta
 
+  @pytest.mark.asyncio
+  async def test_on_state_change_callback_disabled(
+      self,
+      bq_plugin_inst,
+      mock_write_client,
+      callback_context,
+  ):
+    """Test that no event is logged when log_state_changes is False."""
+    bq_plugin_inst.config.log_state_changes = False
+    state_delta = {"key": "value", "new_key": 123}
+    bigquery_agent_analytics_plugin.TraceManager.push_span(callback_context)
+    await bq_plugin_inst.on_state_change_callback(
+        callback_context=callback_context, state_delta=state_delta
+    )
+    await asyncio.sleep(0.01)
+    mock_write_client.append_rows.assert_not_called()
+
   @pytest.mark.asyncio
   async def test_log_event_with_session_metadata(
       self,