diff --git a/.changeset/plugin-storage-bulk-upserts.md b/.changeset/plugin-storage-bulk-upserts.md
new file mode 100644
index 0000000000..4265a60e98
--- /dev/null
+++ b/.changeset/plugin-storage-bulk-upserts.md
@@ -0,0 +1,7 @@
+---
+"@executor-js/fumadb": patch
+"@executor-js/sdk": patch
+---
+
+Add a FumaDB bulk upsert query path and route plugin-storage bulk writes through
+it so existing rows are updated without delete/reinsert churn.
diff --git a/packages/core/fumadb/src/adapters/drizzle/query.ts b/packages/core/fumadb/src/adapters/drizzle/query.ts
index 9471ecda0d..21c7bc1e95 100644
--- a/packages/core/fumadb/src/adapters/drizzle/query.ts
+++ b/packages/core/fumadb/src/adapters/drizzle/query.ts
@@ -22,6 +22,77 @@ type P_DBType = PostgreSQL.PgDatabase<
 
 const CREATE_MANY_BATCH_SIZE = 500;
 
+// A multi-row write binds (rows * columns) parameters in one statement, and
+// engines cap bound parameters per statement (older SQLite: 999, Cloudflare
+// D1: 100). When the adapter does not advertise its limit, budget against the
+// conservative 999 floor so a wide table cannot overflow with "too many SQL
+// variables" on stricter engines.
+const DEFAULT_MAX_BOUND_PARAMETERS = 999;
+
+function parameterBoundedBatchSize(
+  table: AnyTable,
+  columnsPerRow: number,
+  reservedParameters: number,
+  maxBoundParameters: number | undefined
+): number {
+  const budget = maxBoundParameters ?? DEFAULT_MAX_BOUND_PARAMETERS;
+  const rowsPerStatement = Math.floor((budget - reservedParameters) / columnsPerRow);
+  if (rowsPerStatement < 1) {
+    // Even a single row cannot fit the advertised budget. Clamping to one row
+    // anyway would silently emit a statement the engine may reject, so fail
+    // fast and name the numbers instead.
+    const reservedNote =
+      reservedParameters > 0 ? ` and the predicate reserves ${reservedParameters} more` : "";
+    // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- boundary: adapter rejects writes that cannot fit the engine's bound-parameter budget
+    throw new Error(
+      `[FumaDB Drizzle] Cannot write table "${table.ormName}": one row binds ${columnsPerRow} bound parameters${reservedNote}, which exceeds the ${budget}-parameter budget per statement.`
+    );
+  }
+  return Math.min(CREATE_MANY_BATCH_SIZE, rowsPerStatement);
+}
+
+// SQLite transaction scopes run as raw BEGIN/SAVEPOINT statements on one
+// shared connection, so two scopes started concurrently would interleave
+// their control statements: two top-level BEGINs collide, and sibling
+// savepoints at one nesting depth reuse each other's names — one sibling's
+// ROLLBACK TO can undo work the other already released. Serialize scope
+// execution per connection handle and nesting depth: same-depth scopes run
+// one after another, while a parent scope at depth N can still open its
+// child scope at depth N + 1 without deadlocking. SQLite is single-writer,
+// so this serialization costs no real concurrency.
+const transactionScopeQueues = new WeakMap<object, Map<number, Promise<void>>>();
+
+async function runSerializedScope<T>(
+  handle: object,
+  depth: number,
+  fn: () => Promise<T>,
+): Promise<T> {
+  let queues = transactionScopeQueues.get(handle);
+  if (!queues) {
+    queues = new Map();
+    transactionScopeQueues.set(handle, queues);
+  }
+  const previous = queues.get(depth) ?? Promise.resolve();
+  let release!: () => void;
+  queues.set(
+    depth,
+    new Promise<void>((resolve) => {
+      release = resolve;
+    }),
+  );
+  await previous;
+  // oxlint-disable-next-line executor/no-try-catch-or-throw -- scope release must survive the wrapped failure
+  try {
+    return await fn();
+  } finally {
+    release();
+  }
+}
+
+// Savepoint names are invocation-unique as defense in depth: even if two
+// scopes ever interleave, a ROLLBACK TO can only target its own savepoint.
+let savepointSequence = 0;
+
 function buildWhere(
   toDrizzle: (col: AnyColumn) => ColumnType,
   condition: Condition
@@ -121,6 +192,16 @@ function buildWhere(
   );
 }
 
+function countConditionParameters(condition: Condition): number {
+  if (condition.type === ConditionType.Compare) {
+    if (condition.b instanceof Column) return 0;
+    if (Array.isArray(condition.b)) return condition.b.length;
+    return 1;
+  }
+  if (condition.type === ConditionType.Not) return countConditionParameters(condition.item);
+  return condition.items.reduce((count, item) => count + countConditionParameters(item), 0);
+}
+
 function mapValues(
   values: Record<string, unknown>,
   table: AnyTable
@@ -171,7 +252,8 @@ export function fromDrizzle(
   _db: unknown,
   provider: SQLProvider,
   interactiveTransactions: boolean = true,
-  maxBoundParameters?: number
+  maxBoundParameters?: number,
+  transactionDepth: number = 0
 ): AbstractQuery<AnySchema> {
   const [db, drizzleTables] = parseDrizzle(_db);
 
@@ -195,6 +277,41 @@ export function fromDrizzle(
     throw new Error("[FumaDB Drizzle] Database cannot execute raw transaction statements.");
   }
 
+  // Runs `fn` atomically when the engine allows it. SQLite drivers get raw
+  // BEGIN/COMMIT on the shared connection — or a SAVEPOINT when this adapter
+  // instance already lives inside a transaction, so nested use (e.g. a bulk
+  // upsert splitting into predicate groups inside a caller's transaction)
+  // does not issue a second BEGIN. Other providers get a driver transaction
+  // whose handle must be used for every statement inside `fn`. When the
+  // engine rejects interactive transactions (Cloudflare D1), statements
+  // auto-commit — that engine constraint is documented on `transaction`.
+  async function runAtomically<T>(fn: (handle: typeof db) => Promise<T>): Promise<T> {
+    if (!interactiveTransactions) {
+      return fn(db);
+    }
+    if (provider === "sqlite") {
+      return runSerializedScope(db, transactionDepth, async () => {
+        savepointSequence += 1;
+        const savepoint = transactionDepth > 0 ? `fumadb_tx_${savepointSequence}` : undefined;
+        await executeRaw(savepoint ? `SAVEPOINT ${savepoint}` : "BEGIN");
+        try {
+          const result = await fn(db);
+          await executeRaw(savepoint ? `RELEASE SAVEPOINT ${savepoint}` : "COMMIT");
+          return result;
+        } catch (e) {
+          if (savepoint) {
+            await executeRaw(`ROLLBACK TO SAVEPOINT ${savepoint}`);
+            await executeRaw(`RELEASE SAVEPOINT ${savepoint}`);
+          } else {
+            await executeRaw("ROLLBACK");
+          }
+          throw e;
+        }
+      });
+    }
+    return db.transaction((tx) => fn(tx as unknown as typeof db));
+  }
+
   function toDrizzle(v: AnyTable): TableType {
     const out = drizzleTables[v.names.drizzle];
     if (out) return out;
@@ -303,6 +420,117 @@ export function fromDrizzle(
         await this.createMany(table, [v.create]);
       }
     },
+    async upsertMany(table, v) {
+      if (v.values.length === 0) return;
+      if (v.target.length === 0) {
+        // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- boundary: adapter rejects invalid upsert shape
+        throw new Error("[FumaDB] upsertMany requires at least one target column.");
+      }
+      if (v.update.length === 0) {
+        // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- boundary: adapter rejects invalid upsert shape
+        throw new Error("[FumaDB] upsertMany requires at least one update column.");
+      }
+      if (provider !== "sqlite" && provider !== "postgresql") {
+        // This path issues several statements per row, so a failure halfway
+        // through must not leave a prefix of the rows committed. These
+        // providers have native driver transactions; `transaction` routes
+        // every per-row statement through one of them.
+        await this.transaction(async (scoped) => {
+          for (const value of v.values) {
+            const targetCondition: Condition = {
+              type: ConditionType.And,
+              items: v.target.map((column) => ({
+                type: ConditionType.Compare,
+                a: column,
+                operator: "=",
+                b: value[column.ormName],
+              })),
+            };
+            await scoped.internal.upsert(table, {
+              where: v.where
+                ? { type: ConditionType.And, items: [targetCondition, v.where] }
+                : targetCondition,
+              update: Object.fromEntries(
+                v.update.map((column) => [column.ormName, value[column.ormName]]),
+              ),
+              create: value,
+            });
+          }
+        });
+        return;
+      }
+
+      const drizzleTable = toDrizzle(table);
+      const values = v.values.map((value) => mapValues(value, table));
+      const where = v.where ? buildWhere(toDrizzleColumn, v.where) : undefined;
+      const whereParameters = v.where ? countConditionParameters(v.where) : 0;
+      const columnsPerRow = values.length > 0 ? Math.max(1, Object.keys(values[0]!).length) : 1;
+      const batchSize = parameterBoundedBatchSize(
+        table,
+        columnsPerRow,
+        whereParameters,
+        maxBoundParameters,
+      );
+      const target = v.target.map((column) => drizzleTable[column.names.drizzle]);
+      const set = Object.fromEntries(
+        v.update.map((column) => [
+          column.names.drizzle,
+          // `sql.identifier` quotes the physical column name; a raw
+          // interpolation would emit e.g. `excluded.display-name` unquoted,
+          // which the engine parses as an expression.
+          Drizzle.sql`excluded.${Drizzle.sql.identifier(column.names.sql)}`,
+        ]),
+      );
+
+      const buildStatements = (handle: typeof db): unknown[] => {
+        const statements: unknown[] = [];
+        for (let i = 0; i < values.length; i += batchSize) {
+          const batch = values.slice(i, i + batchSize);
+          const insert = handle.insert(drizzleTable).values(batch) as unknown as {
+            onConflictDoUpdate: (input: {
+              readonly target: typeof target;
+              readonly set: typeof set;
+              readonly where?: typeof where;
+            }) => unknown;
+          };
+          statements.push(
+            insert.onConflictDoUpdate({
+              target,
+              set,
+              ...(where === undefined ? {} : { where }),
+            }),
+          );
+        }
+        return statements;
+      };
+      const executeStatements = async (handle: typeof db) => {
+        for (const statement of buildStatements(handle)) {
+          await statement;
+        }
+      };
+      const statementCount = Math.ceil(values.length / batchSize);
+
+      // D1 rejects interactive transactions but its native batch API executes
+      // prepared statements as one transaction. Drizzle exposes that API on
+      // the database handle, so keep parameter-bounded upserts atomic instead
+      // of auto-committing each statement independently.
+      const nativeBatch = db as unknown as {
+        readonly batch?: (statements: readonly unknown[]) => Promise<unknown>;
+      };
+      if (!interactiveTransactions && statementCount > 1 && nativeBatch.batch) {
+        await nativeBatch.batch(buildStatements(db));
+        return;
+      }
+
+      if (statementCount === 1) {
+        await executeStatements(db);
+        return;
+      }
+
+      // One logical upsert split into several parameter-bounded statements
+      // stays atomic: run them inside one transaction.
+      await runAtomically(executeStatements);
+    },
     async findMany(table, v) {
       return (
         await db.query[table.names.drizzle].findMany(buildQueryConfig(table, v))
@@ -353,43 +581,48 @@ export function fromDrizzle(
     },
 
     async createMany(table, values) {
+      if (values.length === 0) return [];
       const idField = table.getIdColumn().names.drizzle;
       const drizzleTable = toDrizzle(table);
       values = values.map((v) => mapValues(v, table));
-      // A multi-row insert binds (rows * columns) parameters in one statement.
-      // Some engines cap bound parameters per query (Cloudflare D1: 100), so
-      // size the batch by PARAMETER count, not row count — otherwise a wide
-      // table (e.g. tools) overflows with "too many SQL variables". Engines
-      // without a tight cap keep the row-count batch.
-      const columnsPerRow = values.length > 0 ? Math.max(1, Object.keys(values[0]!).length) : 1;
-      const batchSize = maxBoundParameters
-        ? Math.max(1, Math.min(CREATE_MANY_BATCH_SIZE, Math.floor(maxBoundParameters / columnsPerRow)))
-        : CREATE_MANY_BATCH_SIZE;
+      const columnsPerRow = Math.max(1, Object.keys(values[0]!).length);
+      const batchSize = parameterBoundedBatchSize(table, columnsPerRow, 0, maxBoundParameters);
       const batches: (typeof values)[] = [];
       for (let i = 0; i < values.length; i += batchSize) {
         batches.push(values.slice(i, i + batchSize));
       }
 
-      if (provider === "sqlite" || provider === "postgresql") {
-        const out: { _id: unknown }[] = [];
-        for (const batch of batches) {
-          out.push(
-            ...(await (db as unknown as P_DBType)
-              .insert(drizzleTable as unknown as P_TableType)
-              .values(batch)
-              .returning({
-                _id: (drizzleTable as unknown as P_TableType)[idField],
-              })),
-          );
+      const insertBatches = async (handle: typeof db): Promise<{ _id: unknown }[]> => {
+        if (provider === "sqlite" || provider === "postgresql") {
+          const out: { _id: unknown }[] = [];
+          for (const batch of batches) {
+            out.push(
+              ...(await (handle as unknown as P_DBType)
+                .insert(drizzleTable as unknown as P_TableType)
+                .values(batch)
+                .returning({
+                  _id: (drizzleTable as unknown as P_TableType)[idField],
+                })),
+            );
+          }
+          return out;
         }
-        return out;
-      }
 
-      const results: Record<string, unknown>[] = [];
-      for (const batch of batches) {
-        results.push(...(await db.insert(drizzleTable).values(batch).$returningId()));
-      }
-      return results.map((result) => ({ _id: result[idField] }));
+        const results: Record<string, unknown>[] = [];
+        for (const batch of batches) {
+          results.push(...(await handle.insert(drizzleTable).values(batch).$returningId()));
+        }
+        return results.map((result) => ({ _id: result[idField] }));
+      };
+
+      if (batches.length === 1) return insertBatches(db);
+      // One logical insert split into parameter-bounded statements must stay
+      // atomic: a later batch's constraint failure rolls back the earlier
+      // batches. (Engines without interactive transactions auto-commit each
+      // statement; `createMany` needs the inserted ids back, which the native
+      // batch API's result shape does not guarantee across drivers, so those
+      // engines keep sequential statements — their documented constraint.)
+      return runAtomically(insertBatches);
     },
 
     async deleteMany(table, v) {
@@ -409,23 +642,44 @@ export function fromDrizzle(
       // each statement auto-commits, so there is no atomic rollback (the
       // engine's constraint, not ours). libSQL/Postgres keep real transactions.
       if (!interactiveTransactions) {
-        return run(fromDrizzle(schema, _db, provider, interactiveTransactions, maxBoundParameters));
+        return run(
+          fromDrizzle(
+            schema,
+            _db,
+            provider,
+            interactiveTransactions,
+            maxBoundParameters,
+            transactionDepth
+          )
+        );
       }
 
       if (provider === "sqlite") {
-        await executeRaw("BEGIN");
-        try {
-          const result = await run(fromDrizzle(schema, _db, provider, interactiveTransactions, maxBoundParameters));
-          await executeRaw("COMMIT");
-          return result;
-        } catch (e) {
-          await executeRaw("ROLLBACK");
-          throw e;
-        }
+        return runAtomically(() =>
+          run(
+            fromDrizzle(
+              schema,
+              _db,
+              provider,
+              interactiveTransactions,
+              maxBoundParameters,
+              transactionDepth + 1
+            )
+          )
+        );
       }
 
       return db.transaction((tx) =>
-        run(fromDrizzle(schema, tx, provider, interactiveTransactions, maxBoundParameters))
+        run(
+          fromDrizzle(
+            schema,
+            tx,
+            provider,
+            interactiveTransactions,
+            maxBoundParameters,
+            transactionDepth + 1
+          )
+        )
       );
     },
   });
diff --git a/packages/core/fumadb/src/adapters/drizzle/upsert-many-generic.test.ts b/packages/core/fumadb/src/adapters/drizzle/upsert-many-generic.test.ts
new file mode 100644
index 0000000000..333d4a2f7d
--- /dev/null
+++ b/packages/core/fumadb/src/adapters/drizzle/upsert-many-generic.test.ts
@@ -0,0 +1,120 @@
+import { expect, test } from "@effect/vitest";
+
+import { column, idColumn, schema, table } from "../../schema";
+import { fromDrizzle } from "./query";
+
+// The generic bulk-upsert path (providers without ON CONFLICT support, e.g.
+// MySQL) issues several statements per row. The unit harness has no MySQL
+// server, so a recording fake of the Drizzle handle asserts the transaction
+// routing instead: every per-row statement must run on the driver
+// transaction's handle, and a failing row must roll the earlier rows back.
+const v1 = schema({
+  version: "1.0.0",
+  tables: {
+    rows: table("generic_rows", {
+      id: idColumn("id", "varchar(255)"),
+      value: column("value", "string"),
+    }),
+  },
+});
+
+interface FakeDbOptions {
+  // A row whose `value` matches makes its INSERT fail.
+  readonly failOnValue?: string;
+}
+
+const createFakeMysqlDb = (options: FakeDbOptions = {}) => {
+  const committed: Record<string, unknown>[] = [];
+  const events: string[] = [];
+  const fakeTable = { id: { name: "id" }, value: { name: "value" } };
+
+  const makeHandle = (label: string, sink: Record<string, unknown>[]) => ({
+    _: { fullSchema: { rows: fakeTable } },
+    select: () => {
+      const builder = {
+        from: () => builder,
+        limit: () => builder,
+        where: () => builder,
+        execute: async (): Promise<Record<string, unknown>[]> => [],
+      };
+      return builder;
+    },
+    insert: () => ({
+      values: (rows: Record<string, unknown>[]) => ({
+        $returningId: async () => {
+          events.push(`${label}:insert`);
+          for (const row of rows) {
+            if (options.failOnValue !== undefined && row["value"] === options.failOnValue) {
+              // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- test forces a row's statement to fail
+              throw new Error("statement failure");
+            }
+            sink.push(row);
+          }
+          return rows.map((row) => ({ id: row["id"] }));
+        },
+      }),
+    }),
+    transaction: async <T>(callback: (tx: unknown) => Promise<T>): Promise<T> => {
+      events.push("transaction:begin");
+      const staged: Record<string, unknown>[] = [];
+      const tx = makeHandle("tx", staged);
+      // oxlint-disable-next-line executor/no-try-catch-or-throw -- fake driver mirrors commit/rollback
+      try {
+        const result = await callback(tx);
+        committed.push(...staged);
+        events.push("transaction:commit");
+        return result;
+      } catch (error) {
+        events.push("transaction:rollback");
+        throw error;
+      }
+    },
+  });
+
+  // Statements on the root handle auto-commit, exactly like a driver outside
+  // an explicit transaction.
+  return { db: makeHandle("root", committed), committed, events };
+};
+
+const upsertAll = async (db: unknown, values: readonly { id: string; value: string }[]) => {
+  const orm = fromDrizzle(v1, db, "mysql");
+  const rows = v1.tables.rows;
+  expect(orm.internal.upsertMany).toBeDefined();
+  await orm.internal.upsertMany?.(rows, {
+    target: [rows.columns.id],
+    update: [rows.columns.value],
+    values: values.map((value) => ({ ...value })),
+  });
+};
+
+test("generic bulk upsert runs every row inside one driver transaction", async () => {
+  const { db, committed, events } = createFakeMysqlDb();
+
+  await upsertAll(db, [
+    { id: "r1", value: "a" },
+    { id: "r2", value: "b" },
+    { id: "r3", value: "c" },
+  ]);
+
+  expect(committed.map((row) => row["id"])).toEqual(["r1", "r2", "r3"]);
+  expect(events).toContain("transaction:begin");
+  expect(events).toContain("transaction:commit");
+  // No row statement may run on the auto-committing root handle.
+  expect(events).not.toContain("root:insert");
+});
+
+test("generic bulk upsert rolls earlier rows back when a later row fails", async () => {
+  const { db, committed, events } = createFakeMysqlDb({ failOnValue: "poison" });
+
+  await expect(
+    upsertAll(db, [
+      { id: "r1", value: "a" },
+      { id: "r2", value: "b" },
+      { id: "r3", value: "poison" },
+    ]),
+  ).rejects.toThrow("statement failure");
+
+  // A single logical bulk upsert must not leave a partial prefix committed.
+  expect(committed).toEqual([]);
+  expect(events).toContain("transaction:rollback");
+});
diff --git a/packages/core/fumadb/src/adapters/memory/index.ts b/packages/core/fumadb/src/adapters/memory/index.ts
index a596373f5c..205655e9fa 100644
--- a/packages/core/fumadb/src/adapters/memory/index.ts
+++ b/packages/core/fumadb/src/adapters/memory/index.ts
@@ -174,6 +174,39 @@ export function memoryAdapter(options: MemoryAdapterOptions = {}): FumaDBAdapter
           }
           await this.create(table, v.create);
         },
+        async upsertMany(table, v) {
+          if (v.target.length === 0) {
+            // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- boundary: adapter rejects invalid upsert shape
+            throw new Error("[FumaDB] upsertMany requires at least one target column.");
+          }
+          if (v.update.length === 0) {
+            // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- boundary: adapter rejects invalid upsert shape
+            throw new Error("[FumaDB] upsertMany requires at least one update column.");
+          }
+          for (const value of v.values) {
+            // Mirror SQL `ON CONFLICT ... DO UPDATE ... WHERE`: the unique
+            // target alone detects the conflict, and the predicate only
+            // decides whether the conflicting row may be updated. A
+            // policy-excluded conflict skips the row; it never inserts a
+            // duplicate of the unique target.
+            const conflicting = tableRows(db, table).find((row) =>
+              v.target.every((column) => row[column.ormName] === value[column.ormName]),
+            );
+            if (conflicting) {
+              if (!matchesCondition(conflicting, v.where)) continue;
+              Object.assign(
+                conflicting,
+                cloneValue(
+                  Object.fromEntries(
+                    v.update.map((column) => [column.ormName, value[column.ormName]]),
+                  ),
+                ),
+              );
+              continue;
+            }
+            await this.create(table, value);
+          }
+        },
         async create(table, values) {
           const row = applyDefaults(table, values);
           tableRows(db, table).push(row);
diff --git a/packages/core/fumadb/src/query/index.ts b/packages/core/fumadb/src/query/index.ts
index 2128adb0ef..45633dfb94 100644
--- a/packages/core/fumadb/src/query/index.ts
+++ b/packages/core/fumadb/src/query/index.ts
@@ -174,6 +174,22 @@ export interface AbstractQuery<S extends AnySchema> {
     }
   ) => Promise<void>;
 
+  /**
+   * Bulk upsert rows by a unique column target.
+   *
+   * Adapters with native conflict support should implement this as one or more
+   * `INSERT ... ON CONFLICT ... DO UPDATE` statements. Other adapters may fall
+   * back to per-row `upsert`.
+   */
+  upsertMany: <TableName extends keyof S["tables"]>(
+    table: TableName,
+    v: {
+      target: (keyof S["tables"][TableName]["columns"])[];
+      update: (keyof S["tables"][TableName]["columns"])[];
+      values: TableToInsertValues<S["tables"][TableName]>[];
+    }
+  ) => Promise<void>;
+
   /**
    * Note: you cannot update the id of a row, some databases don't support that (including MongoDB).
    */
diff --git a/packages/core/fumadb/src/query/orm/index.ts b/packages/core/fumadb/src/query/orm/index.ts
index f0bac42e78..1136b5e318 100644
--- a/packages/core/fumadb/src/query/orm/index.ts
+++ b/packages/core/fumadb/src/query/orm/index.ts
@@ -4,6 +4,7 @@ import type {
   AnySchema,
   AnyTable,
 } from "../../schema";
+import { Column } from "../../schema";
 import type {
   AbstractQuery,
   AnySelectClause,
@@ -12,7 +13,12 @@ import type {
   JoinBuilder,
   OrderBy,
 } from "..";
-import { buildCondition, createBuilder, type Condition } from "../condition-builder";
+import {
+  buildCondition,
+  createBuilder,
+  type Condition,
+  ConditionType,
+} from "../condition-builder";
 
 export interface CompiledJoin {
   relation: AnyRelation;
@@ -231,6 +237,68 @@ const applyUpdatePolicies = async (
   return nextWhere;
 };
 
+// Structural equality over predicate values. Policy predicates may carry any
+// column value shape — string, number, bigint, boolean, null, Date, binary,
+// JSON objects, and arrays of those — so serializing them to string keys
+// (e.g. JSON.stringify) either throws (bigint) or collides (values with the
+// same serialized form). Comparing structurally is unambiguous; a false
+// negative only splits a group and never merges rows under the wrong
+// predicate.
+export const predicateValuesEqual = (a: unknown, b: unknown): boolean => {
+  if (Object.is(a, b)) return true;
+  if (a instanceof Date || b instanceof Date) {
+    return a instanceof Date && b instanceof Date && a.getTime() === b.getTime();
+  }
+  if (a instanceof Uint8Array || b instanceof Uint8Array) {
+    return (
+      a instanceof Uint8Array &&
+      b instanceof Uint8Array &&
+      a.length === b.length &&
+      a.every((byte, index) => byte === b[index])
+    );
+  }
+  if (Array.isArray(a) || Array.isArray(b)) {
+    if (!Array.isArray(a) || !Array.isArray(b) || a.length !== b.length) return false;
+    // Compare by index instead of `.every`, which skips holes in sparse
+    // arrays — `Array(1)` would otherwise equal `[123]` and merge rows under
+    // the wrong predicate group. A hole and a present element are unequal.
+    for (let index = 0; index < a.length; index += 1) {
+      const aHas = index in a;
+      if (aHas !== index in b) return false;
+      if (aHas && !predicateValuesEqual(a[index], b[index])) return false;
+    }
+    return true;
+  }
+  if (isRecord(a) && isRecord(b)) {
+    const aKeys = Object.keys(a);
+    const bKeys = Object.keys(b);
+    return (
+      aKeys.length === bKeys.length &&
+      aKeys.every((key) => Object.hasOwn(b, key) && predicateValuesEqual(a[key], b[key]))
+    );
+  }
+  return false;
+};
+
+const conditionsEqual = (a: Condition | undefined, b: Condition | undefined): boolean => {
+  if (a === undefined || b === undefined) return a === b;
+  if (a.type === ConditionType.Compare || b.type === ConditionType.Compare) {
+    if (a.type !== ConditionType.Compare || b.type !== ConditionType.Compare) return false;
+    if (a.a !== b.a || a.operator !== b.operator) return false;
+    if (a.b instanceof Column || b.b instanceof Column) return a.b === b.b;
+    return predicateValuesEqual(a.b, b.b);
+  }
+  if (a.type === ConditionType.Not || b.type === ConditionType.Not) {
+    if (a.type !== ConditionType.Not || b.type !== ConditionType.Not) return false;
+    return conditionsEqual(a.item, b.item);
+  }
+  return (
+    a.type === b.type &&
+    a.items.length === b.items.length &&
+    a.items.every((item, index) => conditionsEqual(item, b.items[index]))
+  );
+};
+
 const applyDeletePolicies = async (
   table: AnyTable,
   where: Condition | undefined,
@@ -284,6 +352,16 @@ export interface ORMAdapter<S extends AnySchema = AnySchema> {
     },
   ) => Promise<void>;
 
+  upsertMany?: (
+    table: AnyTable,
+    v: {
+      target: AnyColumn[];
+      update: AnyColumn[];
+      values: Record<string, unknown>[];
+      where?: Condition;
+    },
+  ) => Promise<void>;
+
   create: (
     table: AnyTable,
     values: Record<string, unknown>,
@@ -367,6 +445,110 @@ export function toORM<S extends AnySchema>(
         ...options,
       });
     },
+    async upsertMany(name, { target, update, values }) {
+      const table = toTable(name);
+      if (values.length === 0) return;
+      if (target.length === 0) {
+        // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- boundary: public query rejects invalid upsert shape
+        throw new Error("[FumaDB] upsertMany requires at least one target column.");
+      }
+      if (update.length === 0) {
+        // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- boundary: public query rejects invalid upsert shape
+        throw new Error("[FumaDB] upsertMany requires at least one update column.");
+      }
+
+      const targetColumns = target.map((columnName) => {
+        const column = table.columns[columnName as string];
+        if (!column) throw new Error(`[FumaDB] unknown column name ${String(columnName)}.`);
+        return column;
+      });
+      const updateColumns = update.map((columnName) => {
+        const column = table.columns[columnName as string];
+        if (!column) throw new Error(`[FumaDB] unknown column name ${String(columnName)}.`);
+        return column;
+      });
+
+      const builder = createBuilder(table.columns);
+      const permittedRows: {
+        readonly value: Record<string, unknown>;
+        readonly where: Condition | undefined;
+      }[] = [];
+      for (const value of values) {
+        const updateValues = Object.fromEntries(
+          updateColumns.map((column) => [column.ormName, value[column.ormName]]),
+        );
+        const constrainedWhere = await applyUpdatePolicies(
+          table,
+          undefined,
+          updateValues,
+          context,
+          "upsert",
+          value,
+        );
+        if (constrainedWhere === false) continue;
+        await runCreatePolicies(table, value, context);
+        permittedRows.push({ value, where: constrainedWhere });
+      }
+      if (permittedRows.length === 0) return;
+
+      if (internal.upsertMany) {
+        const groups: {
+          readonly where: Condition | undefined;
+          readonly values: Record<string, unknown>[];
+        }[] = [];
+        for (const row of permittedRows) {
+          const group = groups.find((candidate) => conditionsEqual(candidate.where, row.where));
+          if (group) {
+            group.values.push(row.value);
+          } else {
+            groups.push({ where: row.where, values: [row.value] });
+          }
+        }
+        const runGroups = async (adapter: ORMAdapter<S>): Promise<void> => {
+          if (!adapter.upsertMany) {
+            // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- boundary: a transaction adapter must mirror the base adapter's upsertMany support
+            throw new Error("[FumaDB] Transaction adapter does not support upsertMany.");
+          }
+          for (const group of groups) {
+            await adapter.upsertMany(table, {
+              target: targetColumns,
+              update: updateColumns,
+              values: group.values,
+              where: group.where,
+            });
+          }
+        };
+
+        // A single group is one adapter call, which is as atomic as the
+        // engine allows. Multiple predicate groups are separate adapter
+        // calls, so run them inside one transaction: a later group's failure
+        // must not leave earlier groups committed.
+        if (groups.length === 1) {
+          await runGroups(internal);
+          return;
+        }
+        await internal.transaction(async (transactionInstance) => {
+          await runGroups(transactionInstance.internal);
+        });
+        return;
+      }
+
+      for (const row of permittedRows) {
+        const value = row.value;
+        const targetWhere = builder.and(
+          ...targetColumns.map((column) => builder(column.ormName, "=", value[column.ormName])),
+        );
+        const where = builder.and(targetWhere, row.where ?? true);
+        if (where === false) continue;
+        await internal.upsert(table, {
+          where: where === true ? undefined : where,
+          update: Object.fromEntries(
+            updateColumns.map((column) => [column.ormName, value[column.ormName]]),
+          ),
+          create: value,
+        });
+      }
+    },
     async create(name, values) {
       const table = toTable(name);
       await runCreatePolicies(table, values, context);
diff --git a/packages/core/fumadb/src/query/orm/predicate-values-equal.test.ts b/packages/core/fumadb/src/query/orm/predicate-values-equal.test.ts
new file mode 100644
index 0000000000..21c5aaff0c
--- /dev/null
+++ b/packages/core/fumadb/src/query/orm/predicate-values-equal.test.ts
@@ -0,0 +1,34 @@
+import { describe, expect, it } from "@effect/vitest";
+
+import { predicateValuesEqual } from "./index";
+
+describe("predicateValuesEqual", () => {
+  it("treats a sparse-array hole as unequal to a present element, in both orders", () => {
+    // oxlint-disable-next-line unicorn/no-new-array -- the sparse array is the case under test
+    expect(predicateValuesEqual(Array(1), [123])).toBe(false);
+    // oxlint-disable-next-line unicorn/no-new-array -- the sparse array is the case under test
+    expect(predicateValuesEqual([123], Array(1))).toBe(false);
+  });
+
+  it("treats arrays with matching holes as equal", () => {
+    // oxlint-disable-next-line unicorn/no-new-array -- the sparse array is the case under test
+    expect(predicateValuesEqual(Array(2), Array(2))).toBe(true);
+  });
+
+  it("keeps structural equality for dense arrays", () => {
+    expect(
+      predicateValuesEqual([BigInt(7), new Date(5), Number.NaN], [BigInt(7), new Date(5), Number.NaN]),
+    ).toBe(true);
+    expect(predicateValuesEqual([1, 2], [1, 3])).toBe(false);
+    expect(predicateValuesEqual([1, undefined], [1, undefined])).toBe(true);
+  });
+
+  it("keeps the existing Date, Uint8Array, and record handling", () => {
+    expect(predicateValuesEqual(new Date(5), new Date(5))).toBe(true);
+    expect(predicateValuesEqual(new Date(5), new Date(6))).toBe(false);
+    expect(predicateValuesEqual(new Uint8Array([1, 2]), new Uint8Array([1, 2]))).toBe(true);
+    expect(predicateValuesEqual(new Uint8Array([1, 2]), new Uint8Array([1, 3]))).toBe(false);
+    expect(predicateValuesEqual({ a: BigInt(1) }, { a: BigInt(1) })).toBe(true);
+    expect(predicateValuesEqual({ a: 1 }, { a: 1, b: 2 })).toBe(false);
+  });
+});
diff --git a/packages/core/fumadb/src/query/table-policy.test.ts b/packages/core/fumadb/src/query/table-policy.test.ts
index 51d0c2b8c5..df3d39c9a7 100644
--- a/packages/core/fumadb/src/query/table-policy.test.ts
+++ b/packages/core/fumadb/src/query/table-policy.test.ts
@@ -8,6 +8,7 @@ import {
   createDrizzleRuntimeSchemaSqlFromTables,
   drizzleAdapter,
 } from "@executor-js/fumadb/adapters/drizzle";
+import { memoryAdapter } from "@executor-js/fumadb/adapters/memory";
 import { withQueryContext, type AbstractQuery } from "@executor-js/fumadb/query";
 import { column, idColumn, schema, table } from "@executor-js/fumadb/schema";
 
@@ -16,6 +17,7 @@ interface TenantPolicyContext {
   readonly deniedTables: ReadonlySet<string>;
   readonly marker: string;
   readonly observed: string[];
+  readonly allowedRegionId?: bigint;
 }
 
 const observe = (context: TenantPolicyContext, event: string) => {
@@ -103,16 +105,68 @@ const comments = table("policy_comments", {
   },
 });
 
+// Policy predicate carries a bigint value; string-keyed grouping of these
+// predicates (e.g. JSON.stringify) throws on bigint.
+const quotas = table("policy_quotas", {
+  id: idColumn("id", "varchar(255)"),
+  region: column("region", "bigint"),
+  label: column("label", "string"),
+}).policy<TenantPolicyContext>({
+  name: "tenant.quotas",
+  onUpdate: ({ builder, context }) => builder("region", "=", context.allowedRegionId ?? BigInt(-1)),
+});
+
+// The update policy predicate depends on the row itself, so rows with
+// different `shard` values compile to different predicates and split one
+// upsertMany call into separate predicate groups.
+const shards = table("policy_shards", {
+  id: idColumn("id", "varchar(255)"),
+  shard: column("shard", "string"),
+  authorId: column("author_id", "varchar(255)"),
+}).policy<TenantPolicyContext>({
+  name: "tenant.shards",
+  onUpdate: ({ builder, set }) => builder("shard", "=", set.shard ?? ""),
+});
+
+// Ten columns, so a row-count-sized batch would bind ten parameters per row.
+const wideRows = table("policy_wide_rows", {
+  id: idColumn("id", "varchar(255)"),
+  c1: column("c1", "string"),
+  c2: column("c2", "string"),
+  c3: column("c3", "string"),
+  c4: column("c4", "string"),
+  c5: column("c5", "string"),
+  c6: column("c6", "string"),
+  c7: column("c7", "string"),
+  c8: column("c8", "string"),
+  c9: column("c9", "string"),
+});
+
+// The physical column name needs identifier quoting; the bulk upsert conflict
+// SET clause references it as `excluded.<name>` and must quote it.
+const quotedNames = table("policy_quoted_names", {
+  id: idColumn("id", "varchar(255)"),
+  displayName: column("display-name", "string"),
+});
+
 const v1 = schema({
   version: "1.0.0",
   tables: {
     authors,
     posts,
     comments,
+    quotas,
+    shards,
+    wideRows,
+    quotedNames,
   },
   relations: {
     authors: ({ many }) => ({
       posts: many("posts"),
+      shards: many("shards"),
+    }),
+    shards: ({ one }) => ({
+      author: one("authors", ["authorId", "id"]).foreignKey(),
     }),
     posts: ({ one, many }) => ({
       author: one("authors", ["authorId", "id"]).foreignKey(),
@@ -142,7 +196,10 @@ const makeContext = (
   observed: [],
 });
 
-const makeHarness = async () => {
+const makeHarness = async (options?: {
+  readonly nativeBatch?: boolean;
+  readonly maxBoundParameters?: number;
+}) => {
   const sqlite = new Database(":memory:");
   sqlite.pragma("foreign_keys = ON");
   const runtimeSchema = createDrizzleRuntimeSchemaFromTables({
@@ -151,7 +208,25 @@ const makeHarness = async () => {
     version: "1.0.0",
     provider: "sqlite",
   });
-  const drizzleDb = drizzle(sqlite, { schema: runtimeSchema });
+  const statements: { readonly sql: string; readonly paramCount: number }[] = [];
+  const drizzleDb = drizzle(sqlite, {
+    schema: runtimeSchema,
+    logger: {
+      logQuery: (query: string, params: unknown[]) => {
+        statements.push({ sql: query, paramCount: params.length });
+      },
+    },
+  });
+  let batchCalls = 0;
+
+  if (options?.nativeBatch) {
+    Object.assign(drizzleDb, {
+      batch: async (queries: readonly { run: () => unknown }[]) => {
+        batchCalls += 1;
+        return sqlite.transaction(() => queries.map((query) => query.run()))();
+      },
+    });
+  }
 
   for (const statement of createDrizzleRuntimeSchemaSqlFromTables({
     tables: v1.tables,
@@ -166,11 +241,16 @@ const makeHarness = async () => {
     drizzleAdapter({
       db: drizzleDb,
       provider: "sqlite",
+      interactiveTransactions: options?.nativeBatch ? false : undefined,
+      maxBoundParameters: options?.maxBoundParameters,
     }),
   );
 
   return {
     orm: client.orm("1.0.0"),
+    getBatchCalls: () => batchCalls,
+    getStatements: (): readonly { readonly sql: string; readonly paramCount: number }[] =>
+      statements,
     close: async () => {
       sqlite.close();
     },
@@ -179,11 +259,45 @@ const makeHarness = async () => {
 
 const useHarness = <A>(run: (orm: TablePolicyQuery) => Promise<A>) =>
   Effect.acquireUseRelease(
-    Effect.promise(makeHarness),
+    Effect.promise(() => makeHarness()),
+    ({ orm }) => Effect.promise(() => run(orm)),
+    ({ close }) => Effect.promise(close),
+  );
+
+const useStatementHarness = <A>(
+  run: (harness: {
+    readonly orm: TablePolicyQuery;
+    readonly getStatements: () => readonly { readonly sql: string; readonly paramCount: number }[];
+  }) => Promise<A>,
+) =>
+  Effect.acquireUseRelease(
+    Effect.promise(() => makeHarness()),
+    (harness) => Effect.promise(() => run(harness)),
+    ({ close }) => Effect.promise(close),
+  );
+
+const useBudgetHarness = <A>(
+  maxBoundParameters: number,
+  run: (orm: TablePolicyQuery) => Promise<A>,
+) =>
+  Effect.acquireUseRelease(
+    Effect.promise(() => makeHarness({ maxBoundParameters })),
     ({ orm }) => Effect.promise(() => run(orm)),
     ({ close }) => Effect.promise(close),
   );
 
+const useNativeBatchHarness = <A>(
+  run: (harness: {
+    readonly orm: TablePolicyQuery;
+    readonly getBatchCalls: () => number;
+  }) => Promise<A>,
+) =>
+  Effect.acquireUseRelease(
+    Effect.promise(() => makeHarness({ nativeBatch: true, maxBoundParameters: 8 })),
+    (harness) => Effect.promise(() => run(harness)),
+    ({ close }) => Effect.promise(close),
+  );
+
 const seedTenants = async (orm: TablePolicyQuery) => {
   const seed = withQueryContext(orm, makeContext(["tenant-a", "tenant-b"], "seed"));
 
@@ -382,6 +496,24 @@ describe("FumaDB table policies", () => {
             title: "A Three",
           },
         });
+        await tenantA.upsertMany("posts", {
+          target: ["id"],
+          update: ["title"],
+          values: [
+            {
+              id: "post-a-1",
+              tenantId: "tenant-a",
+              authorId: "author-a",
+              title: "tenant-a-bulk-upserted",
+            },
+            {
+              id: "post-a-4",
+              tenantId: "tenant-a",
+              authorId: "author-a",
+              title: "A Four",
+            },
+          ],
+        });
 
         await expect(
           tenantA.findMany("posts", {
@@ -391,7 +523,7 @@ describe("FumaDB table policies", () => {
         ).resolves.toEqual([
           {
             id: "post-a-1",
-            title: "tenant-a-updated",
+            title: "tenant-a-bulk-upserted",
           },
           {
             id: "post-a-2",
@@ -401,6 +533,10 @@ describe("FumaDB table policies", () => {
             id: "post-a-3",
             title: "A Three",
           },
+          {
+            id: "post-a-4",
+            title: "A Four",
+          },
         ]);
 
         expect(tenantAContext.observed).toEqual(
@@ -460,6 +596,395 @@ describe("FumaDB table policies", () => {
     }),
   );
 
+  it.effect("rejects invalid bulk upsert conflict shapes", () =>
+    useHarness(async (orm) => {
+      await seedTenants(orm);
+      const tenantA = withQueryContext(orm, makeContext(["tenant-a"], "tenant-a"));
+      const values = [
+        {
+          id: "post-a-bulk-upsert",
+          tenantId: "tenant-a",
+          authorId: "author-a",
+          title: "A bulk upsert",
+        },
+      ];
+
+      await expect(
+        tenantA.upsertMany("posts", {
+          target: [],
+          update: ["title"],
+          values,
+        }),
+      ).rejects.toThrow("[FumaDB] upsertMany requires at least one target column.");
+
+      await expect(
+        tenantA.upsertMany("posts", {
+          target: ["id"],
+          update: [],
+          values,
+        }),
+      ).rejects.toThrow("[FumaDB] upsertMany requires at least one update column.");
+    }),
+  );
+
+  it.effect("keeps every bulk upsert statement inside the bound-variable budget", () =>
+    useStatementHarness(async ({ orm, getStatements }) => {
+      // 250 rows * 10 columns = 2,500 bound variables in a single statement —
+      // over older SQLite's 999-variable cap. The adapter advertises no limit
+      // here, so batching must fall back to the conservative 999 budget.
+      const values = Array.from({ length: 250 }, (_, index) => ({
+        id: `wide-${String(index).padStart(3, "0")}`,
+        c1: `value-${index}-1`,
+        c2: `value-${index}-2`,
+        c3: `value-${index}-3`,
+        c4: `value-${index}-4`,
+        c5: `value-${index}-5`,
+        c6: `value-${index}-6`,
+        c7: `value-${index}-7`,
+        c8: `value-${index}-8`,
+        c9: `value-${index}-9`,
+      }));
+
+      await orm.upsertMany("wideRows", {
+        target: ["id"],
+        update: ["c1"],
+        values,
+      });
+
+      const inserts = getStatements().filter((statement) =>
+        statement.sql.toLowerCase().startsWith("insert into"),
+      );
+      expect(inserts.length).toBeGreaterThanOrEqual(3);
+      for (const statement of inserts) {
+        expect(statement.paramCount).toBeLessThanOrEqual(999);
+      }
+      await expect(orm.count("wideRows")).resolves.toBe(250);
+    }),
+  );
+
+  it.effect("bulk upserts a column whose physical name needs identifier quoting", () =>
+    useHarness(async (orm) => {
+      await orm.createMany("quotedNames", [{ id: "row-1", displayName: "before" }]);
+
+      // The conflict SET clause references the update column through the
+      // `excluded` pseudo-table; an unquoted physical name like
+      // `display-name` is a syntax error there.
+      await orm.upsertMany("quotedNames", {
+        target: ["id"],
+        update: ["displayName"],
+        values: [
+          { id: "row-1", displayName: "after" },
+          { id: "row-2", displayName: "created" },
+        ],
+      });
+
+      await expect(
+        orm.findMany("quotedNames", {
+          select: ["id", "displayName"],
+          orderBy: ["id", "asc"],
+        }),
+      ).resolves.toEqual([
+        { id: "row-1", displayName: "after" },
+        { id: "row-2", displayName: "created" },
+      ]);
+    }),
+  );
+
+  it.effect("keeps a successful sibling nested transaction when the other rolls back", () =>
+    useHarness(async (orm) => {
+      const writer = withQueryContext(orm, makeContext(["tenant-a"], "siblings"));
+
+      await writer.transaction(async (tx) => {
+        // Two sibling savepoint scopes started concurrently on the one shared
+        // SQLite connection: the failing sibling's ROLLBACK TO must not undo
+        // work the successful sibling already released.
+        const failing = tx.transaction(async (inner) => {
+          await inner.create("authors", {
+            id: "author-rolled-back",
+            tenantId: "tenant-a",
+            name: "Rolled Back",
+          });
+          // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- test forces one sibling to roll back
+          throw new Error("sibling failure");
+        });
+        const succeeding = tx.transaction(async (inner) => {
+          await inner.create("authors", {
+            id: "author-committed",
+            tenantId: "tenant-a",
+            name: "Committed",
+          });
+        });
+
+        const [failed, succeeded] = await Promise.allSettled([failing, succeeding]);
+        expect(failed.status).toBe("rejected");
+        expect(succeeded.status).toBe("fulfilled");
+      });
+
+      await expect(
+        writer.findMany("authors", {
+          select: ["id"],
+          where: (builder) =>
+            builder.or(
+              builder("id", "=", "author-rolled-back"),
+              builder("id", "=", "author-committed"),
+            ),
+        }),
+      ).resolves.toEqual([{ id: "author-committed" }]);
+    }),
+  );
+
+  it.effect("bulk upserts through a policy predicate that contains a bigint", () =>
+    useHarness(async (orm) => {
+      const region = withQueryContext(orm, {
+        ...makeContext(["tenant-a"], "region"),
+        allowedRegionId: BigInt(7),
+      });
+
+      await region.createMany("quotas", [
+        { id: "quota-in-region", region: BigInt(7), label: "before" },
+        { id: "quota-out-of-region", region: BigInt(9), label: "before" },
+      ]);
+
+      await region.upsertMany("quotas", {
+        target: ["id"],
+        update: ["label"],
+        values: [
+          { id: "quota-in-region", region: BigInt(7), label: "after" },
+          { id: "quota-out-of-region", region: BigInt(9), label: "after" },
+          { id: "quota-created", region: BigInt(7), label: "created" },
+        ],
+      });
+
+      await expect(
+        region.findMany("quotas", {
+          select: ["id", "label"],
+          orderBy: ["id", "asc"],
+        }),
+      ).resolves.toEqual([
+        { id: "quota-created", label: "created" },
+        { id: "quota-in-region", label: "after" },
+        { id: "quota-out-of-region", label: "before" },
+      ]);
+    }),
+  );
+
+  it.effect("skips policy-excluded conflicts identically on drizzle and memory adapters", () => {
+    // `post-b-1` already exists for tenant-b, so its unique target conflicts,
+    // and the tenant-a update policy excludes the conflicting row. SQL detects
+    // the conflict first and the predicate only gates the update, so the row
+    // is skipped — never duplicated.
+    const upsertAcrossPolicyExcludedConflict = async (orm: TablePolicyQuery) => {
+      await seedTenants(orm);
+      const tenantA = withQueryContext(orm, makeContext(["tenant-a"], "tenant-a"));
+      await tenantA.upsertMany("posts", {
+        target: ["id"],
+        update: ["title"],
+        values: [
+          {
+            id: "post-b-1",
+            tenantId: "tenant-a",
+            authorId: "author-a",
+            title: "hijack attempt",
+          },
+          {
+            id: "post-a-9",
+            tenantId: "tenant-a",
+            authorId: "author-a",
+            title: "A Nine",
+          },
+        ],
+      });
+      const allTenants = withQueryContext(orm, makeContext(["tenant-a", "tenant-b"], "all"));
+      return allTenants.findMany("posts", {
+        select: ["id", "tenantId", "title"],
+        orderBy: ["id", "asc"],
+      });
+    };
+
+    const expected = [
+      { id: "post-a-1", tenantId: "tenant-a", title: "A One" },
+      { id: "post-a-2", tenantId: "tenant-a", title: "A Two" },
+      { id: "post-a-9", tenantId: "tenant-a", title: "A Nine" },
+      { id: "post-b-1", tenantId: "tenant-b", title: "B One" },
+    ];
+
+    return useHarness(async (orm) => {
+      await expect(upsertAcrossPolicyExcludedConflict(orm)).resolves.toEqual(expected);
+
+      const memoryOrm = tablePolicyDB.client(memoryAdapter()).orm("1.0.0");
+      await expect(upsertAcrossPolicyExcludedConflict(memoryOrm)).resolves.toEqual(expected);
+    });
+  });
+
+  it.effect("rolls back every bounded upsert statement when a native batch fails", () =>
+    useNativeBatchHarness(async ({ orm, getBatchCalls }) => {
+      await seedTenants(orm);
+      const tenantA = withQueryContext(orm, makeContext(["tenant-a"], "tenant-a"));
+
+      await expect(
+        tenantA.upsertMany("posts", {
+          target: ["id"],
+          update: ["title"],
+          values: [
+            {
+              id: "post-a-batch-1",
+              tenantId: "tenant-a",
+              authorId: "author-a",
+              title: "A batch one",
+            },
+            {
+              id: "post-a-batch-2",
+              tenantId: "tenant-a",
+              authorId: "author-a",
+              title: "A batch two",
+            },
+            {
+              id: "post-a-batch-invalid",
+              tenantId: "tenant-a",
+              authorId: "missing-author",
+              title: "Must roll back",
+            },
+          ],
+        }),
+      ).rejects.toThrow();
+
+      expect(getBatchCalls()).toBe(1);
+      await expect(
+        tenantA.findMany("posts", {
+          where: (builder) => builder("id", "starts with", "post-a-batch-"),
+          select: ["id"],
+        }),
+      ).resolves.toEqual([]);
+    }),
+  );
+
+  it.effect("rolls back earlier predicate groups when a later group fails", () =>
+    useHarness(async (orm) => {
+      await seedTenants(orm);
+      const writer = withQueryContext(orm, makeContext(["tenant-a"], "groups"));
+
+      // Two distinct per-row predicates split this call into two internal
+      // groups. The second group's row references a missing author, so its
+      // insert violates the foreign key — the first group's rows must roll
+      // back with it instead of staying committed.
+      await expect(
+        writer.upsertMany("shards", {
+          target: ["id"],
+          update: ["shard"],
+          values: [
+            { id: "shard-a-1", shard: "a", authorId: "author-a" },
+            { id: "shard-b-1", shard: "b", authorId: "missing-author" },
+          ],
+        }),
+      ).rejects.toThrow();
+
+      await expect(orm.count("shards")).resolves.toBe(0);
+    }),
+  );
+
+  it.effect("rolls back earlier createMany batches when a later batch fails", () =>
+    useBudgetHarness(8, async (orm) => {
+      await seedTenants(orm);
+      const tenantA = withQueryContext(orm, makeContext(["tenant-a"], "tenant-a"));
+
+      // Four columns per row against an 8-parameter budget forces two-row
+      // batches, so this insert runs as two statements. The last row reuses
+      // the seeded `post-a-1` id, so the second statement violates the
+      // primary key — the first statement's rows must roll back with it.
+      await expect(
+        tenantA.createMany("posts", [
+          { id: "post-a-n1", tenantId: "tenant-a", authorId: "author-a", title: "N1" },
+          { id: "post-a-n2", tenantId: "tenant-a", authorId: "author-a", title: "N2" },
+          { id: "post-a-n3", tenantId: "tenant-a", authorId: "author-a", title: "N3" },
+          { id: "post-a-1", tenantId: "tenant-a", authorId: "author-a", title: "Duplicate" },
+        ]),
+      ).rejects.toThrow();
+
+      await expect(
+        tenantA.findMany("posts", {
+          where: (builder) => builder("id", "starts with", "post-a-n"),
+          select: ["id"],
+        }),
+      ).resolves.toEqual([]);
+    }),
+  );
+
+  it.effect("fails fast when a single row exceeds the bound-parameter budget", () =>
+    useBudgetHarness(8, async (orm) => {
+      await expect(
+        orm.createMany("wideRows", [
+          {
+            id: "wide-overflow",
+            c1: "1",
+            c2: "2",
+            c3: "3",
+            c4: "4",
+            c5: "5",
+            c6: "6",
+            c7: "7",
+            c8: "8",
+            c9: "9",
+          },
+        ]),
+      ).rejects.toThrow(
+        'one row binds 10 bound parameters, which exceeds the 8-parameter budget',
+      );
+
+      await expect(orm.count("wideRows")).resolves.toBe(0);
+    }),
+  );
+
+  it.effect("fails fast when reserved predicate parameters consume the budget", () =>
+    useBudgetHarness(4, async (orm) => {
+      const seed = withQueryContext(orm, makeContext(["tenant-a"], "seed"));
+      await seed.createMany("authors", [{ id: "author-a", tenantId: "tenant-a", name: "Ada" }]);
+
+      const tenantA = withQueryContext(orm, makeContext(["tenant-a"], "tenant-a"));
+      // A posts row alone fits the 4-parameter budget exactly, but the update
+      // policy predicate reserves one more bound parameter per statement.
+      await expect(
+        tenantA.upsertMany("posts", {
+          target: ["id"],
+          update: ["title"],
+          values: [{ id: "post-a-r1", tenantId: "tenant-a", authorId: "author-a", title: "R1" }],
+        }),
+      ).rejects.toThrow("the predicate reserves 1 more");
+
+      await expect(tenantA.count("posts")).resolves.toBe(0);
+    }),
+  );
+
+  it.effect("applies intra-call conflicts identically on drizzle and memory adapters", () => {
+    // The second value's unique target conflicts with a row created by the
+    // first value in the SAME upsertMany call: both adapters must update the
+    // freshly created row instead of duplicating or dropping it.
+    const upsertIntraCallConflict = async (orm: TablePolicyQuery) => {
+      await seedTenants(orm);
+      const tenantA = withQueryContext(orm, makeContext(["tenant-a"], "tenant-a"));
+      await tenantA.upsertMany("posts", {
+        target: ["id"],
+        update: ["title"],
+        values: [
+          { id: "post-a-dup", tenantId: "tenant-a", authorId: "author-a", title: "first write" },
+          { id: "post-a-dup", tenantId: "tenant-a", authorId: "author-a", title: "second write" },
+        ],
+      });
+      return tenantA.findMany("posts", {
+        where: (builder) => builder("id", "=", "post-a-dup"),
+        select: ["id", "title"],
+      });
+    };
+
+    const expected = [{ id: "post-a-dup", title: "second write" }];
+
+    return useHarness(async (orm) => {
+      await expect(upsertIntraCallConflict(orm)).resolves.toEqual(expected);
+
+      const memoryOrm = tablePolicyDB.client(memoryAdapter()).orm("1.0.0");
+      await expect(upsertIntraCallConflict(memoryOrm)).resolves.toEqual(expected);
+    });
+  });
+
   it.effect("fails closed when a query wrapper does not forward context rebinding", () =>
     useHarness(async (orm) => {
       const wrapped = { ...orm };
@@ -471,7 +996,7 @@ describe("FumaDB table policies", () => {
   );
 
   it.effect(
-    "rejects out-of-context writes across createMany, updateMany, upsert, and transactions",
+    "rejects out-of-context writes across createMany, updateMany, upsert, upsertMany, and transactions",
     () =>
       useHarness(async (orm) => {
         await seedTenants(orm);
@@ -524,6 +1049,47 @@ describe("FumaDB table policies", () => {
           }),
         ).rejects.toThrow("tenant tenant-b is not allowed for posts");
 
+        await expect(
+          tenantA.upsertMany("posts", {
+            target: ["id"],
+            update: ["title"],
+            values: [
+              {
+                id: "post-a-bulk-upsert",
+                tenantId: "tenant-a",
+                authorId: "author-a",
+                title: "A bulk upsert",
+              },
+              {
+                id: "post-b-bulk-upsert",
+                tenantId: "tenant-b",
+                authorId: "author-b",
+                title: "B bulk upsert",
+              },
+            ],
+          }),
+        ).rejects.toThrow("tenant tenant-b is not allowed for posts");
+        await expect(
+          tenantA.findFirst("posts", {
+            where: (builder) => builder("id", "=", "post-a-bulk-upsert"),
+          }),
+        ).resolves.toBeNull();
+
+        await expect(
+          tenantA.upsertMany("posts", {
+            target: ["id"],
+            update: ["tenantId"],
+            values: [
+              {
+                id: "post-a-1",
+                tenantId: "tenant-b",
+                authorId: "author-b",
+                title: "tenant move",
+              },
+            ],
+          }),
+        ).rejects.toThrow("tenant tenant-b is not allowed for posts");
+
         await expect(
           tenantA.transaction(async (tx) => {
             await tx.create("posts", {
diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts
index d6a2b1c4ba..02e44a68dd 100644
--- a/packages/core/sdk/src/executor.ts
+++ b/packages/core/sdk/src/executor.ts
@@ -196,7 +196,6 @@ import { makeShapeMemory, observedShapeToJsonSchema, SHAPE_MEMORY_PLUGIN_ID } fr
 import { isUnauthorizedToolFailure } from "./auth-tool-failure";
 
 const PLUGIN_STORAGE_DELETE_KEY_BATCH_SIZE = 90;
-const PLUGIN_STORAGE_CREATE_ROW_BATCH_SIZE = 90;
 const MAX_APPROVAL_ARGUMENT_PREVIEW_CHARS = 4_000;
 
 // ---------------------------------------------------------------------------
@@ -1075,6 +1074,14 @@ type LooseStorageDb = {
     tableName: string,
     rows: readonly Record<string, unknown>[],
   ) => Promise<readonly unknown[]>;
+  readonly upsertMany: (
+    tableName: string,
+    options: {
+      readonly target: readonly string[];
+      readonly update: readonly string[];
+      readonly values: readonly Record<string, unknown>[];
+    },
+  ) => Promise<void>;
   readonly deleteMany: (tableName: string, options?: unknown) => Promise<void>;
   readonly findFirst: (
     tableName: string,
@@ -1090,6 +1097,8 @@ type LooseStorageDb = {
 const asLooseStorageDb = (db: unknown): LooseStorageDb => db as LooseStorageDb;
 
 const makeCoreDb = (fuma: ReturnType<typeof makeFumaClient>) => ({
+  transaction: <A, E>(effect: Effect.Effect<A, E>): Effect.Effect<A, E | StorageFailure> =>
+    fuma.transaction(effect),
   count: <TName extends CoreTableName>(
     tableName: TName,
     options?: { readonly where?: CoreWhere },
@@ -1111,6 +1120,19 @@ const makeCoreDb = (fuma: ReturnType<typeof makeFumaClient>) => ({
       : fuma
           .use(`${tableName}.createMany`, (db) => asLooseStorageDb(db).createMany(tableName, rows))
           .pipe(Effect.asVoid),
+  upsertMany: <TName extends CoreTableName>(
+    tableName: TName,
+    options: {
+      readonly target: readonly string[];
+      readonly update: readonly string[];
+      readonly values: readonly Record<string, unknown>[];
+    },
+  ): Effect.Effect<void, StorageFailure> =>
+    options.values.length === 0
+      ? Effect.void
+      : fuma.use(`${tableName}.upsertMany`, (db) =>
+          asLooseStorageDb(db).upsertMany(tableName, options),
+        ),
   deleteMany: <TName extends CoreTableName>(
     tableName: TName,
     options: { readonly where?: CoreWhere } = {},
@@ -1439,42 +1461,33 @@ const makePluginStorageFacade = (input: {
       readonly data: unknown;
     }[],
   ) =>
-    Effect.gen(function* () {
-      const os = ownerSubject(owner);
-      if (!os) {
-        return yield* new StorageError({
-          message: `Cannot write plugin storage for owner "user": executor has no subject.`,
-          cause: undefined,
-        });
-      }
-      const entriesById = new Map(
-        entries.map((entry) => [
-          pluginStorageId({
-            pluginId: input.pluginId,
-            collection: entry.collection,
-            key: entry.key,
-          }),
-          entry,
-        ]),
-      );
-      const uniqueEntries = [...entriesById.values()];
-      if (uniqueEntries.length === 0) return;
-
-      yield* deleteManyImpl(owner, os.subject, uniqueEntries);
-
-      const now = new Date();
-      for (
-        let offset = 0;
-        offset < uniqueEntries.length;
-        offset += PLUGIN_STORAGE_CREATE_ROW_BATCH_SIZE
-      ) {
-        const batchEntries = uniqueEntries.slice(
-          offset,
-          offset + PLUGIN_STORAGE_CREATE_ROW_BATCH_SIZE,
+    input.core.transaction(
+      Effect.gen(function* () {
+        const os = ownerSubject(owner);
+        if (!os) {
+          return yield* new StorageError({
+            message: `Cannot write plugin storage for owner "user": executor has no subject.`,
+            cause: undefined,
+          });
+        }
+        const entriesById = new Map(
+          entries.map((entry) => [
+            pluginStorageId({
+              pluginId: input.pluginId,
+              collection: entry.collection,
+              key: entry.key,
+            }),
+            entry,
+          ]),
         );
-        yield* input.core.createMany(
-          "plugin_storage",
-          batchEntries.map((entry) => ({
+        const uniqueEntries = [...entriesById.values()];
+        if (uniqueEntries.length === 0) return;
+
+        const now = new Date();
+        yield* input.core.upsertMany("plugin_storage", {
+          target: ["tenant", "owner", "subject", "plugin_id", "collection", "key"],
+          update: ["data", "updated_at"],
+          values: uniqueEntries.map((entry) => ({
             tenant,
             owner: os.owner,
             subject: os.subject,
@@ -1485,9 +1498,9 @@ const makePluginStorageFacade = (input: {
             created_at: now,
             updated_at: now,
           })),
-        );
-      }
-    });
+        });
+      }),
+    );
 
   const removeManyImpl = (
     owner: Owner,
diff --git a/packages/core/sdk/src/fuma-runtime.ts b/packages/core/sdk/src/fuma-runtime.ts
index ae5c4ebcc5..19852d48d3 100644
--- a/packages/core/sdk/src/fuma-runtime.ts
+++ b/packages/core/sdk/src/fuma-runtime.ts
@@ -285,6 +285,7 @@ const makeSafeFumaQuery = <TSchema extends AnySchema>(
       db.transaction((transactionDb) => run(makeSafeFumaQuery(transactionDb, options))),
     updateMany: (name, value) => db.updateMany(table(name), value),
     upsert: (name, value) => db.upsert(table(name), value),
+    upsertMany: (name, value) => db.upsertMany(table(name), value),
   };
 
   return Object.freeze(query);
diff --git a/packages/core/sdk/src/plugin-storage.test.ts b/packages/core/sdk/src/plugin-storage.test.ts
index 95c7c0b714..3cc1d0793e 100644
--- a/packages/core/sdk/src/plugin-storage.test.ts
+++ b/packages/core/sdk/src/plugin-storage.test.ts
@@ -1,7 +1,8 @@
 import { describe, expect, it } from "@effect/vitest";
 import { Cause, Effect, Exit, Schema } from "effect";
 
-import { StorageError } from "./fuma-runtime";
+import { createExecutor } from "./executor";
+import { StorageError, type FumaDb } from "./fuma-runtime";
 import { Owner } from "./ids";
 import { definePlugin } from "./plugin";
 import {
@@ -10,7 +11,7 @@ import {
   type PluginStorageCollectionQueryInput,
   type PluginStorageCollectionWhere,
 } from "./plugin-storage";
-import { makeTestExecutor } from "./testing";
+import { makeTestConfig, makeTestExecutor } from "./testing";
 
 const ToolCall = Schema.Struct({
   runId: Schema.String,
@@ -113,6 +114,55 @@ const call = (input: {
   durationMs: input.durationMs ?? 0,
 });
 
+// A FumaDB that commits the FIRST row of a multi-row `plugin_storage` bulk
+// write and then fails. Injecting the fault mid-write, rather than before it,
+// is what makes the two rollback cases below meaningful: the row is really on
+// disk when the failure lands, so only the enclosing transaction can take it
+// back. The fault is unconditional (not armed by entering a transaction) so
+// that dropping the transaction is a visible failure and not a silently
+// disarmed test.
+const failPluginStorageBulkWriteAfterFirstRow = (db: FumaDb): FumaDb => {
+  const wrap = (source: FumaDb): FumaDb =>
+    new Proxy(source, {
+      get(target, property, receiver) {
+        if (property === "withContext") {
+          const withContext = target.withContext;
+          return withContext === undefined
+            ? undefined
+            : (context: unknown) => wrap(withContext(context));
+        }
+        if (property === "transaction") {
+          const transaction: FumaDb["transaction"] = (run) =>
+            target.transaction((transactionDb) => run(wrap(transactionDb)));
+          return transaction;
+        }
+        if (property === "upsertMany") {
+          const upsertMany: FumaDb["upsertMany"] = async (table, options) => {
+            if (table !== "plugin_storage" || options.values.length < 2) {
+              return target.upsertMany(table, options);
+            }
+
+            await target.upsertMany(table, {
+              ...options,
+              values: options.values.slice(0, 1),
+            });
+            // oxlint-disable-next-line executor/no-promise-reject -- boundary: fault-injecting FumaDB adapter must reject to exercise transaction rollback
+            return Promise.reject(
+              new StorageError({
+                message: "Injected plugin storage bulk-write failure.",
+                cause: undefined,
+              }),
+            );
+          };
+          return upsertMany;
+        }
+        return Reflect.get(target, property, receiver);
+      },
+    });
+
+  return wrap(db);
+};
+
 describe("plugin storage collections", () => {
   it.effect("queries declared indexes through the executor's SQLite FumaDB target", () =>
     Effect.gen(function* () {
@@ -222,6 +272,164 @@ describe("plugin storage collections", () => {
     }),
   );
 
+  it.effect("stores and overwrites every row when a bulk write spans multiple batches", () =>
+    Effect.gen(function* () {
+      const executor = yield* makeTestExecutor({
+        backend: "sqlite",
+        plugins: [executionHistoryPlugin] as const,
+      });
+      // A plugin_storage row binds ~9 values, so 300 rows exceed one
+      // 999-bound-variable statement budget and must span several batches.
+      const entries = (status: "ok" | "failed") =>
+        Array.from({ length: 300 }, (_, index) => ({
+          key: `batched-call-${String(index).padStart(3, "0")}`,
+          data: call({
+            runId: "run-batched",
+            toolId: "browser",
+            status,
+            startedAt: new Date(Date.UTC(2026, 4, 29, 12, 0, index)).toISOString(),
+          }),
+        }));
+
+      yield* executor.executionHistory.recordMany("org", entries("ok"));
+      const total = yield* executor.executionHistory.count({
+        where: { runId: "run-batched" },
+      });
+      expect(total).toBe(300);
+
+      yield* executor.executionHistory.recordMany("org", entries("failed"));
+      const failed = yield* executor.executionHistory.count({
+        where: { runId: "run-batched", status: "failed" },
+      });
+      expect(failed).toBe(300);
+      const totalAfterOverwrite = yield* executor.executionHistory.count({
+        where: { runId: "run-batched" },
+      });
+      expect(totalAfterOverwrite).toBe(300);
+    }),
+  );
+
+  it.effect("rolls back every plugin storage row when a bulk write fails", () =>
+    Effect.gen(function* () {
+      const config = makeTestConfig({
+        backend: "sqlite",
+        plugins: [executionHistoryPlugin] as const,
+      });
+      const executor = yield* Effect.acquireRelease(
+        createExecutor({
+          ...config,
+          db: failPluginStorageBulkWriteAfterFirstRow(config.db),
+        }),
+        (instance) =>
+          instance
+            .close()
+            .pipe(
+              Effect.ignore,
+              Effect.andThen(Effect.promise(() => config.testDb.close()).pipe(Effect.ignore)),
+            ),
+      );
+
+      const exit = yield* Effect.exit(
+        executor.executionHistory.recordMany("org", [
+          {
+            key: "call-first",
+            data: call({
+              runId: "run-rollback",
+              toolId: "browser",
+              status: "ok",
+              startedAt: "2026-05-29T12:00:00.000Z",
+            }),
+          },
+          {
+            key: "call-second",
+            data: call({
+              runId: "run-rollback",
+              toolId: "shell",
+              status: "ok",
+              startedAt: "2026-05-29T12:01:00.000Z",
+            }),
+          },
+        ]),
+      );
+      expect(Exit.isFailure(exit)).toBe(true);
+
+      const stored = yield* executor.executionHistory.query({
+        where: { runId: "run-rollback" },
+      });
+      expect(stored).toEqual([]);
+    }),
+  );
+
+  // The hazard this whole change exists to remove: the previous implementation
+  // deleted every target key and only then re-created the rows, so a failure
+  // between the two halves destroyed data the caller never meant to touch. An
+  // upsert inside a transaction cannot lose a row it did not successfully
+  // replace, so the ORIGINAL values must still be readable after the failure —
+  // not merely absent-and-consistent like the rolled-back insert above.
+  it.effect("leaves pre-existing rows intact when a bulk overwrite fails mid-batch", () =>
+    Effect.gen(function* () {
+      const config = makeTestConfig({
+        backend: "sqlite",
+        plugins: [executionHistoryPlugin] as const,
+      });
+      const executor = yield* Effect.acquireRelease(
+        createExecutor({
+          ...config,
+          db: failPluginStorageBulkWriteAfterFirstRow(config.db),
+        }),
+        (instance) =>
+          instance
+            .close()
+            .pipe(
+              Effect.ignore,
+              Effect.andThen(Effect.promise(() => config.testDb.close()).pipe(Effect.ignore)),
+            ),
+      );
+
+      // Seeded one row at a time, so the seeding itself never goes through the
+      // bulk path the fault injector breaks.
+      const original = [
+        {
+          key: "call-first",
+          data: call({
+            runId: "run-preexisting",
+            toolId: "browser",
+            status: "ok",
+            startedAt: "2026-05-29T12:00:00.000Z",
+          }),
+        },
+        {
+          key: "call-second",
+          data: call({
+            runId: "run-preexisting",
+            toolId: "shell",
+            status: "ok",
+            startedAt: "2026-05-29T12:01:00.000Z",
+          }),
+        },
+      ];
+      for (const row of original) {
+        yield* executor.executionHistory.record("org", row.key, row.data);
+      }
+
+      const exit = yield* Effect.exit(
+        executor.executionHistory.recordMany(
+          "org",
+          original.map((row) => ({
+            key: row.key,
+            data: { ...row.data, toolId: "overwritten", status: "failed" as const },
+          })),
+        ),
+      );
+      expect(Exit.isFailure(exit)).toBe(true);
+
+      const first = yield* executor.executionHistory.get("call-first");
+      const second = yield* executor.executionHistory.get("call-second");
+      expect(first?.data).toEqual(original[0]!.data);
+      expect(second?.data).toEqual(original[1]!.data);
+    }),
+  );
+
   it.effect(
     "bulk puts large plugin storage row sets in bounded batches",
     () =>
diff --git a/packages/core/sdk/src/test-config.ts b/packages/core/sdk/src/test-config.ts
index f452b94db4..cbb64e1dad 100644
--- a/packages/core/sdk/src/test-config.ts
+++ b/packages/core/sdk/src/test-config.ts
@@ -53,6 +53,14 @@ const makeLazyTestFumaDb = (options: {
     transaction: async (run) => (await start()).db.internal.transaction(run),
     updateMany: async (table, value) => (await start()).db.internal.updateMany(table, value),
     upsert: async (table, value) => (await start()).db.internal.upsert(table, value),
+    upsertMany: async (table, value) => {
+      const actual = await start();
+      if (!actual.db.internal.upsertMany) {
+        // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- boundary: lazy test DB must expose the current FumaDB adapter surface
+        throw new Error("[FumaDB] upsertMany is not supported by this adapter.");
+      }
+      return actual.db.internal.upsertMany(table, value);
+    },
   };
 
   const queryMethods = new Set<PropertyKey>([
@@ -65,6 +73,7 @@ const makeLazyTestFumaDb = (options: {
     "transaction",
     "updateMany",
     "upsert",
+    "upsertMany",
   ]);
 
   const makeDb = (context?: ExecutorOwnerPolicyContext): FumaDb =>