Feat: negotiate the sessions database schema version and add an in-place upgrade step

core/src/index.ts

@@ -40,6 +40,7 @@ export {
 } from './code_executors/unsafe_local_code_executor.js';
 export * from './common.js';
 export {DatabaseSessionService} from './sessions/database_session_service.js';
+export {upgradeSessionDatabaseSchema} from './sessions/db/schema_version.js';
 export {getSessionServiceFromUri} from './sessions/registry.js';
 export {VertexAiSessionService} from './sessions/vertex_ai_session_service.js';
 export type {

core/src/sessions/database_session_service.ts

@@ -27,7 +27,6 @@ import {
 import {
   ensureDatabaseCreated,
   getConnectionOptionsFromUri,
-  validateDatabaseSchemaVersion,
 } from './db/operations.js';
 import {
   ENTITIES,
@@ -36,6 +35,7 @@ import {
   StorageSession,
   StorageUserState,
 } from './db/schema.js';
+import {validateDatabaseSchemaVersion} from './db/schema_version.js';
 import {createSession, Session} from './session.js';
 import {State} from './state.js';
 

core/src/sessions/db/README.md

@@ -0,0 +1,76 @@
+# Adding a session database schema version
+
+`DatabaseSessionService` stores its schema version in the `adk_internal_metadata`
+table, under the key `schema_version`. `schema_version.ts` owns the negotiation:
+
+- `LATEST_SCHEMA_VERSION` — the version stamped on a database this build creates.
+- `SUPPORTED_SCHEMA_VERSIONS` — the versions this build can read and write.
+- `upgradeSessionDatabaseSchema()` — the operator entry point that brings an
+  older database up to the latest version in place.
+
+The table name and the key string are a cross-language contract. adk-python reads
+and writes the same row, so neither may change.
+
+## Backward compatibility policy
+
+adk-python states the policy this package follows: "The `DatabaseSessionService`
+is designed to be backward-compatible with the previous schema for a few releases
+(at least 2)." A version therefore stays in `SUPPORTED_SCHEMA_VERSIONS` for at
+least two releases after it stops being the latest.
+
+## Steps to add version N+1
+
+Assume the current version is `1` and you are adding `2`.
+
+1. **Keep the change additive.** Add a nullable column, a new table, or a new
+   index. `ensureDatabaseCreated()` runs `updateSchema({safe: true})`, which
+   applies exactly those and never drops anything. An additive change is readable
+   by both the old and the new client. A non-additive change is a much larger job
+   — see "If the change cannot be additive" below.
+2. **Edit the entities** in `schema.ts`.
+3. **Add the constant** in `schema_version.ts`:
+   `export const SCHEMA_VERSION_2 = '2';`
+4. **Point `LATEST_SCHEMA_VERSION` at it** and add it to
+   `SUPPORTED_SCHEMA_VERSIONS`. Keep `'1'` in the set for its deprecation window.
+5. **Branch the business logic** in `DatabaseSessionService` on the stored
+   version if a `'1'` database needs different reads or writes. Use
+   `readSchemaVersion()`; do not add a second entity set.
+6. **Warn on the older version.** `validateDatabaseSchemaVersion()` accepts an
+   older supported version silently, because there is no supported-but-older
+   version yet. Add
+   `if (version !== LATEST_SCHEMA_VERSION) logger.warn(...)` after the
+   `assertCompatibleVersion()` call, pointing operators at
+   `upgradeSessionDatabaseSchema()`.
+7. **Test the pair.** Cover a `'1'` database opened by the new build, and the
+   upgrade from `'1'` to `'2'`.
+8. **Deprecate.** After at least two releases, remove `'1'` from
+   `SUPPORTED_SCHEMA_VERSIONS`. A `'1'` database then fails to open with the
+   error that names `upgradeSessionDatabaseSchema()`.
+
+## Why there is only one entity set
+
+adk-python keeps a complete SQLAlchemy model set per version (`schemas/v0.py`,
+`schemas/v1.py`) and selects one at runtime with its `_SchemaClasses` helper.
+Two declarative metadatas can coexist in one process and either can be bound to
+an engine at call time.
+
+MikroORM cannot do that. `MikroORM.init()` discovers and freezes entity metadata,
+and `DatabaseSessionService` passes `entities: ENTITIES` in before init. Selecting
+an entity set per database would need, in order:
+
+1. a throwaway pre-init connection purely to read `adk_internal_metadata`;
+2. closing it and re-initialising the real ORM with the selected set;
+3. an indirection object over every `em.find` / `em.create` call in
+   `DatabaseSessionService`, mirroring `_SchemaClasses`;
+4. duplicate entity classes with duplicate `tableName` declarations, which
+   MikroORM's discovery rejects unless the lists are disjoint.
+
+That cost is paid on every startup, so this package keeps one entity set and
+makes each bump additive instead.
+
+## If the change cannot be additive
+
+A retyped column, a removed column, or a changed payload encoding cannot be read
+by both clients. That is the only case that forces the `_SchemaClasses`
+equivalent, at the four costs above. Consider a new table that the old client
+ignores before you take that on.

core/src/sessions/db/operations.ts

@@ -6,12 +6,7 @@
 
 import {MikroORM, Options as MikroORMOptions} from '@mikro-orm/core';
 import {redactUriPassword} from '../../utils/redact_uri.js';
-import {
-  ENTITIES,
-  SCHEMA_VERSION_1_JSON,
-  SCHEMA_VERSION_KEY,
-  StorageMetadata,
-} from './schema.js';
+import {ENTITIES} from './schema.js';
 
 /**
  * Parses a database connection URI and returns MikroORM Options.
@@ -75,32 +70,3 @@ export async function ensureDatabaseCreated(orm: MikroORM): Promise<void> {
   // creates tables if they don't exist. Safe mode prevents dropping columns or tables.
   await orm.schema.updateSchema({safe: true});
 }
-
-/**
- * Validates the schema version.
- *
- * @param orm The MikroORM instance.
- * @throws Error if the schema version is not compatible.
- */
-export async function validateDatabaseSchemaVersion(orm: MikroORM) {
-  const em = orm.em.fork();
-  const existing = await em.findOne(StorageMetadata, {
-    key: SCHEMA_VERSION_KEY,
-  });
-
-  if (existing) {
-    if (existing.value !== SCHEMA_VERSION_1_JSON) {
-      throw new Error(
-        `ADK Database schema version ${existing.value} is not compatible.`,
-      );
-    }
-    return;
-  }
-
-  const newVersion = em.create(StorageMetadata, {
-    key: SCHEMA_VERSION_KEY,
-    value: SCHEMA_VERSION_1_JSON,
-  });
-
-  await em.persist(newVersion).flush();
-}

core/src/sessions/db/schema.ts

@@ -11,8 +11,6 @@ import {
   transformToSnakeCaseEvent,
 } from '../../events/event.js';
 
-export const SCHEMA_VERSION_KEY = 'schema_version';
-export const SCHEMA_VERSION_1_JSON = '1';
 export const STORAGE_KEY_COLUMN_LENGTH = 191;
 
 /**

core/src/sessions/db/schema_version.ts

@@ -0,0 +1,161 @@
+/**
+ * @license
+ * Copyright 2026 Google LLC
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+import {MikroORM, Options as MikroORMOptions} from '@mikro-orm/core';
+import {logger} from '../../utils/logger.js';
+import {
+  ensureDatabaseCreated,
+  getConnectionOptionsFromUri,
+} from './operations.js';
+import {ENTITIES, StorageMetadata} from './schema.js';
+
+/** Key of the `adk_internal_metadata` row that holds the schema version. */
+export const SCHEMA_VERSION_KEY = 'schema_version';
+
+/** JSON-serialized event payload schema, shared with adk-python's v1. */
+export const SCHEMA_VERSION_1_JSON = '1';
+
+/** Version stamped on a database this build creates. */
+export const LATEST_SCHEMA_VERSION = SCHEMA_VERSION_1_JSON;
+
+/**
+ * Versions this build can read and write.
+ *
+ * A release keeps the previous version here for its deprecation window rather
+ * than removing it, so an existing database keeps opening after a bump. See
+ * `core/src/sessions/db/README.md`.
+ */
+export const SUPPORTED_SCHEMA_VERSIONS: ReadonlySet<string> = new Set([
+  SCHEMA_VERSION_1_JSON,
+]);
+
+/**
+ * Throws when the database holds a version this build cannot read.
+ *
+ * An absent version row is a new database, which is compatible by definition.
+ *
+ * @param version The stored version, or undefined when the row is absent.
+ * @throws Error naming the accepted versions and both remedies.
+ */
+function assertCompatibleVersion(version: string | undefined): void {
+  if (version === undefined || SUPPORTED_SCHEMA_VERSIONS.has(version)) {
+    return;
+  }
+
+  const supported = [...SUPPORTED_SCHEMA_VERSIONS].join(', ');
+  throw new Error(
+    `ADK Database schema version ${version} is not compatible. ` +
+      `This build of ADK supports schema version(s) ${supported}. ` +
+      `Upgrade the @google/adk package if the database was written by a ` +
+      `newer release, or call upgradeSessionDatabaseSchema() to bring an ` +
+      `older database up to version ${LATEST_SCHEMA_VERSION}.`,
+  );
+}
+
+/**
+ * Reads the schema version stored in the database.
+ *
+ * @param orm The MikroORM instance.
+ * @returns The stored version, or undefined when the row is absent.
+ */
+export async function readSchemaVersion(
+  orm: MikroORM,
+): Promise<string | undefined> {
+  const em = orm.em.fork();
+  const existing = await em.findOne(StorageMetadata, {key: SCHEMA_VERSION_KEY});
+
+  return existing?.value;
+}
+
+/**
+ * Writes the schema version, replacing any value already stored.
+ *
+ * @param orm The MikroORM instance.
+ * @param version The version to store.
+ */
+export async function stampSchemaVersion(
+  orm: MikroORM,
+  version: string,
+): Promise<void> {
+  const em = orm.em.fork();
+  const existing = await em.findOne(StorageMetadata, {key: SCHEMA_VERSION_KEY});
+
+  if (existing) {
+    existing.value = version;
+    return em.flush();
+  }
+
+  const created = em.create(StorageMetadata, {
+    key: SCHEMA_VERSION_KEY,
+    value: version,
+  });
+
+  return em.persist(created).flush();
+}
+
+/**
+ * Validates the schema version of an open database, stamping a new one.
+ *
+ * An existing version row is never rewritten; moving a database to a newer
+ * version is `upgradeSessionDatabaseSchema`'s job.
+ *
+ * @param orm The MikroORM instance.
+ * @throws Error if the stored version is outside `SUPPORTED_SCHEMA_VERSIONS`.
+ */
+export async function validateDatabaseSchemaVersion(
+  orm: MikroORM,
+): Promise<void> {
+  const version = await readSchemaVersion(orm);
+  assertCompatibleVersion(version);
+
+  if (version === undefined) {
+    await stampSchemaVersion(orm, LATEST_SCHEMA_VERSION);
+  }
+}
+
+/**
+ * Brings a session database up to `LATEST_SCHEMA_VERSION` in place.
+ *
+ * Applies the additive DDL this build expects, then stamps the version row.
+ * Idempotent: a database already at the latest version is left untouched. The
+ * connection is closed on every exit path.
+ *
+ * @param connectionStringOrOptions A database URI, or MikroORM options
+ *   carrying a driver.
+ * @throws Error if the stored version is outside `SUPPORTED_SCHEMA_VERSIONS`.
+ */
+export async function upgradeSessionDatabaseSchema(
+  connectionStringOrOptions: string | MikroORMOptions,
+): Promise<void> {
+  const options =
+    typeof connectionStringOrOptions === 'string'
+      ? await getConnectionOptionsFromUri(connectionStringOrOptions)
+      : {...connectionStringOrOptions, entities: ENTITIES};
+
+  if (!options.driver) {
+    throw new Error('Driver is required when passing options object.');
+  }
+
+  const orm = await MikroORM.init(options);
+
+  try {
+    await ensureDatabaseCreated(orm);
+
+    const version = await readSchemaVersion(orm);
+    assertCompatibleVersion(version);
+
+    if (version === LATEST_SCHEMA_VERSION) {
+      logger.debug(
+        `Session database is already at schema version ${LATEST_SCHEMA_VERSION}.`,
+      );
+      return;
+    }
+
+    await stampSchemaVersion(orm, LATEST_SCHEMA_VERSION);
+  } finally {
+    await orm.close();
+  }
+}

core/test/sessions/database_session_service_test.ts

@@ -15,7 +15,7 @@ import {MikroORM} from '@mikro-orm/core';
 import {SqliteDriver} from '@mikro-orm/sqlite';
 import {afterEach, beforeEach, describe, expect, it} from 'vitest';
 import {isDatabaseConnectionString} from '../../src/sessions/database_session_service.js';
-import {validateDatabaseSchemaVersion} from '../../src/sessions/db/operations.js';
+import {validateDatabaseSchemaVersion} from '../../src/sessions/db/schema_version.js';
 
 describe('DatabaseSessionService', () => {
   let service: DatabaseSessionService;