entry : rowIndexToErrorMessage.entrySet()) {
+ logger.severe(
+ String.format("Row error at index %d: %s", entry.getKey(), entry.getValue()));
+ }
+ } else {
+ logger.severe(
+ "AppendSerializationError occurred, but no row-specific errors were provided.");
+ }
+ }
+ } catch (Exception e) {
+ if (e instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
+ logger.log(Level.SEVERE, "Failed to write batch to BigQuery", e);
+ } finally {
+ // Clear the vectors to release the memory.
+ root.clear();
+ }
+ } finally {
+ flushLock.set(false);
+ if (queue.size() >= batchSize && !flushLock.get()) {
+ executor.execute(this::flush);
+ }
+ }
+ }
+
+ private void populateVector(FieldVector vector, int index, Object value) {
+ if (value == null || (value instanceof JsonNode jsonNode && jsonNode.isNull())) {
+ vector.setNull(index);
+ return;
+ }
+ if (vector instanceof VarCharVector varCharVector) {
+ String strValue = (value instanceof JsonNode jsonNode) ? jsonNode.asText() : value.toString();
+ varCharVector.setSafe(index, strValue.getBytes(UTF_8));
+ } else if (vector instanceof BigIntVector bigIntVector) {
+ long longValue;
+ if (value instanceof JsonNode jsonNode) {
+ longValue = jsonNode.asLong();
+ } else if (value instanceof Number number) {
+ longValue = number.longValue();
+ } else {
+ longValue = Long.parseLong(value.toString());
+ }
+ bigIntVector.setSafe(index, longValue);
+ } else if (vector instanceof BitVector bitVector) {
+ boolean boolValue =
+ (value instanceof JsonNode jsonNode) ? jsonNode.asBoolean() : (Boolean) value;
+ bitVector.setSafe(index, boolValue ? 1 : 0);
+ } else if (vector instanceof TimeStampVector timeStampVector) {
+ if (value instanceof Instant instant) {
+ long micros =
+ SECONDS.toMicros(instant.getEpochSecond()) + NANOSECONDS.toMicros(instant.getNano());
+ timeStampVector.setSafe(index, micros);
+ } else if (value instanceof JsonNode jsonNode) {
+ timeStampVector.setSafe(index, jsonNode.asLong());
+ } else if (value instanceof Long longValue) {
+ timeStampVector.setSafe(index, longValue);
+ }
+ } else if (vector instanceof ListVector listVector) {
+ int start = listVector.startNewValue(index);
+ if (value instanceof ArrayNode arrayNode) {
+ for (int i = 0; i < arrayNode.size(); i++) {
+ populateVector(listVector.getDataVector(), start + i, arrayNode.get(i));
+ }
+ listVector.endValue(index, arrayNode.size());
+ } else if (value instanceof List) {
+ List list = (List) value;
+ for (int i = 0; i < list.size(); i++) {
+ populateVector(listVector.getDataVector(), start + i, list.get(i));
+ }
+ listVector.endValue(index, list.size());
+ }
+ } else if (vector instanceof StructVector structVector) {
+ structVector.setIndexDefined(index);
+ if (value instanceof ObjectNode objectNode) {
+ for (FieldVector child : structVector.getChildrenFromFields()) {
+ populateVector(child, index, objectNode.get(child.getName()));
+ }
+ } else if (value instanceof Map) {
+ Map map = (Map) value;
+ for (FieldVector child : structVector.getChildrenFromFields()) {
+ populateVector(child, index, map.get(child.getName()));
+ }
+ }
+ }
+ }
+
+ @Override
+ public void close() {
+ if (this.queue != null && !this.queue.isEmpty()) {
+ this.flush();
+ }
+ if (this.allocator != null) {
+ try {
+ this.allocator.close();
+ } catch (RuntimeException e) {
+ logger.log(Level.SEVERE, "Failed to close Buffer allocator", e);
+ }
+ }
+ if (this.root != null) {
+ try {
+ this.root.close();
+ } catch (RuntimeException e) {
+ logger.log(Level.SEVERE, "Failed to close VectorSchemaRoot", e);
+ }
+ }
+ if (this.writer != null) {
+ try {
+ this.writer.close();
+ } catch (RuntimeException e) {
+ logger.log(Level.SEVERE, "Failed to close BigQuery writer", e);
+ }
+ }
+ }
+}
diff --git a/core/src/main/java/com/google/adk/plugins/agentanalytics/BigQueryAgentAnalyticsPlugin.java b/core/src/main/java/com/google/adk/plugins/agentanalytics/BigQueryAgentAnalyticsPlugin.java
new file mode 100644
index 000000000..68b5fb5a1
--- /dev/null
+++ b/core/src/main/java/com/google/adk/plugins/agentanalytics/BigQueryAgentAnalyticsPlugin.java
@@ -0,0 +1,436 @@
+/*
+ * Copyright 2026 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.adk.plugins.agentanalytics;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+
+import com.google.adk.agents.BaseAgent;
+import com.google.adk.agents.CallbackContext;
+import com.google.adk.agents.InvocationContext;
+import com.google.adk.events.Event;
+import com.google.adk.models.LlmRequest;
+import com.google.adk.models.LlmResponse;
+import com.google.adk.plugins.BasePlugin;
+import com.google.adk.tools.BaseTool;
+import com.google.adk.tools.ToolContext;
+import com.google.api.gax.core.FixedCredentialsProvider;
+import com.google.api.gax.retrying.RetrySettings;
+import com.google.auth.oauth2.GoogleCredentials;
+import com.google.cloud.bigquery.BigQuery;
+import com.google.cloud.bigquery.BigQueryException;
+import com.google.cloud.bigquery.BigQueryOptions;
+import com.google.cloud.bigquery.Clustering;
+import com.google.cloud.bigquery.Schema;
+import com.google.cloud.bigquery.StandardTableDefinition;
+import com.google.cloud.bigquery.Table;
+import com.google.cloud.bigquery.TableId;
+import com.google.cloud.bigquery.TableInfo;
+import com.google.cloud.bigquery.storage.v1.BigQueryWriteClient;
+import com.google.cloud.bigquery.storage.v1.BigQueryWriteSettings;
+import com.google.cloud.bigquery.storage.v1.StreamWriter;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.VerifyException;
+import com.google.common.collect.ImmutableList;
+import com.google.genai.types.Content;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.SpanContext;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Maybe;
+import java.io.IOException;
+import java.time.Instant;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+import org.threeten.bp.Duration;
+
+/**
+ * BigQuery Agent Analytics Plugin for Java.
+ *
+ * Logs agent execution events directly to a BigQuery table using the Storage Write API.
+ */
+public class BigQueryAgentAnalyticsPlugin extends BasePlugin {
+ private static final Logger logger =
+ Logger.getLogger(BigQueryAgentAnalyticsPlugin.class.getName());
+ private static final ImmutableList DEFAULT_AUTH_SCOPES =
+ ImmutableList.of("https://www.googleapis.com/auth/cloud-platform");
+ private static final AtomicLong threadCounter = new AtomicLong(0);
+
+ private final BigQueryLoggerConfig config;
+ private final BigQuery bigQuery;
+ private final BigQueryWriteClient writeClient;
+ private final ScheduledExecutorService executor;
+ private final Object tableEnsuredLock = new Object();
+ @VisibleForTesting final BatchProcessor batchProcessor;
+ private volatile boolean tableEnsured = false;
+
+ public BigQueryAgentAnalyticsPlugin(BigQueryLoggerConfig config) throws IOException {
+ this(config, createBigQuery(config));
+ }
+
+ public BigQueryAgentAnalyticsPlugin(BigQueryLoggerConfig config, BigQuery bigQuery)
+ throws IOException {
+ super("bigquery_agent_analytics");
+ this.config = config;
+ this.bigQuery = bigQuery;
+ ThreadFactory threadFactory =
+ r -> new Thread(r, "bq-analytics-plugin-" + threadCounter.getAndIncrement());
+ this.executor = Executors.newScheduledThreadPool(1, threadFactory);
+ this.writeClient = createWriteClient(config);
+
+ if (config.enabled()) {
+ StreamWriter writer = createWriter(config);
+ this.batchProcessor =
+ new BatchProcessor(
+ writer,
+ config.batchSize(),
+ config.batchFlushInterval(),
+ config.queueMaxSize(),
+ executor);
+ this.batchProcessor.start();
+ } else {
+ this.batchProcessor = null;
+ }
+ }
+
+ private static BigQuery createBigQuery(BigQueryLoggerConfig config) throws IOException {
+ BigQueryOptions.Builder builder = BigQueryOptions.newBuilder();
+ if (config.credentials() != null) {
+ builder.setCredentials(config.credentials());
+ } else {
+ builder.setCredentials(
+ GoogleCredentials.getApplicationDefault().createScoped(DEFAULT_AUTH_SCOPES));
+ }
+ return builder.build().getService();
+ }
+
+ private void ensureTableExistsOnce() {
+ if (!tableEnsured) {
+ synchronized (tableEnsuredLock) {
+ if (!tableEnsured) {
+ // Table creation is expensive, so we only do it once per plugin instance.
+ tableEnsured = true;
+ ensureTableExists(bigQuery, config);
+ }
+ }
+ }
+ }
+
+ private void ensureTableExists(BigQuery bigQuery, BigQueryLoggerConfig config) {
+ TableId tableId = TableId.of(config.projectId(), config.datasetId(), config.tableName());
+ Schema schema = BigQuerySchema.getEventsSchema();
+ try {
+ Table table = bigQuery.getTable(tableId);
+ logger.info("BigQuery table: " + tableId);
+ if (table == null) {
+ logger.info("Creating BigQuery table: " + tableId);
+ StandardTableDefinition.Builder tableDefinitionBuilder =
+ StandardTableDefinition.newBuilder().setSchema(schema);
+ if (!config.clusteringFields().isEmpty()) {
+ tableDefinitionBuilder.setClustering(
+ Clustering.newBuilder().setFields(config.clusteringFields()).build());
+ }
+ TableInfo tableInfo = TableInfo.newBuilder(tableId, tableDefinitionBuilder.build()).build();
+ bigQuery.create(tableInfo);
+ } else if (config.autoSchemaUpgrade()) {
+ // TODO(b/491851868): Implement auto-schema upgrade.
+ logger.info("BigQuery table already exists and auto-schema upgrade is enabled: " + tableId);
+ logger.info("Auto-schema upgrade is not implemented yet.");
+ }
+ } catch (BigQueryException e) {
+ if (e.getMessage().contains("invalid_grant")) {
+ logger.log(
+ Level.SEVERE,
+ "Failed to authenticate with BigQuery. Please run 'gcloud auth application-default"
+ + " login' to refresh your credentials or provide valid credentials in"
+ + " BigQueryLoggerConfig.",
+ e);
+ } else {
+ logger.log(
+ Level.WARNING, "Failed to check or create/upgrade BigQuery table: " + tableId, e);
+ }
+ } catch (RuntimeException e) {
+ logger.log(Level.WARNING, "Failed to check or create/upgrade BigQuery table: " + tableId, e);
+ }
+ }
+
+ protected BigQueryWriteClient createWriteClient(BigQueryLoggerConfig config) throws IOException {
+ if (config.credentials() != null) {
+ return BigQueryWriteClient.create(
+ BigQueryWriteSettings.newBuilder()
+ .setCredentialsProvider(FixedCredentialsProvider.create(config.credentials()))
+ .build());
+ }
+ return BigQueryWriteClient.create();
+ }
+
+ protected String getStreamName(BigQueryLoggerConfig config) {
+ return String.format(
+ "projects/%s/datasets/%s/tables/%s/streams/_default",
+ config.projectId(), config.datasetId(), config.tableName());
+ }
+
+ protected StreamWriter createWriter(BigQueryLoggerConfig config) {
+ BigQueryLoggerConfig.RetryConfig retryConfig = config.retryConfig();
+ RetrySettings retrySettings =
+ RetrySettings.newBuilder()
+ .setMaxAttempts(retryConfig.maxRetries())
+ .setInitialRetryDelay(Duration.ofMillis(retryConfig.initialDelay().toMillis()))
+ .setRetryDelayMultiplier(retryConfig.multiplier())
+ .setMaxRetryDelay(Duration.ofMillis(retryConfig.maxDelay().toMillis()))
+ .build();
+
+ String streamName = getStreamName(config);
+ try {
+ return StreamWriter.newBuilder(streamName, writeClient)
+ .setRetrySettings(retrySettings)
+ .setWriterSchema(BigQuerySchema.getArrowSchema())
+ .build();
+ } catch (Exception e) {
+ throw new VerifyException("Failed to create StreamWriter for " + streamName, e);
+ }
+ }
+
+ private void logEvent(
+ String eventType,
+ InvocationContext invocationContext,
+ Optional callbackContext,
+ Object content,
+ Map extraAttributes) {
+ if (batchProcessor == null) {
+ return;
+ }
+
+ ensureTableExistsOnce();
+
+ Map row = new HashMap<>();
+ row.put("timestamp", Instant.now());
+ row.put("event_type", eventType);
+ row.put(
+ "agent",
+ callbackContext.map(CallbackContext::agentName).orElse(invocationContext.agent().name()));
+ row.put("session_id", invocationContext.session().id());
+ row.put("invocation_id", invocationContext.invocationId());
+ row.put("user_id", invocationContext.userId());
+
+ if (content instanceof Content contentParts) {
+ row.put(
+ "content_parts",
+ JsonFormatter.formatContentParts(Optional.of(contentParts), config.maxContentLength()));
+ row.put(
+ "content", JsonFormatter.smartTruncate(content, config.maxContentLength()).toString());
+ } else if (content != null) {
+ row.put(
+ "content", JsonFormatter.smartTruncate(content, config.maxContentLength()).toString());
+ }
+
+ Map attributes = new HashMap<>(config.customTags());
+ if (extraAttributes != null) {
+ attributes.putAll(extraAttributes);
+ }
+ row.put(
+ "attributes",
+ JsonFormatter.smartTruncate(attributes, config.maxContentLength()).toString());
+
+ addTraceDetails(row);
+ batchProcessor.append(row);
+ }
+
+ // TODO(b/491849911): Implement own trace management functionality.
+ private void addTraceDetails(Map row) {
+ SpanContext spanContext = Span.current().getSpanContext();
+ if (spanContext.isValid()) {
+ row.put("trace_id", spanContext.getTraceId());
+ row.put("span_id", spanContext.getSpanId());
+ }
+ }
+
+ @Override
+ public Completable close() {
+ if (batchProcessor != null) {
+ batchProcessor.close();
+ }
+ if (writeClient != null) {
+ writeClient.close();
+ }
+ try {
+ executor.shutdown();
+ if (!executor.awaitTermination(config.shutdownTimeout().toMillis(), MILLISECONDS)) {
+ executor.shutdownNow();
+ }
+ } catch (InterruptedException e) {
+ executor.shutdownNow();
+ Thread.currentThread().interrupt();
+ }
+ return Completable.complete();
+ }
+
+ @Override
+ public Maybe onUserMessageCallback(
+ InvocationContext invocationContext, Content userMessage) {
+ return Maybe.fromAction(
+ () -> logEvent("USER_MESSAGE", invocationContext, Optional.empty(), userMessage, null));
+ }
+
+ @Override
+ public Maybe beforeRunCallback(InvocationContext invocationContext) {
+ return Maybe.fromAction(
+ () -> logEvent("INVOCATION_START", invocationContext, Optional.empty(), null, null));
+ }
+
+ @Override
+ public Maybe onEventCallback(InvocationContext invocationContext, Event event) {
+ return Maybe.fromAction(
+ () -> {
+ Map attrs = new HashMap<>();
+ attrs.put("event_author", event.author());
+ logEvent(
+ "EVENT", invocationContext, Optional.empty(), event.content().orElse(null), attrs);
+ });
+ }
+
+ @Override
+ public Completable afterRunCallback(InvocationContext invocationContext) {
+ return Completable.fromAction(
+ () -> {
+ logEvent("INVOCATION_END", invocationContext, Optional.empty(), null, null);
+ batchProcessor.flush();
+ });
+ }
+
+ @Override
+ public Maybe beforeAgentCallback(BaseAgent agent, CallbackContext callbackContext) {
+ return Maybe.fromAction(
+ () ->
+ logEvent(
+ "AGENT_START",
+ callbackContext.invocationContext(),
+ Optional.of(callbackContext),
+ null,
+ null));
+ }
+
+ @Override
+ public Maybe afterAgentCallback(BaseAgent agent, CallbackContext callbackContext) {
+ return Maybe.fromAction(
+ () ->
+ logEvent(
+ "AGENT_END",
+ callbackContext.invocationContext(),
+ Optional.of(callbackContext),
+ null,
+ null));
+ }
+
+ @Override
+ public Maybe beforeModelCallback(
+ CallbackContext callbackContext, LlmRequest.Builder llmRequest) {
+ return Maybe.fromAction(
+ () -> {
+ Map attrs = new HashMap<>();
+ LlmRequest req = llmRequest.build();
+ attrs.put("model", req.model().orElse("unknown"));
+ logEvent(
+ "MODEL_REQUEST",
+ callbackContext.invocationContext(),
+ Optional.of(callbackContext),
+ req,
+ attrs);
+ });
+ }
+
+ @Override
+ public Maybe afterModelCallback(
+ CallbackContext callbackContext, LlmResponse llmResponse) {
+ return Maybe.fromAction(
+ () -> {
+ Map attrs = new HashMap<>();
+ llmResponse.usageMetadata().ifPresent(u -> attrs.put("usage_metadata", u));
+ logEvent(
+ "MODEL_RESPONSE",
+ callbackContext.invocationContext(),
+ Optional.of(callbackContext),
+ llmResponse,
+ attrs);
+ });
+ }
+
+ @Override
+ public Maybe onModelErrorCallback(
+ CallbackContext callbackContext, LlmRequest.Builder llmRequest, Throwable error) {
+ return Maybe.fromAction(
+ () -> {
+ Map attrs = new HashMap<>();
+ attrs.put("error_message", error.getMessage());
+ logEvent(
+ "MODEL_ERROR",
+ callbackContext.invocationContext(),
+ Optional.of(callbackContext),
+ null,
+ attrs);
+ });
+ }
+
+ @Override
+ public Maybe> beforeToolCallback(
+ BaseTool tool, Map toolArgs, ToolContext toolContext) {
+ return Maybe.fromAction(
+ () -> {
+ Map attrs = new HashMap<>();
+ attrs.put("tool_name", tool.name());
+ logEvent(
+ "TOOL_START",
+ toolContext.invocationContext(),
+ Optional.of(toolContext),
+ toolArgs,
+ attrs);
+ });
+ }
+
+ @Override
+ public Maybe> afterToolCallback(
+ BaseTool tool,
+ Map toolArgs,
+ ToolContext toolContext,
+ Map result) {
+ return Maybe.fromAction(
+ () -> {
+ Map attrs = new HashMap<>();
+ attrs.put("tool_name", tool.name());
+ logEvent(
+ "TOOL_END", toolContext.invocationContext(), Optional.of(toolContext), result, attrs);
+ });
+ }
+
+ @Override
+ public Maybe> onToolErrorCallback(
+ BaseTool tool, Map toolArgs, ToolContext toolContext, Throwable error) {
+ return Maybe.fromAction(
+ () -> {
+ Map attrs = new HashMap<>();
+ attrs.put("tool_name", tool.name());
+ attrs.put("error_message", error.getMessage());
+ logEvent(
+ "TOOL_ERROR", toolContext.invocationContext(), Optional.of(toolContext), null, attrs);
+ });
+ }
+}
diff --git a/core/src/main/java/com/google/adk/plugins/agentanalytics/BigQueryLoggerConfig.java b/core/src/main/java/com/google/adk/plugins/agentanalytics/BigQueryLoggerConfig.java
new file mode 100644
index 000000000..aa5bf37de
--- /dev/null
+++ b/core/src/main/java/com/google/adk/plugins/agentanalytics/BigQueryLoggerConfig.java
@@ -0,0 +1,204 @@
+/*
+ * Copyright 2026 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.adk.plugins.agentanalytics;
+
+import com.google.auth.Credentials;
+import com.google.auto.value.AutoValue;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import java.time.Duration;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.function.BiFunction;
+import javax.annotation.Nullable;
+
+/** Configuration for the BigQueryAgentAnalyticsPlugin. */
+@AutoValue
+public abstract class BigQueryLoggerConfig {
+ // Whether the plugin is enabled.
+ public abstract boolean enabled();
+
+ // List of event types to log. If None, all are allowed
+ // TODO(b/491852782): Implement allowlist/denylist for event types.
+ @Nullable
+ public abstract ImmutableList eventAllowlist();
+
+ // List of event types to ignore.
+ // TODO(b/491852782): Implement allowlist/denylist for event types.
+ @Nullable
+ public abstract ImmutableList eventDenylist();
+
+ // Max length for text content before truncation.
+ public abstract int maxContentLength();
+
+ // Project ID for the BigQuery table.
+ public abstract String projectId();
+
+ // Dataset ID for the BigQuery table.
+ public abstract String datasetId();
+
+ // Table name for the BigQuery table.
+ public abstract String tableName();
+
+ // Fields to cluster the table by.
+ public abstract ImmutableList clusteringFields();
+
+ // Whether to log multi-modal content.
+ // TODO(b/491852782): Implement logging of multi-modal content.
+ public abstract boolean logMultiModalContent();
+
+ // Retry configuration for BigQuery writes.
+ public abstract RetryConfig retryConfig();
+
+ // Number of rows to batch before flushing.
+ public abstract int batchSize();
+
+ // Duration to wait before flushing the queue.
+ public abstract Duration batchFlushInterval();
+
+ // Max time to wait for shutdown.
+ public abstract Duration shutdownTimeout();
+
+ // Max size of the batch processor queue.
+ public abstract int queueMaxSize();
+
+ // Optional custom formatter for content.
+ // TODO(b/491852782): Implement content formatter.
+ @Nullable
+ public abstract BiFunction contentFormatter();
+
+ // TODO(b/491852782): Implement connection id.
+ public abstract Optional connectionId();
+
+ // Toggle for session metadata (e.g. gchat thread-id).
+ // TODO(b/491852782): Implement logging of session metadata.
+ public abstract boolean logSessionMetadata();
+
+ // Static custom tags (e.g. {"agent_role": "sales"}).
+ // TODO(b/491852782): Implement custom tags.
+ public abstract ImmutableMap customTags();
+
+ // Automatically add new columns to existing tables when the plugin
+ // schema evolves. Only additive changes are made (columns are never
+ // dropped or altered).
+ // TODO(b/491852782): Implement auto-schema upgrade.
+ public abstract boolean autoSchemaUpgrade();
+
+ @Nullable
+ public abstract Credentials credentials();
+
+ public static Builder builder() {
+ return new AutoValue_BigQueryLoggerConfig.Builder()
+ .setEnabled(true)
+ .setMaxContentLength(500 * 1024)
+ .setDatasetId("agent_analytics")
+ .setTableName("events")
+ .setClusteringFields(ImmutableList.of("event_type", "agent", "user_id"))
+ .setLogMultiModalContent(true)
+ .setRetryConfig(RetryConfig.builder().build())
+ .setBatchSize(1)
+ .setBatchFlushInterval(Duration.ofSeconds(1))
+ .setShutdownTimeout(Duration.ofSeconds(10))
+ .setQueueMaxSize(10000)
+ .setLogSessionMetadata(true)
+ .setCustomTags(ImmutableMap.of())
+ // TODO(b/491851868): Enable auto-schema upgrade once implemented.
+ .setAutoSchemaUpgrade(false);
+ }
+
+ /** Builder for {@link BigQueryLoggerConfig}. */
+ @AutoValue.Builder
+ public abstract static class Builder {
+ public abstract Builder setEnabled(boolean enabled);
+
+ public abstract Builder setEventAllowlist(@Nullable List eventAllowlist);
+
+ public abstract Builder setEventDenylist(@Nullable List eventDenylist);
+
+ public abstract Builder setMaxContentLength(int maxContentLength);
+
+ public abstract Builder setProjectId(String projectId);
+
+ public abstract Builder setDatasetId(String datasetId);
+
+ public abstract Builder setTableName(String tableName);
+
+ public abstract Builder setClusteringFields(List clusteringFields);
+
+ public abstract Builder setLogMultiModalContent(boolean logMultiModalContent);
+
+ public abstract Builder setRetryConfig(RetryConfig retryConfig);
+
+ public abstract Builder setBatchSize(int batchSize);
+
+ public abstract Builder setBatchFlushInterval(Duration batchFlushInterval);
+
+ public abstract Builder setShutdownTimeout(Duration shutdownTimeout);
+
+ public abstract Builder setQueueMaxSize(int queueMaxSize);
+
+ public abstract Builder setContentFormatter(
+ @Nullable BiFunction contentFormatter);
+
+ public abstract Builder setConnectionId(String connectionId);
+
+ public abstract Builder setLogSessionMetadata(boolean logSessionMetadata);
+
+ public abstract Builder setCustomTags(Map customTags);
+
+ public abstract Builder setAutoSchemaUpgrade(boolean autoSchemaUpgrade);
+
+ public abstract Builder setCredentials(Credentials credentials);
+
+ public abstract BigQueryLoggerConfig build();
+ }
+
+ /** Retry configuration for BigQuery writes. */
+ @AutoValue
+ public abstract static class RetryConfig {
+ public abstract int maxRetries();
+
+ public abstract Duration initialDelay();
+
+ public abstract double multiplier();
+
+ public abstract Duration maxDelay();
+
+ public static Builder builder() {
+ return new AutoValue_BigQueryLoggerConfig_RetryConfig.Builder()
+ .setMaxRetries(3)
+ .setInitialDelay(Duration.ofSeconds(1))
+ .setMultiplier(2.0)
+ .setMaxDelay(Duration.ofSeconds(10));
+ }
+
+ /** Builder for {@link RetryConfig}. */
+ @AutoValue.Builder
+ public abstract static class Builder {
+ public abstract Builder setMaxRetries(int maxRetries);
+
+ public abstract Builder setInitialDelay(Duration initialDelay);
+
+ public abstract Builder setMultiplier(double multiplier);
+
+ public abstract Builder setMaxDelay(Duration maxDelay);
+
+ public abstract RetryConfig build();
+ }
+ }
+}
diff --git a/core/src/main/java/com/google/adk/plugins/agentanalytics/BigQuerySchema.java b/core/src/main/java/com/google/adk/plugins/agentanalytics/BigQuerySchema.java
new file mode 100644
index 000000000..81181a1e0
--- /dev/null
+++ b/core/src/main/java/com/google/adk/plugins/agentanalytics/BigQuerySchema.java
@@ -0,0 +1,304 @@
+/*
+ * Copyright 2026 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.adk.plugins.agentanalytics;
+
+import static com.google.common.collect.ImmutableList.toImmutableList;
+
+import com.google.cloud.bigquery.Field;
+import com.google.cloud.bigquery.FieldList;
+import com.google.cloud.bigquery.Schema;
+import com.google.cloud.bigquery.StandardSQLTypeName;
+import com.google.cloud.bigquery.storage.v1.TableFieldSchema;
+import com.google.cloud.bigquery.storage.v1.TableFieldSchema.Mode;
+import com.google.cloud.bigquery.storage.v1.TableSchema;
+import com.google.common.base.VerifyException;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.protobuf.ByteString;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.channels.Channels;
+import org.apache.arrow.vector.ipc.WriteChannel;
+import org.apache.arrow.vector.ipc.message.MessageSerializer;
+import org.apache.arrow.vector.types.DateUnit;
+import org.apache.arrow.vector.types.FloatingPointPrecision;
+import org.apache.arrow.vector.types.TimeUnit;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.FieldType;
+
+/** Utility for defining the BigQuery events table schema. */
+public final class BigQuerySchema {
+
+ private BigQuerySchema() {}
+
+ private static final ImmutableMap>
+ FIELD_TYPE_TO_ARROW_FIELD_METADATA =
+ ImmutableMap.of(
+ StandardSQLTypeName.JSON,
+ ImmutableMap.of("ARROW:extension:name", "google:sqlType:json"),
+ StandardSQLTypeName.DATETIME,
+ ImmutableMap.of("ARROW:extension:name", "google:sqlType:datetime"),
+ StandardSQLTypeName.GEOGRAPHY,
+ ImmutableMap.of(
+ "ARROW:extension:name",
+ "google:sqlType:geography",
+ "ARROW:extension:metadata",
+ "{\"encoding\": \"WKT\"}"));
+
+ /** Returns the BigQuery schema for the events table. */
+ // TODO(b/491848381): Rely on the same schema defined for python plugin.
+ public static Schema getEventsSchema() {
+ return Schema.of(
+ Field.newBuilder("timestamp", StandardSQLTypeName.TIMESTAMP)
+ .setMode(Field.Mode.REQUIRED)
+ .setDescription("The UTC timestamp when the event occurred.")
+ .build(),
+ Field.newBuilder("event_type", StandardSQLTypeName.STRING)
+ .setMode(Field.Mode.NULLABLE)
+ .setDescription("The category of the event.")
+ .build(),
+ Field.newBuilder("agent", StandardSQLTypeName.STRING)
+ .setMode(Field.Mode.NULLABLE)
+ .setDescription("The name of the agent that generated this event.")
+ .build(),
+ Field.newBuilder("session_id", StandardSQLTypeName.STRING)
+ .setMode(Field.Mode.NULLABLE)
+ .setDescription("A unique identifier for the entire conversation session.")
+ .build(),
+ Field.newBuilder("invocation_id", StandardSQLTypeName.STRING)
+ .setMode(Field.Mode.NULLABLE)
+ .setDescription("A unique identifier for a single turn or execution.")
+ .build(),
+ Field.newBuilder("user_id", StandardSQLTypeName.STRING)
+ .setMode(Field.Mode.NULLABLE)
+ .setDescription("The identifier of the end-user.")
+ .build(),
+ Field.newBuilder("trace_id", StandardSQLTypeName.STRING)
+ .setMode(Field.Mode.NULLABLE)
+ .setDescription("OpenTelemetry trace ID.")
+ .build(),
+ Field.newBuilder("span_id", StandardSQLTypeName.STRING)
+ .setMode(Field.Mode.NULLABLE)
+ .setDescription("OpenTelemetry span ID.")
+ .build(),
+ Field.newBuilder("parent_span_id", StandardSQLTypeName.STRING)
+ .setMode(Field.Mode.NULLABLE)
+ .setDescription("OpenTelemetry parent span ID.")
+ .build(),
+ Field.newBuilder("content", StandardSQLTypeName.JSON)
+ .setMode(Field.Mode.NULLABLE)
+ .setDescription("The primary payload of the event.")
+ .build(),
+ Field.newBuilder(
+ "content_parts",
+ StandardSQLTypeName.STRUCT,
+ FieldList.of(
+ Field.newBuilder("mime_type", StandardSQLTypeName.STRING)
+ .setMode(Field.Mode.NULLABLE)
+ .setDescription("The MIME type of the content part.")
+ .build(),
+ Field.newBuilder("uri", StandardSQLTypeName.STRING)
+ .setMode(Field.Mode.NULLABLE)
+ .setDescription("The URI of the content part if stored externally.")
+ .build(),
+ Field.newBuilder(
+ "object_ref",
+ StandardSQLTypeName.STRUCT,
+ FieldList.of(
+ Field.newBuilder("uri", StandardSQLTypeName.STRING)
+ .setMode(Field.Mode.NULLABLE)
+ .build(),
+ Field.newBuilder("version", StandardSQLTypeName.STRING)
+ .setMode(Field.Mode.NULLABLE)
+ .build(),
+ Field.newBuilder("authorizer", StandardSQLTypeName.STRING)
+ .setMode(Field.Mode.NULLABLE)
+ .build(),
+ Field.newBuilder("details", StandardSQLTypeName.JSON)
+ .setMode(Field.Mode.NULLABLE)
+ .build()))
+ .setMode(Field.Mode.NULLABLE)
+ .setDescription("The ObjectRef of the content part if stored externally.")
+ .build(),
+ Field.newBuilder("text", StandardSQLTypeName.STRING)
+ .setMode(Field.Mode.NULLABLE)
+ .setDescription("The raw text content.")
+ .build(),
+ Field.newBuilder("part_index", StandardSQLTypeName.INT64)
+ .setMode(Field.Mode.NULLABLE)
+ .setDescription("The zero-based index of this part.")
+ .build(),
+ Field.newBuilder("part_attributes", StandardSQLTypeName.STRING)
+ .setMode(Field.Mode.NULLABLE)
+ .setDescription("Additional metadata as a JSON object string.")
+ .build(),
+ Field.newBuilder("storage_mode", StandardSQLTypeName.STRING)
+ .setMode(Field.Mode.NULLABLE)
+ .setDescription("Indicates how the content part is stored.")
+ .build()))
+ .setMode(Field.Mode.REPEATED)
+ .setDescription("Multi-modal events content parts.")
+ .build(),
+ Field.newBuilder("attributes", StandardSQLTypeName.JSON)
+ .setMode(Field.Mode.NULLABLE)
+ .setDescription("A JSON object containing arbitrary key-value pairs.")
+ .build(),
+ Field.newBuilder("latency_ms", StandardSQLTypeName.JSON)
+ .setMode(Field.Mode.NULLABLE)
+ .setDescription("A JSON object containing latency measurements.")
+ .build(),
+ Field.newBuilder("status", StandardSQLTypeName.STRING)
+ .setMode(Field.Mode.NULLABLE)
+ .setDescription("The outcome of the event.")
+ .build(),
+ Field.newBuilder("error_message", StandardSQLTypeName.STRING)
+ .setMode(Field.Mode.NULLABLE)
+ .setDescription("Detailed error message if the status is 'ERROR'.")
+ .build(),
+ Field.newBuilder("is_truncated", StandardSQLTypeName.BOOL)
+ .setMode(Field.Mode.NULLABLE)
+ .setDescription("Indicates if the 'content' field was truncated.")
+ .build());
+ }
+
+ /** Returns the Arrow schema for the events table. */
+ public static org.apache.arrow.vector.types.pojo.Schema getArrowSchema() {
+ return new org.apache.arrow.vector.types.pojo.Schema(
+ getEventsSchema().getFields().stream()
+ .map(BigQuerySchema::convertToArrowField)
+ .collect(toImmutableList()));
+ }
+
+ /** Returns the serialized Arrow schema for the events table. */
+ public static ByteString getSerializedArrowSchema() {
+ try (ByteArrayOutputStream out = new ByteArrayOutputStream()) {
+ MessageSerializer.serialize(new WriteChannel(Channels.newChannel(out)), getArrowSchema());
+ return ByteString.copyFrom(out.toByteArray());
+ } catch (IOException e) {
+ throw new VerifyException("Failed to serialize arrow schema", e);
+ }
+ }
+
+ private static org.apache.arrow.vector.types.pojo.Field convertToArrowField(Field field) {
+ ArrowType arrowType = convertTypeToArrow(field.getType().getStandardType());
+ ImmutableList children = null;
+ if (field.getSubFields() != null) {
+ children =
+ field.getSubFields().stream()
+ .map(BigQuerySchema::convertToArrowField)
+ .collect(toImmutableList());
+ }
+
+ ImmutableMap metadata =
+ FIELD_TYPE_TO_ARROW_FIELD_METADATA.get(field.getType().getStandardType());
+
+ FieldType fieldType =
+ new FieldType(field.getMode() != Field.Mode.REQUIRED, arrowType, null, metadata);
+ org.apache.arrow.vector.types.pojo.Field arrowField =
+ new org.apache.arrow.vector.types.pojo.Field(field.getName(), fieldType, children);
+
+ if (field.getMode() == Field.Mode.REPEATED) {
+ return new org.apache.arrow.vector.types.pojo.Field(
+ field.getName(),
+ new FieldType(false, new ArrowType.List(), null),
+ ImmutableList.of(
+ new org.apache.arrow.vector.types.pojo.Field(
+ "element", arrowField.getFieldType(), arrowField.getChildren())));
+ }
+ return arrowField;
+ }
+
+ private static ArrowType convertTypeToArrow(StandardSQLTypeName type) {
+ return switch (type) {
+ case BOOL -> new ArrowType.Bool();
+ case BYTES -> new ArrowType.Binary();
+ case DATE -> new ArrowType.Date(DateUnit.DAY);
+ case DATETIME ->
+ // Arrow doesn't have a direct DATETIME, often mapped to Timestamp or Utf8
+ new ArrowType.Utf8();
+ case FLOAT64 -> new ArrowType.FloatingPoint(FloatingPointPrecision.DOUBLE);
+ case INT64 -> new ArrowType.Int(64, true);
+ case NUMERIC, BIGNUMERIC -> new ArrowType.Decimal(38, 9, 128);
+ case GEOGRAPHY, STRING, JSON -> new ArrowType.Utf8();
+ case STRUCT -> new ArrowType.Struct();
+ case TIME -> new ArrowType.Time(TimeUnit.MICROSECOND, 64);
+ case TIMESTAMP -> new ArrowType.Timestamp(TimeUnit.MICROSECOND, "UTC");
+ default -> new ArrowType.Null();
+ };
+ }
+
+ /** Returns names of fields to cluster by default. */
+ public static ImmutableList getDefaultClusteringFields() {
+ return ImmutableList.of("event_type", "agent", "user_id");
+ }
+
+ /** Returns the BigQuery TableSchema for the events table (Storage Write API). */
+ public static TableSchema getEventsTableSchema() {
+ return convertTableSchema(getEventsSchema());
+ }
+
+ private static TableSchema convertTableSchema(Schema schema) {
+ TableSchema.Builder result = TableSchema.newBuilder();
+ for (int i = 0; i < schema.getFields().size(); i++) {
+ result.addFields(i, convertFieldSchema(schema.getFields().get(i)));
+ }
+ return result.build();
+ }
+
+ private static TableFieldSchema convertFieldSchema(Field field) {
+ TableFieldSchema.Builder result = TableFieldSchema.newBuilder();
+ Field.Mode mode = field.getMode() != null ? field.getMode() : Field.Mode.NULLABLE;
+
+ Mode resultMode = Mode.valueOf(mode.name());
+ result.setMode(resultMode).setName(field.getName());
+
+ StandardSQLTypeName standardType = field.getType().getStandardType();
+ TableFieldSchema.Type resultType = convertType(standardType);
+ result.setType(resultType);
+
+ if (field.getDescription() != null) {
+ result.setDescription(field.getDescription());
+ }
+ if (field.getSubFields() != null) {
+ for (int i = 0; i < field.getSubFields().size(); i++) {
+ result.addFields(i, convertFieldSchema(field.getSubFields().get(i)));
+ }
+ }
+ return result.build();
+ }
+
+ private static TableFieldSchema.Type convertType(StandardSQLTypeName type) {
+ return switch (type) {
+ case BOOL -> TableFieldSchema.Type.BOOL;
+ case BYTES -> TableFieldSchema.Type.BYTES;
+ case DATE -> TableFieldSchema.Type.DATE;
+ case DATETIME -> TableFieldSchema.Type.DATETIME;
+ case FLOAT64 -> TableFieldSchema.Type.DOUBLE;
+ case GEOGRAPHY -> TableFieldSchema.Type.GEOGRAPHY;
+ case INT64 -> TableFieldSchema.Type.INT64;
+ case NUMERIC -> TableFieldSchema.Type.NUMERIC;
+ case STRING -> TableFieldSchema.Type.STRING;
+ case STRUCT -> TableFieldSchema.Type.STRUCT;
+ case TIME -> TableFieldSchema.Type.TIME;
+ case TIMESTAMP -> TableFieldSchema.Type.TIMESTAMP;
+ case BIGNUMERIC -> TableFieldSchema.Type.BIGNUMERIC;
+ case JSON -> TableFieldSchema.Type.JSON;
+ case INTERVAL -> TableFieldSchema.Type.INTERVAL;
+ default -> TableFieldSchema.Type.TYPE_UNSPECIFIED;
+ };
+ }
+}
diff --git a/core/src/main/java/com/google/adk/plugins/agentanalytics/JsonFormatter.java b/core/src/main/java/com/google/adk/plugins/agentanalytics/JsonFormatter.java
new file mode 100644
index 000000000..b4b4a1049
--- /dev/null
+++ b/core/src/main/java/com/google/adk/plugins/agentanalytics/JsonFormatter.java
@@ -0,0 +1,111 @@
+/*
+ * Copyright 2026 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.adk.plugins.agentanalytics;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.collect.ImmutableList;
+import com.google.genai.types.Blob;
+import com.google.genai.types.Content;
+import com.google.genai.types.FileData;
+import com.google.genai.types.Part;
+import java.util.List;
+import java.util.Optional;
+
+/** Utility for formatting and truncating content for BigQuery logging. */
+final class JsonFormatter {
+ private static final ObjectMapper mapper = new ObjectMapper().findAndRegisterModules();
+
+ private JsonFormatter() {}
+
+ /** Formats Content parts into an ArrayNode for BigQuery logging. */
+ public static ArrayNode formatContentParts(Optional content, int maxLength) {
+ ArrayNode partsArray = mapper.createArrayNode();
+ if (content.isEmpty() || content.get().parts() == null) {
+ return partsArray;
+ }
+
+ List parts = content.get().parts().orElse(ImmutableList.of());
+
+ for (int i = 0; i < parts.size(); i++) {
+ Part part = parts.get(i);
+ ObjectNode partObj = mapper.createObjectNode();
+ partObj.put("part_index", i);
+ partObj.put("storage_mode", "INLINE");
+
+ if (part.text().isPresent()) {
+ partObj.put("mime_type", "text/plain");
+ partObj.put("text", truncateString(part.text().get(), maxLength));
+ } else if (part.inlineData().isPresent()) {
+ Blob blob = part.inlineData().get();
+ partObj.put("mime_type", blob.mimeType().orElse(""));
+ partObj.put("text", "[BINARY DATA]");
+ } else if (part.fileData().isPresent()) {
+ FileData fileData = part.fileData().get();
+ partObj.put("mime_type", fileData.mimeType().orElse(""));
+ partObj.put("uri", fileData.fileUri().orElse(""));
+ partObj.put("storage_mode", "EXTERNAL_URI");
+ }
+ partsArray.add(partObj);
+ }
+ return partsArray;
+ }
+
+ /** Recursively truncates long strings inside an object and returns a Jackson JsonNode. */
+ public static JsonNode smartTruncate(Object obj, int maxLength) {
+ if (obj == null) {
+ return mapper.nullNode();
+ }
+ try {
+ return recursiveSmartTruncate(mapper.valueToTree(obj), maxLength);
+ } catch (IllegalArgumentException e) {
+ // Fallback for types that mapper can't handle directly as a tree
+ return mapper.valueToTree(String.valueOf(obj));
+ }
+ }
+
+ private static JsonNode recursiveSmartTruncate(JsonNode node, int maxLength) {
+ if (node.isTextual()) {
+ return mapper.valueToTree(truncateString(node.asText(), maxLength));
+ } else if (node.isObject()) {
+ ObjectNode newNode = mapper.createObjectNode();
+ node.properties()
+ .iterator()
+ .forEachRemaining(
+ entry -> {
+ newNode.set(entry.getKey(), recursiveSmartTruncate(entry.getValue(), maxLength));
+ });
+ return newNode;
+ } else if (node.isArray()) {
+ ArrayNode newNode = mapper.createArrayNode();
+ for (JsonNode element : node) {
+ newNode.add(recursiveSmartTruncate(element, maxLength));
+ }
+ return newNode;
+ }
+ return node;
+ }
+
+ private static String truncateString(String s, int maxLength) {
+ if (s == null || s.length() <= maxLength) {
+ return s;
+ }
+ return s.substring(0, maxLength) + "...[truncated]";
+ }
+}
diff --git a/core/src/main/java/com/google/adk/runner/Runner.java b/core/src/main/java/com/google/adk/runner/Runner.java
index 5859c4786..2bfbca881 100644
--- a/core/src/main/java/com/google/adk/runner/Runner.java
+++ b/core/src/main/java/com/google/adk/runner/Runner.java
@@ -52,6 +52,7 @@
import com.google.genai.types.Part;
import io.opentelemetry.api.trace.Span;
import io.opentelemetry.api.trace.StatusCode;
+import io.opentelemetry.context.Context;
import io.reactivex.rxjava3.core.Completable;
import io.reactivex.rxjava3.core.Flowable;
import io.reactivex.rxjava3.core.Maybe;
@@ -312,6 +313,7 @@ private Single appendNewMessageToSession(
throw new IllegalArgumentException("No parts in the new_message.");
}
+ Completable saveArtifactsFlow = Completable.complete();
if (this.artifactService != null && saveInputBlobsAsArtifacts) {
// The runner directly saves the artifacts (if applicable) in the user message and replaces
// the artifact data with a file name placeholder.
@@ -321,9 +323,11 @@ private Single appendNewMessageToSession(
continue;
}
String fileName = "artifact_" + invocationContext.invocationId() + "_" + i;
- var unused =
- this.artifactService.saveArtifact(
- this.appName, session.userId(), session.id(), fileName, part);
+ saveArtifactsFlow =
+ saveArtifactsFlow.andThen(
+ this.artifactService
+ .saveArtifact(this.appName, session.userId(), session.id(), fileName, part)
+ .ignoreElement());
newMessage
.parts()
@@ -348,7 +352,8 @@ private Single appendNewMessageToSession(
EventActions.builder().stateDelta(new ConcurrentHashMap<>(stateDelta)).build());
}
- return this.sessionService.appendEvent(session, eventBuilder.build());
+ return saveArtifactsFlow.andThen(
+ this.sessionService.appendEvent(session, eventBuilder.build()));
}
/** See {@link #runAsync(String, String, Content, RunConfig, Map)}. */
@@ -375,20 +380,25 @@ public Flowable runAsync(
Content newMessage,
RunConfig runConfig,
@Nullable Map stateDelta) {
- Maybe