diff --git a/source/common/coroutine/BUILD b/source/common/coroutine/BUILD
index 3656e900a165c..3f1b00daadf95 100644
--- a/source/common/coroutine/BUILD
+++ b/source/common/coroutine/BUILD
@@ -83,14 +83,47 @@ envoy_cc_library(
     ],
 )
 
+envoy_cc_library(
+    name = "semaphore_lib",
+    srcs = ["semaphore.cc"],
+    hdrs = ["semaphore.h"],
+    deps = [
+        ":context_lib",
+        ":launch_lib",
+        ":leaf_awaitable_lib",
+        ":task_lib",
+        "//source/common/common:assert_lib",
+        "@abseil-cpp//absl/functional:any_invocable",
+        "@abseil-cpp//absl/status",
+        "@abseil-cpp//absl/status:statusor",
+    ],
+)
+
+envoy_cc_library(
+    name = "async_queue_lib",
+    hdrs = ["async_queue.h"],
+    deps = [
+        ":leaf_awaitable_lib",
+        ":semaphore_lib",
+        ":status_macros_lib",
+        ":task_lib",
+        "//source/common/common:assert_lib",
+        "@abseil-cpp//absl/functional:any_invocable",
+        "@abseil-cpp//absl/status",
+        "@abseil-cpp//absl/status:statusor",
+    ],
+)
+
 envoy_cc_library(
     name = "coroutine_lib",
     deps = [
+        ":async_queue_lib",
         ":context_lib",
         ":dispatcher_executor_lib",
         ":executor_lib",
         ":launch_lib",
         ":leaf_awaitable_lib",
+        ":semaphore_lib",
         ":status_macros_lib",
         ":task_lib",
     ],
diff --git a/source/common/coroutine/async_queue.h b/source/common/coroutine/async_queue.h
new file mode 100644
index 0000000000000..168670cdfecdd
--- /dev/null
+++ b/source/common/coroutine/async_queue.h
@@ -0,0 +1,429 @@
+#pragma once
+
+#include <algorithm>
+#include <cstdint>
+#include <deque>
+#include <limits>
+#include <list>
+#include <memory>
+#include <optional>
+#include <type_traits>
+#include <utility>
+
+#include "source/common/common/assert.h"
+#include "source/common/coroutine/launch.h"
+#include "source/common/coroutine/leaf_awaitable.h"
+#include "source/common/coroutine/semaphore.h"
+#include "source/common/coroutine/status_macros.h"
+#include "source/common/coroutine/task.h"
+
+#include "absl/cleanup/cleanup.h"
+#include "absl/functional/any_invocable.h"
+#include "absl/status/status.h"
+#include "absl/status/statusor.h"
+
+namespace Envoy {
+namespace Coroutine {
+
+using Capacity = Semaphore;
+using CapacityReservation = SemaphoreReservation;
+using CapacityPtr = SemaphorePtr;
+
+template <typename T> struct DefaultItemSize {
+  uint64_t operator()(const T&) const { return 1; }
+};
+
+/**
+ * AsyncQueue is an asynchronous, bounded or unbounded FIFO queue for Envoy coroutines.
+ *
+ * Concurrency & Ownership Model:
+ * - Designed for multi-producer, single-consumer or single-producer, single-consumer patterns,
+ *   where "producer" and "consumer" refer to independent coroutines pinned to the same
+ *   executor/dispatcher thread, calling `push()` and `pop()` respectively.
+ * - The queue is move-only and owned exclusively by the single consumer coroutine.
+ * - Producer coroutines access the queue via lightweight `PushAccessor` instances, guaranteeing
+ *   deterministic queue teardown when the consumer coroutine finishes.
+ *
+ * Resumption model on push and pop:
+ * - `push()` might synchronously resume an awaiting pop() without buffering or consuming
+ *   Semaphore capacity. The call stack can be as deep as the chain of queues that are connected
+ *   by pop-push operations.
+ * - `pop()` drain an item from the head of the queue if there is any. It doesn't synchronously
+ *   resume any pending `push()`. Pending `pending()` are awaken from a clean call stack in FIFO
+ *   order.
+ *
+ * This design is to avoid unbounded stack, and simplify reentrancy handling. By design, `push()`
+ * and `pop()` of the same queue should be from different coroutines otherwise it might block a
+ * queue indefinitely. This is helpful to push data down the chain as quickly as possible.
+ *
+ * Reentrancy & Memory Safety:
+ * - Synchronous reentrancy occurs at two points: direct data handoff in `tryHandoff()` and EOF
+ *   notification in `close()`.
+ * - In the single consumer coroutine model, call stack depth is strictly bounded at O(1).
+ * - In-line queue destruction by a resumed consumer coroutine is safe: pop_waiters_ is updated
+ *   prior to resumption, and no `this` members are touched after the callback returns.
+ * - Any in-flight producer coroutines suspended on Semaphore acquisition observe `*alive_ == false`
+ *   or `closed_ == true` upon waking up and terminate cleanly with FailedPreconditionError.
+ *
+ * Memory usage:
+ * - The queue's memory usage is bound to O(N + M), where N is the bound, and M is the number of
+ *   pending `push()`.
+ */
+template <typename T, typename SizeFunc = DefaultItemSize<T>> class AsyncQueue {
+public:
+  static_assert(
+      std::is_invocable_r_v<uint64_t, SizeFunc, const T&>,
+      "SizeFunc must be callable with 'const T&' and return a type convertible to uint64_t");
+
+private:
+  struct Core : public std::enable_shared_from_this<Core> {
+    struct QueuedItem {
+      explicit QueuedItem(T item_val) : item(std::move(item_val)) {}
+
+      std::optional<T> item;
+      std::optional<CapacityReservation> reservation;
+    };
+
+    struct PopWaiter {
+      absl::AnyInvocable<void(absl::StatusOr<std::optional<T>>)> cb;
+    };
+
+    Core(CapacityPtr capacity, SizeFunc size_func)
+        : capacity_(capacity != nullptr ? std::move(capacity)
+                                        : std::make_shared<Capacity>(std::nullopt)),
+          size_func_(std::move(size_func)) {}
+
+    ~Core() {
+      ASSERT(pop_waiters_.empty(),
+             "under single consumer assumption, popper cannot be waiting upon destruction");
+      if (in_handoff_ > 0) {
+        ASSERT(queue_.empty(),
+               "no queued items can exist when queue is destroyed during direct handoff");
+      }
+      *alive_ = false;
+      close();
+      queue_.clear();
+      current_size_ = 0;
+    }
+
+    void close() {
+      if (closed_) {
+        return;
+      }
+      closed_ = true;
+      std::list<std::shared_ptr<PopWaiter>> waiters;
+      waiters.swap(pop_waiters_);
+      for (auto& w : waiters) {
+        if (w->cb) {
+          auto cb = std::move(w->cb);
+          cb(std::optional<T>(std::nullopt));
+        }
+      }
+    }
+
+    bool closed() const { return closed_; }
+
+    bool empty() const { return queue_.empty(); }
+
+    uint64_t itemCount() const { return queue_.size(); }
+
+    uint64_t currentSize() const { return current_size_; }
+    std::optional<uint64_t> maxSize() const { return capacity_->maxPermits(); }
+    CapacityPtr capacity() const { return capacity_; }
+
+    template <typename U = T> bool tryHandoff(U&& item) {
+      while (!pop_waiters_.empty()) {
+        if (!pop_waiters_.front()->cb) {
+          pop_waiters_.pop_front();
+          continue;
+        }
+        auto waiter = std::move(pop_waiters_.front());
+        pop_waiters_.pop_front();
+
+        auto cb = std::move(waiter->cb);
+        if (cb) {
+          auto alive = alive_;
+          ++in_handoff_;
+          cb(std::optional<T>(std::move(item)));
+          if (*alive) {
+            --in_handoff_;
+          }
+        }
+        return true;
+      }
+      return false;
+    }
+
+    Task<absl::Status> push(T item) {
+      if (closed_) {
+        co_return absl::FailedPreconditionError("queue is closed");
+      }
+
+      // 1. Direct handoff to a waiting popper if available.
+      if (tryHandoff(item)) {
+        co_return absl::OkStatus();
+      }
+
+      // 2. Put item in queue_ as a pending entry and account for its size.
+      const uint64_t size = size_func_(item);
+      current_size_ += size;
+      auto queued_item = std::make_shared<QueuedItem>(std::move(item));
+      auto it = queue_.insert(queue_.end(), queued_item);
+
+      // 3. Acquire capacity from Capacity. Any pop() arriving while suspended will steal from
+      // queue_.
+      auto alive = alive_;
+      auto cap_res = co_await capacity_->acquire(size);
+      if (!*alive) {
+        queued_item->item.reset();
+        co_return absl::FailedPreconditionError("queue is closed");
+      }
+
+      // If a pop() stole the item during rendezvous, we are done!
+      if (!queued_item->item.has_value()) {
+        // cap_res is destroyed by RAII, returning permits back to capacity_.
+        co_return absl::OkStatus();
+      }
+
+      if (!cap_res.ok() || closed_) {
+        queue_.erase(it);
+        if (queued_item->item.has_value()) {
+          current_size_ -= size;
+          queued_item->item.reset();
+        }
+        if (!cap_res.ok()) {
+          co_return cap_res.status();
+        }
+        co_return absl::FailedPreconditionError("queue is closed");
+      }
+
+      // Attach the acquired reservation.
+      queued_item->reservation = std::move(cap_res.value());
+      co_return absl::OkStatus();
+    }
+
+    template <typename U = T> bool tryPush(U&& item) {
+      if (closed_) {
+        return false;
+      }
+
+      if (tryHandoff(std::forward<U>(item))) {
+        return true;
+      }
+
+      const uint64_t size = size_func_(item);
+      auto res_opt = capacity_->tryAcquire(size);
+      if (!res_opt.has_value()) {
+        return false;
+      }
+
+      current_size_ += size;
+      auto queued_item = std::make_shared<QueuedItem>(std::forward<U>(item));
+      queued_item->reservation = std::move(*res_opt);
+      queue_.push_back(std::move(queued_item));
+      return true;
+    }
+
+    std::optional<T> tryPop() {
+      if (queue_.empty()) {
+        return std::nullopt;
+      }
+      auto queued_item = std::move(queue_.front());
+      queue_.pop_front();
+
+      auto item = std::move(*queued_item->item);
+      queued_item->item.reset();
+      current_size_ -= size_func_(item);
+      queued_item->reservation.reset();
+
+      return std::optional<T>(std::move(item));
+    }
+
+    CapacityPtr capacity_;
+    SizeFunc size_func_;
+    uint64_t current_size_{0};
+    uint64_t in_handoff_{0};
+    std::list<std::shared_ptr<QueuedItem>> queue_;
+    std::list<std::shared_ptr<PopWaiter>> pop_waiters_;
+    bool closed_{false};
+    std::shared_ptr<bool> alive_{std::make_shared<bool>(true)};
+  };
+
+public:
+  /**
+   * PushAccessor provides non-owning access to an AsyncQueue.
+   * Producers can push to the queue without holding ownership, allowing the consumer (the popper)
+   * to be the sole owner of the queue lifetime.
+   */
+  class PushAccessor {
+  public:
+    PushAccessor() = default;
+    explicit PushAccessor(std::weak_ptr<Core> core) : core_(std::move(core)) {}
+
+    Task<absl::Status> push(T item) {
+      auto core = core_.lock();
+      if (!core) {
+        co_return absl::FailedPreconditionError("queue is closed");
+      }
+      auto push_task = core->push(std::move(item));
+      core.reset();
+      co_return co_await std::move(push_task);
+    }
+
+    template <typename U = T> bool tryPush(U&& item) {
+      auto core = core_.lock();
+      if (!core) {
+        return false;
+      }
+      Core* core_ptr = core.get();
+      core.reset();
+      return core_ptr->tryPush(std::forward<U>(item));
+    }
+
+    void close() {
+      auto core = core_.lock();
+      if (core) {
+        core->close();
+      }
+    }
+
+    bool closed() const {
+      auto core = core_.lock();
+      return !core || core->closed();
+    }
+
+    bool empty() const {
+      auto core = core_.lock();
+      return !core || core->empty();
+    }
+
+    uint64_t currentSize() const {
+      auto core = core_.lock();
+      return core ? core->currentSize() : 0;
+    }
+
+    uint64_t itemCount() const {
+      auto core = core_.lock();
+      return core ? core->itemCount() : 0;
+    }
+
+    CapacityPtr capacity() const {
+      auto core = core_.lock();
+      return core ? core->capacity() : nullptr;
+    }
+
+  private:
+    std::weak_ptr<Core> core_;
+  };
+
+  explicit AsyncQueue(CapacityPtr capacity = nullptr, SizeFunc size_func = SizeFunc())
+      : core_(std::make_shared<Core>(std::move(capacity), std::move(size_func))) {}
+
+  explicit AsyncQueue(uint64_t max_size, SizeFunc size_func = SizeFunc())
+      : AsyncQueue(std::make_shared<Capacity>(max_size), std::move(size_func)) {}
+
+  ~AsyncQueue() {
+    if (core_) {
+      core_->close();
+    }
+  }
+
+  // Move-only semantics: steals core_, leaving other.core_ == nullptr.
+  AsyncQueue(AsyncQueue&& other) noexcept : core_(std::move(other.core_)) {}
+  AsyncQueue& operator=(AsyncQueue&& other) noexcept {
+    if (this != &other) {
+      if (core_) {
+        core_->close();
+      }
+      core_ = std::move(other.core_);
+    }
+    return *this;
+  }
+  AsyncQueue(const AsyncQueue&) = delete;
+  AsyncQueue& operator=(const AsyncQueue&) = delete;
+
+  PushAccessor pushAccessor() const { return PushAccessor(core_); }
+
+  uint64_t itemCount() const { return core_ ? core_->itemCount() : 0; }
+  uint64_t currentSize() const { return core_ ? core_->currentSize() : 0; }
+  std::optional<uint64_t> maxSize() const { return core_ ? core_->maxSize() : std::nullopt; }
+  bool closed() const { return !core_ || core_->closed(); }
+  bool empty() const { return !core_ || core_->empty(); }
+  CapacityPtr capacity() const { return core_ ? core_->capacity() : nullptr; }
+
+  Task<absl::Status> push(T item) {
+    if (!core_) {
+      co_return absl::FailedPreconditionError("queue is closed");
+    }
+    co_return co_await core_->push(std::move(item));
+  }
+
+  template <typename U = T> bool tryPush(U&& item) {
+    if (!core_) {
+      return false;
+    }
+    return core_->tryPush(std::forward<U>(item));
+  }
+
+  Task<absl::StatusOr<std::optional<T>>> pop() {
+    if (!core_) {
+      co_return std::optional<T>(std::nullopt);
+    }
+    co_return co_await PopAwaitable(core_);
+  }
+
+  std::optional<T> tryPop() { return core_ ? core_->tryPop() : std::nullopt; }
+
+  void close() {
+    if (core_) {
+      core_->close();
+    }
+  }
+
+private:
+  class PopAwaitable : public LeafAwaitable<absl::StatusOr<std::optional<T>>> {
+  public:
+    explicit PopAwaitable(std::shared_ptr<Core> core) : core_(std::move(core)) {
+      ASSERT(core_ != nullptr);
+    }
+
+  protected:
+    std::optional<absl::StatusOr<std::optional<T>>> tryImmediate() override {
+      std::optional<T> item = core_->tryPop();
+      if (item.has_value()) {
+        return item;
+      }
+      if (core_->closed()) {
+        // Return immediate EOF without suspending. Note that returning
+        // std::optional<T>(std::nullopt) wraps an empty inner optional (EOF) inside a present outer
+        // optional, whereas returning std::nullopt would produce an empty outer optional that
+        // instructs LeafAwaitable to suspend.
+        return std::optional<T>(std::nullopt);
+      }
+      // Queue is open but empty: return an empty outer optional to suspend and wait for a producer.
+      return std::nullopt;
+    }
+
+    void onStart() override {
+      waiter_ = std::make_shared<typename Core::PopWaiter>();
+      waiter_->cb = [this](absl::StatusOr<std::optional<T>> res) {
+        this->complete(std::move(res));
+      };
+      core_->pop_waiters_.push_back(waiter_);
+    }
+
+    void onCancel() override {
+      ASSERT(waiter_ != nullptr);
+      waiter_->cb = nullptr;
+    }
+
+  private:
+    std::shared_ptr<Core> core_;
+    std::shared_ptr<typename Core::PopWaiter> waiter_;
+  };
+
+  std::shared_ptr<Core> core_;
+};
+
+} // namespace Coroutine
+} // namespace Envoy
diff --git a/source/common/coroutine/context.h b/source/common/coroutine/context.h
index a3fb512fadea7..cb0956e5daa83 100644
--- a/source/common/coroutine/context.h
+++ b/source/common/coroutine/context.h
@@ -2,6 +2,7 @@
 
 #include <memory>
 
+#include "source/common/common/assert.h"
 #include "source/common/coroutine/executor.h"
 
 #include "absl/functional/any_invocable.h"
@@ -47,10 +48,12 @@ class CancellationState {
   }
 
   // Registers a cancellation callback. This is called by a leaf awaitable while it
-  // is suspended. If the scope is already cancelled, the callback fires synchronously.
+  // is suspended. This must never execute the callback inline on the stack; the callback
+  // is only invoked when cancel() is explicitly called.
   void setCancelCallback(absl::AnyInvocable<void()> cb) {
     if (cancelled_) {
-      cb();
+      IS_ENVOY_BUG(
+          "setCancelCallback called on an already-cancelled CancellationState. Ignoring callback.");
       return;
     }
     on_cancel_ = std::move(cb);
@@ -84,6 +87,7 @@ class CoroutineContext {
       : executor_(std::move(executor)), cancel_(std::move(cancel)) {}
 
   Executor& executor() const { return *executor_; }
+  const std::shared_ptr<Executor>& executorShared() const { return executor_; }
   const CancellationStatePtr& cancellation() const { return cancel_; }
 
 private:
diff --git a/source/common/coroutine/leaf_awaitable.h b/source/common/coroutine/leaf_awaitable.h
index 6f08322d6a183..79b230563ea7d 100644
--- a/source/common/coroutine/leaf_awaitable.h
+++ b/source/common/coroutine/leaf_awaitable.h
@@ -44,11 +44,31 @@ template <typename T> class LeafAwaitable {
                 "cancellation can be delivered as an aborted value");
 
 public:
-  // Fail-fast: if the scope is already cancelled, don't even start.
-  bool await_ready() { return context_->cancellation()->cancelled(); }
+  LeafAwaitable() = default;
+  LeafAwaitable(const LeafAwaitable&) = delete;
+  LeafAwaitable& operator=(const LeafAwaitable&) = delete;
+  LeafAwaitable(LeafAwaitable&&) = delete;
+  LeafAwaitable& operator=(LeafAwaitable&&) = delete;
+  virtual ~LeafAwaitable() = default;
+
+  // Fail-fast: if the scope is already cancelled, or if an immediate non-blocking attempt
+  // produces a result, don't even start/suspend.
+  bool await_ready() {
+    if (context_->cancellation()->cancelled()) {
+      return true;
+    }
+    result_ = tryImmediate();
+    return result_.has_value();
+  }
 
-  void await_suspend(std::coroutine_handle<> continuation) {
+  bool await_suspend(std::coroutine_handle<> continuation) {
     continuation_ = continuation;
+    if (context_->cancellation()->cancelled()) {
+      // If already cancelled, do not suspend and do not call onStart().
+      // Returning false immediately resumes the coroutine on the current stack,
+      // which will invoke await_resume() and return abortedValue().
+      return false;
+    }
     // Register the cancel action while this is the pending leaf.
     context_->cancellation()->setCancelCallback([this] {
       cancelling_ = true;
@@ -56,13 +76,19 @@ template <typename T> class LeafAwaitable {
       finish(abortedValue());
     });
     onStart(); // derived kicks off the async op; must eventually call complete().
+    return true;
   }
 
   // [[nodiscard]]: the result carries success/failure/cancellation, so a
   // `co_await leaf;` that drops it is almost always a bug.
   [[nodiscard]] T await_resume() {
-    // On the fail-fast path (await_ready true) await_suspend never ran, so
-    // result_ is empty and we resume with the aborted value.
+    // If result_ contains a value (e.g. from tryImmediate()), return that result rather than
+    // abortedValue() even if cancellation occurred. This ensures any non-blocking side effects
+    // (such as popped items or acquired capacity) remain visible to the caller so that they can
+    // be processed, stored, or undone as needed. The caller coroutine will either return all the
+    // way up or encounter a subsequent awaitable that is cancelled without executing further side
+    // effects.
+    // If result_ is empty (the fail-fast pre-cancelled path), return abortedValue().
     return result_ ? std::move(*result_) : abortedValue();
   }
 
@@ -75,6 +101,13 @@ template <typename T> class LeafAwaitable {
   virtual void onStart() PURE;  // launch the op; arrange to call complete(value).
   virtual void onCancel() PURE; // cancel the pending op (honor its cancel contract).
 
+  // Optional non-blocking / immediate attempt: derived classes can perform an immediate
+  // operation (such as a non-blocking check, tryAcquire, or tryPop). If the operation completes
+  // immediately, returning a value avoids suspension overhead and resumes the coroutine directly.
+  // Returning std::nullopt indicates that the operation must suspend and arrange asynchronous
+  // completion via onStart().
+  virtual std::optional<T> tryImmediate() { return std::nullopt; }
+
   // Called by derived when the real event fires.
   void complete(T value) {
     if (!cancelling_) {
@@ -86,8 +119,6 @@ template <typename T> class LeafAwaitable {
 
   CoroutineContext& context() { return *context_; }
 
-  virtual ~LeafAwaitable() = default;
-
 private:
   static T abortedValue() { return absl::CancelledError("coroutine cancelled"); }
 
diff --git a/source/common/coroutine/semaphore.cc b/source/common/coroutine/semaphore.cc
new file mode 100644
index 0000000000000..dab40a1d11f62
--- /dev/null
+++ b/source/common/coroutine/semaphore.cc
@@ -0,0 +1,201 @@
+#include "source/common/coroutine/semaphore.h"
+
+namespace Envoy {
+namespace Coroutine {
+
+SemaphoreReservation::SemaphoreReservation(std::weak_ptr<Semaphore> sem, uint64_t permits)
+    : sem_(std::move(sem)), permits_(permits) {}
+
+SemaphoreReservation::~SemaphoreReservation() { release(); }
+
+SemaphoreReservation::SemaphoreReservation(SemaphoreReservation&& other) noexcept
+    : sem_(std::move(other.sem_)), permits_(std::exchange(other.permits_, 0)) {}
+
+SemaphoreReservation& SemaphoreReservation::operator=(SemaphoreReservation&& other) noexcept {
+  if (this != &other) {
+    if (permits_ > 0) {
+      IS_ENVOY_BUG("SemaphoreReservation should not overwrite an active reservation");
+      release();
+    }
+    sem_ = std::move(other.sem_);
+    permits_ = std::exchange(other.permits_, 0);
+  }
+  return *this;
+}
+
+void SemaphoreReservation::release() {
+  const uint64_t permits = std::exchange(permits_, 0);
+  std::shared_ptr<Semaphore> sem = sem_.lock();
+  sem_.reset();
+  if (permits > 0 && sem != nullptr) {
+    sem->release(permits);
+  }
+}
+
+Semaphore::SemaphoreAwaitable::SemaphoreAwaitable(Semaphore& sem, uint64_t permits)
+    : sem_(sem), permits_(permits) {}
+
+std::optional<absl::StatusOr<SemaphoreReservation>> Semaphore::SemaphoreAwaitable::tryImmediate() {
+  std::optional<SemaphoreReservation> res = sem_.tryAcquire(permits_);
+  if (res.has_value()) {
+    return res;
+  }
+  return std::nullopt;
+}
+
+void Semaphore::SemaphoreAwaitable::onStart() {
+  waiter_ = std::make_shared<Semaphore::Waiter>();
+  waiter_->permits = permits_;
+  waiter_->executor = this->context().executorShared();
+  waiter_->cb = [this](absl::StatusOr<SemaphoreReservation> res) {
+    this->complete(std::move(res));
+  };
+  sem_.waiters_.push_back(waiter_);
+}
+
+void Semaphore::SemaphoreAwaitable::onCancel() {
+  ASSERT(waiter_ != nullptr);
+  waiter_->cb = nullptr;
+  sem_.scheduleProcessWaiters();
+}
+
+Semaphore::Semaphore(std::optional<uint64_t> max_permits) : max_permits_(max_permits) {
+  if (max_permits_.has_value()) {
+    ASSERT(*max_permits_ > 0, "max_permits must be positive if specified");
+  }
+}
+
+Semaphore::~Semaphore() {
+  *alive_ = false;
+  if (process_handle_.has_value()) {
+    // Note: cancel() is a no-op here because runScheduledProcessWaiters does not await any
+    // leaf awaitables and executes synchronously once scheduled. Setting *alive_ = false
+    // above guarantees the task will observe destruction and return immediately.
+    process_handle_->cancel();
+    // Resetting process_handle_ drops the DetachedHandle. The underlying RootTask frame
+    // becomes self-owned and will automatically clean itself up once it runs on the executor.
+    process_handle_.reset();
+  }
+  std::list<std::shared_ptr<Waiter>> waiters;
+  waiters.swap(waiters_);
+  for (std::shared_ptr<Waiter>& w : waiters) {
+    if (w->cb) {
+      absl::AnyInvocable<void(absl::StatusOr<SemaphoreReservation>)> cb = std::move(w->cb);
+      cb(absl::FailedPreconditionError("Semaphore is destroyed"));
+    }
+  }
+}
+
+bool Semaphore::hasPermits(uint64_t additional_permits) const {
+  ASSERT(max_permits_.has_value());
+  if (current_permits_ > *max_permits_) {
+    return false;
+  }
+  if (additional_permits > std::numeric_limits<uint64_t>::max() - current_permits_) {
+    return false;
+  }
+  return (current_permits_ + additional_permits) <= *max_permits_;
+}
+
+bool Semaphore::canAcquire(uint64_t permits) const {
+  if (!max_permits_.has_value()) {
+    return true;
+  }
+  if (current_permits_ == 0 && permits > *max_permits_) {
+    return true;
+  }
+  return hasPermits(permits);
+}
+
+void Semaphore::popCancelledWaiters() {
+  while (!waiters_.empty() && !waiters_.front()->cb) {
+    waiters_.pop_front();
+  }
+}
+
+std::optional<SemaphoreReservation> Semaphore::tryAcquire(uint64_t permits) {
+  popCancelledWaiters();
+  if (waiters_.empty() && canAcquire(permits)) {
+    current_permits_ += permits;
+    return SemaphoreReservation(shared_from_this(), permits);
+  }
+  return std::nullopt;
+}
+
+Semaphore::SemaphoreAwaitable Semaphore::acquire(uint64_t permits) {
+  return SemaphoreAwaitable(*this, permits);
+}
+
+void Semaphore::release(uint64_t permits) {
+  ASSERT(permits > 0);
+  ASSERT(current_permits_ >= permits);
+  current_permits_ -= permits;
+  scheduleProcessWaiters();
+}
+
+Task<absl::Status> Semaphore::runScheduledProcessWaiters(std::weak_ptr<Semaphore> weak_self,
+                                                         std::shared_ptr<bool> alive) {
+  if (!*alive) {
+    co_return absl::OkStatus();
+  }
+  std::shared_ptr<Semaphore> sem = weak_self.lock();
+  ASSERT(sem != nullptr);
+  sem->process_handle_.reset();
+  sem->processWaiters();
+  co_return absl::OkStatus();
+}
+
+void Semaphore::scheduleProcessWaiters() {
+  if (process_handle_.has_value() || waiters_.empty()) {
+    return;
+  }
+
+  popCancelledWaiters();
+  if (waiters_.empty()) {
+    return;
+  }
+
+  if (!canAcquire(waiters_.front()->permits)) {
+    return;
+  }
+
+  std::shared_ptr<Executor> exec;
+  for (const std::shared_ptr<Waiter>& w : waiters_) {
+    if (w->cb && w->executor != nullptr) {
+      exec = w->executor;
+      break;
+    }
+  }
+  ASSERT(exec != nullptr);
+
+  process_handle_ = launch(
+      runScheduledProcessWaiters(weak_from_this(), alive_), std::move(exec), [](absl::Status) {},
+      StartMode::Scheduled);
+}
+
+void Semaphore::processWaiters() {
+  std::shared_ptr<bool> alive = alive_;
+  while (*alive && !waiters_.empty()) {
+    popCancelledWaiters();
+    if (waiters_.empty()) {
+      break;
+    }
+
+    std::shared_ptr<Waiter> head = waiters_.front();
+    if (!canAcquire(head->permits)) {
+      break;
+    }
+
+    waiters_.pop_front();
+    current_permits_ += head->permits;
+    SemaphoreReservation reservation(shared_from_this(), head->permits);
+
+    absl::AnyInvocable<void(absl::StatusOr<SemaphoreReservation>)> cb = std::move(head->cb);
+    if (cb) {
+      cb(std::move(reservation));
+    }
+  }
+}
+
+} // namespace Coroutine
+} // namespace Envoy
diff --git a/source/common/coroutine/semaphore.h b/source/common/coroutine/semaphore.h
new file mode 100644
index 0000000000000..f22616952e218
--- /dev/null
+++ b/source/common/coroutine/semaphore.h
@@ -0,0 +1,135 @@
+#pragma once
+
+#include <cstdint>
+#include <limits>
+#include <list>
+#include <memory>
+#include <optional>
+#include <utility>
+
+#include "source/common/common/assert.h"
+#include "source/common/coroutine/launch.h"
+#include "source/common/coroutine/leaf_awaitable.h"
+#include "source/common/coroutine/task.h"
+
+#include "absl/functional/any_invocable.h"
+#include "absl/status/status.h"
+#include "absl/status/statusor.h"
+
+namespace Envoy {
+namespace Coroutine {
+
+class Semaphore;
+
+/**
+ * SemaphoreReservation is an RAII guard holding a reservation against a Semaphore.
+ * When destroyed or explicitly released, it automatically decrements the reserved permits
+ * in Semaphore and triggers waiter processing.
+ */
+class SemaphoreReservation {
+public:
+  SemaphoreReservation() = default;
+  SemaphoreReservation(std::weak_ptr<Semaphore> sem, uint64_t permits);
+  ~SemaphoreReservation();
+
+  SemaphoreReservation(const SemaphoreReservation&) = delete;
+  SemaphoreReservation& operator=(const SemaphoreReservation&) = delete;
+
+  SemaphoreReservation(SemaphoreReservation&& other) noexcept;
+  SemaphoreReservation& operator=(SemaphoreReservation&& other) noexcept;
+
+  uint64_t permits() const { return permits_; }
+  bool hasPermits() const { return !sem_.expired() && permits_ > 0; }
+  void release();
+
+private:
+  std::weak_ptr<Semaphore> sem_;
+  uint64_t permits_{0};
+};
+
+/**
+ * Semaphore is an asynchronous FIFO weighted semaphore for Envoy coroutines.
+ *
+ * It preserves strict FIFO arrival ordering for permit acquisition. Waiters that cannot be
+ * immediately satisfied are queued. When permits are released, processing resumes pending
+ * waiters in strict FIFO order in the next event loop iteration.
+ *
+ * Anti-starvation for oversized requests:
+ * When the semaphore is completely idle (`currentPermits() == 0`), a single oversized
+ * acquisition (`permits > maxPermits()`) is allowed to acquire and proceed. This ensures that
+ * items larger than the nominal capacity limit are not permanently starved or deadlocked.
+ * While an oversized reservation is active, subsequent acquisitions are blocked until it is
+ * released.
+ */
+class Semaphore : public std::enable_shared_from_this<Semaphore> {
+public:
+  struct Waiter {
+    uint64_t permits{0};
+    absl::AnyInvocable<void(absl::StatusOr<SemaphoreReservation>)> cb;
+    std::shared_ptr<Executor> executor;
+  };
+
+  class SemaphoreAwaitable : public LeafAwaitable<absl::StatusOr<SemaphoreReservation>> {
+  public:
+    SemaphoreAwaitable(Semaphore& sem, uint64_t permits);
+
+  protected:
+    std::optional<absl::StatusOr<SemaphoreReservation>> tryImmediate() override;
+    void onStart() override;
+    void onCancel() override;
+
+  private:
+    Semaphore& sem_;
+    uint64_t permits_;
+    std::shared_ptr<Waiter> waiter_;
+  };
+
+  explicit Semaphore(std::optional<uint64_t> max_permits = std::nullopt);
+  ~Semaphore();
+
+  Semaphore(const Semaphore&) = delete;
+  Semaphore& operator=(const Semaphore&) = delete;
+  Semaphore(Semaphore&&) = delete;
+  Semaphore& operator=(Semaphore&&) = delete;
+
+  std::optional<uint64_t> maxPermits() const { return max_permits_; }
+  uint64_t currentPermits() const { return current_permits_; }
+
+  /**
+   * Attempts to synchronously acquire `permits` without suspending.
+   * If the semaphore has sufficient capacity (or is completely idle for a single oversized
+   * acquisition), returns a SemaphoreReservation. Otherwise returns std::nullopt.
+   */
+  std::optional<SemaphoreReservation> tryAcquire(uint64_t permits = 1);
+
+  /**
+   * Asynchronously acquires `permits` following strict FIFO ordering.
+   * If capacity is not immediately available, suspends the coroutine until sufficient permits
+   * are released. Supports a single oversized acquisition when the semaphore is fully drained.
+   */
+  SemaphoreAwaitable acquire(uint64_t permits = 1);
+
+private:
+  friend class SemaphoreReservation;
+  friend class SemaphoreAwaitable;
+
+  void release(uint64_t permits);
+  bool hasPermits(uint64_t additional_permits) const;
+  bool canAcquire(uint64_t permits) const;
+  void popCancelledWaiters();
+  void processWaiters();
+  void scheduleProcessWaiters();
+  static Task<absl::Status> runScheduledProcessWaiters(std::weak_ptr<Semaphore> weak_self,
+                                                       std::shared_ptr<bool> alive);
+
+  const std::optional<uint64_t> max_permits_;
+  uint64_t current_permits_{0};
+  std::list<std::shared_ptr<Waiter>> waiters_;
+  std::optional<DetachedHandle> process_handle_;
+  std::shared_ptr<bool> alive_{std::make_shared<bool>(true)};
+};
+
+using SemaphorePtr = std::shared_ptr<Semaphore>;
+
+} // namespace Coroutine
+} // namespace Envoy
diff --git a/source/common/coroutine/task.h b/source/common/coroutine/task.h
index b28c01464de7e..47963a2ac226d 100644
--- a/source/common/coroutine/task.h
+++ b/source/common/coroutine/task.h
@@ -34,6 +34,57 @@ namespace Coroutine {
  * only live until C++26 is widely adopted, which provides STL support of
  * coroutine in `std::execution`. It is intentionally kept small and lean so
  * that it is easier to maintain and migrate off eventually.
+ *
+ * ===========================================================================
+ * Core Architectural & Lifecycle Design Invariants:
+ * ===========================================================================
+ * Coroutines in Envoy are NEVER destroyed while suspended.
+ *
+ * External handles (such as `DetachedHandle`) never invoke `destroy()` on suspended
+ * frames. Instead, cancellation in Envoy coroutines operates strictly via
+ * cooperative cancellation with structured synchronous stack unwinding:
+ *
+ * 1. Cooperative Cancellation:
+ *    When cancellation is requested (e.g. via `DetachedHandle::cancel()` or
+ *    `CancellationState::cancel()`), cancellation invokes the pending leaf awaitable's
+ *    `onCancel()` hook to cancel/disarm the underlying asynchronous operation (e.g.
+ *    disarming a timer), completes the awaitable with an aborted status
+ *    (`absl::CancelledError`), and resumes the coroutine synchronously via
+ *    `continuation_.resume()`.
+ *
+ * 2. Structured Synchronous Stack Unwinding:
+ *    Upon resumption, the coroutine receives the aborted status and unwinds through
+ *    standard structured C++ control flow (such as `CO_RETURN_IF_ERROR` or
+ *    `ASSIGN_OR_CO_RETURN`). As the call stack unwinds, all local variables and
+ *    RAII objects (including awaitables, cleanup guards, and locks) have
+ *    their destructors executed deterministically in reverse order of declaration,
+ *    propagating up to `co_return` and reaching `final_suspend()`.
+ *
+ * 3. Awaitable Lifecycle Implications:
+ *    - Awaitable destructors NEVER run mid-suspension: Because coroutine frames are
+ *      never destroyed while suspended, an awaitable object's destructor only runs
+ *      after the coroutine has resumed (either via normal event completion or via
+ *      cancellation resumption) and control flow leaves its lexical scope.
+ *    - Soundness of Defaulted Destructors: Because awaitables never destruct while
+ *      suspended waiting for an event, defaulted virtual destructors on awaitables
+ *      (`virtual ~LeafAwaitable() = default;`) are sound. Derived awaitables do not
+ *      need complex destructor logic to unhook themselves from event sources or
+ *      cancel in-flight operations during destruction; all cancellation and resource
+ *      disarming are handled cleanly through `onCancel()`.
+ *
+ * 4. Handle Ownership & Frame Management:
+ *    - DetachedHandle: `DetachedHandle` (returned by `launch()`) does not own the
+ *      coroutine frame; it only holds a shared reference to the `CancellationState`.
+ *      Dropping a `DetachedHandle` at any time is completely safe: it does not cancel
+ *      the coroutine nor destroy the suspended frame. The coroutine continues
+ *      executing to completion.
+ *    - RootTask: The root coroutine frame is self-owning and destroys itself only
+ *      when it completes at `final_suspend` (via `std::suspend_never` in
+ *      `RootTask::promise_type`), ensuring `on_done` always runs with the final status.
+ *    - Task<T>: For `Task<T>`, the only times `handle_.destroy()` is invoked are when
+ *      an unstarted `Task` is discarded before launch/await (at `initial_suspend()`),
+ *      or when an awaiting caller destroys a completed frame that has already reached
+ *      `final_suspend()`.
  */
 
 // ---------------------------------------------------------------------------
diff --git a/test/common/coroutine/BUILD b/test/common/coroutine/BUILD
index b278661af120b..11b210a4b8d6f 100644
--- a/test/common/coroutine/BUILD
+++ b/test/common/coroutine/BUILD
@@ -57,6 +57,39 @@ envoy_cc_test(
     ],
 )
 
+envoy_cc_test(
+    name = "semaphore_test",
+    srcs = ["semaphore_test.cc"],
+    rbe_pool = "6gig",
+    deps = [
+        ":manual_executor_lib",
+        "//source/common/coroutine:coroutine_lib",
+        "//source/common/coroutine:semaphore_lib",
+        "//test/test_common:status_utility_lib",
+        "//test/test_common:utility_lib",
+        "@abseil-cpp//absl/status",
+        "@abseil-cpp//absl/status:statusor",
+    ],
+)
+
+envoy_cc_test(
+    name = "async_queue_test",
+    srcs = ["async_queue_test.cc"],
+    rbe_pool = "6gig",
+    deps = [
+        ":manual_executor_lib",
+        "//source/common/coroutine:async_queue_lib",
+        "//source/common/coroutine:coroutine_lib",
+        "//source/common/coroutine:dispatcher_executor_lib",
+        "//source/common/coroutine:leaf_awaitable_lib",
+        "//test/test_common:simulated_time_system_lib",
+        "//test/test_common:status_utility_lib",
+        "//test/test_common:utility_lib",
+        "@abseil-cpp//absl/status",
+        "@abseil-cpp//absl/status:statusor",
+    ],
+)
+
 # Echo-server micro-benchmark: a callback state machine vs. a coroutine body, both
 # over a real dispatcher + IoHandle, driven by a shared blocking two-thread client.
 envoy_cc_benchmark_binary(
diff --git a/test/common/coroutine/async_queue_test.cc b/test/common/coroutine/async_queue_test.cc
new file mode 100644
index 0000000000000..28982acf6fae6
--- /dev/null
+++ b/test/common/coroutine/async_queue_test.cc
@@ -0,0 +1,1262 @@
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "source/common/coroutine/async_queue.h"
+#include "source/common/coroutine/launch.h"
+#include "source/common/coroutine/task.h"
+
+#include "test/common/coroutine/manual_executor.h"
+#include "test/test_common/status_utility.h"
+#include "test/test_common/utility.h"
+
+#include "gmock/gmock.h"
+#include "gtest/gtest.h"
+
+namespace Envoy {
+namespace Coroutine {
+namespace {
+
+using ::Envoy::StatusHelpers::HasStatusCode;
+
+struct TestByteItem {
+  std::string data;
+};
+
+struct TestByteSizeFunc {
+  uint64_t operator()(const TestByteItem& item) const { return item.data.size(); }
+};
+
+class AsyncQueueTest : public testing::Test {
+public:
+  AsyncQueueTest() : executor_(std::make_shared<ManualExecutor>()) {}
+
+  void drain() { executor_->drain(); }
+
+  void launchTaskOk(Task<absl::Status> task) {
+    handles_.push_back(
+        launch(std::move(task), executor_, [](absl::Status status) { EXPECT_OK(status); }));
+  }
+
+  template <typename T, typename SizeFunc, typename U = T>
+  static Task<absl::Status> pushTask(AsyncQueue<T, SizeFunc>& queue, U item, bool* done = nullptr) {
+    CO_RETURN_IF_ERROR(co_await queue.push(std::move(item)));
+    if (done != nullptr) {
+      *done = true;
+    }
+    co_return absl::OkStatus();
+  }
+
+  template <typename T, typename SizeFunc, typename U = T, typename TrackType = T>
+  static Task<absl::Status> pushTrackTask(AsyncQueue<T, SizeFunc>& queue, U item,
+                                          std::vector<TrackType>* order_vec) {
+    TrackType tracked = item;
+    CO_RETURN_IF_ERROR(co_await queue.push(std::move(item)));
+    if (order_vec != nullptr) {
+      order_vec->push_back(std::move(tracked));
+    }
+    co_return absl::OkStatus();
+  }
+
+  template <typename T, typename SizeFunc>
+  static Task<absl::Status> popTask(AsyncQueue<T, SizeFunc>& queue,
+                                    std::optional<T>* out_val = nullptr, bool* eof_seen = nullptr) {
+    ASSIGN_OR_CO_RETURN(auto res, co_await queue.pop());
+    if (res.has_value()) {
+      if (out_val != nullptr) {
+        *out_val = std::move(*res);
+      }
+    } else {
+      if (eof_seen != nullptr) {
+        *eof_seen = true;
+      }
+    }
+    co_return absl::OkStatus();
+  }
+
+  template <typename T, typename SizeFunc, typename Container>
+  static Task<absl::Status> popMultipleTask(AsyncQueue<T, SizeFunc>& queue, size_t count,
+                                            Container* out_vec) {
+    for (size_t i = 0; i < count; ++i) {
+      ASSIGN_OR_CO_RETURN(auto val_or, co_await queue.pop());
+      if (!val_or.has_value()) {
+        break;
+      }
+      if (out_vec != nullptr) {
+        out_vec->push_back(std::move(*val_or));
+      }
+    }
+    co_return absl::OkStatus();
+  }
+
+  template <typename T, typename SizeFunc, typename U = T>
+  void launchPush(AsyncQueue<T, SizeFunc>& queue, U item, bool* done = nullptr) {
+    launchTaskOk(pushTask(queue, std::move(item), done));
+  }
+
+  template <typename T, typename SizeFunc, typename U = T, typename TrackType>
+  void launchPushTrack(AsyncQueue<T, SizeFunc>& queue, U item, std::vector<TrackType>& order_vec) {
+    launchTaskOk(pushTrackTask(queue, std::move(item), &order_vec));
+  }
+
+  template <typename T, typename SizeFunc>
+  void launchPop(AsyncQueue<T, SizeFunc>& queue, std::optional<T>* out_val = nullptr,
+                 bool* eof_seen = nullptr) {
+    launchTaskOk(popTask(queue, out_val, eof_seen));
+  }
+
+  template <typename T, typename SizeFunc>
+  void launchPop(AsyncQueue<T, SizeFunc>& queue, std::nullptr_t, bool* eof_seen) {
+    launchTaskOk(popTask(queue, static_cast<std::optional<T>*>(nullptr), eof_seen));
+  }
+
+  template <typename T, typename SizeFunc>
+  void launchPop(AsyncQueue<T, SizeFunc>& queue, std::optional<T>& out_val) {
+    launchPop(queue, &out_val, nullptr);
+  }
+
+  template <typename T, typename SizeFunc, typename Container>
+  void launchPopMultiple(AsyncQueue<T, SizeFunc>& queue, size_t count, Container& out_vec) {
+    launchTaskOk(popMultipleTask(queue, count, &out_vec));
+  }
+
+  std::shared_ptr<ManualExecutor> executor_;
+  std::vector<DetachedHandle> handles_;
+};
+
+// ============================================================================
+// 1. Basic Push, Pop, and FIFO Ordering
+// ============================================================================
+
+TEST_F(AsyncQueueTest, UnboundedQueuePushPopFIFO) {
+  AsyncQueue<std::string> queue;
+
+  EXPECT_TRUE(queue.empty());
+  EXPECT_EQ(queue.currentSize(), 0);
+  EXPECT_EQ(queue.itemCount(), 0);
+
+  bool push1_done = false;
+  bool push2_done = false;
+  bool push3_done = false;
+  launchPush(queue, "item1", &push1_done);
+  launchPush(queue, "item2", &push2_done);
+  launchPush(queue, "item3", &push3_done);
+  drain();
+  EXPECT_TRUE(push1_done);
+  EXPECT_TRUE(push2_done);
+  EXPECT_TRUE(push3_done);
+  EXPECT_EQ(queue.itemCount(), 3);
+  EXPECT_EQ(queue.currentSize(), 3);
+
+  std::vector<std::string> received;
+  launchPopMultiple(queue, 3, received);
+  drain();
+  EXPECT_THAT(received, testing::ElementsAre("item1", "item2", "item3"));
+  EXPECT_TRUE(queue.empty());
+}
+
+TEST_F(AsyncQueueTest, BoundedQueueBlocksPusherWhenFull) {
+  // Capacity = 2 items
+  AsyncQueue<int> queue(2);
+
+  std::vector<int> pushed;
+  auto push_task = [&queue, &pushed]() -> Task<absl::Status> {
+    pushed.push_back(1);
+    CO_RETURN_IF_ERROR(co_await queue.push(1));
+    pushed.push_back(2);
+    CO_RETURN_IF_ERROR(co_await queue.push(2));
+    // 3rd push should suspend because capacity is 2
+    pushed.push_back(3);
+    CO_RETURN_IF_ERROR(co_await queue.push(3));
+    pushed.push_back(4);
+    CO_RETURN_IF_ERROR(co_await queue.push(4));
+    co_return absl::OkStatus();
+  };
+
+  launchTaskOk(push_task());
+  drain();
+
+  // Pushes 1 and 2 completed; 3 is waiting in queue_ pending capacity
+  EXPECT_EQ(queue.itemCount(), 3);
+  EXPECT_EQ(pushed.size(), 3);
+
+  // Pop one item, which frees space and unblocks push of 3
+  std::optional<int> pop1;
+  launchPop(queue, pop1);
+  drain();
+  EXPECT_EQ(pop1, 1);
+
+  // Now push of 3 has completed, and push of 4 suspended in queue_ (items: 2, 3 committed + 4
+  // pending)
+  EXPECT_EQ(queue.itemCount(), 3);
+  EXPECT_EQ(pushed.size(), 4);
+
+  // Pop remaining items
+  std::vector<int> remaining_popped;
+  launchPopMultiple(queue, 3, remaining_popped);
+  drain();
+  EXPECT_THAT(remaining_popped, testing::ElementsAre(2, 3, 4));
+  EXPECT_TRUE(queue.empty());
+}
+
+TEST_F(AsyncQueueTest, AbstractCapacityUnitBytes) {
+  // Capacity = 100 bytes
+  AsyncQueue<TestByteItem, TestByteSizeFunc> queue(100);
+
+  EXPECT_EQ(queue.currentSize(), 0);
+
+  bool push1_done = false;
+  bool push2_done = false;
+
+  // 60 bytes
+  launchPush(queue, TestByteItem{std::string(60, 'a')}, &push1_done);
+  // 50 bytes -> 60 + 50 = 110 > 100 bytes, suspends in queue_
+  launchPush(queue, TestByteItem{std::string(50, 'b')}, &push2_done);
+  drain();
+
+  EXPECT_TRUE(push1_done);
+  EXPECT_FALSE(push2_done);
+  // Both items (60 committed + 50 pending) are in queue_ and accounted in currentSize and itemCount
+  EXPECT_EQ(queue.currentSize(), 110);
+  EXPECT_EQ(queue.itemCount(), 2);
+
+  // Pop first item (60 bytes)
+  std::optional<TestByteItem> popped;
+  launchPop(queue, popped);
+  drain();
+
+  ASSERT_TRUE(popped.has_value());
+  EXPECT_EQ(popped->data.size(), 60);
+
+  // Now push2 should have unblocked
+  EXPECT_TRUE(push2_done);
+  EXPECT_EQ(queue.currentSize(), 50);
+  EXPECT_EQ(queue.itemCount(), 1);
+}
+
+TEST_F(AsyncQueueTest, DirectHandoffToWaitingPopper) {
+  AsyncQueue<std::string> queue(1);
+
+  std::optional<std::string> popped;
+  launchPop(queue, popped);
+  drain();
+  EXPECT_FALSE(popped.has_value());
+
+  // Push directly hands off to the waiting popper without queueing
+  launchPush(queue, "direct_message");
+  drain();
+  EXPECT_TRUE(popped.has_value());
+  EXPECT_EQ(*popped, "direct_message");
+  EXPECT_TRUE(queue.empty());
+}
+
+TEST_F(AsyncQueueTest, TryPushAndTryPop) {
+  AsyncQueue<int> queue(1);
+
+  EXPECT_TRUE(queue.tryPush(10));
+  // Queue full (capacity 1)
+  EXPECT_FALSE(queue.tryPush(20));
+
+  auto val = queue.tryPop();
+  EXPECT_TRUE(val.has_value());
+  EXPECT_EQ(*val, 10);
+
+  // Queue now empty
+  EXPECT_FALSE(queue.tryPop().has_value());
+}
+
+TEST_F(AsyncQueueTest, AsyncQueueMoveOnlyTypes) {
+  AsyncQueue<std::unique_ptr<int>> queue;
+
+  auto p1 = std::make_unique<int>(100);
+  auto p2 = std::make_unique<int>(200);
+
+  EXPECT_TRUE(queue.tryPush(std::move(p1)));
+  EXPECT_TRUE(queue.tryPush(std::move(p2)));
+  EXPECT_EQ(queue.itemCount(), 2);
+
+  auto r1 = queue.tryPop();
+  ASSERT_TRUE(r1.has_value());
+  EXPECT_EQ(**r1, 100);
+
+  auto r2 = queue.tryPop();
+  ASSERT_TRUE(r2.has_value());
+  EXPECT_EQ(**r2, 200);
+}
+
+// ============================================================================
+// 2. Close and Cancellation Semantics
+// ============================================================================
+
+TEST_F(AsyncQueueTest, CloseSignalsEOF) {
+  AsyncQueue<std::string> queue;
+
+  queue.tryPush("msg");
+  queue.close();
+
+  // Subsequent push fails
+  EXPECT_FALSE(queue.tryPush("msg2"));
+
+  // First pop gets queued item
+  std::optional<std::string> val1;
+  launchPop(queue, val1);
+  drain();
+  EXPECT_EQ(val1, "msg");
+
+  // Second pop gets EOF (nullopt)
+  bool eof_seen = false;
+  launchPop(queue, nullptr, &eof_seen);
+  drain();
+  EXPECT_TRUE(eof_seen);
+}
+
+TEST_F(AsyncQueueTest, SuspendedPusherFailsWhenQueueIsClosed) {
+  AsyncQueue<int> queue(1);
+  EXPECT_TRUE(queue.tryPush(1));
+
+  absl::Status push_status;
+  auto push_task = [&queue, &push_status]() -> Task<absl::Status> {
+    push_status = co_await queue.push(2);
+    co_return push_status;
+  };
+
+  handles_.push_back(launch(push_task(), executor_, [](absl::Status) {}));
+  drain();
+
+  queue.close();
+  drain();
+
+  // Pop item 1 to free capacity and unblock push(2), which checks closed_ and returns
+  // FailedPrecondition
+  auto item = queue.tryPop();
+  ASSERT_TRUE(item.has_value());
+  EXPECT_EQ(*item, 1);
+  drain();
+
+  EXPECT_THAT(push_status, HasStatusCode(absl::StatusCode::kFailedPrecondition));
+}
+
+TEST_F(AsyncQueueTest, AsyncQueueCloseIdempotent) {
+  AsyncQueue<int> queue;
+  EXPECT_FALSE(queue.closed());
+  queue.close();
+  EXPECT_TRUE(queue.closed());
+  // Idempotent second close
+  queue.close();
+  EXPECT_TRUE(queue.closed());
+}
+
+TEST_F(AsyncQueueTest, PopCancellationUnregistersWaiter) {
+  AsyncQueue<std::string> queue;
+
+  std::optional<absl::StatusOr<std::optional<std::string>>> pop_result;
+  auto pop_task = [&queue, &pop_result]() -> Task<absl::Status> {
+    pop_result = co_await queue.pop();
+    co_return absl::OkStatus();
+  };
+
+  DetachedHandle handle =
+      launch(pop_task(), executor_, [](absl::Status status) { EXPECT_OK(status); });
+  drain();
+  EXPECT_FALSE(pop_result.has_value());
+
+  // Cancel the pop operation
+  handle.cancel();
+  ASSERT_TRUE(pop_result.has_value());
+  EXPECT_TRUE(absl::IsCancelled(pop_result->status()));
+
+  // Ensure pushing now doesn't send to cancelled waiter
+  EXPECT_TRUE(queue.tryPush("item"));
+  EXPECT_EQ(queue.itemCount(), 1);
+}
+
+TEST_F(AsyncQueueTest, PushCancellationUnregistersWaiter) {
+  // Shared capacity limit = 10 units
+  auto shared_cap = std::make_shared<Capacity>(10);
+  AsyncQueue<TestByteItem, TestByteSizeFunc> q1(shared_cap);
+  AsyncQueue<TestByteItem, TestByteSizeFunc> q2(shared_cap);
+
+  // Fill 5 units in q1 -> shared_cap has 5 units free
+  EXPECT_TRUE(q1.tryPush(TestByteItem{std::string(5, 'a')}));
+  EXPECT_EQ(shared_cap->currentPermits(), 5);
+
+  std::optional<absl::Status> push1_status;
+  bool push2_done = false;
+
+  // Push 1 on q2: 20 bytes -> exceeds 10 bytes capacity, suspends
+  auto push1_task = [&q2, &push1_status]() -> Task<absl::Status> {
+    push1_status = co_await q2.push(TestByteItem{std::string(20, 'b')});
+    co_return *push1_status;
+  };
+
+  DetachedHandle h1 = launch(push1_task(), executor_,
+                             [](absl::Status status) { EXPECT_TRUE(absl::IsCancelled(status)); });
+  // Push 2 on q2: 3 bytes -> would fit in 5 free units, but suspended behind push 1
+  launchPush(q2, TestByteItem{std::string(3, 'c')}, &push2_done);
+  drain();
+
+  EXPECT_FALSE(push1_status.has_value());
+  EXPECT_FALSE(push2_done);
+
+  // Cancel push 1
+  h1.cancel();
+  ASSERT_TRUE(push1_status.has_value());
+  EXPECT_TRUE(absl::IsCancelled(*push1_status));
+
+  drain();
+
+  // Push 2 must now be unblocked immediately because push 1 is cancelled
+  EXPECT_TRUE(push2_done);
+  EXPECT_EQ(q2.itemCount(), 1);
+  EXPECT_EQ(shared_cap->currentPermits(), 8); // 5 + 3
+}
+
+// ============================================================================
+// 3. Shared Capacity Across Multiple Queues & Pipelines
+// ============================================================================
+
+TEST_F(AsyncQueueTest, CapacityAcrossMultipleQueues) {
+  auto shared_cap = std::make_shared<Capacity>(3);
+  AsyncQueue<int> q1(shared_cap);
+  AsyncQueue<int> q2(shared_cap);
+
+  EXPECT_EQ(q1.capacity(), shared_cap);
+  EXPECT_EQ(q2.capacity(), shared_cap);
+  EXPECT_EQ(q1.maxSize(), 3);
+  EXPECT_EQ(q2.maxSize(), 3);
+
+  EXPECT_TRUE(q1.tryPush(10));
+  EXPECT_TRUE(q2.tryPush(20));
+  EXPECT_TRUE(q1.tryPush(30));
+
+  // Total shared capacity (3 items) reached.
+  EXPECT_EQ(shared_cap->currentPermits(), 3);
+  EXPECT_EQ(q1.currentSize(), 2);
+  EXPECT_EQ(q2.currentSize(), 1);
+
+  // Pushing into q2 should suspend
+  bool q2_push_done = false;
+  launchPush(q2, 40, &q2_push_done);
+  drain();
+
+  EXPECT_FALSE(q2_push_done);
+  EXPECT_EQ(shared_cap->currentPermits(), 3);
+
+  // Pop from q1, which frees 1 slot in shared capacity and unblocks q2's push
+  auto pop1 = q1.tryPop();
+  ASSERT_TRUE(pop1.has_value());
+  EXPECT_EQ(*pop1, 10);
+
+  drain();
+  EXPECT_TRUE(q2_push_done);
+  EXPECT_EQ(shared_cap->currentPermits(), 3);
+  EXPECT_EQ(q1.currentSize(), 1);
+  EXPECT_EQ(q2.currentSize(), 2);
+}
+
+TEST_F(AsyncQueueTest, CapacityByteBudgetAcrossChainedQueues) {
+  auto shared_cap = std::make_shared<Capacity>(100);
+  AsyncQueue<TestByteItem, TestByteSizeFunc> q1(shared_cap);
+  AsyncQueue<TestByteItem, TestByteSizeFunc> q2(shared_cap);
+
+  EXPECT_TRUE(q1.tryPush(TestByteItem{std::string(60, 'a')}));
+  EXPECT_TRUE(q2.tryPush(TestByteItem{std::string(40, 'b')}));
+  EXPECT_EQ(shared_cap->currentPermits(), 100);
+
+  // Push 30 bytes to q1 -> exceeds 100 byte limit, suspends
+  bool q1_push_done = false;
+  launchPush(q1, TestByteItem{std::string(30, 'c')}, &q1_push_done);
+  drain();
+
+  EXPECT_FALSE(q1_push_done);
+
+  // Pop 40 bytes from q2
+  auto pop_item = q2.tryPop();
+  ASSERT_TRUE(pop_item.has_value());
+  EXPECT_EQ(pop_item->data.size(), 40);
+
+  drain();
+  // q1 unblocks and completes
+  EXPECT_TRUE(q1_push_done);
+  EXPECT_EQ(shared_cap->currentPermits(), 90); // 60 + 30
+}
+
+TEST_F(AsyncQueueTest, CapacityQueueDestructionReleasesCapacity) {
+  auto shared_cap = std::make_shared<Capacity>(2);
+  auto q1 = std::make_unique<AsyncQueue<int>>(shared_cap);
+  AsyncQueue<int> q2(shared_cap);
+
+  EXPECT_TRUE(q1->tryPush(1));
+  EXPECT_TRUE(q1->tryPush(2));
+  EXPECT_EQ(shared_cap->currentPermits(), 2);
+
+  bool q2_push_done = false;
+  launchPush(q2, 3, &q2_push_done);
+  drain();
+  EXPECT_FALSE(q2_push_done);
+
+  // Destroying q1 releases its 2 units from shared capacity
+  q1.reset();
+
+  drain();
+  EXPECT_TRUE(q2_push_done);
+  EXPECT_EQ(shared_cap->currentPermits(), 1);
+}
+
+TEST_F(AsyncQueueTest, DirectHandoffBypassesCapacityWhenFull) {
+  auto shared_cap = std::make_shared<Capacity>(1);
+  AsyncQueue<int> q1(shared_cap);
+  AsyncQueue<int> q2(shared_cap);
+
+  // Fill shared capacity using q1
+  EXPECT_TRUE(q1.tryPush(100));
+  EXPECT_EQ(shared_cap->currentPermits(), 1);
+
+  // Start waiting popper on q2
+  std::optional<int> q2_popped;
+  launchPop(q2, q2_popped);
+  drain();
+  EXPECT_FALSE(q2_popped.has_value());
+
+  // q2.tryPush should succeed via direct handoff even though Capacity is full
+  EXPECT_TRUE(q2.tryPush(200));
+  EXPECT_TRUE(q2_popped.has_value());
+  EXPECT_EQ(*q2_popped, 200);
+  EXPECT_EQ(shared_cap->currentPermits(), 1);
+  EXPECT_TRUE(q2.empty());
+}
+
+TEST_F(AsyncQueueTest, ChainedQueuesPipelineStreamingUnderCapacityConstraint) {
+  // 3-stage pipeline sharing 1 capacity unit: Q1 -> F1 -> Q2 -> F2 -> Q3 -> Sink
+  auto shared_cap = std::make_shared<Capacity>(1);
+  AsyncQueue<int> q1(shared_cap);
+  AsyncQueue<int> q2(shared_cap);
+  AsyncQueue<int> q3(shared_cap);
+
+  std::vector<int> sink_received;
+
+  // Filter 1: pops from q1, multiplies by 10, pushes to q2
+  auto filter1_task = [&q1, &q2]() -> Task<absl::Status> {
+    while (true) {
+      ASSIGN_OR_CO_RETURN(auto item_or, co_await q1.pop());
+      if (!item_or.has_value()) {
+        q2.close();
+        break;
+      }
+      CO_RETURN_IF_ERROR(co_await q2.push(*item_or * 10));
+    }
+    co_return absl::OkStatus();
+  };
+
+  // Filter 2: pops from q2, adds 1, pushes to q3
+  auto filter2_task = [&q2, &q3]() -> Task<absl::Status> {
+    while (true) {
+      ASSIGN_OR_CO_RETURN(auto item_or, co_await q2.pop());
+      if (!item_or.has_value()) {
+        q3.close();
+        break;
+      }
+      CO_RETURN_IF_ERROR(co_await q3.push(*item_or + 1));
+    }
+    co_return absl::OkStatus();
+  };
+
+  // Sink: pops from q3, collects into sink_received
+  auto sink_task = [&q3, &sink_received]() -> Task<absl::Status> {
+    while (true) {
+      ASSIGN_OR_CO_RETURN(auto item_or, co_await q3.pop());
+      if (!item_or.has_value()) {
+        break;
+      }
+      sink_received.push_back(*item_or);
+    }
+    co_return absl::OkStatus();
+  };
+
+  launchTaskOk(filter1_task());
+  launchTaskOk(filter2_task());
+  launchTaskOk(sink_task());
+
+  // Push 3 items into Q1
+  for (int i = 1; i <= 3; ++i) {
+    launchPush(q1, i);
+  }
+
+  drain();
+  q1.close();
+  drain();
+
+  // (1*10+1=11, 2*10+1=21, 3*10+1=31)
+  EXPECT_THAT(sink_received, testing::ElementsAre(11, 21, 31));
+}
+
+TEST_F(AsyncQueueTest, GlobalTemporalFIFOCapacityDistribution) {
+  // Shared capacity limit = 1 unit
+  auto shared_cap = std::make_shared<Capacity>(1);
+  AsyncQueue<std::string> q1(shared_cap);
+  AsyncQueue<std::string> q2(shared_cap);
+  AsyncQueue<std::string> q3(shared_cap);
+
+  // Initial fill
+  EXPECT_TRUE(q1.tryPush("init"));
+  EXPECT_EQ(shared_cap->currentPermits(), 1);
+
+  std::vector<std::string> order_granted;
+
+  // Queue up 2 pushes per queue in interleaved temporal order:
+  // q1_1, q2_1, q3_1, q1_2, q3_2, q2_2
+  launchPushTrack(q1, "q1_1", order_granted);
+  launchPushTrack(q2, "q2_1", order_granted);
+  launchPushTrack(q3, "q3_1", order_granted);
+  launchPushTrack(q1, "q1_2", order_granted);
+  launchPushTrack(q3, "q3_2", order_granted);
+  launchPushTrack(q2, "q2_2", order_granted);
+  drain();
+
+  EXPECT_TRUE(order_granted.empty());
+
+  // Pop initial item to unblock first waiter (q1_1)
+  EXPECT_TRUE(q1.tryPop().has_value());
+  drain();
+  EXPECT_THAT(order_granted, testing::ElementsAre("q1_1"));
+
+  // Pop q1_1 -> unblocks q2_1
+  EXPECT_TRUE(q1.tryPop().has_value());
+  drain();
+  EXPECT_THAT(order_granted, testing::ElementsAre("q1_1", "q2_1"));
+
+  // Pop q2_1 -> unblocks q3_1
+  EXPECT_TRUE(q2.tryPop().has_value());
+  drain();
+  EXPECT_THAT(order_granted, testing::ElementsAre("q1_1", "q2_1", "q3_1"));
+
+  // Pop q3_1 -> unblocks q1_2
+  EXPECT_TRUE(q3.tryPop().has_value());
+  drain();
+  EXPECT_THAT(order_granted, testing::ElementsAre("q1_1", "q2_1", "q3_1", "q1_2"));
+
+  // Pop q1_2 -> unblocks q3_2
+  EXPECT_TRUE(q1.tryPop().has_value());
+  drain();
+  EXPECT_THAT(order_granted, testing::ElementsAre("q1_1", "q2_1", "q3_1", "q1_2", "q3_2"));
+
+  // Pop q3_2 -> unblocks q2_2
+  EXPECT_TRUE(q3.tryPop().has_value());
+  drain();
+  EXPECT_THAT(order_granted, testing::ElementsAre("q1_1", "q2_1", "q3_1", "q1_2", "q3_2", "q2_2"));
+}
+
+// ============================================================================
+// 4. Oversized Items and Anti-Starvation
+// ============================================================================
+
+TEST_F(AsyncQueueTest, LargeChunkAntiStarvation) {
+  // Shared capacity limit = 100 bytes
+  auto shared_cap = std::make_shared<Capacity>(100);
+  AsyncQueue<TestByteItem, TestByteSizeFunc> q1(shared_cap);
+  AsyncQueue<TestByteItem, TestByteSizeFunc> q2(shared_cap);
+
+  // Fill 60 bytes in q1 (40 bytes free in shared_cap)
+  EXPECT_TRUE(q1.tryPush(TestByteItem{std::string(60, 'a')}));
+  EXPECT_EQ(shared_cap->currentPermits(), 60);
+
+  bool large_push_done = false;
+  bool small_push_done = false;
+
+  // Waiter 1 (q1): large chunk of 80 bytes (exceeds 40 free bytes, so suspends)
+  launchPush(q1, TestByteItem{std::string(80, 'b')}, &large_push_done);
+  // Waiter 2 (q2): small chunk of 20 bytes (would fit into 40 free bytes, but behind Waiter 1 in
+  // FIFO)
+  launchPush(q2, TestByteItem{std::string(20, 'c')}, &small_push_done);
+  drain();
+
+  // Strict Head-of-Line FIFO: small push MUST NOT bypass the large push ahead of it
+  EXPECT_FALSE(large_push_done);
+  EXPECT_FALSE(small_push_done);
+  EXPECT_EQ(shared_cap->currentPermits(), 60);
+
+  // Pop 60 bytes from q1 -> current_size becomes 0 -> 100 bytes free
+  auto item = q1.tryPop();
+  ASSERT_TRUE(item.has_value());
+  EXPECT_EQ(item->data.size(), 60);
+
+  drain();
+
+  // Now both should have been granted in order:
+  // 1. Large chunk (80 bytes) granted first -> currentSize becomes 80
+  // 2. Small chunk (20 bytes) granted second -> currentSize becomes 100 (80 + 20)
+  EXPECT_TRUE(large_push_done);
+  EXPECT_TRUE(small_push_done);
+  EXPECT_EQ(shared_cap->currentPermits(), 100);
+  EXPECT_EQ(q1.itemCount(), 1);
+  EXPECT_EQ(q2.itemCount(), 1);
+}
+
+TEST_F(AsyncQueueTest, OversizedChunkAdmissionAfterDrain) {
+  // Shared capacity limit = 50 bytes
+  auto shared_cap = std::make_shared<Capacity>(50);
+  AsyncQueue<TestByteItem, TestByteSizeFunc> q1(shared_cap);
+  AsyncQueue<TestByteItem, TestByteSizeFunc> q2(shared_cap);
+
+  // q1 holds 30 bytes
+  EXPECT_TRUE(q1.tryPush(TestByteItem{std::string(30, 'a')}));
+  EXPECT_EQ(shared_cap->currentPermits(), 30);
+
+  bool oversized_push_done = false;
+  bool normal_push_done = false;
+
+  // Waiter 1 (q2): oversized chunk of 100 bytes (> maxSize 50).
+  // Because current_size == 30 > 0, it suspends and waits in the wait queue.
+  launchPush(q2, TestByteItem{std::string(100, 'b')}, &oversized_push_done);
+
+  // Waiter 2 (q1): normal chunk of 10 bytes (behind oversized chunk in wait queue)
+  launchPush(q1, TestByteItem{std::string(10, 'c')}, &normal_push_done);
+  drain();
+
+  EXPECT_FALSE(oversized_push_done);
+  EXPECT_FALSE(normal_push_done);
+
+  // Pop 30 bytes from q1 -> current_size becomes 0
+  auto item1 = q1.tryPop();
+  ASSERT_TRUE(item1.has_value());
+  EXPECT_EQ(item1->data.size(), 30);
+
+  drain();
+
+  // Oversized chunk is granted when current_size == 0
+  EXPECT_TRUE(oversized_push_done);
+  // Normal chunk is still waiting because current_size is 100 >= 50
+  EXPECT_FALSE(normal_push_done);
+  EXPECT_EQ(shared_cap->currentPermits(), 100);
+
+  // Pop the oversized chunk from q2 -> current_size becomes 0
+  auto item2 = q2.tryPop();
+  ASSERT_TRUE(item2.has_value());
+  EXPECT_EQ(item2->data.size(), 100);
+
+  drain();
+
+  // Now normal chunk is granted
+  EXPECT_TRUE(normal_push_done);
+  EXPECT_EQ(shared_cap->currentPermits(), 10);
+}
+
+// ============================================================================
+// 5. Direct Capacity Semaphore Semantics
+// ============================================================================
+
+TEST_F(AsyncQueueTest, CapacityDirectAcquireRelease) {
+  auto cap = std::make_shared<Capacity>(2);
+  auto res1 = cap->tryAcquire(2);
+  EXPECT_TRUE(res1.has_value());
+  EXPECT_FALSE(cap->tryAcquire(1).has_value());
+  EXPECT_EQ(cap->currentPermits(), 2);
+
+  res1->release();
+  EXPECT_EQ(cap->currentPermits(), 0);
+
+  auto res2 = cap->tryAcquire(1);
+  EXPECT_TRUE(res2.has_value());
+  EXPECT_EQ(cap->currentPermits(), 1);
+
+  res2->release();
+  EXPECT_EQ(cap->currentPermits(), 0);
+}
+
+TEST_F(AsyncQueueTest, CapacityRequestAndCancel) {
+  auto cap = std::make_shared<Capacity>(1);
+  auto init_res = cap->tryAcquire(1);
+  EXPECT_TRUE(init_res.has_value());
+
+  bool task1_done = false;
+  bool task2_done = false;
+  std::optional<absl::Status> task1_status;
+
+  auto t1 = [&]() -> Task<absl::Status> {
+    auto res = co_await cap->acquire(1);
+    task1_status = res.status();
+    task1_done = true;
+    co_return *task1_status;
+  };
+  std::optional<CapacityReservation> task2_res;
+  auto t2 = [&]() -> Task<absl::Status> {
+    ASSIGN_OR_CO_RETURN(task2_res, co_await cap->acquire(1));
+    task2_done = true;
+    co_return absl::OkStatus();
+  };
+
+  DetachedHandle h1 = launch(t1(), executor_, [](absl::Status) {});
+  DetachedHandle h2 = launch(t2(), executor_, [](absl::Status) {});
+  drain();
+
+  EXPECT_FALSE(task1_done);
+  EXPECT_FALSE(task2_done);
+
+  // Cancel task1 (at head of capacity wait list)
+  h1.cancel();
+  drain();
+
+  ASSERT_TRUE(task1_status.has_value());
+  EXPECT_TRUE(absl::IsCancelled(*task1_status));
+
+  // Release capacity: task2 (now head) must unblock
+  init_res->release();
+  drain();
+
+  EXPECT_TRUE(task2_done);
+  EXPECT_EQ(cap->currentPermits(), 1);
+}
+
+TEST_F(AsyncQueueTest, PushOnClosedQueueReturnsError) {
+  AsyncQueue<int> queue;
+  queue.close();
+  EXPECT_FALSE(queue.tryPush(1));
+
+  absl::Status push_status;
+  auto push_task = [&queue, &push_status]() -> Task<absl::Status> {
+    push_status = co_await queue.push(1);
+    co_return push_status;
+  };
+  handles_.push_back(launch(push_task(), executor_, [](absl::Status) {}));
+  drain();
+  EXPECT_THAT(push_status, HasStatusCode(absl::StatusCode::kFailedPrecondition));
+}
+
+TEST_F(AsyncQueueTest, CapacityDestructionAbortsAcquireWaiters) {
+  auto cap = std::make_shared<Capacity>(1);
+  auto init_res = cap->tryAcquire(1);
+  EXPECT_TRUE(init_res.has_value());
+
+  absl::Status waiter_status;
+  auto acquire_task = [&cap, &waiter_status]() -> Task<absl::Status> {
+    auto res = co_await cap->acquire(1);
+    waiter_status = res.status();
+    co_return waiter_status;
+  };
+
+  handles_.push_back(launch(acquire_task(), executor_, [](absl::Status) {}));
+  drain();
+  EXPECT_TRUE(waiter_status.ok());
+
+  // Destroy cap while waiter is pending
+  cap.reset();
+  drain();
+  EXPECT_THAT(waiter_status, HasStatusCode(absl::StatusCode::kFailedPrecondition));
+}
+
+TEST_F(AsyncQueueTest, CapacityReleaseZero) {
+  auto cap = std::make_shared<Capacity>(10);
+  auto zero_res = cap->tryAcquire(0);
+  ASSERT_TRUE(zero_res.has_value());
+  EXPECT_EQ(cap->currentPermits(), 0);
+  zero_res->release();
+  EXPECT_EQ(cap->currentPermits(), 0);
+
+  auto res = cap->tryAcquire(5);
+  EXPECT_TRUE(res.has_value());
+  EXPECT_EQ(cap->currentPermits(), 5);
+
+  res->release();
+  EXPECT_EQ(cap->currentPermits(), 0);
+}
+
+// ============================================================================
+// 6. Ownership, PushAccessor, Move Semantics, and Destruction Safety
+// ============================================================================
+
+TEST_F(AsyncQueueTest, MovedFromQueueIsInert) {
+  AsyncQueue<std::string> q_src(10);
+  AsyncQueue<std::string> queue = std::move(q_src); // q_src is now moved-from, core_ == nullptr
+
+  EXPECT_TRUE(q_src.empty());
+  EXPECT_TRUE(q_src.closed());
+  EXPECT_EQ(q_src.itemCount(), 0);
+  EXPECT_EQ(q_src.currentSize(), 0);
+  EXPECT_EQ(q_src.maxSize(), std::nullopt);
+  EXPECT_EQ(q_src.capacity(), nullptr);
+
+  // tryPush fails on moved-from queue
+  EXPECT_FALSE(q_src.tryPush("hello"));
+
+  // tryPop returns nullopt on moved-from queue
+  EXPECT_FALSE(q_src.tryPop().has_value());
+
+  // pop() returns immediate EOF on moved-from queue
+  bool eof_seen = false;
+  launchPop(q_src, nullptr, &eof_seen);
+  drain();
+  EXPECT_TRUE(eof_seen);
+
+  // push() returns FailedPreconditionError on moved-from queue
+  absl::Status push_status;
+  auto push_task = [&q_src, &push_status]() -> Task<absl::Status> {
+    push_status = co_await q_src.push("hello");
+    co_return push_status;
+  };
+  handles_.push_back(launch(push_task(), executor_, [](absl::Status) {}));
+  drain();
+  EXPECT_THAT(push_status, HasStatusCode(absl::StatusCode::kFailedPrecondition));
+
+  // close() is a safe no-op on moved-from queue
+  q_src.close();
+  EXPECT_TRUE(q_src.closed());
+
+  // pushAccessor returns closed accessor
+  auto pusher = q_src.pushAccessor();
+  EXPECT_TRUE(pusher.closed());
+  EXPECT_FALSE(pusher.tryPush("hello"));
+}
+
+TEST_F(AsyncQueueTest, MoveConstructionTransfersCoreAndLeavesSourceInert) {
+  AsyncQueue<std::string> q1(2);
+  EXPECT_TRUE(q1.tryPush("item1"));
+  auto p1 = q1.pushAccessor();
+
+  // Move-construct q2 from q1
+  AsyncQueue<std::string> q2 = std::move(q1);
+
+  // q1 is now inert
+  EXPECT_TRUE(q1.empty());
+  EXPECT_TRUE(q1.closed());
+  EXPECT_FALSE(q1.tryPush("item_fail"));
+  EXPECT_FALSE(q1.tryPop().has_value());
+
+  // q2 has the item and accepts new items
+  EXPECT_EQ(q2.itemCount(), 1);
+  EXPECT_TRUE(q2.tryPush("item2"));
+  EXPECT_EQ(q2.itemCount(), 2);
+
+  // Existing PushAccessor created on q1 pushes into q2's core
+  EXPECT_FALSE(p1.tryPush("item3")); // queue is full (capacity 2)
+
+  auto pop1 = q2.tryPop();
+  ASSERT_TRUE(pop1.has_value());
+  EXPECT_EQ(*pop1, "item1");
+
+  auto pop2 = q2.tryPop();
+  ASSERT_TRUE(pop2.has_value());
+  EXPECT_EQ(*pop2, "item2");
+}
+
+TEST_F(AsyncQueueTest, MoveAssignmentClosesPreviousCoreAndAdoptsNew) {
+  AsyncQueue<int> q1(2);
+  EXPECT_TRUE(q1.tryPush(42));
+
+  AsyncQueue<int> q2(2);
+  EXPECT_TRUE(q2.tryPush(99));
+
+  bool old_q2_eof = false;
+  launchPop(q2, nullptr, &old_q2_eof);
+  // Pop item 99 from q2, so popper is now suspended on q2 waiting for next item
+  auto pop_old = q2.tryPop();
+  ASSERT_TRUE(pop_old.has_value());
+  EXPECT_EQ(*pop_old, 99);
+  drain();
+  EXPECT_FALSE(old_q2_eof);
+
+  // Move-assign q1 into q2: this closes old q2's core and unblocks old_q2_eof!
+  q2 = std::move(q1);
+  drain();
+  EXPECT_TRUE(old_q2_eof);
+
+  // q1 is now inert
+  EXPECT_TRUE(q1.closed());
+  EXPECT_TRUE(q1.empty());
+
+  // q2 now contains 42 from q1
+  auto pop_new = q2.tryPop();
+  ASSERT_TRUE(pop_new.has_value());
+  EXPECT_EQ(*pop_new, 42);
+}
+
+TEST_F(AsyncQueueTest, PushAccessorBasicPushAndPop) {
+  AsyncQueue<std::string> queue(2);
+  auto pusher = queue.pushAccessor();
+
+  EXPECT_FALSE(pusher.closed());
+  EXPECT_TRUE(pusher.empty());
+  EXPECT_EQ(pusher.currentSize(), 0);
+
+  EXPECT_TRUE(pusher.tryPush("item1"));
+  EXPECT_EQ(queue.itemCount(), 1);
+  EXPECT_EQ(pusher.itemCount(), 1);
+
+  bool push2_done = false;
+  auto push_task = [&pusher, &push2_done]() -> Task<absl::Status> {
+    CO_RETURN_IF_ERROR(co_await pusher.push("item2"));
+    push2_done = true;
+    co_return absl::OkStatus();
+  };
+  handles_.push_back(
+      launch(push_task(), executor_, [](absl::Status status) { EXPECT_OK(status); }));
+  drain();
+  EXPECT_TRUE(push2_done);
+
+  // Pop from the owner (queue)
+  std::vector<std::string> popped;
+  launchPopMultiple(queue, 2, popped);
+  drain();
+  EXPECT_THAT(popped, testing::ElementsAre("item1", "item2"));
+}
+
+TEST_F(AsyncQueueTest, PushAccessorDestructionSafety) {
+  auto queue = std::make_unique<AsyncQueue<int>>(2);
+  auto pusher = queue->pushAccessor();
+
+  EXPECT_TRUE(pusher.tryPush(10));
+  EXPECT_FALSE(pusher.closed());
+
+  // Destroy the owner queue
+  queue.reset();
+
+  // PushAccessor should now detect that the underlying queue is destroyed
+  EXPECT_TRUE(pusher.closed());
+  EXPECT_FALSE(pusher.tryPush(20));
+
+  absl::Status push_status;
+  auto push_task = [&pusher, &push_status]() -> Task<absl::Status> {
+    push_status = co_await pusher.push(30);
+    co_return push_status;
+  };
+  handles_.push_back(launch(push_task(), executor_, [](absl::Status) {}));
+  drain();
+  EXPECT_THAT(push_status, HasStatusCode(absl::StatusCode::kFailedPrecondition));
+}
+
+TEST_F(AsyncQueueTest, PushAccessorSuspendedPushDoesNotPreventCoreDestruction) {
+  auto cap = std::make_shared<Capacity>(1);
+  auto queue = std::make_unique<AsyncQueue<int>>(cap);
+  auto pusher = queue->pushAccessor();
+
+  // Hold capacity
+  auto hold = cap->tryAcquire(1);
+  ASSERT_TRUE(hold.has_value());
+
+  absl::Status push_status;
+  auto push_task = [&pusher, &push_status]() -> Task<absl::Status> {
+    push_status = co_await pusher.push(42);
+    co_return push_status;
+  };
+  handles_.push_back(launch(push_task(), executor_, [](absl::Status) {}));
+  drain();
+
+  // Pusher is suspended waiting for capacity
+  EXPECT_FALSE(pusher.closed());
+
+  // Destroy the owner queue while pusher is suspended on capacity
+  queue.reset();
+
+  // The underlying Core must be immediately destroyed (weak_ptr expired)
+  EXPECT_TRUE(pusher.closed());
+
+  // Release capacity so pusher can resume
+  hold->release();
+  drain();
+
+  // Pusher wakes up, observes Core is destroyed, and returns FailedPreconditionError
+  EXPECT_THAT(push_status, HasStatusCode(absl::StatusCode::kFailedPrecondition));
+}
+
+TEST_F(AsyncQueueTest, DirectHandoffConsumerDestroysQueue) {
+  auto consumer_task = [](std::unique_ptr<AsyncQueue<int>>& q, int expected) -> Task<absl::Status> {
+    auto res = co_await q->pop();
+    EXPECT_TRUE(res.ok());
+    EXPECT_EQ(*res.value(), expected);
+    // Destroy the owner queue inside the callback
+    q.reset();
+    co_return absl::OkStatus();
+  };
+
+  // Case 1: PushAccessor::tryPush triggers direct handoff and consumer destroys queue
+  {
+    auto queue = std::make_unique<AsyncQueue<int>>(10);
+    auto pusher = queue->pushAccessor();
+    launchTaskOk(consumer_task(queue, 42));
+    drain();
+    EXPECT_TRUE(pusher.tryPush(42));
+    EXPECT_EQ(queue, nullptr);
+    EXPECT_TRUE(pusher.closed());
+  }
+
+  // Case 2: AsyncQueue::tryPush triggers direct handoff and consumer destroys queue
+  {
+    auto queue = std::make_unique<AsyncQueue<int>>(10);
+    launchTaskOk(consumer_task(queue, 99));
+    drain();
+    EXPECT_TRUE(queue->tryPush(99));
+    EXPECT_EQ(queue, nullptr);
+  }
+}
+
+TEST_F(AsyncQueueTest, MultiplePushersSinglePopper) {
+  // Multiple producers, one consumer (the owner queue)
+  AsyncQueue<std::string> queue(10);
+  auto p1 = queue.pushAccessor();
+  auto p2 = queue.pushAccessor();
+  auto p3 = queue.pushAccessor();
+
+  EXPECT_TRUE(p1.tryPush("p1_msg"));
+  EXPECT_TRUE(p2.tryPush("p2_msg"));
+  EXPECT_TRUE(p3.tryPush("p3_msg"));
+  EXPECT_EQ(queue.itemCount(), 3);
+
+  std::vector<std::string> received;
+  launchPopMultiple(queue, 3, received);
+  drain();
+  EXPECT_THAT(received, testing::ElementsAre("p1_msg", "p2_msg", "p3_msg"));
+}
+
+TEST_F(AsyncQueueTest, PushAccessorClose) {
+  AsyncQueue<int> queue;
+  auto pusher = queue.pushAccessor();
+
+  EXPECT_TRUE(pusher.tryPush(1));
+  pusher.close();
+
+  EXPECT_TRUE(queue.closed());
+  EXPECT_TRUE(pusher.closed());
+  EXPECT_FALSE(pusher.tryPush(2));
+
+  // Pop remaining item then EOF
+  auto item = queue.tryPop();
+  ASSERT_TRUE(item.has_value());
+  EXPECT_EQ(*item, 1);
+  EXPECT_FALSE(queue.tryPop().has_value());
+}
+
+TEST_F(AsyncQueueTest, DirectResumptionSynchronousExecution) {
+  AsyncQueue<int> queue(1);
+  std::optional<int> popped_value;
+  bool popper_resumed = false;
+
+  auto popper_task = [&]() -> Task<absl::Status> {
+    ASSIGN_OR_CO_RETURN(auto item, co_await queue.pop());
+    popped_value = item;
+    popper_resumed = true;
+    co_return absl::OkStatus();
+  };
+
+  // Launch popper, which suspends waiting for item
+  handles_.push_back(
+      launch(popper_task(), executor_, [](absl::Status status) { EXPECT_OK(status); }));
+  drain();
+  EXPECT_FALSE(popper_resumed);
+  EXPECT_FALSE(popped_value.has_value());
+
+  // Push synchronously wakes up the popper inline during push!
+  EXPECT_TRUE(queue.tryPush(42));
+  EXPECT_TRUE(popper_resumed);
+  EXPECT_EQ(popped_value, 42);
+  EXPECT_TRUE(queue.empty());
+}
+
+TEST_F(AsyncQueueTest, PushPopRendezvousWhenBlockedOnCapacity) {
+  // Shared capacity limit = 1 item
+  auto shared_cap = std::make_shared<Capacity>(1);
+  AsyncQueue<std::unique_ptr<int>> queue(shared_cap);
+
+  // Fill capacity directly
+  auto init_res = shared_cap->tryAcquire(1);
+  ASSERT_TRUE(init_res.has_value());
+  EXPECT_EQ(shared_cap->currentPermits(), 1);
+
+  // Push into queue: capacity is full, so pusher suspends
+  bool push_done = false;
+  launchPush(queue, std::make_unique<int>(99), &push_done);
+  drain();
+  EXPECT_FALSE(push_done);
+  EXPECT_EQ(queue.itemCount(), 1);
+  EXPECT_EQ(queue.currentSize(), 1);
+  EXPECT_FALSE(queue.empty());
+
+  // Pop arrives while pusher is blocked on capacity:
+  // pop() steals the item directly via rendezvous!
+  std::optional<std::unique_ptr<int>> popped;
+  launchPop(queue, popped);
+  drain();
+
+  ASSERT_TRUE(popped.has_value());
+  ASSERT_NE(*popped, nullptr);
+  EXPECT_EQ(**popped, 99);
+  EXPECT_EQ(queue.itemCount(), 0);
+  EXPECT_EQ(queue.currentSize(), 0);
+  EXPECT_TRUE(queue.empty());
+
+  // Releasing capacity unblocks the suspended pusher, which observes the item was
+  // already delivered and completes cleanly with OkStatus without re-buffering.
+  init_res->release();
+  drain();
+
+  EXPECT_TRUE(push_done);
+  EXPECT_EQ(shared_cap->currentPermits(), 0);
+  EXPECT_TRUE(queue.empty());
+}
+
+TEST_F(AsyncQueueTest, TryPushFailurePreservesCallerItem) {
+  // Queue with capacity limit of 1, filled by holding capacity
+  AsyncQueue<std::unique_ptr<int>> queue(1);
+  auto hold = queue.capacity()->tryAcquire(1);
+  ASSERT_TRUE(hold.has_value());
+
+  auto item = std::make_unique<int>(123);
+  EXPECT_FALSE(queue.tryPush(std::move(item)));
+  // item must NOT be dropped on the floor
+  ASSERT_NE(item, nullptr);
+  EXPECT_EQ(*item, 123);
+
+  // Closed queue: tryPush will fail and preserve caller's item
+  AsyncQueue<std::unique_ptr<int>> closed_queue(10);
+  closed_queue.close();
+  EXPECT_FALSE(closed_queue.tryPush(std::move(item)));
+  ASSERT_NE(item, nullptr);
+  EXPECT_EQ(*item, 123);
+
+  // When capacity is released, tryPush succeeds and moves item
+  hold->release();
+  EXPECT_TRUE(queue.tryPush(std::move(item)));
+  EXPECT_EQ(item, nullptr);
+
+  auto popped = queue.tryPop();
+  ASSERT_TRUE(popped.has_value());
+  ASSERT_NE(*popped, nullptr);
+  EXPECT_EQ(**popped, 123);
+}
+
+TEST_F(AsyncQueueTest, QueueDestroyedWithStolenItemAndPendingPusher) {
+  auto cap = std::make_shared<Capacity>(1);
+  auto queue = std::make_unique<AsyncQueue<std::unique_ptr<int>>>(cap);
+
+  auto hold = cap->tryAcquire(1);
+  ASSERT_TRUE(hold.has_value());
+
+  bool push_done = false;
+  handles_.push_back(
+      launch(pushTask(*queue, std::make_unique<int>(42), &push_done), executor_,
+             [](absl::Status status) { EXPECT_TRUE(absl::IsFailedPrecondition(status)); }));
+  drain();
+
+  // Item is in queue_ pending capacity
+  EXPECT_EQ(queue->itemCount(), 1);
+
+  // Consumer steals item via rendezvous
+  std::optional<std::unique_ptr<int>> popped;
+  launchPop(*queue, popped);
+  drain();
+
+  ASSERT_TRUE(popped.has_value());
+  ASSERT_NE(*popped, nullptr);
+  EXPECT_EQ(**popped, 42);
+
+  // Destroy queue while pusher is still suspended on capacity!
+  queue.reset();
+
+  // Now release capacity: pusher wakes up, sees !*alive, and completes cleanly
+  hold->release();
+  drain();
+
+  EXPECT_FALSE(push_done); // Completed with FailedPreconditionError, push_done not set
+}
+
+} // namespace
+} // namespace Coroutine
+} // namespace Envoy
diff --git a/test/common/coroutine/coroutine_test.cc b/test/common/coroutine/coroutine_test.cc
index 699101f30970c..3cfbad18b2c76 100644
--- a/test/common/coroutine/coroutine_test.cc
+++ b/test/common/coroutine/coroutine_test.cc
@@ -138,12 +138,14 @@ TEST(CancellationStateTest, CancelSetsFlagAndIsIdempotent) {
   EXPECT_EQ(1, fired);
 }
 
-TEST(CancellationStateTest, SetCallbackAfterCancelFiresSynchronously) {
+TEST(CancellationStateTest, SetCallbackAfterCancelDoesNotFireOnStack) {
   CancellationState state;
   state.cancel();
   int fired = 0;
-  state.setCancelCallback([&fired] { ++fired; });
-  EXPECT_EQ(1, fired);
+  EXPECT_ENVOY_BUG(
+      { state.setCancelCallback([&fired] { ++fired; }); },
+      "setCancelCallback called on an already-cancelled CancellationState");
+  EXPECT_EQ(0, fired);
 }
 
 TEST(CancellationStateTest, ClearedCallbackDoesNotFire) {
@@ -582,6 +584,171 @@ TEST(StatusMacrosTest, CoReturnIfErrorFailure) {
   EXPECT_EQ(result->message(), "status error");
 }
 
+// ---------------------------------------------------------------------------
+// Additional edge cases and coverage tests
+// ---------------------------------------------------------------------------
+
+TEST(TaskTest, MoveAssignment) {
+  Task<absl::StatusOr<int>> t1 = returnsValue(10);
+  Task<absl::StatusOr<int>> t2 = returnsValue(20);
+  // Overwrite an active task with another active task
+  t1 = std::move(t2);
+
+  auto exec = std::make_shared<ManualExecutor>();
+  std::optional<absl::StatusOr<int>> result;
+  DetachedHandle handle = launch(
+      std::move(t1), exec, [&result](absl::StatusOr<int> val) { result = val; }, StartMode::Inline);
+  ASSERT_TRUE(result.has_value());
+  EXPECT_EQ(**result, 20);
+
+  // Self-assignment
+  Task<absl::StatusOr<int>>* t_ptr = &t1;
+  t1 = std::move(*t_ptr);
+}
+
+namespace {
+Task<absl::StatusOr<std::unique_ptr<int>>> returnsMoveOnly(int val) {
+  co_return std::make_unique<int>(val);
+}
+
+Task<absl::StatusOr<std::unique_ptr<int>>> awaitMoveOnly(int val) {
+  ASSIGN_OR_CO_RETURN(auto ptr, co_await returnsMoveOnly(val));
+  co_return ptr;
+}
+} // namespace
+
+TEST(TaskTest, MoveOnlyReturnType) {
+  auto exec = std::make_shared<ManualExecutor>();
+  std::optional<absl::StatusOr<std::unique_ptr<int>>> result;
+  DetachedHandle handle = launch(
+      awaitMoveOnly(99), exec,
+      [&result](absl::StatusOr<std::unique_ptr<int>> res) { result = std::move(res); },
+      StartMode::Inline);
+  ASSERT_TRUE(result.has_value());
+  ASSERT_TRUE(result->ok());
+  ASSERT_NE(result->value(), nullptr);
+  EXPECT_EQ(*result->value(), 99);
+}
+
+TEST(LaunchTest, DetachedHandleMoveAssignmentAndNull) {
+  auto exec = std::make_shared<ManualExecutor>();
+  bool ran = false;
+  DetachedHandle h1(nullptr);
+  // Cancel on null handle is a safe no-op
+  h1.cancel();
+
+  DetachedHandle h2 = launch(returnsOk(ran), exec, [](absl::Status) {});
+  h1 = std::move(h2);
+  exec->drain();
+  EXPECT_TRUE(ran);
+}
+
+TEST(LeafAwaitableTest, MultipleCompleteCallsAreIdempotent) {
+  auto exec = std::make_shared<ManualExecutor>();
+  LeafController controller;
+  std::optional<absl::Status> result;
+  DetachedHandle handle = launch(awaitLeaf(controller), exec,
+                                 [&result](absl::Status status) { result = std::move(status); });
+  exec->drain();
+  ASSERT_TRUE(controller.started);
+
+  // First completion succeeds
+  controller.completeWith(absl::OkStatus());
+  ASSERT_TRUE(result.has_value());
+  EXPECT_TRUE(result->ok());
+}
+
+TEST(TaskTest, FinalAwaiterAndTaskAwaiterCoverage) {
+  FinalAwaiter final_awaiter;
+  EXPECT_FALSE(final_awaiter.await_ready());
+  final_awaiter.await_resume();
+
+  bool ran = false;
+  Task<absl::Status> t = returnsOk(ran);
+  auto awaiter = std::move(t).operator co_await();
+  EXPECT_FALSE(awaiter.await_ready());
+}
+
+class ImmediateLeaf : public LeafAwaitable<absl::StatusOr<int>> {
+public:
+  ImmediateLeaf(std::optional<int> immediate_val, bool cancel_during_immediate = false)
+      : immediate_val_(immediate_val), cancel_during_immediate_(cancel_during_immediate) {}
+
+  bool started_ = false;
+
+protected:
+  std::optional<absl::StatusOr<int>> tryImmediate() override {
+    if (cancel_during_immediate_) {
+      context().cancellation()->cancel();
+    }
+    if (immediate_val_.has_value()) {
+      return *immediate_val_;
+    }
+    return std::nullopt;
+  }
+
+  void onStart() override {
+    started_ = true;
+    complete(999);
+  }
+  void onCancel() override {}
+
+private:
+  std::optional<int> immediate_val_;
+  bool cancel_during_immediate_ = false;
+};
+
+TEST(LeafAwaitableTest, TryImmediateSuccessAvoidsSuspension) {
+  auto exec = std::make_shared<ManualExecutor>();
+  bool ran = false;
+  std::optional<absl::StatusOr<int>> result;
+
+  auto coro = [&]() -> Task<absl::Status> {
+    ImmediateLeaf leaf(42);
+    ASSIGN_OR_CO_RETURN(int val, co_await leaf);
+    EXPECT_FALSE(leaf.started_);
+    result = val;
+    ran = true;
+    co_return absl::OkStatus();
+  };
+
+  DetachedHandle handle = launch(coro(), exec, [](absl::Status) {}, StartMode::Inline);
+  EXPECT_TRUE(ran);
+  ASSERT_TRUE(result.has_value());
+  EXPECT_OK(*result);
+  EXPECT_EQ(result->value(), 42);
+}
+
+TEST(LeafAwaitableTest, CancellationDuringTryImmediatePreservesResultAndSubsequentAwaitAborts) {
+  auto exec = std::make_shared<ManualExecutor>();
+  bool after_first_await_reached = false;
+  bool after_second_await_reached = false;
+  std::optional<int> received_val;
+  std::optional<absl::Status> final_status;
+
+  auto coro = [&]() -> Task<absl::Status> {
+    ImmediateLeaf leaf1(42, /*cancel_during_immediate=*/true);
+    ASSIGN_OR_CO_RETURN(int val, co_await leaf1);
+    received_val = val;
+    after_first_await_reached = true;
+
+    // Second awaitable must fail-fast due to the cancellation triggered during leaf1
+    ImmediateLeaf leaf2(100);
+    ASSIGN_OR_CO_RETURN(int val2, co_await leaf2);
+    (void)val2;
+    after_second_await_reached = true;
+    co_return absl::OkStatus();
+  };
+
+  DetachedHandle handle = launch(
+      coro(), exec, [&final_status](absl::Status s) { final_status = s; }, StartMode::Inline);
+  EXPECT_TRUE(after_first_await_reached);
+  EXPECT_EQ(received_val, 42);
+  EXPECT_FALSE(after_second_await_reached);
+  ASSERT_TRUE(final_status.has_value());
+  EXPECT_TRUE(absl::IsCancelled(*final_status));
+}
+
 } // namespace
 } // namespace Coroutine
 } // namespace Envoy
diff --git a/test/common/coroutine/perf_test.cc b/test/common/coroutine/perf_test.cc
index 2409c5bf63e88..c1d2a468fabef 100644
--- a/test/common/coroutine/perf_test.cc
+++ b/test/common/coroutine/perf_test.cc
@@ -402,9 +402,7 @@ Coroutine::Task<absl::StatusOr<uint64_t>> AsyncSocket::read(Buffer::Instance& bu
       co_return 0;
     }
     ++read_blocks_;
-    if (absl::Status s = co_await whenReady(Read); !s.ok()) {
-      co_return s;
-    }
+    CO_RETURN_IF_ERROR(co_await whenReady(Read));
   }
 }
 
@@ -418,9 +416,7 @@ Coroutine::Task<absl::StatusOr<uint64_t>> AsyncSocket::write(Buffer::Instance& b
       co_return absl::InternalError("write() failed");
     }
     ++write_blocks_;
-    if (absl::Status s = co_await whenReady(Write); !s.ok()) {
-      co_return s;
-    }
+    CO_RETURN_IF_ERROR(co_await whenReady(Write));
   }
 }
 
@@ -430,23 +426,16 @@ Coroutine::Task<absl::Status> coroEcho(AsyncSocket& sock, uint32_t high) {
   while (true) {
     if (buf.length() < high) {
       bool await = buf.length() == 0;
-      absl::StatusOr<uint64_t> n = co_await sock.read(buf, ReadSize, await);
-      if (!n.ok()) {
-        co_return n.status();
-      }
-      if (*n == 0 && await) {
+      ASSIGN_OR_CO_RETURN(uint64_t n, co_await sock.read(buf, ReadSize, await));
+      if (n == 0 && await) {
         co_return absl::InternalError("peer closed before echo completed");
       }
     }
     if (buf.length() > 0) {
-      absl::StatusOr<uint64_t> wrote = co_await sock.write(buf);
-      if (!wrote.ok()) {
-        co_return wrote.status();
-      }
-    }
-    if (absl::Status s = co_await YieldToNextIteration(sock); !s.ok()) {
-      co_return s;
+      ASSIGN_OR_CO_RETURN(auto wrote, co_await sock.write(buf));
+      (void)wrote;
     }
+    CO_RETURN_IF_ERROR(co_await YieldToNextIteration(sock));
   }
 }
 
diff --git a/test/common/coroutine/semaphore_test.cc b/test/common/coroutine/semaphore_test.cc
new file mode 100644
index 0000000000000..a9690b18f0e9a
--- /dev/null
+++ b/test/common/coroutine/semaphore_test.cc
@@ -0,0 +1,376 @@
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "source/common/coroutine/launch.h"
+#include "source/common/coroutine/semaphore.h"
+#include "source/common/coroutine/task.h"
+
+#include "test/common/coroutine/manual_executor.h"
+#include "test/test_common/status_utility.h"
+#include "test/test_common/utility.h"
+
+#include "gmock/gmock.h"
+#include "gtest/gtest.h"
+
+namespace Envoy {
+namespace Coroutine {
+namespace {
+
+using ::Envoy::StatusHelpers::HasStatusCode;
+
+class SemaphoreTest : public testing::Test {
+public:
+  SemaphoreTest() : executor_(std::make_shared<ManualExecutor>()) {}
+
+  void drain() { executor_->drain(); }
+
+  void launchTaskOk(Task<absl::Status> task) {
+    handles_.push_back(
+        launch(std::move(task), executor_, [](absl::Status status) { EXPECT_OK(status); }));
+  }
+
+  std::shared_ptr<ManualExecutor> executor_;
+  std::vector<DetachedHandle> handles_;
+};
+
+TEST_F(SemaphoreTest, UnboundedSemaphoreAcquireRelease) {
+  auto sem = std::make_shared<Semaphore>();
+  EXPECT_FALSE(sem->maxPermits().has_value());
+  EXPECT_EQ(sem->currentPermits(), 0);
+
+  auto res1 = sem->tryAcquire(5);
+  ASSERT_TRUE(res1.has_value());
+  EXPECT_EQ(res1->permits(), 5);
+  EXPECT_EQ(sem->currentPermits(), 5);
+
+  auto res2 = sem->tryAcquire(10);
+  ASSERT_TRUE(res2.has_value());
+  EXPECT_EQ(res2->permits(), 10);
+  EXPECT_EQ(sem->currentPermits(), 15);
+
+  res1->release();
+  EXPECT_EQ(res1->permits(), 0);
+  EXPECT_EQ(sem->currentPermits(), 10);
+
+  res2.reset();
+  EXPECT_EQ(sem->currentPermits(), 0);
+}
+
+TEST_F(SemaphoreTest, BoundedSemaphoreTryAcquire) {
+  auto sem = std::make_shared<Semaphore>(10);
+  EXPECT_EQ(sem->maxPermits().value(), 10);
+  EXPECT_EQ(sem->currentPermits(), 0);
+
+  auto res1 = sem->tryAcquire(6);
+  ASSERT_TRUE(res1.has_value());
+  EXPECT_EQ(sem->currentPermits(), 6);
+
+  // Exceeds remaining capacity (4 available, requesting 5)
+  auto res2 = sem->tryAcquire(5);
+  EXPECT_FALSE(res2.has_value());
+  EXPECT_EQ(sem->currentPermits(), 6);
+
+  // Fits in remaining capacity
+  auto res3 = sem->tryAcquire(4);
+  ASSERT_TRUE(res3.has_value());
+  EXPECT_EQ(sem->currentPermits(), 10);
+
+  res1.reset();
+  EXPECT_EQ(sem->currentPermits(), 4);
+  drain();
+
+  // Now 5 fits
+  auto res4 = sem->tryAcquire(5);
+  ASSERT_TRUE(res4.has_value());
+  EXPECT_EQ(sem->currentPermits(), 9);
+}
+
+TEST_F(SemaphoreTest, AsyncAcquireAndFifoOrder) {
+  auto sem = std::make_shared<Semaphore>(10);
+  auto hold = sem->tryAcquire(10);
+  ASSERT_TRUE(hold.has_value());
+
+  std::vector<int> acquired_order;
+  std::vector<SemaphoreReservation> held;
+
+  auto acquireTask = [](SemaphorePtr s, uint64_t permits, int id, std::vector<int>* order,
+                        std::vector<SemaphoreReservation>* held_res) -> Task<absl::Status> {
+    ASSIGN_OR_CO_RETURN(auto res, co_await s->acquire(permits));
+    order->push_back(id);
+    held_res->push_back(std::move(res));
+    co_return absl::OkStatus();
+  };
+
+  launchTaskOk(acquireTask(sem, 5, 1, &acquired_order, &held));
+  launchTaskOk(acquireTask(sem, 5, 2, &acquired_order, &held));
+  launchTaskOk(acquireTask(sem, 5, 3, &acquired_order, &held));
+  drain();
+  EXPECT_TRUE(acquired_order.empty());
+
+  // Release initial reservation; waiter 1 and waiter 2 should be satisfied up to capacity 10.
+  hold.reset();
+  drain();
+  EXPECT_THAT(acquired_order, testing::ElementsAre(1, 2));
+
+  // Release waiter 1's permits; waiter 3 should now be satisfied.
+  held[0].release();
+  drain();
+  EXPECT_THAT(acquired_order, testing::ElementsAre(1, 2, 3));
+}
+
+TEST_F(SemaphoreTest, FifoHeadOfLineBlocking) {
+  auto sem = std::make_shared<Semaphore>(10);
+  auto hold = sem->tryAcquire(10);
+  ASSERT_TRUE(hold.has_value());
+
+  std::vector<int> acquired_order;
+  std::vector<SemaphoreReservation> held;
+
+  auto acquireTask = [](SemaphorePtr s, uint64_t permits, int id, std::vector<int>* order,
+                        std::vector<SemaphoreReservation>* held_res) -> Task<absl::Status> {
+    ASSIGN_OR_CO_RETURN(auto res, co_await s->acquire(permits));
+    order->push_back(id);
+    held_res->push_back(std::move(res));
+    co_return absl::OkStatus();
+  };
+
+  // Waiter 1 requests 8 permits. Waiter 2 requests 2 permits.
+  launchTaskOk(acquireTask(sem, 8, 1, &acquired_order, &held));
+  launchTaskOk(acquireTask(sem, 2, 2, &acquired_order, &held));
+  drain();
+  EXPECT_TRUE(acquired_order.empty());
+
+  // Release hold (10 permits); both waiter 1 (8 permits) and waiter 2 (2 permits) fit.
+  hold.reset();
+  drain();
+  EXPECT_THAT(acquired_order, testing::ElementsAre(1, 2));
+}
+
+TEST_F(SemaphoreTest, CancellationUnblocksNextWaiters) {
+  auto sem = std::make_shared<Semaphore>(10);
+  auto hold = sem->tryAcquire(10);
+  ASSERT_TRUE(hold.has_value());
+
+  std::vector<int> acquired_order;
+  std::vector<SemaphoreReservation> held;
+
+  auto acquireTask = [](SemaphorePtr s, uint64_t permits, int id, std::vector<int>* order,
+                        std::vector<SemaphoreReservation>* held_res) -> Task<absl::Status> {
+    ASSIGN_OR_CO_RETURN(auto res, co_await s->acquire(permits));
+    order->push_back(id);
+    held_res->push_back(std::move(res));
+    co_return absl::OkStatus();
+  };
+
+  DetachedHandle h1 =
+      launch(acquireTask(sem, 8, 1, &acquired_order, &held), executor_, [](absl::Status status) {
+        EXPECT_THAT(status, HasStatusCode(absl::StatusCode::kCancelled));
+      });
+  DetachedHandle h2 = launch(acquireTask(sem, 4, 2, &acquired_order, &held), executor_,
+                             [](absl::Status status) { EXPECT_OK(status); });
+  drain();
+  EXPECT_TRUE(acquired_order.empty());
+
+  // Cancel waiter 1 (head of line).
+  h1.cancel();
+  drain();
+
+  // Release hold (10 permits); waiter 2 is now the head and acquires 4 permits.
+  hold.reset();
+  drain();
+  EXPECT_THAT(acquired_order, testing::ElementsAre(2));
+}
+
+TEST_F(SemaphoreTest, DestructionFailsWaiters) {
+  auto sem = std::make_shared<Semaphore>(10);
+  auto hold = sem->tryAcquire(10);
+  ASSERT_TRUE(hold.has_value());
+
+  bool waiter_failed = false;
+  auto acquireTask = [](Semaphore& s, bool* failed) -> Task<absl::Status> {
+    auto res = co_await s.acquire(5);
+    if (!res.ok() && res.status().code() == absl::StatusCode::kFailedPrecondition) {
+      *failed = true;
+    }
+    co_return absl::OkStatus();
+  };
+
+  launchTaskOk(acquireTask(*sem, &waiter_failed));
+  drain();
+  EXPECT_FALSE(waiter_failed);
+
+  // Destroy semaphore while waiter is pending.
+  sem.reset();
+  drain();
+  EXPECT_TRUE(waiter_failed);
+}
+
+TEST_F(SemaphoreTest, ReservationLifecycle) {
+  // Empty reservation
+  SemaphoreReservation empty_res;
+  EXPECT_FALSE(empty_res.hasPermits());
+
+  // Active reservation acquires permits and hasPermits() is true
+  auto sem = std::make_shared<Semaphore>(10);
+  auto res1 = sem->tryAcquire(5);
+  ASSERT_TRUE(res1.has_value());
+  EXPECT_TRUE(res1->hasPermits());
+  EXPECT_EQ(res1->permits(), 5);
+
+  // Explicit release clears permits
+  res1->release();
+  EXPECT_FALSE(res1->hasPermits());
+  EXPECT_EQ(sem->currentPermits(), 0);
+
+  // Reservation outliving semaphore releases safely without crash or leak
+  SemaphoreReservation outliving_res;
+  {
+    auto scoped_sem = std::make_shared<Semaphore>(10);
+    auto opt = scoped_sem->tryAcquire(4);
+    ASSERT_TRUE(opt.has_value());
+    outliving_res = std::move(*opt);
+    EXPECT_EQ(outliving_res.permits(), 4);
+    // `scoped_sem` is destroyed here.
+  }
+  EXPECT_FALSE(outliving_res.hasPermits()); // `sem` is destroyed
+  outliving_res.release();                  // Dropping after destruction is safe
+  EXPECT_EQ(outliving_res.permits(), 0);
+
+  // Move-assignment over an active reservation triggers ENVOY_BUG
+  auto r1 = sem->tryAcquire(4);
+  auto r2 = sem->tryAcquire(3);
+  ASSERT_TRUE(r1.has_value());
+  ASSERT_TRUE(r2.has_value());
+  EXPECT_EQ(sem->currentPermits(), 7);
+  EXPECT_ENVOY_BUG(*r1 = std::move(*r2),
+                   "SemaphoreReservation should not overwrite an active reservation");
+}
+
+TEST_F(SemaphoreTest, UnboundedAsyncAcquire) {
+  auto sem = std::make_shared<Semaphore>(); // unbounded
+  auto hold = sem->tryAcquire(100);
+  ASSERT_TRUE(hold.has_value());
+
+  auto acquireTask = [](SemaphorePtr s, bool* acquired) -> Task<absl::Status> {
+    ASSIGN_OR_CO_RETURN(auto res, co_await s->acquire(50));
+    *acquired = true;
+    co_return absl::OkStatus();
+  };
+
+  bool acquired = false;
+  launchTaskOk(acquireTask(sem, &acquired));
+  drain();
+  EXPECT_TRUE(acquired);
+}
+
+TEST_F(SemaphoreTest, PermitBoundaryConditions) {
+  auto sem = std::make_shared<Semaphore>(10);
+
+  // Zero-permit acquire and release are no-ops
+  auto zero_res1 = sem->tryAcquire(0);
+  ASSERT_TRUE(zero_res1.has_value());
+  EXPECT_FALSE(zero_res1->hasPermits());
+  zero_res1->release();
+  EXPECT_EQ(sem->currentPermits(), 0);
+
+  auto res = sem->tryAcquire(5);
+  ASSERT_TRUE(res.has_value());
+  EXPECT_EQ(sem->currentPermits(), 5);
+
+  auto zero_res2 = sem->tryAcquire(0);
+  ASSERT_TRUE(zero_res2.has_value());
+  EXPECT_FALSE(zero_res2->hasPermits());
+  zero_res2->release();
+  EXPECT_EQ(sem->currentPermits(), 5);
+
+  // When partially in use, oversized/overflow acquire requests are rejected
+  uint64_t huge_permits = std::numeric_limits<uint64_t>::max();
+  EXPECT_FALSE(sem->tryAcquire(huge_permits).has_value());
+
+  res->release();
+  EXPECT_EQ(sem->currentPermits(), 0);
+}
+
+TEST_F(SemaphoreTest, ReleaseWhenAllWaitersCancelled) {
+  auto sem = std::make_shared<Semaphore>(1);
+  auto hold = sem->tryAcquire(1);
+  ASSERT_TRUE(hold.has_value());
+
+  auto acquireTask = [](Semaphore& s) -> Task<absl::Status> {
+    auto res = co_await s.acquire(1);
+    co_return absl::OkStatus();
+  };
+
+  auto handle = launch(acquireTask(*sem), executor_, [](absl::Status) {});
+  drain();
+
+  // Cancel the pending waiter
+  handle.cancel();
+
+  // Release capacity: scheduleProcessWaiters pops the cancelled waiter and sees empty list
+  hold->release();
+  drain();
+  EXPECT_EQ(sem->currentPermits(), 0);
+}
+
+TEST_F(SemaphoreTest, DestructionWhileProcessWaitersScheduled) {
+  auto sem = std::make_shared<Semaphore>(10);
+  auto hold = sem->tryAcquire(10);
+  ASSERT_TRUE(hold.has_value());
+
+  auto acquireTask = [](Semaphore& s, bool* failed) -> Task<absl::Status> {
+    auto res = co_await s.acquire(5);
+    if (!res.ok()) {
+      *failed = true;
+    }
+    co_return absl::OkStatus();
+  };
+
+  bool waiter_failed = false;
+  launchTaskOk(acquireTask(*sem, &waiter_failed));
+  drain(); // Waiter is now queued in waiters_
+
+  // Release permits: this schedules runScheduledProcessWaiters on executor_
+  hold->release();
+  // Destroy semaphore while scheduled task is in executor queue:
+  sem.reset();
+  // Now drain: scheduled task runs, observes !*alive, and completes without crash
+  drain();
+  EXPECT_TRUE(waiter_failed);
+}
+
+TEST_F(SemaphoreTest, CancellationDuringScheduledProcessing) {
+  auto sem = std::make_shared<Semaphore>(10);
+  auto hold = sem->tryAcquire(10);
+  ASSERT_TRUE(hold.has_value());
+
+  auto acquireTask = [](SemaphorePtr s,
+                        SemaphoreReservation* out_res = nullptr) -> Task<absl::Status> {
+    ASSIGN_OR_CO_RETURN(auto res, co_await s->acquire(5));
+    if (out_res != nullptr) {
+      *out_res = std::move(res);
+    }
+    co_return absl::OkStatus();
+  };
+
+  SemaphoreReservation w1_res;
+  launchTaskOk(acquireTask(sem, &w1_res));
+  DetachedHandle h2 = launch(acquireTask(sem), executor_, [](absl::Status) {});
+  DetachedHandle h3 = launch(acquireTask(sem), executor_, [](absl::Status) {});
+  drain(); // 3 waiters queued
+
+  // Cancel trailing waiters h2 and h3
+  h2.cancel();
+  h3.cancel();
+
+  // Release hold: processWaiters satisfies w1 and prunes cancelled waiters h2 and h3
+  hold->release();
+  drain();
+  EXPECT_TRUE(w1_res.hasPermits());
+  EXPECT_EQ(sem->currentPermits(), 5);
+}
+
+} // namespace
+} // namespace Coroutine
+} // namespace Envoy