feat(bq-plugin): Add schema auto-upgrade, tool provenance, and HITL tracing

Implements three enhancements for the BigQuery Agent Analytics plugin:

1. Schema Auto-Upgrade: Opt-in config (`auto_schema_upgrade`) that
   automatically adds missing columns to existing BQ tables. Uses
   table labels for schema version governance.

2. Tool Provenance: Adds `tool_origin` field to TOOL_* event content,
   distinguishing LOCAL, MCP, SUB_AGENT, TRANSFER_AGENT, and UNKNOWN
   origins via isinstance() checks on the tool object.

3. HITL Tracing: Emits additional HITL-specific events
   (HITL_CONFIRMATION_REQUEST, HITL_CREDENTIAL_REQUEST,
   HITL_INPUT_REQUEST + _COMPLETED variants) alongside normal TOOL_*
   events when adk_request_* tools are invoked.

Closes #4554 (plugin items 3, 4, 5)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: caohy1988

src/google/adk/plugins/bigquery_agent_analytics_plugin.py

@@ -71,6 +71,24 @@
     "google.adk.plugins.bigquery_agent_analytics", __version__
 )
 
+# Bumped when the schema changes. Used as a table label for
+# governance and to decide whether auto-upgrade should run.
+_SCHEMA_VERSION = "3"
+_SCHEMA_VERSION_LABEL_KEY = "adk_schema_version"
+
+# Human-in-the-loop (HITL) tool names that receive additional
+# dedicated event types alongside the normal TOOL_* events.
+_HITL_TOOL_NAMES = frozenset({
+    "adk_request_credential",
+    "adk_request_confirmation",
+    "adk_request_input",
+})
+_HITL_EVENT_MAP = MappingProxyType({
+    "adk_request_credential": "HITL_CREDENTIAL_REQUEST",
+    "adk_request_confirmation": "HITL_CONFIRMATION_REQUEST",
+    "adk_request_input": "HITL_INPUT_REQUEST",
+})
+
 
 def _safe_callback(func):
   """Decorator that catches and logs exceptions in plugin callbacks.
@@ -132,6 +150,40 @@ def _format_content(
   return " | ".join(parts), truncated
 
 
+def _get_tool_origin(tool: "BaseTool") -> str:
+  """Returns the provenance category of a tool.
+
+  Uses lazy imports to avoid circular dependencies.
+
+  Args:
+      tool: The tool instance.
+
+  Returns:
+      One of LOCAL, MCP, SUB_AGENT, TRANSFER_AGENT, or UNKNOWN.
+  """
+  # Import lazily to avoid circular dependencies.
+  # pylint: disable=g-import-not-at-top
+  from ..tools.agent_tool import AgentTool  # pytype: disable=import-error
+  from ..tools.function_tool import FunctionTool  # pytype: disable=import-error
+  from ..tools.transfer_to_agent_tool import TransferToAgentTool  # pytype: disable=import-error
+
+  try:
+    from ..tools.mcp_tool.mcp_tool import McpTool  # pytype: disable=import-error
+  except ImportError:
+    McpTool = None
+
+  # Order matters: TransferToAgentTool is a subclass of FunctionTool.
+  if McpTool is not None and isinstance(tool, McpTool):
+    return "MCP"
+  if isinstance(tool, TransferToAgentTool):
+    return "TRANSFER_AGENT"
+  if isinstance(tool, AgentTool):
+    return "SUB_AGENT"
+  if isinstance(tool, FunctionTool):
+    return "LOCAL"
+  return "UNKNOWN"
+
+
 def _recursive_smart_truncate(
     obj: Any, max_len: int, seen: Optional[set[int]] = None
 ) -> tuple[Any, bool]:
@@ -435,6 +487,9 @@ class BigQueryLoggerConfig:
   log_session_metadata: bool = True
   # Static custom tags (e.g. {"agent_role": "sales"})
   custom_tags: dict[str, Any] = field(default_factory=dict)
+  # If True, automatically add new columns to existing tables
+  # when the plugin schema evolves. Only additive changes are made.
+  auto_schema_upgrade: bool = False
 
 
 # ==============================================================================
@@ -1822,16 +1877,25 @@ def _atexit_cleanup(batch_processor: "BatchProcessor") -> None:
       )
 
   def _ensure_schema_exists(self) -> None:
-    """Ensures the BigQuery table exists with the correct schema."""
+    """Ensures the BigQuery table exists with the correct schema.
+
+    When ``config.auto_schema_upgrade`` is True and the table already
+    exists, missing columns are added automatically (additive only).
+    A ``adk_schema_version`` label is written for governance.
+    """
     try:
-      self.client.get_table(self.full_table_id)
+      existing_table = self.client.get_table(self.full_table_id)
+      if self.config.auto_schema_upgrade:
+        self._maybe_upgrade_schema(existing_table)
     except cloud_exceptions.NotFound:
       logger.info("Table %s not found, creating table.", self.full_table_id)
       tbl = bigquery.Table(self.full_table_id, schema=self._schema)
       tbl.time_partitioning = bigquery.TimePartitioning(
-          type_=bigquery.TimePartitioningType.DAY, field="timestamp"
+          type_=bigquery.TimePartitioningType.DAY,
+          field="timestamp",
       )
       tbl.clustering_fields = self.config.clustering_fields
+      tbl.labels = {_SCHEMA_VERSION_LABEL_KEY: _SCHEMA_VERSION}
       try:
         self.client.create_table(tbl)
       except cloud_exceptions.Conflict:
@@ -1851,6 +1915,50 @@ def _ensure_schema_exists(self) -> None:
           exc_info=True,
       )
 
+  def _maybe_upgrade_schema(self, existing_table: bigquery.Table) -> None:
+    """Adds missing columns to an existing table (additive only).
+
+    Args:
+        existing_table: The current BigQuery table object.
+    """
+    stored_version = (existing_table.labels or {}).get(
+        _SCHEMA_VERSION_LABEL_KEY
+    )
+    if stored_version == _SCHEMA_VERSION:
+      return
+
+    existing_names = {f.name for f in existing_table.schema}
+    new_fields = [f for f in self._schema if f.name not in existing_names]
+
+    updated = False
+    if new_fields:
+      merged = list(existing_table.schema) + new_fields
+      existing_table.schema = merged
+      updated = True
+      logger.info(
+          "Auto-upgrading table %s: adding columns %s",
+          self.full_table_id,
+          [f.name for f in new_fields],
+      )
+
+    # Always stamp the version label so we skip on next run.
+    labels = dict(existing_table.labels or {})
+    labels[_SCHEMA_VERSION_LABEL_KEY] = _SCHEMA_VERSION
+    existing_table.labels = labels
+    updated = True
+
+    if updated:
+      try:
+        update_fields = ["schema", "labels"]
+        self.client.update_table(existing_table, update_fields)
+      except Exception as e:
+        logger.error(
+            "Schema auto-upgrade failed for %s: %s",
+            self.full_table_id,
+            e,
+            exc_info=True,
+        )
+
   async def shutdown(self, timeout: float | None = None) -> None:
     """Shuts down the plugin and releases resources.
 
@@ -2460,14 +2568,28 @@ async def before_tool_callback(
     args_truncated, is_truncated = _recursive_smart_truncate(
         tool_args, self.config.max_content_length
     )
-    content_dict = {"tool": tool.name, "args": args_truncated}
+    tool_origin = _get_tool_origin(tool)
+    content_dict = {
+        "tool": tool.name,
+        "args": args_truncated,
+        "tool_origin": tool_origin,
+    }
     TraceManager.push_span(tool_context, "tool")
     await self._log_event(
         "TOOL_STARTING",
         tool_context,
         raw_content=content_dict,
         is_truncated=is_truncated,
     )
+    # Emit additional HITL event alongside the normal TOOL_STARTING.
+    hitl_event = _HITL_EVENT_MAP.get(tool.name)
+    if hitl_event:
+      await self._log_event(
+          hitl_event,
+          tool_context,
+          raw_content=content_dict,
+          is_truncated=is_truncated,
+      )
 
   @_safe_callback
   async def after_tool_callback(
@@ -2489,21 +2611,37 @@ async def after_tool_callback(
     resp_truncated, is_truncated = _recursive_smart_truncate(
         result, self.config.max_content_length
     )
-    content_dict = {"tool": tool.name, "result": resp_truncated}
+    tool_origin = _get_tool_origin(tool)
+    content_dict = {
+        "tool": tool.name,
+        "result": resp_truncated,
+        "tool_origin": tool_origin,
+    }
     span_id, duration = TraceManager.pop_span()
     parent_span_id, _ = TraceManager.get_current_span_and_parent()
 
+    event_data = EventData(
+        latency_ms=duration,
+        span_id_override=span_id,
+        parent_span_id_override=parent_span_id,
+    )
     await self._log_event(
         "TOOL_COMPLETED",
         tool_context,
         raw_content=content_dict,
         is_truncated=is_truncated,
-        event_data=EventData(
-            latency_ms=duration,
-            span_id_override=span_id,
-            parent_span_id_override=parent_span_id,
-        ),
+        event_data=event_data,
     )
+    # Emit additional HITL completion event.
+    hitl_event = _HITL_EVENT_MAP.get(tool.name)
+    if hitl_event:
+      await self._log_event(
+          hitl_event + "_COMPLETED",
+          tool_context,
+          raw_content=content_dict,
+          is_truncated=is_truncated,
+          event_data=event_data,
+      )
 
   @_safe_callback
   async def on_tool_error_callback(
@@ -2525,7 +2663,12 @@ async def on_tool_error_callback(
     args_truncated, is_truncated = _recursive_smart_truncate(
         tool_args, self.config.max_content_length
     )
-    content_dict = {"tool": tool.name, "args": args_truncated}
+    tool_origin = _get_tool_origin(tool)
+    content_dict = {
+        "tool": tool.name,
+        "args": args_truncated,
+        "tool_origin": tool_origin,
+    }
     _, duration = TraceManager.pop_span()
     await self._log_event(
         "TOOL_ERROR",