Add async execution engine and AsyncDriver

runtime/src/main/java/dev/cel/runtime/CelRuntimeImpl.java

@@ -191,8 +191,11 @@ public ListenableFuture<Object> evalAsync(PartialVars partialVars) {
 
       @Override
       public ListenableFuture<Object> evalAsync(Message message) {
-        throw new UnsupportedOperationException(
-            "evalAsync is not supported by this Program implementation.");
+        checkNotNull(message, "message");
+        return program.evalAsync(
+            ProtoMessageActivationFactory.fromProto(message, program.options()),
+            CelFunctionResolver.EMPTY,
+            /* partialVars= */ null);
       }
 
       @Override
@@ -580,6 +583,7 @@ public CelRuntime build() {
               container(),
               options(),
               lateBoundFunctionNamesBuilder().build(),
+              runtimeEquality,
               asyncEvaluationOptions(),
               asyncExecutor().orElse(null));
       setPlanner(planner);

runtime/src/main/java/dev/cel/runtime/LiteRuntimeImpl.java

@@ -183,10 +183,10 @@ public CelLiteRuntime build() {
       ImmutableMap.Builder<String, CelFunctionBinding> functionBindingsBuilder =
           ImmutableMap.builder();
 
+      RuntimeHelpers runtimeHelpers = RuntimeHelpers.create();
+      RuntimeEquality runtimeEquality = RuntimeEquality.create(runtimeHelpers, celOptions);
       ImmutableSet<CelStandardFunction> standardFunctions = standardFunctionBuilder.build();
       if (!standardFunctions.isEmpty()) {
-        RuntimeHelpers runtimeHelpers = RuntimeHelpers.create();
-        RuntimeEquality runtimeEquality = RuntimeEquality.create(runtimeHelpers, celOptions);
         for (CelStandardFunction standardFunction : standardFunctions) {
           ImmutableSet<CelFunctionBinding> standardFunctionBinding =
               standardFunction.newFunctionBindings(celOptions, runtimeEquality);
@@ -230,6 +230,7 @@ public CelLiteRuntime build() {
               container,
               celOptions,
               lateBoundFunctionNamesBuilder.build(),
+              runtimeEquality,
               // TODO: Support async eval in lite runtime.
               CelAsyncEvaluationOptions.defaultOptions(),
               /* asyncExecutor= */ null);

runtime/src/main/java/dev/cel/runtime/planner/BUILD.bazel

@@ -61,6 +61,7 @@ java_library(
         "//runtime:evaluation_exception_builder",
         "//runtime:function_overload",
         "//runtime:resolved_overload",
+        "//runtime:runtime_equality",
         "@maven//:com_google_code_findbugs_annotations",
         "@maven//:com_google_errorprone_error_prone_annotations",
         "@maven//:com_google_guava_guava",
@@ -74,6 +75,9 @@ java_library(
     tags = [
     ],
     deps = [
+        ":async_call_state_tracker",
+        ":async_completion_coordinator",
+        ":async_gate",
         ":error_metadata",
         ":localized_evaluation_exception",
         ":planned_interpretable",
@@ -82,6 +86,7 @@ java_library(
         "//common/annotations",
         "//common/exceptions:runtime_exception",
         "//common/values",
+        "//runtime:accumulated_unknowns",
         "//runtime:activation",
         "//runtime:async_options",
         "//runtime:evaluation_exception",
@@ -92,6 +97,7 @@ java_library(
         "//runtime:interpreter_util",
         "//runtime:partial_vars",
         "//runtime:program",
+        "//runtime:runtime_equality",
         "//runtime:variable_resolver",
         "@maven//:com_google_errorprone_error_prone_annotations",
         "@maven//:com_google_guava_guava",
@@ -250,10 +256,19 @@ java_library(
     name = "eval_async_call",
     srcs = ["EvalAsyncCall.java"],
     deps = [
+        ":eval_helpers",
+        ":localized_evaluation_exception",
         ":planned_interpretable",
+        "//common:error_codes",
         "//common/ast",
+        "//common/exceptions:overload_not_found",
+        "//common/exceptions:runtime_exception",
+        "//common/values",
+        "//runtime:accumulated_unknowns",
         "//runtime:evaluation_exception",
+        "//runtime:function_overload",
         "//runtime:interpretable",
+        "//runtime:resolved_overload",
         "@maven//:com_google_errorprone_error_prone_annotations",
         "@maven//:com_google_guava_guava",
     ],
@@ -457,6 +472,7 @@ java_library(
         "//common/values",
         "//runtime:accumulated_unknowns",
         "//runtime:evaluation_exception",
+        "//runtime:function_overload",
         "//runtime:interpretable",
         "//runtime:resolved_overload",
         "@maven//:com_google_guava_guava",
@@ -601,6 +617,7 @@ java_library(
         "PlannedInterpretable.java",
     ],
     deps = [
+        ":async_call_state_tracker",
         ":localized_evaluation_exception",
         "//common:options",
         "//common/ast",
@@ -613,6 +630,7 @@ java_library(
         "//runtime:partial_vars",
         "//runtime:resolved_overload",
         "@maven//:com_google_errorprone_error_prone_annotations",
+        "@maven//:com_google_guava_guava",
         "@maven//:org_jspecify_jspecify",
     ],
 )
@@ -670,6 +688,7 @@ cel_android_library(
         "//runtime:evaluation_exception_builder",
         "//runtime:function_overload_android",
         "//runtime:resolved_overload_android",
+        "//runtime:runtime_equality_android",
         "@maven//:com_google_errorprone_error_prone_annotations",
         "@maven//:org_jspecify_jspecify",
         "@maven_android//:com_google_guava_guava",
@@ -682,6 +701,9 @@ cel_android_library(
     tags = [
     ],
     deps = [
+        ":async_call_state_tracker_android",
+        ":async_completion_coordinator_android",
+        ":async_gate_android",
         ":error_metadata_android",
         ":localized_evaluation_exception_android",
         ":planned_interpretable_android",
@@ -690,11 +712,13 @@ cel_android_library(
         "//common/annotations",
         "//common/exceptions:runtime_exception",
         "//common/values:values_android",
+        "//runtime:accumulated_unknowns_android",
         "//runtime:activation_android",
         "//runtime:async_options_android",
         "//runtime:evaluation_exception",
         "//runtime:evaluation_exception_builder",
         "//runtime:interpretable_android",
+        "//runtime:runtime_equality_android",
         "//runtime:variable_resolver",
         "//runtime/src/main/java/dev/cel/runtime:evaluation_listener_android",
         "//runtime/src/main/java/dev/cel/runtime:function_resolver_android",
@@ -860,10 +884,19 @@ cel_android_library(
     name = "eval_async_call_android",
     srcs = ["EvalAsyncCall.java"],
     deps = [
+        ":eval_helpers_android",
+        ":localized_evaluation_exception_android",
         ":planned_interpretable_android",
+        "//common:error_codes",
         "//common/ast:ast_android",
+        "//common/exceptions:overload_not_found",
+        "//common/exceptions:runtime_exception",
+        "//common/values:values_android",
+        "//runtime:accumulated_unknowns_android",
         "//runtime:evaluation_exception",
+        "//runtime:function_overload_android",
         "//runtime:interpretable_android",
+        "//runtime:resolved_overload_android",
         "@maven//:com_google_errorprone_error_prone_annotations",
         "@maven_android//:com_google_guava_guava",
     ],
@@ -1066,6 +1099,7 @@ cel_android_library(
         "//common/exceptions:overload_not_found",
         "//common/values:values_android",
         "//runtime:evaluation_exception",
+        "//runtime:function_overload_android",
         "//runtime:interpretable_android",
         "//runtime:resolved_overload_android",
         "//runtime/src/main/java/dev/cel/runtime:accumulated_unknowns_android",
@@ -1206,6 +1240,7 @@ cel_android_library(
         "PlannedInterpretable.java",
     ],
     deps = [
+        ":async_call_state_tracker_android",
         ":localized_evaluation_exception_android",
         "//common:options",
         "//common/ast:ast_android",
@@ -1219,5 +1254,6 @@ cel_android_library(
         "//runtime/src/main/java/dev/cel/runtime:partial_vars_android",
         "@maven//:com_google_errorprone_error_prone_annotations",
         "@maven//:org_jspecify_jspecify",
+        "@maven_android//:com_google_guava_guava",
     ],
 )

runtime/src/main/java/dev/cel/runtime/planner/EvalAsyncCall.java

@@ -15,33 +15,106 @@
 package dev.cel.runtime.planner;
 
 import static com.google.common.base.Preconditions.checkNotNull;
+import static dev.cel.runtime.planner.EvalHelpers.evalStrictly;
 
+import com.google.common.collect.ImmutableList;
 import com.google.errorprone.annotations.Immutable;
+import dev.cel.common.CelErrorCode;
 import dev.cel.common.ast.CelExpr;
+import dev.cel.common.exceptions.CelOverloadNotFoundException;
+import dev.cel.common.exceptions.CelRuntimeException;
+import dev.cel.common.values.CelValueConverter;
+import dev.cel.runtime.AccumulatedUnknowns;
+import dev.cel.runtime.CelAsyncFunctionOverload;
 import dev.cel.runtime.CelEvaluationException;
+import dev.cel.runtime.CelFunctionOverload;
+import dev.cel.runtime.CelResolvedOverload;
 import dev.cel.runtime.GlobalResolver;
 
 /** Evaluates an asynchronous function call within a planned program. */
 @Immutable
 final class EvalAsyncCall extends PlannedInterpretable {
 
   private final String functionName;
+  private final CelResolvedOverload resolvedOverload;
+  private final CelAsyncFunctionOverload overload;
 
-  static EvalAsyncCall create(CelExpr expr, String functionName) {
-    return new EvalAsyncCall(expr, functionName);
+  @SuppressWarnings("Immutable") // Array not mutated
+  private final PlannedInterpretable[] args;
+
+  private final CelValueConverter celValueConverter;
+
+  static EvalAsyncCall create(
+      CelExpr expr,
+      String functionName,
+      CelResolvedOverload resolvedOverload,
+      CelAsyncFunctionOverload overload,
+      PlannedInterpretable[] args,
+      CelValueConverter celValueConverter) {
+    return new EvalAsyncCall(
+        expr, functionName, resolvedOverload, overload, args, celValueConverter);
   }
 
   @Override
   Object evalInternal(GlobalResolver resolver, ExecutionFrame frame) throws CelEvaluationException {
-    throw new CelEvaluationException(
-        String.format(
-            "Async function '%s' evaluated in synchronous mode. Asynchronous functions are only"
-                + " supported via evalAsync.",
-            functionName));
+    if (!frame.isAsync()) {
+      throw new CelEvaluationException(
+          String.format(
+              "Async function '%s' evaluated in synchronous mode. Asynchronous functions are only"
+                  + " supported via evalAsync.",
+              functionName));
+    }
+
+    Object[] evaluatedArgs = new Object[args.length];
+    AccumulatedUnknowns accumulatedUnknowns = null;
+
+    for (int i = 0; i < args.length; i++) {
+      Object argVal = evalStrictly(args[i], resolver, frame);
+      accumulatedUnknowns = AccumulatedUnknowns.maybeMerge(accumulatedUnknowns, argVal);
+      evaluatedArgs[i] = argVal;
+    }
+
+    if (accumulatedUnknowns != null) {
+      return accumulatedUnknowns;
+    }
+
+    if (!CelFunctionOverload.canHandle(
+        evaluatedArgs, resolvedOverload.getParameterTypes(), /* isStrict= */ true)) {
+      throw new LocalizedEvaluationException(
+          new CelOverloadNotFoundException(
+              functionName, ImmutableList.of(resolvedOverload.getOverloadId())),
+          expr().id());
+    }
+
+    try {
+      return frame
+          .asyncTracker()
+          .recordOrGet(
+              expr().id(),
+              functionName,
+              resolvedOverload.getOverloadId(),
+              evaluatedArgs,
+              overload,
+              celValueConverter);
+    } catch (CelRuntimeException e) {
+      throw new LocalizedEvaluationException(e, expr().id());
+    } catch (RuntimeException e) {
+      throw new LocalizedEvaluationException(e, CelErrorCode.INTERNAL_ERROR, expr().id());
+    }
   }
 
-  private EvalAsyncCall(CelExpr expr, String functionName) {
+  private EvalAsyncCall(
+      CelExpr expr,
+      String functionName,
+      CelResolvedOverload resolvedOverload,
+      CelAsyncFunctionOverload overload,
+      PlannedInterpretable[] args,
+      CelValueConverter celValueConverter) {
     super(expr);
     this.functionName = checkNotNull(functionName);
+    this.resolvedOverload = checkNotNull(resolvedOverload);
+    this.overload = checkNotNull(overload);
+    this.args = checkNotNull(args);
+    this.celValueConverter = checkNotNull(celValueConverter);
   }
 }

runtime/src/main/java/dev/cel/runtime/planner/EvalLateBoundCall.java

@@ -21,6 +21,7 @@
 import dev.cel.common.exceptions.CelOverloadNotFoundException;
 import dev.cel.common.values.CelValueConverter;
 import dev.cel.runtime.AccumulatedUnknowns;
+import dev.cel.runtime.CelAsyncFunctionOverload;
 import dev.cel.runtime.CelEvaluationException;
 import dev.cel.runtime.CelResolvedOverload;
 import dev.cel.runtime.GlobalResolver;
@@ -56,6 +57,14 @@ Object evalInternal(GlobalResolver resolver, ExecutionFrame frame) throws CelEva
             .findOverload(functionName, overloadIds, argVals)
             .orElseThrow(() -> new CelOverloadNotFoundException(functionName, overloadIds));
 
+    if (resolvedOverload.getDefinition() instanceof CelAsyncFunctionOverload) {
+      throw new CelEvaluationException(
+          String.format(
+              "Async function '%s' cannot be late-bound. Late-bound functions must be"
+                  + " synchronous.",
+              functionName));
+    }
+
     return EvalHelpers.dispatch(functionName, resolvedOverload, celValueConverter, argVals);
   }