BigQuery Agent Analytics: close fail-open privacy, GCS concurrency, and startup-loss gaps

[caohy1988]

Summary

A fresh review of the Python BigQuery Agent Analytics (BQAA) plugin at google/adk-python@9d306f5d322eccbc5022ac72407fa8a994f64cf6 found four high-priority correctness/privacy gaps and several operational follow-ups.

The most important problems are:

1. a configured privacy formatter fails open and logs the original content when it raises;
2. sensitive-key filtering is not applied to the final assembled attributes tree;
3. concurrent GCS offloads share mutable trace/span path state and can overwrite another event's object;
4. table-readiness failures are swallowed, after which the plugin can mark itself started and lose every row.

This is related to #5112, but it is intentionally narrower: #5112 covers the framework-wide design for non-persistent secret state. The BQAA output boundary should be made safe independently and should not wait for that design.

P1: privacy, data integrity, and silent-loss fixes

1. Make content_formatter failure fail closed

When a configured formatter raises, _log_event logs a warning but leaves raw_content unchanged. The original content is then parsed and written to BigQuery:

• Implementation:

  adk-python/src/google/adk/plugins/bigquery_agent_analytics_plugin.py

  Lines 3570 to 3575 in 9d306f5

  	if self.config.content_formatter:
  	try:
  	raw_content = self.config.content_formatter(raw_content, event_type)
  	except Exception as e:
  	logger.warning("Content formatter failed: %s", e)

• The current test explicitly expects the raw "Secret message" to be written after formatter failure:

  adk-python/tests/unittests/plugins/test_bigquery_agent_analytics_plugin.py

  Lines 727 to 763 in 9d306f5

  	async def test_content_formatter_error(
  	self,
  	mock_write_client,
  	invocation_context,
  	mock_auth_default,
  	mock_bq_client,
  	mock_to_arrow_schema,
  	dummy_arrow_schema,
  	mock_asyncio_to_thread,
  	):
  	"""Test content formatter error handling."""
  	_ = mock_auth_default
  	_ = mock_bq_client
  	def error_formatter(content, event_type):
  	raise ValueError("Formatter failed")
  	config = bigquery_agent_analytics_plugin.BigQueryLoggerConfig(
  	content_formatter=error_formatter
  	)
  	async with managed_plugin(
  	PROJECT_ID, DATASET_ID, table_id=TABLE_ID, config=config
  	) as plugin:
  	await plugin._ensure_started()
  	mock_write_client.append_rows.reset_mock()
  	user_message = types.Content(parts=[types.Part(text="Secret message")])
  	bigquery_agent_analytics_plugin.TraceManager.push_span(invocation_context)
  	await plugin.on_user_message_callback(
  	invocation_context=invocation_context, user_message=user_message
  	)
  	await asyncio.sleep(0.01)
  	mock_write_client.append_rows.assert_called_once()
  	log_entry = await _get_captured_event_dict_async(
  	mock_write_client, dummy_arrow_schema
  	)
  	# If formatter fails, it logs a warning and continues with original content.
  	assert log_entry["content"] == '{"text_summary": "Secret message"}'

For a formatter used as a redaction/privacy boundary, failure must never fall back to the unformatted payload.

Proposed behavior:

• Preserve the event's non-content metadata, but replace content with a fixed sentinel or omit it.
• Increment an observable formatter_failed counter and log the exception without the original payload.
• Add tests for user messages, model content, tool arguments/results, and custom formatter return values that are not JSON-serializable.

2. Sanitize the complete attributes tree immediately before serialization

_recursive_smart_truncate redacts known sensitive dictionary keys, but only the producers that call it receive that protection. Several producers add values directly:

• state_delta is copied directly into extra_attributes:

  adk-python/src/google/adk/plugins/bigquery_agent_analytics_plugin.py

  Lines 3767 to 3775 in 9d306f5

  	# --- State delta logging ---
  	if event.actions.state_delta:
  	await self._log_event(
  	"STATE_DELTA",
  	callback_ctx,
  	event_data=EventData(
  	source_event=event,
  	extra_attributes={"state_delta": dict(event.actions.state_delta)},
  	),

• extra_attributes and static custom_tags are copied directly into the assembled map:

  adk-python/src/google/adk/plugins/bigquery_agent_analytics_plugin.py

  Lines 3412 to 3469 in 9d306f5

  	attrs: dict[str, Any] = dict(event_data.extra_attributes)
  	adk_envelope = self._build_adk_envelope(
  	callback_context, event_data.source_event
  	)
  	# Merge producer-supplied adk_extras (long-running pair keys etc.)
  	# INTO the adk envelope so consumer SQL on
  	# ``$.adk.pause_kind`` / ``$.adk.function_call_id`` resolves.
  	# adk_envelope wins on key conflict — producer-derived envelope
  	# is the source of truth for identity fields like source_event_id.
  	for k, v in event_data.adk_extras.items():
  	adk_envelope.setdefault(k, v)
  	attrs["adk"] = adk_envelope
  	attrs["root_agent_name"] = TraceManager.get_root_agent_name()
  	if event_data.model:
  	attrs["model"] = event_data.model
  	if event_data.model_version:
  	attrs["model_version"] = event_data.model_version
  	if event_data.usage_metadata:
  	usage_dict, _ = _recursive_smart_truncate(
  	event_data.usage_metadata, self.config.max_content_length
  	)
  	if isinstance(usage_dict, dict):
  	attrs["usage_metadata"] = usage_dict
  	else:
  	attrs["usage_metadata"] = event_data.usage_metadata
  	if event_data.cache_metadata:
  	cache_meta_dict, _ = _recursive_smart_truncate(
  	event_data.cache_metadata, self.config.max_content_length
  	)
  	if isinstance(cache_meta_dict, dict):
  	attrs["cache_metadata"] = cache_meta_dict
  	else:
  	attrs["cache_metadata"] = event_data.cache_metadata
  	if self.config.log_session_metadata:
  	try:
  	session = callback_context._invocation_context.session
  	session_meta = {
  	"session_id": session.id,
  	"app_name": session.app_name,
  	"user_id": session.user_id,
  	}
  	# Include session state if non-empty (contains user-set metadata
  	# like gchat thread-id, customer_id, etc.)
  	if session.state:
  	truncated_state, _ = _recursive_smart_truncate(
  	dict(session.state),
  	self.config.max_content_length,
  	)
  	session_meta["state"] = truncated_state
  	attrs["session_metadata"] = session_meta
  	except Exception:
  	pass
  	if self.config.custom_tags:
  	attrs["custom_tags"] = self.config.custom_tags

• The assembled map is serialized without a final sanitizer pass:

  adk-python/src/google/adk/plugins/bigquery_agent_analytics_plugin.py

  Lines 3601 to 3616 in 9d306f5

  	latency_json = self._extract_latency(event_data)
  	attributes = self._enrich_attributes(event_data, callback_context)
  	# Capture allowlisted custom_metadata into attributes.custom_metadata.
  	# Runs for every row emitted from a source Event (incl. AGENT_RESPONSE,
  	# which does not otherwise read custom_metadata), through the same safety
  	# pipeline. Truncation here also flips is_truncated.
  	if self._custom_metadata_exact or self._custom_metadata_prefixes:
  	meta_truncated = self._capture_custom_metadata(event_data, attributes)
  	is_truncated = is_truncated or meta_truncated
  	# Serialize attributes to JSON string
  	try:
  	attributes_json = json.dumps(attributes)
  	except (TypeError, ValueError):
  	attributes_json = json.dumps(attributes, default=str)

• Sensitive-key matching only runs inside _recursive_smart_truncate:

  adk-python/src/google/adk/plugins/bigquery_agent_analytics_plugin.py

  Lines 422 to 478 in 9d306f5

  	def _recursive_smart_truncate(
  	obj: Any, max_len: int, seen: Optional[set[int]] = None
  	) -> tuple[Any, bool]:
  	"""Recursively truncates string values within a dict or list.
  	Redacts sensitive keys corresponding to OAuth tokens and secrets
  	prior to serialization into BigQuery JSON strings.
  	Args:
  	obj: The object to truncate.
  	max_len: Maximum length for string values.
  	seen: Set of object IDs visited in the current recursion stack.
  	Returns:
  	A tuple of (truncated_object, is_truncated).
  	"""
  	if seen is None:
  	seen = set()
  	obj_id = id(obj)
  	if obj_id in seen:
  	return "[CIRCULAR_REFERENCE]", False
  	# Track compound objects to detect cycles
  	is_compound = (
  	isinstance(obj, (dict, list, tuple))
  	or (dataclasses.is_dataclass(obj) and not isinstance(obj, type))
  	or hasattr(obj, "model_dump")
  	or hasattr(obj, "dict")
  	or hasattr(obj, "to_dict")
  	)
  	if is_compound:
  	seen.add(obj_id)
  	try:
  	if isinstance(obj, str):
  	if max_len != -1 and len(obj) > max_len:
  	return obj[:max_len] + "...[TRUNCATED]", True
  	return obj, False
  	elif isinstance(obj, dict):
  	truncated_any = False
  	# Use dict comprehension for potentially slightly better performance,
  	# but explicit loop is fine for clarity given recursive nature.
  	new_dict = {}
  	for k, v in obj.items():
  	if isinstance(k, str):
  	k_lower = k.lower()
  	if k_lower in _SENSITIVE_KEYS or k_lower.startswith("temp:"):
  	new_dict[k] = "[REDACTED]"
  	continue
  	val, trunc = _recursive_smart_truncate(v, max_len, seen)
  	if trunc:
  	truncated_any = True
  	new_dict[k] = val
  	return new_dict, truncated_any

As a result, values such as state_delta.access_token, custom_tags.password, and nested label/extra-attribute secrets can reach BigQuery unchanged. The same helper also does not inspect JSON-encoded string blobs such as credential caches, which is already discussed in #5112.

Proposed behavior:

• Run one sanitizer over the complete attributes tree immediately before json.dumps.
• Propagate truncation from that final pass into is_truncated.
• Redact temp: and any accepted future secret: scope by contract.
• Cover JSON-encoded credential blobs and configurable/customer-specific sensitive keys without logging their values.
• Keep producer-local sanitization where it prevents expensive oversized work, but treat the final pass as the mandatory safety boundary.

3. Remove mutable per-call IDs from the shared async content parser

The plugin reuses one HybridContentParser, mutating parser.trace_id and parser.span_id immediately before each awaited parse:

• Shared mutable assignment:

  adk-python/src/google/adk/plugins/bigquery_agent_analytics_plugin.py

  Lines 3593 to 3598 in 9d306f5

  	# Update parser's trace/span IDs for GCS pathing (reuse instance)
  	self.parser.trace_id = trace_id or "no_trace"
  	self.parser.span_id = span_id or "no_span"
  	content_json, content_parts, parser_truncated = await self.parser.parse(
  	raw_content
  	)

• Mutable parser fields:

  adk-python/src/google/adk/plugins/bigquery_agent_analytics_plugin.py

  Lines 1623 to 1635 in 9d306f5

  	def __init__(
  	self,
  	offloader: Optional[GCSOffloader],
  	trace_id: str,
  	span_id: str,
  	max_length: int = 20000,
  	connection_id: Optional[str] = None,
  	):
  	self.offloader = offloader
  	self.trace_id = trace_id
  	self.span_id = span_id
  	self.max_length = max_length
  	self.connection_id = connection_id

• GCS paths read those fields between awaited uploads:

  adk-python/src/google/adk/plugins/bigquery_agent_analytics_plugin.py

  Lines 1672 to 1680 in 9d306f5

  	# CASE B: It is Binary/Inline Data (Image/Blob)
  	elif hasattr(part, "inline_data") and part.inline_data:
  	if self.offloader:
  	ext = mimetypes.guess_extension(part.inline_data.mime_type) or ".bin"
  	path = f"{datetime.now().date()}/{self.trace_id}/{self.span_id}_p{idx}{ext}"
  	try:
  	uri = await self.offloader.upload_content(
  	part.inline_data.data, part.inline_data.mime_type, path
  	)

  and

  adk-python/src/google/adk/plugins/bigquery_agent_analytics_plugin.py

  Lines 1700 to 1719 in 9d306f5

  	# CASE C: Text
  	elif hasattr(part, "text") and part.text:
  	char_len = len(part.text)
  	byte_len = len(part.text.encode("utf-8"))
  	# Decide whether to offload using each limit in its own
  	# unit. inline_text_limit is a byte-based storage guard;
  	# max_length is a character-based truncation limit.
  	exceeds_inline_byte_limit = byte_len > self.inline_text_limit
  	exceeds_char_limit = (
  	self.max_length != -1 and char_len > self.max_length
  	)
  	if self.offloader and (exceeds_inline_byte_limit or exceeds_char_limit):
  	# Text is too big, treat as file
  	path = f"{datetime.now().date()}/{self.trace_id}/{self.span_id}_p{idx}.txt"
  	try:
  	uri = await self.offloader.upload_content(
  	part.text, "text/plain", path
  	)

With two concurrent events containing multiple offloaded parts, event A can resume after event B changes the parser IDs. A's next part then uses B's object name. Both events can write the same trace-b/span-b_p1 object, so the later upload overwrites the earlier one and a BigQuery row can point at another event's content.

Proposed behavior:

• Pass trace_id and span_id as immutable per-call arguments, or create an immutable parse context.
• Construct every path from call-local values; do not store request identity on the shared parser.
• Add a deterministic overlap test with two two-part events and a controlled await between uploads. Assert all object paths are unique and each returned URI remains owned by its source event.

4. Treat table readiness as a startup requirement and retry failures

_ensure_schema_exists catches general get/create failures and returns normally:

• Swallowed table-check/create failures:

  adk-python/src/google/adk/plugins/bigquery_agent_analytics_plugin.py

  Lines 2765 to 2802 in 9d306f5

  	try:
  	existing_table = self.client.get_table(self.full_table_id)
  	if self.config.auto_schema_upgrade:
  	self._maybe_upgrade_schema(existing_table)
  	if self.config.create_views:
  	self._create_analytics_views()
  	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",
  	)
  	tbl.clustering_fields = self.config.clustering_fields
  	tbl.labels = {_SCHEMA_VERSION_LABEL_KEY: _SCHEMA_VERSION}
  	table_ready = False
  	try:
  	self.client.create_table(tbl)
  	table_ready = True
  	except cloud_exceptions.Conflict:
  	# Another process created it concurrently — still usable.
  	table_ready = True
  	except Exception as e:
  	logger.error(
  	"Could not create table %s: %s",
  	self.full_table_id,
  	e,
  	exc_info=True,
  	)
  	if table_ready and self.config.create_views:
  	self._create_analytics_views()
  	except Exception as e:
  	logger.error(
  	"Error checking for table %s: %s",
  	self.full_table_id,
  	e,
  	exc_info=True,
  	)

• _lazy_setup does not inspect a readiness result and continues creating the writer:

  adk-python/src/google/adk/plugins/bigquery_agent_analytics_plugin.py

  Lines 2678 to 2723 in 9d306f5

  	self.full_table_id = f"{self.project_id}.{self.dataset_id}.{self.table_id}"
  	if not self._schema:
  	# Project out denied payload columns schema-first, so the table
  	# schema, Arrow schema, row dict, and views all stay consistent.
  	self._schema = _project_schema(_get_events_schema(), self._denied_columns)
  	await loop.run_in_executor(self._executor, self._ensure_schema_exists)
  	if not self.parser:
  	self.arrow_schema = to_arrow_schema(self._schema)
  	if not self.arrow_schema:
  	raise RuntimeError("Failed to convert BigQuery schema to Arrow schema.")
  	self.offloader = None
  	if self.config.gcs_bucket_name:
  	if "content_parts" in self._denied_columns:
  	# GCS offload stores its object reference in the
  	# ``content_parts`` column. With ``content_parts`` projected out,
  	# an upload would be orphaned -- payload leaks to GCS and incurs
  	# cost with no retained reference. Disable offload and keep
  	# content inline (truncated) instead.
  	logger.warning(
  	"GCS offload disabled: payload_column_denylist drops"
  	" 'content_parts', which holds the offloaded object reference;"
  	" large/binary content is kept inline (truncated) instead of"
  	" being uploaded to %s.",
  	self.config.gcs_bucket_name,
  	)
  	else:
  	self.offloader = GCSOffloader(
  	self.project_id,
  	self.config.gcs_bucket_name,
  	self._executor,
  	storage_client=storage.Client(
  	project=self.project_id, credentials=self._credentials
  	),
  	)
  	self.parser = HybridContentParser(
  	self.offloader,
  	"",
  	"",
  	max_length=self.config.max_content_length,
  	connection_id=self.config.connection_id,
  	)
  	await self._get_loop_state()

• _ensure_started then marks the plugin started:

  adk-python/src/google/adk/plugins/bigquery_agent_analytics_plugin.py

  Lines 3161 to 3177 in 9d306f5

  	if not self._started:
  	# Kept original lock name as it was not explicitly changed.
  	if self._setup_lock is None:
  	self._setup_lock = asyncio.Lock()
  	async with self._setup_lock:
  	if not self._started:
  	try:
  	await self._lazy_setup(**kwargs)
  	self._started = True
  	self._startup_error = None
  	# Record the current PID so fork detection works for
  	# the rest of this instance's lifetime.
  	if self._init_pid == 0:
  	self._init_pid = os.getpid()
  	except Exception as e:
  	self._startup_error = e
  	logger.error("Failed to initialize BigQuery Plugin: %s", e)

A transient control-plane error or failed table creation can therefore leave the plugin permanently "started" against a missing/unready table. Subsequent writes fail and analytics are lost until restart/shutdown.

Proposed behavior:

• Return verified readiness or raise from schema/table setup; do not set _started=True unless the table is usable.
• Preserve Conflict handling for concurrent creation, then verify/read the created table as needed.
• Coalesce concurrent setup, use bounded retry/backoff, and retry on a later event after failure.
• Count rows skipped while setup is unavailable with a specific drop reason.

P2: operational hardening

• Honor enabled=False on every entry point. before_run_callback calls _ensure_started() before _log_event can apply the enabled guard, and __aenter__ also starts unconditionally:

  adk-python/src/google/adk/plugins/bigquery_agent_analytics_plugin.py

  Lines 3143 to 3148 in 9d306f5

  	async def __aenter__(self) -> BigQueryAgentAnalyticsPlugin:
  	await self._ensure_started()
  	return self
  	async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
  	await self.shutdown()

  and

  adk-python/src/google/adk/plugins/bigquery_agent_analytics_plugin.py

  Lines 4038 to 4053 in 9d306f5

  	@_safe_callback
  	async def before_run_callback(
  	self, *, invocation_context: "InvocationContext"
  	) -> None:
  	"""Callback before the agent run starts.
  	Args:
  	invocation_context: The context of the current invocation.
  	"""
  	await self._ensure_started()
  	callback_ctx = CallbackContext(invocation_context)
  	TraceManager.ensure_invocation_span(callback_ctx)
  	await self._log_event(
  	"INVOCATION_STARTING",
  	callback_ctx,
  	)

  . Disabled mode should perform no ADC lookup, BigQuery/GCS client creation, table RPC, or background-task creation.
• Bound or decouple first-use setup from agent callbacks. before_run_callback awaits the entire setup path, so slow BigQuery control-plane calls delay the agent run. Provide explicit startup/preflight or a bounded asynchronous readiness path with observable loss behavior.
• Validate runtime settings at construction. The dataclasses accept invalid batch, duration, queue, and retry values, while the constructor validates only view/projection-related fields:

  adk-python/src/google/adk/plugins/bigquery_agent_analytics_plugin.py

  Lines 668 to 798 in 9d306f5

  	@dataclass
  	class RetryConfig:
  	"""Configuration for retrying failed BigQuery write operations.
  	Attributes:
  	max_retries: Maximum number of retry attempts.
  	initial_delay: Initial delay between retries in seconds.
  	multiplier: Multiplier for exponential backoff.
  	max_delay: Maximum delay between retries in seconds.
  	"""
  	max_retries: int = 3
  	initial_delay: float = 1.0
  	multiplier: float = 2.0
  	max_delay: float = 10.0
  	@dataclass
  	class BigQueryLoggerConfig:
  	"""Configuration for the BigQueryAgentAnalyticsPlugin.
  	Attributes:
  	enabled: Whether logging is enabled.
  	event_allowlist: list of event types to log. If None, all are allowed.
  	event_denylist: list of event types to ignore.
  	max_content_length: Max length for text content before truncation.
  	table_id: BigQuery table ID.
  	clustering_fields: Fields to cluster the table by.
  	log_multi_modal_content: Whether to log detailed content parts.
  	retry_config: Retry configuration for writes.
  	batch_size: Number of rows per batch.
  	batch_flush_interval: Max time to wait before flushing a batch.
  	shutdown_timeout: Max time to wait for shutdown.
  	queue_max_size: Max size of the in-memory queue.
  	content_formatter: Optional custom formatter for content.
  	gcs_bucket_name: GCS bucket for offloading large content.
  	connection_id: BigQuery connection ID for ObjectRef columns.
  	log_session_metadata: Whether to log session metadata.
  	custom_tags: Static custom tags to attach to every event.
  	auto_schema_upgrade: Whether to auto-add new columns on schema evolution.
  	create_views: Whether to auto-create per-event-type views.
  	view_prefix: Prefix for auto-created view names. Default ``"v"`` produces
  	views like ``v_llm_request``. Set a distinct prefix per table when
  	multiple plugin instances share one dataset to avoid view-name
  	collisions.
  	enable_otel_correlation: When ``True``, capture the ambient OpenTelemetry
  	span context at row-emission time into ``attributes.otel.{span_id,
  	trace_id}`` (a best-effort Cloud Trace join key, not a foreign key).
  	``False`` (the default) emits no ``attributes.otel``. Has no effect when
  	``attributes`` is projected out via ``payload_column_denylist``.
  	custom_metadata_allowlist: Keys to capture from ``event.custom_metadata``
  	into ``attributes.custom_metadata.*``. Entries are exact keys, or
  	explicit prefix patterns ending in ``*`` (e.g. ``"a2a:*"``). ``None`` /
  	empty preserves today's behavior (only the built-in ``a2a:*`` path
  	runs). Captured values pass the same safety pipeline (truncation,
  	sensitive-key redaction, circular-reference handling) as all other
  	logged content.
  	payload_column_denylist: Payload columns to project OUT of the table at
  	write time. Only the projectable payload columns ``content`` /
  	``content_parts`` / ``attributes`` / ``latency_ms`` may be listed;
  	identity / correlation columns are protected and raise ``ValueError`` if
  	listed. Applied schema-first (table schema, Arrow schema, row dict, and
  	views all stay consistent); views that reference a denied column drop
  	the dependent derived columns. NOTE: denying ``attributes`` also
  	disables ``attributes.otel`` and ``attributes.custom_metadata``;
  	combining it with a non-empty ``custom_metadata_allowlist`` is
  	rejected at construction.
  	"""
  	enabled: bool = True
  	# V1 Configuration Parity
  	event_allowlist: list[str] | None = None
  	event_denylist: list[str] | None = None
  	max_content_length: int = 500 * 1024 # Defaults to 500KB per text block
  	table_id: str = "agent_events"
  	# V2 Configuration
  	clustering_fields: list[str] = field(
  	default_factory=lambda: ["event_type", "agent", "user_id"]
  	)
  	log_multi_modal_content: bool = True
  	retry_config: RetryConfig = field(default_factory=RetryConfig)
  	batch_size: int = 1
  	batch_flush_interval: float = 1.0
  	shutdown_timeout: float = 10.0
  	queue_max_size: int = 10000
  	content_formatter: Optional[Callable[[Any, str], Any]] = None
  	# If provided, large content (images, audio, video, large text) will be offloaded to this GCS bucket.
  	gcs_bucket_name: Optional[str] = None
  	# If provided, this connection ID will be used as the authorizer for ObjectRef columns.
  	# Format: "location.connection_id" (e.g. "us.my-connection")
  	connection_id: Optional[str] = None
  	# Toggle for session metadata (e.g. gchat thread-id)
  	log_session_metadata: bool = True
  	# Static custom tags (e.g. {"agent_role": "sales"})
  	custom_tags: dict[str, Any] = field(default_factory=dict)
  	# Automatically add new columns to existing tables when the plugin
  	# schema evolves. Only additive changes are made (columns are never
  	# dropped or altered). Safe to leave enabled; a version label on the
  	# table ensures the diff runs at most once per schema version.
  	auto_schema_upgrade: bool = True
  	# Automatically create per-event-type BigQuery views that unnest
  	# JSON columns into typed, queryable columns.
  	create_views: bool = True
  	# Prefix for auto-created per-event-type view names.
  	# Default "v" produces views like ``v_llm_request``. Set a distinct
  	# prefix per table when multiple plugin instances share one dataset
  	# to avoid view-name collisions (e.g. ``"v_staging"`` →
  	# ``v_staging_llm_request``).
  	view_prefix: str = "v"
  	# --- span-level Cloud Trace correlation ---
  	# When True, capture the ambient OpenTelemetry span context into
  	# ``attributes.otel.{span_id,trace_id}`` at row-emission time. Off by
  	# default; no plugin-owned span is created.
  	enable_otel_correlation: bool = False
  	# --- generic custom_metadata capture (allowlist) ---
  	# Exact keys and/or explicit ``*``-suffixed prefix patterns to capture
  	# from ``event.custom_metadata`` into ``attributes.custom_metadata.*``.
  	# None/empty preserves today's behavior (only the built-in ``a2a:*`` path).
  	custom_metadata_allowlist: list[str] | None = None
  	# --- physical column projection (denylist-first) ---
  	# Payload columns to omit from the table at write time. Only the
  	# projectable payload columns are accepted; identity/correlation columns
  	# are protected (see ``_PROJECTABLE_PAYLOAD_COLUMNS``).
  	payload_column_denylist: list[str] | None = None

  and

  adk-python/src/google/adk/plugins/bigquery_agent_analytics_plugin.py

  Lines 2424 to 2456 in 9d306f5

  	self.config = config or BigQueryLoggerConfig()
  	# Override config with kwargs if provided
  	for key, value in kwargs.items():
  	if hasattr(self.config, key):
  	setattr(self.config, key, value)
  	else:
  	logger.warning(f"Unknown configuration parameter: {key}")
  	if not self.config.view_prefix:
  	raise ValueError("view_prefix must be a non-empty string.")
  	# Pre-parse the custom_metadata allowlist into exact keys + prefixes.
  	self._custom_metadata_exact, self._custom_metadata_prefixes = (
  	_parse_custom_metadata_allowlist(self.config.custom_metadata_allowlist)
  	)
  	# Validate (fail-closed on protected/unknown columns) the projection.
  	self._denied_columns = _validate_payload_column_denylist(
  	self.config.payload_column_denylist
  	)
  	# Capturing custom_metadata into the attributes column is
  	# incompatible with projecting attributes out -- the captured payload
  	# would be silently dropped (and is_truncated could still flip). Fail
  	# fast rather than do useless work.
  	if "attributes" in self._denied_columns and (
  	self._custom_metadata_exact or self._custom_metadata_prefixes
  	):
  	raise ValueError(
  	"custom_metadata_allowlist captures into the 'attributes' column,"
  	" but 'attributes' is in payload_column_denylist -- the captured"
  	" metadata would be dropped. Remove 'attributes' from"
  	" payload_column_denylist or clear custom_metadata_allowlist."
  	)

  . Validate batch_size, flush/shutdown durations, queue bounds, max_retries, retry delays, and multiplier. In particular, max_retries < 0 skips the write loop entirely.
• Extend loss accounting beyond an existing BatchProcessor. Setup failures and rows skipped before a processor exists currently cannot appear in get_drop_stats(). Preserve counters across loop cleanup/shutdown and include setup, formatter, shutdown-race, and rejected-append reasons.

Acceptance tests

• A raising content_formatter never writes the original content.
• Final attributes sanitization covers state deltas, custom tags, labels, generic extra attributes, session state, nested structures, and JSON-encoded secret blobs.
• Concurrent two-part GCS offloads retain call-local trace/span paths and never share an object name.
• Failed table get/create leaves _started=False, records loss, and succeeds on a later retry without recreating one setup RPC per event.
• enabled=False causes zero auth/client/table/view/writer/background-task side effects through Runner callbacks and async context-manager use.
• Slow setup cannot block an agent run beyond the documented bound.
• Invalid batch, queue, duration, and retry settings fail during plugin construction.
• Drop stats remain queryable after shutdown and include pre-processor failure reasons.

Verification signal

• Focused Python BQAA suite at the pinned commit: 306 passed, 6 skipped.
• A deterministic local repro confirmed the shared-parser path collision, unredacted final attributes, disabled-mode startup call, and acceptance of invalid runtime config.
• The existing formatter-error test currently codifies fail-open behavior by asserting that the original secret message is written.

These green tests do not cover the safety/lifecycle invariants above.

Related issues

• "Secret" session state scope #5112 — broader non-persistent secret: session-state design and JSON-blob/arbitrary-key redaction gaps.
• BigQuery Agent Analytics Plugin: 6 validated bugs (status reporting, schema upgrade, shutdown, PII, truncation, dead code) #4694 — earlier six-item BQAA review; its status/schema/multi-loop/session-truncation findings have since been addressed. The findings in this issue are distinct current-main gaps.