fix: add log_state_changes toggle and deduplicate STATE_DELTA logging

- Add log_state_changes config flag (default False) to
  BigQueryLoggerConfig for explicit opt-in to STATE_DELTA logging
  via the existing after_tool_callback inline path
- Add event ID dedup guard in Runner._exec_with_plugin to prevent
  the same event from triggering on_state_change_callback twice
- Add tests for toggle enabled and disabled 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"})
@@ -2510,7 +2512,7 @@ async def after_tool_callback(
         parent_span_id_override=parent_span_id,
     )
 
-    if tool_context.actions.state_delta:
+    if tool_context.actions.state_delta and self.config.log_state_changes:
       await self._log_event(
           "STATE_DELTA",
           tool_context,

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

@@ -1563,6 +1563,8 @@ async def test_after_tool_callback_logs_correctly(
   async def test_after_tool_callback_state_delta_logging(
       self, bq_plugin_inst, mock_write_client, tool_context, dummy_arrow_schema
   ):
+    """STATE_DELTA is logged inline when log_state_changes is True."""
+    bq_plugin_inst.config.log_state_changes = True
     mock_tool = mock.create_autospec(
         base_tool_lib.BaseTool, instance=True, spec_set=True
     )
@@ -1608,29 +1610,32 @@ async def test_after_tool_callback_state_delta_logging(
     assert state_delta_event["content"] is None
 
   @pytest.mark.asyncio
-  async def test_on_state_change_callback_logs_correctly(
-      self,
-      bq_plugin_inst,
-      mock_write_client,
-      callback_context,
-      dummy_arrow_schema,
+  async def test_after_tool_callback_state_delta_disabled(
+      self, bq_plugin_inst, mock_write_client, tool_context, dummy_arrow_schema
   ):
-    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
+    """STATE_DELTA is not logged when log_state_changes is False (default)."""
+    mock_tool = mock.create_autospec(
+        base_tool_lib.BaseTool, instance=True, spec_set=True
+    )
+    type(mock_tool).name = mock.PropertyMock(return_value="StateTool")
+    type(mock_tool).description = mock.PropertyMock(return_value="Sets state")
+
+    tool_context.actions.state_delta["new_key"] = "new_value"
+
+    bigquery_agent_analytics_plugin.TraceManager.push_span(tool_context)
+    await bq_plugin_inst.after_tool_callback(
+        tool=mock_tool,
+        tool_args={"arg1": "val1"},
+        tool_context=tool_context,
+        result={"res": "success"},
     )
     await asyncio.sleep(0.01)
+
+    # Only TOOL_COMPLETED should be logged, not STATE_DELTA
     log_entry = await _get_captured_event_dict_async(
         mock_write_client, dummy_arrow_schema
     )
-    _assert_common_fields(log_entry, "STATE_DELTA")
-    # content should be None (as raw_content was not passed)
-    assert log_entry["content"] is None
-
-    # state_delta should be in attributes
-    attributes = json.loads(log_entry["attributes"])
-    assert attributes["state_delta"] == state_delta
+    assert log_entry["event_type"] == "TOOL_COMPLETED"
 
   @pytest.mark.asyncio
   async def test_log_event_with_session_metadata(