-U postgres -v ON_ERROR_STOP=1 -f postgresql-setup.sql
+--
+-- It creates no tables. Those come from DatabaseSessionService.migrate(), or from running the
+-- module's postgresql/V1__create_session_schema.sql as adk_sessions; the README has both ways.
+--
+-- Replace both passwords before use.
+
+-- The owner: creates the tables, so it is the role for migrate() and for running the schema
+-- script by hand. It may also be the only role, used by the application as well.
+CREATE ROLE adk_sessions LOGIN PASSWORD 'change-me';
+
+-- A database of its own keeps the tables, whose names are as plain as "sessions" and "events",
+-- away from the application's. template0 lets the encoding be UTF8 whatever the cluster default.
+CREATE DATABASE adk_sessions OWNER adk_sessions ENCODING 'UTF8' TEMPLATE template0;
+
+-- Optional: a second role for the application at run time. It reads and writes the rows and
+-- cannot create, alter or drop a table. Leave this part out to use adk_sessions for everything.
+CREATE ROLE adk_sessions_app LOGIN PASSWORD 'change-me-too';
+
+\connect adk_sessions
+
+-- Applies to the tables adk_sessions creates from now on, so run this script before the tables
+-- exist. For tables that are already there, grant the same four rights with
+-- GRANT SELECT, INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA public TO adk_sessions_app;
+ALTER DEFAULT PRIVILEGES FOR ROLE adk_sessions IN SCHEMA public
+ GRANT SELECT, INSERT, UPDATE, DELETE ON TABLES TO adk_sessions_app;
diff --git a/contrib/database-session-service/src/main/java/com/google/adk/sessions/DatabaseDiagnostics.java b/contrib/database-session-service/src/main/java/com/google/adk/sessions/DatabaseDiagnostics.java
new file mode 100644
index 000000000..54c578b4a
--- /dev/null
+++ b/contrib/database-session-service/src/main/java/com/google/adk/sessions/DatabaseDiagnostics.java
@@ -0,0 +1,191 @@
+/*
+ * 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.sessions;
+
+import com.google.adk.sessions.DatabaseSessionService.Dialect;
+import java.sql.Connection;
+import java.sql.SQLException;
+import org.jspecify.annotations.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Reports stable diagnostic fields without passing a driver exception to a caller. The driver's own
+ * exception, whose message can quote SQL or data, is logged only at DEBUG, which an operator turns
+ * on deliberately.
+ *
+ * A database failure therefore reaches two audiences in two forms. The caller gets a {@link
+ * SessionException} with a message from this module and no cause. The operator gets one WARN line
+ * with the operation, the phase it failed in, the dialect, the exception's class, its SQLSTATE and
+ * its vendor code, which is what is needed to look the failure up.
+ */
+final class DatabaseDiagnostics {
+ private static final Logger logger = LoggerFactory.getLogger(DatabaseDiagnostics.class);
+
+ private DatabaseDiagnostics() {}
+
+ /** The step of an operation in which a failure happened, in the order the steps run. */
+ enum Phase {
+ /** Borrowing a connection from the data source. */
+ CONNECTION,
+ /** Checking the borrowed connection, setting isolation, and beginning the transaction. */
+ SETUP,
+ /** Checking that the schema is the one this version reads and writes. */
+ SCHEMA,
+ /** Running the statements of the operation. */
+ STATEMENT,
+ /** Committing. */
+ COMMIT,
+ /** Restoring the connection's settings and returning it. */
+ CLEANUP,
+ /** Running the schema migration, which is not part of an operation. */
+ MIGRATION
+ }
+
+ /**
+ * Requires a connection this module can run its own transaction on.
+ *
+ *
A connection with auto-commit off is, as far as can be told, in the middle of the caller's
+ * transaction: committing on it would commit the caller's work, and rolling back would discard
+ * it. It is refused before anything is executed.
+ *
+ * @param write whether the operation changes data, which a read-only connection cannot do
+ * @throws Configuration with a message naming what the data source has to provide
+ */
+ static void requireConnection(Connection connection, boolean write) throws SQLException {
+ if (!connection.getAutoCommit()) {
+ throw new Configuration("ADK requires a borrowed connection with auto-commit enabled.");
+ }
+ if (write && connection.isReadOnly()) {
+ throw new Configuration("ADK writes require a writable database connection.");
+ }
+ }
+
+ /**
+ * Logs a failure and returns the exception to throw in its place.
+ *
+ * @param message the text for the caller; written by this module, without data from {@code
+ * failure}
+ * @param failure what actually went wrong; reported to the log, and deliberately not attached as
+ * the cause
+ */
+ static SessionException failure(
+ String operation, Phase phase, @Nullable Dialect dialect, String message, Throwable failure) {
+ report(operation, phase, dialect, failure);
+ return new SessionException(message);
+ }
+
+ /**
+ * Logs a failure: its coded fields at WARN, and the exception itself with its stack trace at
+ * DEBUG.
+ *
+ * @param dialect {@code null} while the dialect is still being detected
+ */
+ static void report(String operation, Phase phase, @Nullable Dialect dialect, Throwable failure) {
+ SqlDiagnostic code = sqlDiagnostic(failure);
+ logger.warn(
+ "ADK JDBC failure: operation={} phase={} dialect={} exceptionType={} sqlState={} vendorCode={}",
+ operation,
+ phase,
+ dialect,
+ failure.getClass().getName(),
+ code.state(),
+ code.vendorCode());
+ if (logger.isDebugEnabled()) {
+ logger.debug("ADK JDBC failure detail: operation={} phase={}", operation, phase, failure);
+ }
+ }
+
+ /**
+ * The two coded fields of a JDBC failure.
+ *
+ * @param state the five-character SQLSTATE, or {@code "unavailable"}
+ * @param vendorCode the engine's own error number, or 0
+ */
+ record SqlDiagnostic(String state, int vendorCode) {}
+
+ /**
+ * Extracts the coded fields of the first {@link SQLException} in the cause chain. Messages are
+ * never read.
+ */
+ static SqlDiagnostic sqlDiagnostic(Throwable failure) {
+ SQLException sql = sqlException(failure);
+ if (sql == null) {
+ return new SqlDiagnostic("unavailable", 0);
+ }
+ String state = sql.getSQLState();
+ return new SqlDiagnostic(isSqlState(state) ? state : "unavailable", sql.getErrorCode());
+ }
+
+ /**
+ * Whether the database itself gave up on the transaction because another one held conflicting
+ * locks: a deadlock victim, a serialization failure, or a lock wait that timed out.
+ */
+ static boolean concurrencyConflict(Throwable failure, Dialect dialect) {
+ SQLException sql = sqlException(failure);
+ if (sql == null) {
+ return false;
+ }
+ String state = sql.getSQLState();
+ // Class 40 is the standard "transaction rollback" class on every engine.
+ if (state != null && state.startsWith("40")) {
+ return true;
+ }
+ int code = sql.getErrorCode();
+ return switch (dialect) {
+ case POSTGRESQL -> "55P03".equals(state); // lock_not_available, including lock_timeout
+ case MYSQL, MARIADB -> code == 1205 || code == 1213; // lock wait timeout, deadlock
+ // SQLITE_BUSY and SQLITE_LOCKED; the low byte also matches their extended codes.
+ case SQLITE -> (code & 0xFF) == 5 || (code & 0xFF) == 6;
+ };
+ }
+
+ /** Finds the first {@link SQLException} among a failure and its causes, a few levels deep. */
+ private static @Nullable SQLException sqlException(Throwable failure) {
+ // Flyway may wrap the SQLException. Bound traversal and never read exception messages.
+ Throwable cause = failure;
+ for (int depth = 0; cause != null && depth < 8; depth++, cause = cause.getCause()) {
+ if (cause instanceof SQLException sql) {
+ return sql;
+ }
+ }
+ return null;
+ }
+
+ /** SQLSTATE is five upper-case letters or digits; anything else is not a code to repeat. */
+ private static boolean isSqlState(@Nullable String state) {
+ if (state == null || state.length() != 5) {
+ return false;
+ }
+ for (int i = 0; i < 5; i++) {
+ char c = state.charAt(i);
+ if (!(c >= '0' && c <= '9') && !(c >= 'A' && c <= 'Z')) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ /**
+ * A known configuration error whose message is defined by this module. Because the text is the
+ * module's own, it is the one failure whose message is handed to the caller unchanged.
+ */
+ static final class Configuration extends RuntimeException {
+ Configuration(String message) {
+ super(message);
+ }
+ }
+}
diff --git a/contrib/database-session-service/src/main/java/com/google/adk/sessions/DatabaseDialect.java b/contrib/database-session-service/src/main/java/com/google/adk/sessions/DatabaseDialect.java
new file mode 100644
index 000000000..9e88c19cf
--- /dev/null
+++ b/contrib/database-session-service/src/main/java/com/google/adk/sessions/DatabaseDialect.java
@@ -0,0 +1,149 @@
+/*
+ * 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
+ *
+ * https://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.sessions;
+
+import com.google.adk.sessions.DatabaseDiagnostics.Phase;
+import com.google.adk.sessions.DatabaseSessionService.Dialect;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.sql.Types;
+import java.util.Locale;
+import javax.sql.DataSource;
+
+/**
+ * The places where the four engines need different SQL or a different JDBC call. Everything else in
+ * {@link DatabaseSessionStore} is the same statement for all of them.
+ *
+ *
The table definitions differ too; those are in the per-dialect migration scripts under {@code
+ * com/google/adk/sessions/migration}.
+ */
+final class DatabaseDialect {
+ private DatabaseDialect() {}
+
+ /**
+ * The statement that reads the two rows identifying the schema: the layout version, which the
+ * Python v1 schema defines, and the marker saying that this module's Java format is stored in it.
+ * {@code key} is a reserved word in MySQL and MariaDB and has to be quoted there.
+ */
+ static String metadataQuery(Dialect dialect) {
+ String key = dialect == Dialect.MYSQL || dialect == Dialect.MARIADB ? "`key`" : "key";
+ return "SELECT "
+ + key
+ + ",value FROM adk_internal_metadata WHERE "
+ + key
+ + " IN ('schema_version','adk_java_storage_format')";
+ }
+
+ /**
+ * Binds JSON text to a state or event column. PostgreSQL's {@code json} column does not accept a
+ * parameter typed as text, so the value is sent untyped and the server takes it as JSON. The
+ * other engines keep JSON in text columns.
+ */
+ static void bindJson(PreparedStatement statement, int index, String json, Dialect dialect)
+ throws SQLException {
+ if (dialect == Dialect.POSTGRESQL) {
+ statement.setObject(index, json, Types.OTHER);
+ } else {
+ statement.setString(index, json);
+ }
+ }
+
+ /**
+ * The clause that makes a {@code SELECT} lock the rows it reads until the transaction ends.
+ * SQLite has no such clause and needs none: a write transaction there is the only writer.
+ */
+ static String lock(Dialect dialect) {
+ return dialect == Dialect.SQLITE ? "" : " FOR UPDATE";
+ }
+
+ /**
+ * The statement that inserts an empty shared-state row unless the row is already there.
+ *
+ *
Parameters: the app name, the user ID if {@code user}, then the JSON document. The update
+ * time is the database's clock.
+ *
+ *
MySQL and MariaDB have no {@code ON CONFLICT} clause. Their counterpart is {@code ON
+ * DUPLICATE KEY UPDATE} with an assignment that changes nothing. On an existing row it takes an
+ * exclusive lock right away, where a plain {@code INSERT} that hits a duplicate takes a shared
+ * lock first, which is how two concurrent inserters of the same key come to deadlock.
+ *
+ * @param table {@code app_states} or {@code user_states}
+ * @param user whether the row is per user
+ */
+ static String insertShared(String table, boolean user, Dialect dialect) {
+ String columns = user ? "app_name,user_id,state,update_time" : "app_name,state,update_time";
+ String values = (user ? "?,?,?," : "?,?,") + utcNow(dialect);
+ String conflict =
+ dialect == Dialect.MYSQL || dialect == Dialect.MARIADB
+ ? " ON DUPLICATE KEY UPDATE app_name=app_name"
+ : " ON CONFLICT DO NOTHING";
+ return "INSERT INTO " + table + " (" + columns + ") VALUES (" + values + ")" + conflict;
+ }
+
+ /**
+ * An expression for the database's current time as a UTC value without a zone, at the microsecond
+ * precision of the time columns. It does not depend on the time zone of the session, the server
+ * or the JVM.
+ */
+ static String utcNow(Dialect dialect) {
+ return switch (dialect) {
+ case POSTGRESQL -> "(CURRENT_TIMESTAMP AT TIME ZONE 'UTC')";
+ case MYSQL, MARIADB -> "UTC_TIMESTAMP(6)";
+ // SQLite's %f has three fractional digits; use the same six-digit format as JDBC writes.
+ case SQLITE -> "strftime('%Y-%m-%d %H:%M:%f','now') || '000'";
+ };
+ }
+
+ /**
+ * Detects the dialect from the product name the JDBC driver reports, borrowing one connection.
+ * The JDBC URL is never looked at: a pool or proxy may not expose it, and it can hold
+ * credentials.
+ *
+ * @throws SessionException if no connection could be borrowed or the product is not one of the
+ * four supported ones; the message does not repeat the product name
+ */
+ static Dialect detect(DataSource dataSource) {
+ String productName;
+ try (Connection connection = dataSource.getConnection()) {
+ productName = connection.getMetaData().getDatabaseProductName();
+ } catch (SQLException | RuntimeException driverFailure) {
+ throw DatabaseDiagnostics.failure(
+ "detectDialect",
+ Phase.CONNECTION,
+ null,
+ "Unable to determine the database dialect from JDBC metadata.",
+ driverFailure);
+ }
+
+ if (productName != null) {
+ switch (productName.strip().toLowerCase(Locale.ROOT)) {
+ case "sqlite":
+ return Dialect.SQLITE;
+ case "postgresql":
+ return Dialect.POSTGRESQL;
+ case "mysql":
+ return Dialect.MYSQL;
+ case "mariadb":
+ return Dialect.MARIADB;
+ default:
+ break;
+ }
+ }
+ throw new SessionException(
+ "Unsupported database product; select a supported dialect explicitly.");
+ }
+}
diff --git a/contrib/database-session-service/src/main/java/com/google/adk/sessions/DatabaseSchemaMigrator.java b/contrib/database-session-service/src/main/java/com/google/adk/sessions/DatabaseSchemaMigrator.java
new file mode 100644
index 000000000..6647c502c
--- /dev/null
+++ b/contrib/database-session-service/src/main/java/com/google/adk/sessions/DatabaseSchemaMigrator.java
@@ -0,0 +1,217 @@
+/*
+ * 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
+ *
+ * https://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.sessions;
+
+import com.google.adk.sessions.DatabaseDiagnostics.Phase;
+import com.google.adk.sessions.DatabaseSessionService.Dialect;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.HashSet;
+import java.util.Locale;
+import java.util.Set;
+import javax.sql.DataSource;
+import org.flywaydb.core.api.configuration.FluentConfiguration;
+
+/**
+ * Isolated, explicit Flyway bootstrap. Never adopts tables merely because their names match.
+ *
+ *
Flyway is used as an embedded library over the application's data source, kept apart from any
+ * Flyway the application runs for itself:
+ *
+ *
+ * - The scripts are in {@code com/google/adk/sessions/migration/}, not in Flyway's
+ * default {@code db/migration}, so an application's own Flyway does not pick them up.
+ *
- The history is kept in {@code adk_schema_history}, not in the application's {@code
+ * flyway_schema_history}.
+ *
- A schema that already holds other tables is baselined at version 0, so that version 1 of
+ * this module still runs in it.
+ *
+ *
+ * The table names are those of the Python v1 schema and are common words. Before and after
+ * migrating, the tables found under those names are therefore checked to be this module's own; see
+ * {@link #validateOwnership}.
+ *
+ *
This is the only class that refers to Flyway. A deployment that creates the tables from the
+ * SQL scripts itself, and never calls {@code migrate()}, does not load it.
+ */
+final class DatabaseSchemaMigrator {
+ /** Value of {@code schema_version}: the layout version, shared with the Python v1 schema. */
+ static final String SCHEMA_VERSION = "1";
+
+ /**
+ * Value of {@code adk_java_storage_format}: says that the events and times in these tables were
+ * written by this module. A schema created by ADK Python has the same {@link #SCHEMA_VERSION} but
+ * stores events in another JSON form, and lacks this row.
+ */
+ static final String STORAGE_FORMAT = "jdbc-v1";
+
+ /** The tables of the session schema together with this module's Flyway history table. */
+ private static final Set TABLES =
+ Set.of(
+ "adk_internal_metadata",
+ "sessions",
+ "events",
+ "app_states",
+ "user_states",
+ "adk_schema_history");
+
+ private DatabaseSchemaMigrator() {}
+
+ /**
+ * Brings the session schema to the current version, creating it in an empty or foreign schema and
+ * doing nothing when it is up to date.
+ *
+ * @throws SessionException if tables of the same names exist that this module did not create, the
+ * schema is only partly there, or the migration itself fails
+ */
+ static void migrate(DataSource dataSource, Dialect dialect) {
+ Phase phase = Phase.SCHEMA;
+ try {
+ // First, so that a foreign "sessions" or "events" table stops the migration before Flyway
+ // creates anything next to it.
+ validateOwnership(dataSource, dialect);
+ phase = Phase.MIGRATION;
+ // The class loader of this module, so that the scripts are found in its jar whatever the
+ // thread's context class loader is.
+ new FluentConfiguration(DatabaseSchemaMigrator.class.getClassLoader())
+ .dataSource(dataSource)
+ .locations(
+ "classpath:com/google/adk/sessions/migration/"
+ + dialect.name().toLowerCase(Locale.ROOT))
+ .table("adk_schema_history")
+ // Without a baseline Flyway refuses a schema that has tables but no history table,
+ // which is what an application's own database looks like on the first run.
+ .baselineOnMigrate(true)
+ .baselineVersion("0")
+ // This module must never be able to drop the tables of the schema it shares.
+ .cleanDisabled(true)
+ .failOnMissingLocations(true)
+ .validateMigrationNaming(true)
+ // An empty list of loggers keeps Flyway quiet; its own log lines include the JDBC URL.
+ .loggers(new String[0])
+ .load()
+ .migrate();
+ phase = Phase.SCHEMA;
+ // On a new connection, as the service will use: a data source that hands out private
+ // in-memory databases would have migrated one that nobody can reach afterwards.
+ if (!validateOwnership(dataSource, dialect)) {
+ throw new DatabaseDiagnostics.Configuration(
+ "The migrated ADK schema is not visible on a new database connection.");
+ }
+ } catch (DatabaseDiagnostics.Configuration configuration) {
+ throw new SessionException(configuration.getMessage());
+ } catch (SQLException | RuntimeException e) {
+ throw DatabaseDiagnostics.failure(
+ "migrate",
+ phase,
+ dialect,
+ phase == Phase.SCHEMA
+ ? "Unable to verify ADK schema ownership before or after migration."
+ : "Unable to migrate the ADK database schema; verify database and Flyway configuration.",
+ e);
+ }
+ }
+
+ /**
+ * Looks at the tables that carry one of this module's names in the schema the data source
+ * connects to, and decides whose they are.
+ *
+ *
+ * - None of them exists: nothing to own yet.
+ *
- All of them exist and the metadata rows are this module's: the schema is ours.
+ *
- Anything in between is refused: tables without the metadata table belong to someone else,
+ * metadata without the Java marker is a Python schema, and an incomplete set is a migration
+ * that stopped half way or a table that was dropped.
+ *
+ *
+ * @return whether the complete schema of this module is present
+ * @throws DatabaseDiagnostics.Configuration if tables are present that are not this module's
+ * complete schema
+ */
+ private static boolean validateOwnership(DataSource dataSource, Dialect dialect)
+ throws SQLException {
+ try (Connection connection = dataSource.getConnection()) {
+ DatabaseDiagnostics.requireConnection(connection, true);
+ // Only the catalog and schema of this connection: tables of the same names elsewhere on
+ // the server are none of this module's business. Names are compared in lower case because
+ // engines differ in how they report the case of an unquoted name.
+ Set existing = new HashSet<>();
+ try (ResultSet tables =
+ connection
+ .getMetaData()
+ .getTables(
+ connection.getCatalog(),
+ connection.getSchema(),
+ null,
+ new String[] {"TABLE", "VIEW"})) {
+ while (tables.next()) {
+ String name = tables.getString("TABLE_NAME").toLowerCase(Locale.ROOT);
+ if (TABLES.contains(name)) {
+ existing.add(name);
+ }
+ }
+ }
+ if (existing.isEmpty()) {
+ return false;
+ }
+ if (!existing.contains("adk_internal_metadata")) {
+ throw new DatabaseDiagnostics.Configuration(
+ "The database session schema was not created by ADK Java.");
+ }
+ try (Statement statement = connection.createStatement();
+ ResultSet metadata = statement.executeQuery(DatabaseDialect.metadataQuery(dialect))) {
+ // A Python V1 schema has the same version but no Java storage-format marker.
+ requireMetadata(metadata);
+ }
+ if (!existing.equals(TABLES)) {
+ throw new DatabaseDiagnostics.Configuration(
+ "ADK migration found a partial or conflicting schema.");
+ }
+ return true;
+ }
+ }
+
+ /**
+ * Requires both identifying rows of {@code adk_internal_metadata} to have the values this version
+ * writes. The store calls it at the start of every transaction, with the rows read inside that
+ * transaction.
+ *
+ * @param metadata key and value columns, as selected by {@link DatabaseDialect#metadataQuery}
+ * @throws DatabaseDiagnostics.Configuration if the Java marker is missing or different, or the
+ * layout version is not the supported one
+ */
+ static void requireMetadata(ResultSet metadata) throws SQLException {
+ String format = null;
+ String version = null;
+ while (metadata.next()) {
+ switch (metadata.getString(1)) {
+ case "adk_java_storage_format" -> format = metadata.getString(2);
+ case "schema_version" -> version = metadata.getString(2);
+ default -> {}
+ }
+ }
+ if (!STORAGE_FORMAT.equals(format)) {
+ throw new DatabaseDiagnostics.Configuration(
+ "The database session schema was not created by ADK Java.");
+ }
+ if (!SCHEMA_VERSION.equals(version)) {
+ throw new DatabaseDiagnostics.Configuration(
+ "The ADK database schema version is missing or unsupported.");
+ }
+ }
+}
diff --git a/contrib/database-session-service/src/main/java/com/google/adk/sessions/DatabaseSessionService.java b/contrib/database-session-service/src/main/java/com/google/adk/sessions/DatabaseSessionService.java
new file mode 100644
index 000000000..e791984f5
--- /dev/null
+++ b/contrib/database-session-service/src/main/java/com/google/adk/sessions/DatabaseSessionService.java
@@ -0,0 +1,652 @@
+/*
+ * 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
+ *
+ * https://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.sessions;
+
+import com.google.adk.events.Event;
+import com.google.adk.platform.UuidProvider;
+import com.google.errorprone.annotations.CanIgnoreReturnValue;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Maybe;
+import io.reactivex.rxjava3.core.Single;
+import java.time.Duration;
+import java.time.Instant;
+import java.time.InstantSource;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.ConcurrentMap;
+import javax.sql.DataSource;
+import org.jspecify.annotations.Nullable;
+
+/**
+ * JDBC session persistence for SQLite, PostgreSQL, MySQL and MariaDB using an application-owned
+ * {@link DataSource}.
+ *
+ * Construction never borrows a connection, runs migrations, or creates a connection pool. The
+ * application remains responsible for the lifetime of its data source and its providers.
+ *
+ *
{@code
+ * DatabaseSessionService sessions = new DatabaseSessionService(dataSource);
+ * sessions.migrate().blockingAwait(); // once, before workers start
+ *
+ * Runner runner =
+ * Runner.builder().appName("my-app").agent(agent).sessionService(sessions).build();
+ * }
+ *
+ * Execution. Every operation is cold: nothing happens until the returned {@code Single},
+ * {@code Maybe} or {@code Completable} is subscribed, and each subscription runs the operation
+ * again. The JDBC calls block the subscribing thread; move them with {@code subscribeOn} if that
+ * thread must not block.
+ *
+ *
State. Keys with the {@link State#APP_PREFIX} and {@link State#USER_PREFIX} prefixes
+ * are shared by all sessions of the app, or of the app and user; other keys belong to one session.
+ * Keys with {@link State#TEMP_PREFIX} are never stored.
+ *
+ *
Errors. Operations fail only with {@link SessionException} or {@link
+ * SessionNotFoundException}. Their messages are written by this module: they never quote a driver
+ * message, an identifier, or a payload, and they carry no cause. What the driver reported is in the
+ * log.
+ *
+ *
Instances are safe to share between threads. One instance per data source is enough.
+ */
+public final class DatabaseSessionService implements BaseSessionService {
+
+ /** Database engines with built-in SQL dialects. */
+ public enum Dialect {
+ /** SQLite through a file-backed database; one writer at a time. */
+ SQLITE,
+ /** PostgreSQL. */
+ POSTGRESQL,
+ /** MySQL with InnoDB. */
+ MYSQL,
+ /** MariaDB with InnoDB. Its SQL differs from MySQL's in the collation of the keys. */
+ MARIADB
+ }
+
+ private static final int DEFAULT_STATEMENT_TIMEOUT_SECONDS = 15;
+ // SQLite takes its busy timeout in milliseconds as an int.
+ private static final int MAX_STATEMENT_TIMEOUT_SECONDS = Integer.MAX_VALUE / 1000;
+
+ private final DataSource dataSource;
+ private final InstantSource instantSource;
+ private final UuidProvider uuidProvider;
+ private final boolean autoMigrate;
+ private final int statementTimeoutSeconds;
+
+ // The three fields below are resolved on first use, not at construction, and are guarded by
+ // this instance's monitor.
+ private @Nullable Dialect resolvedDialect;
+ private boolean initialized;
+ private @Nullable DatabaseSessionStore store;
+
+ /**
+ * Configures a service with migrations disabled and lazy JDBC metadata dialect detection.
+ *
+ *
The schema must already exist when the first operation runs: call {@link #migrate()} once,
+ * or create the tables with the SQL shipped in this module.
+ *
+ * @throws SessionException if the data source is null
+ */
+ public DatabaseSessionService(DataSource dataSource) {
+ this(builder(dataSource));
+ }
+
+ private DatabaseSessionService(Builder builder) {
+ this.dataSource = builder.dataSource;
+ this.instantSource = builder.instantSource;
+ this.uuidProvider = builder.uuidProvider;
+ this.autoMigrate = builder.autoMigrate;
+ this.statementTimeoutSeconds = builder.statementTimeoutSeconds;
+ this.resolvedDialect = builder.dialect;
+ }
+
+ /** Returns a builder without accessing the supplied data source. */
+ public static Builder builder(DataSource dataSource) {
+ return new Builder(dataSource);
+ }
+
+ /**
+ * Migrates only this module's schema when subscribed. Run once before starting concurrent
+ * workers. Does not run application migrations or take ownership of the data source.
+ *
+ *
The migration creates the five session tables and its own history table, {@code
+ * adk_schema_history}, in the schema the data source connects to. Repeating it on an up-to-date
+ * schema changes nothing. It refuses to touch tables of the same names that it did not create.
+ *
+ *
It needs Flyway on the classpath, together with Flyway's module for the database in use. The
+ * user of the data source needs the right to create tables.
+ *
+ * @return a cold {@code Completable} that fails with {@link SessionException} if the schema could
+ * not be verified or migrated
+ */
+ public Completable migrate() {
+ return Completable.fromAction(this::migrateNow);
+ }
+
+ /** Runs the migration now and remembers that this instance needs no automatic one. */
+ private synchronized void migrateNow() {
+ try {
+ DatabaseSchemaMigrator.migrate(dataSource, resolveDialect());
+ } catch (LinkageError missingMigrationLibrary) {
+ // The migrator is the only class of this module that touches Flyway, so an application
+ // that provisions the schema itself gets this far only if it calls migrate() after all.
+ throw new SessionException(
+ "Unable to load the database migration library; verify the consumer dependencies.");
+ }
+ initialized = true;
+ }
+
+ /**
+ * Runs the opted-in migration before the first database operation of this instance. A failed
+ * attempt is not remembered, so the next operation tries again.
+ */
+ synchronized void initialize() {
+ if (!initialized) {
+ if (autoMigrate) {
+ migrateNow();
+ }
+ initialized = true;
+ }
+ }
+
+ /**
+ * Returns the store, creating it on first use. This is where a lazily detected dialect costs its
+ * one borrowed connection.
+ */
+ private synchronized DatabaseSessionStore store() {
+ initialize();
+ if (store == null) {
+ store = new DatabaseSessionStore(dataSource, resolveDialect(), statementTimeoutSeconds);
+ }
+ return store;
+ }
+
+ /**
+ * {@inheritDoc}
+ *
+ * @deprecated Use {@link #createSession(String, String, Map, String)}, which this delegates to.
+ */
+ @Override
+ @Deprecated
+ public Single createSession(
+ String appName,
+ String userId,
+ @Nullable ConcurrentMap state,
+ @Nullable String sessionId) {
+ return createSession(appName, userId, (Map) state, sessionId);
+ }
+
+ /**
+ * Creates a session and stores its initial state.
+ *
+ * {@code app:} and {@code user:} keys of {@code state} are written to the shared state, where
+ * every other session of the app or user sees them. {@code temp:} keys are returned on the
+ * created session and not stored. A key whose value is {@code null} or {@link State#REMOVED} is
+ * not stored, and removes an {@code app:} or {@code user:} key that was shared before.
+ *
+ * @param appName at most 128 code points; compared exactly, including case and trailing spaces
+ * @param userId at most 128 code points, compared exactly
+ * @param state the initial state, or {@code null} for none; it is not modified
+ * @param sessionId the ID to use, at most 128 code points; {@code null} or blank for a generated
+ * one
+ * @return a cold {@code Single} of the created session, whose state includes what the app and
+ * user already shared
+ * @throws SessionException signaled through the returned {@code Single}: with {@link
+ * SessionException#SESSION_ALREADY_EXISTS} if the ID is taken for this app and user, with
+ * another message if an argument is invalid or the database fails
+ */
+ @Override
+ public Single createSession(
+ String appName,
+ String userId,
+ @Nullable Map state,
+ @Nullable String sessionId) {
+ return Single.fromCallable(
+ () -> {
+ // As in InMemorySessionService, a blank ID asks for a generated one. An ID with any
+ // other content is stored exactly as given, including its surrounding spaces.
+ String id = sessionId == null || sessionId.trim().isEmpty() ? newIdentifier() : sessionId;
+ SessionKey key = key(appName, userId, id);
+ Instant now;
+ try {
+ now = instantSource.instant();
+ } catch (RuntimeException failure) {
+ throw new SessionException("Unable to obtain the session creation time.");
+ }
+ // temp: values go straight from the argument to the returned session. They are kept
+ // by reference and never encoded, so they need not be something JSON can express.
+ Map temporary = new HashMap<>();
+ try {
+ if (state != null)
+ state.forEach(
+ (name, value) -> {
+ if (name.startsWith(State.TEMP_PREFIX)
+ && value != null
+ && value != State.REMOVED) {
+ temporary.put(name, value);
+ }
+ });
+ } catch (RuntimeException invalidState) {
+ throw new SessionException("Unable to snapshot initial session state.");
+ }
+ now = DatabaseTime.forStorage(now);
+ Session created = store().create(key, state == null ? Map.of() : state, now);
+ created.state().putAll(temporary);
+ return created;
+ });
+ }
+
+ /**
+ * Loads a session with its state and history from one consistent snapshot of the database.
+ *
+ * The state merges the session's own keys with the shared {@code app:} and {@code user:} keys
+ * as they are now. The history is in the order the events were appended. With a {@code config},
+ * the last {@code numRecentEvents} appended are taken first, and of those the events at or after
+ * {@code afterTimestamp} are returned; {@code numRecentEvents(0)} returns the session without
+ * history.
+ *
+ *
A session loaded with a filter that leaves out its last event cannot be appended to; see
+ * {@link #appendEvent(Session, Event)}.
+ *
+ * @return a cold {@code Maybe} of a session that shares nothing with other loaded copies; empty
+ * if there is no such session
+ * @throws SessionException signaled through the returned {@code Maybe} if an argument is invalid
+ * or the database fails
+ */
+ @Override
+ public Maybe getSession(
+ String appName, String userId, String sessionId, Optional config) {
+ return Maybe.fromCallable(
+ () -> {
+ SessionKey key = key(appName, userId, sessionId);
+ if (config == null) {
+ throw new SessionException("Session configuration cannot be null.");
+ }
+ return store().get(key, config.orElseGet(() -> GetSessionConfig.builder().build()));
+ });
+ }
+
+ /**
+ * Lists the sessions of one user of one app, with their state and without their events.
+ *
+ * @return a cold {@code Single} of the sessions ordered by last update, oldest first, then by ID;
+ * an empty list if the user has none
+ * @throws SessionException signaled through the returned {@code Single} if an argument is invalid
+ * or the database fails
+ */
+ @Override
+ public Single listSessions(String appName, String userId) {
+ return Single.fromCallable(
+ () -> {
+ validateKey(appName);
+ validateKey(userId);
+ return ListSessionsResponse.builder().sessions(store().list(appName, userId)).build();
+ });
+ }
+
+ /**
+ * Deletes a session and its events. The shared {@code app:} and {@code user:} state is kept,
+ * since other sessions use it.
+ *
+ * @return a cold {@code Completable}
+ * @throws SessionNotFoundException signaled through the returned {@code Completable} if there is
+ * no such session
+ * @throws SessionException signaled the same way if an argument is invalid or the database fails
+ */
+ @Override
+ public Completable deleteSession(String appName, String userId, String sessionId) {
+ return Completable.fromAction(
+ () -> {
+ SessionKey key = key(appName, userId, sessionId);
+ store().delete(key);
+ });
+ }
+
+ /**
+ * Lists every event of a session in the order they were appended. The response carries no page
+ * token: the whole history is returned at once.
+ *
+ * @return a cold {@code Single} of the events
+ * @throws SessionNotFoundException signaled through the returned {@code Single} if there is no
+ * such session
+ * @throws SessionException signaled the same way if an argument is invalid or the database fails
+ */
+ @Override
+ public Single listEvents(String appName, String userId, String sessionId) {
+ return Single.fromCallable(
+ () -> {
+ SessionKey key = key(appName, userId, sessionId);
+ return ListEventsResponse.builder().events(store().listEvents(key)).build();
+ });
+ }
+
+ /**
+ * Persists atomically, then updates the supplied session. The supplied history must contain the
+ * last stored event (or both histories must be empty); reload a stale or filtered snapshot.
+ *
+ * In one transaction the event is added to the history, its state delta is applied to the
+ * session's own state and to the shared {@code app:} and {@code user:} state, and the session's
+ * update time becomes the event's timestamp. Only after the commit is {@code session} changed:
+ * the event is added to it and the delta applied to its state. If the call fails, {@code session}
+ * is exactly as it was.
+ *
+ *
{@link State#TEMP_PREFIX} keys of the event's delta are applied to the supplied session but
+ * never stored, so they are gone when the session is loaded again.
+ *
+ *
A partial event is returned as it is and nothing is stored. An event ID can be used once per
+ * session. Do not change the event or its delta while the call is running.
+ *
+ *
When a failure message says the write may have been committed, load the session again before
+ * deciding whether to repeat the call. When it says nothing was written, the same call can be
+ * repeated with the same session.
+ *
+ * @param session a session of this service, as created or loaded; it is locked for the call
+ * @param event the event to append; its ID is required, at most 128 code points
+ * @return a cold {@code Single} of {@code event}
+ * @throws SessionNotFoundException signaled through the returned {@code Single} if the session no
+ * longer exists
+ * @throws SessionException signaled the same way if the supplied history is not current, the
+ * event ID is already used, an argument is invalid, or the database fails
+ */
+ @CanIgnoreReturnValue
+ @Override
+ public Single appendEvent(Session session, Event event) {
+ return Single.fromCallable(
+ () -> {
+ if (session == null || event == null) {
+ throw new SessionException("Session and event cannot be null.");
+ }
+ if (event.partial().orElse(false)) {
+ return event;
+ }
+ // One append at a time per Session object: the second of two concurrent calls has to
+ // see the event the first one added, or it would be rejected as stale.
+ synchronized (session) {
+ SessionKey key = key(session.appName(), session.userId(), session.id());
+ validateKey(event.id());
+ // The invocation_id column is NOT NULL in the Python v1 layout; an event without one
+ // is stored with an empty string there, and without the field in its JSON.
+ String invocationId = Objects.requireNonNullElse(event.invocationId(), "");
+ if (!validText(invocationId, 256)) {
+ throw new SessionException(
+ "Invocation IDs must be valid Unicode without NUL and at most 256 code points.");
+ }
+ Instant timestamp = DatabaseTime.forStorage(Instant.ofEpochMilli(event.timestamp()));
+ // Copy the delta once. What is stored and what is then applied to the session must be
+ // the same entries, even if the caller's map changes in the meantime.
+ Map delta = snapshotDelta(event);
+ store().append(key, lastEventId(session), event, delta, timestamp, invocationId);
+ // Reached only after a confirmed commit.
+ // temp: values are never stored, but as in InMemorySessionService they reach the
+ // supplied session, where later steps of the same invocation read them.
+ delta.forEach(
+ (name, value) -> {
+ if (value == null || value == State.REMOVED) {
+ session.state().remove(name);
+ } else {
+ session.state().put(name, value);
+ }
+ });
+ session.lastUpdateTime(timestamp);
+ session.addEvent(event);
+ }
+ return event;
+ });
+ }
+
+ /**
+ * Returns the ID of the last event the caller's session holds, or {@code null} if it holds none.
+ * The store compares it with the last stored event to tell a current copy from a stale one; no
+ * extra field on {@link Session} is needed for that.
+ */
+ @SuppressWarnings("deprecation") // Use Session's list monitor without copying the whole history.
+ private static @Nullable String lastEventId(Session session) {
+ List events = session.events();
+ synchronized (events) {
+ if (events.isEmpty()) {
+ return null;
+ }
+ Event tail = events.get(events.size() - 1);
+ if (tail == null || tail.id() == null) {
+ throw new SessionException("Session history is invalid; reload before appending.");
+ }
+ return tail.id();
+ }
+ }
+
+ /** Copies the entries of the event's state delta; empty if the event has none. */
+ private static Map snapshotDelta(Event event) {
+ try {
+ Map delta = event.actions() == null ? null : event.actions().stateDelta();
+ return delta == null ? Map.of() : new HashMap<>(delta);
+ } catch (RuntimeException invalidDelta) {
+ throw new SessionException("Unable to snapshot session event state.");
+ }
+ }
+
+ /** Asks the configured provider for a session ID and holds it to the rules for any other ID. */
+ private String newIdentifier() {
+ try {
+ String value = uuidProvider.newUuid();
+ validateKey(value);
+ return value;
+ } catch (RuntimeException failure) {
+ throw new SessionException("Unable to obtain a valid session identifier.");
+ }
+ }
+
+ /** Validates the three parts of a session's identity and combines them. */
+ private static SessionKey key(String appName, String userId, String id) {
+ validateKey(appName);
+ validateKey(userId);
+ validateKey(id);
+ return new SessionKey(appName, userId, id);
+ }
+
+ /**
+ * Requires an identifier that fits a key column, {@code VARCHAR(128)} on every engine. The
+ * message names the rule and never the value, which may be private.
+ */
+ private static void validateKey(@Nullable String value) {
+ if (!validText(value, 128)) {
+ throw new SessionException(
+ "Identifiers must be valid Unicode without NUL and at most 128 code points.");
+ }
+ }
+
+ /**
+ * Whether every supported engine stores the text and gives the same text back. Checked here so
+ * that all four behave alike, instead of each rejecting, truncating or altering in its own way.
+ *
+ *
+ * - Length is counted in code points, as the engines count {@code VARCHAR} length, not in
+ * Java's UTF-16 units.
+ *
- NUL is refused: PostgreSQL cannot store it in text.
+ *
- A surrogate without its pair is refused: it is not valid Unicode and cannot be encoded as
+ * UTF-8.
+ *
+ */
+ private static boolean validText(@Nullable String value, int maxCodePoints) {
+ if (value == null
+ || value.codePointCount(0, value.length()) > maxCodePoints
+ || value.indexOf('\0') >= 0) {
+ return false;
+ }
+ // A high surrogate must be followed by a low one, and a low one must follow a high one.
+ for (int i = 0; i < value.length(); i++) {
+ char current = value.charAt(i);
+ if (Character.isHighSurrogate(current)) {
+ if (++i >= value.length() || !Character.isLowSurrogate(value.charAt(i))) {
+ return false;
+ }
+ } else if (Character.isLowSurrogate(current)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ /**
+ * Configures immutable service settings; a builder itself is not thread-safe.
+ *
+ * {@code
+ * DatabaseSessionService sessions =
+ * DatabaseSessionService.builder(dataSource)
+ * .dialect(DatabaseSessionService.Dialect.POSTGRESQL)
+ * .statementTimeout(Duration.ofSeconds(30))
+ * .build();
+ * }
+ *
+ * Each setter rejects {@code null} with a {@link SessionException}. A builder can be reused; a
+ * service keeps the settings it was built with.
+ */
+ public static final class Builder {
+ private final DataSource dataSource;
+ private InstantSource instantSource = InstantSource.system();
+ private UuidProvider uuidProvider = UuidProvider.SYSTEM;
+ private boolean autoMigrate;
+ private int statementTimeoutSeconds = DEFAULT_STATEMENT_TIMEOUT_SECONDS;
+ private @Nullable Dialect dialect;
+
+ private Builder(DataSource dataSource) {
+ this.dataSource = requireComponent(dataSource, "dataSource");
+ }
+
+ /**
+ * Selects a dialect explicitly, without inspecting JDBC URLs or metadata.
+ *
+ *
Without this, the dialect is detected on first use from the database product name that the
+ * driver reports, which costs one borrowed connection. Set it when the product name is not one
+ * of the four that are recognized, for example behind a proxy or a compatible engine.
+ */
+ @CanIgnoreReturnValue
+ public Builder dialect(Dialect dialect) {
+ this.dialect = requireComponent(dialect, "dialect");
+ return this;
+ }
+
+ /**
+ * Selects the source used when creating a session; this does not read the clock.
+ *
+ *
It supplies the creation time of a session, which is also its first update time. Later
+ * update times come from the timestamps of the appended events, and the times of the shared
+ * state come from the database's own clock.
+ */
+ @CanIgnoreReturnValue
+ public Builder instantSource(InstantSource instantSource) {
+ this.instantSource = requireComponent(instantSource, "instantSource");
+ return this;
+ }
+
+ /**
+ * Selects a thread-safe provider of distinct identifiers without invoking it.
+ *
+ *
It is asked for a session ID whenever a session is created without one. An ID it returns
+ * has to meet the rules for any session ID, at most 128 code points.
+ */
+ @CanIgnoreReturnValue
+ public Builder uuidProvider(UuidProvider uuidProvider) {
+ this.uuidProvider = requireComponent(uuidProvider, "uuidProvider");
+ return this;
+ }
+
+ /**
+ * Enables schema migration before the first database operation; disabled by default.
+ *
+ *
The builder and constructor remain side-effect free even when this is enabled. Arrange for
+ * one migration owner before starting workers. This flag does not coordinate bootstrap across
+ * services or processes, particularly with SQLite or non-transactional DDL.
+ *
+ *
Leave it off when the schema is created by {@link DatabaseSessionService#migrate()} at
+ * deployment time, or by running the module's SQL yourself.
+ */
+ @CanIgnoreReturnValue
+ public Builder autoMigrate(boolean autoMigrate) {
+ this.autoMigrate = autoMigrate;
+ return this;
+ }
+
+ /**
+ * Limits each SQL statement of an operation, and how long SQLite waits for a competing writer;
+ * 15 seconds by default.
+ *
+ *
This is not an end-to-end deadline: one operation runs several statements, and connection
+ * and network timeouts stay with the application's data source.
+ *
+ * @throws SessionException unless the timeout is a whole number of seconds from 1 to 2147483
+ */
+ @CanIgnoreReturnValue
+ public Builder statementTimeout(Duration statementTimeout) {
+ requireComponent(statementTimeout, "statementTimeout");
+ long seconds = statementTimeout.getSeconds();
+ if (statementTimeout.getNano() != 0
+ || seconds < 1
+ || seconds > MAX_STATEMENT_TIMEOUT_SECONDS) {
+ throw new SessionException(
+ "statementTimeout must be a whole number of seconds from 1 to "
+ + MAX_STATEMENT_TIMEOUT_SECONDS
+ + ".");
+ }
+ this.statementTimeoutSeconds = (int) seconds;
+ return this;
+ }
+
+ /** Takes a snapshot of these settings without accessing the database or providers. */
+ public DatabaseSessionService build() {
+ return new DatabaseSessionService(this);
+ }
+ }
+
+ // The settings this instance was built with, readable by the module's own tests.
+
+ DataSource dataSource() {
+ return dataSource;
+ }
+
+ InstantSource instantSource() {
+ return instantSource;
+ }
+
+ UuidProvider uuidProvider() {
+ return uuidProvider;
+ }
+
+ boolean autoMigrate() {
+ return autoMigrate;
+ }
+
+ Duration statementTimeout() {
+ return Duration.ofSeconds(statementTimeoutSeconds);
+ }
+
+ /** Resolves once per instance. A failed attempt is not cached, allowing a later retry. */
+ synchronized Dialect resolveDialect() {
+ if (resolvedDialect == null) {
+ resolvedDialect = DatabaseDialect.detect(dataSource);
+ }
+ return resolvedDialect;
+ }
+
+ /** Rejects a missing setting with the one exception type this service uses. */
+ private static T requireComponent(@Nullable T value, String name) {
+ if (value == null) {
+ throw new SessionException(name + " cannot be null.");
+ }
+ return value;
+ }
+}
diff --git a/contrib/database-session-service/src/main/java/com/google/adk/sessions/DatabaseSessionStore.java b/contrib/database-session-service/src/main/java/com/google/adk/sessions/DatabaseSessionStore.java
new file mode 100644
index 000000000..44a1d1ce8
--- /dev/null
+++ b/contrib/database-session-service/src/main/java/com/google/adk/sessions/DatabaseSessionStore.java
@@ -0,0 +1,944 @@
+/*
+ * 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
+ *
+ * https://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.sessions;
+
+import com.google.adk.events.Event;
+import com.google.adk.sessions.DatabaseDiagnostics.Phase;
+import com.google.adk.sessions.DatabaseSessionService.Dialect;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import javax.sql.DataSource;
+import org.jspecify.annotations.Nullable;
+
+/**
+ * The SQL of the session service: one short transaction per operation, and the mapping between rows
+ * and {@link Session}s. No pool, executor, or ORM.
+ *
+ * Tables. A session's state is kept in three JSON documents, as in the Python v1 schema:
+ * {@code app_states} (one row per app, the {@code app:} keys without their prefix), {@code
+ * user_states} (one row per app and user, the {@code user:} keys) and {@code sessions.state} (the
+ * unprefixed keys). {@code events} holds the history, one row per event, with the whole event as
+ * JSON in {@code event_data}.
+ *
+ *
Transactions. Every operation borrows one connection, runs inside {@link #transaction},
+ * and returns the connection in the state it was borrowed in. Reads see one snapshot; writes lock
+ * exactly the rows they change.
+ *
+ *
Lock order. Writers take row locks in the order app, user, session, and skip the scopes
+ * they do not change. Keeping to one order is what keeps two writers from deadlocking, so a new
+ * statement that locks a row has to respect it. SQLite has no row locks; there a write transaction
+ * is the only writer of the file ({@code BEGIN IMMEDIATE}).
+ *
+ *
Errors. Callers get a {@link SessionException} whose message this class wrote. A
+ * rejection that the caller can act on (duplicate, missing, stale) travels as {@link Rejected};
+ * anything from the driver is reported by {@link DatabaseDiagnostics} and never passed on.
+ *
+ *
Instances are immutable and safe to share between threads.
+ */
+final class DatabaseSessionStore {
+ /** Primary key of {@code sessions}; bound by {@link #bindScope}. */
+ private static final String SCOPE = "app_name=? AND user_id=? AND id=?";
+
+ /** The same key as it appears in {@code events}, where the session's ID is a foreign key. */
+ private static final String EVENT_SCOPE = "app_name=? AND user_id=? AND session_id=?";
+
+ /**
+ * One session together with the two shared documents it is merged with. The joins are outer so
+ * that a missing shared row shows up as a NULL document, which fails to decode, instead of making
+ * the session itself look absent. A {@code WHERE} clause on {@code s} follows.
+ */
+ private static final String VIEW =
+ "SELECT s.id,s.state AS local_state,s.update_time,"
+ + "a.state AS app_state,u.state AS user_state FROM sessions s "
+ + "LEFT JOIN app_states a ON a.app_name=s.app_name "
+ + "LEFT JOIN user_states u ON u.app_name=s.app_name AND u.user_id=s.user_id ";
+
+ /** Message of a failure that left nothing behind and can be answered by calling again. */
+ static final String CONFLICT =
+ "The database rolled the operation back because of a concurrent transaction;"
+ + " nothing was written, so it can be retried.";
+
+ /**
+ * Both shared documents of one app and user in a single row, for a listing: joining them to every
+ * session, as {@link #VIEW} does for one, would send the same two documents once per session. A
+ * scalar subquery yields NULL for a missing row.
+ */
+ private static final String SHARED =
+ "SELECT (SELECT state FROM app_states WHERE app_name=?),"
+ + "(SELECT state FROM user_states WHERE app_name=? AND user_id=?)";
+
+ private final DataSource source;
+ private final Dialect dialect;
+ private final int timeoutSeconds;
+ private final SessionDataCodec codec = new SessionDataCodec();
+
+ /**
+ * @param source the application's data source; a connection is borrowed per operation and never
+ * kept
+ * @param dialect the engine behind {@code source}
+ * @param timeoutSeconds limit for each statement, and for SQLite's wait on a competing writer
+ */
+ DatabaseSessionStore(DataSource source, Dialect dialect, int timeoutSeconds) {
+ this.source = source;
+ this.dialect = dialect;
+ this.timeoutSeconds = timeoutSeconds;
+ }
+
+ /**
+ * Inserts a session and writes the {@code app:} and {@code user:} keys of its initial state to
+ * the shared documents, creating their rows if this is the first session of the app or user.
+ *
+ * @param state the initial state with prefixed keys; {@code temp:} keys are ignored
+ * @param now creation time, already at storage precision
+ * @return the session as stored, with the shared state merged in
+ * @throws SessionException with {@link SessionException#SESSION_ALREADY_EXISTS} if the key is
+ * taken, in which case the shared documents are left untouched as well
+ */
+ Session create(SessionKey key, Map state, Instant now) {
+ SessionDataCodec.ScopedState split = codec.splitState(state);
+ return transaction(
+ "create",
+ true,
+ connection -> {
+ // A fast rejection is safe here; the insert still enforces uniqueness under races.
+ if (sessionExists(connection, key)) {
+ throw new Rejected(Reason.DUPLICATE);
+ }
+ // App before user, the lock order of this class. Both rows are created here, once, so
+ // that an append can lock a row that is certain to exist instead of racing to insert it.
+ ensureShared(connection, key, false);
+ updateShared(connection, key, false, split.app());
+ ensureShared(connection, key, true);
+ updateShared(connection, key, true, split.user());
+ try (PreparedStatement insert =
+ prepare(
+ connection,
+ "INSERT INTO sessions (app_name,user_id,id,state,create_time,update_time) VALUES (?,?,?,?,?,?)")) {
+ bindScope(insert, key);
+ DatabaseDialect.bindJson(insert, 4, codec.writeState(split.local()), dialect);
+ DatabaseTime.bind(insert, 5, now, dialect);
+ DatabaseTime.bind(insert, 6, now, dialect);
+ insert.executeUpdate();
+ } catch (SQLException collision) {
+ // Two creators of the same key can both pass the check above. The loser fails on the
+ // primary key: unique_violation on PostgreSQL, ER_DUP_ENTRY on MySQL and MariaDB.
+ // SQLite cannot get here, since its write transaction excludes the other creator.
+ if ((dialect == Dialect.POSTGRESQL && "23505".equals(collision.getSQLState()))
+ || ((dialect == Dialect.MYSQL || dialect == Dialect.MARIADB)
+ && collision.getErrorCode() == 1062)) {
+ throw new Rejected(Reason.DUPLICATE);
+ }
+ throw collision;
+ }
+ // Read the row back rather than echo the arguments: the caller gets what the database
+ // holds, including shared keys that other sessions wrote earlier.
+ return load(connection, key);
+ });
+ }
+
+ /**
+ * Reads a session with its merged state and the part of its history that {@code config} asks for,
+ * all from one snapshot.
+ *
+ * @return the session, or {@code null} if there is none under this key
+ */
+ @Nullable Session get(SessionKey key, GetSessionConfig config) {
+ return transaction(
+ "get",
+ false,
+ connection -> {
+ Session session = load(connection, key);
+ if (session != null) {
+ session.addEvents(events(connection, key, config));
+ }
+ return session;
+ });
+ }
+
+ /**
+ * Lists the sessions of one user with their merged state and without their history, the least
+ * recently updated first and by ID within the same update time.
+ *
+ * Two statements whatever the number of sessions: the session rows, then the shared state
+ * once.
+ */
+ List list(String app, String user) {
+ return transaction(
+ "list",
+ false,
+ connection -> {
+ // The rows are held until the shared state is known; a session cannot be built without.
+ record Row(String id, String state, Instant updateTime) {}
+ List rows = new ArrayList<>();
+ try (PreparedStatement query =
+ prepare(
+ connection,
+ "SELECT id,state,update_time FROM sessions WHERE app_name=? AND user_id=?"
+ + " ORDER BY update_time,id")) {
+ query.setString(1, app);
+ query.setString(2, user);
+ try (ResultSet found = query.executeQuery()) {
+ while (found.next()) {
+ rows.add(
+ new Row(
+ found.getString(1),
+ found.getString(2),
+ DatabaseTime.read(found, "update_time", dialect)));
+ }
+ }
+ }
+ List sessions = new ArrayList<>(rows.size());
+ if (rows.isEmpty()) {
+ // A user without sessions may have no shared rows at all; do not go looking for them.
+ return sessions;
+ }
+ // The read transaction gives both statements the same snapshot.
+ String appState;
+ String userState;
+ try (PreparedStatement query = prepare(connection, SHARED)) {
+ query.setString(1, app);
+ query.setString(2, app);
+ query.setString(3, user);
+ try (ResultSet shared = query.executeQuery()) {
+ if (!shared.next()) {
+ throw new SQLException();
+ }
+ appState = shared.getString(1);
+ userState = shared.getString(2);
+ }
+ }
+ for (Row row : rows) {
+ sessions.add(
+ session(row.id(), app, user, row.updateTime(), appState, userState, row.state()));
+ }
+ return sessions;
+ });
+ }
+
+ /**
+ * Deletes a session and, through the foreign key's {@code ON DELETE CASCADE}, its events. The
+ * shared app and user documents stay: other sessions read them.
+ *
+ * @throws SessionNotFoundException if there is no session under this key
+ */
+ void delete(SessionKey key) {
+ transaction(
+ "delete",
+ true,
+ connection -> {
+ try (PreparedStatement delete =
+ prepare(connection, "DELETE FROM sessions WHERE " + SCOPE)) {
+ bindScope(delete, key);
+ if (delete.executeUpdate() == 0) {
+ throw new Rejected(Reason.MISSING);
+ }
+ }
+ return null;
+ });
+ }
+
+ /**
+ * Reads the whole history of a session in append order.
+ *
+ * @throws SessionNotFoundException if there is no session under this key, which an empty history
+ * alone could not tell apart from a session without events
+ */
+ List listEvents(SessionKey key) {
+ return transaction(
+ "listEvents",
+ false,
+ connection -> {
+ if (!sessionExists(connection, key)) {
+ throw new Rejected(Reason.MISSING);
+ }
+ return events(connection, key, GetSessionConfig.builder().build());
+ });
+ }
+
+ /**
+ * Appends one event and applies its state delta, atomically: the event row, the session's {@code
+ * update_time} and state, and the shared documents its delta touches are committed together or
+ * not at all.
+ *
+ * The writer has to have seen the current end of the history. Its last known event is compared
+ * with the stored one while the session row is locked, so of two writers that loaded the same
+ * history the second is rejected rather than appended after an event it never saw.
+ *
+ * @param expectedTailId the ID of the last event in the caller's copy of the session, or {@code
+ * null} if that copy has no events
+ * @param liveDelta the event's state delta as it was when the call started, with prefixed keys
+ * @param timestamp the event's time at storage precision; it becomes the session's {@code
+ * update_time}
+ * @param invocationId the value of the {@code invocation_id} column, empty if the event has none
+ * @throws SessionNotFoundException if the session does not exist
+ * @throws SessionException if the caller's history is stale, or the session already holds an
+ * event with this ID
+ */
+ void append(
+ SessionKey key,
+ @Nullable String expectedTailId,
+ Event event,
+ Map liveDelta,
+ Instant timestamp,
+ String invocationId) {
+ // Encode before borrowing a connection: an event that cannot be written should not cost a
+ // transaction, and no lock is held while JSON is produced.
+ String json = codec.writeEvent(event);
+ SessionDataCodec.ScopedState delta = codec.splitState(liveDelta);
+ transaction(
+ "append",
+ true,
+ connection -> {
+ // Global lock order is app -> user -> session, matching create. Unchanged scopes are not
+ // locked, ensured, or updated. In particular a local event never locks the app-wide row.
+ updateShared(connection, key, false, delta.app());
+ updateShared(connection, key, true, delta.user());
+ // Most events carry no local delta. Lock the session either way, but leave its state
+ // document unread and unwritten then: rewriting it would cost a full decode, encode and
+ // row/TOAST write per event, growing with the size of the session state.
+ boolean localChanged = !delta.local().isEmpty();
+ Map local = Map.of();
+ try (PreparedStatement query =
+ prepare(
+ connection,
+ "SELECT "
+ + (localChanged ? "state" : "1")
+ + " FROM sessions WHERE "
+ + SCOPE
+ + DatabaseDialect.lock(dialect))) {
+ bindScope(query, key);
+ try (ResultSet row = query.executeQuery()) {
+ if (!row.next()) {
+ throw new Rejected(Reason.MISSING);
+ }
+ if (localChanged) {
+ local = codec.readState(row.getString(1));
+ }
+ }
+ }
+ // The locked session (or SQLite's write transaction) serializes tail reads and inserts.
+ long next = nextSequence(connection, key, expectedTailId);
+ try (PreparedStatement update =
+ prepare(
+ connection,
+ "UPDATE sessions SET "
+ + (localChanged ? "state=?," : "")
+ + "update_time=? WHERE "
+ + SCOPE)) {
+ int index = 1;
+ if (localChanged) {
+ DatabaseDialect.bindJson(
+ update,
+ index++,
+ codec.writeState(codec.applyState(local, delta.local())),
+ dialect);
+ }
+ DatabaseTime.bind(update, index++, timestamp, dialect);
+ bindScope(update, index, key);
+ // The session exists and is locked; a no-op update can report zero changed rows.
+ update.executeUpdate();
+ }
+ // An event ID may be used once per session (the primary key of events). PostgreSQL and
+ // SQLite are asked to skip a duplicate, which shows as zero inserted rows; on PostgreSQL
+ // a failed statement would otherwise abort the whole transaction. MySQL and MariaDB have
+ // no ON CONFLICT clause and report ER_DUP_ENTRY, which is caught below.
+ boolean mysqlFamily = dialect == Dialect.MYSQL || dialect == Dialect.MARIADB;
+ String conflict =
+ mysqlFamily ? "" : " ON CONFLICT (id,app_name,user_id,session_id) DO NOTHING";
+ try (PreparedStatement insert =
+ prepare(
+ connection,
+ "INSERT INTO events (app_name,user_id,session_id,event_sequence,id,invocation_id,timestamp,event_data) VALUES (?,?,?,?,?,?,?,?)"
+ + conflict)) {
+ bindScope(insert, key);
+ insert.setLong(4, next);
+ insert.setString(5, event.id());
+ insert.setString(6, invocationId);
+ DatabaseTime.bind(insert, 7, timestamp, dialect);
+ DatabaseDialect.bindJson(insert, 8, json, dialect);
+ if (insert.executeUpdate() == 0) {
+ throw new Rejected(Reason.DUPLICATE_EVENT);
+ }
+ } catch (SQLException collision) {
+ if (mysqlFamily && collision.getErrorCode() == 1062) {
+ throw new Rejected(Reason.DUPLICATE_EVENT);
+ }
+ throw collision;
+ }
+ return null;
+ });
+ }
+
+ /** Reads one session with its merged state and no events; {@code null} if it does not exist. */
+ private @Nullable Session load(Connection connection, SessionKey key) throws SQLException {
+ try (PreparedStatement query =
+ prepare(connection, VIEW + "WHERE s.app_name=? AND s.user_id=? AND s.id=?")) {
+ bindScope(query, key);
+ try (ResultSet rows = query.executeQuery()) {
+ return rows.next()
+ ? session(
+ rows.getString("id"),
+ key.appName(),
+ key.userId(),
+ DatabaseTime.read(rows, "update_time", dialect),
+ rows.getString("app_state"),
+ rows.getString("user_state"),
+ rows.getString("local_state"))
+ : null;
+ }
+ }
+ }
+
+ /**
+ * Builds a session from its three stored documents. They are decoded for this session alone, so
+ * that no two sessions share a value, and a NULL document (a missing shared row) fails to decode
+ * instead of being read as empty state.
+ */
+ private Session session(
+ String id,
+ String app,
+ String user,
+ Instant updateTime,
+ @Nullable String appState,
+ @Nullable String userState,
+ @Nullable String localState) {
+ return Session.builder(id)
+ .appName(app)
+ .userId(user)
+ .lastUpdateTime(updateTime)
+ .state(
+ codec.mergeState(
+ codec.readState(appState), codec.readState(userState), codec.readState(localState)))
+ .build();
+ }
+
+ /**
+ * Reads a session's events in append order, filtered the way {@code InMemorySessionService}
+ * filters them: first keep the last {@code numRecentEvents} appended, then drop those older than
+ * {@code afterTimestamp}. An event at exactly that time is kept.
+ *
+ * The statement has one of two shapes, each with or without the time predicate in brackets:
+ *
+ *
{@code
+ * SELECT event_data FROM events WHERE scope [AND timestamp>=?] ORDER BY event_sequence
+ *
+ * SELECT event_data FROM (
+ * SELECT ... FROM events WHERE scope ORDER BY event_sequence DESC LIMIT ?) recent
+ * [WHERE timestamp>=?] ORDER BY event_sequence
+ * }
+ *
+ * Order and limit use {@code event_sequence}, never {@code timestamp}: event times have
+ * millisecond resolution, tie often, and may run backwards.
+ */
+ private List events(Connection connection, SessionKey key, GetSessionConfig config)
+ throws SQLException {
+ List events = new ArrayList<>();
+ if (config.numRecentEvents().orElse(-1) == 0) {
+ // Asked for the session without history; there is nothing to query.
+ return events;
+ }
+ Instant lowerBound = null;
+ if (config.afterTimestamp().isPresent()) {
+ Instant cursor = config.afterTimestamp().orElseThrow();
+ // A cursor outside the range a column can hold cannot be bound. Past the range no stored
+ // event can match; before it every stored event does, so no predicate is needed.
+ if (cursor.isAfter(DatabaseTime.MAX)) {
+ return events;
+ }
+ if (cursor.isAfter(DatabaseTime.MIN)) {
+ // Columns hold microseconds. Rounding the cursor up keeps ">=" exact for a cursor with
+ // nanoseconds: an event in the same microsecond but before the cursor is left out.
+ lowerBound = DatabaseTime.ceilToMicros(cursor);
+ }
+ }
+ boolean limited = config.numRecentEvents().isPresent();
+ // The subquery limits append positions BEFORE timestamp filtering; pushing the predicate into
+ // that subquery changes the Java contract for out-of-order timestamps.
+ String sql =
+ limited
+ ? "SELECT event_data FROM (SELECT event_sequence,timestamp,event_data FROM events WHERE "
+ + EVENT_SCOPE
+ + " ORDER BY event_sequence DESC LIMIT ?) recent"
+ : "SELECT event_data FROM events WHERE " + EVENT_SCOPE;
+ if (lowerBound != null) {
+ sql += (limited ? " WHERE " : " AND ") + "timestamp>=?";
+ }
+ sql += " ORDER BY event_sequence";
+ try (PreparedStatement query = prepare(connection, sql)) {
+ // Parameters in the order they appear in either shape: the scope, the limit, the cursor.
+ bindScope(query, key);
+ int index = 4;
+ if (limited) {
+ query.setInt(index++, config.numRecentEvents().orElseThrow());
+ }
+ if (lowerBound != null) {
+ DatabaseTime.bind(query, index, lowerBound, dialect);
+ }
+ try (ResultSet rows = query.executeQuery()) {
+ while (rows.next()) {
+ events.add(codec.readEvent(rows.getString(1)));
+ }
+ }
+ }
+ return events;
+ }
+
+ /** Whether a session row exists, without locking it. */
+ private boolean sessionExists(Connection connection, SessionKey key) throws SQLException {
+ try (PreparedStatement query = prepare(connection, "SELECT 1 FROM sessions WHERE " + SCOPE)) {
+ bindScope(query, key);
+ try (ResultSet rows = query.executeQuery()) {
+ return rows.next();
+ }
+ }
+ }
+
+ /**
+ * Checks that the caller's last event is the stored last event, and returns the position for the
+ * next one. Positions start at 1 and are dense within a session.
+ *
+ * Call it only while the session row is locked (on SQLite, inside the write transaction): that
+ * is what keeps a second writer from reading the same tail and taking the same position. The
+ * unique constraint on the position is a second line of defense, not the mechanism.
+ *
+ * @param expectedTailId the caller's last event ID, or {@code null} for an empty history
+ */
+ private long nextSequence(Connection connection, SessionKey key, @Nullable String expectedTailId)
+ throws SQLException {
+ try (PreparedStatement query =
+ prepare(
+ connection,
+ "SELECT id,event_sequence FROM events WHERE "
+ + EVENT_SCOPE
+ + " ORDER BY event_sequence DESC LIMIT 1")) {
+ bindScope(query, key);
+ try (ResultSet rows = query.executeQuery()) {
+ boolean hasTail = rows.next();
+ String actual = hasTail ? rows.getString("id") : null;
+ // Both empty, or the same last event. Anything else means the caller's copy is behind
+ // the stored history, or was loaded with a filter that cut its end off.
+ if (!Objects.equals(actual, expectedTailId)) {
+ throw new Rejected(Reason.STALE);
+ }
+ long sequence = hasTail ? rows.getLong("event_sequence") : 0;
+ if (sequence == Long.MAX_VALUE) {
+ throw new Rejected(Reason.SEQUENCE_EXHAUSTED);
+ }
+ return sequence + 1;
+ }
+ }
+ }
+
+ /**
+ * Makes sure the app-wide or per-user state row exists, inserting an empty document if it does
+ * not. Two sessions created at the same moment may both try; the insert is written so that the
+ * second one waits for the first and then does nothing.
+ *
+ * @param user {@code true} for the user's row in {@code user_states}, {@code false} for the app's
+ * row in {@code app_states}
+ */
+ private void ensureShared(Connection connection, SessionKey key, boolean user)
+ throws SQLException {
+ String table = user ? "user_states" : "app_states";
+ if (dialect == Dialect.MYSQL || dialect == Dialect.MARIADB) {
+ // InnoDB takes an exclusive lock even for a no-op duplicate-key update. Existing shared
+ // rows are never deleted by the service, so skip the upsert on a successful consistent read.
+ String scope = "app_name=?" + (user ? " AND user_id=?" : "");
+ try (PreparedStatement query =
+ prepare(connection, "SELECT 1 FROM " + table + " WHERE " + scope)) {
+ bindShared(query, 1, key, user);
+ try (ResultSet rows = query.executeQuery()) {
+ if (rows.next()) {
+ return;
+ }
+ }
+ }
+ }
+ try (PreparedStatement insert =
+ prepare(connection, DatabaseDialect.insertShared(table, user, dialect))) {
+ insert.setString(1, key.appName());
+ int stateIndex = 2;
+ if (user) {
+ insert.setString(2, key.userId());
+ stateIndex++;
+ }
+ DatabaseDialect.bindJson(insert, stateIndex, "{}", dialect);
+ insert.executeUpdate();
+ }
+ }
+
+ /**
+ * Applies a delta to the app-wide or per-user document: lock the row, read the document, change
+ * the keys of the delta, write the document back. The lock is what keeps two sessions that change
+ * different keys of the same document from overwriting each other.
+ *
+ *
An empty delta does nothing at all. That is the common case, and the reason an ordinary
+ * event never waits on the one row that every session of an app shares.
+ *
+ * @param user {@code true} for {@code user_states}, {@code false} for {@code app_states}
+ * @param delta keys without their scope prefix; a {@link State#REMOVED} value deletes the key
+ */
+ private void updateShared(
+ Connection connection, SessionKey key, boolean user, Map delta)
+ throws SQLException {
+ if (delta.isEmpty()) {
+ return;
+ }
+ String table = user ? "user_states" : "app_states";
+ String scope = "app_name=?" + (user ? " AND user_id=?" : "");
+ Map current;
+ try (PreparedStatement query =
+ prepare(
+ connection,
+ "SELECT state FROM " + table + " WHERE " + scope + DatabaseDialect.lock(dialect))) {
+ bindShared(query, 1, key, user);
+ try (ResultSet rows = query.executeQuery()) {
+ if (!rows.next()) {
+ // createSession made this row and nothing in the service deletes it, so its absence
+ // means the tables were changed from outside. Fail rather than quietly recreate it.
+ throw new SQLException();
+ }
+ current = codec.readState(rows.getString(1));
+ }
+ }
+ // update_time is the database's own UTC clock, not the event's time: it records when the
+ // shared document last changed, whichever session and whichever application host wrote it.
+ try (PreparedStatement update =
+ prepare(
+ connection,
+ "UPDATE "
+ + table
+ + " SET state=?,update_time="
+ + DatabaseDialect.utcNow(dialect)
+ + " WHERE "
+ + scope)) {
+ DatabaseDialect.bindJson(
+ update, 1, codec.writeState(codec.applyState(current, delta)), dialect);
+ bindShared(update, 2, key, user);
+ update.executeUpdate();
+ }
+ }
+
+ /**
+ * Binds the key of a shared row from {@code index} on: the app, then the user if there is one.
+ */
+ private void bindShared(PreparedStatement statement, int index, SessionKey key, boolean user)
+ throws SQLException {
+ statement.setString(index, key.appName());
+ if (user) {
+ statement.setString(index + 1, key.userId());
+ }
+ }
+
+ /** Binds the three parts of a session key as the first three parameters. */
+ private void bindScope(PreparedStatement statement, SessionKey key) throws SQLException {
+ bindScope(statement, 1, key);
+ }
+
+ /** Binds app, user and session ID, in the order of {@link #SCOPE}, from {@code index} on. */
+ private void bindScope(PreparedStatement statement, int index, SessionKey key)
+ throws SQLException {
+ statement.setString(index, key.appName());
+ statement.setString(index + 1, key.userId());
+ statement.setString(index + 2, key.id());
+ }
+
+ /**
+ * Prepares a statement with the configured timeout. Every statement of this class goes through
+ * here; SQL text is built only from constants of this module, and all values are bound.
+ */
+ private PreparedStatement prepare(Connection connection, String sql) throws SQLException {
+ PreparedStatement statement = connection.prepareStatement(sql);
+ try {
+ statement.setQueryTimeout(timeoutSeconds);
+ return statement;
+ } catch (SQLException | RuntimeException failure) {
+ // The caller's try-with-resources never received the statement, so close it here.
+ statement.close();
+ throw failure;
+ }
+ }
+
+ /**
+ * Runs {@code work} in one transaction on a borrowed connection and maps every outcome to what a
+ * caller may see.
+ *
+ * In order: borrow; require a connection that is not inside someone else's transaction; set
+ * the isolation level and begin; check that the schema is the one this version writes; run the
+ * work; commit; put the connection's settings back; close.
+ *
+ *
+ * - Isolation. A read uses REPEATABLE READ so that its statements see one snapshot. A
+ * write uses READ COMMITTED and explicit row locks. With a snapshot taken at its start, a
+ * writer that had to wait for a row lock would afterwards still read the history as it was
+ * before the wait (MySQL, MariaDB) or be refused outright (PostgreSQL). SQLite gets its
+ * consistency from being a single writer.
+ *
- SQLite. The transaction is opened with an explicit {@code BEGIN}, immediate for a
+ * write so that the file's write lock is taken before anything is read. Foreign keys are
+ * switched on, because the cascade from a session to its events depends on a per-connection
+ * setting that is off by default.
+ *
- Schema check. The metadata row is read inside the transaction, every time, so that
+ * a process still running after the schema has been migrated to a newer version stops
+ * instead of writing the old format into it.
+ *
- Commit. Whether the commit was not attempted, sent, or acknowledged decides what
+ * the caller is told: nothing written, possibly written, or written.
+ *
- Cleanup. Auto-commit, isolation and the SQLite settings are restored to what they
+ * were at borrow time, so the pool gets back the connection it handed out. If that fails
+ * the connection is aborted rather than returned in an unknown state.
+ *
+ *
+ * @param operation name for diagnostics, such as {@code "append"}
+ * @param write whether the work changes data; decides the isolation level and the lock taken on
+ * SQLite, and whether a failed commit is ambiguous
+ * @throws SessionNotFoundException if the work rejected a missing session
+ * @throws SessionException for every other failure, with a message written by this class
+ */
+ private T transaction(String operation, boolean write, SqlWork work) {
+ Phase phase = Phase.CONNECTION;
+ CommitOutcome commitOutcome = CommitOutcome.NOT_ATTEMPTED;
+ // phase and commitOutcome are read by the catch blocks at the end: they say how far the
+ // transaction got when it failed, which is what the message to the caller depends on.
+ try (Connection connection = source.getConnection()) {
+ phase = Phase.SETUP;
+ DatabaseDiagnostics.requireConnection(connection, write);
+ int isolation = connection.getTransactionIsolation();
+ int required =
+ dialect == Dialect.SQLITE
+ ? Connection.TRANSACTION_SERIALIZABLE
+ : write
+ ? Connection.TRANSACTION_READ_COMMITTED
+ : Connection.TRANSACTION_REPEATABLE_READ;
+ // A server connection already at the required level needs neither the change nor the
+ // restore, each a round trip of its own. SQLite's level is driver-side state that is cheap
+ // to assert, so it is always set.
+ boolean changeIsolation = dialect == Dialect.SQLITE || isolation != required;
+ // The SQLite settings as borrowed, to restore afterwards; -1 until they have been read.
+ int foreignKeys = -1;
+ int busyTimeout = -1;
+ boolean begun = false;
+ @Nullable Throwable operationFailure = null;
+ try {
+ if (changeIsolation) {
+ connection.setTransactionIsolation(required);
+ }
+ if (dialect == Dialect.SQLITE) {
+ foreignKeys = pragma(connection, "foreign_keys");
+ busyTimeout = pragma(connection, "busy_timeout");
+ control(connection, "PRAGMA foreign_keys=ON");
+ control(connection, "PRAGMA busy_timeout=" + timeoutSeconds * 1000);
+ // SQLite accepts this PRAGMA silently when it cannot honor it, for instance inside an
+ // open transaction or in a build without foreign keys. Deleting a session would then
+ // leave its events behind, so read the setting back.
+ if (pragma(connection, "foreign_keys") != 1) {
+ throw new SQLException();
+ }
+ // Do not combine manual BEGIN with setAutoCommit(false): Xerial would issue nested BEGIN.
+ control(connection, write ? "BEGIN IMMEDIATE" : "BEGIN");
+ } else {
+ connection.setAutoCommit(false);
+ }
+ begun = true;
+ phase = Phase.SCHEMA;
+ try (PreparedStatement query = prepare(connection, DatabaseDialect.metadataQuery(dialect));
+ ResultSet metadata = query.executeQuery()) {
+ DatabaseSchemaMigrator.requireMetadata(metadata);
+ }
+ phase = Phase.STATEMENT;
+ T result = work.run(connection);
+ phase = Phase.COMMIT;
+ // UNCONFIRMED is set before the commit is sent and CONFIRMED only once it returns. A
+ // failure in between leaves UNCONFIRMED: the server may have committed and the reply
+ // been lost, so the caller must not be told that nothing was written.
+ if (dialect == Dialect.SQLITE) {
+ try (Statement commit = connection.createStatement()) {
+ commitOutcome = CommitOutcome.UNCONFIRMED;
+ commit.execute("COMMIT");
+ // Closing the statement is cleanup, not part of the commit acknowledgement.
+ commitOutcome = CommitOutcome.CONFIRMED;
+ phase = Phase.CLEANUP;
+ }
+ } else {
+ commitOutcome = CommitOutcome.UNCONFIRMED;
+ connection.commit();
+ commitOutcome = CommitOutcome.CONFIRMED;
+ phase = Phase.CLEANUP;
+ }
+ return result;
+ } catch (SQLException | RuntimeException failure) {
+ operationFailure = failure;
+ throw failure;
+ } finally {
+ // Runs on success and on failure alike: undo what was not committed, then give the
+ // connection back the settings it was borrowed with.
+ try {
+ if (begun && commitOutcome != CommitOutcome.CONFIRMED) {
+ if (dialect == Dialect.SQLITE) {
+ control(connection, "ROLLBACK");
+ } else {
+ connection.rollback();
+ }
+ }
+ if (dialect == Dialect.SQLITE) {
+ if (foreignKeys >= 0) {
+ control(connection, "PRAGMA foreign_keys=" + foreignKeys);
+ }
+ if (busyTimeout >= 0) {
+ control(connection, "PRAGMA busy_timeout=" + busyTimeout);
+ }
+ } else {
+ connection.setAutoCommit(true);
+ }
+ if (changeIsolation) {
+ connection.setTransactionIsolation(isolation);
+ }
+ } catch (SQLException | RuntimeException cleanupFailure) {
+ // Never enable auto-commit after a failed rollback: that could commit a partial write.
+ try {
+ connection.abort(Runnable::run);
+ } catch (SQLException | RuntimeException ignored) {
+ // Closing the borrowed connection is still attempted by try-with-resources.
+ }
+ // With no failure so far, or only a rejection the caller could have acted on, the
+ // cleanup failure is the one to report: the connection is broken and the caller should
+ // know. Otherwise the earlier failure stays the reported one and this is only logged.
+ if (operationFailure == null
+ || operationFailure instanceof Rejected
+ || operationFailure instanceof DatabaseDiagnostics.Configuration) {
+ phase = Phase.CLEANUP;
+ throw cleanupFailure;
+ }
+ // Report cleanup separately without masking the original statement/commit failure.
+ DatabaseDiagnostics.report(operation, Phase.CLEANUP, dialect, cleanupFailure);
+ }
+ }
+ } catch (Rejected rejection) {
+ // Rejections are decisions of this class, not database failures: nothing is logged, and
+ // the messages are fixed texts without any of the caller's data.
+ throw switch (rejection.reason) {
+ case DUPLICATE -> new SessionException(SessionException.SESSION_ALREADY_EXISTS);
+ case DUPLICATE_EVENT ->
+ new SessionException("An event with this ID already exists in the session.");
+ case MISSING -> new SessionNotFoundException("Session not found.");
+ case STALE ->
+ new SessionException(
+ "Session history is stale or filtered; reload the session before appending.");
+ case SEQUENCE_EXHAUSTED -> new SessionException("Session event sequence is exhausted.");
+ };
+ } catch (DatabaseDiagnostics.Configuration configuration) {
+ throw new SessionException(configuration.getMessage());
+ } catch (SessionDataCodec.StorageFailure failure) {
+ // A resource close may have attached an unsafe suppressed driver exception.
+ throw DatabaseDiagnostics.failure(operation, phase, dialect, failure.getMessage(), failure);
+ } catch (SQLException | RuntimeException failure) {
+ // What the caller needs to know first is whether the write happened. Only when it certainly
+ // did not is the phase that failed worth naming.
+ String message;
+ if (write && commitOutcome == CommitOutcome.CONFIRMED) {
+ message = "The database write committed, but cleanup failed; reload the session.";
+ } else if (write && commitOutcome == CommitOutcome.UNCONFIRMED) {
+ message = "The database write may have committed; reload the session before retrying.";
+ } else if (commitOutcome == CommitOutcome.NOT_ATTEMPTED
+ && DatabaseDiagnostics.concurrencyConflict(failure, dialect)) {
+ // Nothing was committed and the supplied session is unchanged, so the same call can be
+ // repeated as it is.
+ message = CONFLICT;
+ } else {
+ message =
+ switch (phase) {
+ case CONNECTION -> "Unable to borrow a database connection.";
+ case SETUP -> "Unable to configure the database session transaction.";
+ case SCHEMA ->
+ "Unable to verify the ADK schema; check the database and run migrations if needed.";
+ case CLEANUP -> "Unable to restore the borrowed database connection.";
+ default -> "Unable to complete the database session operation.";
+ };
+ }
+ throw DatabaseDiagnostics.failure(operation, phase, dialect, message, failure);
+ }
+ }
+
+ /** Reads the current integer value of a SQLite per-connection setting. */
+ private static int pragma(Connection connection, String name) throws SQLException {
+ try (Statement query = connection.createStatement();
+ ResultSet rows = query.executeQuery("PRAGMA " + name)) {
+ if (!rows.next()) {
+ throw new SQLException();
+ }
+ return rows.getInt(1);
+ }
+ }
+
+ /**
+ * Executes a SQLite transaction or settings command. These take no parameters, and their text
+ * comes only from this class and from integers it read from the connection.
+ */
+ private static void control(Connection connection, String sql) throws SQLException {
+ try (Statement statement = connection.createStatement()) {
+ statement.execute(sql);
+ }
+ }
+
+ /** How far a commit got, which is what a caller of a failed write most needs to know. */
+ private enum CommitOutcome {
+ /** The transaction failed, or is still running, before any commit was sent. */
+ NOT_ATTEMPTED,
+ /** The commit was sent and no answer has been seen: it may or may not have been applied. */
+ UNCONFIRMED,
+ /** The database acknowledged the commit; a later failure can only concern cleanup. */
+ CONFIRMED
+ }
+
+ /** Why an operation was refused although the database itself worked. */
+ private enum Reason {
+ /** A session with this key exists. */
+ DUPLICATE,
+ /** The session already has an event with this ID. */
+ DUPLICATE_EVENT,
+ /** There is no session with this key. */
+ MISSING,
+ /** The caller's last event is not the stored last event. */
+ STALE,
+ /** The session has used every position a {@code BIGINT} can number. */
+ SEQUENCE_EXHAUSTED
+ }
+
+ /**
+ * Thrown inside {@link SqlWork} to abandon the transaction for a {@link Reason}. It is caught in
+ * {@link #transaction}, after the rollback, and turned into the exception the caller sees; it
+ * never leaves this class.
+ */
+ private static final class Rejected extends RuntimeException {
+ final Reason reason;
+
+ Rejected(Reason reason) {
+ this.reason = reason;
+ }
+ }
+
+ /** The statements of one operation, run by {@link #transaction} between begin and commit. */
+ @FunctionalInterface
+ private interface SqlWork {
+ T run(Connection connection) throws SQLException;
+ }
+}
diff --git a/contrib/database-session-service/src/main/java/com/google/adk/sessions/DatabaseTime.java b/contrib/database-session-service/src/main/java/com/google/adk/sessions/DatabaseTime.java
new file mode 100644
index 000000000..87b72a344
--- /dev/null
+++ b/contrib/database-session-service/src/main/java/com/google/adk/sessions/DatabaseTime.java
@@ -0,0 +1,105 @@
+/*
+ * 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.sessions;
+
+import com.google.adk.sessions.DatabaseSessionService.Dialect;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
+import java.time.format.ResolverStyle;
+import java.time.temporal.ChronoUnit;
+import java.util.Locale;
+import org.jspecify.annotations.Nullable;
+
+/**
+ * UTC wall times in the common SQL range, without the JVM or connection time zone.
+ *
+ * The time columns have no time zone, as in the Python v1 schema, and hold UTC by convention. An
+ * {@link Instant} is therefore written as the date and time it has in UTC, through {@link
+ * LocalDateTime}. A {@code java.sql.Timestamp} would not do: drivers convert it using the time zone
+ * of the JVM or of the connection, and the stored value would then depend on where the application
+ * runs.
+ *
+ *
Precision is microseconds, the finest that all four engines keep.
+ */
+final class DatabaseTime {
+ /** Earliest time every engine stores; MySQL's {@code DATETIME} starts at the year 1000. */
+ static final Instant MIN = Instant.parse("1000-01-01T00:00:00Z");
+
+ /** Latest time every engine stores, at the last microsecond of the year 9999. */
+ static final Instant MAX = Instant.parse("9999-12-31T23:59:59.999999Z");
+
+ /**
+ * SQLite has no time type, so times are text. All six fraction digits are always written: text of
+ * one fixed width sorts and compares in the order of the times it stands for, which the {@code
+ * ORDER BY} and {@code >=} of the store rely on. It is also the form in which SQLAlchemy, and
+ * with it the Python service, writes its SQLite times.
+ */
+ private static final DateTimeFormatter SQLITE_FORMAT =
+ DateTimeFormatter.ofPattern("uuuu-MM-dd HH:mm:ss.SSSSSS", Locale.ROOT)
+ .withResolverStyle(ResolverStyle.STRICT);
+
+ private DatabaseTime() {}
+
+ /**
+ * Returns the time as it will be stored, cut to microseconds, so that a caller holds the same
+ * value that a later read returns.
+ *
+ * @throws SessionException if the time is {@code null} or outside {@link #MIN} to {@link #MAX}
+ */
+ static Instant forStorage(@Nullable Instant time) {
+ if (time == null || time.isBefore(MIN) || time.truncatedTo(ChronoUnit.MICROS).isAfter(MAX)) {
+ throw new SessionException("Database session times must be in UTC years 1000 through 9999.");
+ }
+ return time.truncatedTo(ChronoUnit.MICROS);
+ }
+
+ /**
+ * Rounds a time up to the next microsecond unless it is already on one.
+ *
+ *
For a lower bound of a query. Stored times are whole microseconds, so "at or after the
+ * cursor" is the same set of rows as "at or after the cursor rounded up", while rounding down
+ * would let in a row from just before the cursor.
+ */
+ static Instant ceilToMicros(Instant cursor) {
+ Instant floor = cursor.truncatedTo(ChronoUnit.MICROS);
+ return floor.equals(cursor) ? floor : floor.plus(1, ChronoUnit.MICROS);
+ }
+
+ /** Binds a time as its UTC date and time: as text for SQLite, as a zone-free value otherwise. */
+ static void bind(PreparedStatement statement, int index, Instant time, Dialect dialect)
+ throws SQLException {
+ LocalDateTime utc = LocalDateTime.ofInstant(time, ZoneOffset.UTC);
+ if (dialect == Dialect.SQLITE) {
+ statement.setString(index, SQLITE_FORMAT.format(utc));
+ } else {
+ statement.setObject(index, utc);
+ }
+ }
+
+ /** Reads a time column, taking the stored date and time to be UTC. */
+ static Instant read(ResultSet rows, String column, Dialect dialect) throws SQLException {
+ LocalDateTime utc =
+ dialect == Dialect.SQLITE
+ ? LocalDateTime.parse(rows.getString(column), SQLITE_FORMAT)
+ : rows.getObject(column, LocalDateTime.class);
+ return utc.toInstant(ZoneOffset.UTC);
+ }
+}
diff --git a/contrib/database-session-service/src/main/java/com/google/adk/sessions/SessionDataCodec.java b/contrib/database-session-service/src/main/java/com/google/adk/sessions/SessionDataCodec.java
new file mode 100644
index 000000000..7b17a76de
--- /dev/null
+++ b/contrib/database-session-service/src/main/java/com/google/adk/sessions/SessionDataCodec.java
@@ -0,0 +1,419 @@
+/*
+ * 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
+ *
+ * https://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.sessions;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.JsonStreamContext;
+import com.fasterxml.jackson.core.JsonToken;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.ObjectReader;
+import com.fasterxml.jackson.databind.ObjectWriter;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.adk.JsonBaseModel;
+import com.google.adk.events.Event;
+import com.google.adk.events.EventActions;
+import java.io.IOException;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import org.jspecify.annotations.Nullable;
+
+/**
+ * JSON boundary and scope rules shared by all JDBC dialects. Never mutates caller-owned data.
+ *
+ *
Two things are stored as JSON text: state documents and events.
+ *
+ *
+ * - A state document is a JSON object for one scope: an app, a user, or a session. Its
+ * keys carry no scope prefix; the table it is in says which scope it is. A key is either
+ * present with a value or absent, so a top-level {@code null} is never written and, if found,
+ * is read as absent. Inside a value, {@code null} is ordinary data.
+ *
- An event is the whole Java {@link Event}, written with ADK's own mapper so that it
+ * reads back equal. One part is written differently, its state delta: {@code temp:} keys are
+ * left out, and a removed key is written as {@code null}.
+ *
+ *
+ * The methods come in the order data flows: a caller's map is split into scoped snapshots,
+ * snapshots are applied to stored documents and written, stored documents are read and merged into
+ * the map a caller sees.
+ *
+ *
Failures carry a fixed message and no cause: a Jackson message can quote the value it failed
+ * on. Instances are immutable and safe to share between threads.
+ */
+final class SessionDataCodec {
+ /**
+ * Name of the per-call attribute that holds the event being written. The delta serializer
+ * compares it with the object it is nested in, to act on that event's own delta only.
+ */
+ private static final String EVENT_ROOT = "com.google.adk.sessions.persistedEvent";
+
+ /** A JSON object as a map that keeps the order of its keys. */
+ private static final TypeReference> STATE_TYPE =
+ new TypeReference<>() {};
+
+ // Configure a private copy once: changing ADK's shared mapper would affect unrelated services.
+ private final ObjectMapper mapper = JsonBaseModel.getMapper().copy();
+ // Event-only rules must not change how arbitrary model objects inside state are encoded.
+ private final ObjectWriter eventWriter =
+ mapper.copy().addMixIn(EventActions.class, PersistedActions.class).writer();
+ // Text after the document, such as a second object, means the column does not hold what was
+ // written; fail instead of reading only its beginning.
+ private final ObjectReader treeReader =
+ mapper.readerFor(JsonNode.class).with(DeserializationFeature.FAIL_ON_TRAILING_TOKENS);
+ private final ObjectReader eventReader =
+ mapper.readerFor(Event.class).with(DeserializationFeature.FAIL_ON_TRAILING_TOKENS);
+
+ /**
+ * Scoped delta snapshots. Only the outermost prefix is a scope; suffixes are literal keys.
+ *
+ * So {@code app:temp:x} is the app's key {@code temp:x}. Each map is unmodifiable, holds
+ * values in their JSON shape (maps, lists, strings, numbers, booleans), and marks a key to remove
+ * with {@link State#REMOVED}.
+ *
+ * @param app the {@code app:} keys without the prefix
+ * @param user the {@code user:} keys without the prefix
+ * @param local the keys without a scope prefix
+ */
+ record ScopedState(
+ Map app, Map user, Map local) {}
+
+ /**
+ * A safe codec message from a storage operation, not an arbitrary driver SessionException.
+ *
+ * The store passes the message of this type on to the caller as it is. Any other {@link
+ * SessionException} that surfaces during a transaction, for instance one thrown by the
+ * application's data source, is not trusted that way and gets a message of the store's own.
+ */
+ static final class StorageFailure extends SessionException {
+ private StorageFailure(String message) {
+ super(message);
+ }
+ }
+
+ /**
+ * Splits a combined state or an event's state delta by scope, and takes a snapshot of it.
+ *
+ *
The values are put through JSON once, here. That detaches them from the caller's objects,
+ * which may change while a transaction runs, gives them the shape they will have when read back,
+ * and fails now, before any connection is borrowed, if a value cannot be written. {@code temp:}
+ * entries are dropped before that, so their values are never looked at.
+ *
+ * @param state keys with scope prefixes; {@code null} is the same as empty
+ * @throws SessionException if a value cannot be written as JSON
+ */
+ ScopedState splitState(@Nullable Map state) {
+ try {
+ Map app = new LinkedHashMap<>();
+ Map user = new LinkedHashMap<>();
+ Map local = new LinkedHashMap<>();
+ if (state != null) {
+ state.forEach(
+ (key, value) -> {
+ if (key.startsWith(State.APP_PREFIX)) {
+ app.put(key.substring(State.APP_PREFIX.length()), value);
+ } else if (key.startsWith(State.USER_PREFIX)) {
+ user.put(key.substring(State.USER_PREFIX.length()), value);
+ } else if (!key.startsWith(State.TEMP_PREFIX)) {
+ local.put(key, value);
+ }
+ });
+ }
+ return new ScopedState(snapshotDelta(app), snapshotDelta(user), snapshotDelta(local));
+ } catch (JsonProcessingException | RuntimeException e) {
+ throw new SessionException("Unable to split session state.");
+ }
+ }
+
+ /**
+ * Writes an already scoped document, not a combined prefixed state or an event delta.
+ *
+ * Entries whose value is {@code null} or {@link State#REMOVED} are left out: in a stored
+ * document a key that is not there is how a removed key looks.
+ */
+ String writeState(Map state) {
+ try {
+ return mapper.writeValueAsString(activeValues(state));
+ } catch (JsonProcessingException | RuntimeException e) {
+ throw new StorageFailure("Unable to encode session state.");
+ }
+ }
+
+ /**
+ * Reads an active document. Only top-level nulls mean absence; nested nulls are user data.
+ *
+ * @param json the text of a state column
+ * @return a new, mutable map that nothing else refers to
+ * @throws StorageFailure if the text is {@code null}, not a single JSON object, or not JSON; a
+ * NULL column therefore fails rather than reading as an empty state
+ */
+ Map readState(@Nullable String json) {
+ try {
+ return activeValues(readObject(json));
+ } catch (JsonProcessingException | RuntimeException e) {
+ throw new StorageFailure("Unable to decode session state.");
+ }
+ }
+
+ /**
+ * Applies a scoped delta as replacement/deletion, not recursive map merging.
+ *
+ * Both arguments are this codec's own snapshots, {@code current} from {@link #readState} and
+ * {@code delta} from {@link #splitState}, so their values already have their JSON shape. The
+ * result is a new map over those same values, meant only for {@link #writeState}; encoding and
+ * decoding it once more here would double the cost of every state write.
+ */
+ Map applyState(Map current, @Nullable Map delta) {
+ try {
+ Map updated = activeValues(current);
+ if (delta != null) {
+ delta.forEach(
+ (key, value) -> {
+ if (value == null || value == State.REMOVED) {
+ updated.remove(key);
+ } else {
+ updated.put(key, value);
+ }
+ });
+ }
+ return updated;
+ } catch (RuntimeException e) {
+ throw new StorageFailure("Unable to apply session state.");
+ }
+ }
+
+ /**
+ * Builds a session view; shared documents are authoritative for their own scopes.
+ *
+ * The view keeps the values of the three documents instead of copying them through JSON a
+ * second time. Pass documents decoded by {@link #readState} for this one view, so that no two
+ * sessions end up sharing a nested value.
+ */
+ ConcurrentHashMap mergeState(
+ Map app, Map user, Map local) {
+ try {
+ ConcurrentHashMap combined = new ConcurrentHashMap<>();
+ // The session's own document should hold no prefixed keys. If one is there all the same,
+ // it must not hide or resurrect a shared key, so it is skipped and the shared documents,
+ // added next, decide what the prefixed keys are.
+ activeValues(local)
+ .forEach(
+ (key, value) -> {
+ if (!key.startsWith(State.APP_PREFIX)
+ && !key.startsWith(State.USER_PREFIX)
+ && !key.startsWith(State.TEMP_PREFIX)) {
+ combined.put(key, value);
+ }
+ });
+ activeValues(app).forEach((key, value) -> combined.put(State.APP_PREFIX + key, value));
+ activeValues(user).forEach((key, value) -> combined.put(State.USER_PREFIX + key, value));
+ return combined;
+ } catch (RuntimeException e) {
+ throw new StorageFailure("Unable to merge session state.");
+ }
+ }
+
+ /**
+ * Stores the full Java event, with a narrowly scoped state-delta serialization override.
+ *
+ * The result is what {@code event.toJson()} produces, except inside the event's own state
+ * delta, where {@code temp:} keys are left out and removals are written as {@code null}; see
+ * {@link PersistedDeltaSerializer}. The event itself is not changed.
+ *
+ * @throws SessionException if the event has no ID, without which it could not be read back, or
+ * holds a value that cannot be written as JSON
+ */
+ String writeEvent(Event event) {
+ try {
+ if (event.id() == null) {
+ throw new IllegalArgumentException();
+ }
+ // The attribute tells the delta serializer which object is the event being stored.
+ return eventWriter.withAttribute(EVENT_ROOT, event).writeValueAsString(event);
+ } catch (JsonProcessingException | RuntimeException e) {
+ throw new SessionException("Unable to encode session event.");
+ }
+ }
+
+ /**
+ * Reads an event written by {@link #writeEvent}.
+ *
+ * @param json the text of an {@code event_data} column
+ * @throws StorageFailure if the text is not one JSON object that holds an event with an ID and a
+ * timestamp; a NULL column, which the Python v1 layout allows, is such a case
+ */
+ Event readEvent(@Nullable String json) {
+ try {
+ requireEventIdentity(json);
+ // EventActions.Builder restores top-level null delta values as State.REMOVED. A literal
+ // sentinel-looking string is deliberately never interpreted as a deletion.
+ // Decode the original bytes, not an intermediate tree: a floating-point JsonNode can lose
+ // nanoseconds from typed metadata Instants before their native deserializer sees the value.
+ Event restored = eventReader.readValue(json);
+ // writeEvent never stores a temp: key. Should a row hold one anyway, written by something
+ // else, it still must not come back as part of a loaded session.
+ if (restored.actions() != null && restored.actions().stateDelta() != null) {
+ restored.actions().stateDelta().keySet().removeIf(key -> key.startsWith(State.TEMP_PREFIX));
+ }
+ return restored;
+ } catch (IOException | RuntimeException e) {
+ throw new StorageFailure("Unable to decode session event.");
+ }
+ }
+
+ /**
+ * Requires a JSON object whose {@code id} is text and whose {@code timestamp} fits a long. The
+ * native builder supplies the current wall clock when timestamp is missing; a damaged record must
+ * fail, not silently become a different event when read from storage.
+ *
+ *
History is the largest payload of a session read, so this scans tokens instead of building a
+ * second, full tree of every event. As in a tree, the last duplicate of a field decides.
+ */
+ private void requireEventIdentity(@Nullable String json) throws IOException {
+ boolean id = false;
+ boolean timestamp = false;
+ try (JsonParser parser = mapper.getFactory().createParser(json)) {
+ if (parser.nextToken() != JsonToken.START_OBJECT) {
+ throw new IllegalArgumentException();
+ }
+ // Walk the fields of the root object only. Each turn of the loop is at a field name,
+ // looks at the first token of its value, and then steps over the rest of that value.
+ while (parser.nextToken() == JsonToken.FIELD_NAME) {
+ String field = parser.currentName();
+ JsonToken value = parser.nextToken();
+ if (field.equals("id")) {
+ id = value == JsonToken.VALUE_STRING;
+ } else if (field.equals("timestamp")) {
+ // An integer that does not fit a long is reported as BIG_INTEGER.
+ timestamp =
+ value == JsonToken.VALUE_NUMBER_INT
+ && parser.getNumberType() != JsonParser.NumberType.BIG_INTEGER;
+ }
+ // Does nothing after a scalar; after the start of an object or array it moves to its
+ // end, so that an "id" nested inside some other field is never taken for the event's.
+ parser.skipChildren();
+ }
+ // The object must close, and nothing may follow it.
+ if (parser.currentToken() != JsonToken.END_OBJECT || parser.nextToken() != null) {
+ throw new IllegalArgumentException();
+ }
+ }
+ if (!id || !timestamp) {
+ throw new IllegalArgumentException();
+ }
+ }
+
+ /**
+ * Copies one scope of a delta through JSON and back. Removals take a detour: the sentinel {@link
+ * State#REMOVED} is not something JSON can carry, so it is written as {@code null} and restored
+ * afterwards, which also turns a plain {@code null} value into a removal.
+ */
+ private Map snapshotDelta(Map delta)
+ throws JsonProcessingException {
+ Map normalized = new LinkedHashMap<>();
+ delta.forEach((key, value) -> normalized.put(key, value == State.REMOVED ? null : value));
+ Map snapshot = readObject(mapper.writeValueAsString(normalized));
+ snapshot.replaceAll((key, value) -> value == null ? State.REMOVED : value);
+ return Collections.unmodifiableMap(snapshot);
+ }
+
+ /** Copies the entries that have a value, dropping those that are {@code null} or removed. */
+ private static Map activeValues(Map state) {
+ Map active = new LinkedHashMap<>();
+ state.forEach(
+ (key, value) -> {
+ if (value != null && value != State.REMOVED) {
+ active.put(key, value);
+ }
+ });
+ return active;
+ }
+
+ /** Parses one JSON object into maps, lists and scalar values. */
+ private Map readObject(@Nullable String json) throws JsonProcessingException {
+ return mapper.convertValue(readObjectNode(json), STATE_TYPE);
+ }
+
+ /** Parses text that must be exactly one JSON object: not an array, a scalar, or {@code null}. */
+ private ObjectNode readObjectNode(@Nullable String json) throws JsonProcessingException {
+ JsonNode node = treeReader.readValue(json);
+ if (!(node instanceof ObjectNode object)) {
+ throw new IllegalArgumentException();
+ }
+ return object;
+ }
+
+ /**
+ * Jackson mix-in that gives {@link EventActions#stateDelta()} its own serializer in the event
+ * writer, without touching the class in core or the mapper other code uses.
+ */
+ private abstract static class PersistedActions {
+ @JsonProperty("stateDelta")
+ @JsonSerialize(using = PersistedDeltaSerializer.class)
+ abstract Map stateDelta();
+ }
+
+ /**
+ * Writes the state delta of the event being stored: {@code temp:} entries are skipped without
+ * looking at their values, and a removed key is written as {@code null}, which the builder of
+ * {@link EventActions} turns back into {@link State#REMOVED} on reading.
+ *
+ * It applies only to the delta of the stored event itself. An {@link EventActions} can also
+ * turn up as plain data, inside state or a tool's output, and is then written as usual.
+ */
+ private static final class PersistedDeltaSerializer extends JsonSerializer