/*

 * Copyright 2021 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.

 */

#include "eval/compiler/flat_expr_builder.h"

#include <algorithm>

#include <cstddef>

#include <cstdint>

#include <deque>

#include <iterator>

#include <limits>

#include <memory>

#include <optional>

#include <stack>

#include <string>

#include <type_traits>

#include <utility>

#include <vector>

#include "absl/algorithm/container.h"

#include "absl/base/nullability.h"

#include "absl/container/flat_hash_map.h"

#include "absl/container/flat_hash_set.h"

#include "absl/container/node_hash_map.h"

#include "absl/functional/any_invocable.h"

#include "absl/log/absl_check.h"

#include "absl/log/check.h"

#include "absl/status/status.h"

#include "absl/status/statusor.h"

#include "absl/strings/match.h"

#include "absl/strings/numbers.h"

#include "absl/strings/str_cat.h"

#include "absl/strings/string_view.h"

#include "absl/strings/strip.h"

#include "absl/types/span.h"

#include "absl/types/variant.h"

#include "base/ast.h"

#include "base/builtins.h"

#include "base/type_provider.h"

#include "common/ast.h"

#include "common/ast_traverse.h"

#include "common/ast_visitor.h"

#include "common/constant.h"

#include "common/expr.h"

#include "common/kind.h"

#include "common/type.h"

#include "common/type_spec_resolver.h"

#include "common/value.h"

#include "eval/compiler/check_ast_extensions.h"

#include "eval/compiler/flat_expr_builder_extensions.h"

#include "eval/compiler/resolver.h"

#include "eval/eval/comprehension_step.h"

#include "eval/eval/container_access_step.h"

#include "eval/eval/create_list_step.h"

#include "eval/eval/create_map_step.h"

#include "eval/eval/create_struct_step.h"

#include "eval/eval/evaluator_core.h"

#include "eval/eval/expression_step_logic.h"

#include "eval/eval/function_step.h"

#include "eval/eval/optional_or_step.h"

#include "eval/eval/select_step.h"

#include "eval/eval/shadowable_value_step.h"

#include "eval/eval/ternary_step.h"

#include "internal/status_macros.h"

#include "runtime/internal/convert_constant.h"

#include "runtime/internal/issue_collector.h"

#include "runtime/runtime_issue.h"

#include "runtime/runtime_options.h"

#include "runtime/type_registry.h"

#include "google/protobuf/arena.h"

namespace google::api::expr::runtime {

namespace {

using ::cel::Ast;

using ::cel::AstTraverse;

using ::cel::RuntimeIssue;

using ::cel::Value;

using ::cel::runtime_internal::ConvertConstant;

using ::cel::runtime_internal::GetLegacyRuntimeTypeProvider;

using ::cel::runtime_internal::GetRuntimeTypeProvider;

using ::cel::runtime_internal::IssueCollector;

constexpr absl::string_view kOptionalOrFn = "or";

constexpr absl::string_view kOptionalOrValueFn = "orValue";

constexpr absl::string_view kBlock = "cel.@block";

// Forward declare to resolve circular dependency for short_circuiting visitors.

class FlatExprVisitor;

// Helper for bookkeeping variables mapped to indexes.

class IndexManager {

 public:

  IndexManager() : next_free_slot_(0), max_slot_count_(0) {}

  size_t ReserveSlots(size_t n) {

    size_t result = next_free_slot_;

    next_free_slot_ += n;

    if (next_free_slot_ > max_slot_count_) {

      max_slot_count_ = next_free_slot_;

    }

    return result;

  }

  size_t ReleaseSlots(size_t n) {

    next_free_slot_ -= n;

    return next_free_slot_;

  }

  size_t max_slot_count() const { return max_slot_count_; }

 private:

  size_t next_free_slot_;

  size_t max_slot_count_;

};

// Helper for computing jump offsets.

//

// Jumps should be self-contained to a single expression node -- jumping

// outside that range is a bug.

struct ProgramStepIndex {

  int index;

  ProgramBuilder::Subexpression* subexpression;

};

// A convenience wrapper for offset-calculating logic.

class Jump {

 public:

  static absl::StatusOr<int> CalculateOffset(ProgramStepIndex base,

                                             ProgramStepIndex target) {

    if (target.subexpression != base.subexpression) {

      return absl::InternalError(

          "Jump target must be contained in the parent"

          "subexpression");

    }

    int offset = base.subexpression->CalculateOffset(base.index, target.index);

    // The offset may be packed into a 31-bit int field. Technically, the offset

    // could be larger for a well-formed expression with more than 2**30 steps,

    // but such a case is unlikely to exist at this time.

    constexpr int kMaxJumpOffset = (1 << 30) - 1;

    constexpr int kMinJumpOffset = -(1 << 30);

    if (offset < kMinJumpOffset || offset > kMaxJumpOffset) {

      return absl::InternalError("Jump offset exceeds supported range: " +

                                 absl::StrCat(offset));

    }

    return offset;

  }

};

class CondVisitor {

 public:

  virtual ~CondVisitor() = default;

  virtual void PreVisit(const cel::Expr* expr) = 0;

  virtual void PostVisitArg(int arg_num, const cel::Expr* expr) = 0;

  virtual void PostVisit(const cel::Expr* expr) = 0;

  virtual void PostVisitTarget(const cel::Expr* expr) {}

};

// Visitor managing the "&&" and "||" (boolean logic) operations.

// Implements short-circuiting if enabled.

//

// With short-circuiting enabled, generates a program like:

//   +-------------+------------------------+-----------------------+

//   | PC          | Step                   | Stack                 |

//   +-------------+------------------------+-----------------------+

//   | i + 0       | <Arg1>                 | arg1                  |

//   | i + 1       | ConditionalJump i + 4  | arg1                  |

//   | i + 2       | <Arg2>                 | arg1, arg2            |

//   | i + 3       | BooleanOperator        | Op(arg1, arg2)        |

//   | i + 4       | <rest of program>      | arg1 | Op(arg1, arg2) |

//   +-------------+------------------------+------------------------+

class LogicalCondVisitor : public CondVisitor {

 public:

  explicit LogicalCondVisitor(FlatExprVisitor* visitor, bool is_or,

                              bool short_circuiting)

      : visitor_(visitor), is_or_(is_or), short_circuiting_(short_circuiting) {}

  void PreVisit(const cel::Expr* expr) override;

  void PostVisitArg(int arg_num, const cel::Expr* expr) override;

  void PostVisit(const cel::Expr* expr) override;

 private:

  FlatExprVisitor* visitor_;

  const bool is_or_;

  std::vector<ProgramStepIndex> jump_steps_;

  bool short_circuiting_;

};

// Visitor managing optional "or" and "orValue" operations.

// Implements short-circuiting if enabled.

class OptionalOrCondVisitor : public CondVisitor {

 public:

  explicit OptionalOrCondVisitor(FlatExprVisitor* visitor, bool is_or_value,

                                 bool short_circuiting)

      : visitor_(visitor),

        is_or_value_(is_or_value),

        short_circuiting_(short_circuiting) {}

  void PreVisit(const cel::Expr* expr) override;

  void PostVisitArg(int arg_num, const cel::Expr* expr) override {}

  void PostVisitTarget(const cel::Expr* expr) override;

  void PostVisit(const cel::Expr* expr) override;

 private:

  FlatExprVisitor* visitor_;

  const bool is_or_value_;

  OptionalHasValueJumpStep* jump_step_ = nullptr;

  ProgramStepIndex jump_step_pos_{-1, nullptr};

  bool short_circuiting_;

};

class TernaryCondVisitor : public CondVisitor {

 public:

  explicit TernaryCondVisitor(FlatExprVisitor* visitor) : visitor_(visitor) {}

  void PreVisit(const cel::Expr* expr) override;

  void PostVisitArg(int arg_num, const cel::Expr* expr) override;

  void PostVisit(const cel::Expr* expr) override;

 private:

  FlatExprVisitor* visitor_;

  std::optional<ProgramStepIndex> cond_jump_pos_;

  std::optional<ProgramStepIndex> jump_after_first_pos_;

};

class ExhaustiveTernaryCondVisitor : public CondVisitor {

 public:

  explicit ExhaustiveTernaryCondVisitor(FlatExprVisitor* visitor)

      : visitor_(visitor) {}

  void PreVisit(const cel::Expr* expr) override;

  void PostVisitArg(int arg_num, const cel::Expr* expr) override {}

  void PostVisit(const cel::Expr* expr) override;

 private:

  FlatExprVisitor* visitor_;

};

// Returns a hint for the number of program nodes (steps or subexpressions) that

// will be created for this expr.

size_t SizeHint(const cel::Expr& expr) {

  switch (expr.kind_case()) {

    case cel::ExprKindCase::kConstant:

      return 1;

    case cel::ExprKindCase::kIdentExpr:

      return 1;

    case cel::ExprKindCase::kSelectExpr:

      return 2;

    case cel::ExprKindCase::kCallExpr:

      return expr.call_expr().args().size() +

             (expr.call_expr().has_target() ? 2 : 1);

    case cel::ExprKindCase::kListExpr:

      return expr.list_expr().elements().size() + 1;

    case cel::ExprKindCase::kStructExpr:

      return expr.struct_expr().fields().size() + 1;

    case cel::ExprKindCase::kMapExpr:

      return 2 * expr.struct_expr().fields().size() + 1;

    default:

      return 1;

  }

  return 0;

}

// Returns whether this comprehension appears to be a standard map/filter

// macro implementation. It is not exhaustive, so it is unsafe to use with

// custom comprehensions outside of the standard macros or hand crafted ASTs.

bool IsOptimizableListAppend(const cel::ComprehensionExpr* comprehension,

                             bool enable_comprehension_list_append) {

  if (!enable_comprehension_list_append) {

    return false;

  }

  absl::string_view accu_var = comprehension->accu_var();

  if (accu_var.empty() ||

      comprehension->result().ident_expr().name() != accu_var) {

    return false;

  }

  if (!comprehension->accu_init().has_list_expr() ||

      !comprehension->accu_init().list_expr().elements().empty()) {

    return false;

  }

  if (!comprehension->loop_step().has_call_expr()) {

    return false;

  }

  // Macro loop_step for a filter() will contain a ternary:

  //   filter ? accu_var + [elem] : accu_var

  // Macro loop_step for a map() will contain a list concat operation:

  //   accu_var + [elem]

  const auto* call_expr = &comprehension->loop_step().call_expr();

  if (call_expr->function() == cel::builtin::kTernary &&

      call_expr->args().size() == 3) {

    if (!call_expr->args()[1].has_call_expr()) {

      return false;

    }

    call_expr = &(call_expr->args()[1].call_expr());

  }

  return call_expr->function() == cel::builtin::kAdd &&

         call_expr->args().size() == 2 &&

         call_expr->args()[0].has_ident_expr() &&

         call_expr->args()[0].ident_expr().name() == accu_var &&

         call_expr->args()[1].has_list_expr() &&

         call_expr->args()[1].list_expr().elements().size() == 1;

}

// Assuming `IsOptimizableListAppend()` return true, return a pointer to the

// call `accu_var + [elem]`.

const cel::CallExpr* GetOptimizableListAppendCall(

    const cel::ComprehensionExpr* comprehension) {

  ABSL_DCHECK(IsOptimizableListAppend(

      comprehension, /*enable_comprehension_list_append=*/true));

  // Macro loop_step for a filter() will contain a ternary:

  //   filter ? accu_var + [elem] : accu_var

  // Macro loop_step for a map() will contain a list concat operation:

  //   accu_var + [elem]

  const auto* call_expr = &comprehension->loop_step().call_expr();

  if (call_expr->function() == cel::builtin::kTernary &&

      call_expr->args().size() == 3) {

    call_expr = &(call_expr->args()[1].call_expr());

  }

  return call_expr;

}

// Assuming `IsOptimizableListAppend()` return true, return a pointer to the

// node `[elem]`.

const cel::Expr* GetOptimizableListAppendOperand(

    const cel::ComprehensionExpr* comprehension) {

  return &GetOptimizableListAppendCall(comprehension)->args()[1];

}

// Returns whether this comprehension appears to be a macro implementation for

// map transformations. It is not exhaustive, so it is unsafe to use with custom

// comprehensions outside of the standard macros or hand crafted ASTs.

bool IsOptimizableMapInsert(const cel::ComprehensionExpr* comprehension,

                            bool enable_comprehension_mutable_map) {

  if (!enable_comprehension_mutable_map) {

    return false;

  }

  if (comprehension->iter_var().empty() || comprehension->iter_var2().empty()) {

    return false;

  }

  absl::string_view accu_var = comprehension->accu_var();

  if (accu_var.empty() || !comprehension->has_result() ||

      !comprehension->result().has_ident_expr() ||

      comprehension->result().ident_expr().name() != accu_var) {

    return false;

  }

  if (!comprehension->accu_init().has_map_expr()) {

    return false;

  }

  if (!comprehension->loop_step().has_call_expr()) {

    return false;

  }

  const auto* call_expr = &comprehension->loop_step().call_expr();

  if (call_expr->function() == cel::builtin::kTernary &&

      call_expr->args().size() == 3) {

    if (!call_expr->args()[1].has_call_expr()) {

      return false;

    }

    call_expr = &(call_expr->args()[1].call_expr());

  }

  return call_expr->function() == "cel.@mapInsert" &&

         (call_expr->args().size() == 2 || call_expr->args().size() == 3) &&

         call_expr->args()[0].has_ident_expr() &&

         call_expr->args()[0].ident_expr().name() == accu_var;

}

bool IsBind(const cel::ComprehensionExpr* comprehension) {

  static constexpr absl::string_view kUnusedIterVar = "#unused";

  return comprehension->loop_condition().const_expr().has_bool_value() &&

         comprehension->loop_condition().const_expr().bool_value() == false &&

         comprehension->iter_var() == kUnusedIterVar &&

         comprehension->iter_var2().empty() &&

         comprehension->iter_range().has_list_expr() &&

         comprehension->iter_range().list_expr().elements().empty();

}

bool IsBlock(const cel::CallExpr* call) { return call->function() == kBlock; }

// Visitor for Comprehension expressions.

class ComprehensionVisitor {

 public:

  explicit ComprehensionVisitor(FlatExprVisitor* visitor, bool is_trivial,

                                size_t iter_slot, size_t iter2_slot,

                                size_t accu_slot)

      : visitor_(visitor),

        init_step_(nullptr),

        is_trivial_(is_trivial),

        accu_init_extracted_(false),

        iter_slot_(iter_slot),

        iter2_slot_(iter2_slot),

        accu_slot_(accu_slot) {}

  void PreVisit(const cel::Expr* expr);

  absl::Status PostVisitArg(cel::ComprehensionArg arg_num,

                            const cel::Expr* comprehension_expr) {

    if (is_trivial_) {

      PostVisitArgTrivial(arg_num, comprehension_expr);

      return absl::OkStatus();

    } else {

      return PostVisitArgDefault(arg_num, comprehension_expr);

    }

  }

  void PostVisit(const cel::Expr* expr);

  void MarkAccuInitExtracted() { accu_init_extracted_ = true; }

 private:

  void PostVisitArgTrivial(cel::ComprehensionArg arg_num,

                           const cel::Expr* comprehension_expr);

  absl::Status PostVisitArgDefault(cel::ComprehensionArg arg_num,

                                   const cel::Expr* comprehension_expr);

  ComprehensionCondStep* absl_nullable GetCondStep();

  ComprehensionNextStep* absl_nullable GetNextStep();

  FlatExprVisitor* visitor_;

  ComprehensionInitStep* init_step_;

  ProgramStepIndex init_step_pos_;

  std::optional<ProgramStepIndex> next_step_pos_;

  std::optional<ProgramStepIndex> cond_step_pos_;

  bool is_trivial_;

  bool accu_init_extracted_;

  size_t iter_slot_;

  size_t iter2_slot_;

  size_t accu_slot_;

};

absl::flat_hash_set<int32_t> MakeOptionalIndicesSet(

    const cel::StructExpr& create_struct_expr) {

  absl::flat_hash_set<int32_t> optional_indices;

  for (size_t i = 0; i < create_struct_expr.fields().size(); ++i) {

    if (create_struct_expr.fields()[i].optional()) {

      optional_indices.insert(static_cast<int32_t>(i));

    }

  }

  return optional_indices;

}

absl::flat_hash_set<int32_t> MakeOptionalIndicesSet(

    const cel::MapExpr& map_expr) {

  absl::flat_hash_set<int32_t> optional_indices;

  for (size_t i = 0; i < map_expr.entries().size(); ++i) {

    if (map_expr.entries()[i].optional()) {

      optional_indices.insert(static_cast<int32_t>(i));

    }

  }

  return optional_indices;

}

class FlatExprVisitor : public cel::AstVisitor {

 public:

  enum class CallHandlerResult {

    // The call was intercepted, no additional processing is needed.

    kIntercepted,

    // The call was not intercepted, continue with the default processing.

    kNotIntercepted,

  };

  // Handler for functions with builtin implementations.

  // This is used to replace the usual dispatcher step that applies

  // the arguments to a candidate function from the function registry.

  using CallHandler = absl::AnyInvocable<CallHandlerResult(

      const cel::Expr&, const cel::CallExpr&)>;

  FlatExprVisitor(

      const Resolver& resolver, const cel::RuntimeOptions& options,

      std::vector<std::unique_ptr<ProgramOptimizer>> program_optimizers,

      const absl::flat_hash_map<int64_t, cel::TypeSpec>& type_map,

      const cel::TypeProvider& type_provider, IssueCollector& issue_collector,

      ProgramBuilder& program_builder, PlannerContext& extension_context,

      bool enable_optional_types)

      : resolver_(resolver),

        type_provider_(type_provider),

        progress_status_(absl::OkStatus()),

        resolved_select_expr_(nullptr),

        options_(options),

        program_optimizers_(std::move(program_optimizers)),

        type_map_(type_map),

        issue_collector_(issue_collector),

        program_builder_(program_builder),

        extension_context_(extension_context),

        enable_optional_types_(enable_optional_types) {

    constexpr size_t kCallHandlerSizeHint = 11;

    call_handlers_.reserve(kCallHandlerSizeHint);

    call_handlers_[cel::builtin::kIndex] = [this](const cel::Expr& expr,

                                                  const cel::CallExpr& call) {

      return HandleIndex(expr, call);

    };

    call_handlers_[kBlock] = [this](const cel::Expr& expr,

                                    const cel::CallExpr& call) {

      return HandleBlock(expr, call);

    };

    call_handlers_[cel::builtin::kAdd] = [this](const cel::Expr& expr,

                                                const cel::CallExpr& call) {

      return HandleListAppend(expr, call);

    };

    if (options_.enable_fast_builtins) {

      call_handlers_[cel::builtin::kNotStrictlyFalse] =

          [this](const cel::Expr& expr, const cel::CallExpr& call) {

            return HandleNotStrictlyFalse(expr, call);

          };

      call_handlers_[cel::builtin::kNotStrictlyFalseDeprecated] =

          [this](const cel::Expr& expr, const cel::CallExpr& call) {

            return HandleNotStrictlyFalse(expr, call);

          };

      call_handlers_[cel::builtin::kNot] = [this](const cel::Expr& expr,

                                                  const cel::CallExpr& call) {

        return HandleNot(expr, call);

      };

      if (options_.enable_heterogeneous_equality) {

        for (const auto& in_op :

             {cel::builtin::kIn, cel::builtin::kInDeprecated,

              cel::builtin::kInFunction}) {

          call_handlers_[in_op] = [this](const cel::Expr& expr,

                                         const cel::CallExpr& call) {

            return HandleHeterogeneousEqualityIn(expr, call);

          };

        }

        // Try to detect if the environment is setup with a custom equality

        // implementation.

        if (resolver_

                .FindOverloads(cel::builtin::kEqual,

                               /*receiver_style=*/false,

                               {cel::Kind::kAny, cel::Kind::kAny})

                .empty()) {

          call_handlers_[cel::builtin::kEqual] =

              [this](const cel::Expr& expr, const cel::CallExpr& call) {

                return HandleHeterogeneousEquality(expr, call,

                                                   /*inequality=*/false);

              };

          call_handlers_[cel::builtin::kInequal] =

              [this](const cel::Expr& expr, const cel::CallExpr& call) {

                return HandleHeterogeneousEquality(expr, call,

                                                   /*inequality=*/true);

              };

        }

      }

    }

  }

  void SetResolvedType(const cel::Expr& expr, cel::Type type) {

    resolved_types_[&expr] = std::move(type);

  }

  std::optional<cel::Type> GetResolvedType(const cel::Expr* expr) const {

    if (expr == nullptr) {

      return std::nullopt;

    }

    auto it = resolved_types_.find(expr);

    if (it != resolved_types_.end()) {

      return it->second;

    }

    return std::nullopt;

  }

  void PreVisitExpr(const cel::Expr& expr) override {

    ValidateOrError(!absl::holds_alternative<cel::UnspecifiedExpr>(expr.kind()),

                    "Invalid empty expression");

    if (!progress_status_.ok()) {

      return;

    }

    if (resume_from_suppressed_branch_ == nullptr &&

        suppressed_branches_.find(&expr) != suppressed_branches_.end()) {

      resume_from_suppressed_branch_ = &expr;

    }

    if (options_.enable_typed_field_access) {

      MaybeResolveType(expr);

    }

    if (block_.has_value()) {

      BlockInfo& block = *block_;

      if (block.in && block.bindings_set.contains(&expr)) {

        block.current_binding = &expr;

      }

    }

    auto* subexpression =

        program_builder_.EnterSubexpression(&expr, SizeHint(expr));

    if (subexpression == nullptr) {

      progress_status_.Update(

          absl::InternalError("same CEL expr visited twice"));

      return;

    }

    for (const std::unique_ptr<ProgramOptimizer>& optimizer :

         program_optimizers_) {

      absl::Status status = optimizer->OnPreVisit(extension_context_, expr);

      if (!status.ok()) {

        SetProgressStatusIfError(status);

      }

    }

  }

  void PostVisitExpr(const cel::Expr& expr) override {

    if (!progress_status_.ok()) {

      return;

    }

    if (&expr == resume_from_suppressed_branch_) {

      resume_from_suppressed_branch_ = nullptr;

    }

    for (const std::unique_ptr<ProgramOptimizer>& optimizer :

         program_optimizers_) {

      absl::Status status = optimizer->OnPostVisit(extension_context_, expr);

      if (!status.ok()) {

        SetProgressStatusIfError(status);

        return;

      }

    }

    program_builder_.ExitSubexpression(&expr);

    if (!comprehension_stack_.empty() &&

        comprehension_stack_.back().is_optimizable_bind &&

        (&comprehension_stack_.back().comprehension->accu_init() == &expr)) {

      SetProgressStatusIfError(

          MaybeExtractSubexpression(&expr, comprehension_stack_.back()));

    }

    if (block_.has_value()) {

      BlockInfo& block = *block_;

      if (block.current_binding == &expr) {

        int index = program_builder_.ExtractSubexpression(&expr);

        if (index == -1) {

          SetProgressStatusIfError(

              absl::InvalidArgumentError("failed to extract subexpression"));

          return;

        }

        block.subexpressions[block.current_index++] = index;

        block.current_binding = nullptr;

      }

    }

  }

  void PostVisitConst(const cel::Expr& expr,

                      const cel::Constant& const_expr) override {

    if (!progress_status_.ok()) {

      return;

    }

    absl::StatusOr<cel::Value> converted_value =

        ConvertConstant(const_expr, extension_context_.MutableArena());

    if (!converted_value.ok()) {

      SetProgressStatusIfError(converted_value.status());

      return;

    }

    AddStep(ExpressionStep::MakeConstant(std::move(converted_value).value(),

                                         expr.id()));

  }

  struct SlotLookupResult {

    int slot;

    int subexpression;

  };

  // Helper to lookup a variable mapped to a slot.

  //

  // If lazy evaluation enabled and ided as a lazy expression,

  // subexpression and slot will be set.

  SlotLookupResult LookupSlot(absl::string_view path) {

    // If there's a leading dot, it cannot resolve to a local variable.

    if (absl::StartsWith(path, ".")) {

      return {-1, -1};

    }

    if (block_.has_value()) {

      const BlockInfo& block = *block_;

      if (block.in) {

        absl::string_view index_suffix = path;

        if (absl::ConsumePrefix(&index_suffix, "@index")) {

          size_t index;

          if (!absl::SimpleAtoi(index_suffix, &index)) {

            SetProgressStatusIfError(

                issue_collector_.AddIssue(RuntimeIssue::CreateError(

                    absl::InvalidArgumentError("bad @index"))));

            return {-1, -1};

          }

          if (index >= block.size) {

            SetProgressStatusIfError(

                issue_collector_.AddIssue(RuntimeIssue::CreateError(

                    absl::InvalidArgumentError(absl::StrCat(

                        "invalid @index greater than number of bindings: ",

                        index, " >= ", block.size)))));

            return {-1, -1};

          }

          if (index >= block.current_index) {

            SetProgressStatusIfError(

                issue_collector_.AddIssue(RuntimeIssue::CreateError(

                    absl::InvalidArgumentError(absl::StrCat(

                        "@index references current or future binding: ", index,

                        " >= ", block.current_index)))));

            return {-1, -1};

          }

          return {static_cast<int>(block.index + index),

                  block.subexpressions[index]};

        }

      }

    }

    if (!comprehension_stack_.empty()) {

      for (int i = comprehension_stack_.size() - 1; i >= 0; i--) {

        const ComprehensionStackRecord& record = comprehension_stack_[i];

        if (record.iter_var_in_scope &&

            record.comprehension->iter_var() == path) {

          if (record.is_optimizable_bind) {

            SetProgressStatusIfError(issue_collector_.AddIssue(

                RuntimeIssue::CreateWarning(absl::InvalidArgumentError(

                    "Unexpected iter_var access in trivial comprehension"))));

            return {-1, -1};

          }

          return {static_cast<int>(record.iter_slot), -1};

        }

        if (record.iter_var2_in_scope &&

            record.comprehension->iter_var2() == path) {

          return {static_cast<int>(record.iter2_slot), -1};

        }

        if (record.accu_var_in_scope &&

            record.comprehension->accu_var() == path) {

          int slot = record.accu_slot;

          int subexpression = -1;

          if (record.is_optimizable_bind) {

            subexpression = record.subexpression;

          }

          return {slot, subexpression};

        }

      }

    }

    if (absl::StartsWith(path, "@it:") || absl::StartsWith(path, "@it2:") ||

        absl::StartsWith(path, "@ac:")) {

      // If we see a CSE generated comprehension variable that was not

      // resolvable through the normal comprehension scope resolution, reject it

      // now rather than surfacing errors at activation time.

      SetProgressStatusIfError(

          issue_collector_.AddIssue(RuntimeIssue::CreateError(

              absl::InvalidArgumentError("out of scope reference to CSE "

                                         "generated comprehension variable"))));

    }

    return {-1, -1};

  }

  // Ident node handler.

  // Invoked after child nodes are processed.

  void PostVisitIdent(const cel::Expr& expr,

                      const cel::IdentExpr& ident_expr) override {

    if (!progress_status_.ok()) {

      return;

    }

    absl::string_view path = ident_expr.name();

    if (!ValidateOrError(

            !path.empty(),

            "Invalid expression: identifier 'name' must not be empty")) {

      return;

    }

    // Check if this is a local variable first (since it should shadow most

    // other interpretations).

    SlotLookupResult slot = LookupSlot(path);

    if (slot.subexpression >= 0) {

      auto* subexpression =

          program_builder_.GetExtractedSubexpression(slot.subexpression);

      if (subexpression == nullptr) {

        SetProgressStatusIfError(

            absl::InternalError("bad subexpression reference"));

        return;

      }

        // Off by one since mainline expression will be index 0.

        AddStep(ExpressionStep::MakeLazyInitStep(

            slot.slot, slot.subexpression + 1, expr.id()));

      return;

    } else if (slot.slot >= 0) {

        AddStep(ExpressionStep::MakeReadSlotStep(slot.slot, expr.id()));

      return;

    }

    // Attempt to resolve a select expression as a namespaced identifier for an

    // enum or type constant value.

    std::optional<cel::Value> const_value;

    int64_t select_root_id = -1;

    std::string path_candidate;

    while (!namespace_stack_.empty()) {

      const auto& select_node = namespace_stack_.front();

      // Generate path in format "<ident>.<field 0>.<field 1>...".

      const cel::Expr* select_expr = select_node.first;

      path_candidate = absl::StrCat(path, ".", select_node.second);

      // Attempt to find a constant enum or type value which matches the

      // qualified path present in the expression. Whether the identifier

      // can be resolved to a type instance depends on whether the option to

      // 'enable_qualified_type_identifiers' is set to true.

      const_value = resolver_.FindConstant(path_candidate, select_expr->id());

      if (const_value) {

        resolved_select_expr_ = select_expr;

        select_root_id = select_expr->id();

        path = path_candidate;

        namespace_stack_.clear();

        break;

      }

      namespace_stack_.pop_front();

    }

    if (!const_value) {

      // Attempt to resolve a simple identifier as an enum or type constant

      // value.

      const_value = resolver_.FindConstant(path, expr.id());

      select_root_id = expr.id();

    }

    // TODO(issues/97): Need to add support for resolving packaged names at

    // runtime if Parse-only. For checked, checker should have reported the

    // expected interpretation.

    if (const_value) {

      // If the path starts with a dot, strip it.

      absl::string_view name = absl::StripPrefix(path, ".");

      AddStep(CreateShadowableValueStep(name, std::move(const_value).value()),

              select_root_id);

      return;

    }

    absl::string_view ident_name = absl::StripPrefix(ident_expr.name(), ".");

      AddStep(ExpressionStep::MakeIdentifierStep(ident_name, expr.id()));

  }

  void PreVisitSelect(const cel::Expr& expr,

                      const cel::SelectExpr& select_expr) override {

    if (!progress_status_.ok()) {

      return;

    }

    if (!ValidateOrError(

            !select_expr.field().empty(),

            "invalid expression: select 'field' must not be empty")) {

      return;

    }

    if (!ValidateOrError(

            select_expr.has_operand() &&

                select_expr.operand().kind_case() !=

                    cel::ExprKindCase::kUnspecifiedExpr,

            "invalid expression: select must specify an operand")) {

      return;

    }

    // Not exactly the cleanest solution - we peek into child of

    // select_expr.

    // Chain of multiple SELECT ending with IDENT can represent namespaced

    // entity.

    if (!select_expr.test_only() && (select_expr.operand().has_ident_expr() ||

                                     select_expr.operand().has_select_expr())) {

      // select expressions are pushed in reverse order:

      // google.type.Expr is pushed as:

      // - field: 'Expr'

      // - field: 'type'

      // - id: 'google'

      //

      // The search order though is as follows:

      // - id: 'google.type.Expr'

      // - id: 'google.type', field: 'Expr'

      // - id: 'google', field: 'type', field: 'Expr'

      for (size_t i = 0; i < namespace_stack_.size(); i++) {

        auto ns = namespace_stack_[i];

        namespace_stack_[i] = {

            ns.first, absl::StrCat(select_expr.field(), ".", ns.second)};

      }

      namespace_stack_.push_back({&expr, select_expr.field()});

    } else {

      namespace_stack_.clear();

    }

  }

  // Select node handler.

  // Invoked after child nodes are processed.

  void PostVisitSelect(const cel::Expr& expr,

                       const cel::SelectExpr& select_expr) override {

    if (!progress_status_.ok()) {

      return;

    }

    // Check if we are "in the middle" of namespaced name.

    // This is currently enum specific. Constant expression that corresponds

    // to resolved enum value has been already created, thus preceding chain

    // of selects is no longer relevant.

    if (resolved_select_expr_) {

      if (&expr == resolved_select_expr_) {

        resolved_select_expr_ = nullptr;

      }

      return;

    }

    std::string field = select_expr.field();

    std::optional<cel::StructType> struct_type;

    std::optional<cel::StructTypeField> field_type;

    if (options_.enable_typed_field_access) {

      std::optional<cel::Type> operand_type =

          GetResolvedType(&select_expr.operand());

      if (operand_type.has_value() && operand_type->IsStruct()) {

        struct_type = operand_type->GetStruct();

        if (struct_type.has_value()) {

          auto field_lookup =

              extension_context_.type_reflector().FindStructTypeFieldByName(

                  *struct_type, select_expr.field());

          // Swallow error to fallback to duck typing behavior.

          if (field_lookup.ok() && field_lookup->has_value()) {

            field_type = *std::move(field_lookup);

          }

        }

      }

    }

    if (field_type.has_value()) {

      AddStep(

          CreateTypedSelectStep(std::move(field), *struct_type,

                                *std::move(field_type), select_expr.test_only(),

                                options_.enable_empty_wrapper_null_unboxing,

                                enable_optional_types_),

          expr.id());

      return;

    }

    AddStep(CreateSelectStep(std::move(field), select_expr.test_only(),

                             options_.enable_empty_wrapper_null_unboxing,

                             enable_optional_types_),

            expr.id());

  }

  // Call node handler group.

  // We provide finer granularity for Call node callbacks to allow special

  // handling for short-circuiting

  // PreVisitCall is invoked before child nodes are processed.

  void PreVisitCall(const cel::Expr& expr,

                    const cel::CallExpr& call_expr) override {

    if (!progress_status_.ok()) {

      return;

    }

    std::unique_ptr<CondVisitor> cond_visitor;

    if (call_expr.function() == cel::builtin::kAnd) {

      cond_visitor = std::make_unique<LogicalCondVisitor>(

          this, /*is_or=*/false, options_.short_circuiting);

    } else if (call_expr.function() == cel::builtin::kOr) {

      cond_visitor = std::make_unique<LogicalCondVisitor>(

          this, /*is_or=*/true, options_.short_circuiting);

    } else if (call_expr.function() == cel::builtin::kTernary) {

      if (options_.short_circuiting) {

        cond_visitor = std::make_unique<TernaryCondVisitor>(this);

      } else {

        cond_visitor = std::make_unique<ExhaustiveTernaryCondVisitor>(this);

      }

    } else if (enable_optional_types_ &&

               call_expr.function() == kOptionalOrFn &&

               call_expr.has_target() && call_expr.args().size() == 1) {

      cond_visitor = std::make_unique<OptionalOrCondVisitor>(

          this, /*is_or_value=*/false, options_.short_circuiting);

    } else if (enable_optional_types_ &&

               call_expr.function() == kOptionalOrValueFn &&

               call_expr.has_target() && call_expr.args().size() == 1) {

      cond_visitor = std::make_unique<OptionalOrCondVisitor>(

          this, /*is_or_value=*/true, options_.short_circuiting);

    } else if (IsBlock(&call_expr)) {

      // cel.@block

      if (block_.has_value()) {

        // There can only be one for now.

        SetProgressStatusIfError(

            absl::InvalidArgumentError("multiple cel.@block are not allowed"));

        return;

      }

      block_ = BlockInfo();

      BlockInfo& block = *block_;

      block.in = true;

      if (call_expr.args().empty()) {