diff --git a/common/BUILD b/common/BUILD
index f75c82f17..5b6349624 100644
--- a/common/BUILD
+++ b/common/BUILD
@@ -504,6 +504,12 @@ cc_library(
     hdrs = ["memory.h"],
     deps = [
         ":allocator",
+        ":arena",
+        ":data",
+        ":reference_count",
+        "//common/internal:metadata",
+        "//common/internal:reference_count",
+        "//internal:exceptions",
         "//internal:to_address",
         "@com_google_absl//absl/base:core_headers",
         "@com_google_absl//absl/base:no_destructor",
@@ -514,6 +520,21 @@ cc_library(
     ],
 )
 
+cc_test(
+    name = "memory_test",
+    srcs = ["memory_test.cc"],
+    deps = [
+        ":allocator",
+        ":data",
+        ":memory",
+        "//common/internal:reference_count",
+        "//internal:testing",
+        "@com_google_absl//absl/base:nullability",
+        "@com_google_protobuf//:protobuf",
+        "@com_google_protobuf//:struct_cc_proto",
+    ],
+)
+
 cc_library(
     name = "memory_testing",
     testonly = True,
diff --git a/common/memory.h b/common/memory.h
index 967bc57e2..615bce519 100644
--- a/common/memory.h
+++ b/common/memory.h
@@ -16,16 +16,25 @@
 #define THIRD_PARTY_CEL_CPP_COMMON_MEMORY_H_
 
 #include <cstddef>
+#include <cstdint>
 #include <memory>
 #include <ostream>
 #include <type_traits>
 #include <utility>
 
 #include "absl/base/attributes.h"
+#include "absl/base/macros.h"
 #include "absl/base/nullability.h"
+#include "absl/base/optimization.h"
 #include "absl/log/absl_check.h"
 #include "absl/numeric/bits.h"
 #include "common/allocator.h"
+#include "common/arena.h"
+#include "common/data.h"
+#include "common/internal/metadata.h"
+#include "common/internal/reference_count.h"
+#include "common/reference_count.h"
+#include "internal/exceptions.h"
 #include "internal/to_address.h"  // IWYU pragma: keep
 #include "google/protobuf/arena.h"
 
@@ -48,10 +57,1252 @@ enum class MemoryManagement {
 
 std::ostream& operator<<(std::ostream& out, MemoryManagement memory_management);
 
+class ABSL_ATTRIBUTE_TRIVIAL_ABI [[nodiscard]] Owner;
+class Borrower;
+template <typename T>
+class ABSL_ATTRIBUTE_TRIVIAL_ABI [[nodiscard]] Unique;
+template <typename T>
+class ABSL_ATTRIBUTE_TRIVIAL_ABI [[nodiscard]] Owned;
+template <typename T>
+class Borrowed;
+template <typename T>
+struct Ownable;
+template <typename T>
+struct Borrowable;
+
 class MemoryManager;
 class ReferenceCountingMemoryManager;
 class PoolingMemoryManager;
 
+namespace common_internal {
+template <typename T>
+inline constexpr bool kNotMessageLiteAndNotData =
+    std::conjunction_v<std::negation<std::is_base_of<google::protobuf::MessageLite, T>>,
+                       std::negation<std::is_base_of<Data, T>>>;
+template <typename To, typename From>
+inline constexpr bool kIsPointerConvertible = std::is_convertible_v<From*, To*>;
+template <typename To, typename From>
+inline constexpr bool kNotSameAndIsPointerConvertible =
+    std::conjunction_v<std::negation<std::is_same<To, From>>,
+                       std::bool_constant<kIsPointerConvertible<To, From>>>;
+
+// Clears the contents of `owner`, and returns the reference count if in use.
+const ReferenceCount* absl_nullable OwnerRelease(Owner owner) noexcept;
+const ReferenceCount* absl_nullable BorrowerRelease(Borrower borrower) noexcept;
+template <typename T>
+Owned<const T> WrapEternal(const T* value);
+
+// Pointer tag used by `cel::Unique` to indicate that the destructor needs to be
+// registered with the arena, but it has not been done yet. Must be done when
+// releasing.
+inline constexpr uintptr_t kUniqueArenaUnownedBit = uintptr_t{1} << 0;
+inline constexpr uintptr_t kUniqueArenaBits = kUniqueArenaUnownedBit;
+inline constexpr uintptr_t kUniqueArenaPointerMask = ~kUniqueArenaBits;
+}  // namespace common_internal
+
+template <typename T, typename... Args>
+Owned<T> AllocateShared(Allocator<> allocator, Args&&... args);
+
+template <typename T>
+Owned<T> WrapShared(T* object, Allocator<> allocator);
+
+// `Owner` represents a reference to some co-owned data, of which this owner is
+// one of the co-owners. When using reference counting, `Owner` performs
+// increment/decrement where appropriate similar to `std::shared_ptr`.
+// `Borrower` is similar to `Owner`, except that it is always trivially
+// copyable/destructible. In that sense, `Borrower` is similar to
+// `std::reference_wrapper<const Owner>`.
+class ABSL_ATTRIBUTE_TRIVIAL_ABI [[nodiscard]] Owner final {
+ private:
+  static constexpr uintptr_t kNone = common_internal::kMetadataOwnerNone;
+  static constexpr uintptr_t kReferenceCountBit =
+      common_internal::kMetadataOwnerReferenceCountBit;
+  static constexpr uintptr_t kArenaBit =
+      common_internal::kMetadataOwnerArenaBit;
+  static constexpr uintptr_t kBits = common_internal::kMetadataOwnerBits;
+  static constexpr uintptr_t kPointerMask =
+      common_internal::kMetadataOwnerPointerMask;
+
+ public:
+  static Owner None() noexcept { return Owner(); }
+
+  static Owner Allocator(Allocator<> allocator) noexcept {
+    auto* arena = allocator.arena();
+    return arena != nullptr ? Arena(arena) : None();
+  }
+
+  static Owner Arena(google::protobuf::Arena* absl_nonnull arena
+                         ABSL_ATTRIBUTE_LIFETIME_BOUND) noexcept {
+    ABSL_DCHECK(arena != nullptr);
+    return Owner(reinterpret_cast<uintptr_t>(arena) | kArenaBit);
+  }
+
+  static Owner Arena(std::nullptr_t) = delete;
+
+  static Owner ReferenceCount(const ReferenceCount* absl_nonnull reference_count
+                                  ABSL_ATTRIBUTE_LIFETIME_BOUND) noexcept {
+    ABSL_DCHECK(reference_count != nullptr);
+    common_internal::StrongRef(*reference_count);
+    return Owner(reinterpret_cast<uintptr_t>(reference_count) |
+                 kReferenceCountBit);
+  }
+
+  static Owner ReferenceCount(std::nullptr_t) = delete;
+
+  Owner() = default;
+
+  Owner(const Owner& other) noexcept : Owner(CopyFrom(other.ptr_)) {}
+
+  Owner(Owner&& other) noexcept : Owner(MoveFrom(other.ptr_)) {}
+
+  template <typename T>
+  // NOLINTNEXTLINE(google-explicit-constructor)
+  Owner(const Owned<T>& owned) noexcept;
+
+  template <typename T>
+  // NOLINTNEXTLINE(google-explicit-constructor)
+  Owner(Owned<T>&& owned) noexcept;
+
+  explicit Owner(Borrower borrower) noexcept;
+
+  template <typename T>
+  explicit Owner(Borrowed<T> borrowed) noexcept;
+
+  ~Owner() { Destroy(ptr_); }
+
+  Owner& operator=(const Owner& other) noexcept {
+    if (ptr_ != other.ptr_) {
+      Destroy(ptr_);
+      ptr_ = CopyFrom(other.ptr_);
+    }
+    return *this;
+  }
+
+  Owner& operator=(Owner&& other) noexcept {
+    if (ABSL_PREDICT_TRUE(this != &other)) {
+      Destroy(ptr_);
+      ptr_ = MoveFrom(other.ptr_);
+    }
+    return *this;
+  }
+
+  template <typename T>
+  // NOLINTNEXTLINE(google-explicit-constructor)
+  Owner& operator=(const Owned<T>& owned) noexcept;
+
+  template <typename T>
+  // NOLINTNEXTLINE(google-explicit-constructor)
+  Owner& operator=(Owned<T>&& owned) noexcept;
+
+  explicit operator bool() const noexcept { return !IsNone(ptr_); }
+
+  google::protobuf::Arena* absl_nullable arena() const noexcept {
+    return (ptr_ & Owner::kBits) == Owner::kArenaBit
+               ? reinterpret_cast<google::protobuf::Arena*>(ptr_ & Owner::kPointerMask)
+               : nullptr;
+  }
+
+  void reset() noexcept {
+    Destroy(ptr_);
+    ptr_ = 0;
+  }
+
+  // Tests whether two owners have ownership over the same data, that is they
+  // are co-owners.
+  friend bool operator==(const Owner& lhs, const Owner& rhs) noexcept {
+    // A reference count and arena can never occupy the same memory address, so
+    // we can compare for equality without masking off the bits.
+    return lhs.ptr_ == rhs.ptr_;
+  }
+
+ private:
+  template <typename T>
+  friend class Unique;
+  friend class Borrower;
+  template <typename T, typename... Args>
+  friend Owned<T> AllocateShared(cel::Allocator<> allocator, Args&&... args);
+  template <typename T>
+  friend Owned<T> WrapShared(T* object, cel::Allocator<> allocator);
+  template <typename U>
+  friend struct Ownable;
+  friend const common_internal::ReferenceCount* absl_nullable
+  common_internal::OwnerRelease(Owner owner) noexcept;
+  friend const common_internal::ReferenceCount* absl_nullable
+  common_internal::BorrowerRelease(Borrower borrower) noexcept;
+  friend struct ArenaTraits<Owner>;
+
+  constexpr explicit Owner(uintptr_t ptr) noexcept : ptr_(ptr) {}
+
+  static constexpr bool IsNone(uintptr_t ptr) noexcept { return ptr == kNone; }
+
+  static constexpr bool IsArena(uintptr_t ptr) noexcept {
+    return (ptr & kArenaBit) != kNone;
+  }
+
+  static constexpr bool IsReferenceCount(uintptr_t ptr) noexcept {
+    return (ptr & kReferenceCountBit) != kNone;
+  }
+
+  ABSL_ATTRIBUTE_RETURNS_NONNULL
+  static google::protobuf::Arena* absl_nonnull AsArena(uintptr_t ptr) noexcept {
+    ABSL_ASSERT(IsArena(ptr));
+    return reinterpret_cast<google::protobuf::Arena*>(ptr & kPointerMask);
+  }
+
+  ABSL_ATTRIBUTE_RETURNS_NONNULL
+  static const common_internal::ReferenceCount* absl_nonnull AsReferenceCount(
+      uintptr_t ptr) noexcept {
+    ABSL_ASSERT(IsReferenceCount(ptr));
+    return reinterpret_cast<const common_internal::ReferenceCount*>(
+        ptr & kPointerMask);
+  }
+
+  static uintptr_t CopyFrom(uintptr_t other) noexcept { return Own(other); }
+
+  static uintptr_t MoveFrom(uintptr_t& other) noexcept {
+    return std::exchange(other, kNone);
+  }
+
+  static void Destroy(uintptr_t ptr) noexcept { Unown(ptr); }
+
+  static uintptr_t Own(uintptr_t ptr) noexcept {
+    if (IsReferenceCount(ptr)) {
+      const auto* refcount = Owner::AsReferenceCount(ptr);
+      ABSL_ASSUME(refcount != nullptr);
+      common_internal::StrongRef(refcount);
+    }
+    return ptr;
+  }
+
+  static void Unown(uintptr_t ptr) noexcept {
+    if (IsReferenceCount(ptr)) {
+      const auto* reference_count = AsReferenceCount(ptr);
+      ABSL_ASSUME(reference_count != nullptr);
+      common_internal::StrongUnref(reference_count);
+    }
+  }
+
+  uintptr_t ptr_ = kNone;
+};
+
+inline bool operator!=(const Owner& lhs, const Owner& rhs) noexcept {
+  return !operator==(lhs, rhs);
+}
+
+namespace common_internal {
+
+inline const ReferenceCount* absl_nullable OwnerRelease(Owner owner) noexcept {
+  uintptr_t ptr = std::exchange(owner.ptr_, kMetadataOwnerNone);
+  if (Owner::IsReferenceCount(ptr)) {
+    return Owner::AsReferenceCount(ptr);
+  }
+  return nullptr;
+}
+
+}  // namespace common_internal
+
+template <>
+struct ArenaTraits<Owner> {
+  static bool trivially_destructible(const Owner& owner) {
+    return !Owner::IsReferenceCount(owner.ptr_);
+  }
+};
+
+// `Borrower` represents a reference to some borrowed data, where the data has
+// at least one owner. When using reference counting, `Borrower` does not
+// participate in incrementing/decrementing the reference count. Thus `Borrower`
+// will not keep the underlying data alive.
+class Borrower final {
+ public:
+  static Borrower None() noexcept { return Borrower(); }
+
+  static Borrower Allocator(Allocator<> allocator) noexcept {
+    auto* arena = allocator.arena();
+    return arena != nullptr ? Arena(arena) : None();
+  }
+
+  static Borrower Arena(google::protobuf::Arena* absl_nonnull arena
+                            ABSL_ATTRIBUTE_LIFETIME_BOUND) noexcept {
+    ABSL_DCHECK(arena != nullptr);
+    return Borrower(reinterpret_cast<uintptr_t>(arena) | Owner::kArenaBit);
+  }
+
+  static Borrower Arena(std::nullptr_t) = delete;
+
+  static Borrower ReferenceCount(
+      const ReferenceCount* absl_nonnull reference_count
+          ABSL_ATTRIBUTE_LIFETIME_BOUND) noexcept {
+    ABSL_DCHECK(reference_count != nullptr);
+    return Borrower(reinterpret_cast<uintptr_t>(reference_count) |
+                    Owner::kReferenceCountBit);
+  }
+
+  static Borrower ReferenceCount(std::nullptr_t) = delete;
+
+  Borrower() = default;
+  Borrower(const Borrower&) = default;
+  Borrower(Borrower&&) = default;
+  Borrower& operator=(const Borrower&) = default;
+  Borrower& operator=(Borrower&&) = default;
+
+  template <typename T>
+  // NOLINTNEXTLINE(google-explicit-constructor)
+  Borrower(const Owned<T>& owned ABSL_ATTRIBUTE_LIFETIME_BOUND) noexcept;
+
+  template <typename T>
+  // NOLINTNEXTLINE(google-explicit-constructor)
+  Borrower(Borrowed<T> borrowed) noexcept;
+
+  // NOLINTNEXTLINE(google-explicit-constructor)
+  Borrower(const Owner& owner ABSL_ATTRIBUTE_LIFETIME_BOUND) noexcept
+      : ptr_(owner.ptr_) {}
+
+  // NOLINTNEXTLINE(google-explicit-constructor)
+  Borrower& operator=(
+      const Owner& owner ABSL_ATTRIBUTE_LIFETIME_BOUND) noexcept {
+    ptr_ = owner.ptr_;
+    return *this;
+  }
+
+  Borrower& operator=(Owner&&) = delete;
+
+  template <typename T>
+  Borrower& operator=(
+      const Owned<T>& owned ABSL_ATTRIBUTE_LIFETIME_BOUND) noexcept;
+
+  template <typename T>
+  Borrower& operator=(Owned<T>&&) = delete;
+
+  template <typename T>
+  // NOLINTNEXTLINE(google-explicit-constructor)
+  Borrower& operator=(Borrowed<T> borrowed) noexcept;
+
+  explicit operator bool() const noexcept { return !Owner::IsNone(ptr_); }
+
+  google::protobuf::Arena* absl_nullable arena() const noexcept {
+    return (ptr_ & Owner::kBits) == Owner::kArenaBit
+               ? reinterpret_cast<google::protobuf::Arena*>(ptr_ & Owner::kPointerMask)
+               : nullptr;
+  }
+
+  void reset() noexcept { ptr_ = 0; }
+
+  // Tests whether two borrowers are borrowing the same data.
+  friend bool operator==(Borrower lhs, Borrower rhs) noexcept {
+    // A reference count and arena can never occupy the same memory address, so
+    // we can compare for equality without masking off the bits.
+    return lhs.ptr_ == rhs.ptr_;
+  }
+
+ private:
+  friend class Owner;
+  template <typename U>
+  friend struct Borrowable;
+  friend const common_internal::ReferenceCount* absl_nullable
+  common_internal::BorrowerRelease(Borrower borrower) noexcept;
+
+  constexpr explicit Borrower(uintptr_t ptr) noexcept : ptr_(ptr) {}
+
+  uintptr_t ptr_ = Owner::kNone;
+};
+
+inline bool operator!=(Borrower lhs, Borrower rhs) noexcept {
+  return !operator==(lhs, rhs);
+}
+
+inline bool operator==(Borrower lhs, const Owner& rhs) noexcept {
+  return operator==(lhs, Borrower(rhs));
+}
+
+inline bool operator==(const Owner& lhs, Borrower rhs) noexcept {
+  return operator==(Borrower(lhs), rhs);
+}
+
+inline bool operator!=(Borrower lhs, const Owner& rhs) noexcept {
+  return !operator==(lhs, rhs);
+}
+
+inline bool operator!=(const Owner& lhs, Borrower rhs) noexcept {
+  return !operator==(lhs, rhs);
+}
+
+inline Owner::Owner(Borrower borrower) noexcept
+    : ptr_(Owner::Own(borrower.ptr_)) {}
+
+namespace common_internal {
+
+inline const ReferenceCount* absl_nullable BorrowerRelease(
+    Borrower borrower) noexcept {
+  uintptr_t ptr = borrower.ptr_;
+  if (Owner::IsReferenceCount(ptr)) {
+    return Owner::AsReferenceCount(ptr);
+  }
+  return nullptr;
+}
+
+}  // namespace common_internal
+
+template <typename T, typename... Args>
+Unique<T> AllocateUnique(Allocator<> allocator, Args&&... args);
+
+// Wrap an already created `T` in `Unique`. Requires that `T` is not const,
+// otherwise `GetArena()` may return slightly unexpected results depending on if
+// it is the default value.
+template <typename T>
+std::enable_if_t<!std::is_const_v<T>, Unique<T>> WrapUnique(T* object);
+
+template <typename T>
+Unique<T> WrapUnique(T* object, Allocator<> allocator);
+
+// `Unique<T>` points to an object which was allocated using `Allocator<>` or
+// `Allocator<T>`. It has ownership over the object, and will perform any
+// destruction and deallocation required. `Unique` must not outlive the
+// underlying arena, if any. Unlike `Owned` and `Borrowed`, `Unique` supports
+// arena incompatible objects. It is very similar to `std::unique_ptr` when
+// using a custom deleter.
+//
+// IMPLEMENTATION NOTES:
+// When utilizing arenas, we optionally perform a risky optimization via
+// `AllocateUnique`. We do not use `Arena::Create`, instead we directly allocate
+// the bytes and construct it in place ourselves. This avoids registering the
+// destructor when required. Instead we register the destructor ourselves, if
+// required, during `Unique::release`. This allows us to avoid deferring
+// destruction of the object until the arena is destroyed, avoiding the cost
+// involved in doing so.
+template <typename T>
+class ABSL_ATTRIBUTE_TRIVIAL_ABI [[nodiscard]] Unique final {
+ public:
+  using element_type = T;
+
+  static_assert(!std::is_array_v<T>, "T must not be an array");
+  static_assert(!std::is_reference_v<T>, "T must not be a reference");
+  static_assert(!std::is_volatile_v<T>, "T must not be volatile qualified");
+
+  Unique() = default;
+  Unique(const Unique&) = delete;
+  Unique& operator=(const Unique&) = delete;
+
+  explicit Unique(T* ptr) noexcept
+      : Unique(ptr, common_internal::GetArena(ptr)) {}
+
+  // NOLINTNEXTLINE(google-explicit-constructor)
+  Unique(std::nullptr_t) noexcept : Unique() {}
+
+  Unique(Unique&& other) noexcept : Unique(other.ptr_, other.arena_) {
+    other.ptr_ = nullptr;
+  }
+
+  template <typename U,
+            typename = std::enable_if_t<
+                common_internal::kNotSameAndIsPointerConvertible<T, U>>>
+  // NOLINTNEXTLINE(google-explicit-constructor)
+  Unique(Unique<U>&& other) noexcept : Unique(other.ptr_, other.arena_) {
+    other.ptr_ = nullptr;
+  }
+
+  ~Unique() { Delete(); }
+
+  Unique& operator=(Unique&& other) noexcept {
+    if (ABSL_PREDICT_TRUE(this != &other)) {
+      Delete();
+      ptr_ = other.ptr_;
+      arena_ = other.arena_;
+      other.ptr_ = nullptr;
+    }
+    return *this;
+  }
+
+  template <typename U,
+            typename = std::enable_if_t<
+                common_internal::kNotSameAndIsPointerConvertible<T, U>>>
+  // NOLINTNEXTLINE(google-explicit-constructor)
+  Unique& operator=(U* other) noexcept {
+    reset(other);
+    return *this;
+  }
+
+  template <typename U,
+            typename = std::enable_if_t<
+                common_internal::kNotSameAndIsPointerConvertible<T, U>>>
+  // NOLINTNEXTLINE(google-explicit-constructor)
+  Unique& operator=(Unique<U>&& other) noexcept {
+    Delete();
+    ptr_ = other.ptr_;
+    arena_ = other.arena_;
+    other.ptr_ = nullptr;
+    return *this;
+  }
+
+  // NOLINTNEXTLINE(google-explicit-constructor)
+  Unique& operator=(std::nullptr_t) noexcept {
+    reset();
+    return *this;
+  }
+
+  T& operator*() const noexcept ABSL_ATTRIBUTE_LIFETIME_BOUND {
+    ABSL_DCHECK(static_cast<bool>(*this));
+    return *get();
+  }
+
+  T* absl_nonnull operator->() const noexcept ABSL_ATTRIBUTE_LIFETIME_BOUND {
+    ABSL_DCHECK(static_cast<bool>(*this));
+    return get();
+  }
+
+  // Relinquishes ownership of `T*`, returning it. If `T` was allocated and
+  // constructed using an arena, no further action is required. If `T` was
+  // allocated and constructed without an arena, the caller must eventually call
+  // `delete`.
+  ABSL_MUST_USE_RESULT T* release() noexcept {
+    PreRelease();
+    return std::exchange(ptr_, nullptr);
+  }
+
+  void reset() noexcept { reset(nullptr); }
+
+  void reset(T* ptr) noexcept {
+    Delete();
+    ptr_ = ptr;
+    arena_ = reinterpret_cast<uintptr_t>(common_internal::GetArena(ptr));
+  }
+
+  void reset(std::nullptr_t) noexcept {
+    Delete();
+    ptr_ = nullptr;
+    arena_ = 0;
+  }
+
+  explicit operator bool() const noexcept { return get() != nullptr; }
+
+  google::protobuf::Arena* absl_nullable arena() const noexcept {
+    return reinterpret_cast<google::protobuf::Arena*>(
+        arena_ & common_internal::kUniqueArenaPointerMask);
+  }
+
+  friend void swap(Unique& lhs, Unique& rhs) noexcept {
+    using std::swap;
+    swap(lhs.ptr_, rhs.ptr_);
+    swap(lhs.arena_, rhs.arena_);
+  }
+
+ private:
+  template <typename U>
+  friend class Unique;
+  template <typename U>
+  friend class Owned;
+  template <typename U, typename... Args>
+  friend Unique<U> AllocateUnique(Allocator<> allocator, Args&&... args);
+  template <typename U>
+  friend Unique<U> WrapUnique(U* object, Allocator<> allocator);
+  friend class ReferenceCountingMemoryManager;
+  friend class PoolingMemoryManager;
+  friend struct std::pointer_traits<Unique<T>>;
+  friend struct ArenaTraits<Unique<T>>;
+
+  static constexpr bool kNeedsArenaDestructor =
+      !std::is_trivially_destructible_v<T> &&
+      !google::protobuf::Arena::is_destructor_skippable<T>::value &&
+      !std::is_base_of_v<google::protobuf::MessageLite, T>;
+
+  Unique(T* ptr, uintptr_t arena) noexcept : ptr_(ptr), arena_(arena) {}
+
+  Unique(T* ptr, google::protobuf::Arena* arena, bool unowned = false) noexcept
+      : Unique(ptr,
+               reinterpret_cast<uintptr_t>(arena) |
+                   (unowned ? common_internal::kUniqueArenaUnownedBit : 0)) {
+    ABSL_ASSERT(!unowned || (unowned && arena != nullptr));
+  }
+
+  Unique(google::protobuf::Arena* arena, T* ptr, bool unowned = false) noexcept
+      : Unique(ptr, arena, unowned) {}
+
+  T* get() const noexcept { return ptr_; }
+
+  void Delete() const noexcept {
+    if (static_cast<bool>(*this)) {
+      if (arena_ != 0) {
+        if ((arena_ & common_internal::kUniqueArenaBits) ==
+            common_internal::kUniqueArenaUnownedBit) {
+          // We never registered the destructor, call it if necessary.
+          if constexpr (kNeedsArenaDestructor) {
+            std::destroy_at(ptr_);
+          }
+        }
+      } else {
+        delete ptr_;
+      }
+    }
+  }
+
+  void PreRelease() noexcept {
+    if constexpr (kNeedsArenaDestructor) {
+      if (static_cast<bool>(*this) &&
+          (arena_ & common_internal::kUniqueArenaBits) ==
+              common_internal::kUniqueArenaUnownedBit) {
+        // We never registered the destructor, call it if necessary.
+        arena()->OwnDestructor(const_cast<std::remove_const_t<T>*>(ptr_));
+        arena_ &= common_internal::kUniqueArenaPointerMask;
+      }
+    }
+  }
+
+  void Release(T** ptr, Owner* owner) noexcept {
+    if (ptr_ == nullptr) {
+      *ptr = nullptr;
+      return;
+    }
+    PreRelease();
+    *ptr = std::exchange(ptr_, nullptr);
+    if (arena_ == 0) {
+      owner->ptr_ = reinterpret_cast<uintptr_t>(
+                        common_internal::MakeDeletingReferenceCount(*ptr)) |
+                    common_internal::kMetadataOwnerReferenceCountBit;
+    } else {
+      owner->ptr_ = reinterpret_cast<uintptr_t>(arena()) |
+                    common_internal::kMetadataOwnerArenaBit;
+    }
+  }
+
+  T* ptr_ = nullptr;
+  // Potentially tagged pointer to `google::protobuf::Arena`. The tag is used to determine
+  // whether we still need to register the destructor with the `google::protobuf::Arena`.
+  uintptr_t arena_ = 0;
+};
+
+template <typename T>
+Unique(T*) -> Unique<T>;
+
+template <typename T, typename... Args>
+Unique<T> AllocateUnique(Allocator<> allocator, Args&&... args) {
+  using U = std::remove_cv_t<T>;
+  static_assert(!std::is_reference_v<U>, "T must not be a reference");
+  static_assert(!std::is_array_v<U>, "T must not be an array");
+
+  U* object;
+  google::protobuf::Arena* absl_nullable arena = allocator.arena();
+  bool unowned;
+  if constexpr (google::protobuf::Arena::is_arena_constructable<U>::value) {
+    object = google::protobuf::Arena::Create<U>(arena, std::forward<Args>(args)...);
+    // For arena-compatible proto types, let the Arena::Create handle
+    // registering the destructor call.
+    // Otherwise, Unique<T> retains a pointer to the owning arena so it may
+    // conditionally register T::~T depending on usage.
+    unowned = false;
+  } else {
+    void* p = allocator.allocate_bytes(sizeof(U), alignof(U));
+    CEL_INTERNAL_TRY {
+      if constexpr (ArenaTraits<>::constructible<U>()) {
+        object = ::new (p) U(arena, std::forward<Args>(args)...);
+      } else {
+        object = ::new (p) U(std::forward<Args>(args)...);
+      }
+    }
+    CEL_INTERNAL_CATCH_ANY {
+      allocator.deallocate_bytes(p, sizeof(U), alignof(U));
+      CEL_INTERNAL_RETHROW;
+    }
+    unowned =
+        arena != nullptr && !ArenaTraits<>::trivially_destructible(*object);
+  }
+  return Unique<T>(object, arena, unowned);
+}
+
+template <typename T>
+std::enable_if_t<!std::is_const_v<T>, Unique<T>> WrapUnique(T* object) {
+  return Unique<T>(object);
+}
+
+template <typename T>
+Unique<T> WrapUnique(T* object, Allocator<> allocator) {
+  return Unique<T>(object, allocator.arena());
+}
+
+template <typename T>
+inline bool operator==(const Unique<T>& lhs, std::nullptr_t) {
+  return !static_cast<bool>(lhs);
+}
+
+template <typename T>
+inline bool operator==(std::nullptr_t, const Unique<T>& rhs) {
+  return !static_cast<bool>(rhs);
+}
+
+template <typename T>
+inline bool operator!=(const Unique<T>& lhs, std::nullptr_t) {
+  return static_cast<bool>(lhs);
+}
+
+template <typename T>
+inline bool operator!=(std::nullptr_t, const Unique<T>& rhs) {
+  return static_cast<bool>(rhs);
+}
+
+}  // namespace cel
+
+namespace std {
+
+template <typename T>
+struct pointer_traits<cel::Unique<T>> {
+  using pointer = cel::Unique<T>;
+  using element_type = typename cel::Unique<T>::element_type;
+  using difference_type = ptrdiff_t;
+
+  template <typename U>
+  using rebind = cel::Unique<U>;
+
+  static element_type* to_address(const pointer& p) noexcept { return p.ptr_; }
+};
+
+}  // namespace std
+
+namespace cel {
+
+template <typename T>
+struct ArenaTraits<Unique<T>> {
+  static bool trivially_destructible(const Unique<T>& unique) {
+    return unique.arena_ != 0 &&
+           (unique.arena_ & common_internal::kUniqueArenaBits) == 0;
+  }
+};
+
+// `Owned<T>` points to an object which was allocated using `Allocator<>` or
+// `Allocator<T>`. It has co-ownership over the object. `T` must meet the named
+// requirement `ArenaConstructable`.
+template <typename T>
+class ABSL_ATTRIBUTE_TRIVIAL_ABI [[nodiscard]] Owned final {
+ public:
+  using element_type = T;
+
+  static_assert(!std::is_array_v<T>, "T must not be an array");
+  static_assert(!std::is_reference_v<T>, "T must not be a reference");
+  static_assert(!std::is_volatile_v<T>, "T must not be volatile qualified");
+  static_assert(!std::is_void_v<T>, "T must not be void");
+
+  Owned() = default;
+  Owned(const Owned&) = default;
+  Owned& operator=(const Owned&) = default;
+
+  Owned(Owned&& other) noexcept
+      : Owned(std::exchange(other.value_, nullptr), std::move(other.owner_)) {}
+
+  template <typename U,
+            typename = std::enable_if_t<
+                common_internal::kNotSameAndIsPointerConvertible<T, U>>>
+  // NOLINTNEXTLINE(google-explicit-constructor)
+  Owned(const Owned<U>& other) noexcept : Owned(other.value_, other.owner_) {}
+
+  template <typename U,
+            typename = std::enable_if_t<
+                common_internal::kNotSameAndIsPointerConvertible<T, U>>>
+  // NOLINTNEXTLINE(google-explicit-constructor)
+  Owned(Owned<U>&& other) noexcept
+      : Owned(std::exchange(other.value_, nullptr), std::move(other.owner_)) {}
+
+  template <typename U, typename = std::enable_if_t<
+                            common_internal::kIsPointerConvertible<T, U>>>
+  explicit Owned(Borrowed<U> other) noexcept;
+
+  template <typename U, typename = std::enable_if_t<
+                            common_internal::kIsPointerConvertible<T, U>>>
+  // NOLINTNEXTLINE(google-explicit-constructor)
+  Owned(Unique<U>&& other) : Owned() {
+    other.Release(&value_, &owner_);
+  }
+
+  Owned(Owner owner, T* value ABSL_ATTRIBUTE_LIFETIME_BOUND) noexcept
+      : Owned(value, std::move(owner)) {}
+
+  // NOLINTNEXTLINE(google-explicit-constructor)
+  Owned(std::nullptr_t) noexcept : Owned() {}
+
+  Owned& operator=(Owned&& other) noexcept {
+    if (ABSL_PREDICT_TRUE(this != &other)) {
+      value_ = std::exchange(other.value_, nullptr);
+      owner_ = std::move(other.owner_);
+    }
+    return *this;
+  }
+
+  template <typename U,
+            typename = std::enable_if_t<
+                common_internal::kNotSameAndIsPointerConvertible<T, U>>>
+  // NOLINTNEXTLINE(google-explicit-constructor)
+  Owned& operator=(const Owned<U>& other) noexcept {
+    value_ = other.value_;
+    owner_ = other.owner_;
+    return *this;
+  }
+
+  template <typename U,
+            typename = std::enable_if_t<
+                common_internal::kNotSameAndIsPointerConvertible<T, U>>>
+  // NOLINTNEXTLINE(google-explicit-constructor)
+  Owned& operator=(Owned<U>&& other) noexcept {
+    value_ = std::exchange(other.value_, nullptr);
+    owner_ = std::move(other.owner_);
+    return *this;
+  }
+
+  template <typename U, typename = std::enable_if_t<
+                            common_internal::kIsPointerConvertible<T, U>>>
+  // NOLINTNEXTLINE(google-explicit-constructor)
+  Owned& operator=(Borrowed<U> other) noexcept;
+
+  template <typename U, typename = std::enable_if_t<
+                            common_internal::kIsPointerConvertible<T, U>>>
+  // NOLINTNEXTLINE(google-explicit-constructor)
+  Owned& operator=(Unique<U>&& other) {
+    owner_.reset();
+    other.Release(&value_, &owner_);
+    return *this;
+  }
+
+  // NOLINTNEXTLINE(google-explicit-constructor)
+  Owned& operator=(std::nullptr_t) noexcept {
+    reset();
+    return *this;
+  }
+
+  T& operator*() const noexcept ABSL_ATTRIBUTE_LIFETIME_BOUND {
+    ABSL_DCHECK(static_cast<bool>(*this));
+    return *get();
+  }
+
+  T* absl_nonnull operator->() const noexcept ABSL_ATTRIBUTE_LIFETIME_BOUND {
+    ABSL_DCHECK(static_cast<bool>(*this));
+    return get();
+  }
+
+  void reset() noexcept {
+    value_ = nullptr;
+    owner_.reset();
+  }
+
+  google::protobuf::Arena* absl_nullable arena() const noexcept { return owner_.arena(); }
+
+  explicit operator bool() const noexcept { return get() != nullptr; }
+
+  friend void swap(Owned& lhs, Owned& rhs) noexcept {
+    using std::swap;
+    swap(lhs.value_, rhs.value_);
+    swap(lhs.owner_, rhs.owner_);
+  }
+
+ private:
+  friend class Owner;
+  friend class Borrower;
+  template <typename U>
+  friend class Owned;
+  template <typename U>
+  friend class Borrowed;
+  template <typename U>
+  friend struct Ownable;
+  template <typename U, typename... Args>
+  friend Owned<U> AllocateShared(Allocator<> allocator, Args&&... args);
+  template <typename U>
+  friend Owned<U> WrapShared(U* object, Allocator<> allocator);
+  template <typename U>
+  friend Owned<const U> common_internal::WrapEternal(const U* value);
+  friend struct std::pointer_traits<Owned<T>>;
+  friend struct ArenaTraits<Owned<T>>;
+
+  Owned(T* value, Owner owner) noexcept
+      : value_(value), owner_(std::move(owner)) {}
+
+  T* get() const noexcept { return value_; }
+
+  T* value_ = nullptr;
+  Owner owner_;
+};
+
+template <typename T>
+Owned(T*) -> Owned<T>;
+template <typename T>
+Owned(Unique<T>) -> Owned<T>;
+template <typename T>
+Owned(Owner, T*) -> Owned<T>;
+template <typename T>
+Owned(Borrowed<T>) -> Owned<T>;
+
+}  // namespace cel
+
+namespace std {
+
+template <typename T>
+struct pointer_traits<cel::Owned<T>> {
+  using pointer = cel::Owned<T>;
+  using element_type = typename cel::Owned<T>::element_type;
+  using difference_type = ptrdiff_t;
+
+  template <typename U>
+  using rebind = cel::Owned<U>;
+
+  static element_type* to_address(const pointer& p) noexcept {
+    return p.value_;
+  }
+};
+
+}  // namespace std
+
+namespace cel {
+
+template <typename T>
+struct ArenaTraits<Owned<T>> {
+  static bool trivially_destructible(const Owned<T>& owned) {
+    return ArenaTraits<>::trivially_destructible(owned.owner_);
+  }
+};
+
+template <typename T>
+Owner::Owner(const Owned<T>& owned) noexcept : Owner(owned.owner_) {}
+
+template <typename T>
+Owner::Owner(Owned<T>&& owned) noexcept : Owner(std::move(owned.owner_)) {
+  owned.value_ = nullptr;
+}
+
+template <typename T>
+Owner& Owner::operator=(const Owned<T>& owned) noexcept {
+  *this = owned.owner_;
+  return *this;
+}
+
+template <typename T>
+Owner& Owner::operator=(Owned<T>&& owned) noexcept {
+  *this = std::move(owned.owner_);
+  owned.value_ = nullptr;
+  return *this;
+}
+
+template <typename T>
+bool operator==(const Owned<T>& lhs, std::nullptr_t) noexcept {
+  return !static_cast<bool>(lhs);
+}
+
+template <typename T>
+bool operator==(std::nullptr_t, const Owned<T>& rhs) noexcept {
+  return rhs == nullptr;
+}
+
+template <typename T>
+bool operator!=(const Owned<T>& lhs, std::nullptr_t) noexcept {
+  return !operator==(lhs, nullptr);
+}
+
+template <typename T>
+bool operator!=(std::nullptr_t, const Owned<T>& rhs) noexcept {
+  return !operator==(nullptr, rhs);
+}
+
+template <typename T, typename... Args>
+Owned<T> AllocateShared(Allocator<> allocator, Args&&... args) {
+  using U = std::remove_cv_t<T>;
+  static_assert(!std::is_reference_v<U>, "T must not be a reference");
+  static_assert(!std::is_array_v<U>, "T must not be an array");
+
+  U* object;
+  Owner owner;
+  if (google::protobuf::Arena* absl_nullable arena = allocator.arena();
+      arena != nullptr) {
+    object = ArenaAllocator(arena).template new_object<U>(
+        std::forward<Args>(args)...);
+    owner.ptr_ = reinterpret_cast<uintptr_t>(arena) |
+                 common_internal::kMetadataOwnerArenaBit;
+  } else {
+    const common_internal::ReferenceCount* refcount;
+    std::tie(object, refcount) = common_internal::MakeEmplacedReferenceCount<U>(
+        std::forward<Args>(args)...);
+    owner.ptr_ = reinterpret_cast<uintptr_t>(refcount) |
+                 common_internal::kMetadataOwnerReferenceCountBit;
+  }
+  return Owned<U>(object, std::move(owner));
+}
+
+template <typename T>
+Owned<T> WrapShared(T* object, Allocator<> allocator) {
+  Owner owner;
+  if (object == nullptr) {
+  } else if (allocator.arena() != nullptr) {
+    owner.ptr_ = reinterpret_cast<uintptr_t>(
+                     static_cast<google::protobuf::Arena*>(allocator.arena())) |
+                 common_internal::kMetadataOwnerArenaBit;
+  } else {
+    owner.ptr_ = reinterpret_cast<uintptr_t>(
+                     common_internal::MakeDeletingReferenceCount(object)) |
+                 common_internal::kMetadataOwnerReferenceCountBit;
+  }
+  return Owned<T>(object, std::move(owner));
+}
+
+template <typename T>
+std::enable_if_t<!std::is_const_v<T>, Owned<T>> WrapShared(T* object) {
+  return WrapShared(object, object->GetArena());
+}
+
+namespace common_internal {
+
+template <typename T>
+Owned<const T> WrapEternal(const T* value) {
+  return Owned<const T>(value, Owner::None());
+}
+
+}  // namespace common_internal
+
+// `Borrowed<T>` points to an object which was allocated using `Allocator<>` or
+// `Allocator<T>`. It has no ownership over the object, and is only valid so
+// long as one or more owners of the object exist. `T` must meet the named
+// requirement `ArenaConstructable`.
+template <typename T>
+class Borrowed final {
+ public:
+  using element_type = T;
+
+  static_assert(!std::is_array_v<T>, "T must not be an array");
+  static_assert(!std::is_reference_v<T>, "T must not be a reference");
+  static_assert(!std::is_volatile_v<T>, "T must not be volatile qualified");
+  static_assert(!std::is_void_v<T>, "T must not be void");
+
+  Borrowed() = default;
+  Borrowed(const Borrowed&) = default;
+  Borrowed(Borrowed&&) = default;
+  Borrowed& operator=(const Borrowed&) = default;
+  Borrowed& operator=(Borrowed&&) = default;
+
+  template <typename U,
+            typename = std::enable_if_t<
+                common_internal::kNotSameAndIsPointerConvertible<T, U>>>
+  // NOLINTNEXTLINE(google-explicit-constructor)
+  Borrowed(const Borrowed<U>& other) noexcept
+      : Borrowed(other.value_, other.borrower_) {}
+
+  template <typename U,
+            typename = std::enable_if_t<
+                common_internal::kNotSameAndIsPointerConvertible<T, U>>>
+  // NOLINTNEXTLINE(google-explicit-constructor)
+  Borrowed(Borrowed<U>&& other) noexcept
+      : Borrowed(other.value_, other.borrower_) {}
+
+  template <typename U, typename = std::enable_if_t<
+                            common_internal::kIsPointerConvertible<T, U>>>
+  // NOLINTNEXTLINE(google-explicit-constructor)
+  Borrowed(const Owned<U>& other ABSL_ATTRIBUTE_LIFETIME_BOUND) noexcept
+      : Borrowed(other.value_, other.owner_) {}
+
+  Borrowed(Borrower borrower, T* ptr) noexcept : Borrowed(ptr, borrower) {}
+
+  // NOLINTNEXTLINE(google-explicit-constructor)
+  Borrowed(std::nullptr_t) noexcept : Borrowed() {}
+
+  template <typename U,
+            typename = std::enable_if_t<
+                common_internal::kNotSameAndIsPointerConvertible<T, U>>>
+  // NOLINTNEXTLINE(google-explicit-constructor)
+  Borrowed& operator=(const Borrowed<U>& other) noexcept {
+    value_ = other.value_;
+    borrower_ = other.borrower_;
+    return *this;
+  }
+
+  template <typename U,
+            typename = std::enable_if_t<
+                common_internal::kNotSameAndIsPointerConvertible<T, U>>>
+  // NOLINTNEXTLINE(google-explicit-constructor)
+  Borrowed& operator=(Borrowed<U>&& other) noexcept {
+    value_ = other.value_;
+    borrower_ = other.borrower_;
+    return *this;
+  }
+
+  template <typename U, typename = std::enable_if_t<
+                            common_internal::kIsPointerConvertible<T, U>>>
+  // NOLINTNEXTLINE(google-explicit-constructor)
+  Borrowed& operator=(
+      const Owned<U>& other ABSL_ATTRIBUTE_LIFETIME_BOUND) noexcept {
+    value_ = other.value_;
+    borrower_ = other.borrower_;
+    return *this;
+  }
+
+  template <typename U, typename = std::enable_if_t<
+                            common_internal::kIsPointerConvertible<T, U>>>
+  // NOLINTNEXTLINE(google-explicit-constructor)
+  Borrowed& operator=(Owned<U>&&) = delete;
+
+  // NOLINTNEXTLINE(google-explicit-constructor)
+  Borrowed& operator=(std::nullptr_t) noexcept {
+    reset();
+    return *this;
+  }
+
+  T& operator*() const noexcept {
+    ABSL_DCHECK(static_cast<bool>(*this));
+    return *get();
+  }
+
+  T* absl_nonnull operator->() const noexcept {
+    ABSL_DCHECK(static_cast<bool>(*this));
+    return get();
+  }
+
+  void reset() noexcept {
+    value_ = nullptr;
+    borrower_.reset();
+  }
+
+  google::protobuf::Arena* absl_nullable arena() const noexcept {
+    return borrower_.arena();
+  }
+
+  explicit operator bool() const noexcept { return get() != nullptr; }
+
+ private:
+  friend class Owner;
+  friend class Borrower;
+  template <typename U>
+  friend class Owned;
+  template <typename U>
+  friend class Borrowed;
+  template <typename U>
+  friend struct Borrowable;
+  friend struct std::pointer_traits<Borrowed<T>>;
+
+  constexpr Borrowed(T* value, Borrower borrower) noexcept
+      : value_(value), borrower_(borrower) {}
+
+  T* get() const noexcept { return value_; }
+
+  T* value_ = nullptr;
+  Borrower borrower_;
+};
+
+template <typename T>
+Borrowed(T*) -> Borrowed<T>;
+template <typename T>
+Borrowed(Borrower, T*) -> Borrowed<T>;
+template <typename T>
+Borrowed(Owned<T>) -> Borrowed<T>;
+
+}  // namespace cel
+
+namespace std {
+
+template <typename T>
+struct pointer_traits<cel::Borrowed<T>> {
+  using pointer = cel::Borrowed<T>;
+  using element_type = typename cel::Borrowed<T>::element_type;
+  using difference_type = ptrdiff_t;
+
+  template <typename U>
+  using rebind = cel::Borrowed<U>;
+
+  static element_type* to_address(pointer p) noexcept { return p.value_; }
+};
+
+}  // namespace std
+
+namespace cel {
+
+template <typename T>
+Owner::Owner(Borrowed<T> borrowed) noexcept : Owner(borrowed.borrower_) {}
+
+template <typename T>
+Borrower::Borrower(const Owned<T>& owned ABSL_ATTRIBUTE_LIFETIME_BOUND) noexcept
+    : Borrower(owned.owner_) {}
+
+template <typename T>
+Borrower::Borrower(Borrowed<T> borrowed) noexcept
+    : Borrower(borrowed.borrower_) {}
+
+template <typename T>
+Borrower& Borrower::operator=(
+    const Owned<T>& owned ABSL_ATTRIBUTE_LIFETIME_BOUND) noexcept {
+  *this = owned.owner_;
+  return *this;
+}
+
+template <typename T>
+Borrower& Borrower::operator=(Borrowed<T> borrowed) noexcept {
+  *this = borrowed.borrower_;
+  return *this;
+}
+
+template <typename T>
+bool operator==(Borrowed<T> lhs, std::nullptr_t) noexcept {
+  return !static_cast<bool>(lhs);
+}
+
+template <typename T>
+bool operator==(std::nullptr_t, Borrowed<T> rhs) noexcept {
+  return rhs == nullptr;
+}
+
+template <typename T>
+bool operator!=(Borrowed<T> lhs, std::nullptr_t) noexcept {
+  return !operator==(lhs, nullptr);
+}
+
+template <typename T>
+bool operator!=(std::nullptr_t, Borrowed<T> rhs) noexcept {
+  return !operator==(nullptr, rhs);
+}
+
+template <typename T>
+template <typename U, typename>
+Owned<T>::Owned(Borrowed<U> other) noexcept
+    : Owned(other.value_, Owner(other.borrower_)) {}
+
+template <typename T>
+template <typename U, typename>
+Owned<T>& Owned<T>::operator=(Borrowed<U> other) noexcept {
+  value_ = other.value_;
+  owner_ = Owner(other.borrower_);
+  return *this;
+}
+
+// `Ownable<T>` is a mixin for enabling the ability to get `Owned` that refer to
+// this.
+template <typename T>
+struct Ownable {
+ protected:
+  Owned<const T> Own() const noexcept {
+    static_assert(std::is_base_of_v<Data, T>, "T must be derived from Data");
+    const T* const that = static_cast<const T*>(this);
+    return Owned<const T>(
+        Owner(Owner::Own(static_cast<const Data*>(that)->owner_)), that);
+  }
+
+  Owned<T> Own() noexcept {
+    static_assert(std::is_base_of_v<Data, T>, "T must be derived from Data");
+    T* const that = static_cast<T*>(this);
+    return Owned<T>(Owner(Owner::Own(static_cast<Data*>(that)->owner_)), that);
+  }
+
+  ABSL_DEPRECATED("Use Own")
+  Owned<const T> shared_from_this() const noexcept { return Own(); }
+
+  ABSL_DEPRECATED("Use Own")
+  Owned<T> shared_from_this() noexcept { return Own(); }
+};
+
+// `Borrowable<T>` is a mixin for enabling the ability to get `Borrowed` that
+// refer to this.
+template <typename T>
+struct Borrowable {
+ protected:
+  Borrowed<const T> Borrow() const noexcept {
+    static_assert(std::is_base_of_v<Data, T>, "T must be derived from Data");
+    const T* const that = static_cast<const T*>(this);
+    return Borrowed<const T>(Borrower(static_cast<const Data*>(that)->owner_),
+                             that);
+  }
+
+  Borrowed<T> Borrow() noexcept {
+    static_assert(std::is_base_of_v<Data, T>, "T must be derived from Data");
+    T* const that = static_cast<T*>(this);
+    return Borrowed<T>(Borrower(static_cast<Data*>(that)->owner_), that);
+  }
+};
+
 // `ReferenceCountingMemoryManager` is a `MemoryManager` which employs automatic
 // memory management through reference counting.
 class ReferenceCountingMemoryManager final {
diff --git a/common/memory_test.cc b/common/memory_test.cc
new file mode 100644
index 000000000..7f3e7a82a
--- /dev/null
+++ b/common/memory_test.cc
@@ -0,0 +1,466 @@
+// Copyright 2023 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     https://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// This header contains primitives for reference counting, roughly equivalent to
+// the primitives used to implement `std::shared_ptr`. These primitives should
+// not be used directly in most cases, instead `cel::ManagedMemory` should be
+// used instead.
+
+#include "common/memory.h"
+
+#include <utility>
+
+#include "google/protobuf/struct.pb.h"
+#include "absl/base/nullability.h"
+#include "common/allocator.h"
+#include "common/data.h"
+#include "common/internal/reference_count.h"
+#include "internal/testing.h"
+#include "google/protobuf/arena.h"
+
+#ifdef ABSL_HAVE_EXCEPTIONS
+#include <stdexcept>
+#endif
+
+namespace cel {
+namespace {
+
+using ::testing::IsFalse;
+using ::testing::IsNull;
+using ::testing::IsTrue;
+using ::testing::TestParamInfo;
+using ::testing::TestWithParam;
+
+TEST(Owner, None) {
+  EXPECT_THAT(Owner::None(), IsFalse());
+  EXPECT_THAT(Owner::None().arena(), IsNull());
+}
+
+TEST(Owner, Allocator) {
+  google::protobuf::Arena arena;
+  EXPECT_THAT(Owner::Allocator(NewDeleteAllocator<>{}), IsFalse());
+  EXPECT_THAT(Owner::Allocator(ArenaAllocator<>{&arena}), IsTrue());
+}
+
+TEST(Owner, Arena) {
+  google::protobuf::Arena arena;
+  EXPECT_THAT(Owner::Arena(&arena), IsTrue());
+  EXPECT_EQ(Owner::Arena(&arena).arena(), &arena);
+}
+
+TEST(Owner, ReferenceCount) {
+  auto* refcount = new common_internal::ReferenceCounted();
+  EXPECT_THAT(Owner::ReferenceCount(refcount), IsTrue());
+  EXPECT_THAT(Owner::ReferenceCount(refcount).arena(), IsNull());
+  common_internal::StrongUnref(refcount);
+}
+
+TEST(Owner, Equality) {
+  google::protobuf::Arena arena1;
+  google::protobuf::Arena arena2;
+  EXPECT_EQ(Owner::None(), Owner::None());
+  EXPECT_EQ(Owner::Allocator(NewDeleteAllocator<>{}), Owner::None());
+  EXPECT_EQ(Owner::Arena(&arena1), Owner::Arena(&arena1));
+  EXPECT_NE(Owner::Arena(&arena1), Owner::None());
+  EXPECT_NE(Owner::None(), Owner::Arena(&arena1));
+  EXPECT_NE(Owner::Arena(&arena1), Owner::Arena(&arena2));
+  EXPECT_EQ(Owner::Allocator(ArenaAllocator<>{&arena1}), Owner::Arena(&arena1));
+}
+
+TEST(Borrower, None) {
+  EXPECT_THAT(Borrower::None(), IsFalse());
+  EXPECT_THAT(Borrower::None().arena(), IsNull());
+}
+
+TEST(Borrower, Allocator) {
+  google::protobuf::Arena arena;
+  EXPECT_THAT(Borrower::Allocator(NewDeleteAllocator<>{}), IsFalse());
+  EXPECT_THAT(Borrower::Allocator(ArenaAllocator<>{&arena}), IsTrue());
+}
+
+TEST(Borrower, Arena) {
+  google::protobuf::Arena arena;
+  EXPECT_THAT(Borrower::Arena(&arena), IsTrue());
+  EXPECT_EQ(Borrower::Arena(&arena).arena(), &arena);
+}
+
+TEST(Borrower, ReferenceCount) {
+  auto* refcount = new common_internal::ReferenceCounted();
+  EXPECT_THAT(Borrower::ReferenceCount(refcount), IsTrue());
+  EXPECT_THAT(Borrower::ReferenceCount(refcount).arena(), IsNull());
+  common_internal::StrongUnref(refcount);
+}
+
+TEST(Borrower, Equality) {
+  google::protobuf::Arena arena1;
+  google::protobuf::Arena arena2;
+  EXPECT_EQ(Borrower::None(), Borrower::None());
+  EXPECT_EQ(Borrower::Allocator(NewDeleteAllocator<>{}), Borrower::None());
+  EXPECT_EQ(Borrower::Arena(&arena1), Borrower::Arena(&arena1));
+  EXPECT_NE(Borrower::Arena(&arena1), Borrower::None());
+  EXPECT_NE(Borrower::None(), Borrower::Arena(&arena1));
+  EXPECT_NE(Borrower::Arena(&arena1), Borrower::Arena(&arena2));
+  EXPECT_EQ(Borrower::Allocator(ArenaAllocator<>{&arena1}),
+            Borrower::Arena(&arena1));
+}
+
+TEST(OwnerBorrower, CopyConstruct) {
+  auto* refcount = new common_internal::ReferenceCounted();
+  Owner owner1 = Owner::ReferenceCount(refcount);
+  common_internal::StrongUnref(refcount);
+  Owner owner2(owner1);
+  Borrower borrower(owner1);
+  EXPECT_EQ(owner1, owner2);
+  EXPECT_EQ(owner1, borrower);
+  EXPECT_EQ(borrower, owner1);
+}
+
+TEST(OwnerBorrower, MoveConstruct) {
+  auto* refcount = new common_internal::ReferenceCounted();
+  Owner owner1 = Owner::ReferenceCount(refcount);
+  common_internal::StrongUnref(refcount);
+  Owner owner2(std::move(owner1));
+  Borrower borrower(owner2);
+  EXPECT_EQ(owner2, borrower);
+  EXPECT_EQ(borrower, owner2);
+}
+
+TEST(OwnerBorrower, CopyAssign) {
+  auto* refcount = new common_internal::ReferenceCounted();
+  Owner owner1 = Owner::ReferenceCount(refcount);
+  common_internal::StrongUnref(refcount);
+  Owner owner2;
+  owner2 = owner1;
+  Borrower borrower(owner1);
+  EXPECT_EQ(owner1, owner2);
+  EXPECT_EQ(owner1, borrower);
+  EXPECT_EQ(borrower, owner1);
+}
+
+TEST(OwnerBorrower, MoveAssign) {
+  auto* refcount = new common_internal::ReferenceCounted();
+  Owner owner1 = Owner::ReferenceCount(refcount);
+  common_internal::StrongUnref(refcount);
+  Owner owner2;
+  owner2 = std::move(owner1);
+  Borrower borrower(owner2);
+  EXPECT_EQ(owner2, borrower);
+  EXPECT_EQ(borrower, owner2);
+}
+
+TEST(Unique, ToAddress) {
+  Unique<bool> unique;
+  EXPECT_EQ(cel::to_address(unique), nullptr);
+  unique = AllocateUnique<bool>(NewDeleteAllocator<>{});
+  EXPECT_EQ(cel::to_address(unique), unique.operator->());
+}
+
+class OwnedTest : public TestWithParam<AllocatorKind> {
+ public:
+  Allocator<> GetAllocator() {
+    switch (GetParam()) {
+      case AllocatorKind::kArena:
+        return ArenaAllocator<>{&arena_};
+      case AllocatorKind::kNewDelete:
+        return NewDeleteAllocator<>{};
+    }
+  }
+
+ private:
+  google::protobuf::Arena arena_;
+};
+
+TEST_P(OwnedTest, Default) {
+  Owned<Data> owned;
+  EXPECT_FALSE(owned);
+  EXPECT_EQ(cel::to_address(owned), nullptr);
+  EXPECT_FALSE(owned != nullptr);
+  EXPECT_FALSE(nullptr != owned);
+}
+
+class TestData final : public Data {
+ public:
+  using InternalArenaConstructable_ = void;
+  using DestructorSkippable_ = void;
+
+  TestData() noexcept : Data() {}
+
+  explicit TestData(google::protobuf::Arena* absl_nullable arena) noexcept
+      : Data(arena) {}
+};
+
+TEST_P(OwnedTest, AllocateSharedData) {
+  auto owned = AllocateShared<TestData>(GetAllocator());
+  EXPECT_EQ(owned->GetArena(), GetAllocator().arena());
+  EXPECT_EQ(Owner(owned).arena(), GetAllocator().arena());
+  EXPECT_EQ(Borrower(owned).arena(), GetAllocator().arena());
+}
+
+TEST_P(OwnedTest, AllocateSharedMessageLite) {
+  auto owned = AllocateShared<google::protobuf::Value>(GetAllocator());
+  EXPECT_EQ(owned->GetArena(), GetAllocator().arena());
+  EXPECT_EQ(Owner(owned).arena(), GetAllocator().arena());
+  EXPECT_EQ(Borrower(owned).arena(), GetAllocator().arena());
+}
+
+TEST_P(OwnedTest, WrapSharedData) {
+  auto owned =
+      WrapShared(google::protobuf::Arena::Create<TestData>(GetAllocator().arena()));
+  EXPECT_EQ(owned->GetArena(), GetAllocator().arena());
+  EXPECT_EQ(Owner(owned).arena(), GetAllocator().arena());
+  EXPECT_EQ(Borrower(owned).arena(), GetAllocator().arena());
+}
+
+TEST_P(OwnedTest, WrapSharedMessageLite) {
+  auto owned = WrapShared(
+      google::protobuf::Arena::Create<google::protobuf::Value>(GetAllocator().arena()));
+  EXPECT_EQ(owned->GetArena(), GetAllocator().arena());
+  EXPECT_EQ(Owner(owned).arena(), GetAllocator().arena());
+  EXPECT_EQ(Borrower(owned).arena(), GetAllocator().arena());
+}
+
+TEST_P(OwnedTest, SharedFromUniqueData) {
+  auto owned = Owned(AllocateUnique<TestData>(GetAllocator()));
+  EXPECT_EQ(owned->GetArena(), GetAllocator().arena());
+  EXPECT_EQ(Owner(owned).arena(), GetAllocator().arena());
+  EXPECT_EQ(Borrower(owned).arena(), GetAllocator().arena());
+}
+
+TEST_P(OwnedTest, SharedFromUniqueMessageLite) {
+  auto owned = Owned(AllocateUnique<google::protobuf::Value>(GetAllocator()));
+  EXPECT_EQ(owned->GetArena(), GetAllocator().arena());
+  EXPECT_EQ(Owner(owned).arena(), GetAllocator().arena());
+  EXPECT_EQ(Borrower(owned).arena(), GetAllocator().arena());
+}
+
+TEST_P(OwnedTest, CopyConstruct) {
+  auto owned = Owned(AllocateUnique<TestData>(GetAllocator()));
+  EXPECT_EQ(owned->GetArena(), GetAllocator().arena());
+  Owned<TestData> copied_owned(owned);
+  EXPECT_EQ(copied_owned->GetArena(), GetAllocator().arena());
+}
+
+TEST_P(OwnedTest, MoveConstruct) {
+  auto owned = Owned(AllocateUnique<TestData>(GetAllocator()));
+  EXPECT_EQ(owned->GetArena(), GetAllocator().arena());
+  Owned<TestData> moved_owned(std::move(owned));
+  EXPECT_EQ(moved_owned->GetArena(), GetAllocator().arena());
+}
+
+TEST_P(OwnedTest, CopyConstructOther) {
+  auto owned = Owned(AllocateUnique<TestData>(GetAllocator()));
+  EXPECT_EQ(owned->GetArena(), GetAllocator().arena());
+  Owned<Data> copied_owned(owned);
+  EXPECT_EQ(copied_owned->GetArena(), GetAllocator().arena());
+}
+
+TEST_P(OwnedTest, MoveConstructOther) {
+  auto owned = Owned(AllocateUnique<TestData>(GetAllocator()));
+  EXPECT_EQ(owned->GetArena(), GetAllocator().arena());
+  Owned<Data> moved_owned(std::move(owned));
+  EXPECT_EQ(moved_owned->GetArena(), GetAllocator().arena());
+}
+
+TEST_P(OwnedTest, ConstructBorrowed) {
+  auto owned = Owned(AllocateUnique<TestData>(GetAllocator()));
+  EXPECT_EQ(owned->GetArena(), GetAllocator().arena());
+  Owned<TestData> borrowed_owned(Borrowed<TestData>{owned});
+  EXPECT_EQ(borrowed_owned->GetArena(), GetAllocator().arena());
+}
+
+TEST_P(OwnedTest, ConstructOwner) {
+  auto owned = Owned(AllocateUnique<TestData>(GetAllocator()));
+  EXPECT_EQ(owned->GetArena(), GetAllocator().arena());
+  Owned<TestData> owner_owned(Owner(owned), cel::to_address(owned));
+  EXPECT_EQ(owner_owned->GetArena(), GetAllocator().arena());
+}
+
+TEST_P(OwnedTest, ConstructNullPtr) {
+  Owned<Data> owned(nullptr);
+  EXPECT_EQ(owned, nullptr);
+}
+
+TEST_P(OwnedTest, CopyAssign) {
+  auto owned = Owned(AllocateUnique<TestData>(GetAllocator()));
+  EXPECT_EQ(owned->GetArena(), GetAllocator().arena());
+  Owned<TestData> copied_owned;
+  copied_owned = owned;
+  EXPECT_EQ(copied_owned->GetArena(), GetAllocator().arena());
+}
+
+TEST_P(OwnedTest, MoveAssign) {
+  auto owned = Owned(AllocateUnique<TestData>(GetAllocator()));
+  EXPECT_EQ(owned->GetArena(), GetAllocator().arena());
+  Owned<TestData> moved_owned;
+  moved_owned = std::move(owned);
+  EXPECT_EQ(moved_owned->GetArena(), GetAllocator().arena());
+}
+
+TEST_P(OwnedTest, CopyAssignOther) {
+  auto owned = Owned(AllocateUnique<TestData>(GetAllocator()));
+  EXPECT_EQ(owned->GetArena(), GetAllocator().arena());
+  Owned<Data> copied_owned;
+  copied_owned = owned;
+  EXPECT_EQ(copied_owned->GetArena(), GetAllocator().arena());
+}
+
+TEST_P(OwnedTest, MoveAssignOther) {
+  auto owned = Owned(AllocateUnique<TestData>(GetAllocator()));
+  EXPECT_EQ(owned->GetArena(), GetAllocator().arena());
+  Owned<Data> moved_owned;
+  moved_owned = std::move(owned);
+  EXPECT_EQ(moved_owned->GetArena(), GetAllocator().arena());
+}
+
+TEST_P(OwnedTest, AssignBorrowed) {
+  auto owned = Owned(AllocateUnique<TestData>(GetAllocator()));
+  EXPECT_EQ(owned->GetArena(), GetAllocator().arena());
+  Owned<TestData> borrowed_owned;
+  borrowed_owned = Borrowed<TestData>{owned};
+  EXPECT_EQ(borrowed_owned->GetArena(), GetAllocator().arena());
+}
+
+TEST_P(OwnedTest, AssignUnique) {
+  Owned<TestData> owned;
+  owned = AllocateUnique<TestData>(GetAllocator());
+  EXPECT_EQ(owned->GetArena(), GetAllocator().arena());
+}
+
+TEST_P(OwnedTest, AssignNullPtr) {
+  auto owned = Owned(AllocateUnique<TestData>(GetAllocator()));
+  EXPECT_EQ(owned->GetArena(), GetAllocator().arena());
+  EXPECT_TRUE(owned);
+  owned = nullptr;
+  EXPECT_FALSE(owned);
+}
+
+INSTANTIATE_TEST_SUITE_P(OwnedTest, OwnedTest,
+                         ::testing::Values(AllocatorKind::kArena,
+                                           AllocatorKind::kNewDelete));
+
+class BorrowedTest : public TestWithParam<AllocatorKind> {
+ public:
+  Allocator<> GetAllocator() {
+    switch (GetParam()) {
+      case AllocatorKind::kArena:
+        return ArenaAllocator<>{&arena_};
+      case AllocatorKind::kNewDelete:
+        return NewDeleteAllocator<>{};
+    }
+  }
+
+ private:
+  google::protobuf::Arena arena_;
+};
+
+TEST_P(BorrowedTest, Default) {
+  Borrowed<Data> borrowed;
+  EXPECT_FALSE(borrowed);
+  EXPECT_EQ(cel::to_address(borrowed), nullptr);
+  EXPECT_FALSE(borrowed != nullptr);
+  EXPECT_FALSE(nullptr != borrowed);
+}
+
+TEST_P(BorrowedTest, CopyConstruct) {
+  auto owned = Owned(AllocateUnique<TestData>(GetAllocator()));
+  auto borrowed = Borrowed(owned);
+  EXPECT_EQ(borrowed->GetArena(), GetAllocator().arena());
+  Borrowed<TestData> copied_borrowed(borrowed);
+  EXPECT_EQ(copied_borrowed->GetArena(), GetAllocator().arena());
+}
+
+TEST_P(BorrowedTest, MoveConstruct) {
+  auto owned = Owned(AllocateUnique<TestData>(GetAllocator()));
+  auto borrowed = Borrowed(owned);
+  EXPECT_EQ(borrowed->GetArena(), GetAllocator().arena());
+  Borrowed<TestData> moved_borrowed(std::move(borrowed));
+  EXPECT_EQ(moved_borrowed->GetArena(), GetAllocator().arena());
+}
+
+TEST_P(BorrowedTest, CopyConstructOther) {
+  auto owned = Owned(AllocateUnique<TestData>(GetAllocator()));
+  auto borrowed = Borrowed(owned);
+  EXPECT_EQ(borrowed->GetArena(), GetAllocator().arena());
+  Borrowed<Data> copied_borrowed(borrowed);
+  EXPECT_EQ(copied_borrowed->GetArena(), GetAllocator().arena());
+}
+
+TEST_P(BorrowedTest, MoveConstructOther) {
+  auto owned = Owned(AllocateUnique<TestData>(GetAllocator()));
+  auto borrowed = Borrowed(owned);
+  EXPECT_EQ(borrowed->GetArena(), GetAllocator().arena());
+  Borrowed<Data> moved_borrowed(std::move(borrowed));
+  EXPECT_EQ(moved_borrowed->GetArena(), GetAllocator().arena());
+}
+
+TEST_P(BorrowedTest, ConstructNullPtr) {
+  Borrowed<TestData> borrowed(nullptr);
+  EXPECT_FALSE(borrowed);
+}
+
+TEST_P(BorrowedTest, CopyAssign) {
+  auto owned = Owned(AllocateUnique<TestData>(GetAllocator()));
+  auto borrowed = Borrowed(owned);
+  EXPECT_EQ(borrowed->GetArena(), GetAllocator().arena());
+  Borrowed<TestData> copied_borrowed;
+  copied_borrowed = borrowed;
+  EXPECT_EQ(copied_borrowed->GetArena(), GetAllocator().arena());
+}
+
+TEST_P(BorrowedTest, MoveAssign) {
+  auto owned = Owned(AllocateUnique<TestData>(GetAllocator()));
+  auto borrowed = Borrowed(owned);
+  EXPECT_EQ(borrowed->GetArena(), GetAllocator().arena());
+  Borrowed<TestData> moved_borrowed;
+  moved_borrowed = std::move(borrowed);
+  EXPECT_EQ(moved_borrowed->GetArena(), GetAllocator().arena());
+}
+
+TEST_P(BorrowedTest, CopyAssignOther) {
+  auto owned = Owned(AllocateUnique<TestData>(GetAllocator()));
+  auto borrowed = Borrowed(owned);
+  EXPECT_EQ(borrowed->GetArena(), GetAllocator().arena());
+  Borrowed<Data> copied_borrowed;
+  copied_borrowed = borrowed;
+  EXPECT_EQ(copied_borrowed->GetArena(), GetAllocator().arena());
+}
+
+TEST_P(BorrowedTest, MoveAssignOther) {
+  auto owned = Owned(AllocateUnique<TestData>(GetAllocator()));
+  auto borrowed = Borrowed(owned);
+  EXPECT_EQ(borrowed->GetArena(), GetAllocator().arena());
+  Borrowed<Data> moved_borrowed;
+  moved_borrowed = std::move(borrowed);
+  EXPECT_EQ(moved_borrowed->GetArena(), GetAllocator().arena());
+}
+
+TEST_P(BorrowedTest, AssignOwned) {
+  auto owned = Owned(AllocateUnique<TestData>(GetAllocator()));
+  EXPECT_EQ(owned->GetArena(), GetAllocator().arena());
+  Borrowed<Data> borrowed = owned;
+  EXPECT_EQ(borrowed->GetArena(), GetAllocator().arena());
+}
+
+TEST_P(BorrowedTest, AssignNullPtr) {
+  Borrowed<TestData> borrowed;
+  borrowed = nullptr;
+  EXPECT_FALSE(borrowed);
+}
+
+INSTANTIATE_TEST_SUITE_P(BorrowedTest, BorrowedTest,
+                         ::testing::Values(AllocatorKind::kArena,
+                                           AllocatorKind::kNewDelete));
+
+}  // namespace
+}  // namespace cel
diff --git a/common/value.cc b/common/value.cc
index fa937bdf0..cc3a0825f 100644
--- a/common/value.cc
+++ b/common/value.cc
@@ -43,6 +43,7 @@
 #include "absl/types/source_location.h"
 #include "absl/types/variant.h"
 #include "common/allocator.h"
+#include "common/memory.h"
 #include "common/optional_ref.h"
 #include "common/type.h"
 #include "common/value_kind.h"
@@ -1243,8 +1244,7 @@ struct OwningWellKnownTypesValueVisitor {
                 cloned->CopyFrom(*value);
                 return ParsedJsonListValue(cloned, arena);
               }
-              return ParsedJsonListValue(value.try_as_arena_ptr()->get(),
-                                         arena);
+              return ParsedJsonListValue(value.release(), arena);
             }),
         well_known_types::AsVariant(std::move(value)));
   }
@@ -1258,23 +1258,23 @@ struct OwningWellKnownTypesValueVisitor {
               return ParsedJsonMapValue(cloned, arena);
             },
             [&](well_known_types::StructPtr value) -> MapValue {
-              if (value.GetOwningArena() != arena) {
+              if (value.arena() != arena) {
                 auto* cloned = value->New(arena);
                 cloned->CopyFrom(*value);
                 return ParsedJsonMapValue(cloned, arena);
               }
-              return ParsedJsonMapValue(value.try_as_arena_ptr()->get(), arena);
+              return ParsedJsonMapValue(value.release(), arena);
             }),
         well_known_types::AsVariant(std::move(value)));
   }
 
-  Value operator()(google::protobuf::Arena::UniquePtr<google::protobuf::Message> value) const {
+  Value operator()(Unique<google::protobuf::Message> value) const {
     if (value->GetArena() != arena) {
       auto* cloned = value->New(arena);
       cloned->CopyFrom(*value);
       return ParsedMessageValue(cloned, arena);
     }
-    return ParsedMessageValue(value.try_as_arena_ptr()->get(), arena);
+    return ParsedMessageValue(value.release(), arena);
   }
 
   template <typename T>
@@ -1344,8 +1344,7 @@ struct BorrowingWellKnownTypesValueVisitor {
                 cloned->CopyFrom(*value);
                 return ParsedJsonListValue(cloned, arena);
               }
-              return ParsedJsonListValue(value.try_as_arena_ptr()->get(),
-                                         arena);
+              return ParsedJsonListValue(value.release(), arena);
             }),
         well_known_types::AsVariant(std::move(value)));
   }
@@ -1363,18 +1362,18 @@ struct BorrowingWellKnownTypesValueVisitor {
                 cloned->CopyFrom(*value);
                 return ParsedJsonMapValue(cloned, arena);
               }
-              return ParsedJsonMapValue(value.try_as_arena_ptr()->get(), arena);
+              return ParsedJsonMapValue(value.release(), arena);
             }),
         well_known_types::AsVariant(std::move(value)));
   }
 
-  Value operator()(google::protobuf::Arena::UniquePtr<google::protobuf::Message>&& value) const {
+  Value operator()(Unique<google::protobuf::Message>&& value) const {
     if (value->GetArena() != arena) {
       auto* cloned = value->New(arena);
       cloned->CopyFrom(*value);
       return ParsedMessageValue(cloned, arena);
     }
-    return ParsedMessageValue(value.try_as_arena_ptr()->get(), arena);
+    return ParsedMessageValue(value.release(), arena);
   }
 
   template <typename T>
diff --git a/internal/BUILD b/internal/BUILD
index e49dd67e3..d6781b538 100644
--- a/internal/BUILD
+++ b/internal/BUILD
@@ -727,6 +727,7 @@ cc_test(
         ":testing_descriptor_pool",
         ":testing_message_factory",
         ":well_known_types",
+        "//common:memory",
         "@com_google_absl//absl/base:core_headers",
         "@com_google_absl//absl/base:nullability",
         "@com_google_absl//absl/log:die_if_null",
@@ -736,6 +737,7 @@ cc_test(
         "@com_google_absl//absl/strings:cord",
         "@com_google_absl//absl/strings:string_view",
         "@com_google_absl//absl/time",
+        "@com_google_absl//absl/types:variant",
         "@com_google_cel_spec//proto/cel/expr/conformance/proto3:test_all_types_cc_proto",
         "@com_google_protobuf//:any_cc_proto",
         "@com_google_protobuf//:duration_cc_proto",
@@ -840,12 +842,12 @@ cc_test(
         ":testing_descriptor_pool",
         ":testing_message_factory",
         ":well_known_types",
+        "//common:allocator",
         "//common:memory",
         "@com_google_absl//absl/base:no_destructor",
         "@com_google_absl//absl/base:nullability",
         "@com_google_absl//absl/log:absl_check",
         "@com_google_absl//absl/log:die_if_null",
-        "@com_google_absl//absl/memory",
         "@com_google_absl//absl/status:status_matchers",
         "@com_google_absl//absl/strings",
         "@com_google_absl//absl/strings:cord",
diff --git a/internal/message_equality.cc b/internal/message_equality.cc
index 921753222..34080e6b1 100644
--- a/internal/message_equality.cc
+++ b/internal/message_equality.cc
@@ -462,7 +462,7 @@ absl::StatusOr<EquatableValue> MapValueAsEquatableValue(
     MessageFactory* absl_nonnull factory, EquatableValueReflection& reflection,
     const google::protobuf::MapValueConstRef& value,
     const FieldDescriptor* absl_nonnull field, std::string& scratch,
-    google::protobuf::Arena::UniquePtr<Message>& unpacked) {
+    Unique<Message>& unpacked) {
   if (IsAnyField(field)) {
     CEL_ASSIGN_OR_RETURN(unpacked, well_known_types::UnpackAnyIfResolveable(
                                        arena, reflection.any_reflection,
@@ -516,8 +516,7 @@ absl::StatusOr<EquatableValue> RepeatedFieldAsEquatableValue(
     google::protobuf::Arena* absl_nonnull arena, const DescriptorPool* absl_nonnull pool,
     MessageFactory* absl_nonnull factory, EquatableValueReflection& reflection,
     const Message& message, const FieldDescriptor* absl_nonnull field,
-    int index, std::string& scratch,
-    google::protobuf::Arena::UniquePtr<Message>& unpacked) {
+    int index, std::string& scratch, Unique<Message>& unpacked) {
   if (IsAnyField(field)) {
     const auto& field_value =
         message.GetReflection()->GetRepeatedMessage(message, field, index);
@@ -835,8 +834,8 @@ class MessageEqualsState final {
     auto rhs_well_known_type = rhs_descriptor->well_known_type();
     const Message* absl_nonnull lhs_ptr = &lhs;
     const Message* absl_nonnull rhs_ptr = &rhs;
-    google::protobuf::Arena::UniquePtr<Message> lhs_unpacked;
-    google::protobuf::Arena::UniquePtr<Message> rhs_unpacked;
+    Unique<Message> lhs_unpacked;
+    Unique<Message> rhs_unpacked;
     // Deal with any first. We could in theory check if we should bother
     // unpacking, but that is more complicated. We can always implement it
     // later.
@@ -907,9 +906,9 @@ class MessageEqualsState final {
     }
     auto lhs_begin = ConstMapBegin(*lhs_reflection, lhs, *lhs_field);
     const auto lhs_end = ConstMapEnd(*lhs_reflection, lhs, *lhs_field);
-    google::protobuf::Arena::UniquePtr<Message> lhs_unpacked;
+    Unique<Message> lhs_unpacked;
     EquatableValue lhs_value;
-    google::protobuf::Arena::UniquePtr<Message> rhs_unpacked;
+    Unique<Message> rhs_unpacked;
     EquatableValue rhs_value;
     google::protobuf::MapKey rhs_map_key;
     google::protobuf::MapValueConstRef rhs_map_value;
@@ -961,9 +960,9 @@ class MessageEqualsState final {
     if (size != rhs_reflection->FieldSize(rhs, rhs_field)) {
       return false;
     }
-    google::protobuf::Arena::UniquePtr<Message> lhs_unpacked;
+    Unique<Message> lhs_unpacked;
     EquatableValue lhs_value;
-    google::protobuf::Arena::UniquePtr<Message> rhs_unpacked;
+    Unique<Message> rhs_unpacked;
     EquatableValue rhs_value;
     for (int i = 0; i < size; ++i) {
       CEL_ASSIGN_OR_RETURN(lhs_value,
@@ -1008,8 +1007,8 @@ class MessageEqualsState final {
     }
     const Message* absl_nonnull lhs_ptr = &lhs;
     const Message* absl_nonnull rhs_ptr = &rhs;
-    google::protobuf::Arena::UniquePtr<Message> lhs_unpacked;
-    google::protobuf::Arena::UniquePtr<Message> rhs_unpacked;
+    Unique<Message> lhs_unpacked;
+    Unique<Message> rhs_unpacked;
     if (lhs_field != nullptr && IsAnyField(lhs_field)) {
       CEL_ASSIGN_OR_RETURN(lhs_unpacked,
                            well_known_types::UnpackAnyIfResolveable(
@@ -1094,7 +1093,7 @@ class MessageEqualsState final {
         return false;
       }
       const Message* absl_nullable rhs_packed = nullptr;
-      google::protobuf::Arena::UniquePtr<Message> rhs_unpacked;
+      Unique<Message> rhs_unpacked;
       if (rhs_field != nullptr && IsAnyField(rhs_field)) {
         rhs_packed = &rhs.GetReflection()->GetMessage(rhs, rhs_field);
       } else if (rhs_field == nullptr && IsAny(rhs)) {
@@ -1175,7 +1174,7 @@ class MessageEqualsState final {
         return false;
       }
       const Message* absl_nullable lhs_packed = nullptr;
-      google::protobuf::Arena::UniquePtr<Message> lhs_unpacked;
+      Unique<Message> lhs_unpacked;
       if (lhs_field != nullptr && IsAnyField(lhs_field)) {
         lhs_packed = &lhs.GetReflection()->GetMessage(lhs, lhs_field);
       } else if (lhs_field == nullptr && IsAny(lhs)) {
@@ -1263,7 +1262,7 @@ class MessageEqualsState final {
         return false;
       }
       const Message* absl_nullable rhs_packed = nullptr;
-      google::protobuf::Arena::UniquePtr<Message> rhs_unpacked;
+      Unique<Message> rhs_unpacked;
       if (rhs_field != nullptr && IsAnyField(rhs_field)) {
         rhs_packed = &rhs.GetReflection()->GetMessage(rhs, rhs_field);
       } else if (rhs_field == nullptr && IsAny(rhs)) {
@@ -1343,7 +1342,7 @@ class MessageEqualsState final {
         return false;
       }
       const Message* absl_nullable lhs_packed = nullptr;
-      google::protobuf::Arena::UniquePtr<Message> lhs_unpacked;
+      Unique<Message> lhs_unpacked;
       if (lhs_field != nullptr && IsAnyField(lhs_field)) {
         lhs_packed = &lhs.GetReflection()->GetMessage(lhs, lhs_field);
       } else if (lhs_field == nullptr && IsAny(lhs)) {
diff --git a/internal/message_equality_test.cc b/internal/message_equality_test.cc
index c08116181..092edd71b 100644
--- a/internal/message_equality_test.cc
+++ b/internal/message_equality_test.cc
@@ -14,7 +14,6 @@
 
 #include "internal/message_equality.h"
 
-#include <memory>
 #include <string>
 #include <utility>
 #include <vector>
@@ -28,12 +27,12 @@
 #include "absl/base/nullability.h"
 #include "absl/log/absl_check.h"
 #include "absl/log/die_if_null.h"
-#include "absl/memory/memory.h"
 #include "absl/status/status_matchers.h"
 #include "absl/strings/cord.h"
 #include "absl/strings/str_cat.h"
 #include "absl/strings/string_view.h"
 #include "absl/types/optional.h"
+#include "common/allocator.h"
 #include "common/memory.h"
 #include "internal/message_type_name.h"
 #include "internal/parse_text_proto.h"
@@ -395,7 +394,7 @@ void PackMessageTo(const google::protobuf::Message& message, google::protobuf::M
   reflection.SetValue(instance, value);
 }
 
-absl::optional<std::pair<std::unique_ptr<google::protobuf::Message>,
+absl::optional<std::pair<Owned<google::protobuf::Message>,
                          const google::protobuf::FieldDescriptor* absl_nonnull>>
 PackTestAllTypesProto3Field(const google::protobuf::Message& message,
                             const google::protobuf::FieldDescriptor* absl_nonnull field) {
@@ -406,7 +405,7 @@ PackTestAllTypesProto3Field(const google::protobuf::Message& message,
       field->type() == google::protobuf::FieldDescriptor::TYPE_MESSAGE) {
     const auto* descriptor = message.GetDescriptor();
     const auto* any_field = descriptor->FindFieldByName("repeated_any");
-    auto packed = absl::WrapUnique(message.New());
+    auto packed = WrapShared(message.New(), NewDeleteAllocator<>{});
     const int size = message.GetReflection()->FieldSize(message, field);
     for (int i = 0; i < size; ++i) {
       PackMessageTo(
@@ -414,17 +413,17 @@ PackTestAllTypesProto3Field(const google::protobuf::Message& message,
           packed->GetReflection()->AddMessage(cel::to_address(packed),
                                               any_field));
     }
-    return std::pair{std::move(packed), any_field};
+    return std::pair{packed, any_field};
   }
   if (!field->is_repeated() &&
       field->type() == google::protobuf::FieldDescriptor::TYPE_MESSAGE) {
     const auto* descriptor = message.GetDescriptor();
     const auto* any_field = descriptor->FindFieldByName("single_any");
-    auto packed = absl::WrapUnique(message.New());
+    auto packed = WrapShared(message.New(), NewDeleteAllocator<>{});
     PackMessageTo(message.GetReflection()->GetMessage(message, field),
                   packed->GetReflection()->MutableMessage(
                       cel::to_address(packed), any_field));
-    return std::pair{std::move(packed), any_field};
+    return std::pair{packed, any_field};
   }
   return std::nullopt;
 }
@@ -500,10 +499,10 @@ TEST_P(UnaryMessageFieldEqualsTest, Equals) {
             << rhs_field->name();
       }
       // Test `google.protobuf.Any`.
-      absl::optional<std::pair<google::protobuf::Arena::UniquePtr<google::protobuf::Message>,
+      absl::optional<std::pair<Owned<google::protobuf::Message>,
                                const google::protobuf::FieldDescriptor* absl_nonnull>>
           lhs_any = PackTestAllTypesProto3Field(*lhs_message, lhs_field);
-      absl::optional<std::pair<google::protobuf::Arena::UniquePtr<google::protobuf::Message>,
+      absl::optional<std::pair<Owned<google::protobuf::Message>,
                                const google::protobuf::FieldDescriptor* absl_nonnull>>
           rhs_any = PackTestAllTypesProto3Field(*rhs_message, rhs_field);
       if (lhs_any) {
diff --git a/internal/well_known_types.cc b/internal/well_known_types.cc
index f48d77f7f..5232bb6ad 100644
--- a/internal/well_known_types.cc
+++ b/internal/well_known_types.cc
@@ -1489,7 +1489,7 @@ google::protobuf::Message* absl_nonnull ValueReflection::MutableStructValue(
   return message->GetReflection()->MutableMessage(message, struct_value_field_);
 }
 
-google::protobuf::Arena::UniquePtr<google::protobuf::Message> ValueReflection::ReleaseListValue(
+Unique<google::protobuf::Message> ValueReflection::ReleaseListValue(
     google::protobuf::Message* absl_nonnull message) const {
   ABSL_DCHECK(IsInitialized());
   ABSL_DCHECK_EQ(message->GetDescriptor(), descriptor_);
@@ -1497,12 +1497,12 @@ google::protobuf::Arena::UniquePtr<google::protobuf::Message> ValueReflection::R
   if (!reflection->HasField(*message, list_value_field_)) {
     reflection->MutableMessage(message, list_value_field_);
   }
-  return google::protobuf::Arena::UnsafeWrapUniquePtr(
-      message->GetArena(),
-      reflection->UnsafeArenaReleaseMessage(message, list_value_field_));
+  return WrapUnique(
+      reflection->UnsafeArenaReleaseMessage(message, list_value_field_),
+      message->GetArena());
 }
 
-google::protobuf::Arena::UniquePtr<google::protobuf::Message> ValueReflection::ReleaseStructValue(
+Unique<google::protobuf::Message> ValueReflection::ReleaseStructValue(
     google::protobuf::Message* absl_nonnull message) const {
   ABSL_DCHECK(IsInitialized());
   ABSL_DCHECK_EQ(message->GetDescriptor(), descriptor_);
@@ -1510,9 +1510,9 @@ google::protobuf::Arena::UniquePtr<google::protobuf::Message> ValueReflection::R
   if (!reflection->HasField(*message, struct_value_field_)) {
     reflection->MutableMessage(message, struct_value_field_);
   }
-  return google::protobuf::Arena::UnsafeWrapUniquePtr(
-      message->GetArena(),
-      reflection->UnsafeArenaReleaseMessage(message, struct_value_field_));
+  return WrapUnique(
+      reflection->UnsafeArenaReleaseMessage(message, struct_value_field_),
+      message->GetArena());
 }
 
 absl::StatusOr<ValueReflection> GetValueReflection(
@@ -1899,9 +1899,9 @@ namespace {
 // it as `ListValue`. If adapted is empty, we return as a reference. If adapted
 // is present, message must be a reference to the value held in adapted and it
 // will be returned by value.
-absl::StatusOr<ListValue> AdaptListValue(
-    google::protobuf::Arena* absl_nullable arena, const google::protobuf::Message& message,
-    google::protobuf::Arena::UniquePtr<google::protobuf::Message> adapted) {
+absl::StatusOr<ListValue> AdaptListValue(google::protobuf::Arena* absl_nullable arena,
+                                         const google::protobuf::Message& message,
+                                         Unique<google::protobuf::Message> adapted) {
   ABSL_DCHECK(!adapted || &message == cel::to_address(adapted));
   const auto* descriptor = message.GetDescriptor();
   if (ABSL_PREDICT_FALSE(descriptor == nullptr)) {
@@ -1922,9 +1922,9 @@ absl::StatusOr<ListValue> AdaptListValue(
 // as `Struct`. If adapted is empty, we return as a reference. If adapted is
 // present, message must be a reference to the value held in adapted and it will
 // be returned by value.
-absl::StatusOr<Struct> AdaptStruct(
-    google::protobuf::Arena* absl_nullable arena, const google::protobuf::Message& message,
-    google::protobuf::Arena::UniquePtr<google::protobuf::Message> adapted) {
+absl::StatusOr<Struct> AdaptStruct(google::protobuf::Arena* absl_nullable arena,
+                                   const google::protobuf::Message& message,
+                                   Unique<google::protobuf::Message> adapted) {
   ABSL_DCHECK(!adapted || &message == cel::to_address(adapted));
   const auto* descriptor = message.GetDescriptor();
   if (ABSL_PREDICT_FALSE(descriptor == nullptr)) {
@@ -1942,14 +1942,14 @@ absl::StatusOr<Struct> AdaptStruct(
 
 // AdaptAny recursively unpacks a protocol buffer message which is an instance
 // of `google.protobuf.Any`.
-absl::StatusOr<google::protobuf::Arena::UniquePtr<google::protobuf::Message>> AdaptAny(
+absl::StatusOr<Unique<google::protobuf::Message>> AdaptAny(
     google::protobuf::Arena* absl_nullable arena, AnyReflection& reflection,
     const google::protobuf::Message& message, const Descriptor* absl_nonnull descriptor,
     const DescriptorPool* absl_nonnull pool,
     google::protobuf::MessageFactory* absl_nonnull factory, bool error_if_unresolveable) {
   ABSL_DCHECK_EQ(descriptor->well_known_type(), Descriptor::WELLKNOWNTYPE_ANY);
   const google::protobuf::Message* absl_nonnull to_unwrap = &message;
-  google::protobuf::Arena::UniquePtr<google::protobuf::Message> unwrapped;
+  Unique<google::protobuf::Message> unwrapped;
   std::string type_url_scratch;
   std::string value_scratch;
   do {
@@ -1979,8 +1979,7 @@ absl::StatusOr<google::protobuf::Arena::UniquePtr<google::protobuf::Message>> Ad
           "unable to build prototype for type name: ", type_url_view));
     }
     BytesValue value = reflection.GetValue(*to_unwrap, value_scratch);
-    google::protobuf::Arena::UniquePtr<google::protobuf::Message> unpacked =
-        google::protobuf::Arena::UnsafeWrapUniquePtr(arena, prototype->New(arena));
+    Unique<google::protobuf::Message> unpacked = WrapUnique(prototype->New(arena), arena);
     // TODO(b/557267722): Extensions that are not included in the same
     // descriptor pool as the resolved descriptor will be treated as unknown
     // fields. Extending messages like this should be exceedingly rare and is
@@ -2015,7 +2014,7 @@ absl::StatusOr<google::protobuf::Arena::UniquePtr<google::protobuf::Message>> Ad
 
 }  // namespace
 
-absl::StatusOr<google::protobuf::Arena::UniquePtr<google::protobuf::Message>> UnpackAnyFrom(
+absl::StatusOr<Unique<google::protobuf::Message>> UnpackAnyFrom(
     google::protobuf::Arena* absl_nullable arena, AnyReflection& reflection,
     const google::protobuf::Message& message,
     const google::protobuf::DescriptorPool* absl_nonnull pool,
@@ -2026,12 +2025,11 @@ absl::StatusOr<google::protobuf::Arena::UniquePtr<google::protobuf::Message>> Un
                   factory, /*error_if_unresolveable=*/true);
 }
 
-absl::StatusOr<google::protobuf::Arena::UniquePtr<google::protobuf::Message>>
-UnpackAnyIfResolveable(google::protobuf::Arena* absl_nullable arena,
-                       AnyReflection& reflection,
-                       const google::protobuf::Message& message,
-                       const google::protobuf::DescriptorPool* absl_nonnull pool,
-                       google::protobuf::MessageFactory* absl_nonnull factory) {
+absl::StatusOr<Unique<google::protobuf::Message>> UnpackAnyIfResolveable(
+    google::protobuf::Arena* absl_nullable arena, AnyReflection& reflection,
+    const google::protobuf::Message& message,
+    const google::protobuf::DescriptorPool* absl_nonnull pool,
+    google::protobuf::MessageFactory* absl_nonnull factory) {
   ABSL_DCHECK_EQ(message.GetDescriptor()->well_known_type(),
                  Descriptor::WELLKNOWNTYPE_ANY);
   return AdaptAny(arena, reflection, message, message.GetDescriptor(), pool,
@@ -2049,7 +2047,7 @@ absl::StatusOr<well_known_types::Value> AdaptFromMessage(
                      message.GetTypeName()));
   }
   const google::protobuf::Message* absl_nonnull to_adapt;
-  google::protobuf::Arena::UniquePtr<google::protobuf::Message> adapted;
+  Unique<google::protobuf::Message> adapted;
   Descriptor::WellKnownType well_known_type = descriptor->well_known_type();
   if (well_known_type == Descriptor::WELLKNOWNTYPE_ANY) {
     AnyReflection reflection;
diff --git a/internal/well_known_types.h b/internal/well_known_types.h
index 3e09b286b..f63e5e76b 100644
--- a/internal/well_known_types.h
+++ b/internal/well_known_types.h
@@ -50,6 +50,7 @@
 #include "absl/time/time.h"
 #include "absl/types/variant.h"
 #include "common/any.h"
+#include "common/memory.h"
 #include "google/protobuf/arena.h"
 #include "google/protobuf/descriptor.h"
 #include "google/protobuf/map_field.h"
@@ -1006,10 +1007,10 @@ class ValueReflection final {
   google::protobuf::Message* absl_nonnull MutableStructValue(
       google::protobuf::Message* absl_nonnull message) const;
 
-  google::protobuf::Arena::UniquePtr<google::protobuf::Message> ReleaseListValue(
+  Unique<google::protobuf::Message> ReleaseListValue(
       google::protobuf::Message* absl_nonnull message) const;
 
-  google::protobuf::Arena::UniquePtr<google::protobuf::Message> ReleaseStructValue(
+  Unique<google::protobuf::Message> ReleaseStructValue(
       google::protobuf::Message* absl_nonnull message) const;
 
  private:
@@ -1277,18 +1278,18 @@ absl::StatusOr<FieldMaskReflection> GetFieldMaskReflection(
     const google::protobuf::Descriptor* absl_nonnull descriptor
         ABSL_ATTRIBUTE_LIFETIME_BOUND);
 
-using ListValuePtr = google::protobuf::Arena::UniquePtr<google::protobuf::Message>;
+using ListValuePtr = Unique<google::protobuf::Message>;
 
 using ListValueConstRef = std::reference_wrapper<const google::protobuf::Message>;
 
-using StructPtr = google::protobuf::Arena::UniquePtr<google::protobuf::Message>;
+using StructPtr = Unique<google::protobuf::Message>;
 
 using StructConstRef = std::reference_wrapper<const google::protobuf::Message>;
 
 // Variant holding `std::reference_wrapper<const
-// google::protobuf::Message>` or `google::protobuf::Arena::UniquePtr<google::protobuf::Message>`, either of
-// which is an instance of `google.protobuf.ListValue` which is either a
-// generated message or dynamic message.
+// google::protobuf::Message>` or `Unique<google::protobuf::Message>`, either of which is an
+// instance of `google.protobuf.ListValue` which is either a generated message
+// or dynamic message.
 class ListValue final : public absl::variant<ListValueConstRef, ListValuePtr> {
   using absl::variant<ListValueConstRef, ListValuePtr>::variant;
 };
@@ -1315,9 +1316,9 @@ inline absl::variant<ListValueConstRef, ListValuePtr>&& AsVariant(
 }
 
 // Variant holding `std::reference_wrapper<const
-// google::protobuf::Message>` or `google::protobuf::Arena::UniquePtr<google::protobuf::Message>`, either of
-// which is an instance of `google.protobuf.Struct` which is either a generated
-// message or dynamic message.
+// google::protobuf::Message>` or `Unique<google::protobuf::Message>`, either of which is an
+// instance of `google.protobuf.Struct` which is either a generated message or
+// dynamic message.
 class Struct final : public absl::variant<StructConstRef, StructPtr> {
  public:
   using absl::variant<StructConstRef, StructPtr>::variant;
@@ -1341,14 +1342,13 @@ inline absl::variant<StructConstRef, StructPtr>&& AsVariant(Struct&& value) {
 }
 
 // Variant capable of representing any unwrapped well known type or message.
-using Value =
-    absl::variant<absl::monostate, std::nullptr_t, bool, int32_t, int64_t,
-                  uint32_t, uint64_t, float, double, StringValue, BytesValue,
-                  absl::Duration, absl::Time, ListValue, Struct,
-                  google::protobuf::Arena::UniquePtr<google::protobuf::Message>>;
+using Value = absl::variant<absl::monostate, std::nullptr_t, bool, int32_t,
+                            int64_t, uint32_t, uint64_t, float, double,
+                            StringValue, BytesValue, absl::Duration, absl::Time,
+                            ListValue, Struct, Unique<google::protobuf::Message>>;
 
 // Unpacks the given instance of `google.protobuf.Any`.
-absl::StatusOr<google::protobuf::Arena::UniquePtr<google::protobuf::Message>> UnpackAnyFrom(
+absl::StatusOr<Unique<google::protobuf::Message>> UnpackAnyFrom(
     google::protobuf::Arena* absl_nullable arena ABSL_ATTRIBUTE_LIFETIME_BOUND,
     AnyReflection& reflection, const google::protobuf::Message& message,
     const google::protobuf::DescriptorPool* absl_nonnull pool
@@ -1356,8 +1356,7 @@ absl::StatusOr<google::protobuf::Arena::UniquePtr<google::protobuf::Message>> Un
     google::protobuf::MessageFactory* absl_nonnull factory ABSL_ATTRIBUTE_LIFETIME_BOUND);
 
 // Unpacks the given instance of `google.protobuf.Any` if it is resolvable.
-absl::StatusOr<google::protobuf::Arena::UniquePtr<google::protobuf::Message>>
-UnpackAnyIfResolveable(
+absl::StatusOr<Unique<google::protobuf::Message>> UnpackAnyIfResolveable(
     google::protobuf::Arena* absl_nullable arena ABSL_ATTRIBUTE_LIFETIME_BOUND,
     AnyReflection& reflection, const google::protobuf::Message& message,
     const google::protobuf::DescriptorPool* absl_nonnull pool
diff --git a/internal/well_known_types_test.cc b/internal/well_known_types_test.cc
index aae66c51c..5ffc06ee9 100644
--- a/internal/well_known_types_test.cc
+++ b/internal/well_known_types_test.cc
@@ -34,6 +34,8 @@
 #include "absl/strings/cord.h"
 #include "absl/strings/string_view.h"
 #include "absl/time/time.h"
+#include "absl/types/variant.h"
+#include "common/memory.h"
 #include "internal/message_type_name.h"
 #include "internal/minimal_descriptor_pool.h"
 #include "internal/parse_text_proto.h"
@@ -1025,10 +1027,8 @@ TEST_F(AdaptFromMessageTest, Any_Struct) {
 TEST_F(AdaptFromMessageTest, Any_TestAllTypesProto3) {
   auto message = DynamicParseTextProto<google::protobuf::Any>(
       R"pb(type_url: "type.googleapis.com/cel.expr.conformance.proto3.TestAllTypes")pb");
-  EXPECT_THAT(
-      AdaptFromMessage(*message),
-      IsOkAndHolds(
-          VariantWith<google::protobuf::Arena::UniquePtr<google::protobuf::Message>>(NotNull())));
+  EXPECT_THAT(AdaptFromMessage(*message),
+              IsOkAndHolds(VariantWith<Unique<google::protobuf::Message>>(NotNull())));
 }
 
 TEST_F(AdaptFromMessageTest, Any_BadTypeUrlDomain) {