path_runtime.h

#pragma once
 
#include <tess/core/assert.h>
#include <tess/core/fail_fast.h>
#include <tess/path/field_product_cache.h>
#include <tess/path/path.h>
#include <tess/path/portal_route.h>
#include <tess/path/portal_segment_cache.h>
#include <tess/path/precheck.h>
 
#include <algorithm>
#include <cstddef>
#include <cstdint>
#include <limits>
#include <optional>
#include <span>
#include <utility>
#include <vector>
 
namespace tess {

struct PathTicket {
  std::size_t value = 0;
  std::uint64_t generation = 0;
};
 
namespace detail {
 
// Portal-first eligibility: the class must use the ordinary adjacent-step
// policy because chunk portals do not model custom regular-step rules.
template <typename Class>
struct portal_replan_tags {
  static constexpr bool eligible =
      std::derived_from<Class, movement::movement_class_tag> &&
      std::same_as<movement::step_policy_of<Class>, movement::DefaultSteps>;
};
 
}  // namespace detail
 
// How single-goal weighted replans are served.
//
// ExactAStar: the batch's singleton fallback runs raw weighted A*; results are
// optimal.
//
// PortalFirst: eligible singletons (dense orthogonal-lattice worlds,
// default adjacent transitions, and explicit movement classes — Manhattan
// is an admissible lower bound on that lattice, and the premium
// cap's guarantee rests on it) first try a chunk-portal route
// stitched through the runtime's segment cache. Accepted routes are legal
// and verified but may exceed the optimal cost, bounded by the premium
// cap below; every other outcome — no candidate, a failed segment, a cap
// rejection, or an ineligible request — falls back to exact A* with
// byte-identical results. A rejection costs the portal work PLUS the
// exact search, so the cap is a route-quality contract, not a latency
// bound; the stats record every outcome.
enum class WeightedReplanStrategy : std::uint8_t {
  ExactAStar,
  PortalFirst,
};

struct PathRuntimeCachePolicy {
  MissingChunkPolicy missing_chunk_policy =
      MissingChunkPolicy::ReportIndeterminate;
  // In ScopedFeasible staleness mode world changes do not advance this
  // deep-clear counter: its rationale — bounding staleness accumulated
  // behind the 64-bit fingerprint — is superseded by exact per-chunk
  // validation. Note the periodic clear it would have fired sweeps ALL
  // runtime caches (clear_caches), so scoped mode also retains
  // field-product and portal entries longer; both self-validate their
  // dependencies and stay budget-bounded.
  std::size_t clear_every_world_change = 0;
  bool invalidate_unit_route_cache_on_world_change = true;
  // Opt-in scoped staleness for the unit route cache (see
  // UnitRouteStaleness): surviving routes are legal with truthful cost and
  // were optimal when stored; an edit elsewhere can leave them suboptimal
  // until retired. Default preserves exact whole-world invalidation.
  UnitRouteStaleness unit_route_staleness = UnitRouteStaleness::WholeWorldExact;
  // Opt-in portal-first serving for single-goal weighted replans (see
  // WeightedReplanStrategy). The premium cap alpha = num/den accepts a
  // portal route only when its cost is at most alpha times the request's
  // Manhattan distance — an admissible lower bound on the optimal cost for
  // the eligible model class, so acceptance guarantees cost <= alpha x
  // optimal. num == 0 disables the cap (accept every verified route).
  WeightedReplanStrategy weighted_replan_strategy =
      WeightedReplanStrategy::ExactAStar;
  std::uint32_t portal_premium_limit_num = 4;
  std::uint32_t portal_premium_limit_den = 3;
  bool use_unit_field_product_cache = false;
  std::size_t unit_field_product_min_goal_reuse = 2;
  std::size_t unit_field_product_min_start_chunks = 2;
  // Unit and weighted products share one runtime cache. The latest processing
  // pass applies its corresponding budget to the combined retained footprint,
  // so lowering this value may evict weighted products too.
  std::size_t unit_field_product_cache_byte_budget =
      std::numeric_limits<std::size_t>::max();
  bool use_weighted_field_product_cache = false;
  std::size_t weighted_field_product_min_goal_reuse = 2;
  std::size_t weighted_field_product_min_start_chunks = 2;
  // See unit_field_product_cache_byte_budget: this pass likewise budgets the
  // shared cache and may evict retained unit products.
  std::size_t weighted_field_product_cache_byte_budget =
      std::numeric_limits<std::size_t>::max();
  std::size_t max_route_entries = UnitRouteCache::default_max_entries;
  std::size_t max_route_path_nodes = UnitRouteCache::default_max_path_nodes;
  // Scoped staleness only: total budget for stored (chunk, content version)
  // dependency pairs, with the same oversized-skip / cap-invalidate
  // lifecycle as the path-node cap.
  std::size_t max_route_dependency_pairs =
      UnitRouteCache::default_max_path_nodes / 8u;
  std::size_t portal_segment_budget =
      WeightedPortalSegmentCache::default_segment_budget;
};
 
// Portal-first single-goal replan accounting (see
// PathRuntimeCachePolicy::weighted_replan_strategy). The identity
// attempts == accepted + no_candidates + verification_failures +
// premium_rejections holds; exact_fallbacks counts every attempt that was
// ultimately served by exact A* (all non-accepted attempts).
struct WeightedPortalReplanStats {
  std::size_t attempts = 0;
  std::size_t accepted = 0;
  std::size_t no_candidates = 0;
  std::size_t verification_failures = 0;
  std::size_t premium_rejections = 0;
  std::size_t exact_fallbacks = 0;
  // Requests processed while PortalFirst was requested but the batch's
  // instantiation is ineligible (sparse world, custom provider, or
  // non-default-step class): they take the exact path without an
  // attempt, and this counter is what distinguishes a misconfigured policy
  // from a disabled one. Not part of the attempts identity above.
  std::size_t ineligible_fallbacks = 0;
};

struct PathRuntimeStats {
  std::size_t submitted = 0;
  std::size_t completed = 0;
  std::size_t found = 0;
  std::size_t invalid_start = 0;
  std::size_t invalid_goal = 0;
  std::size_t no_path = 0;
  std::size_t not_computed = 0;
  std::size_t no_candidate = 0;
  // Sparse worlds: the search could not rule out a route through a
  // non-resident chunk (PathStatus::Indeterminate). Kept distinct from
  // no_path so a stale/partial residency set is never counted as "no route".
  std::size_t indeterminate = 0;
  std::size_t cost_overflow = 0;
  // Requests an optional topology precheck proved unreachable before A*, so no
  // grid was expanded for them. A SUBSET of no_path (each ruled-out request is
  // also counted there): the result is the same NoPath A* would have returned,
  // this counter only measures how many were resolved without searching.
  std::size_t precheck_ruled_out = 0;
  std::size_t world_cache_invalidations = 0;
  // Unit-cache clears forced by processing with a different movement class
  // than the runtime was last bound to (see process_unit_cached). Correct but
  // wasteful: keep one runtime per (world, class) to stay at zero.
  std::size_t class_cache_invalidations = 0;
  std::size_t cache_clears = 0;
  std::size_t path_nodes = 0;
  UnitRouteCacheStats route_cache{};
  FieldProductCacheStats field_product_cache{};
  std::size_t field_product_candidate_groups = 0;
  std::size_t field_product_used_groups = 0;
  std::size_t field_product_skipped_groups = 0;
  WeightedPathBatchStats weighted_batch{};
  PortalSegmentCacheStats portal_segment_cache{};
  WeightedPortalReplanStats portal_replan{};
};

class PathRequestRuntime {
 public:
  void reserve_requests(std::size_t count) {
    requests_.reserve(count);
    results_.reserve(count);
    offsets_.reserve(count);
    sizes_.reserve(count);
    processed_.reserve(count);
    request_group_.reserve(count);
    group_members_.reserve(count);
    group_start_chunks_.reserve(count);
    precheck_survivors_.reserve(count);
    survivor_original_.reserve(count);
    weighted_batch_.reserve_requests(count);
    unit_field_goals_.reserve(1);
  }

  void reserve_path_nodes(std::size_t count) {
    paths_.reserve(count);
    unit_route_cache_.reserve_path_nodes(count);
    unit_field_scratch_.reserve_nodes(count);
    unit_field_product_.reserve_nodes(count);
    weighted_batch_.reserve_path_nodes(count);
    portal_segment_cache_.reserve_path_nodes(count);
    portal_replan_product_.reserve_path_nodes(count);
  }

  void reserve_search_nodes(std::size_t count) {
    unit_scratch_.reserve_nodes(count);
    unit_field_scratch_.reserve_nodes(count);
    unit_field_product_.reserve_nodes(count);
    weighted_batch_.reserve_search_nodes(count);
  }

  void reserve_unit_routes(std::size_t count) {
    unit_route_cache_.reserve_routes(count);
  }

  void reserve_unit_field_products(std::size_t count) {
    unit_field_product_cache_.reserve_entries(count);
  }

  void reserve_unit_field_product_dependencies(std::size_t count) {
    unit_field_product_.reserve_dependencies(count);
  }

  void reserve_weighted_field_products(std::size_t count) {
    unit_field_product_cache_.reserve_entries(count);
  }

  void reserve_weighted_field_product_dependencies(std::size_t count) {
    unit_field_product_.reserve_dependencies(count);
  }

  void reserve_portal_segments(std::size_t count) {
    portal_segment_cache_.reserve_segments(count);
    portal_replan_product_.reserve_waypoints(count);
  }

  void clear_requests() noexcept {
    requests_.clear();
    clear_results();
    ++generation_;
  }

  void clear_caches() noexcept {
    unit_route_cache_.clear();
    unit_field_product_cache_.clear();
    portal_segment_cache_.clear();
    world_changes_since_clear_ = 0;
    bound_unit_class_ = 0;
    ++cache_clears_;
  }

  [[nodiscard]] auto submit(PathRequest request) -> PathTicket {
    const auto ticket = PathTicket{requests_.size(), generation_};
    requests_.push_back(request);
    return ticket;
  }

  [[nodiscard]] auto requests() const noexcept -> std::span<const PathRequest> {
    return requests_;
  }

  [[nodiscard]] auto results() const noexcept -> std::span<const PathResult> {
    return results_published_ ? std::span<const PathResult>{results_}
                              : std::span<const PathResult>{};
  }

  [[nodiscard]] auto try_result(PathTicket ticket) const noexcept
      -> std::optional<PathResult> {
    if (ticket.generation != generation_ || !results_published_ ||
        ticket.value >= results_.size()) {
      return std::nullopt;
    }
    return results_[ticket.value];
  }

  [[nodiscard]] auto result(PathTicket ticket) const noexcept -> PathResult {
    if (ticket.generation != generation_) {
      detail::fail_fast(
          "PathRequestRuntime::result received a stale PathTicket; use "
          "try_result() for uncertain lookup");
    }
    if (!results_published_) {
      detail::fail_fast(
          "PathRequestRuntime::result has no published result batch; process "
          "requests first or use try_result()");
    }
    if (ticket.value >= results_.size()) {
      detail::fail_fast(
          "PathRequestRuntime::result received an out-of-range PathTicket; "
          "use try_result() for uncertain lookup");
    }
    return results_[ticket.value];
  }

  [[nodiscard]] auto route_cache() noexcept -> UnitRouteCache& {
    return unit_route_cache_;
  }

  [[nodiscard]] auto route_cache() const noexcept -> const UnitRouteCache& {
    return unit_route_cache_;
  }

  [[nodiscard]] auto portal_segment_cache() noexcept
      -> WeightedPortalSegmentCache& {
    return portal_segment_cache_;
  }

  [[nodiscard]] auto portal_segment_cache() const noexcept
      -> const WeightedPortalSegmentCache& {
    return portal_segment_cache_;
  }

  [[nodiscard]] auto stats() const noexcept -> PathRuntimeStats {
    auto stats = stats_;
    stats.submitted = requests_.size();
    if (results_published_) {
      stats.completed = results_.size();
      stats.path_nodes = paths_.size();
    } else {
      stats.completed = 0;
      stats.found = 0;
      stats.invalid_start = 0;
      stats.invalid_goal = 0;
      stats.no_path = 0;
      stats.not_computed = 0;
      stats.no_candidate = 0;
      stats.indeterminate = 0;
      stats.cost_overflow = 0;
      stats.precheck_ruled_out = 0;
      stats.path_nodes = 0;
      stats.field_product_candidate_groups = 0;
      stats.field_product_used_groups = 0;
      stats.field_product_skipped_groups = 0;
      stats.portal_replan = {};
    }
    stats.route_cache = unit_route_cache_.stats();
    stats.field_product_cache = unit_field_product_cache_.stats();
    stats.weighted_batch =
        results_published_ ? weighted_batch_.stats() : WeightedPathBatchStats{};
    stats.portal_segment_cache = portal_segment_cache_.stats();
    stats.cache_clears = cache_clears_;
    stats.class_cache_invalidations = class_cache_invalidations_;
    return stats;
  }

  template <typename World, typename ClassOrTag>
  [[nodiscard]] auto process_unit_cached(
      const World& world, PathRuntimeCachePolicy policy = {},
      const RegionGraphT<typename World::residency_type>* graph = nullptr)
      -> std::span<const PathResult> {
    clear_results();
    results_.resize(requests_.size());
    offsets_.assign(requests_.size(), 0);
    sizes_.assign(requests_.size(), 0);
    processed_.assign(requests_.size(), 0);
    stats_ = {};
    // Bind AFTER prepare_process: a policy-triggered clear_caches() there
    // zeroes the binding, and binding first would let this call refill the
    // caches under an unbound (0) identity that a later class could then
    // silently reuse.
    prepare_process(world, policy);
    bind_unit_class(
        detail::tag_identity<movement::movement_class_of<ClassOrTag>>());
    if (graph != nullptr) {
      precheck_prepass<ClassOrTag>(world, *graph, policy.missing_chunk_policy,
                                   AdjacentTransitions{});
    }
    if constexpr (std::is_same_v<typename World::residency_type,
                                 AlwaysResident>) {
      // The unit field-product cache is dense-only (it sizes distance arrays by
      // the global tile count). if constexpr, not a runtime if, so the
      // dense-only process_repeated_goal_fields is never instantiated for a
      // sparse world; sparse unit path processing routes each request through
      // cached_astar_path instead.
      if (policy.use_unit_field_product_cache) {
        process_repeated_goal_fields<World, ClassOrTag>(world, policy);
      }
    }
 
    for (std::size_t i = 0; i < requests_.size(); ++i) {
      if (processed_[i] != 0) {
        continue;
      }
      const auto result = cached_astar_path<World, ClassOrTag>(
          world, requests_[i], unit_scratch_, unit_route_cache_,
          policy.missing_chunk_policy);
      copy_result(i, result);
      record_status(result.status);
    }
    refresh_result_spans();
    results_published_ = true;
    return results_;
  }

  template <typename World, typename ClassOrTag, typename Provider>
  [[nodiscard]] auto process_unit_cached(
      const World& world, PathRuntimeCachePolicy policy,
      const RegionGraphT<typename World::residency_type>* graph,
      const Provider& provider) -> std::span<const PathResult> {
    clear_results();
    results_.resize(requests_.size());
    offsets_.assign(requests_.size(), 0);
    sizes_.assign(requests_.size(), 0);
    processed_.assign(requests_.size(), 0);
    stats_ = {};
    prepare_process(world, policy);
    bind_unit_class(
        detail::tag_identity<movement::movement_class_of<ClassOrTag>>());
    if (graph != nullptr && graph->matches_provider(provider)) {
      precheck_prepass<ClassOrTag>(world, *graph, policy.missing_chunk_policy,
                                   provider);
    }
 
    // Provider-aware products are public, but the runtime deliberately uses
    // its exact route cache here until provider revision is threaded through
    // the repeated-goal selection policy and its counters.
    for (std::size_t i = 0; i < requests_.size(); ++i) {
      if (processed_[i] != 0) {
        continue;
      }
      const auto result = cached_astar_path<World, ClassOrTag, Provider>(
          world, requests_[i], unit_scratch_, unit_route_cache_, provider,
          policy.missing_chunk_policy);
      copy_result(i, result);
      record_status(result.status);
    }
    refresh_result_spans();
    results_published_ = true;
    return results_;
  }

  template <typename World, typename Class, std::uint32_t MaxCost>
  [[nodiscard]] auto process_weighted_batch(
      const World& world, PathRuntimeCachePolicy policy = {},
      const RegionGraphT<typename World::residency_type>* graph = nullptr)
      -> std::span<const PathResult> {
    static_assert(std::derived_from<Class, movement::movement_class_tag>,
                  "process_weighted_batch<World, Class, MaxCost> requires a "
                  "MovementClass; pass a movement class such as "
                  "PositiveCostFieldMovement.");
    return process_weighted_batch_impl<World, Class, MaxCost, Class>(
        world, policy, graph, AdjacentTransitions{});
  }

  template <typename World, typename Class, std::uint32_t MaxCost,
            typename Provider>
  [[nodiscard]] auto process_weighted_batch(
      const World& world, PathRuntimeCachePolicy policy,
      const RegionGraphT<typename World::residency_type>* graph,
      const Provider& provider) -> std::span<const PathResult> {
    static_assert(std::derived_from<Class, movement::movement_class_tag>,
                  "process_weighted_batch<World, Class, MaxCost> requires a "
                  "MovementClass; pass a movement class such as "
                  "PositiveCostFieldMovement.");
    return process_weighted_batch_impl<World, Class, MaxCost, Class>(
        world, policy, graph, provider);
  }
 
 private:
  template <typename World, typename BatchClass, std::uint32_t MaxCost,
            typename PrecheckClassOrTag, typename Provider>
  [[nodiscard]] auto process_weighted_batch_impl(
      const World& world, PathRuntimeCachePolicy policy,
      const RegionGraphT<typename World::residency_type>* graph,
      const Provider& provider) -> std::span<const PathResult> {
    clear_results();
    results_.resize(requests_.size());
    offsets_.assign(requests_.size(), 0);
    sizes_.assign(requests_.size(), 0);
    processed_.assign(requests_.size(), 0);
    stats_ = {};
    prepare_process(world, policy);
 
    if (graph != nullptr && !graph->matches_provider(provider)) {
      graph = nullptr;
    }
    if (graph != nullptr) {
      precheck_prepass<PrecheckClassOrTag>(
          world, *graph, policy.missing_chunk_policy, provider);
    }
    if constexpr (std::is_same_v<typename World::residency_type,
                                 AlwaysResident>) {
      if (policy.use_weighted_field_product_cache) {
        process_repeated_goal_weighted_fields<World, BatchClass>(world, policy,
                                                                 provider);
      }
    }
 
    // Portal-first pass over eligible single-goal survivors: an accepted
    // portal route is copied into result storage immediately (the returned
    // path borrows the shared product workspace) and marks the request
    // processed; every other outcome leaves the request untouched so it
    // flows into the exact batch below — fallback parity by construction.
    if constexpr (std::is_same_v<typename World::residency_type,
                                 AlwaysResident> &&
                  std::is_same_v<typename ShapeTraits<
                                     typename World::shape_type>::lattice_type,
                                 lattice::Orthogonal> &&
                  std::is_same_v<Provider, AdjacentTransitions> &&
                  detail::portal_replan_tags<BatchClass>::eligible) {
      if (policy.weighted_replan_strategy ==
          WeightedReplanStrategy::PortalFirst) {
        process_portal_first_singletons<World, BatchClass>(world, policy);
      }
    } else {
      // PortalFirst requested on an ineligible instantiation: everything
      // takes the exact path, and the counter is what separates a
      // misconfigured policy from a disabled one.
      if (policy.weighted_replan_strategy ==
          WeightedReplanStrategy::PortalFirst) {
        for (std::size_t i = 0; i < requests_.size(); ++i) {
          if (processed_[i] == 0) {
            ++stats_.portal_replan.ineligible_fallbacks;
          }
        }
      }
    }
 
    // Precheck and product reuse partition the batch. Only survivors run
    // through the existing weighted batch, then scatter to original slots.
    precheck_survivors_.clear();
    survivor_original_.clear();
    for (std::size_t i = 0; i < requests_.size(); ++i) {
      if (processed_[i] == 0) {
        survivor_original_.push_back(i);
        precheck_survivors_.push_back(requests_[i]);
      }
    }
    const auto batch =
        weighted_path_batch<World, BatchClass, MaxCost, Provider>(
            world, precheck_survivors_, weighted_batch_,
            policy.missing_chunk_policy, provider);
    for (std::size_t s = 0; s < batch.size(); ++s) {
      const auto i = survivor_original_[s];
      copy_result(i, batch[s]);
      record_status(batch[s].status);
    }
    refresh_result_spans();
    results_published_ = true;
    return results_;
  }
 
  template <typename World, typename Class, typename Provider>
  void process_repeated_goal_weighted_fields(const World& world,
                                             PathRuntimeCachePolicy policy,
                                             const Provider& provider) {
    using Shape = typename World::shape_type;
    if (policy.weighted_field_product_min_goal_reuse < 2) {
      policy.weighted_field_product_min_goal_reuse = 2;
    }
    if (policy.weighted_field_product_min_start_chunks == 0) {
      policy.weighted_field_product_min_start_chunks = 1;
    }
    unit_field_product_cache_.set_byte_budget(
        policy.weighted_field_product_cache_byte_budget);
    const auto product_distance_fits =
        unit_field_product_cache_.can_fit_distance_storage(
            detail::NodeIndexSpace<World>{world}.capacity_hint());
 
    constexpr auto no_group = std::numeric_limits<std::uint32_t>::max();
    group_goals_.clear();
    group_counts_.clear();
    request_group_.assign(requests_.size(), no_group);
    auto slot_capacity = goal_group_slots_.size() < 16u
                             ? std::size_t{16}
                             : goal_group_slots_.size();
    while (slot_capacity < (requests_.size() + 1u) * 2u) {
      slot_capacity *= 2u;
    }
    goal_group_slots_.assign(slot_capacity, 0u);
    const auto slot_mask = slot_capacity - 1u;
    for (std::size_t i = 0; i < requests_.size(); ++i) {
      if (processed_[i] != 0) {
        continue;
      }
      if (!contains<Shape>(requests_[i].start)) {
        auto invalid = PathResult{};
        invalid.status = PathStatus::InvalidStart;
        copy_result(i, invalid);
        record_status(invalid.status);
        processed_[i] = 1;
        continue;
      }
      const auto goal = requests_[i].goal;
      auto slot =
          static_cast<std::size_t>(detail::coord_hash(goal)) & slot_mask;
      auto group = no_group;
      while (goal_group_slots_[slot] != 0u) {
        const auto candidate = goal_group_slots_[slot] - 1u;
        if (group_goals_[candidate] == goal) {
          group = candidate;
          break;
        }
        slot = (slot + 1u) & slot_mask;
      }
      if (group == no_group) {
        group = static_cast<std::uint32_t>(group_goals_.size());
        goal_group_slots_[slot] = group + 1u;
        group_goals_.push_back(goal);
        group_counts_.push_back(0u);
      }
      request_group_[i] = group;
      ++group_counts_[group];
    }
 
    group_offsets_.assign(group_goals_.size() + 1u, 0u);
    for (std::size_t i = 0; i < requests_.size(); ++i) {
      if (request_group_[i] != no_group) {
        ++group_offsets_[request_group_[i] + 1u];
      }
    }
    for (std::size_t group = 1; group < group_offsets_.size(); ++group) {
      group_offsets_[group] += group_offsets_[group - 1u];
    }
    group_cursors_.assign(group_offsets_.begin(), group_offsets_.end());
    group_members_.assign(group_offsets_.back(), 0u);
    for (std::size_t i = 0; i < requests_.size(); ++i) {
      if (request_group_[i] != no_group) {
        group_members_[group_cursors_[request_group_[i]]++] =
            static_cast<std::uint32_t>(i);
      }
    }
 
    for (std::uint32_t group = 0; group < group_goals_.size(); ++group) {
      if (group_counts_[group] < policy.weighted_field_product_min_goal_reuse) {
        continue;
      }
      const auto members_begin = group_offsets_[group];
      const auto members_end = group_offsets_[group + 1u];
      group_start_chunks_.clear();
      for (auto member = members_begin; member < members_end; ++member) {
        const auto& request = requests_[group_members_[member]];
        group_start_chunks_.push_back(
            chunk_key<Shape>(tile_key<Shape>(request.start)).value);
      }
      std::sort(group_start_chunks_.begin(), group_start_chunks_.end());
      const auto start_chunk_count = static_cast<std::size_t>(
          std::unique(group_start_chunks_.begin(), group_start_chunks_.end()) -
          group_start_chunks_.begin());
      ++stats_.field_product_candidate_groups;
      if (start_chunk_count < policy.weighted_field_product_min_start_chunks) {
        ++stats_.field_product_skipped_groups;
        continue;
      }
      if (!product_distance_fits) {
        // The normal weighted batch still builds one shared truncated field.
        // Do not first build a full cache product that cannot possibly fit:
        // an over-budget store would discard it and clear unrelated entries.
        ++stats_.field_product_skipped_groups;
        continue;
      }
 
      unit_field_goals_.clear();
      unit_field_goals_.add(group_goals_[group]);
      auto* product = unit_field_product_cache_
                          .template lookup_weighted<World, Class, Provider>(
                              world, unit_field_goals_, provider);
      if (product == nullptr) {
        const auto field =
            build_weighted_distance_field_product<World, Class, Provider>(
                world, unit_field_goals_, unit_field_product_,
                unit_field_scratch_, provider);
        if (field.status == PathStatus::Found) {
          // See the unit path below: the store returns what it stored, so
          // the second lookup (a full rescan that also counted a spurious
          // hit) is gone, and the product keeps reusable storage.
          product =
              unit_field_product_cache_
                  .template store_weighted_reusing<World, Class, Provider>(
                      unit_field_product_, provider);
        }
      }
      if (product == nullptr) {
        ++stats_.field_product_skipped_groups;
        continue;
      }
 
      ++stats_.field_product_used_groups;
      for (auto member = members_begin; member < members_end; ++member) {
        const auto index = static_cast<std::size_t>(group_members_[member]);
        const auto result =
            weighted_distance_field_product_path<World, Class, Provider>(
                world, requests_[index].start, *product, unit_field_scratch_,
                provider);
        copy_result(index, result);
        record_status(result.status);
        processed_[index] = 1;
      }
    }
  }
 
  // The unit route cache keys entries on (start, goal) — staleness carried
  // by the world fingerprint or, in scoped mode, per-chunk dependency
  // records — and nothing on the movement class, so a runtime reused
  // across classes would serve one class's route for another. Each unit
  // process call binds the runtime to its (normalized) class; a rebind
  // clears the unit caches -- correct even on misuse -- and counts in
  // stats().class_cache_invalidations. One runtime per (world, class) is
  // therefore the PERF contract, not a correctness precondition. The
  // field-product cache already folds the class identity into its keys, and
  // the weighted batch keeps no cross-call cache. The portal segment cache
  // binds through for_class(), so a class change safely clears its entries.
  // Keeping one runtime per (world, class) remains the performance contract:
  // it avoids both kinds of conservative whole-cache rebind invalidation.
  void bind_unit_class(std::uintptr_t identity) noexcept {
    if (bound_unit_class_ == identity) {
      return;
    }
    if (bound_unit_class_ != 0) {
      unit_route_cache_.clear();
      unit_field_product_cache_.clear();
      ++class_cache_invalidations_;
    }
    bound_unit_class_ = identity;
  }
 
  void clear_results() noexcept {
    results_published_ = false;
    results_.clear();
    offsets_.clear();
    sizes_.clear();
    processed_.clear();
    paths_.clear();
    stats_ = {};
  }
 
  template <typename World>
  void prepare_process(const World& world, PathRuntimeCachePolicy policy) {
    unit_route_cache_.set_caps(UnitRouteCacheLimits{
        policy.max_route_entries, policy.max_route_path_nodes});
    unit_route_cache_.set_dependency_cap(policy.max_route_dependency_pairs);
    // A staleness-mode flip clears the unit caches unconditionally (entries
    // stored under one mode's semantics are never served under the
    // other's), independent of the world-change flag below.
    unit_route_cache_.set_staleness(policy.unit_route_staleness);
    portal_segment_cache_.set_segment_budget(policy.portal_segment_budget);
    if (!policy.invalidate_unit_route_cache_on_world_change) {
      return;
    }
    if (!unit_route_cache_.refresh_if_world_changed(world)) {
      return;
    }
    ++stats_.world_cache_invalidations;
    // The deep-clear counter pauses only when scoped validation is
    // actually in effect: sparse worlds fall back to the exact fingerprint
    // lifecycle regardless of the requested policy, so they keep the
    // periodic collision backstop and cache sweep.
    if constexpr (std::is_same_v<typename World::residency_type,
                                 AlwaysResident>) {
      if (policy.unit_route_staleness == UnitRouteStaleness::ScopedFeasible) {
        // Scoped mode: the change armed per-entry validation instead of
        // dropping entries; the deep-clear counter stays put (see the
        // policy field's comment).
        return;
      }
    }
    ++world_changes_since_clear_;
    if (policy.clear_every_world_change != 0 &&
        world_changes_since_clear_ >= policy.clear_every_world_change) {
      clear_caches();
    }
  }
 
  // Groups repeated-goal requests in one O(n) pass: a reusable
  // open-addressed flat map (power-of-two capacity, linear probing) assigns
  // each distinct goal a group id in first-occurrence order, member indices
  // are bucketed with a counting sort, and per-group distinct start chunks
  // come from a sort+unique over reusable scratch. Group processing order
  // and all stats semantics match the previous per-request rescan.
  template <typename World, typename PassableTag>
  void process_repeated_goal_fields(const World& world,
                                    PathRuntimeCachePolicy policy) {
    using Shape = typename World::shape_type;
 
    if (policy.unit_field_product_min_goal_reuse < 2) {
      policy.unit_field_product_min_goal_reuse = 2;
    }
    if (policy.unit_field_product_min_start_chunks == 0) {
      policy.unit_field_product_min_start_chunks = 1;
    }
    unit_field_product_cache_.set_byte_budget(
        policy.unit_field_product_cache_byte_budget);
    const auto product_distance_fits =
        unit_field_product_cache_.can_fit_distance_storage(
            detail::NodeIndexSpace<World>{world}.capacity_hint());
 
    constexpr auto no_group = std::numeric_limits<std::uint32_t>::max();
    group_goals_.clear();
    group_counts_.clear();
    request_group_.assign(requests_.size(), no_group);
    auto slot_capacity = goal_group_slots_.size() < 16u
                             ? std::size_t{16}
                             : goal_group_slots_.size();
    while (slot_capacity < (requests_.size() + 1u) * 2u) {
      slot_capacity *= 2u;
    }
    goal_group_slots_.assign(slot_capacity, 0u);
    const auto slot_mask = slot_capacity - 1u;
    for (std::size_t i = 0; i < requests_.size(); ++i) {
      if (processed_[i] != 0) {
        continue;
      }
      // The grouping fast path converts starts to unchecked tile keys below.
      // Resolve only out-of-shape starts here, as part of the existing O(n)
      // grouping pass. Passability and goal validation remain in the field
      // builder or ordinary A* fallback, avoiding duplicate world reads.
      if (!contains<Shape>(requests_[i].start)) {
        auto invalid = PathResult{};
        invalid.status = PathStatus::InvalidStart;
        copy_result(i, invalid);
        record_status(invalid.status);
        processed_[i] = 1;
        continue;
      }
      const auto goal = requests_[i].goal;
      auto slot =
          static_cast<std::size_t>(detail::coord_hash(goal)) & slot_mask;
      auto group = no_group;
      while (goal_group_slots_[slot] != 0u) {
        const auto candidate = goal_group_slots_[slot] - 1u;
        if (group_goals_[candidate] == goal) {
          group = candidate;
          break;
        }
        slot = (slot + 1u) & slot_mask;
      }
      if (group == no_group) {
        group = static_cast<std::uint32_t>(group_goals_.size());
        goal_group_slots_[slot] = group + 1u;
        group_goals_.push_back(goal);
        group_counts_.push_back(0u);
      }
      request_group_[i] = group;
      ++group_counts_[group];
    }
 
    // Bucket member indices per group; members stay in submission order.
    group_offsets_.assign(group_goals_.size() + 1u, 0u);
    for (std::size_t i = 0; i < requests_.size(); ++i) {
      if (request_group_[i] != no_group) {
        ++group_offsets_[request_group_[i] + 1u];
      }
    }
    for (std::size_t g = 1; g < group_offsets_.size(); ++g) {
      group_offsets_[g] += group_offsets_[g - 1u];
    }
    group_cursors_.assign(group_offsets_.begin(), group_offsets_.end());
    group_members_.assign(group_offsets_.back(), 0u);
    for (std::size_t i = 0; i < requests_.size(); ++i) {
      if (request_group_[i] != no_group) {
        group_members_[group_cursors_[request_group_[i]]++] =
            static_cast<std::uint32_t>(i);
      }
    }
 
    for (std::uint32_t g = 0; g < group_goals_.size(); ++g) {
      if (group_counts_[g] < policy.unit_field_product_min_goal_reuse) {
        continue;
      }
      const auto members_begin = group_offsets_[g];
      const auto members_end = group_offsets_[g + 1u];
      group_start_chunks_.clear();
      for (auto m = members_begin; m < members_end; ++m) {
        const auto& request = requests_[group_members_[m]];
        group_start_chunks_.push_back(
            chunk_key<Shape>(tile_key<Shape>(request.start)).value);
      }
      std::sort(group_start_chunks_.begin(), group_start_chunks_.end());
      const auto start_chunk_count = static_cast<std::size_t>(
          std::unique(group_start_chunks_.begin(), group_start_chunks_.end()) -
          group_start_chunks_.begin());
 
      ++stats_.field_product_candidate_groups;
      if (start_chunk_count < policy.unit_field_product_min_start_chunks) {
        ++stats_.field_product_skipped_groups;
        continue;
      }
      if (!product_distance_fits) {
        // Ordinary A* remains the exact fallback. Avoid building a world-sized
        // product that store() must reject (and whose rejection clears useful
        // cache entries that already fit the configured budget).
        ++stats_.field_product_skipped_groups;
        continue;
      }
 
      unit_field_goals_.clear();
      unit_field_goals_.add(group_goals_[g]);
      auto* product =
          unit_field_product_cache_.template lookup<World, PassableTag>(
              world, unit_field_goals_);
      if (product == nullptr) {
        const auto field = build_distance_field_product<World, PassableTag>(
            world, unit_field_goals_, unit_field_product_, unit_field_scratch_);
        if (field.status == PathStatus::Found) {
          // The store returns what it stored, so no second lookup is
          // needed. That lookup rescanned every entry AND recorded a hit,
          // inflating the published cache-hit rate by one on every build.
          // It also leaves `unit_field_product_` holding storage the cache
          // displaced, so the next rebuild reuses that capacity instead of
          // reallocating a world-sized distance array.
          product = unit_field_product_cache_
                        .template store_reusing<World, PassableTag>(
                            unit_field_product_);
        }
      }
 
      if (product == nullptr) {
        ++stats_.field_product_skipped_groups;
        continue;
      }
 
      ++stats_.field_product_used_groups;
      for (auto m = members_begin; m < members_end; ++m) {
        const auto j = static_cast<std::size_t>(group_members_[m]);
        const auto result = distance_field_product_path<World, PassableTag>(
            world, requests_[j].start, *product, unit_field_scratch_);
        copy_result(j, result);
        record_status(result.status);
        processed_[j] = 1;
      }
    }
  }
 
  // Resolves every not-yet-processed request the region graph proves
  // unreachable to a NoPath result without running A*, marking it processed so
  // downstream field-product grouping and the search loop skip it. The graph's
  // freshness, class-agreement, and no-false-negative guarantees live in
  // precheck_path: a graph stamped for another movement class reads as
  // GraphStale there, so this pass rules nothing out and A* stays
  // authoritative.
  template <typename ClassOrTag, typename World, typename Provider>
  void precheck_prepass(
      const World& world,
      const RegionGraphT<typename World::residency_type>& graph,
      MissingChunkPolicy missing_chunk_policy, const Provider& provider) {
    for (std::size_t i = 0; i < requests_.size(); ++i) {
      if (processed_[i] != 0) {
        continue;
      }
      const auto status = precheck_path<ClassOrTag>(
          graph, world, requests_[i], precheck_scratch_, missing_chunk_policy,
          provider);
      if (!precheck_rules_out_path(status)) {
        continue;
      }
      auto ruled_out = PathResult{};
      ruled_out.status = PathStatus::NoPath;
      copy_result(i, ruled_out);
      record_status(ruled_out.status);
      ++stats_.precheck_ruled_out;
      processed_[i] = 1;
    }
  }
 
  // Serves eligible single-goal weighted requests through the chunk-portal
  // product path (see WeightedReplanStrategy::PortalFirst). Runs only under
  // the compile-time eligibility gate at the call site. Requests whose
  // endpoints are out of shape are skipped without counting an attempt —
  // the batch classifies them — and a goal shared by two unprocessed
  // requests is never a singleton.
  template <typename World, typename BatchClass>
  void process_portal_first_singletons(const World& world,
                                       const PathRuntimeCachePolicy& policy) {
    using Shape = typename World::shape_type;
    // O(n) goal-occurrence counts over unprocessed requests, reusing the
    // grouping scratch (any earlier fields pass has finished with it).
    constexpr auto no_group = std::numeric_limits<std::uint32_t>::max();
    group_goals_.clear();
    group_counts_.clear();
    auto slot_capacity = goal_group_slots_.size() < 16u
                             ? std::size_t{16}
                             : goal_group_slots_.size();
    while (slot_capacity < (requests_.size() + 1u) * 2u) {
      slot_capacity *= 2u;
    }
    goal_group_slots_.assign(slot_capacity, 0u);
    const auto slot_mask = slot_capacity - 1u;
    const auto group_of = [&](Coord3 goal, bool insert) {
      auto slot =
          static_cast<std::size_t>(detail::coord_hash(goal)) & slot_mask;
      while (goal_group_slots_[slot] != 0u) {
        const auto candidate = goal_group_slots_[slot] - 1u;
        if (group_goals_[candidate] == goal) {
          return candidate;
        }
        slot = (slot + 1u) & slot_mask;
      }
      if (!insert) {
        return no_group;
      }
      const auto group = static_cast<std::uint32_t>(group_goals_.size());
      goal_group_slots_[slot] = group + 1u;
      group_goals_.push_back(goal);
      group_counts_.push_back(0u);
      return group;
    };
    for (std::size_t i = 0; i < requests_.size(); ++i) {
      if (processed_[i] != 0) {
        continue;
      }
      ++group_counts_[group_of(requests_[i].goal, true)];
    }
 
    auto& replan_stats = stats_.portal_replan;
    for (std::size_t i = 0; i < requests_.size(); ++i) {
      if (processed_[i] != 0) {
        continue;
      }
      const auto request = requests_[i];
      if (!contains<Shape>(request.start) || !contains<Shape>(request.goal)) {
        continue;
      }
      if (group_counts_[group_of(request.goal, false)] != 1u) {
        continue;
      }
      ++replan_stats.attempts;
      const auto result =
          build_weighted_chunk_portal_route_product_cached<World, BatchClass>(
              world, request, unit_scratch_, portal_segment_cache_,
              portal_replan_product_);
      if (result.status != PathStatus::Found) {
        // Same-chunk requests always select a (waypoint-free) candidate,
        // so an empty waypoint list marks a selection failure only when
        // the endpoints sit in different chunks; a same-chunk failure is
        // a failed direct segment, i.e. a verification failure.
        const auto same_chunk = chunk_coord<Shape>(request.start) ==
                                chunk_coord<Shape>(request.goal);
        if (!same_chunk && portal_replan_product_.waypoints().empty()) {
          ++replan_stats.no_candidates;
        } else {
          ++replan_stats.verification_failures;
        }
        ++replan_stats.exact_fallbacks;
        continue;
      }
      // Premium cap: Manhattan is an admissible lower bound for the
      // eligible scale-1 model class, so cost * den <= manhattan * num
      // implies cost <= (num/den) x optimal. num == 0 means no cap; a
      // zero denominator normalizes to 1 so a misconfigured policy can
      // never silently accept every route through cost * 0 <= bound.
      const auto premium_den = policy.portal_premium_limit_den == 0
                                   ? std::uint32_t{1}
                                   : policy.portal_premium_limit_den;
      const auto lower_bound = static_cast<std::uint64_t>(
          detail::manhattan(request.start, request.goal));
      const auto accept =
          policy.portal_premium_limit_num == 0 ||
          static_cast<std::uint64_t>(result.cost) * premium_den <=
              lower_bound * policy.portal_premium_limit_num;
      if (!accept) {
        ++replan_stats.premium_rejections;
        ++replan_stats.exact_fallbacks;
        continue;
      }
      // The result's path borrows the shared product; copy_result copies
      // it into runtime storage before the next singleton rebuilds. The
      // eligible model class is scale-1, matching the result's default.
      copy_result(i, result);
      record_status(result.status);
      processed_[i] = 1;
      ++replan_stats.accepted;
    }
  }
 
  void copy_result(std::size_t index, PathResult result) {
    offsets_[index] = paths_.size();
    sizes_[index] = result.path.size();
    paths_.insert(paths_.end(), result.path.begin(), result.path.end());
    results_[index] = PathResult{
        result.status,        result.cost, result.expanded_nodes,
        result.reached_nodes, {},          result.cost_scale,
    };
  }
 
  void refresh_result_spans() noexcept {
    for (std::size_t i = 0; i < results_.size(); ++i) {
      if (sizes_[i] == 0) {
        results_[i].path = {};
      } else {
        results_[i].path =
            std::span<const Coord3>{paths_.data() + offsets_[i], sizes_[i]};
      }
    }
  }
 
  void record_status(PathStatus status) noexcept {
    switch (status) {
      case PathStatus::NotComputed:
        ++stats_.not_computed;
        return;
      case PathStatus::Found:
        ++stats_.found;
        return;
      case PathStatus::InvalidStart:
        ++stats_.invalid_start;
        return;
      case PathStatus::InvalidGoal:
        ++stats_.invalid_goal;
        return;
      case PathStatus::NoPath:
        ++stats_.no_path;
        return;
      case PathStatus::Indeterminate:
        ++stats_.indeterminate;
        return;
      case PathStatus::CostOverflow:
        ++stats_.cost_overflow;
        return;
      case PathStatus::NoCandidate:
        ++stats_.no_candidate;
        return;
    }
  }
 
  std::vector<PathRequest> requests_;
  std::vector<PathResult> results_;
  std::vector<std::size_t> offsets_;
  std::vector<std::size_t> sizes_;
  std::vector<std::uint8_t> processed_;
  std::vector<Coord3> paths_;
  bool results_published_ = false;
  // Optional pre-A* topology precheck: reused BFS scratch plus the survivor
  // partition (surviving requests and their original slot indices) that lets
  // the monolithic weighted batch skip proven-unreachable requests.
  RegionGraphScratch precheck_scratch_;
  std::vector<PathRequest> precheck_survivors_;
  std::vector<std::size_t> survivor_original_;
  // Reusable repeated-goal grouping scratch: flat-hash goal map slots,
  // per-group goals/counts/member buckets, and start-chunk dedup storage.
  std::vector<std::uint32_t> goal_group_slots_;
  std::vector<Coord3> group_goals_;
  std::vector<std::uint32_t> group_counts_;
  std::vector<std::uint32_t> group_offsets_;
  std::vector<std::uint32_t> group_cursors_;
  std::vector<std::uint32_t> group_members_;
  std::vector<std::uint32_t> request_group_;
  std::vector<std::uint64_t> group_start_chunks_;
  PathScratch unit_scratch_;
  UnitRouteCache unit_route_cache_;
  DistanceFieldScratch unit_field_scratch_;
  GoalSet unit_field_goals_;
  DistanceFieldProduct unit_field_product_;
  // The historical unit_ name predates weighted product reuse. Both processing
  // paths intentionally share this cache, its capacity, and its active budget.
  FieldProductCache unit_field_product_cache_;
  WeightedPathBatchScratch weighted_batch_;
  WeightedPortalSegmentCache portal_segment_cache_;
  // Portal-first replan workspace, reused across singletons in a batch;
  // every accepted route is copied out via copy_result before the next
  // request rebuilds it (the returned path borrows this product).
  WeightedPortalRouteProduct portal_replan_product_;
  PathRuntimeStats stats_;
  std::size_t world_changes_since_clear_ = 0;
  std::size_t cache_clears_ = 0;
  // Movement-class identity the unit caches are bound to (0 = unbound); see
  // bind_unit_class.
  std::uintptr_t bound_unit_class_ = 0;
  std::size_t class_cache_invalidations_ = 0;
  std::uint64_t generation_ = 0;
};
 
}  // namespace tess