fix(sdk): transact bulk plugin storage writes

Author: aryasaatvik

packages/core/sdk/src/executor.ts

@@ -996,6 +996,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 },
@@ -1358,44 +1360,46 @@ 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,
+    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,
+          ]),
+        );
+        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,
+            plugin_id: input.pluginId,
             collection: entry.collection,
             key: entry.key,
-          }),
-          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,
-          plugin_id: input.pluginId,
-          collection: entry.collection,
-          key: entry.key,
-          data: entry.data,
-          created_at: now,
-          updated_at: now,
-        })),
-      });
-    });
+            data: entry.data,
+            created_at: now,
+            updated_at: now,
+          })),
+        });
+      }),
+    );
 
   const removeManyImpl = (
     owner: Owner,

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,
@@ -109,6 +110,48 @@ const call = (input: {
   durationMs: input.durationMs ?? 0,
 });
 
+const failPluginStorageBulkWriteAfterFirstRow = (db: FumaDb): FumaDb => {
+  const wrap = (source: FumaDb, failBulkWrite: boolean): FumaDb =>
+    new Proxy(source, {
+      get(target, property, receiver) {
+        if (property === "withContext") {
+          const withContext = target.withContext;
+          return withContext === undefined
+            ? undefined
+            : (context: unknown) => wrap(withContext(context), failBulkWrite);
+        }
+        if (property === "transaction") {
+          const transaction: FumaDb["transaction"] = (run) =>
+            target.transaction((transactionDb) => run(wrap(transactionDb, true)));
+          return transaction;
+        }
+        if (property === "upsertMany" && failBulkWrite) {
+          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, false);
+};
+
 describe("plugin storage collections", () => {
   it.effect("queries declared indexes through the executor's SQLite FumaDB target", () =>
     Effect.gen(function* () {
@@ -218,6 +261,57 @@ describe("plugin storage collections", () => {
     }),
   );
 
+  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([]);
+    }),
+  );
+
   it.effect(
     "bulk puts large plugin storage row sets in bounded batches",
     () =>