diff --git a/apps/cli/src/command-internal/db-connection.errors.ts b/apps/cli/src/command-internal/db-connection.errors.ts
index 793e2b28c8..ad2a399090 100644
--- a/apps/cli/src/command-internal/db-connection.errors.ts
+++ b/apps/cli/src/command-internal/db-connection.errors.ts
@@ -36,6 +36,12 @@ export class DbExecError extends Data.TaggedError("DbExecError")<{
    * the batch length for a deferred Sync failure. Absent for `exec`/`query`.
    */
   readonly statementIndex?: number;
+  /**
+   * Which CLI-injected transaction wrapper was in flight when the batch failed —
+   * whether the server rejected BEGIN/COMMIT or the connection was lost while one
+   * was pending — rather than a caller statement. Absent otherwise.
+   */
+  readonly transactionPhase?: "begin" | "commit";
   /**
    * Postgres SQLSTATE (e.g. `42P01` undefined_table), extracted from the driver
    * error's `cause` chain when present. Lets callers match Go's error-code checks
diff --git a/apps/cli/src/command-internal/db-connection.service.ts b/apps/cli/src/command-internal/db-connection.service.ts
index e87e21e0b8..53f4bcece7 100644
--- a/apps/cli/src/command-internal/db-connection.service.ts
+++ b/apps/cli/src/command-internal/db-connection.service.ts
@@ -104,9 +104,12 @@ export interface DbSession {
   /** Run a single SQL statement, ignoring any returned rows. */
   readonly exec: (sql: string) => Effect.Effect<void, DbExecError>;
   /**
-   * Run statements as one extended-protocol batch with a single final Sync.
-   * On failure, {@link DbExecError.statementIndex} is the number of
-   * statements that completed before the error.
+   * Run statements as one extended-protocol batch inside a single explicit
+   * transaction, with a single final Sync — a bare pipeline is not a transaction
+   * block (supabase/cli#6347). On failure a bounded, best-effort rollback runs
+   * before the connection can be reused (a rollback that does not succeed
+   * discards the connection), and {@link DbExecError.statementIndex} is
+   * the number of the caller's statements that completed before the error.
    *
    * A batch runs on its own pooled connection, which the driver checks out per
    * call. Failing to acquire it, or losing it before any of the batch reaches the
diff --git a/apps/cli/src/command-internal/db-connection.sql-pg.integration.test.ts b/apps/cli/src/command-internal/db-connection.sql-pg.integration.test.ts
index 49a14bd053..f2e5337fd9 100644
--- a/apps/cli/src/command-internal/db-connection.sql-pg.integration.test.ts
+++ b/apps/cli/src/command-internal/db-connection.sql-pg.integration.test.ts
@@ -11,7 +11,7 @@ import { describe, expect, it } from "@effect/vitest";
 import { Duration, Effect } from "effect";
 
 import { SUGGEST_ENV_VAR, SUGGEST_LOCAL_STACK } from "./connect-errors.ts";
-import type { DbConnectError, DbExecError } from "./db-connection.errors.ts";
+import { type DbConnectError, DbExecError } from "./db-connection.errors.ts";
 import { type DbSession, type PgConnInput, DbConnection } from "./db-connection.service.ts";
 import {
   acquirePgPool,
@@ -162,6 +162,7 @@ interface FakeBatchServerState {
   readonly frameTypes: Array<string>;
   readonly statements: Array<string>;
   readonly params: Array<ReadonlyArray<string | null>>;
+  readonly simpleQueries: Array<string>;
   syncs: number;
 }
 
@@ -173,8 +174,12 @@ const fakeBatchServer = (
     readonly emptyAt?: number;
     /** Never answer an extended-protocol frame, so a batch hangs until interrupted. */
     readonly stall?: boolean;
+    readonly stallRollback?: boolean;
+    readonly destroyOnRollback?: boolean;
     /** Drop the connection on the first Sync, so a batch dies mid-flight. */
     readonly destroyOnFirstSync?: boolean;
+    /** Drop the connection at the first Execute (BEGIN's), before anything completes. */
+    readonly destroyOnFirstExecute?: boolean;
   } = {},
 ): Promise<{
   readonly port: number;
@@ -187,9 +192,11 @@ const fakeBatchServer = (
       frameTypes: [],
       statements: [],
       params: [],
+      simpleQueries: [],
       syncs: 0,
     };
     const sockets: Array<net.Socket> = [];
+    let destroyedOnExecute = false;
     const server = net.createServer((socket) => {
       sockets.push(socket);
       let sawStartup = false;
@@ -219,6 +226,13 @@ const fakeBatchServer = (
           const body = pending.subarray(5, length + 1);
           pending = pending.subarray(length + 1);
           if (type === "Q") {
+            const sql = body.toString("utf8", 0, body.length - 1);
+            state.simpleQueries.push(sql);
+            if (options.stallRollback === true && sql === "ROLLBACK") continue;
+            if (options.destroyOnRollback === true && sql === "ROLLBACK") {
+              socket.destroy();
+              return;
+            }
             socket.write(Buffer.concat([commandComplete("SELECT 1"), READY_FOR_QUERY]));
             continue;
           }
@@ -250,6 +264,11 @@ const fakeBatchServer = (
           } else if (type === "D") {
             if (!failed) socket.write(NO_DATA);
           } else if (type === "E") {
+            if (options.destroyOnFirstExecute === true && !destroyedOnExecute) {
+              destroyedOnExecute = true;
+              socket.destroy();
+              return;
+            }
             if (!failed) {
               if (activeIndex === options.failExecuteAt) {
                 failed = true;
@@ -547,9 +566,9 @@ describe("dbConnectionSqlPgLayer extended batches", () => {
       Effect.ensuring(Effect.sync(server.close)),
     );
 
-  it.live("sends every statement and parameter set before one Sync", () =>
+  it.live("sends every statement and parameter set inside one BEGIN/COMMIT before one Sync", () =>
     Effect.gen(function* () {
-      const server = yield* Effect.promise(() => fakeBatchServer({ emptyAt: 1 }));
+      const server = yield* Effect.promise(() => fakeBatchServer({ emptyAt: 2 }));
       const values = ["plain", 'quote"', "slash\\", "comma,", "{brace}", "line\nbreak", "NULL", ""];
       yield* runWithBatchServer(server, (session) =>
         session.execBatch([
@@ -562,11 +581,14 @@ describe("dbConnectionSqlPgLayer extended batches", () => {
         ]),
       );
       expect(server.state.statements).toEqual([
+        "BEGIN",
         "SELECT 1",
         "-- comment only",
         "INSERT INTO history(version, name, statements) VALUES($1, $2, $3)",
+        "COMMIT",
       ]);
       expect(server.state.params).toEqual([
+        [],
         [],
         [],
         [
@@ -574,6 +596,7 @@ describe("dbConnectionSqlPgLayer extended batches", () => {
           "name\\two",
           '{"plain","quote\\\"","slash\\\\","comma,","{brace}","line\nbreak","NULL",""}',
         ],
+        [],
       ]);
       expect(server.state.frameTypes).toEqual([
         "P",
@@ -588,15 +611,24 @@ describe("dbConnectionSqlPgLayer extended batches", () => {
         "B",
         "D",
         "E",
+        "P",
+        "B",
+        "D",
+        "E",
+        "P",
+        "B",
+        "D",
+        "E",
         "S",
       ]);
       expect(server.state.syncs).toBe(1);
+      expect(server.state.simpleQueries).not.toContain("ROLLBACK");
     }),
   );
 
   it.live("maps a later parse failure to its statement and keeps its local position", () =>
     Effect.gen(function* () {
-      const server = yield* Effect.promise(() => fakeBatchServer({ emptyAt: 1, failAt: 2 }));
+      const server = yield* Effect.promise(() => fakeBatchServer({ emptyAt: 2, failAt: 3 }));
       yield* runWithBatchServer(server, (session) =>
         Effect.gen(function* () {
           const error = asBatchExecError(
@@ -617,7 +649,7 @@ describe("dbConnectionSqlPgLayer extended batches", () => {
 
   it.live("maps a position-less runtime failure from completed commands", () =>
     Effect.gen(function* () {
-      const server = yield* Effect.promise(() => fakeBatchServer({ failExecuteAt: 1 }));
+      const server = yield* Effect.promise(() => fakeBatchServer({ failExecuteAt: 2 }));
       yield* runWithBatchServer(server, (session) =>
         session
           .execBatch([{ sql: "SELECT 1" }, { sql: "INSERT duplicate" }, { sql: "SELECT 3" }])
@@ -655,6 +687,72 @@ describe("dbConnectionSqlPgLayer extended batches", () => {
     }),
   );
 
+  it.live("rolls a failed batch's transaction back before the pooled client is reused", () =>
+    Effect.gen(function* () {
+      const server = yield* Effect.promise(() => fakeBatchServer({ failExecuteAt: 3 }));
+      yield* runWithBatchServer(server, (session) =>
+        Effect.gen(function* () {
+          yield* session
+            .execBatch([{ sql: "SELECT 1" }, { sql: "SELECT 2" }, { sql: "INSERT duplicate" }])
+            .pipe(Effect.flip);
+          const sockets = server.sockets.length;
+          yield* session.execBatch([{ sql: "SELECT 3" }]);
+          expect(server.sockets.length).toBe(sockets);
+        }),
+      );
+      expect(server.state.simpleQueries).toContain("ROLLBACK");
+      expect(server.state.syncs).toBe(2);
+    }),
+  );
+
+  it.live("absorbs a socket death during the error-path rollback instead of crashing", () =>
+    Effect.gen(function* () {
+      const server = yield* Effect.promise(() =>
+        fakeBatchServer({ failExecuteAt: 3, destroyOnRollback: true }),
+      );
+      yield* runWithBatchServer(server, (session) =>
+        Effect.gen(function* () {
+          yield* session
+            .execBatch([{ sql: "SELECT 1" }, { sql: "SELECT 2" }, { sql: "INSERT duplicate" }])
+            .pipe(Effect.flip);
+          yield* session.execBatch([{ sql: "SELECT 3" }]).pipe(
+            Effect.timeoutOrElse({
+              duration: Duration.seconds(10),
+              orElse: () => Effect.die("the batch after a dead-rollback socket never settled"),
+            }),
+          );
+          expect(server.state.simpleQueries).toContain("ROLLBACK");
+        }),
+      );
+    }),
+  );
+
+  it.live(
+    "bounds a stalled failed-batch rollback and discards the client instead of reusing it",
+    () =>
+      Effect.gen(function* () {
+        const server = yield* Effect.promise(() =>
+          fakeBatchServer({ failExecuteAt: 3, stallRollback: true }),
+        );
+        yield* runWithBatchServer(server, (session) =>
+          Effect.gen(function* () {
+            const before = server.sockets.length;
+            yield* session
+              .execBatch([{ sql: "SELECT 1" }, { sql: "SELECT 2" }, { sql: "INSERT duplicate" }])
+              .pipe(Effect.flip);
+            yield* session.execBatch([{ sql: "SELECT 3" }]).pipe(
+              Effect.timeoutOrElse({
+                duration: Duration.seconds(10),
+                orElse: () => Effect.die("the batch after a stalled rollback never settled"),
+              }),
+            );
+            expect(server.state.simpleQueries).toContain("ROLLBACK");
+            expect(server.sockets.length).toBeGreaterThan(before);
+          }),
+        );
+      }),
+  );
+
   it.live("fails a batch whose connection drops after it was written, then recovers", () =>
     // A socket dropped after the batch was written must fail that batch and must not leave
     // the client to be handed to the next one.
@@ -671,6 +769,35 @@ describe("dbConnectionSqlPgLayer extended batches", () => {
           );
           expect(error._tag).toBe("DbExecError");
           expect(asBatchExecError(error).message).toContain("Connection terminated unexpectedly");
+          // This server acks every statement and dies at Sync, so the loss lands
+          // on COMMIT — marked so the formatter never blames a caller statement.
+          expect(asBatchExecError(error).transactionPhase).toBe("commit");
+          yield* session.execBatch([{ sql: "SELECT 3" }]);
+        }),
+      );
+    }),
+  );
+
+  it.live("marks the begin phase when the connection drops before BEGIN completes", () =>
+    // The loss arrives while BEGIN is still in flight, so no caller statement ran:
+    // the phase marker keeps formatters from rendering `At statement: 0` for it.
+    Effect.gen(function* () {
+      const server = yield* Effect.promise(() => fakeBatchServer({ destroyOnFirstExecute: true }));
+      yield* runWithBatchServer(server, (session) =>
+        Effect.gen(function* () {
+          const error = yield* session.execBatch([{ sql: "SELECT 1" }, { sql: "SELECT 2" }]).pipe(
+            Effect.flip,
+            Effect.timeoutOrElse({
+              duration: Duration.seconds(10),
+              orElse: () => Effect.die("execBatch never settled after the connection died"),
+            }),
+          );
+          expect(error).toBeInstanceOf(DbExecError);
+          expect(asBatchExecError(error).message).toContain("Connection terminated unexpectedly");
+          expect(asBatchExecError(error)).toMatchObject({
+            statementIndex: 0,
+            transactionPhase: "begin",
+          });
           yield* session.execBatch([{ sql: "SELECT 3" }]);
         }),
       );
diff --git a/apps/cli/src/command-internal/db-connection.sql-pg.layer.ts b/apps/cli/src/command-internal/db-connection.sql-pg.layer.ts
index b9a2f43cce..74f796e87d 100644
--- a/apps/cli/src/command-internal/db-connection.sql-pg.layer.ts
+++ b/apps/cli/src/command-internal/db-connection.sql-pg.layer.ts
@@ -2,7 +2,7 @@ import { readFileSync } from "node:fs";
 import * as net from "node:net";
 import type { ConnectionOptions } from "node:tls";
 import { PgClient } from "@effect/sql-pg";
-import { Cause, Duration, Effect, Exit, Layer, Scope } from "effect";
+import { Cause, Duration, Effect, Exit, Layer, Option, Scope } from "effect";
 import * as Reactivity from "effect/unstable/reactivity/Reactivity";
 import { ConnectionError, SqlError } from "effect/unstable/sql/SqlError";
 // `pg` is also `@effect/sql-pg`'s transitive driver; we depend on it directly for
@@ -105,6 +105,12 @@ function needsRoleStepDown(user: string): boolean {
 const TERMINAL_SQLSTATES = new Set(["28P01", "3D000", "42501"]);
 const TLS_GATED_SQLSTATE = "28000";
 
+// Class 08 (connection exception) plus the operator-intervention terminations that
+// close the session; 57014 (query_canceled) stays a statement failure.
+const SESSION_ENDING_SQLSTATES = new Set(["57P01", "57P02", "57P03", "57P04", "57P05"]);
+const isConnectionEndingSqlState = (code: string): boolean =>
+  code.startsWith("08") || SESSION_ENDING_SQLSTATES.has(code);
+
 /**
  * Whether a failed connection attempt should terminate the multi-host fallback
  * chain instead of falling through to the next host. Mirrors pgconn's
@@ -219,7 +225,14 @@ const DB_KEEPALIVE_IDLE_MILLIS = 300_000;
  */
 export function batchFailureError(
   error: Error,
-  batch: { readonly completed: number; readonly outcome: BatchOutcome } | undefined,
+  batch:
+    | {
+        readonly completed: number;
+        readonly outcome: BatchOutcome;
+        readonly began?: boolean;
+        readonly atCommit?: boolean;
+      }
+    | undefined,
   isLocal: boolean,
 ): DbExecError | DbConnectError {
   if (batch === undefined || batch.outcome === "unsent") {
@@ -231,12 +244,34 @@ export function batchFailureError(
     });
   }
   const mapped = toExecError(error);
+  // The phase marker and the relabel are separate: whenever BEGIN or COMMIT was
+  // the statement in flight, none of the caller's statements failed at
+  // `statementIndex`, so the phase is always recorded and formatters must not
+  // blame one. The message is only relabeled when the server rejected the
+  // wrapper itself — a lost connection (including a server-initiated
+  // termination) keeps its own reason. Gated on SQLSTATE class, never the
+  // severity string, which arrives localized (e.g. "FEHLER").
+  const server = extractPgServerError(error);
+  const statementFailure = server !== undefined && !isConnectionEndingSqlState(server.code);
+  const atBegin = batch.outcome === "submitted" && batch.began === false;
+  const atCommit = batch.outcome === "submitted" && batch.atCommit === true;
+  const transactionPhase: "begin" | "commit" | undefined = atBegin
+    ? "begin"
+    : atCommit
+      ? "commit"
+      : undefined;
   return new DbExecError({
-    message: mapped.message,
+    message:
+      atBegin && statementFailure
+        ? `failed to begin the batch transaction: ${mapped.message}`
+        : atCommit && statementFailure
+          ? `failed to commit the batch transaction: ${mapped.message}`
+          : mapped.message,
     code: mapped.code,
     detail: mapped.detail,
     position: mapped.position,
     statementIndex: batch.completed,
+    ...(transactionPhase !== undefined ? { transactionPhase } : {}),
   });
 }
 
@@ -246,18 +281,24 @@ export function batchFailureError(
  * socket is already gone, so the next checkout would write into the same dead connection.
  *
  * A batch that WAS written keeps its client: a statement failure should not cost a redial and
- * a fresh step-down on a single-connection pool. Recovering from a socket that died after the
- * write is left to pg-pool, which drops a released client whose private `_queryable` flag is
- * false — so that is the behavior to re-check if a pg-pool bump ever breaks the recovery this
- * layer's integration tests assert.
+ * a fresh step-down on a single-connection pool. The keep is conditional on `rolledBack` —
+ * a failed submitted batch whose rollback failed or timed out is discarded. Recovering
+ * from a socket that died after the write is additionally backstopped by pg-pool, which
+ * drops a released client whose private `_queryable` flag is false — so that is the behavior
+ * to re-check if a pg-pool bump ever breaks the recovery this layer's integration tests
+ * assert.
  */
 export function shouldDiscardBatchClient(
   batch: { readonly outcome: BatchOutcome } | undefined,
   exit: Exit.Exit<unknown, unknown>,
+  rolledBack: boolean,
 ): boolean {
   return (
     (batch !== undefined && batch.outcome !== "submitted") ||
-    (Exit.isFailure(exit) && (Cause.hasInterrupts(exit.cause) || Cause.hasDies(exit.cause)))
+    (Exit.isFailure(exit) &&
+      (Cause.hasInterrupts(exit.cause) ||
+        Cause.hasDies(exit.cause) ||
+        (batch?.outcome === "submitted" && !rolledBack)))
   );
 }
 
@@ -277,6 +318,12 @@ export class PgBatchQuery implements Pg.Submittable {
   callback: (error: Error | undefined) => void;
   completed = 0;
   outcome: BatchOutcome = "unsent";
+  began = false;
+
+  // An error arriving once every caller statement completed can only be COMMIT's.
+  get atCommit(): boolean {
+    return this.began && this.completed >= this.statements.length;
+  }
 
   constructor(
     statements: ReadonlyArray<DbBatchStatement>,
@@ -296,7 +343,12 @@ export class PgBatchQuery implements Pg.Submittable {
     let started = false;
     connection.stream.cork?.();
     try {
-      for (const { sql, params } of this.statements) {
+      // A bare pipeline is not a transaction block (supabase/cli#6347).
+      for (const { sql, params } of [
+        { sql: "BEGIN", params: [] },
+        ...this.statements,
+        { sql: "COMMIT", params: [] },
+      ]) {
         started = true;
         connection.parse({ name: "", text: sql, types: [] }, true);
         connection.bind({ portal: "", statement: "", values: [...params] }, true);
@@ -327,11 +379,22 @@ export class PgBatchQuery implements Pg.Submittable {
   handlePortalSuspended(): void {}
 
   handleCommandComplete(): void {
-    this.completed += 1;
+    this.recordCompletion();
   }
 
   handleEmptyQuery(): void {
-    this.completed += 1;
+    this.recordCompletion();
+  }
+
+  // BEGIN completes first and COMMIT only after every statement; neither counts.
+  private recordCompletion(): void {
+    if (!this.began) {
+      this.began = true;
+      return;
+    }
+    if (this.completed < this.statements.length) {
+      this.completed += 1;
+    }
   }
 
   handleCopyInResponse(connection: Pg.Connection): void {
@@ -1063,44 +1126,70 @@ const connect = (
 
     const execBatch = (statements: ReadonlyArray<DbBatchStatement>) => {
       if (statements.length === 0) return Effect.void;
-      let batchQuery: PgBatchQuery | undefined;
-      return Effect.acquireUseRelease(
-        Effect.interruptible(acquireBatchClient),
-        (activeClient) => {
-          const onConnectionError = () => {};
-          activeClient.on("error", onConnectionError);
-          return Effect.callback<void, DbExecError | DbConnectError>((resume) => {
-            let done = false;
-            const finish = (error: Error | undefined) => {
-              if (done) return;
-              done = true;
-              if (error === undefined) {
-                resume(Effect.void);
-                return;
+      // Suspended so each evaluation owns fresh batch/rollback state.
+      return Effect.suspend(() => {
+        let batchQuery: PgBatchQuery | undefined;
+        let rolledBack = false;
+        // Spans the whole checkout: an unlistened 'error' kills the process (see
+        // acquireRawClient) and pg-pool detaches its own handler while checked out.
+        const onConnectionError = () => {};
+        return Effect.acquireUseRelease(
+          Effect.interruptible(acquireBatchClient).pipe(
+            Effect.tap((activeClient) =>
+              Effect.sync(() => activeClient.on("error", onConnectionError)),
+            ),
+          ),
+          (activeClient) =>
+            Effect.callback<void, DbExecError | DbConnectError>((resume) => {
+              let done = false;
+              const finish = (error: Error | undefined) => {
+                if (done) return;
+                done = true;
+                if (error === undefined) {
+                  resume(Effect.void);
+                  return;
+                }
+                resume(Effect.fail(batchFailureError(error, batchQuery, options.isLocal)));
+              };
+              batchQuery = new PgBatchQuery(statements, finish);
+              try {
+                activeClient.query(batchQuery);
+              } catch (error) {
+                finish(error instanceof Error ? error : new Error(String(error)));
               }
-              resume(Effect.fail(batchFailureError(error, batchQuery, options.isLocal)));
-            };
-            batchQuery = new PgBatchQuery(statements, finish);
-            try {
-              activeClient.query(batchQuery);
-            } catch (error) {
-              finish(error instanceof Error ? error : new Error(String(error)));
-            }
-            return Effect.sync(() => {
-              done = true;
-            });
-          }).pipe(
-            Effect.ensuring(
-              Effect.sync(() => activeClient.removeListener("error", onConnectionError)),
+              return Effect.sync(() => {
+                done = true;
+              });
+            }).pipe(
+              // Roll a written batch's aborted transaction back while still
+              // interruptible; a rollback that fails or times out leaves the client
+              // to the discard below instead of returning it aborted (25P02). The
+              // rollback's own failure is consumed as that discard policy — it must
+              // never supplant the batch error this tap is observing.
+              Effect.tapError(() =>
+                Effect.suspend(() => {
+                  if (batchQuery?.outcome !== "submitted") return Effect.void;
+                  return Effect.tryPromise(() => activeClient.query("ROLLBACK")).pipe(
+                    Effect.match({ onFailure: () => false, onSuccess: () => true }),
+                    Effect.timeoutOption(1000),
+                    Effect.map((completed) => {
+                      rolledBack = Option.getOrElse(completed, () => false);
+                    }),
+                  );
+                }),
+              ),
             ),
-          );
-        },
-        (activeClient, exit) =>
-          Effect.sync(() => {
-            const discard = shouldDiscardBatchClient(batchQuery, exit);
-            activeClient.release(discard ? new Error("batch connection discarded") : undefined);
-          }),
-      );
+          (activeClient, exit) =>
+            Effect.sync(() => {
+              const discard = shouldDiscardBatchClient(batchQuery, exit, rolledBack);
+              try {
+                activeClient.release(discard ? new Error("batch connection discarded") : undefined);
+              } finally {
+                activeClient.removeListener("error", onConnectionError);
+              }
+            }),
+        );
+      });
     };
 
     const session: DbSession = {
diff --git a/apps/cli/src/command-internal/db-connection.sql-pg.unit.test.ts b/apps/cli/src/command-internal/db-connection.sql-pg.unit.test.ts
index 6ad2dc851f..b4e9dc4b7b 100644
--- a/apps/cli/src/command-internal/db-connection.sql-pg.unit.test.ts
+++ b/apps/cli/src/command-internal/db-connection.sql-pg.unit.test.ts
@@ -6,6 +6,7 @@ import { describe, expect, it } from "vitest";
 
 import { ErrorActionabilityId } from "../shared/telemetry/error-actionability.ts";
 import { SUGGEST_LOCAL_STACK } from "./connect-errors.ts";
+import { DbExecError } from "./db-connection.errors.ts";
 import {
   acquireProbedPool,
   batchFailureError,
@@ -596,7 +597,9 @@ describe("PgBatchQuery.submit", () => {
       frames,
       connection: {
         stream,
-        parse: record("parse"),
+        parse: (query: { text: string }) => {
+          frames.push(`parse(${query.text})`);
+        },
         bind: record("bind"),
         describe: record("describe"),
         execute: record("execute"),
@@ -626,10 +629,26 @@ describe("PgBatchQuery.submit", () => {
       "the connection's socket became unwritable while the batch was flushing",
     );
     expect(batch.outcome).toBe("unsent");
-    expect(frames).toEqual(["cork", "parse", "bind", "describe", "execute", "sync", "uncork"]);
+    expect(frames).toEqual([
+      "cork",
+      "parse(BEGIN)",
+      "bind",
+      "describe",
+      "execute",
+      "parse(select 1)",
+      "bind",
+      "describe",
+      "execute",
+      "parse(COMMIT)",
+      "bind",
+      "describe",
+      "execute",
+      "sync",
+      "uncork",
+    ]);
   });
 
-  it("writes parse/bind/describe/execute per statement and one sync while writable", () => {
+  it("brackets the statements in BEGIN/COMMIT and writes one sync while writable", () => {
     const { connection, frames } = fakeConnection(true);
     const batch = new PgBatchQuery([{ sql: "select 1" }, { sql: "select 2" }], () => {});
 
@@ -637,11 +656,19 @@ describe("PgBatchQuery.submit", () => {
 
     expect(frames).toEqual([
       "cork",
-      "parse",
+      "parse(BEGIN)",
+      "bind",
+      "describe",
+      "execute",
+      "parse(select 1)",
       "bind",
       "describe",
       "execute",
-      "parse",
+      "parse(select 2)",
+      "bind",
+      "describe",
+      "execute",
+      "parse(COMMIT)",
       "bind",
       "describe",
       "execute",
@@ -650,6 +677,18 @@ describe("PgBatchQuery.submit", () => {
     ]);
     expect(batch.outcome).toBe("submitted");
   });
+
+  it("counts neither BEGIN's nor COMMIT's completion toward the statement index", () => {
+    const batch = new PgBatchQuery([{ sql: "select 1" }, { sql: "select 2" }], () => {});
+
+    batch.handleCommandComplete();
+    expect(batch.completed).toBe(0);
+    batch.handleCommandComplete();
+    batch.handleEmptyQuery();
+    expect(batch.completed).toBe(2);
+    batch.handleCommandComplete();
+    expect(batch.completed).toBe(2);
+  });
 });
 
 describe("batchFailureError", () => {
@@ -725,6 +764,124 @@ describe("batchFailureError", () => {
     expect(error).toMatchObject({ message: "Error: serialization blew up", statementIndex: 0 });
   });
 
+  it("names the transaction start when the server rejected the batch before BEGIN completed", () => {
+    const beginRejected = new SqlError({
+      reason: new SqlSyntaxError({
+        cause: Object.assign(new Error("canceling statement due to statement timeout"), {
+          severity: "ERROR",
+          code: "57014",
+        }),
+        message: "Failed to execute statement",
+        operation: "execute",
+      }),
+    });
+    const error = batchFailureError(
+      beginRejected,
+      { completed: 0, outcome: "submitted", began: false },
+      true,
+    );
+
+    expect(error).toBeInstanceOf(DbExecError);
+    expect(error).toMatchObject({
+      message:
+        "failed to begin the batch transaction: " +
+        "ERROR: canceling statement due to statement timeout (SQLSTATE 57014)",
+      statementIndex: 0,
+      transactionPhase: "begin",
+    });
+
+    const lost = batchFailureError(
+      new Error("Connection terminated unexpectedly"),
+      { completed: 0, outcome: "submitted", began: false },
+      true,
+    );
+    expect(lost).toBeInstanceOf(DbExecError);
+    // A connection lost at BEGIN keeps its own reason, but still marks the phase:
+    // no caller statement ran, so nothing downstream may render `At statement: 0`.
+    expect(lost).toMatchObject({
+      message: "Error: Connection terminated unexpectedly",
+      statementIndex: 0,
+      transactionPhase: "begin",
+    });
+
+    const terminated = batchFailureError(
+      new SqlError({
+        reason: new SqlSyntaxError({
+          cause: Object.assign(new Error("terminating connection due to idle-session timeout"), {
+            severity: "FATAL",
+            code: "57P05",
+          }),
+          message: "Failed to execute statement",
+          operation: "execute",
+        }),
+      }),
+      { completed: 0, outcome: "submitted", began: false },
+      true,
+    );
+    expect(terminated).toBeInstanceOf(DbExecError);
+    expect(terminated).toMatchObject({
+      message: "FATAL: terminating connection due to idle-session timeout (SQLSTATE 57P05)",
+      transactionPhase: "begin",
+    });
+
+    const poisoned = batchFailureError(
+      beginRejected,
+      { completed: 0, outcome: "poisoned", began: false },
+      true,
+    );
+    expect(poisoned).toBeInstanceOf(DbExecError);
+    expect(poisoned.message).toBe(
+      "ERROR: canceling statement due to statement timeout (SQLSTATE 57014)",
+    );
+    // A poisoned batch never reached the server, so it stays on the statement path.
+    expect(poisoned).not.toHaveProperty("transactionPhase");
+  });
+
+  it("names the transaction commit when a deferred failure lands on COMMIT", () => {
+    const deferred = new SqlError({
+      reason: new SqlSyntaxError({
+        cause: Object.assign(new Error("deferred constraint failed"), {
+          severity: "ERROR",
+          code: "23514",
+        }),
+        message: "Failed to execute statement",
+        operation: "execute",
+      }),
+    });
+    const error = batchFailureError(
+      deferred,
+      { completed: 2, outcome: "submitted", began: true, atCommit: true },
+      true,
+    );
+    expect(error).toBeInstanceOf(DbExecError);
+    expect(error).toMatchObject({
+      message:
+        "failed to commit the batch transaction: ERROR: deferred constraint failed (SQLSTATE 23514)",
+      statementIndex: 2,
+      transactionPhase: "commit",
+    });
+
+    const dropped = batchFailureError(
+      new SqlError({
+        reason: new SqlSyntaxError({
+          cause: Object.assign(new Error("terminating connection: database dropped"), {
+            severity: "FATAL",
+            code: "57P04",
+          }),
+          message: "Failed to execute statement",
+          operation: "execute",
+        }),
+      }),
+      { completed: 2, outcome: "submitted", began: true, atCommit: true },
+      true,
+    );
+    expect(dropped).toBeInstanceOf(DbExecError);
+    expect(dropped).toMatchObject({
+      message: "FATAL: terminating connection: database dropped (SQLSTATE 57P04)",
+      transactionPhase: "commit",
+    });
+  });
+
   it("keeps server-error mapping and the completed count for a statement failure", () => {
     const error = batchFailureError(
       new SqlError({
@@ -756,20 +913,38 @@ describe("batchFailureError", () => {
 
 describe("shouldDiscardBatchClient", () => {
   it("discards a client whose batch never reached the wire", () => {
-    expect(shouldDiscardBatchClient({ outcome: "unsent" }, Exit.succeed(undefined))).toBe(true);
+    expect(shouldDiscardBatchClient({ outcome: "unsent" }, Exit.succeed(undefined), false)).toBe(
+      true,
+    );
   });
 
-  it("returns a client to the pool once its batch was written, error or not", () => {
-    expect(shouldDiscardBatchClient({ outcome: "submitted" }, Exit.succeed(undefined))).toBe(false);
+  it("keeps a written batch's client on success or once its failure rolled back", () => {
+    expect(shouldDiscardBatchClient({ outcome: "submitted" }, Exit.succeed(undefined), false)).toBe(
+      false,
+    );
     expect(
-      shouldDiscardBatchClient({ outcome: "submitted" }, Exit.fail(new Error("server said no"))),
+      shouldDiscardBatchClient(
+        { outcome: "submitted" },
+        Exit.fail(new Error("server said no")),
+        true,
+      ),
     ).toBe(false);
   });
 
+  it("discards a written batch's client when its failure was not rolled back", () => {
+    expect(
+      shouldDiscardBatchClient(
+        { outcome: "submitted" },
+        Exit.fail(new Error("server said no")),
+        false,
+      ),
+    ).toBe(true);
+  });
+
   it("discards a client whose batch was interrupted or died mid-flight", () => {
-    expect(shouldDiscardBatchClient({ outcome: "submitted" }, Exit.interrupt(1))).toBe(true);
-    expect(shouldDiscardBatchClient({ outcome: "submitted" }, Exit.die("boom"))).toBe(true);
+    expect(shouldDiscardBatchClient({ outcome: "submitted" }, Exit.interrupt(1), true)).toBe(true);
+    expect(shouldDiscardBatchClient({ outcome: "submitted" }, Exit.die("boom"), true)).toBe(true);
     // Interrupted before the batch was even constructed: no batch, still discard.
-    expect(shouldDiscardBatchClient(undefined, Exit.interrupt(1))).toBe(true);
+    expect(shouldDiscardBatchClient(undefined, Exit.interrupt(1), false)).toBe(true);
   });
 });
diff --git a/apps/cli/src/command-internal/migration-apply.ts b/apps/cli/src/command-internal/migration-apply.ts
index 8cfb3dabf5..a7c8225dae 100644
--- a/apps/cli/src/command-internal/migration-apply.ts
+++ b/apps/cli/src/command-internal/migration-apply.ts
@@ -61,9 +61,24 @@ const BOM_CODE_POINT = 0xfeff;
 const CREATE_INDEX_CONCURRENTLY_PATTERN = /^CREATE\s+(?:UNIQUE\s+)?INDEX\s+CONCURRENTLY(?:\s|$)/u;
 const DROP_INDEX_CONCURRENTLY_PATTERN = /^DROP\s+INDEX\s+CONCURRENTLY(?:\s|$)/u;
 const REINDEX_CONCURRENTLY_PATTERN = /^REINDEX(?:\s|\().*\sCONCURRENTLY(?:\s|$)/u;
+const REINDEX_OPTION_CONCURRENTLY_PATTERN = /^REINDEX\s*\([^)]*\bCONCURRENTLY\b[^)]*\)/u;
 const VACUUM_PATTERN = /^VACUUM(?:\s|\(|$)/u;
 const ALTER_SYSTEM_PATTERN = /^ALTER\s+SYSTEM(?:\s|$)/u;
 const CLUSTER_PATTERN = /^CLUSTER(?:\s|$)/u;
+const DATABASE_DDL_PATTERN = /^(?:CREATE|DROP)\s+DATABASE(?:\s|$)/u;
+const TABLESPACE_DDL_PATTERN = /^(?:CREATE|DROP)\s+TABLESPACE(?:\s|$)/u;
+const REINDEX_DATABASE_PATTERN = /^REINDEX(?:\s*\([^)]*\))?\s+(?:DATABASE|SYSTEM|SCHEMA)(?:\s|$)/u;
+const SUBSCRIPTION_DDL_PATTERN = /^(?:CREATE|DROP)\s+SUBSCRIPTION(?:\s|$)/u;
+const DISCARD_ALL_PATTERN = /^DISCARD\s+ALL(?:\s|$)/u;
+const ALTER_DATABASE_TABLESPACE_PATTERN = /^ALTER\s+DATABASE\s[\s\S]*\sSET\s+TABLESPACE(?:\s|$)/u;
+const ALTER_SUBSCRIPTION_REFRESH_PATTERN =
+  /^ALTER\s+SUBSCRIPTION\s[\s\S]*\s(?:REFRESH\s+PUBLICATION|(?:SET|ADD|DROP)\s+PUBLICATION)(?:\s|$)/u;
+const ALL_IN_TABLESPACE_PATTERN =
+  /^ALTER\s+(?:TABLE|INDEX|MATERIALIZED\s+VIEW)\s+ALL\s+IN\s+TABLESPACE(?:\s|$)/u;
+const DETACH_PARTITION_PATTERN =
+  /^ALTER\s+TABLE\s[\s\S]*\sDETACH\s+PARTITION\s[\s\S]*\sCONCURRENTLY(?:\s|$)/u;
+const REFRESH_MATERIALIZED_VIEW_CONCURRENTLY_PATTERN =
+  /^REFRESH\s+MATERIALIZED\s+VIEW\s+CONCURRENTLY(?:\s|$)/u;
 const TRANSACTION_CONTROL_PATTERN =
   /^(?:BEGIN|START\s+TRANSACTION|COMMIT|END|ABORT|PREPARE\s+TRANSACTION)(?:\s|$)/u;
 
@@ -97,10 +112,26 @@ const trimLeadingSqlComments = (sql: string): string => {
 /**
  * Whether a migration statement cannot run inside a transaction block — `CREATE
  * [UNIQUE] INDEX CONCURRENTLY`, `DROP INDEX CONCURRENTLY`, `REINDEX … CONCURRENTLY`,
- * `VACUUM`, `ALTER SYSTEM`, `CLUSTER`. Such statements fail with SQLSTATE 25001
- * inside the implicit transaction
+ * `VACUUM`, `ALTER SYSTEM`, `CLUSTER`, `CREATE`/`DROP DATABASE`,
+ * `CREATE`/`DROP TABLESPACE`, `REINDEX DATABASE`/`SYSTEM`/`SCHEMA`,
+ * `CREATE`/`DROP SUBSCRIPTION`, `DISCARD ALL`,
+ * `ALTER DATABASE … SET TABLESPACE`,
+ * `ALTER SUBSCRIPTION … REFRESH`/`SET`/`ADD`/`DROP PUBLICATION`,
+ * `ALTER TABLE … DETACH PARTITION … CONCURRENTLY`,
+ * `ALTER TABLE`/`INDEX`/`MATERIALIZED VIEW ALL IN TABLESPACE`,
+ * `REFRESH MATERIALIZED VIEW CONCURRENTLY`. Such statements (in
+ * their default forms) fail with
+ * SQLSTATE 25001 inside the transaction
  * created by a migration batch, so `execMigrationBatch` runs them standalone.
- * Port of `isPipelineIncompatible` (`pkg/migration/file.go`, supabase/cli#5156).
+ * Port of `isPipelineIncompatible` (`pkg/migration/file.go`, supabase/cli#5156),
+ * extended with the remaining statement kinds PostgreSQL refuses inside the
+ * explicit transaction the batch runs in since supabase/cli#6347.
+ *
+ * Deliberately loose, keyword-anchored matching — never identifier-aware: a match
+ * inside a comment or literal over-routes the statement standalone, which is always
+ * valid SQL placement and at worst adds a documented flush boundary, while an
+ * under-match is a hard SQLSTATE 25001 failure. Do not tighten these into
+ * identifier parsing.
  */
 export const isPipelineIncompatible = (sql: string): boolean => {
   const upper = trimLeadingSqlComments(sql).toUpperCase();
@@ -108,9 +139,20 @@ export const isPipelineIncompatible = (sql: string): boolean => {
     CREATE_INDEX_CONCURRENTLY_PATTERN.test(upper) ||
     DROP_INDEX_CONCURRENTLY_PATTERN.test(upper) ||
     REINDEX_CONCURRENTLY_PATTERN.test(upper) ||
+    REINDEX_OPTION_CONCURRENTLY_PATTERN.test(upper) ||
     VACUUM_PATTERN.test(upper) ||
     ALTER_SYSTEM_PATTERN.test(upper) ||
-    CLUSTER_PATTERN.test(upper)
+    CLUSTER_PATTERN.test(upper) ||
+    DATABASE_DDL_PATTERN.test(upper) ||
+    TABLESPACE_DDL_PATTERN.test(upper) ||
+    REINDEX_DATABASE_PATTERN.test(upper) ||
+    SUBSCRIPTION_DDL_PATTERN.test(upper) ||
+    DISCARD_ALL_PATTERN.test(upper) ||
+    ALTER_DATABASE_TABLESPACE_PATTERN.test(upper) ||
+    ALTER_SUBSCRIPTION_REFRESH_PATTERN.test(upper) ||
+    DETACH_PARTITION_PATTERN.test(upper) ||
+    ALL_IN_TABLESPACE_PATTERN.test(upper) ||
+    REFRESH_MATERIALIZED_VIEW_CONCURRENTLY_PATTERN.test(upper)
   );
 };
 
@@ -527,8 +569,8 @@ const formattedExecBatchDbError = (error: unknown): DbExecError | undefined => {
 
 /**
  * Runs a single migration/seed file's statements (plus the optional history insert).
- * Statements run inside an implicitly transactional extended-protocol batch,
- * except pipeline-incompatible ones
+ * Statements run inside an explicitly transactional extended-protocol batch
+ * (supabase/cli#6347), except pipeline-incompatible ones
  * (`isPipelineIncompatible` — `CREATE INDEX CONCURRENTLY`, `VACUUM`, …) which
  * cannot run in a transaction block: the open batch is flushed (committed), the
  * statement runs standalone, then batching resumes (supabase/cli#5156). The history
@@ -694,11 +736,14 @@ const execMigrationBatch = <E>(
       // (Go threads the same counter through `ExecBatch`).
       let pending: Array<string> = [];
       let executed = 0;
+      let standaloneRestored = false;
 
       const flushBatch = (final: boolean) =>
         Effect.gen(function* () {
           const recordVersion = final && version.length > 0;
-          const trailingRestore = final ? restoreRole : undefined;
+          // A standalone statement that just restored the role needs no trailing repeat.
+          const trailingRestore =
+            final && !(pending.length === 0 && standaloneRestored) ? restoreRole : undefined;
           if (pending.length === 0 && !recordVersion && trailingRestore === undefined) return;
           const batchStatements = pending;
           const operations: Array<DbBatchStatement> = [];
@@ -742,6 +787,16 @@ const execMigrationBatch = <E>(
               // `statementIndex` is set by every batch failure the driver raises; a
               // session that omits it can only have failed before the first statement.
               const raw = cause.statementIndex ?? 0;
+              // A wrapper (BEGIN/COMMIT) failure, or one past every operation
+              // (e.g. a deferred constraint), blames no statement: keep the
+              // driver's phase-labeled message without an `At statement` tail.
+              if (cause.transactionPhase !== undefined || raw >= operations.length) {
+                const msg = [errorMessage(cause)];
+                if (cause.detail !== undefined && cause.detail.length > 0) {
+                  msg.push(cause.detail);
+                }
+                return formattedExecBatchFailure(msg.join("\n"), cause);
+              }
               const globalIndex = base + raw - (injectedBefore[raw] ?? injected);
               return formatExecBatchError(
                 cause,
@@ -763,9 +818,17 @@ const execMigrationBatch = <E>(
           yield* session
             .exec(statement)
             .pipe(Effect.mapError((cause) => formatExecBatchError(cause, index, statement)));
+          standaloneRestored = false;
+          if (restoreRole !== undefined && revertsToLoginRole(statement)) {
+            yield* session
+              .exec(restoreRole)
+              .pipe(Effect.mapError((cause) => formatExecBatchError(cause, index, restoreRole)));
+            standaloneRestored = true;
+          }
           executed += 1;
         } else {
           pending.push(statement);
+          standaloneRestored = false;
         }
       }
       yield* flushBatch(true);
diff --git a/apps/cli/src/command-internal/migration-apply.unit.test.ts b/apps/cli/src/command-internal/migration-apply.unit.test.ts
index 9e183196b3..79aaf95114 100644
--- a/apps/cli/src/command-internal/migration-apply.unit.test.ts
+++ b/apps/cli/src/command-internal/migration-apply.unit.test.ts
@@ -31,6 +31,7 @@ class FakeExecError extends Data.TaggedError("DbExecError")<{
   readonly detail?: string;
   readonly position?: number;
   readonly statementIndex?: number;
+  readonly transactionPhase?: "begin" | "commit";
 }> {
   get [ErrorActionabilityId](): CliErrorActionabilityDeclaration {
     return actionability.dbFinding;
@@ -41,7 +42,13 @@ function fakeSession(
   opts: {
     failOn?: string;
     failAfterBatch?: boolean;
-    failWith?: { message: string; code?: string; detail?: string; position?: number };
+    failWith?: {
+      message: string;
+      code?: string;
+      detail?: string;
+      position?: number;
+      transactionPhase?: "begin" | "commit";
+    };
     restoreRoleSql?: string;
     batchConnectionLost?: string;
   } = {},
@@ -284,7 +291,58 @@ describe("applyMigrationFile", () => {
     );
   });
 
-  it.effect("defaults a deferred batch failure to the migration history statement", () => {
+  it.effect("reports a begin failure without blaming the first statement", () => {
+    const dir = mkdtempSync(join(tmpdir(), "apply-"));
+    const file = join(dir, "20240101120000_begin.sql");
+    writeFileSync(file, "SELECT 1;");
+    const { session } = fakeSession({
+      failOn: "SELECT 1",
+      failWith: {
+        message: "failed to begin the batch transaction: ERROR: canceling statement",
+        transactionPhase: "begin",
+      },
+    });
+    return run(session, file).pipe(
+      Effect.flip,
+      Effect.tap((error) =>
+        Effect.sync(() => {
+          expect(error.message).toContain("failed to begin the batch transaction");
+          expect(error.message).not.toContain("At statement");
+          expect(error.message).not.toContain("SELECT 1");
+        }),
+      ),
+      Effect.ensuring(Effect.sync(() => rmSync(dir, { recursive: true, force: true }))),
+    );
+  });
+
+  it.effect("reports a connection lost at BEGIN without blaming the first statement", () => {
+    // The driver marks the begin phase without relabeling the message when the
+    // connection died before BEGIN completed; the caller's statement never ran,
+    // so the formatter must surface the loss verbatim with no statement echo.
+    const dir = mkdtempSync(join(tmpdir(), "apply-"));
+    const file = join(dir, "20240101120000_begin_lost.sql");
+    writeFileSync(file, "SELECT 1;");
+    const { session } = fakeSession({
+      failOn: "SELECT 1",
+      failWith: {
+        message: "Error: Connection terminated unexpectedly",
+        transactionPhase: "begin",
+      },
+    });
+    return run(session, file).pipe(
+      Effect.flip,
+      Effect.tap((error) =>
+        Effect.sync(() => {
+          expect(error.message).toContain("Connection terminated unexpectedly");
+          expect(error.message).not.toContain("At statement");
+          expect(error.message).not.toContain("SELECT 1");
+        }),
+      ),
+      Effect.ensuring(Effect.sync(() => rmSync(dir, { recursive: true, force: true }))),
+    );
+  });
+
+  it.effect("reports a deferred commit failure without blaming a statement", () => {
     const dir = mkdtempSync(join(tmpdir(), "apply-"));
     const file = join(dir, "20240101120000_deferred.sql");
     writeFileSync(file, "SELECT 1;");
@@ -293,8 +351,8 @@ describe("applyMigrationFile", () => {
       Effect.flip,
       Effect.tap((error) =>
         Effect.sync(() => {
-          expect(error.message).toContain("At statement: 2");
-          expect(error.message).toContain("INSERT INTO supabase_migrations.schema_migrations");
+          expect(error.message).not.toContain("At statement");
+          expect(error.message).not.toContain("INSERT INTO supabase_migrations.schema_migrations");
         }),
       ),
       Effect.ensuring(Effect.sync(() => rmSync(dir, { recursive: true, force: true }))),
@@ -668,10 +726,9 @@ describe("applyMigrationFile", () => {
     );
   });
 
-  it.effect("keeps the deferred-failure index when a restore op is appended", () => {
-    // Mirrors "defaults a deferred batch failure to the migration history
-    // statement": the restore op between the statements and the insert must not
-    // shift the deferred (post-Sync) index either.
+  it.effect("reports a deferred commit failure cleanly when a restore op is appended", () => {
+    // The restore op between the statements and the insert must not resurrect a
+    // statement tail for a commit-phase failure.
     const dir = mkdtempSync(join(tmpdir(), "apply-"));
     const file = join(dir, "20240101120000_deferred.sql");
     writeFileSync(file, "SELECT 1;");
@@ -683,8 +740,8 @@ describe("applyMigrationFile", () => {
       Effect.flip,
       Effect.tap((error) =>
         Effect.sync(() => {
-          expect(error.message).toContain("At statement: 2");
-          expect(error.message).toContain("INSERT INTO supabase_migrations.schema_migrations");
+          expect(error.message).not.toContain("At statement");
+          expect(error.message).not.toContain("INSERT INTO supabase_migrations.schema_migrations");
           rmSync(dir, { recursive: true, force: true });
         }),
       ),
@@ -822,7 +879,7 @@ describe("applyMigrationFile", () => {
     );
   });
 
-  it.effect("keeps the deferred-failure index when mid-file restores were injected", () => {
+  it.effect("reports a deferred commit failure cleanly with mid-file restores injected", () => {
     // The `injectedBefore[raw] ?? injected` fallback only matters when the
     // deferred (post-Sync) index lands past the ops array AND injections exist.
     const dir = mkdtempSync(join(tmpdir(), "apply-"));
@@ -836,8 +893,8 @@ describe("applyMigrationFile", () => {
       Effect.flip,
       Effect.tap((error) =>
         Effect.sync(() => {
-          expect(error.message).toContain("At statement: 4");
-          expect(error.message).toContain("INSERT INTO supabase_migrations.schema_migrations");
+          expect(error.message).not.toContain("At statement");
+          expect(error.message).not.toContain("INSERT INTO supabase_migrations.schema_migrations");
           rmSync(dir, { recursive: true, force: true });
         }),
       ),
@@ -868,6 +925,49 @@ describe("applyMigrationFile", () => {
     );
   });
 
+  it.effect("re-asserts postgres right after a standalone role-reverting statement", () => {
+    const dir = mkdtempSync(join(tmpdir(), "apply-"));
+    const file = join(dir, "20240101120000_discard.sql");
+    writeFileSync(file, "select 1;\nDISCARD ALL;\nselect 2;");
+    const { session, calls } = fakeSession({ restoreRoleSql: "SET SESSION ROLE postgres" });
+    return run(session, file).pipe(
+      Effect.tap(() =>
+        Effect.sync(() => {
+          const execs = calls.filter((call) => call.kind === "exec").map((call) => call.sql);
+          expect(execs.slice(-2)).toEqual(["DISCARD ALL", "SET SESSION ROLE postgres"]);
+          const batches = calls.filter((call) => call.kind === "batch");
+          expect(batches[0]?.statements?.map(({ sql }) => sql)).toEqual(["select 1"]);
+          expect(batches[1]?.statements?.map(({ sql }) => sql)).toEqual([
+            "select 2",
+            "SET SESSION ROLE postgres",
+            expect.stringContaining("supabase_migrations.schema_migrations"),
+          ]);
+          rmSync(dir, { recursive: true, force: true });
+        }),
+      ),
+    );
+  });
+
+  it.effect("sends no trailing restore when the file ends on a restored standalone", () => {
+    const dir = mkdtempSync(join(tmpdir(), "apply-"));
+    const file = join(dir, "20240101120000_discard_last.sql");
+    writeFileSync(file, "select 1;\nDISCARD ALL;");
+    const { session, calls } = fakeSession({ restoreRoleSql: "SET SESSION ROLE postgres" });
+    return run(session, file).pipe(
+      Effect.tap(() =>
+        Effect.sync(() => {
+          const execs = calls.filter((call) => call.kind === "exec").map((call) => call.sql);
+          expect(execs.slice(-2)).toEqual(["DISCARD ALL", "SET SESSION ROLE postgres"]);
+          const batches = calls.filter((call) => call.kind === "batch");
+          expect(batches.at(-1)?.statements?.map(({ sql }) => sql)).toEqual([
+            expect.stringContaining("supabase_migrations.schema_migrations"),
+          ]);
+          rmSync(dir, { recursive: true, force: true });
+        }),
+      ),
+    );
+  });
+
   it.effect("reports a mid-file restore's own failure at its host statement", () => {
     const dir = mkdtempSync(join(tmpdir(), "apply-"));
     const file = join(dir, "20240101120000_fail.sql");
@@ -1126,6 +1226,85 @@ describe("isPipelineIncompatible", () => {
     ["vacuum with options", "VACUUM (FULL, ANALYZE) public.widgets", true],
     ["alter system", "ALTER SYSTEM SET wal_level = 'logical'", true],
     ["cluster", "CLUSTER public.widgets USING widgets_id_idx", true],
+    ["create database", "CREATE DATABASE demo", true],
+    ["drop database", "DROP DATABASE IF EXISTS demo", true],
+    ["create tablespace", "CREATE TABLESPACE ts LOCATION '/tmp/ts'", true],
+    ["drop tablespace", "DROP TABLESPACE ts", true],
+    ["reindex database", "REINDEX DATABASE postgres", true],
+    ["reindex system with options", "REINDEX (VERBOSE) SYSTEM postgres", true],
+    ["reindex database adjacent options", "REINDEX(VERBOSE) DATABASE postgres", true],
+    ["reindex concurrently as option", "REINDEX (CONCURRENTLY) INDEX widgets_id_idx", true],
+    ["reindex mixed option list", "REINDEX (VERBOSE, CONCURRENTLY) TABLE public.widgets", true],
+    [
+      "reindex concurrently false over-routes conservatively",
+      "REINDEX (CONCURRENTLY FALSE) INDEX widgets_id_idx",
+      true,
+    ],
+    ["reindex verbose option only", "REINDEX (VERBOSE) INDEX widgets_id_idx", false],
+    ["reindex schema", "REINDEX SCHEMA public", true],
+    ["reindex table non-concurrent", "REINDEX TABLE public.widgets", false],
+    ["alter database", "ALTER DATABASE demo SET search_path = public", false],
+    ["alter database set tablespace", "ALTER DATABASE demo SET TABLESPACE fast", true],
+    [
+      "alter table all in tablespace",
+      "ALTER TABLE ALL IN TABLESPACE old_ts SET TABLESPACE fast",
+      true,
+    ],
+    [
+      "alter index all in tablespace",
+      "ALTER INDEX ALL IN TABLESPACE old_ts SET TABLESPACE fast",
+      true,
+    ],
+    [
+      "alter materialized view all in tablespace",
+      "ALTER MATERIALIZED VIEW ALL IN TABLESPACE old_ts SET TABLESPACE fast",
+      true,
+    ],
+    ["alter table set tablespace single", "ALTER TABLE t SET TABLESPACE fast", false],
+    ["alter database set tablespace multiline", "ALTER DATABASE demo\n SET TABLESPACE fast", true],
+    [
+      "alter database with tablespace inside a literal over-routes conservatively",
+      "ALTER DATABASE demo SET application_name TO 'foo SET TABLESPACE bar'",
+      true,
+    ],
+    ["detach partition concurrently", "ALTER TABLE m DETACH PARTITION p CONCURRENTLY", true],
+    ["detach partition finalize stays batched", "ALTER TABLE m DETACH PARTITION p FINALIZE", false],
+    ["detach partition plain", "ALTER TABLE m DETACH PARTITION p", false],
+    [
+      "detach inside a comment over-routes conservatively",
+      "ALTER TABLE m ADD COLUMN x int /* DETACH PARTITION p CONCURRENTLY */",
+      true,
+    ],
+    [
+      "detach partition qualified concurrently",
+      "ALTER TABLE IF EXISTS ONLY s.m\n DETACH PARTITION p CONCURRENTLY",
+      true,
+    ],
+    [
+      "detach partition quoted qualified concurrently",
+      'ALTER TABLE "tenant schema".events DETACH PARTITION "p 1" CONCURRENTLY',
+      true,
+    ],
+    [
+      "detach partition spaced qualification",
+      'ALTER TABLE "tenant schema" . events DETACH PARTITION p CONCURRENTLY',
+      true,
+    ],
+    ["create subscription", "CREATE SUBSCRIPTION sub CONNECTION 'host=h' PUBLICATION pub", true],
+    ["drop subscription", "DROP SUBSCRIPTION IF EXISTS sub", true],
+    ["alter subscription", "ALTER SUBSCRIPTION sub DISABLE", false],
+    ["alter subscription refresh", "ALTER SUBSCRIPTION sub REFRESH PUBLICATION", true],
+    ["alter subscription set publication", "ALTER SUBSCRIPTION sub SET PUBLICATION p", true],
+    ["alter subscription refresh multiline", "ALTER SUBSCRIPTION sub\n REFRESH PUBLICATION", true],
+    ["alter subscription set options", "ALTER SUBSCRIPTION sub SET (slot_name = 's')", false],
+    ["discard all", "DISCARD ALL", true],
+    ["discard temp", "DISCARD TEMP", false],
+    [
+      "refresh materialized view concurrently",
+      "REFRESH MATERIALIZED VIEW CONCURRENTLY public.mv",
+      true,
+    ],
+    ["plain refresh materialized view", "REFRESH MATERIALIZED VIEW public.mv", false],
     [
       "lower-case create index concurrently",
       "create index concurrently widgets_id_idx on public.widgets(id)",
diff --git a/apps/cli/src/commands/db/push/SIDE_EFFECTS.md b/apps/cli/src/commands/db/push/SIDE_EFFECTS.md
index 03616d31b0..6824a61f49 100644
--- a/apps/cli/src/commands/db/push/SIDE_EFFECTS.md
+++ b/apps/cli/src/commands/db/push/SIDE_EFFECTS.md
@@ -25,14 +25,14 @@ before migrations unless `--skip-vault` is set.
 
 ## Database Mutations
 
-| Statement                                                                                                           | When                                                                                                                                                                                                                                    |
-| ------------------------------------------------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
-| `RESET ALL` + migration statements + `INSERT INTO supabase_migrations.schema_migrations(version, name, statements)` | per pending migration (after confirmation); compatible statements use an implicit extended-protocol batch with one final `Sync`, while pipeline-incompatible statements run standalone — see Notes                                      |
-| `CREATE SCHEMA/TABLE … supabase_migrations.schema_migrations`, `ALTER TABLE … ADD COLUMN …`                         | once before applying migrations, when a read-only probe finds the ledger not yet provisioned (idempotent; supabase/cli#6393)                                                                                                            |
-| `roles.sql` statements (no history row)                                                                             | per `--include-roles` globals file (after confirmation); statements use an implicit extended-protocol batch with one final `Sync`                                                                                                       |
-| `SELECT id, name FROM vault.secrets …`, `SELECT vault.update_secret(...)`, `SELECT vault.create_secret(...)`        | when `[db.vault]` has syncable secrets, migrations are applied, and `--skip-vault` is not set                                                                                                                                           |
-| `CREATE TABLE … supabase_migrations.seed_files`, seed statements, `INSERT … seed_files(path, hash) … ON CONFLICT …` | per pending seed file with `--include-seed` (after confirmation; the `seed_files` DDL only when a read-only probe finds that ledger not yet provisioned); a dirty seed only refreshes the hash                                          |
-| `SET SESSION ROLE postgres`                                                                                         | stepped-down sessions only (`cli_login_*`/`supabase_admin`): after each top-level role-reverting statement, at the end of each migration/globals/seed file, and before the history insert and the `seed_files` upsert (CLI-2205, #6236) |
+| Statement                                                                                                           | When                                                                                                                                                                                                                                                         |
+| ------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
+| `RESET ALL` + migration statements + `INSERT INTO supabase_migrations.schema_migrations(version, name, statements)` | per pending migration (after confirmation); compatible statements use one explicitly transactional extended-protocol batch (`BEGIN` … `COMMIT`) with one final `Sync`, while pipeline-incompatible statements run standalone — see Notes                     |
+| `CREATE SCHEMA/TABLE … supabase_migrations.schema_migrations`, `ALTER TABLE … ADD COLUMN …`                         | once before applying migrations, when a read-only probe finds the ledger not yet provisioned (idempotent; supabase/cli#6393)                                                                                                                                 |
+| `roles.sql` statements (no history row)                                                                             | per `--include-roles` globals file (after confirmation); compatible statements use one explicitly transactional extended-protocol batch (`BEGIN` … `COMMIT`) with one final `Sync`, with the same standalone/sequential exceptions as migrations — see Notes |
+| `SELECT id, name FROM vault.secrets …`, `SELECT vault.update_secret(...)`, `SELECT vault.create_secret(...)`        | when `[db.vault]` has syncable secrets, migrations are applied, and `--skip-vault` is not set                                                                                                                                                                |
+| `CREATE TABLE … supabase_migrations.seed_files`, seed statements, `INSERT … seed_files(path, hash) … ON CONFLICT …` | per pending seed file with `--include-seed` (after confirmation; the `seed_files` DDL only when a read-only probe finds that ledger not yet provisioned); a dirty seed only refreshes the hash                                                               |
+| `SET SESSION ROLE postgres`                                                                                         | stepped-down sessions only (`cli_login_*`/`supabase_admin`): after each top-level role-reverting statement, at the end of each migration/globals/seed file, and before the history insert and the `seed_files` upsert (CLI-2205, #6236)                      |
 
 ## API Routes
 
@@ -110,11 +110,19 @@ stdout is payload-only. A single `result` object is emitted:
   load, including decrypted `encrypted:` values. `--skip-vault` leaves them unchanged
   and does not resolve or decrypt their configured values.
 - **Pipeline-incompatible statements**: `CREATE [UNIQUE] INDEX CONCURRENTLY`,
-  `REINDEX … CONCURRENTLY`, `VACUUM`, `ALTER SYSTEM`, and `CLUSTER` cannot run inside a
+  `DROP INDEX CONCURRENTLY`, `REINDEX … CONCURRENTLY`, `VACUUM`, `ALTER SYSTEM`, `CLUSTER`,
+  `CREATE`/`DROP DATABASE`, `CREATE`/`DROP TABLESPACE`, `REINDEX DATABASE`/`SYSTEM`/`SCHEMA`,
+  `CREATE`/`DROP SUBSCRIPTION`, `DISCARD ALL`, `ALTER DATABASE … SET TABLESPACE`, and
+  `ALTER SUBSCRIPTION … REFRESH`/`SET`/`ADD`/`DROP PUBLICATION`,
+  `ALTER TABLE … DETACH PARTITION … CONCURRENTLY`,
+  `ALTER TABLE`/`INDEX`/`MATERIALIZED VIEW ALL IN TABLESPACE`, and
+  `REFRESH MATERIALIZED VIEW CONCURRENTLY`
+  cannot run inside a
   transaction block (SQLSTATE 25001). The apply flushes (commits) the open batch, runs
   the statement standalone outside any transaction, then resumes batching; the history
   insert stays in the final batch so the migration is recorded only after every
-  statement succeeds. Atomicity is therefore lost at each flush boundary: statements
+  statement succeeds. A failed batch's transaction is rolled back (bounded) before its
+  connection is reused; a rollback that fails or times out discards the connection. Atomicity is therefore lost at each flush boundary: statements
   committed in an earlier batch are **not** rolled back if a later statement fails,
   leaving the database partially migrated with **no history row** — a re-run replays
   the whole file from the top (which may then fail on already-applied statements).
diff --git a/apps/cli/src/commands/db/reset/SIDE_EFFECTS.md b/apps/cli/src/commands/db/reset/SIDE_EFFECTS.md
index 9be8bb75b5..7215aa1170 100644
--- a/apps/cli/src/commands/db/reset/SIDE_EFFECTS.md
+++ b/apps/cli/src/commands/db/reset/SIDE_EFFECTS.md
@@ -150,7 +150,10 @@ reported as a lost connection (with the driver's own reason and, locally, the hi
 stack) rather than against a statement that never ran. Once any part of the batch has been written,
 and for the pipeline-incompatible statements the same loop runs on their own (`CREATE INDEX
 CONCURRENTLY`, `VACUUM`, ...), a failure still reports as `At statement: N` with the statement
-echoed, because those may genuinely have reached the server.
+echoed, because those may genuinely have reached the server. The one exception is a failure of the
+batch's own transaction wrapper — a rejected `BEGIN`, or a deferred constraint surfacing at
+`COMMIT` — which reports the phase-labeled driver message with no statement context, since no
+caller statement is to blame.
 
 ## Exit Codes
 
diff --git a/apps/cli/src/commands/migration/up/SIDE_EFFECTS.md b/apps/cli/src/commands/migration/up/SIDE_EFFECTS.md
index 8a700a4c12..932a2bdc1e 100644
--- a/apps/cli/src/commands/migration/up/SIDE_EFFECTS.md
+++ b/apps/cli/src/commands/migration/up/SIDE_EFFECTS.md
@@ -65,10 +65,18 @@ Same structured `applied` result delivered as an NDJSON `result` event.
   a non-linked target).
 - `--include-all` applies all migrations not found on the remote history table.
 - Pipeline-incompatible statements (`CREATE [UNIQUE] INDEX CONCURRENTLY`,
-  `REINDEX … CONCURRENTLY`, `VACUUM`, `ALTER SYSTEM`, `CLUSTER`) run standalone outside
+  `DROP INDEX CONCURRENTLY`, `REINDEX … CONCURRENTLY`, `VACUUM`, `ALTER SYSTEM`,
+  `CLUSTER`, `CREATE`/`DROP DATABASE`, `CREATE`/`DROP TABLESPACE`,
+  `REINDEX DATABASE`/`SYSTEM`/`SCHEMA`, `CREATE`/`DROP SUBSCRIPTION`, `DISCARD ALL`,
+  `ALTER DATABASE … SET TABLESPACE`,
+  `ALTER SUBSCRIPTION … REFRESH`/`SET`/`ADD`/`DROP PUBLICATION`,
+  `ALTER TABLE … DETACH PARTITION … CONCURRENTLY`,
+  `ALTER TABLE`/`INDEX`/`MATERIALIZED VIEW ALL IN TABLESPACE`, and
+  `REFRESH MATERIALIZED VIEW CONCURRENTLY`) run standalone outside
   the migration's transaction batch — they fail with SQLSTATE 25001 inside one. The
   history insert stays in the final batch, so a mid-file failure leaves earlier,
   already-committed batches applied with **no history row**; a re-run replays the file
-  from the top. Prefer idempotent forms (`… IF NOT EXISTS`) for such statements.
+  from the top. A failed batch's transaction is rolled back (bounded) before its
+  connection is reused. Prefer idempotent forms (`… IF NOT EXISTS`) for such statements.
   Intentional fix for supabase/cli#5139, adopted into TS in PR supabase/cli#5671
   (landed on develop as `b48fad60`).