diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md
index b31c208..b65b4cf 100644
--- a/docs/ARCHITECTURE.md
+++ b/docs/ARCHITECTURE.md
@@ -57,10 +57,12 @@ place, in one language.
 ### `Driver` vs `Database` vs `Connection`
 
 - `Driver` is the low-level connection layer (`PgDriver`, `PgliteDriver`,
-  `SqliteDriver`, `DoSqliteDriver`). It exposes `execute(sql)`,
-  `runInSingleConnection(fn)`, `close()`, and a `dialect`. Each lives at its
-  own entry point (`typegres/drivers/*`) so a bundle only ever resolves the
-  optional peer it actually imports.
+  `SqliteDriver`, `DoSqliteDriver`, `OracleDriver`). It exposes `execute(sql)`,
+  `runInTransaction(options, fn)`, `close()`, and a `dialect`. Each driver owns
+  connection pinning plus its database's transaction protocol and supplies the
+  transaction-bound executor to `fn`. Drivers live at separate entry points
+  (`typegres/drivers/*`) so a bundle only ever resolves the optional peer it
+  actually imports.
 - `Database` is the schema handle: provenance identity and the `Table`
   factory, no driver of its own. `typegres()` constructs one synchronously,
   so table classes can be declared at module load without a top-level await.
diff --git a/src/database.ts b/src/database.ts
index 30f86b3..f336516 100644
--- a/src/database.ts
+++ b/src/database.ts
@@ -1,9 +1,15 @@
-import { type Driver, isSyncDriver, type QueryResult } from "./drivers/types";
+import {
+  type Driver,
+  isSyncDriver,
+  type QueryResult,
+  type TransactionIsolation,
+  type TransactionOptions,
+} from "./drivers/types";
 import type { Fromable, RowType, RowTypeToTsType } from "./builder/query";
 import { QueryBuilder, hydrateRows } from "./builder/query";
 import { deserializeRows } from "./util";
 import type { Sql } from "./builder/sql";
-import { compile, sql, Ident } from "./builder/sql";
+import { Ident } from "./builder/sql";
 import { Table, type TableBase, type TableOptions } from "./table";
 import { Values } from "./builder/values";
 import { InsertBuilder } from "./builder/insert";
@@ -15,10 +21,7 @@ import { PgExecutor } from "./live/pg/executor";
 import { StatementExecutor, type Executor } from "./executor";
 import type { DialectName } from "./builder/sql";
 
-export type TransactionIsolation = "read committed" | "repeatable read" | "serializable";
-export type TransactionOptions = {
-  isolation?: TransactionIsolation;
-};
+export type { TransactionIsolation, TransactionOptions } from "./drivers/types";
 
 // Postgres isolation levels are totally ordered. A nested call asking for
 // weaker-or-equal isolation than the active txn flattens harmlessly (caller
@@ -30,10 +33,10 @@ export type TransactionOptions = {
 // `default_transaction_isolation`). We can't prove what level we got, so
 // any *explicit* nested request inside an ambient txn must throw — the
 // alternative would silently downgrade the caller's expectation.
-const ISOLATION: { [K in TransactionIsolation]: { rank: number; begin: Sql } } = {
-  "read committed": { rank: 0, begin: sql`BEGIN ISOLATION LEVEL READ COMMITTED` },
-  "repeatable read": { rank: 1, begin: sql`BEGIN ISOLATION LEVEL REPEATABLE READ` },
-  "serializable":    { rank: 2, begin: sql`BEGIN ISOLATION LEVEL SERIALIZABLE` },
+const ISOLATION: { [K in TransactionIsolation]: { rank: number } } = {
+  "read committed": { rank: 0 },
+  "repeatable read": { rank: 1 },
+  "serializable": { rank: 2 },
 };
 
 // Provenance identity, no driver and no dialect of its own. Construction
@@ -323,61 +326,38 @@ export class Connection<C = undefined> {
       }
       return fn(this);
     }
+    const driver = this.driver;
     const bus = this.#bus;
-    return this.driver.runInSingleConnection(async (execute) => {
-      const driver = this.driver;
-      let txExecutor: Executor;
-      if (this.database.dialect === "postgres") {
-        txExecutor = new PgExecutor(this.database, execute, true);
-      } else if (this.database.dialect === "sqlite") {
-        if (!isSyncDriver(driver)) {
-          throw new Error("unreachable: sqlite Connection without a SyncDriver");
-        }
-        // Bound and pooled are the same channel on sqlite's one handle —
-        // checked, not assumed; the bound executor differs only in event
-        // timing (commit-deferred flush).
-        if (execute !== driver.executeSync) {
-          throw new Error(
-            "sync driver must pass its executeSync to runInSingleConnection — one handle, one channel",
-          );
-        }
-        if (!bus) {
-          throw new Error("sqlite Connection is missing its live bus");
-        }
-        txExecutor = new SqliteLiveExecutor(this.database, driver, bus, true);
-      } else {
-        txExecutor = new StatementExecutor(this.database, execute, true);
-      }
-      const tx = new Connection<C>(this.database, this.driver, txExecutor, opts?.isolation);
-      // Drivers with a native transaction protocol (Durable Objects) own
-      // commit/rollback; everyone else gets BEGIN/COMMIT/ROLLBACK SQL.
-      if (driver.runInTransaction) {
-        try {
-          const result = await driver.runInTransaction(() => fn(tx));
-          txExecutor.onCommit();
-          return result;
-        } catch (e) {
-          txExecutor.onRollback();
-          throw e;
-        }
-      }
-      const runSql = async (s: Sql) => execute(compile(s, { database: this.database }));
-      await runSql(opts?.isolation ? ISOLATION[opts.isolation].begin : sql`BEGIN`);
-      try {
-        const result = await fn(tx);
-        await runSql(sql`COMMIT`);
-        txExecutor.onCommit();
-        return result;
-      } catch (e) {
-        try {
-          await runSql(sql`ROLLBACK`);
-        } catch (rollbackErr) {
-          console.error("ROLLBACK failed after transaction error:", rollbackErr);
+    let txExecutor: Executor | undefined;
+    try {
+      const result = await driver.runInTransaction(opts ?? {}, async (execute) => {
+        if (this.database.dialect === "postgres") {
+          txExecutor = new PgExecutor(this.database, execute, true);
+        } else if (this.database.dialect === "sqlite") {
+          if (!isSyncDriver(driver)) {
+            throw new Error("unreachable: sqlite Connection without a SyncDriver");
+          }
+          if (execute !== driver.executeSync) {
+            throw new Error(
+              "sync driver must pass its executeSync to the transaction — one handle, one channel",
+            );
+          }
+          if (!bus) {
+            throw new Error("sqlite Connection is missing its live bus");
+          }
+          txExecutor = new SqliteLiveExecutor(this.database, driver, bus, true);
+        } else {
+          txExecutor = new StatementExecutor(this.database, execute, true);
         }
-        txExecutor.onRollback();
-        throw e;
-      }
-    });
+        const tx = new Connection<C>(this.database, driver, txExecutor, opts?.isolation);
+        return fn(tx);
+      });
+      txExecutor?.onCommit();
+      return result;
+    } catch (e) {
+      txExecutor?.onRollback();
+      throw e;
+    }
   }
 
   async close(): Promise<void> {
diff --git a/src/drivers/do.ts b/src/drivers/do.ts
index 4550aa0..eabf70e 100644
--- a/src/drivers/do.ts
+++ b/src/drivers/do.ts
@@ -1,5 +1,5 @@
 import type { CompiledSql } from "../builder/sql";
-import type { ExecuteFn, ExecuteSyncFn, QueryResult, SyncDriver } from "./types";
+import type { ExecuteFn, ExecuteSyncFn, QueryResult, SyncDriver, TransactionOptions } from "./types";
 import { normalizeRow } from "./shared-sqlite";
 import { stripMatchedOuterParens } from "./shared";
 
@@ -51,12 +51,10 @@ export class DoSqliteDriver implements SyncDriver {
   };
 
   // storage.transaction() commits on resolution, rolls back on throw.
-  runInTransaction = <T>(cb: () => Promise<T>): Promise<T> => this.storage.transaction(cb);
-
-  // One handle: the single-connection execute IS executeSync (callers
-  // assert this identity — see Connection.transaction).
-  runInSingleConnection = <T>(cb: (execute: ExecuteSyncFn) => Promise<T>): Promise<T> =>
-    cb(this.executeSync);
+  runInTransaction = <T>(
+    _opts: TransactionOptions,
+    cb: (execute: ExecuteSyncFn) => Promise<T>,
+  ): Promise<T> => this.storage.transaction(() => cb(this.executeSync));
 
   close = (): Promise<void> => Promise.resolve();
 }
diff --git a/src/drivers/oracle-transaction.test.ts b/src/drivers/oracle-transaction.test.ts
new file mode 100644
index 0000000..3841c1a
--- /dev/null
+++ b/src/drivers/oracle-transaction.test.ts
@@ -0,0 +1,83 @@
+import { afterAll, beforeAll, describe, expect, test } from "vitest";
+import { sql } from "../builder/sql";
+import { Database, type Connection } from "../database";
+import { Varchar2 } from "../types/oracle";
+import { OracleDriver } from "./oracle";
+import { requireOraclePoolAttributes } from "./oracle-url";
+
+const enabled = process.env["ORACLE_URL"] !== undefined;
+const db = new Database();
+
+class TransactionRows extends db.Table("oracle_transaction_rows") {
+  id = Varchar2.column({ nonNull: true });
+  value = Varchar2.column({ nonNull: true });
+}
+
+describe.skipIf(!enabled)("Oracle transactions", () => {
+  let conn: Connection;
+
+  beforeAll(async () => {
+    conn = db.connect(await OracleDriver.create(requireOraclePoolAttributes()));
+    try {
+      await conn.execute(sql`DROP TABLE ${db.scopedIdent("oracle_transaction_rows")} PURGE`);
+    } catch {
+      // The table does not exist on the first run.
+    }
+    await conn.execute(sql`
+      CREATE TABLE ${db.scopedIdent("oracle_transaction_rows")} (
+        ${db.scopedIdent("id")} VARCHAR2(36) PRIMARY KEY,
+        ${db.scopedIdent("value")} VARCHAR2(100) NOT NULL
+      )
+    `);
+  });
+
+  afterAll(async () => {
+    await conn.execute(sql`DROP TABLE ${db.scopedIdent("oracle_transaction_rows")} PURGE`);
+    await conn.close();
+  });
+
+  test("commits successful transactions", async () => {
+    const result = await conn.transaction(async (tx) => {
+      await TransactionRows.insert({ id: "commit", value: "visible" }).execute(tx);
+      return "committed";
+    });
+
+    expect(result).toBe("committed");
+    expect(await TransactionRows.from()
+      .where(({ oracle_transaction_rows: row }) => row.id.eq("commit"))
+      .select(({ oracle_transaction_rows: row }) => ({ value: row.value }))
+      .execute()).toEqual([{ value: "visible" }]);
+  });
+
+  test("rolls back failed transactions", async () => {
+    await expect(conn.transaction(async (tx) => {
+      await TransactionRows.insert({ id: "rollback", value: "hidden" }).execute(tx);
+      throw new Error("rollback requested");
+    })).rejects.toThrow("rollback requested");
+
+    expect(await TransactionRows.from()
+      .where(({ oracle_transaction_rows: row }) => row.id.eq("rollback"))
+      .select(({ oracle_transaction_rows: row }) => ({ id: row.id }))
+      .execute()).toEqual([]);
+  });
+
+  test("pins one Oracle session and flattens nested transactions", async () => {
+    await conn.transaction(async (tx) => {
+      const first = await tx.execute(sql`
+        SELECT SYS_CONTEXT('USERENV', 'SID') AS ${db.scopedIdent("sid")} FROM DUAL
+      `);
+      await tx.transaction(async (nested) => {
+        const second = await nested.execute(sql`
+          SELECT SYS_CONTEXT('USERENV', 'SID') AS ${db.scopedIdent("sid")} FROM DUAL
+        `);
+        expect(second.rows[0]?.["sid"]).toBe(first.rows[0]?.["sid"]);
+        await TransactionRows.insert({ id: "nested", value: "committed" }).execute(nested);
+      });
+    });
+
+    expect(await TransactionRows.from()
+      .where(({ oracle_transaction_rows: row }) => row.id.eq("nested"))
+      .select(({ oracle_transaction_rows: row }) => ({ value: row.value }))
+      .execute()).toEqual([{ value: "committed" }]);
+  });
+});
diff --git a/src/drivers/oracle.test.ts b/src/drivers/oracle.test.ts
index ecafc6c..36c4dd9 100644
--- a/src/drivers/oracle.test.ts
+++ b/src/drivers/oracle.test.ts
@@ -20,7 +20,7 @@ test("oracle Connection constructs without a live engine", async () => {
   const conn = db.connect({
     dialect: "oracle",
     execute: async () => ({ rows: [{ v: "1" }] }),
-    runInSingleConnection: async () => {
+    runInTransaction: async () => {
       throw new Error("unused");
     },
     close: async () => {},
diff --git a/src/drivers/oracle.ts b/src/drivers/oracle.ts
index e49f30f..280638a 100644
--- a/src/drivers/oracle.ts
+++ b/src/drivers/oracle.ts
@@ -1,8 +1,9 @@
 import type { CompiledSql } from "../builder/sql";
 import type { DialectName } from "../builder/sql";
 import oracledb from "oracledb";
-import type { Driver, ExecuteFn, QueryResult } from "./types";
+import type { Driver, ExecuteFn, QueryResult, TransactionOptions } from "./types";
 import { stripMatchedOuterParens } from "./shared";
+import { runTransaction } from "./transaction";
 
 // node-oracledb adapter (thin mode — no Instant Client). Optional peer,
 // imported statically because this module only loads when the caller
@@ -50,29 +51,35 @@ export class OracleDriver implements Driver {
 
   private constructor(private pool: oracledb.Pool) {}
 
-  async execute({ text, values }: CompiledSql): Promise<QueryResult> {
-    const conn = await this.pool.getConnection();
-    try {
+  private executor(conn: oracledb.Connection, autoCommit: boolean): ExecuteFn {
+    return async ({ text, values }) => {
       const result = await conn.execute(stripMatchedOuterParens(text), oracleBinds(values), {
         outFormat: oracledb.OUT_FORMAT_OBJECT,
-        autoCommit: true,
+        autoCommit,
       });
       return { rows: normalizeRows(result.rows) };
+    };
+  }
+
+  async execute(compiled: CompiledSql): Promise<QueryResult> {
+    const conn = await this.pool.getConnection();
+    try {
+      return await this.executor(conn, true)(compiled);
     } finally {
       await conn.close();
     }
   }
 
-  async runInSingleConnection<T>(cb: (execute: ExecuteFn) => Promise<T>): Promise<T> {
+  async runInTransaction<T>(
+    _opts: TransactionOptions,
+    cb: (execute: ExecuteFn) => Promise<T>,
+  ): Promise<T> {
     const conn = await this.pool.getConnection();
     try {
-      return await cb(async ({ text, values }) => {
-        const result = await conn.execute(stripMatchedOuterParens(text), oracleBinds(values), {
-          outFormat: oracledb.OUT_FORMAT_OBJECT,
-          autoCommit: false,
-        });
-        return { rows: normalizeRows(result.rows) };
-      });
+      return await runTransaction({
+        commit: () => conn.commit(),
+        rollback: () => conn.rollback(),
+      }, () => cb(this.executor(conn, false)));
     } finally {
       await conn.close();
     }
diff --git a/src/drivers/pg.ts b/src/drivers/pg.ts
index d52c12f..b31d206 100644
--- a/src/drivers/pg.ts
+++ b/src/drivers/pg.ts
@@ -1,7 +1,8 @@
 import type { CompiledSql } from "../builder/sql";
 import type { DialectName } from "../builder/sql";
 import pgLib from "pg";
-import type { Driver, ExecuteFn, QueryResult } from "./types";
+import type { Driver, ExecuteFn, QueryResult, TransactionOptions } from "./types";
+import { postgresBeginSql, runSqlTransaction } from "./transaction";
 
 // pg adapter — returns raw text strings (no driver-side deserialization).
 // `pg` is an *optional* peer dep (see package.json#peerDependenciesMeta),
@@ -31,10 +32,14 @@ export class PgDriver implements Driver {
     return this.pool.query(text, values as unknown[]);
   }
 
-  async runInSingleConnection<T>(cb: (execute: ExecuteFn) => Promise<T>): Promise<T> {
+  async runInTransaction<T>(
+    opts: TransactionOptions,
+    cb: (execute: ExecuteFn) => Promise<T>,
+  ): Promise<T> {
     const client = await this.pool.connect();
+    const execute: ExecuteFn = ({ text, values }) => client.query(text, values as unknown[]);
     try {
-      return await cb(({ text, values }) => client.query(text, values as unknown[]));
+      return await runSqlTransaction(execute, postgresBeginSql(opts), () => cb(execute));
     } finally {
       client.release();
     }
diff --git a/src/drivers/pglite.ts b/src/drivers/pglite.ts
index 3ddce73..4e8bd48 100644
--- a/src/drivers/pglite.ts
+++ b/src/drivers/pglite.ts
@@ -1,7 +1,8 @@
 import { PGlite } from "@electric-sql/pglite";
 import type { CompiledSql } from "../builder/sql";
 import type { DialectName } from "../builder/sql";
-import type { Driver, ExecuteFn, QueryResult } from "./types";
+import type { Driver, ExecuteFn, QueryResult, TransactionOptions } from "./types";
+import { postgresBeginSql, runSqlTransaction } from "./transaction";
 
 // pglite adapter — returns raw text strings (no driver-side deserialization).
 // `@electric-sql/pglite` is an optional peer, imported statically because
@@ -35,8 +36,12 @@ export class PgliteDriver implements Driver {
     return this.db.query(text, values as unknown[], { parsers: this.parsers }) as Promise<QueryResult>;
   }
 
-  async runInSingleConnection<T>(cb: (execute: ExecuteFn) => Promise<T>): Promise<T> {
-    return cb(this.execute.bind(this));
+  async runInTransaction<T>(
+    opts: TransactionOptions,
+    cb: (execute: ExecuteFn) => Promise<T>,
+  ): Promise<T> {
+    const execute = this.execute.bind(this);
+    return runSqlTransaction(execute, postgresBeginSql(opts), () => cb(execute));
   }
 
   async close(): Promise<void> {
diff --git a/src/drivers/sqlite.ts b/src/drivers/sqlite.ts
index f3a67e2..61a3ca4 100644
--- a/src/drivers/sqlite.ts
+++ b/src/drivers/sqlite.ts
@@ -1,9 +1,10 @@
 import type { CompiledSql } from "../builder/sql";
 import type { DialectName } from "../builder/sql";
 import BetterSqlite3 from "better-sqlite3";
-import type { ExecuteSyncFn, QueryResult, SyncDriver } from "./types";
+import type { ExecuteSyncFn, QueryResult, SyncDriver, TransactionOptions } from "./types";
 import { normalizeRow } from "./shared-sqlite";
 import { stripMatchedOuterParens } from "./shared";
+import { runSqlTransaction } from "./transaction";
 
 // better-sqlite3 adapter. Synchronous under the hood; wrapped in
 // Promise.resolve for the async Driver contract. `better-sqlite3` is an
@@ -50,10 +51,11 @@ export class SqliteDriver implements SyncDriver {
     return { rows: [] };
   }
 
-  async runInSingleConnection<T>(cb: (execute: ExecuteSyncFn) => Promise<T>): Promise<T> {
-    // One handle: the single-connection execute IS executeSync (callers
-    // assert this identity — see Connection.transaction).
-    return cb(this.executeSync);
+  async runInTransaction<T>(
+    _opts: TransactionOptions,
+    cb: (execute: ExecuteSyncFn) => Promise<T>,
+  ): Promise<T> {
+    return runSqlTransaction(this.executeSync, "BEGIN", () => cb(this.executeSync));
   }
 
   async close(): Promise<void> {
diff --git a/src/drivers/transaction.test.ts b/src/drivers/transaction.test.ts
new file mode 100644
index 0000000..5b42ea6
--- /dev/null
+++ b/src/drivers/transaction.test.ts
@@ -0,0 +1,77 @@
+import { describe, expect, test, vi } from "vitest";
+import type { ExecuteFn, TransactionOptions } from "./types";
+import { postgresBeginSql, runSqlTransaction, runTransaction } from "./transaction";
+
+describe("driver transaction helpers", () => {
+  test("selects PostgreSQL BEGIN SQL from fixed values", () => {
+    expect(postgresBeginSql({})).toBe("BEGIN");
+    expect(postgresBeginSql({ isolation: "read committed" })).toBe(
+      "BEGIN ISOLATION LEVEL READ COMMITTED",
+    );
+    expect(postgresBeginSql({ isolation: "repeatable read" })).toBe(
+      "BEGIN ISOLATION LEVEL REPEATABLE READ",
+    );
+    expect(postgresBeginSql({ isolation: "serializable" })).toBe(
+      "BEGIN ISOLATION LEVEL SERIALIZABLE",
+    );
+    expect(() => postgresBeginSql(
+      { isolation: "serializable; SELECT 1" } as unknown as TransactionOptions,
+    )).toThrow(
+      "Unsupported PostgreSQL transaction isolation level",
+    );
+  });
+
+  test("runs begin, callback, and commit in order", async () => {
+    const events: string[] = [];
+    const result = await runTransaction({
+      begin: () => { events.push("begin"); },
+      commit: async () => { events.push("commit"); },
+      rollback: () => { events.push("rollback"); },
+    }, async () => {
+      events.push("callback");
+      return 42;
+    });
+
+    expect(result).toBe(42);
+    expect(events).toEqual(["begin", "callback", "commit"]);
+  });
+
+  test("rolls back callback and commit failures", async () => {
+    const callbackEvents: string[] = [];
+    await expect(runTransaction({
+      commit: () => { callbackEvents.push("commit"); },
+      rollback: () => { callbackEvents.push("rollback"); },
+    }, async () => {
+      callbackEvents.push("callback");
+      throw new Error("callback failed");
+    })).rejects.toThrow("callback failed");
+    expect(callbackEvents).toEqual(["callback", "rollback"]);
+
+    const commitEvents: string[] = [];
+    await expect(runTransaction({
+      commit: () => {
+        commitEvents.push("commit");
+        throw new Error("commit failed");
+      },
+      rollback: () => { commitEvents.push("rollback"); },
+    }, async () => { commitEvents.push("callback"); })).rejects.toThrow("commit failed");
+    expect(commitEvents).toEqual(["callback", "commit", "rollback"]);
+  });
+
+  test("SQL transactions emit protocol statements", async () => {
+    const statements: string[] = [];
+    const execute: ExecuteFn = vi.fn(async ({ text }) => {
+      statements.push(text);
+      return { rows: [] };
+    });
+
+    await runSqlTransaction(execute, "BEGIN ISOLATION LEVEL SERIALIZABLE", async () => {
+      await execute({ text: "SELECT 1", values: [] });
+    });
+    expect(statements).toEqual([
+      "BEGIN ISOLATION LEVEL SERIALIZABLE",
+      "SELECT 1",
+      "COMMIT",
+    ]);
+  });
+});
diff --git a/src/drivers/transaction.ts b/src/drivers/transaction.ts
new file mode 100644
index 0000000..cd7c857
--- /dev/null
+++ b/src/drivers/transaction.ts
@@ -0,0 +1,54 @@
+import type { AnyExecuteFn, TransactionOptions } from "./types";
+
+export type TransactionLifecycle = {
+  begin?: () => void | Promise<void>;
+  commit: () => void | Promise<void>;
+  rollback: () => void | Promise<void>;
+};
+
+// Shared state machine for drivers whose transaction protocol exposes
+// explicit begin/commit/rollback operations. Connection acquisition and
+// the transaction-bound executor remain driver-specific.
+export const runTransaction = async <T>(
+  lifecycle: TransactionLifecycle,
+  cb: () => Promise<T>,
+): Promise<T> => {
+  await lifecycle.begin?.();
+  try {
+    const result = await cb();
+    await lifecycle.commit();
+    return result;
+  } catch (error) {
+    try {
+      await lifecycle.rollback();
+    } catch (rollbackError) {
+      console.error("Rollback failed after transaction error:", rollbackError);
+    }
+    throw error;
+  }
+};
+
+export const postgresBeginSql = (opts: TransactionOptions): string => {
+  switch (opts.isolation) {
+    case undefined: return "BEGIN";
+    case "read committed": return "BEGIN ISOLATION LEVEL READ COMMITTED";
+    case "repeatable read": return "BEGIN ISOLATION LEVEL REPEATABLE READ";
+    case "serializable": return "BEGIN ISOLATION LEVEL SERIALIZABLE";
+    default: throw new TypeError("Unsupported PostgreSQL transaction isolation level");
+  }
+};
+
+export const runSqlTransaction = <T>(
+  execute: AnyExecuteFn,
+  begin: string,
+  cb: () => Promise<T>,
+): Promise<T> => {
+  const run = async (text: string): Promise<void> => {
+    await execute({ text, values: [] });
+  };
+  return runTransaction({
+    begin: () => run(begin),
+    commit: () => run("COMMIT"),
+    rollback: () => run("ROLLBACK"),
+  }, cb);
+};
diff --git a/src/drivers/types.ts b/src/drivers/types.ts
index b84538e..c101862 100644
--- a/src/drivers/types.ts
+++ b/src/drivers/types.ts
@@ -18,6 +18,9 @@ export type ExecuteSyncFn = (sql: CompiledSql) => QueryResult;
 // Callers `await` either flavor (a no-op on the sync one).
 export type AnyExecuteFn = ExecuteFn | ExecuteSyncFn;
 
+export type TransactionIsolation = "read committed" | "repeatable read" | "serializable";
+export type TransactionOptions = { isolation?: TransactionIsolation };
+
 export interface Driver {
   readonly dialect: DialectName;
   execute: ExecuteFn;
@@ -25,13 +28,13 @@ export interface Driver {
   // SqlStorage). Required by sqlite live capture, which needs multiple
   // statements with no awaits between them.
   executeSync?: ExecuteSyncFn;
-  // Native transaction protocol: commit when `cb` resolves, roll back
-  // when it throws. When present, Connection.transaction() uses this
-  // instead of BEGIN/COMMIT/ROLLBACK SQL (workerd rejects SQL BEGIN).
-  runInTransaction?<T>(cb: () => Promise<T>): Promise<T>;
-  // Sync drivers must pass their executeSync itself as `execute` (one
-  // handle, one channel) — Connection.transaction() asserts the identity.
-  runInSingleConnection<T>(cb: (execute: AnyExecuteFn) => Promise<T>): Promise<T>;
+  // Own connection pinning and the database's transaction protocol. The
+  // callback receives the only executor valid inside that transaction.
+  // Sync drivers must pass executeSync itself (one handle, one channel).
+  runInTransaction<T>(
+    opts: TransactionOptions,
+    cb: (execute: AnyExecuteFn) => Promise<T>,
+  ): Promise<T>;
   close(): Promise<void>;
 }
 
diff --git a/src/live/sqlite/db-live.test.ts b/src/live/sqlite/db-live.test.ts
index 7593b97..f71dfca 100644
--- a/src/live/sqlite/db-live.test.ts
+++ b/src/live/sqlite/db-live.test.ts
@@ -389,9 +389,20 @@ test("live over DoSqliteDriver (fake SqlStorage backed by better-sqlite3)", asyn
     .live(doConn)[Symbol.asyncIterator]();
 
   expect(await takeNext(iter)).toEqual([]);
-  await Notes.insert({ id: 1, user_id: 1, body: "from-do" }).execute(doConn);
+  await doConn.transaction(async (tx) => {
+    await Notes.insert({ id: 1, user_id: 1, body: "from-do" }).execute(tx);
+  });
   expect(await takeNext(iter)).toEqual([{ id: 1, body: "from-do" }]);
 
+  await expect(doConn.transaction(async (tx) => {
+    await Notes.insert({ id: 2, user_id: 1, body: "rolled-back" }).execute(tx);
+    throw new Error("rollback requested");
+  })).rejects.toThrow("rollback requested");
+  expect(await Notes.from()
+    .where(({ notes }) => notes.id.eq(2))
+    .select(({ notes }) => ({ id: notes.id }))
+    .execute(doConn)).toEqual([]);
+
   await iter.return?.();
   raw.close();
 }, 10_000);
diff --git a/src/test-helpers.ts b/src/test-helpers.ts
index 39e798b..a7e4730 100644
--- a/src/test-helpers.ts
+++ b/src/test-helpers.ts
@@ -29,7 +29,7 @@ export const dialectOnlyDriver = (dialect: DialectName): SyncDriver => {
     liveSeq: 0n,
     execute: unsupported,
     executeSync: unsupported,
-    runInSingleConnection: unsupported,
+    runInTransaction: unsupported,
     close: () => Promise.resolve(),
   };
 };