diff --git a/.changeset/confirm-channel.md b/.changeset/confirm-channel.md
new file mode 100644
index 0000000..5c78a63
--- /dev/null
+++ b/.changeset/confirm-channel.md
@@ -0,0 +1,9 @@
+---
+"@effect-messaging/amqp": minor
+---
+
+Add `confirm` option to `AMQPChannelOptions`. When set to `true`, the channel
+is opened in publisher-confirm mode (`createConfirmChannel`) and every
+`publish` call resolves only after the broker has acknowledged the message,
+giving real backpressure and durability guarantees instead of relying on the
+local socket buffer. Defaults to `false` so existing callers are unaffected.
diff --git a/packages/amqp/src/AMQPChannel.ts b/packages/amqp/src/AMQPChannel.ts
index 52392b2..deea428 100644
--- a/packages/amqp/src/AMQPChannel.ts
+++ b/packages/amqp/src/AMQPChannel.ts
@@ -112,128 +112,128 @@ export interface AMQPChannelOptions {
   retryConnectionSchedule?: Schedule.Schedule<unknown, AMQPError.AMQPConnectionError>
   retryConsumptionSchedule?: Schedule.Schedule<unknown, AMQPError.AMQPChannelError>
   waitChannelTimeout?: Duration.DurationInput
+  /**
+   * When `true`, the channel is opened in publisher-confirm mode
+   * (`connection.createConfirmChannel`). Every `publish` call returns only
+   * after the broker has acknowledged the message, providing real
+   * backpressure and durability guarantees. Defaults to `false`.
+   *
+   * @since 0.7.0
+   */
+  confirm?: boolean
 }
 
 /**
  * @category constructors
  * @since 0.1.0
  */
-export const make = (options: AMQPChannelOptions = {}): Effect.Effect<
+export const make = (
+  options: AMQPChannelOptions = {}
+): Effect.Effect<
   AMQPChannel,
   AMQPError.AMQPChannelError | AMQPError.AMQPConnectionError,
   Scope.Scope | AMQPConnection.AMQPConnection
 > =>
-  Effect.gen(
-    function*() {
-      const internalChannel = yield* internal.InternalAMQPChannel
-      const provideInternal = Effect.provideService(internal.InternalAMQPChannel, internalChannel)
+  Effect.gen(function*() {
+    const internalChannel = yield* internal.InternalAMQPChannel
+    const provideInternal = Effect.provideService(internal.InternalAMQPChannel, internalChannel)
 
-      const channel = yield* Effect.acquireRelease(
-        Effect.gen(function*() {
-          yield* internal.initiateChannel
-          const connection = yield* AMQPConnection.AMQPConnection
+    const channel = yield* Effect.acquireRelease(
+      Effect.gen(function*() {
+        yield* internal.initiateChannel
+        const connection = yield* AMQPConnection.AMQPConnection
 
-          return {
-            [TypeId]: TypeId as TypeId,
-            connection,
-            consume: (queueName: string, options?: { readonly prefetch?: number }) =>
-              internal.consume(queueName, options).pipe(provideInternal),
-            ack: (...params: Parameters<Channel["ack"]>) =>
-              internal.wrapChannelMethod("ack", async (channel) => channel.ack(...params)).pipe(provideInternal),
-            ackAll: (...params: Parameters<Channel["ackAll"]>) =>
-              internal.wrapChannelMethod("ackAll", async (channel) => channel.ackAll(...params)).pipe(provideInternal),
-            nack: (...params: Parameters<Channel["nack"]>) =>
-              internal.wrapChannelMethod("nack", async (channel) => channel.nack(...params)).pipe(provideInternal),
-            nackAll: (...params: Parameters<Channel["nackAll"]>) =>
-              internal.wrapChannelMethod("nackAll", async (channel) => channel.nackAll(...params)).pipe(
-                provideInternal
-              ),
-            reject: (...params: Parameters<Channel["reject"]>) =>
-              internal.wrapChannelMethod("reject", async (channel) => channel.reject(...params)).pipe(provideInternal),
-            publish: (...params: Parameters<Channel["publish"]>) => internal.publish(...params).pipe(provideInternal),
-            sendToQueue: (...params: Parameters<Channel["sendToQueue"]>) =>
-              internal.wrapChannelMethod("sendToQueue", async (channel) => channel.sendToQueue(...params)).pipe(
-                provideInternal
-              ),
-            assertQueue: (...params: Parameters<Channel["assertQueue"]>) =>
-              internal.wrapChannelMethod("assertQueue", async (channel) => channel.assertQueue(...params)).pipe(
-                provideInternal
-              ),
-            checkQueue: (...params: Parameters<Channel["checkQueue"]>) =>
-              internal.wrapChannelMethod("checkQueue", async (channel) => channel.checkQueue(...params)).pipe(
-                provideInternal
-              ),
-            deleteQueue: (...params: Parameters<Channel["deleteQueue"]>) =>
-              internal.wrapChannelMethod("deleteQueue", async (channel) => channel.deleteQueue(...params)).pipe(
-                provideInternal
-              ),
-            purgeQueue: (...params: Parameters<Channel["purgeQueue"]>) =>
-              internal.wrapChannelMethod("purgeQueue", async (channel) => channel.purgeQueue(...params)).pipe(
-                provideInternal
-              ),
-            bindQueue: (...params: Parameters<Channel["bindQueue"]>) =>
-              internal.wrapChannelMethod("bindQueue", async (channel) => channel.bindQueue(...params)).pipe(
-                provideInternal
-              ),
-            unbindQueue: (...params: Parameters<Channel["unbindQueue"]>) =>
-              internal.wrapChannelMethod("unbindQueue", async (channel) => channel.unbindQueue(...params)).pipe(
-                provideInternal
-              ),
-            assertExchange: (...params: Parameters<Channel["assertExchange"]>) =>
-              internal.wrapChannelMethod(
-                "assertExchange",
-                async (channel) => channel.assertExchange(...params)
-              ).pipe(provideInternal),
-            checkExchange: (...params: Parameters<Channel["checkExchange"]>) =>
-              internal.wrapChannelMethod(
-                "checkExchange",
-                async (channel) => channel.checkExchange(...params)
-              ).pipe(provideInternal),
-            deleteExchange: (...params: Parameters<Channel["deleteExchange"]>) =>
-              internal.wrapChannelMethod(
-                "deleteExchange",
-                async (channel) => channel.deleteExchange(...params)
-              ).pipe(provideInternal),
-            bindExchange: (...params: Parameters<Channel["bindExchange"]>) =>
-              internal.wrapChannelMethod(
-                "bindExchange",
-                async (channel) => channel.bindExchange(...params)
-              ).pipe(provideInternal),
-            unbindExchange: (...params: Parameters<Channel["unbindExchange"]>) =>
-              internal.wrapChannelMethod(
-                "unbindExchange",
-                async (channel) => channel.unbindExchange(...params)
-              ).pipe(provideInternal),
-            cancel: (...params: Parameters<Channel["cancel"]>) =>
-              internal.wrapChannelMethod("cancel", async (channel) => channel.cancel(...params)).pipe(provideInternal),
-            get: (...params: Parameters<Channel["get"]>) =>
-              internal.wrapChannelMethod("get", async (channel) => channel.get(...params)).pipe(provideInternal),
-            prefetch: (...params: Parameters<Channel["prefetch"]>) =>
-              internal.wrapChannelMethod("prefetch", async (channel) => channel.prefetch(...params)).pipe(
-                provideInternal
-              ),
-            recover: (...params: Parameters<Channel["recover"]>) =>
-              internal.wrapChannelMethod("recover", async (channel) => channel.recover(...params)).pipe(
-                provideInternal
-              ),
-            close: (opts: internal.CloseChannelOptions = {}) => internal.closeChannel(opts).pipe(provideInternal)
-          }
-        }),
-        (channel) => channel.close()
-      )
-      yield* Effect.forkScoped(internal.keepChannelAlive)
-      yield* Effect.forkScoped(internal.monitorChannelErrors)
-      return channel
-    }
-  ).pipe(
-    Effect.provideServiceEffect(internal.InternalAMQPChannel, internal.InternalAMQPChannel.new(options))
-  )
+        return {
+          [TypeId]: TypeId as TypeId,
+          connection,
+          consume: (queueName: string, options?: { readonly prefetch?: number }) =>
+            internal.consume(queueName, options).pipe(provideInternal),
+          ack: (...params: Parameters<Channel["ack"]>) =>
+            internal.wrapChannelMethod("ack", async (channel) => channel.ack(...params)).pipe(provideInternal),
+          ackAll: (...params: Parameters<Channel["ackAll"]>) =>
+            internal.wrapChannelMethod("ackAll", async (channel) => channel.ackAll(...params)).pipe(provideInternal),
+          nack: (...params: Parameters<Channel["nack"]>) =>
+            internal.wrapChannelMethod("nack", async (channel) => channel.nack(...params)).pipe(provideInternal),
+          nackAll: (...params: Parameters<Channel["nackAll"]>) =>
+            internal.wrapChannelMethod("nackAll", async (channel) => channel.nackAll(...params)).pipe(provideInternal),
+          reject: (...params: Parameters<Channel["reject"]>) =>
+            internal.wrapChannelMethod("reject", async (channel) => channel.reject(...params)).pipe(provideInternal),
+          publish: (...params: Parameters<Channel["publish"]>) => internal.publish(...params).pipe(provideInternal),
+          sendToQueue: (...params: Parameters<Channel["sendToQueue"]>) =>
+            internal
+              .wrapChannelMethod("sendToQueue", async (channel) => channel.sendToQueue(...params))
+              .pipe(provideInternal),
+          assertQueue: (...params: Parameters<Channel["assertQueue"]>) =>
+            internal
+              .wrapChannelMethod("assertQueue", async (channel) => channel.assertQueue(...params))
+              .pipe(provideInternal),
+          checkQueue: (...params: Parameters<Channel["checkQueue"]>) =>
+            internal
+              .wrapChannelMethod("checkQueue", async (channel) => channel.checkQueue(...params))
+              .pipe(provideInternal),
+          deleteQueue: (...params: Parameters<Channel["deleteQueue"]>) =>
+            internal
+              .wrapChannelMethod("deleteQueue", async (channel) => channel.deleteQueue(...params))
+              .pipe(provideInternal),
+          purgeQueue: (...params: Parameters<Channel["purgeQueue"]>) =>
+            internal
+              .wrapChannelMethod("purgeQueue", async (channel) => channel.purgeQueue(...params))
+              .pipe(provideInternal),
+          bindQueue: (...params: Parameters<Channel["bindQueue"]>) =>
+            internal
+              .wrapChannelMethod("bindQueue", async (channel) => channel.bindQueue(...params))
+              .pipe(provideInternal),
+          unbindQueue: (...params: Parameters<Channel["unbindQueue"]>) =>
+            internal
+              .wrapChannelMethod("unbindQueue", async (channel) => channel.unbindQueue(...params))
+              .pipe(provideInternal),
+          assertExchange: (...params: Parameters<Channel["assertExchange"]>) =>
+            internal
+              .wrapChannelMethod("assertExchange", async (channel) => channel.assertExchange(...params))
+              .pipe(provideInternal),
+          checkExchange: (...params: Parameters<Channel["checkExchange"]>) =>
+            internal
+              .wrapChannelMethod("checkExchange", async (channel) => channel.checkExchange(...params))
+              .pipe(provideInternal),
+          deleteExchange: (...params: Parameters<Channel["deleteExchange"]>) =>
+            internal
+              .wrapChannelMethod("deleteExchange", async (channel) => channel.deleteExchange(...params))
+              .pipe(provideInternal),
+          bindExchange: (...params: Parameters<Channel["bindExchange"]>) =>
+            internal
+              .wrapChannelMethod("bindExchange", async (channel) => channel.bindExchange(...params))
+              .pipe(provideInternal),
+          unbindExchange: (...params: Parameters<Channel["unbindExchange"]>) =>
+            internal
+              .wrapChannelMethod("unbindExchange", async (channel) => channel.unbindExchange(...params))
+              .pipe(provideInternal),
+          cancel: (...params: Parameters<Channel["cancel"]>) =>
+            internal.wrapChannelMethod("cancel", async (channel) => channel.cancel(...params)).pipe(provideInternal),
+          get: (...params: Parameters<Channel["get"]>) =>
+            internal.wrapChannelMethod("get", async (channel) => channel.get(...params)).pipe(provideInternal),
+          prefetch: (...params: Parameters<Channel["prefetch"]>) =>
+            internal
+              .wrapChannelMethod("prefetch", async (channel) => channel.prefetch(...params))
+              .pipe(provideInternal),
+          recover: (...params: Parameters<Channel["recover"]>) =>
+            internal.wrapChannelMethod("recover", async (channel) => channel.recover(...params)).pipe(provideInternal),
+          close: (opts: internal.CloseChannelOptions = {}) => internal.closeChannel(opts).pipe(provideInternal)
+        }
+      }),
+      (channel) => channel.close()
+    )
+    yield* Effect.forkScoped(internal.keepChannelAlive)
+    yield* Effect.forkScoped(internal.monitorChannelErrors)
+    return channel
+  }).pipe(Effect.provideServiceEffect(internal.InternalAMQPChannel, internal.InternalAMQPChannel.new(options)))
 
 /**
  * @since 0.1.0
  * @category Layers
  */
-export const layer = (options: AMQPChannelOptions = {}): Layer.Layer<
+export const layer = (
+  options: AMQPChannelOptions = {}
+): Layer.Layer<
   AMQPChannel,
   AMQPError.AMQPChannelError | AMQPError.AMQPConnectionError,
   AMQPConnection.AMQPConnection
diff --git a/packages/amqp/src/internal/AMQPChannel.ts b/packages/amqp/src/internal/AMQPChannel.ts
index b9aea9f..d269169 100644
--- a/packages/amqp/src/internal/AMQPChannel.ts
+++ b/packages/amqp/src/internal/AMQPChannel.ts
@@ -1,6 +1,6 @@
 import * as Headers from "@effect/platform/Headers"
 import * as HttpTraceContext from "@effect/platform/HttpTraceContext"
-import type { Channel, ConsumeMessage } from "amqplib"
+import type { Channel, ConfirmChannel, ConsumeMessage } from "amqplib"
 import type { StreamEmit } from "effect"
 import * as Context from "effect/Context"
 import * as Duration from "effect/Duration"
@@ -30,15 +30,17 @@ const ATTR_MESSAGING_MESSAGE_CONVERSATION_ID = "messaging.message.conversation_i
 const ATTR_MESSAGING_AMQP_DESTINATION_ROUTING_KEY = "messaging.amqp.destination.routing_key" as const
 
 /** @internal */
-export class InternalAMQPChannel
-  extends Context.Tag("@effect-messaging/amqp/InternalAMQPChannel")<InternalAMQPChannel, {
-    channelRef: SubscriptionRef.SubscriptionRef<Option.Option<Channel>>
+export class InternalAMQPChannel extends Context.Tag("@effect-messaging/amqp/InternalAMQPChannel")<
+  InternalAMQPChannel,
+  {
+    channelRef: SubscriptionRef.SubscriptionRef<Option.Option<Channel | ConfirmChannel>>
     serverProperties: AMQPConnection.AMQPConnectionServerProperties
     retryConnectionSchedule: Schedule.Schedule<unknown, AMQPConnectionError>
     retryConsumptionSchedule: Schedule.Schedule<unknown, AMQPChannelError>
     waitChannelTimeout: Duration.DurationInput
-  }>()
-{
+    confirm: boolean
+  }
+>() {
   private static defaultRetryConnectionSchedule = Schedule.forever.pipe(Schedule.addDelay(() => 1000))
   private static defaultRetryConsumptionSchedule = Schedule.forever.pipe(Schedule.addDelay(() => 1000))
   private static defaultwaitChannelTimeout = Duration.seconds(5)
@@ -47,13 +49,10 @@ export class InternalAMQPChannel
     retryConnectionSchedule?: Schedule.Schedule<unknown, AMQPConnectionError>
     retryConsumptionSchedule?: Schedule.Schedule<unknown, AMQPChannelError>
     waitChannelTimeout?: Duration.DurationInput
-  }): Effect.Effect<
-    Context.Tag.Service<InternalAMQPChannel>,
-    AMQPConnectionError,
-    AMQPConnection.AMQPConnection
-  > =>
+    confirm?: boolean
+  }): Effect.Effect<Context.Tag.Service<InternalAMQPChannel>, AMQPConnectionError, AMQPConnection.AMQPConnection> =>
     Effect.gen(function*() {
-      const channelRef = yield* SubscriptionRef.make(Option.none<Channel>())
+      const channelRef = yield* SubscriptionRef.make(Option.none<Channel | ConfirmChannel>())
       const connection = yield* AMQPConnection.AMQPConnection
       const serverProperties = yield* connection.serverProperties
       return {
@@ -62,7 +61,8 @@ export class InternalAMQPChannel
         retryConnectionSchedule: options.retryConnectionSchedule ?? InternalAMQPChannel.defaultRetryConnectionSchedule,
         retryConsumptionSchedule: options.retryConsumptionSchedule ??
           InternalAMQPChannel.defaultRetryConsumptionSchedule,
-        waitChannelTimeout: options.waitChannelTimeout ?? InternalAMQPChannel.defaultwaitChannelTimeout
+        waitChannelTimeout: options.waitChannelTimeout ?? InternalAMQPChannel.defaultwaitChannelTimeout,
+        confirm: options.confirm ?? false
       }
     })
 }
@@ -86,17 +86,15 @@ const getOrWaitChannel = Effect.gen(function*() {
 
 /** @internal */
 export const initiateChannel = Effect.gen(function*() {
-  const { channelRef } = yield* InternalAMQPChannel
+  const { channelRef, confirm } = yield* InternalAMQPChannel
   yield* SubscriptionRef.updateEffect(channelRef, () =>
     Effect.gen(function*() {
       const connection = yield* AMQPConnection.AMQPConnection
-      const channel = yield* connection.createChannel
+      const channel = yield* confirm ? connection.createConfirmChannel : connection.createChannel
       return Option.some(channel)
     }))
   yield* Effect.logDebug(`AMQPChannel: channel created`)
-}).pipe(
-  Effect.withSpan("AMQPChannel.initiateChannel")
-)
+}).pipe(Effect.withSpan("AMQPChannel.initiateChannel"))
 
 /** @internal */
 export interface CloseChannelOptions {
@@ -118,9 +116,7 @@ export const closeChannel = ({ removeAllListeners = true }: CloseChannelOptions
         return Option.none()
       }))
     yield* Effect.logDebug("AMQPChannel: channel closed")
-  }).pipe(
-    Effect.withSpan("AMQPChannel.closeChannel")
-  )
+  }).pipe(Effect.withSpan("AMQPChannel.closeChannel"))
 
 /** @internal */
 export const keepChannelAlive = Effect.gen(function*() {
@@ -144,9 +140,11 @@ export const monitorChannelErrors = Effect.gen(function*() {
 })
 
 /** @internal */
-export const publish = (
-  ...[exchange, routingKey, content, options]: Parameters<Channel["publish"]>
-) =>
+const isConfirmChannel = (channel: Channel | ConfirmChannel): channel is ConfirmChannel =>
+  typeof (channel as ConfirmChannel).waitForConfirms === "function"
+
+/** @internal */
+export const publish = (...[exchange, routingKey, content, options]: Parameters<Channel["publish"]>) =>
   Effect.gen(function*() {
     const { serverProperties } = yield* InternalAMQPChannel
     return yield* Effect.useSpan(
@@ -169,15 +167,31 @@ export const publish = (
       (span) =>
         Effect.gen(function*() {
           const channel = yield* getOrWaitChannel
+          const finalOptions = {
+            ...options,
+            headers: Headers.merge(options?.headers ?? {}, HttpTraceContext.toHeaders(span))
+          }
+          if (isConfirmChannel(channel)) {
+            return yield* Effect.async<boolean, AMQPChannelError>((resume) => {
+              try {
+                const accepted = channel.publish(exchange, routingKey, content, finalOptions, (err) => {
+                  if (err) {
+                    resume(
+                      Effect.fail(
+                        new AMQPChannelError({ reason: `Broker nacked or channel closed before confirm`, cause: err })
+                      )
+                    )
+                  } else {
+                    resume(Effect.succeed(accepted))
+                  }
+                })
+              } catch (error) {
+                resume(Effect.fail(new AMQPChannelError({ reason: `Failed to publish on channel`, cause: error })))
+              }
+            })
+          }
           return yield* Effect.try({
-            try: () =>
-              channel.publish(exchange, routingKey, content, {
-                ...options,
-                headers: Headers.merge(
-                  options?.headers ?? {},
-                  HttpTraceContext.toHeaders(span)
-                )
-              }),
+            try: () => channel.publish(exchange, routingKey, content, finalOptions),
             catch: (error) => new AMQPChannelError({ reason: `Failed to publish on channel`, cause: error })
           })
         })
@@ -185,10 +199,7 @@ export const publish = (
   })
 
 /** @internal */
-export const wrapChannelMethod = <A>(
-  methodName: string,
-  callMethod: (channel: Channel) => PromiseLike<A>
-) =>
+export const wrapChannelMethod = <A>(methodName: string, callMethod: (channel: Channel) => PromiseLike<A>) =>
   Effect.gen(function*() {
     const channel = yield* getOrWaitChannel
     return yield* Effect.tryPromise({
@@ -198,42 +209,37 @@ export const wrapChannelMethod = <A>(
   }).pipe(Effect.withSpan(`AMQPChannel.${methodName}`))
 
 /** @internal */
-const initiateConsumption = Effect.fn("initiateConsumption")(
-  function*(
-    channel: Channel,
-    queueName: string,
-    emit: StreamEmit.EmitOpsPush<AMQPChannelError, ConsumeMessage>,
-    options?: { readonly prefetch?: number }
-  ) {
-    yield* Effect.annotateCurrentSpan({
-      [ATTR_MESSAGING_DESTINATION_SUBSCRIPTION_NAME]: queueName
-    })
-    yield* Effect.tryPromise({
-      try: () => channel.prefetch(options?.prefetch ?? DEFAULT_PREFETCH),
-      catch: (error) =>
-        new AMQPChannelError({ reason: `Failed to set prefetch on channel for queue ${queueName}`, cause: error })
-    })
-    const { consumerTag } = yield* Effect.tryPromise({
-      try: () =>
-        channel.consume(queueName, (message) => {
-          if (!message) return
-          emit.single(message)
-        }),
-      catch: (error) => new AMQPChannelError({ reason: `Failed to consume from queue ${queueName}`, cause: error })
-    })
-    yield* Effect.addFinalizer(() =>
-      Effect.tryPromise(() => channel.cancel(consumerTag)).pipe(
-        Effect.tap(Effect.logDebug(`AMQPChannel: consumer ${consumerTag} cancelled for queue ${queueName}`)),
-        Effect.ignore
-      )
+const initiateConsumption = Effect.fn("initiateConsumption")(function*(
+  channel: Channel,
+  queueName: string,
+  emit: StreamEmit.EmitOpsPush<AMQPChannelError, ConsumeMessage>,
+  options?: { readonly prefetch?: number }
+) {
+  yield* Effect.annotateCurrentSpan({ [ATTR_MESSAGING_DESTINATION_SUBSCRIPTION_NAME]: queueName })
+  yield* Effect.tryPromise({
+    try: () => channel.prefetch(options?.prefetch ?? DEFAULT_PREFETCH),
+    catch: (error) =>
+      new AMQPChannelError({ reason: `Failed to set prefetch on channel for queue ${queueName}`, cause: error })
+  })
+  const { consumerTag } = yield* Effect.tryPromise({
+    try: () =>
+      channel.consume(queueName, (message) => {
+        if (!message) return
+        emit.single(message)
+      }),
+    catch: (error) => new AMQPChannelError({ reason: `Failed to consume from queue ${queueName}`, cause: error })
+  })
+  yield* Effect.addFinalizer(() =>
+    Effect.tryPromise(() => channel.cancel(consumerTag)).pipe(
+      Effect.tap(Effect.logDebug(`AMQPChannel: consumer ${consumerTag} cancelled for queue ${queueName}`)),
+      Effect.ignore
     )
-    channel.on("close", () => {
-      emit.end()
-    })
-    yield* Effect.logDebug(`AMQPChannel: consuming from queue ${queueName} with consumer tag ${consumerTag}`)
-  },
-  Effect.withSpan("AMQPChannel.initiateConsumption")
-)
+  )
+  channel.on("close", () => {
+    emit.end()
+  })
+  yield* Effect.logDebug(`AMQPChannel: consuming from queue ${queueName} with consumer tag ${consumerTag}`)
+}, Effect.withSpan("AMQPChannel.initiateConsumption"))
 
 /** @internal */
 export const consume = (queueName: string, options?: { readonly prefetch?: number }) =>
@@ -244,9 +250,7 @@ export const consume = (queueName: string, options?: { readonly prefetch?: numbe
       Stream.flatMap(
         (channel) =>
           Stream.asyncPush<ConsumeMessage, AMQPChannelError>((emit) =>
-            initiateConsumption(channel, queueName, emit, options).pipe(
-              Effect.retry(retryConsumptionSchedule)
-            )
+            initiateConsumption(channel, queueName, emit, options).pipe(Effect.retry(retryConsumptionSchedule))
           ),
         { concurrency: "unbounded" }
       )
diff --git a/packages/amqp/test/AMQPChannel.test.ts b/packages/amqp/test/AMQPChannel.test.ts
index 2e56fcb..5563dd5 100644
--- a/packages/amqp/test/AMQPChannel.test.ts
+++ b/packages/amqp/test/AMQPChannel.test.ts
@@ -7,7 +7,8 @@ import {
   assertTestQueue,
   simulateChannelClose,
   simulateConnectionClose,
-  testChannel
+  testChannel,
+  testConfirmChannel
 } from "./dependencies.js"
 
 describe("AMQPChannel", () => {
@@ -54,9 +55,7 @@ describe("AMQPChannel", () => {
         yield* assertTestQueue
         const channel = yield* AMQPChannel.AMQPChannel
         const result = yield* channel.checkQueue("TEST_QUEUE")
-        expect(result).toMatchObject({
-          queue: "TEST_QUEUE"
-        })
+        expect(result).toMatchObject({ queue: "TEST_QUEUE" })
       }).pipe(Effect.provide(testChannel), TestServices.provideLive))
 
     it.effect("Should return an error when the queue does not exist", () =>
@@ -66,4 +65,40 @@ describe("AMQPChannel", () => {
         expect(exit).toStrictEqual(Exit.fail(expect.any(AMQPChannelError)))
       }).pipe(Effect.provide(testChannel), TestServices.provideLive))
   })
+
+  describe("confirm channel", () => {
+    const EXCHANGE = "TEST_CONFIRM_EXCHANGE"
+    const QUEUE = "TEST_CONFIRM_QUEUE"
+    const ROUTING_KEY = "TEST_CONFIRM_SUBJECT"
+
+    it.effect("publish resolves only after the broker confirms the message", () =>
+      Effect.gen(function*() {
+        const channel = yield* AMQPChannel.AMQPChannel
+        yield* channel.assertExchange(EXCHANGE, "direct", { durable: true })
+        yield* channel.assertQueue(QUEUE, { durable: true })
+        yield* channel.bindQueue(QUEUE, EXCHANGE, ROUTING_KEY)
+        yield* channel.purgeQueue(QUEUE)
+
+        yield* channel.publish(EXCHANGE, ROUTING_KEY, Buffer.from("confirmed-payload"), { persistent: true })
+
+        const message = yield* channel.get(QUEUE, { noAck: true })
+        expect(message).not.toBe(false)
+        if (message !== false) {
+          expect(message.content.toString()).toBe("confirmed-payload")
+        }
+
+        // Cleanup so re-runs don't accumulate state
+        yield* channel.deleteQueue(QUEUE)
+        yield* channel.deleteExchange(EXCHANGE)
+      }).pipe(Effect.provide(testConfirmChannel), TestServices.provideLive))
+
+    it.effect("publish fails when the target exchange does not exist", () =>
+      Effect.gen(function*() {
+        const channel = yield* AMQPChannel.AMQPChannel
+        const exit = yield* channel
+          .publish("NON_EXISTENT_CONFIRM_EXCHANGE", "whatever", Buffer.from("payload"))
+          .pipe(Effect.exit)
+        expect(exit).toStrictEqual(Exit.fail(expect.any(AMQPChannelError)))
+      }).pipe(Effect.provide(testConfirmChannel), TestServices.provideLive))
+  })
 })
diff --git a/packages/amqp/test/dependencies.ts b/packages/amqp/test/dependencies.ts
index 24649fc..cd73c5d 100644
--- a/packages/amqp/test/dependencies.ts
+++ b/packages/amqp/test/dependencies.ts
@@ -12,6 +12,8 @@ export const testConnection = AMQPConnection.layer({
 
 export const testChannel = AMQPChannel.layer().pipe(Layer.provideMerge(testConnection))
 
+export const testConfirmChannel = AMQPChannel.layer({ confirm: true }).pipe(Layer.provideMerge(testConnection))
+
 export const TEST_EXCHANGE = "TEST_EXCHANGE"
 export const TEST_QUEUE = "TEST_QUEUE"
 export const TEST_SUBJECT = "TEST_SUBJECT"