@@ -21,3 +21,15 @@ java_library(
     visibility = ["//:internal"],
     exports = ["//runtime/src/main/java/dev/cel/runtime/planner:planned_program"],
 )
+
+java_library(
+    name = "async_gate",
+    visibility = ["//:internal"],
+    exports = ["//runtime/src/main/java/dev/cel/runtime/planner:async_gate"],
+)
+
+cel_android_library(
+    name = "async_gate_android",
+    visibility = ["//:internal"],
+    exports = ["//runtime/src/main/java/dev/cel/runtime/planner:async_gate_android"],
+)
@@ -0,0 +1,305 @@
+// Copyright 2026 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.
+
+package dev.cel.runtime.planner;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import static com.google.common.util.concurrent.MoreExecutors.directExecutor;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.errorprone.annotations.CanIgnoreReturnValue;
+import javax.annotation.concurrent.ThreadSafe;
+import java.util.Queue;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.jspecify.annotations.Nullable;
+
+/** Regulates the number of concurrent asynchronous function executions based on maxConcurrency. */
+@ThreadSafe
+final class AsyncGate {
+
+  private final @Nullable Semaphore semaphore;
+
+  // CEL-Internal-4
+  private final Queue<QueuedTask> pendingTasks;
+
+  private final AtomicBoolean cancelled;
+  private final ThreadLocal<Boolean> isDraining;
+
+  /** Represents an asynchronous task subject to concurrency gating. */
+  interface GatedTask {
+    /**
+     * Starts the asynchronous task.
+     *
+     * <p>The concurrency gate holds a permit until the returned {@link ListenableFuture} completes.
+     */
+    ListenableFuture<?> start();
+
+    /**
+     * Invoked instead of {@link #start} when the task cannot be admitted (e.g. the gate is
+     * cancelled or the executor rejects execution).
+     */
+    default void reject(Throwable cause) {}
+  }
+
+  static final class QueuedTask {
+    private final GatedTask task;
+    private final Executor executor;
+
+    QueuedTask(GatedTask task, Executor executor) {
+      this.task = checkNotNull(task, "task must not be null");
+      this.executor = checkNotNull(executor, "executor must not be null");
+    }
+  }
+
+  /** Cancels the gate, rejecting all pending queued tasks and preventing new dispatches. */
+  void cancel() {
+    cancelled.set(true);
+    clearPendingTasks(new CancellationException("AsyncGate is cancelled"));
+  }
+
+  /**
+   * Dispatches a gated task for execution.
+   *
+   * <p>If concurrency permits allow, the task is submitted to the executor immediately. Otherwise,
+   * it is enqueued into pending tasks and will be drained when an active task completes.
+   *
+   * @return true if the task was admitted or queued for execution, false if rejected due to
+   *     cancellation.
+   */
+  @CanIgnoreReturnValue
+  boolean dispatch(Executor executor, GatedTask task) {
+    checkNotNull(executor, "executor must not be null");
+    checkNotNull(task, "task must not be null");
+    if (cancelled.get()) {
+      task.reject(new CancellationException("AsyncGate is cancelled"));
+      return false;
+    }
+
+    if (semaphore == null) {
+      AtomicBoolean started = new AtomicBoolean(false);
+      try {
+        executor.execute(
+            () -> {
+              started.set(true);
+              if (cancelled.get()) {
+                task.reject(new CancellationException("AsyncGate is cancelled"));
+                return;
+              }
+              executeTaskWithoutPermit(task);
+            });
+        return true;
+      } catch (Throwable t) {
+        if (!started.get()) {
+          task.reject(t);
+          return false;
+        }
+        throw t;
+      }
+    }
+
+    if (pendingTasks.isEmpty() && semaphore.tryAcquire()) {
+      AtomicBoolean started = new AtomicBoolean(false);
+      try {
+        executor.execute(
+            () -> {
+              started.set(true);
+              if (cancelled.get()) {
+                semaphore.release();
+                task.reject(new CancellationException("AsyncGate is cancelled"));
+                drainPending();
+                return;
+              }
+              runTaskWithPermit(task);
+            });
+        return true;
+      } catch (Throwable t) {
+        if (!started.get()) {
+          semaphore.release();
+          drainPending();
+          task.reject(t);
+          return false;
+        }
+        throw t;
+      }
+    }
+
+    pendingTasks.add(new QueuedTask(task, executor));
+    drainPending();
+    return true;
+  }
+
+  private void runTaskWithPermit(GatedTask task) {
+    ListenableFuture<?> future;
+    try {
+      future = checkNotNull(task.start(), "GatedTask.start() returned null");
+    } catch (Throwable t) {
+      semaphore.release();
+      try {
+        task.reject(t);
+      } catch (Throwable e) {
+        // Suppress secondary exception from reject callback to preserve primary cause t.
+        t.addSuppressed(e);
+      }
+      drainPending();
+      throw t;
+    }
+
+    Futures.addCallback(
+        future,
+        new FutureCallback<Object>() {
+          @Override
+          public void onSuccess(@Nullable Object result) {
+            release();
+          }
+
+          @Override
+          public void onFailure(Throwable t) {
+            release();
+          }
+
+          private void release() {
+            semaphore.release();
+            drainPending();
+          }
+        },
+        directExecutor());
+  }
+
+  @SuppressWarnings("FutureReturnValueIgnored")
+  private void executeTaskWithoutPermit(GatedTask task) {
+    try {
+      checkNotNull(task.start(), "GatedTask.start() returned null");
+    } catch (Throwable t) {
+      try {
+        task.reject(t);
+      } catch (Throwable e) {
+        // Suppress secondary exception from reject callback to preserve primary cause t.
+        t.addSuppressed(e);
+      }
+      throw t;
+    }
+  }
+
+  @VisibleForTesting
+  void drainPending() {
+    if (isDraining.get()) {
+      return;
+    }
+    isDraining.set(true);
+    try {
+      if (cancelled.get()) {
+        clearPendingTasks(new CancellationException("AsyncGate is cancelled"));
+        return;
+      }
+      if (semaphore == null) {
+        return;
+      }
+
+      while (!pendingTasks.isEmpty()) {
+        if (!semaphore.tryAcquire()) {
+          break;
+        }
+        QueuedTask queuedTask = pendingTasks.poll();
+        if (queuedTask == null) {
+          semaphore.release();
+          break;
+        }
+
+        try {
+          queuedTask.executor.execute(
+              () -> {
+                if (cancelled.get()) {
+                  semaphore.release();
+                  queuedTask.task.reject(new CancellationException("AsyncGate is cancelled"));
+                  drainPending();
+                  return;
+                }
+                try {
+                  runTaskWithPermit(queuedTask.task);
+                } catch (Throwable ignored) {
+                  // runTaskWithPermit already releases permit, calls reject, and drains.
+                }
+              });
+        } catch (Throwable e) {
+          semaphore.release();
+          try {
+            queuedTask.task.reject(e);
+          } catch (Throwable t) {
+            // Suppress secondary exception from reject callback to preserve primary cause e.
+            e.addSuppressed(t);
+          }
+        }
+      }
+    } finally {
+      isDraining.set(false);
+    }
+  }
+
+  private void clearPendingTasks(Throwable cause) {
+    QueuedTask queuedTask;
+    while ((queuedTask = pendingTasks.poll()) != null) {
+      try {
+        queuedTask.task.reject(cause);
+      } catch (Throwable t) {
+        // Suppress exception from reject callback to preserve primary cancellation cause.
+        cause.addSuppressed(t);
+      }
+    }
+  }
+
+  @VisibleForTesting
+  boolean isCancelled() {
+    return cancelled.get();
+  }
+
+  @VisibleForTesting
+  int availablePermits() {
+    return semaphore != null ? semaphore.availablePermits() : -1;
+  }
+
+  @VisibleForTesting
+  int queuedTaskCount() {
+    return pendingTasks.size();
+  }
+
+  static AsyncGate create(int maxConcurrency) {
+    return new AsyncGate(maxConcurrency);
+  }
+
+  @VisibleForTesting
+  AsyncGate(int maxConcurrency) {
+    this(maxConcurrency > 0 ? new Semaphore(maxConcurrency) : null, new ConcurrentLinkedQueue<>());
+  }
+
+  @VisibleForTesting
+  AsyncGate(@Nullable Semaphore semaphore, Queue<QueuedTask> pendingTasks) {
+    this.semaphore = semaphore;
+    this.pendingTasks = checkNotNull(pendingTasks, "pendingTasks must not be null");
+    this.cancelled = new AtomicBoolean(false);
+    this.isDraining =
+        new ThreadLocal<Boolean>() {
+          @Override
+          protected Boolean initialValue() {
+            return Boolean.FALSE;
+          }
+        };
+  }
+}
@@ -187,6 +187,19 @@ java_library(
     ],
 )
 
+java_library(
+    name = "async_gate",
+    srcs = ["AsyncGate.java"],
+    tags = [
+    ],
+    deps = [
+        "@maven//:com_google_code_findbugs_annotations",
+        "@maven//:com_google_errorprone_error_prone_annotations",
+        "@maven//:com_google_guava_guava",
+        "@maven//:org_jspecify_jspecify",
+    ],
+)
+
 java_library(
     name = "activation_wrapper",
     srcs = ["ActivationWrapper.java"],
@@ -717,6 +730,20 @@ cel_android_library(
     ],
 )
 
+cel_android_library(
+    name = "async_gate_android",
+    srcs = ["AsyncGate.java"],
+    tags = [
+    ],
+    deps = [
+        "@maven//:com_google_code_findbugs_annotations",
+        "@maven//:com_google_errorprone_error_prone_annotations",
+        "@maven//:com_google_guava_guava",
+        "@maven//:org_jspecify_jspecify",
+        "@maven_android//:com_google_guava_guava",
+    ],
+)
+
 cel_android_library(
     name = "activation_wrapper_android",
     srcs = ["ActivationWrapper.java"],