queued.h

#pragma once
 
#include <tess/block/block.h>
#include <tess/core/assert.h>
#include <tess/core/capacity.h>
#include <tess/core/shape.h>
#include <tess/core/tag_identity.h>
#include <tess/diagnostics/diagnostics.h>
#include <tess/diagnostics/trace.h>
#include <tess/ops/phase_executor.h>
#include <tess/storage/world.h>
 
#include <algorithm>
#include <cstddef>
#include <cstdint>
#include <limits>
#include <optional>
#include <source_location>
#include <span>
#include <stdexcept>
#include <type_traits>
#include <utility>
#include <vector>
 
namespace tess {
 
namespace detail {
 
struct PlannedWorldStamp {
  std::uintptr_t shape_identity = 0;
  std::uint64_t chunk_limit = 0;
};
 
template <typename Shape, std::uint64_t ChunkLimit>
[[nodiscard]] inline auto planned_world_stamp() noexcept
    -> const PlannedWorldStamp* {
  static const auto stamp = PlannedWorldStamp{
      tag_identity<Shape>(),
      ChunkLimit,
  };
  return &stamp;
}
 
template <typename World>
[[nodiscard]] inline auto planned_world_stamp() noexcept
    -> const PlannedWorldStamp* {
  return planned_world_stamp<typename World::shape_type, World::chunk_count>();
}
 
template <typename World>
[[nodiscard]] inline auto validate_planned_world_stamp(
    const PlannedWorldStamp* stamp) noexcept -> PlannedExecutionStatus {
  const auto* expected = planned_world_stamp<World>();
  if (stamp == expected) {
    return PlannedExecutionStatus::Executed;
  }
  if (stamp == nullptr || stamp->shape_identity != expected->shape_identity) {
    return PlannedExecutionStatus::InvalidShape;
  }
  if (stamp->chunk_limit != expected->chunk_limit) {
    return PlannedExecutionStatus::InvalidChunk;
  }
  return PlannedExecutionStatus::Executed;
}
 
}  // namespace detail

struct OpId {
  std::uint64_t value = 0;
 
  friend constexpr bool operator==(OpId lhs, OpId rhs) noexcept = default;
};

struct OpHandle {
  std::uint64_t value = 0;
 
  friend constexpr bool operator==(OpHandle lhs,
                                   OpHandle rhs) noexcept = default;
};

enum class OperationKind : std::uint8_t {
  UpdateField,
  QueryPaths,
  QueryNearest,
  BuildFieldProduct,
  MoveEntities,
  RebuildTopology,
  EnsureResident,
  MarkDirty,
  PublishRenderDeltas,
};
static_assert(sizeof(OperationKind) == sizeof(std::uint8_t));

enum class BackendEligibility : std::uint8_t {
  CpuOnly,
  CpuOrGpu,
};

enum class ExactnessRequirement : std::uint8_t {
  Exact,
  ApproximateAllowed,
};

struct IntentVersions {
  std::uint64_t world = 0;
  std::uint64_t topology = 0;
  std::uint64_t fields = 0;
  std::uint64_t product = 0;
 
  friend constexpr bool operator==(IntentVersions lhs,
                                   IntentVersions rhs) noexcept = default;
};

struct IntentInvalidations {
  DirtyMask dirty_mask{};
  bool topology = false;
  bool paths = false;
  bool render = false;
 
  friend constexpr bool operator==(IntentInvalidations lhs,
                                   IntentInvalidations rhs) noexcept = default;
};

struct IntentPayloadView {
  template <typename T, std::size_t Extent>
  [[nodiscard]] static auto from(std::span<T, Extent> values) noexcept
      -> IntentPayloadView {
    using Item = std::remove_cv_t<T>;
    static_assert(std::is_object_v<Item>);
    static_assert(!std::is_volatile_v<T>);
    return IntentPayloadView{values.data(), values.size(), sizeof(Item),
                             detail::tag_identity<Item>()};
  }

  template <typename T>
  [[nodiscard]] auto holds() const noexcept -> bool {
    using Item = std::remove_cv_t<T>;
    static_assert(std::is_object_v<Item>);
    return item_size == sizeof(Item) &&
           type_identity == detail::tag_identity<Item>();
  }

  [[nodiscard]] auto bound() const noexcept -> bool {
    return type_identity != 0;
  }

  template <typename T>
  [[nodiscard]] auto as() const noexcept -> std::span<const T> {
    using Item = std::remove_cv_t<T>;
    static_assert(std::is_object_v<Item>);
    TESS_ASSERT_MSG(holds<Item>(),
                    "IntentPayloadView::as<T> on a payload that does not "
                    "hold T; check holds<T>() first");
    if (!holds<Item>()) {
      return {};
    }
    return {static_cast<const T*>(data), count};
  }
 
  const void* data = nullptr;
  std::size_t count = 0;
  std::size_t item_size = 0;
  std::uintptr_t type_identity = 0;
};

enum class Priority : std::uint8_t {
  Immediate,
  GameplayCritical,
  VisibleSoon,
  Background,
  Maintenance,
};
static_assert(sizeof(Priority) == sizeof(std::uint8_t));

enum class BudgetPolicy : std::uint8_t {
  MustRun,
  CanDefer,
  CanSkipIfSuperseded,
  BudgetedIncremental,
};
static_assert(sizeof(BudgetPolicy) == sizeof(std::uint8_t));

enum class OperationStatus : std::uint8_t {
  Planned,
  InvalidIdentity,
  InvalidWritePolicy,
  InvalidDomain,
  InvalidFieldAccess,
  HazardConflict,
};
static_assert(sizeof(OperationStatus) == sizeof(std::uint8_t));

enum class OperationFailure : std::uint8_t {
  None,
  NonDenseHandle,
  NonDenseId,
  InvalidWritePolicyValue,
  ExplicitChunkOutOfRange,
  ReadOnlyWriteMask,
  FieldHazardConflict,
};
static_assert(sizeof(OperationFailure) == sizeof(std::uint8_t));

enum class ExecutionPhaseStatus : std::uint8_t {
  Ready,
  UnsupportedWritePolicy,
};
static_assert(sizeof(ExecutionPhaseStatus) == sizeof(std::uint8_t));

enum class DomainKind : std::uint8_t {
  ExplicitChunks,
  DirtyChunks,
  ActiveChunks,
  ResidentChunks,
};
static_assert(sizeof(DomainKind) == sizeof(std::uint8_t));

class DomainDesc {
 public:
  // Explicit chunk keys are stored sorted and deduplicated so a planned
  // operation never visits one chunk twice: repeated keys under
  // UniquePerChunk would otherwise defeat the per-chunk ownership rule
  // that parallel phase planning relies on.
  [[nodiscard]] static auto explicit_chunks(std::span<const ChunkKey> keys)
      -> DomainDesc {
    DomainDesc desc{DomainKind::ExplicitChunks};
    desc.explicit_chunks_.assign(keys.begin(), keys.end());
    std::sort(desc.explicit_chunks_.begin(), desc.explicit_chunks_.end(),
              [](ChunkKey lhs, ChunkKey rhs) { return lhs.value < rhs.value; });
    desc.explicit_chunks_.erase(
        std::unique(desc.explicit_chunks_.begin(), desc.explicit_chunks_.end()),
        desc.explicit_chunks_.end());
    return desc;
  }
 
  [[nodiscard]] static constexpr auto dirty_chunks(DirtyMask mask) noexcept
      -> DomainDesc {
    DomainDesc desc{DomainKind::DirtyChunks};
    desc.mask_ = mask.value;
    return desc;
  }
 
  [[nodiscard]] static constexpr auto active_chunks(ActiveMask mask) noexcept
      -> DomainDesc {
    DomainDesc desc{DomainKind::ActiveChunks};
    desc.mask_ = mask.value;
    return desc;
  }
 
  [[nodiscard]] static constexpr auto resident_chunks() noexcept -> DomainDesc {
    return DomainDesc{DomainKind::ResidentChunks};
  }
 
  [[nodiscard]] constexpr auto kind() const noexcept -> DomainKind {
    return kind_;
  }
 
  [[nodiscard]] constexpr auto dirty_mask() const noexcept -> DirtyMask {
    return DirtyMask{mask_};
  }
 
  [[nodiscard]] constexpr auto active_mask() const noexcept -> ActiveMask {
    return ActiveMask{mask_};
  }

  [[nodiscard]] constexpr auto mask_bits() const noexcept -> std::uint32_t {
    return mask_;
  }
 
  [[nodiscard]] constexpr auto explicit_chunks() const noexcept
      -> std::span<const ChunkKey> {
    return {explicit_chunks_.data(), explicit_chunks_.size()};
  }
 
 private:
  constexpr explicit DomainDesc(DomainKind kind) noexcept : kind_(kind) {}
 
  DomainKind kind_;
  std::uint32_t mask_ = 0;
  std::vector<ChunkKey> explicit_chunks_;
};

struct FieldAccessDesc {
  std::uint32_t read_mask = 0;
  std::uint32_t write_mask = 0;
  DirtyMask dirty_mask = {};
 
  friend constexpr bool operator==(FieldAccessDesc lhs,
                                   FieldAccessDesc rhs) noexcept = default;
};

struct IntentMetadata {
  DomainDesc domain = DomainDesc::resident_chunks();
  FieldAccessDesc field_access{};
  WritePolicy write_policy = WritePolicy::ReadOnly;
  Priority priority = Priority::GameplayCritical;
  BudgetPolicy budget_policy = BudgetPolicy::MustRun;
  IntentVersions versions{};
  IntentInvalidations invalidations{};
  BackendEligibility backend = BackendEligibility::CpuOnly;
  ExactnessRequirement exactness = ExactnessRequirement::Exact;
};

struct PathBatchDesc {
  template <typename Request, std::size_t Extent>
  [[nodiscard]] static auto from(std::span<Request, Extent> requests,
                                 IntentMetadata operation = {})
      -> PathBatchDesc {
    return PathBatchDesc{IntentPayloadView::from(requests),
                         std::move(operation)};
  }
 
  IntentPayloadView requests{};
  IntentMetadata operation{};
};

struct NearestBatchDesc {
  template <typename Request, std::size_t Extent>
  [[nodiscard]] static auto from(std::span<Request, Extent> requests,
                                 IntentMetadata operation = {})
      -> NearestBatchDesc {
    return NearestBatchDesc{IntentPayloadView::from(requests),
                            std::move(operation)};
  }
 
  IntentPayloadView requests{};
  IntentMetadata operation{};
};

struct FieldProductDesc {
  template <typename Request, std::size_t Extent>
  [[nodiscard]] static auto from(std::span<Request, Extent> requests,
                                 IntentMetadata operation = {})
      -> FieldProductDesc {
    return FieldProductDesc{IntentPayloadView::from(requests),
                            std::move(operation)};
  }
 
  IntentPayloadView requests{};
  IntentMetadata operation{};
};

struct MoveBatchDesc {
  template <typename Request, std::size_t Extent>
  [[nodiscard]] static auto from(std::span<Request, Extent> requests,
                                 IntentMetadata operation = {})
      -> MoveBatchDesc {
    return MoveBatchDesc{IntentPayloadView::from(requests),
                         std::move(operation)};
  }
 
  IntentPayloadView requests{};
  IntentMetadata operation{};
};

struct TopologyRebuildDesc {
  IntentMetadata operation{};
};

struct ResidencyDesc {
  template <typename Request, std::size_t Extent>
  [[nodiscard]] static auto from(std::span<Request, Extent> requests,
                                 IntentMetadata operation = {})
      -> ResidencyDesc {
    return ResidencyDesc{IntentPayloadView::from(requests),
                         std::move(operation)};
  }
 
  IntentPayloadView requests{};
  IntentMetadata operation{};
};

struct MarkDirtyDesc {
  DomainDesc domain = DomainDesc::resident_chunks();
  DirtyMask dirty_mask = {};
};

struct RenderDeltaDesc {
  template <typename Request, std::size_t Extent>
  [[nodiscard]] static auto from(std::span<Request, Extent> requests,
                                 IntentMetadata operation = {})
      -> RenderDeltaDesc {
    return RenderDeltaDesc{IntentPayloadView::from(requests),
                           std::move(operation)};
  }
 
  IntentPayloadView requests{};
  IntentMetadata operation{};
};

struct PathBatchHandle {
  OpHandle operation{};
};

struct NearestBatchHandle {
  OpHandle operation{};
};

struct QueuedOperation {
  OperationKind kind = OperationKind::UpdateField;
  OpHandle handle{};
  OpId id{};
  DomainDesc domain = DomainDesc::resident_chunks();
  FieldAccessDesc field_access{};
  WritePolicy write_policy = WritePolicy::ReadOnly;
  Priority priority = Priority::GameplayCritical;
  BudgetPolicy budget_policy = BudgetPolicy::MustRun;
  std::source_location source = std::source_location::current();
  IntentPayloadView payload{};
  IntentVersions versions{};
  IntentInvalidations invalidations{};
  BackendEligibility backend = BackendEligibility::CpuOnly;
  ExactnessRequirement exactness = ExactnessRequirement::Exact;
};

struct OperationAccess {
  WritePolicy write_policy = WritePolicy::ReadOnly;
  DomainKind domain_kind = DomainKind::ResidentChunks;
  std::uint32_t domain_mask = 0;
};

enum class PlannedOperationCreateStatus : std::uint8_t {
  Created,
  InvalidChunk,
};
static_assert(sizeof(PlannedOperationCreateStatus) == sizeof(std::uint8_t));
 
struct PlannedOperationCreateResult;
class ExecutionReport;
class ExecutionPhase;

class PlannedOperation {
 public:
  OperationKind kind = OperationKind::UpdateField;
  OpHandle handle{};
  OpId id{};
  OperationAccess access{};
  FieldAccessDesc field_access{};
  WritePolicy write_policy = WritePolicy::ReadOnly;
  Priority priority = Priority::GameplayCritical;
  BudgetPolicy budget_policy = BudgetPolicy::MustRun;
  IntentPayloadView payload{};
  IntentVersions versions{};
  IntentInvalidations invalidations{};
  BackendEligibility backend = BackendEligibility::CpuOnly;
  ExactnessRequirement exactness = ExactnessRequirement::Exact;
  // Enqueue-site capture carried through planning so run-time completions
  // (result channels) can report where the operation came from.
  std::source_location source = std::source_location::current();

  template <typename World>
  [[nodiscard]] static auto create(const World& world,
                                   const QueuedOperation& operation,
                                   std::span<const ChunkKey> chunks)
      -> PlannedOperationCreateResult;

  [[nodiscard]] constexpr auto chunks() const noexcept
      -> std::span<const ChunkKey> {
    return chunks_;
  }

  template <typename World>
  [[nodiscard]] auto world_validation_status(
      const World& /*world*/) const noexcept -> PlannedExecutionStatus {
    static_assert(
        std::is_same_v<typename World::residency_type, AlwaysResident>,
        "Queued-op validation requires an AlwaysResidentWorld; use direct "
        "sparse-world operations instead.");
    return detail::validate_planned_world_stamp<World>(world_stamp_);
  }
 
 private:
  friend class ExecutionReport;
  friend class ExecutionPhase;
 
  PlannedOperation(const QueuedOperation& operation,
                   std::vector<ChunkKey>&& chunks,
                   const detail::PlannedWorldStamp* world_stamp) noexcept
      : kind(operation.kind),
        handle(operation.handle),
        id(operation.id),
        access(OperationAccess{operation.write_policy, operation.domain.kind(),
                               operation.domain.mask_bits()}),
        field_access(operation.field_access),
        write_policy(operation.write_policy),
        priority(operation.priority),
        budget_policy(operation.budget_policy),
        payload(operation.payload),
        versions(operation.versions),
        invalidations(operation.invalidations),
        backend(operation.backend),
        exactness(operation.exactness),
        source(operation.source),
        chunks_(std::move(chunks)),
        world_stamp_(world_stamp) {}
 
  std::vector<ChunkKey> chunks_;
  const detail::PlannedWorldStamp* world_stamp_ = nullptr;
};

struct PlannedOperationCreateResult {
  PlannedOperationCreateStatus status =
      PlannedOperationCreateStatus::InvalidChunk;
  std::optional<PlannedOperation> operation;
  ChunkKey invalid_chunk{};
};
 
template <typename World>
auto PlannedOperation::create(const World& /*world*/,
                              const QueuedOperation& operation,
                              std::span<const ChunkKey> chunks)
    -> PlannedOperationCreateResult {
  static_assert(std::is_same_v<typename World::residency_type, AlwaysResident>,
                "Queued operations require an AlwaysResidentWorld; use direct "
                "sparse-world operations instead.");
 
  for (const auto key : chunks) {
    if (key.value >= World::chunk_count) {
      return PlannedOperationCreateResult{
          PlannedOperationCreateStatus::InvalidChunk,
          std::nullopt,
          key,
      };
    }
  }
 
  auto validated = std::vector<ChunkKey>{chunks.begin(), chunks.end()};
  std::sort(validated.begin(), validated.end(),
            [](ChunkKey lhs, ChunkKey rhs) { return lhs.value < rhs.value; });
  validated.erase(std::unique(validated.begin(), validated.end()),
                  validated.end());
  auto planned = PlannedOperation{
      operation,
      std::move(validated),
      detail::planned_world_stamp<World>(),
  };
  return PlannedOperationCreateResult{
      PlannedOperationCreateStatus::Created,
      std::optional<PlannedOperation>{std::move(planned)},
      {},
  };
}

class ExecutionPlan {
 public:
  // Lvalue-only for the same reason as ExecutionReport's: the span points
  // into operations_, and a plan reached through a temporary report would
  // outlive it.
  [[nodiscard]] constexpr auto operations() const& noexcept
      -> std::span<const PlannedOperation> {
    return {operations_.data(), operations_.size()};
  }
  auto operations() const&& -> std::span<const PlannedOperation> = delete;
 
  [[nodiscard]] constexpr bool empty() const noexcept {
    return operations_.empty();
  }
 
  [[nodiscard]] constexpr auto size() const noexcept -> std::size_t {
    return operations_.size();
  }
 
 private:
  friend class ExecutionReport;
  friend class ExecutionPhase;
 
  ExecutionPlan() noexcept = default;
  ExecutionPlan(const ExecutionPlan&) = default;
  ExecutionPlan(ExecutionPlan&&) noexcept = default;
 
  // generation_ is this plan's capability epoch and must never be copied from
  // another plan; assignment invalidates it through bump_generation instead.
  // cppcheck-suppress operatorEqVarError
  auto operator=(const ExecutionPlan& other) -> ExecutionPlan& {
    if (this != &other) {
      // Expire every issued capability before a potentially throwing vector
      // copy. A failed assignment must not leave an old phase authorized for
      // whatever state the vector copy preserved or partially replaced.
      bump_generation();
      operations_ = other.operations_;
    }
    return *this;
  }
 
  auto operator=(ExecutionPlan&& other) noexcept -> ExecutionPlan& {
    if (this != &other) {
      operations_ = std::move(other.operations_);
      bump_generation();
    }
    return *this;
  }
 
  constexpr void bump_generation() noexcept { ++generation_; }
 
  std::vector<PlannedOperation> operations_;
  std::uint64_t generation_ = 0;
};

class ExecutionPhase {
 public:
  ExecutionPhase(const ExecutionPhase&) noexcept = default;
  ExecutionPhase(ExecutionPhase&&) noexcept = default;
  auto operator=(const ExecutionPhase&) noexcept -> ExecutionPhase& = default;
  auto operator=(ExecutionPhase&&) noexcept -> ExecutionPhase& = default;
 
  [[nodiscard]] constexpr auto first_operation() const noexcept -> std::size_t {
    return first_operation_;
  }
 
  [[nodiscard]] constexpr auto operation_count() const noexcept -> std::size_t {
    return operation_count_;
  }

  [[nodiscard]] constexpr bool belongs_to(
      const ExecutionPlan& plan) const noexcept {
    return plan_ == &plan && plan_generation_ == plan.generation_;
  }

  template <typename World>
  [[nodiscard]] auto world_validation_status(
      const World& /*world*/) const noexcept -> PlannedExecutionStatus {
    static_assert(
        std::is_same_v<typename World::residency_type, AlwaysResident>,
        "Queued-op validation requires an AlwaysResidentWorld; use direct "
        "sparse-world operations instead.");
    return detail::validate_planned_world_stamp<World>(world_stamp_);
  }

  template <WritePolicy Policy>
  [[nodiscard]] constexpr bool policy_matches() const noexcept {
    static_assert(is_valid_write_policy(Policy));
    return write_policy_mask_ == policy_bit(Policy);
  }
 
 private:
  friend class ExecutionPhasePlan;
 
  [[nodiscard]] static constexpr auto policy_bit(WritePolicy policy) noexcept
      -> std::uint8_t {
    return static_cast<std::uint8_t>(std::uint8_t{1}
                                     << static_cast<std::uint8_t>(policy));
  }
 
  constexpr ExecutionPhase(const ExecutionPlan& plan,
                           std::size_t first_operation,
                           std::size_t operation_count,
                           const PlannedOperation& operation) noexcept
      : plan_(&plan),
        first_operation_(first_operation),
        operation_count_(operation_count),
        plan_generation_(plan.generation_),
        world_stamp_(operation.world_stamp_),
        write_policy_mask_(policy_bit(operation.write_policy)) {}
 
  constexpr void extend(const PlannedOperation& operation) noexcept {
    ++operation_count_;
    write_policy_mask_ |= policy_bit(operation.write_policy);
  }
 
  const ExecutionPlan* plan_;
  std::size_t first_operation_;
  std::size_t operation_count_;
  std::uint64_t plan_generation_;
  const detail::PlannedWorldStamp* world_stamp_;
  std::uint8_t write_policy_mask_;
};

[[nodiscard]] constexpr auto executor_phase_range(
    const ExecutionPhase& phase) noexcept -> ExecutorPhaseRange {
  return ExecutorPhaseRange{
      phase.first_operation(),
      phase.operation_count(),
  };
}

class ExecutionPhasePlan {
 public:
  [[nodiscard]] constexpr auto phases() const noexcept
      -> std::span<const ExecutionPhase> {
    return {phases_.data(), phases_.size()};
  }
 
  [[nodiscard]] constexpr auto status() const noexcept -> ExecutionPhaseStatus {
    return status_;
  }
 
  [[nodiscard]] constexpr bool ok() const noexcept {
    return status_ == ExecutionPhaseStatus::Ready;
  }
 
  [[nodiscard]] constexpr auto failed_operation_index() const noexcept
      -> std::size_t {
    return failed_operation_index_;
  }
 
  [[nodiscard]] constexpr auto failed_write_policy() const noexcept
      -> WritePolicy {
    return failed_write_policy_;
  }
 
 private:
  friend auto plan_parallel_execution_phases(const ExecutionPlan& plan)
      -> ExecutionPhasePlan;
 
  void reserve(std::size_t size) { phases_.reserve(size); }
 
  void push_phase(const ExecutionPlan& plan, std::size_t first_operation,
                  std::size_t operation_count,
                  const PlannedOperation& operation) {
    phases_.push_back(
        ExecutionPhase{plan, first_operation, operation_count, operation});
  }
 
  void extend_last_phase(const PlannedOperation& operation) {
    phases_.back().extend(operation);
  }
 
  std::vector<ExecutionPhase> phases_;
  ExecutionPhaseStatus status_ = ExecutionPhaseStatus::Ready;
  std::size_t failed_operation_index_ = 0;
  WritePolicy failed_write_policy_ = WritePolicy::ReadOnly;
};
 
namespace detail {
 
[[nodiscard]] constexpr bool execution_phase_valid_for(
    const ExecutionPlan& plan, const ExecutionPhase& phase) noexcept {
  const auto operations = plan.operations();
  const auto first = phase.first_operation();
  const auto count = phase.operation_count();
  return phase.belongs_to(plan) && first <= operations.size() &&
         count <= operations.size() - first;
}
 
template <WritePolicy Policy, typename World>
[[nodiscard]] auto execution_phase_validation_status(
    const World& world, const ExecutionPlan& plan,
    const ExecutionPhase& phase) noexcept -> PlannedExecutionStatus {
  if (!execution_phase_valid_for(plan, phase)) {
    return PlannedExecutionStatus::InvalidPhase;
  }
  const auto world_status = phase.world_validation_status(world);
  if (world_status != PlannedExecutionStatus::Executed) {
    return world_status;
  }
  if (!phase.template policy_matches<Policy>()) {
    return PlannedExecutionStatus::PolicyMismatch;
  }
  return PlannedExecutionStatus::Executed;
}
 
inline void record_execution_phase_validation_failure(
    PlannedExecutionStatus status) noexcept {
#if TESS_DIAGNOSTICS_ENABLED
  if (status == PlannedExecutionStatus::InvalidPhase) {
    TESS_DIAG_EVENT(queued_phase_invalid_range);
  } else {
    TESS_DIAG_EVENT(queued_phase_failure);
  }
#else
  (void)status;
#endif
}
 
}  // namespace detail

struct OperationReport {
  OpHandle handle{};
  OpId id{};
  OperationStatus status = OperationStatus::Planned;
  OperationFailure failure = OperationFailure::None;
  OperationAccess access{};
  FieldAccessDesc field_access{};
  IntentVersions versions{};
  IntentInvalidations invalidations{};
  BackendEligibility backend = BackendEligibility::CpuOnly;
  ExactnessRequirement exactness = ExactnessRequirement::Exact;
  ChunkKey detail_chunk{};
  OpHandle conflict_handle{};
  OpId conflict_id{};
  std::uint32_t conflict_mask = 0;
  bool has_detail_chunk = false;
  bool has_conflict = false;
  std::size_t chunk_count = 0;
  std::source_location source = std::source_location::current();
};

struct PlannedDirtyRecord {
  ChunkKey chunk{};
  DirtyMask dirty_mask = {};
  Box3 bounds{};
};

enum class PlannedDirtyRecordStatus : std::uint8_t {
  Recorded,
  IgnoredEmptyMask,
  InvalidShape,
  InvalidChunk,
};
static_assert(sizeof(PlannedDirtyRecordStatus) == sizeof(std::uint8_t));

enum class PlannedDirtyMergeStatus : std::uint8_t {
  Merged,
  InvalidShape,
  InvalidChunk,
  CapacityExceeded,
};
static_assert(sizeof(PlannedDirtyMergeStatus) == sizeof(std::uint8_t));

enum class PlannedDirtyCollectStatus : std::uint8_t {
  Collected,
  InvalidShape,
  InvalidChunk,
  CapacityExceeded,
};
static_assert(sizeof(PlannedDirtyCollectStatus) == sizeof(std::uint8_t));

struct PlannedDirtyCollectResult {
  PlannedDirtyCollectStatus status = PlannedDirtyCollectStatus::Collected;
  std::size_t record_count = 0;
 
  [[nodiscard]] constexpr bool ok() const noexcept {
    return status == PlannedDirtyCollectStatus::Collected;
  }
};

struct PlannedDirtyMergeResult {
  PlannedDirtyMergeStatus status = PlannedDirtyMergeStatus::Merged;
  std::size_t merged_chunk_count = 0;
 
  [[nodiscard]] constexpr bool ok() const noexcept {
    return status == PlannedDirtyMergeStatus::Merged;
  }
};
 
class PlannedDirtyAccumulator;
class PlannedPhaseExecutionScratch;
 
namespace detail {
 
template <bool BindWorld, WritePolicy Policy, typename World, typename Fn>
[[nodiscard]] auto execute_validated_planned_operation_deferred_dirty(
    World& world, const PlannedOperation& operation,
    PlannedDirtyAccumulator& dirty, Fn&& fn) -> PlannedExecutionResult;
 
template <typename World>
[[nodiscard]] auto merge_planned_dirty_after_exception(
    World& world, PlannedPhaseExecutionScratch& scratch) noexcept
    -> PlannedDirtyMergeResult;
 
}  // namespace detail
 
class PlannedDirtyPartitions;

class PlannedDirtyAccumulator {
 public:
  void reserve(std::size_t count) { records_.reserve(count); }
 
  void clear() noexcept {
    records_.clear();
    world_stamp_ = nullptr;
  }

  template <typename World>
  [[nodiscard]] auto record(const World& /*world*/, ChunkKey chunk,
                            DirtyMask dirty_mask, Box3 bounds)
      -> PlannedDirtyRecordStatus {
    static_assert(
        std::is_same_v<typename World::residency_type, AlwaysResident>,
        "Queued-op dirty recording requires an AlwaysResidentWorld; use "
        "direct sparse-world operations instead.");
    if (!dirty_mask) {
      return PlannedDirtyRecordStatus::IgnoredEmptyMask;
    }
    if (chunk.value >= World::chunk_count) {
      return PlannedDirtyRecordStatus::InvalidChunk;
    }
 
    if (world_stamp_ != nullptr) {
      const auto validation =
          detail::validate_planned_world_stamp<World>(world_stamp_);
      if (validation == PlannedExecutionStatus::InvalidShape) {
        return PlannedDirtyRecordStatus::InvalidShape;
      }
      if (validation == PlannedExecutionStatus::InvalidChunk) {
        return PlannedDirtyRecordStatus::InvalidChunk;
      }
    }
 
    records_.push_back(PlannedDirtyRecord{chunk, dirty_mask, bounds});
    world_stamp_ = detail::planned_world_stamp<World>();
    return PlannedDirtyRecordStatus::Recorded;
  }
 
  [[nodiscard]] auto records() const noexcept
      -> std::span<const PlannedDirtyRecord> {
    return records_;
  }

  template <typename World>
  [[nodiscard]] auto validation_status(const World& /*world*/) const noexcept
      -> PlannedDirtyMergeStatus {
    static_assert(
        std::is_same_v<typename World::residency_type, AlwaysResident>,
        "Queued-op dirty validation requires an AlwaysResidentWorld; use "
        "direct sparse-world operations instead.");
    if (world_stamp_ == nullptr) {
      return PlannedDirtyMergeStatus::Merged;
    }
    const auto validation =
        detail::validate_planned_world_stamp<World>(world_stamp_);
    if (validation == PlannedExecutionStatus::InvalidShape) {
      return PlannedDirtyMergeStatus::InvalidShape;
    }
    if (validation == PlannedExecutionStatus::InvalidChunk) {
      return PlannedDirtyMergeStatus::InvalidChunk;
    }
    return PlannedDirtyMergeStatus::Merged;
  }
 
 private:
  friend class PlannedDirtyPartitions;
 
  template <bool BindWorld, WritePolicy Policy, typename World, typename Fn>
  friend auto detail::execute_validated_planned_operation_deferred_dirty(
      World& world, const PlannedOperation& operation,
      PlannedDirtyAccumulator& dirty, Fn&& fn) -> PlannedExecutionResult;
 
  template <WritePolicy Policy, typename World, typename Fn>
  friend auto execute_planned_operation_deferred_dirty(
      World& world, const PlannedOperation& operation,
      PlannedDirtyAccumulator& dirty, Fn&& fn) -> PlannedExecutionResult;
  template <typename World>
  friend auto merge_planned_dirty(World& world,
                                  PlannedDirtyAccumulator& dirty) noexcept
      -> PlannedDirtyMergeResult;
  template <typename World>
  friend auto merge_planned_dirty(World& world,
                                  PlannedPhaseExecutionScratch& scratch)
      -> PlannedDirtyMergeResult;
  friend auto collect_planned_dirty(PlannedDirtyAccumulator& dirty,
                                    PlannedDirtyPartitions& partitions)
      -> PlannedDirtyCollectResult;
 
  template <typename World>
  void bind_validated_world(const World& /*world*/) noexcept {
    if (world_stamp_ == nullptr) {
      world_stamp_ = detail::planned_world_stamp<World>();
    }
  }
 
  template <typename World>
  void prepare_for_validated_world(const World& /*world*/) noexcept {
    records_.clear();
    world_stamp_ = detail::planned_world_stamp<World>();
  }
 
  void record_validated(ChunkKey chunk, DirtyMask dirty_mask, Box3 bounds) {
    if (!dirty_mask) {
      return;
    }
    records_.push_back(PlannedDirtyRecord{chunk, dirty_mask, bounds});
  }
 
  std::vector<PlannedDirtyRecord> records_;
  const detail::PlannedWorldStamp* world_stamp_ = nullptr;
};
 
template <bool BindWorld, WritePolicy Policy, typename World, typename Fn>
auto detail::execute_validated_planned_operation_deferred_dirty(
    World& world, const PlannedOperation& operation,
    PlannedDirtyAccumulator& dirty, Fn&& fn) -> PlannedExecutionResult {
  if constexpr (BindWorld) {
    if (operation.field_access.dirty_mask) {
      dirty.bind_validated_world(world);
    }
  }
  auto ctx = block_ctx<Policy>(world, chunk_domain(operation.chunks()));
 
  std::size_t chunk_count = 0;
  auto&& callback = fn;
  ctx.for_each_chunk([&](auto view) {
    dirty.record_validated(view.key(), operation.field_access.dirty_mask,
                           view.bounds());
    callback(view);
    ++chunk_count;
  });
 
  return PlannedExecutionResult{
      PlannedExecutionStatus::Executed,
      chunk_count,
  };
}

class PlannedDirtyPartitions {
 public:
  void reserve(std::size_t count) { partitions_.reserve(count); }
 
  void resize(std::size_t count) { partitions_.resize(count); }
 
  void clear() noexcept { partitions_.clear(); }
 
  void clear_records() noexcept {
    for (auto& partition : partitions_) {
      partition.clear();
    }
  }
 
  void reserve_records_per_partition(std::size_t count) {
    records_per_partition_reserve_ = count;
    for (auto& partition : partitions_) {
      partition.reserve(count);
    }
  }
 
  [[nodiscard]] auto size() const noexcept -> std::size_t {
    return partitions_.size();
  }
 
  [[nodiscard]] auto partition(std::size_t index) noexcept
      -> PlannedDirtyAccumulator& {
    return partitions_[index];
  }
 
  [[nodiscard]] auto partition(std::size_t index) const noexcept
      -> const PlannedDirtyAccumulator& {
    return partitions_[index];
  }
 
  [[nodiscard]] auto partitions() const noexcept
      -> std::span<const PlannedDirtyAccumulator> {
    return partitions_;
  }
 
 private:
  friend auto collect_planned_dirty(PlannedDirtyAccumulator& dirty,
                                    PlannedDirtyPartitions& partitions)
      -> PlannedDirtyCollectResult;
  friend class PlannedPhaseExecutionScratch;
 
  void prepare(std::size_t count) {
    partitions_.resize(count);
    for (auto& partition : partitions_) {
      partition.clear();
      partition.reserve(records_per_partition_reserve_);
    }
  }
 
  template <typename World>
  void prepare(const World& world, std::size_t count) {
    partitions_.resize(count);
    for (auto& partition : partitions_) {
      partition.prepare_for_validated_world(world);
      partition.reserve(records_per_partition_reserve_);
    }
  }
 
  std::vector<PlannedDirtyAccumulator> partitions_;
  std::size_t records_per_partition_reserve_ = 0;
};
 
// Public because a public accessor returns a span of it:
// PhaseExecutionScratch::dirty_partitions() hands these out, so a caller
// reading per-operation dirty records had to name a `detail` type that
// docs/style.md says carries no source-compatibility guarantee.
// Scratch-owned phase partitions carry no independent world stamp. The
// enclosing scratch object owns one capability stamp, and this record-only
// type cannot be passed to the public dirty-merge APIs for another world.
class PhaseDirtyPartition {
 public:
  void reserve(std::size_t count) { records_.reserve(count); }
 
  void clear() noexcept { records_.clear(); }
 
  void record(ChunkKey chunk, DirtyMask dirty_mask, Box3 bounds) {
    if (dirty_mask) {
      // Phase setup reserves one record for every possible chunk visit.
      TESS_ASSERT(records_.size() < records_.capacity());
      records_.push_back(PlannedDirtyRecord{chunk, dirty_mask, bounds});
    }
  }
 
  [[nodiscard]] auto records() const noexcept
      -> std::span<const PlannedDirtyRecord> {
    return records_;
  }
 
 private:
  std::vector<PlannedDirtyRecord> records_;
};
 
namespace detail {
 
template <WritePolicy Policy, typename World>
using PlannedChunkView =
    ChunkView<typename BlockCtx<World, Policy>::view_world_type>;
 
template <WritePolicy Policy, typename World, typename Fn>
inline constexpr bool planned_callback_is_nothrow =
    std::is_nothrow_invocable_v<Fn&, PlannedChunkView<Policy, World>&>;
 
template <WritePolicy Policy, typename World, typename Fn>
[[nodiscard]] auto execute_validated_phase_operation_deferred_dirty(
    World& world, const PlannedOperation& operation, PhaseDirtyPartition& dirty,
    Fn&& fn) -> PlannedExecutionResult {
  auto ctx = block_ctx<Policy>(world, chunk_domain(operation.chunks()));
 
  std::size_t chunk_count = 0;
  auto&& callback = fn;
  ctx.for_each_chunk([&](auto view) {
    dirty.record(view.key(), operation.field_access.dirty_mask, view.bounds());
    callback(view);
    ++chunk_count;
  });
 
  return PlannedExecutionResult{
      PlannedExecutionStatus::Executed,
      chunk_count,
  };
}
 
// Chunk-keyed index over accepted operations, so hazard detection and
// parallel-phase grouping examine only the operations that actually share
// a chunk with the candidate instead of every operation accepted so far.
//
// Both scans were quadratic in the operation count: `find_hazard` walked
// every accepted operation per new operation, and phase grouping compared
// each operation against every member of the open phase. Per-chunk edits
// -- one operation per dirty chunk, the ordinary consumer shape -- are
// pairwise disjoint, so every one of those comparisons failed on chunk
// overlap only after paying for the check.
//
// An index only pays for itself when operations touch FEW chunks each.
// An operation over a whole domain -- `resident_chunks()`, the default
// selector -- would store one node per chunk and make every later lookup
// walk one chain per chunk, turning an O(n^2) scan into O(n^2 * chunks).
// Worse, that scan was the cheap case: read-only operations have a zero
// hazard mask, so it rejected each pair on the mask alone and never
// touched a chunk list. Measured at 64 read-only resident operations, the
// index cost 168 ms against the scan's 201 us before this bound existed.
//
// So operations wider than `index_max_chunks_per_operation` are kept OUT
// of the index and scanned linearly, and a candidate that wide skips the
// index and scans everything, exactly as the planner used to. Every
// lookup is then at worst the old cost and at best the indexed one.
inline constexpr std::size_t index_max_chunks_per_operation = 64;
 
[[nodiscard]] constexpr bool operation_is_indexable(
    std::span<const ChunkKey> chunks) noexcept {
  return chunks.size() <= index_max_chunks_per_operation;
}
 
// Open-addressed chunk key -> intrusive chain of nodes, power-of-two
// capacity, linear probing. No erase: a plan only grows, and the index is
// cleared wholesale per plan. Each node carries its key so a rehash can
// relink without consulting the operations.
class ChunkOperationIndex {
 public:
  static constexpr auto npos = std::numeric_limits<std::uint32_t>::max();
 
  // Stamped, not rewritten: `slots_` keeps its high-water size across the
  // report reuse this index exists to serve, so clearing it per plan would
  // charge every small batch for the largest batch that came before.
  void clear() noexcept {
    nodes_.clear();
    if (++generation_ != 0) {
      return;
    }
    // Wrapped: stale slots would read as live under the new generation.
    // Reset in place rather than `assign`, which is not noexcept -- and
    // `clear` runs on the report-reuse path, where throwing would leave a
    // half-cleared index behind a plan that reported success.
    for (auto& slot : slots_) {
      slot = Slot{};
    }
    generation_ = 1;
  }
 
  void reserve(std::size_t nodes) {
    nodes_.reserve(nodes);
    rehash(capacity_for(nodes));
  }

  void insert(std::span<const ChunkKey> chunks, std::uint32_t op_index) {
    if (slots_.empty() ||
        (nodes_.size() + chunks.size()) * 2 >= slots_.size()) {
      rehash(capacity_for(nodes_.size() + chunks.size()));
    }
    for (const auto key : chunks) {
      nodes_.push_back(Node{key.value, op_index, npos});
      link(static_cast<std::uint32_t>(nodes_.size() - 1));
    }
  }

  template <typename Visit>
  void for_each_sharing(std::span<const ChunkKey> chunks, Visit&& visit) const {
    if (slots_.empty()) {
      return;
    }
    for (const auto key : chunks) {
      for (auto node = head_for(key.value); node != npos;
           node = nodes_[node].next) {
        if (nodes_[node].key == key.value) {
          visit(nodes_[node].op_index);
        }
      }
    }
  }
 
 private:
  struct Slot {
    std::uint32_t head = npos;
    std::uint32_t generation = 0;
  };
  struct Node {
    std::uint64_t key = 0;
    std::uint32_t op_index = 0;
    std::uint32_t next = npos;
  };
 
  [[nodiscard]] static constexpr auto mix(std::uint64_t key) noexcept
      -> std::uint64_t {
    key ^= key >> 33U;
    key *= 0xff51afd7ed558ccdULL;
    key ^= key >> 33U;
    return key;
  }
 
  [[nodiscard]] static constexpr auto capacity_for(std::size_t nodes) noexcept
      -> std::size_t {
    auto capacity = std::size_t{16};
    while (capacity < nodes * 2) {
      capacity *= 2;
    }
    return capacity;
  }
 
  // Chains are keyed by bucket, and every node carries its key, so a
  // bucket may hold nodes for several keys; `for_each_sharing` filters.
  [[nodiscard]] auto bucket(std::uint64_t key) const noexcept -> std::size_t {
    return static_cast<std::size_t>(mix(key)) & (slots_.size() - 1);
  }
 
  [[nodiscard]] auto head_for(std::uint64_t key) const noexcept
      -> std::uint32_t {
    const auto& slot = slots_[bucket(key)];
    return slot.generation == generation_ ? slot.head : npos;
  }
 
  void link(std::uint32_t node) noexcept {
    auto& slot = slots_[bucket(nodes_[node].key)];
    if (slot.generation != generation_) {
      slot.generation = generation_;
      slot.head = npos;
    }
    nodes_[node].next = slot.head;
    slot.head = node;
  }
 
  void rehash(std::size_t capacity) {
    if (capacity <= slots_.size()) {
      return;
    }
    slots_.assign(capacity, Slot{});
    for (std::size_t i = 0; i < nodes_.size(); ++i) {
      nodes_[i].next = npos;
      link(static_cast<std::uint32_t>(i));
    }
  }
 
  std::vector<Slot> slots_;
  std::vector<Node> nodes_;
  std::uint32_t generation_ = 1;
};
 
}  // namespace detail

template <typename T>
class ResultChannel;

class PlannedPhaseExecutionScratch {
 public:
  void reserve_operations(std::size_t count) {
    dirty_partitions_.reserve(count);
    results_.reserve(count);
  }
 
  void reserve_dirty_records_per_operation(std::size_t count) {
    records_per_partition_reserve_ = count;
    for (auto& partition : dirty_partitions_) {
      partition.reserve(count);
    }
  }
 
  void reserve_merged_dirty_records(std::size_t count) {
    merged_dirty_.reserve(count);
  }
 
  void prepare_for_operation_count(std::size_t count) { prepare(count); }
 
  void clear() noexcept {
    for (auto& partition : dirty_partitions_) {
      partition.clear();
    }
    results_.clear();
    merged_dirty_.clear();
    world_stamp_ = nullptr;
  }
 
  [[nodiscard]] auto operation_count() const noexcept -> std::size_t {
    return results_.size();
  }
 
  [[nodiscard]] auto dirty_partitions() const noexcept
      -> std::span<const PhaseDirtyPartition> {
    return dirty_partitions_;
  }
 
 private:
  template <WritePolicy Policy, typename Executor, typename World, typename Fn>
  friend auto execute_phase_partitioned_dirty_with(
      Executor&& executor, World& world, const ExecutionPlan& plan,
      const ExecutionPhase& phase, PlannedPhaseExecutionScratch& scratch,
      Fn&& fn) -> PlannedExecutionResult;
 
  template <WritePolicy Policy, typename Executor, typename World, typename T,
            typename Fn>
  friend auto execute_phase_partitioned_dirty_with_results(
      Executor&& executor, World& world, const ExecutionPlan& plan,
      const ExecutionPhase& phase, PlannedPhaseExecutionScratch& scratch,
      ResultChannel<T>& channel, Fn&& fn) -> PlannedExecutionResult;
 
  template <typename World>
  friend auto merge_planned_dirty(World& world,
                                  PlannedPhaseExecutionScratch& scratch)
      -> PlannedDirtyMergeResult;
  template <typename World>
  friend auto detail::merge_planned_dirty_after_exception(
      World& world, PlannedPhaseExecutionScratch& scratch) noexcept
      -> PlannedDirtyMergeResult;
 
  void prepare(std::size_t operation_count) {
    prepare_partitions(operation_count);
    results_.assign(operation_count, PlannedExecutionResult{});
    merged_dirty_.clear();
    world_stamp_ = nullptr;
  }
 
  template <typename World>
  void prepare(const World& /*world*/, std::size_t operation_count) {
    prepare_partitions(operation_count);
    results_.assign(operation_count, PlannedExecutionResult{});
    merged_dirty_.clear();
    world_stamp_ = detail::planned_world_stamp<World>();
  }
 
  [[nodiscard]] auto dirty_for_operation(std::size_t index) noexcept
      -> PhaseDirtyPartition& {
    return dirty_partitions_[index];
  }
 
  void prepare_partitions(std::size_t operation_count) {
    dirty_partitions_.resize(operation_count);
    for (auto& partition : dirty_partitions_) {
      partition.clear();
      partition.reserve(records_per_partition_reserve_);
    }
  }
 
  void record_result(std::size_t index, PlannedExecutionResult result) {
    results_[index] = result;
  }
 
  [[nodiscard]] auto results() const noexcept
      -> std::span<const PlannedExecutionResult> {
    return results_;
  }
 
  std::vector<PhaseDirtyPartition> dirty_partitions_;
  std::vector<PlannedExecutionResult> results_;
  PlannedDirtyAccumulator merged_dirty_;
  std::size_t records_per_partition_reserve_ = 0;
  const detail::PlannedWorldStamp* world_stamp_ = nullptr;
};

class ExecutionReport {
 public:
  // plan_operations returns an ExecutionReport BY VALUE, so every accessor
  // handing out a reference or a span into it is lvalue-only. The
  // idiomatic-looking `for (const auto& op :
  // plan_operations(world, ops).plan().operations())` iterated freed
  // memory: the report died at the end of the full expression while the
  // range-for held a span into its member vector. ExecutionPhase already
  // carried a generation check, which made these unprotected accessors
  // read as safe by comparison.
  [[nodiscard]] constexpr auto operations() const& noexcept
      -> std::span<const OperationReport> {
    return {operations_.data(), operations_.size()};
  }
  auto operations() const&& -> std::span<const OperationReport> = delete;
 
  [[nodiscard]] constexpr auto plan() const& noexcept -> const ExecutionPlan& {
    return plan_;
  }
  auto plan() const&& -> const ExecutionPlan& = delete;
 
  auto find(OpHandle handle) const&& -> const OperationReport* = delete;
 
  [[nodiscard]] constexpr auto find(OpHandle handle) const& noexcept
      -> const OperationReport* {
    for (const auto& op : operations_) {
      if (op.handle == handle) {
        return &op;
      }
    }
    return nullptr;
  }
 
  [[nodiscard]] constexpr bool ok() const noexcept {
    return failed_count() == 0;
  }
 
  [[nodiscard]] constexpr bool failed() const noexcept {
    return failed_count() != 0;
  }
 
  [[nodiscard]] constexpr auto planned_count() const noexcept -> std::size_t {
    return plan_.size();
  }
 
  [[nodiscard]] constexpr auto failed_count() const noexcept -> std::size_t {
    std::size_t count = 0;
    for (const auto& op : operations_) {
      if (op.status != OperationStatus::Planned) {
        ++count;
      }
    }
    return count;
  }
 
  // Clears all results while keeping every allocation -- report rows,
  // planned operations, and their chunk lists (parked in a pool) -- so a
  // caller-owned report makes steady-state planning allocation-free.
  void reset() {
    plan_.bump_generation();
    for (auto& planned : plan_.operations_) {
      planned.chunks_.clear();
      chunk_pool_.push_back(std::move(planned.chunks_));
    }
    plan_.operations_.clear();
    operations_.clear();
    chunk_index_.clear();
    wide_operations_.clear();
  }
 
 private:
  template <typename World>
  friend auto plan_operations(const World& world,
                              std::span<const QueuedOperation> operations,
                              ExecutionReport& report)
      -> const ExecutionReport&;
 
  void reserve(std::size_t size) {
    operations_.reserve(size);
    plan_.operations_.reserve(size);
    chunk_index_.reserve(size);
  }
 
  void push_report(OperationReport report) { operations_.push_back(report); }
 
  void push_planned(PlannedOperation planned) {
    const auto op_index = static_cast<std::uint32_t>(plan_.operations_.size());
    if (detail::operation_is_indexable(planned.chunks())) {
      chunk_index_.insert(planned.chunks(), op_index);
    } else {
      wide_operations_.push_back(op_index);
    }
    plan_.operations_.push_back(std::move(planned));
  }
 
  [[nodiscard]] auto chunk_index() const noexcept
      -> const detail::ChunkOperationIndex& {
    return chunk_index_;
  }
 
  [[nodiscard]] auto wide_operations() const noexcept
      -> std::span<const std::uint32_t> {
    return {wide_operations_.data(), wide_operations_.size()};
  }
 
  template <typename World>
  [[nodiscard]] auto make_planned(const QueuedOperation& operation,
                                  std::vector<ChunkKey>&& chunks)
      -> PlannedOperation {
    return PlannedOperation{
        operation,
        std::move(chunks),
        detail::planned_world_stamp<World>(),
    };
  }
 
  [[nodiscard]] auto acquire_chunks() -> std::vector<ChunkKey> {
    if (chunk_pool_.empty()) {
      return {};
    }
    auto chunks = std::move(chunk_pool_.back());
    chunk_pool_.pop_back();
    return chunks;
  }
 
  void recycle_chunks(std::vector<ChunkKey>&& chunks) {
    chunks.clear();
    chunk_pool_.push_back(std::move(chunks));
  }
 
  void recycle_chunks(PlannedOperation&& planned) {
    recycle_chunks(std::move(planned.chunks_));
  }
 
  std::vector<OperationReport> operations_;
  ExecutionPlan plan_;
  std::vector<std::vector<ChunkKey>> chunk_pool_;
  // Chunk -> accepted-operation index, so hazard detection stays linear in
  // the operations that share a chunk rather than in the whole plan. Cleared,
  // not freed, by `reset`.
  detail::ChunkOperationIndex chunk_index_;
  // Accepted operations too wide to index, in plan order. Scanned the way
  // the planner always scanned, which for these is the cheaper path.
  std::vector<std::uint32_t> wide_operations_;
};

class OperationBatch {
 public:
  void reserve_operations(std::size_t count) { operations_.reserve(count); }
 
  [[nodiscard]] auto update_field(
      DomainDesc domain, FieldAccessDesc field_access, WritePolicy write_policy,
      Priority priority = Priority::GameplayCritical,
      BudgetPolicy budget_policy = BudgetPolicy::MustRun,
      std::source_location source = std::source_location::current())
      -> OpHandle {
    auto metadata = IntentMetadata{};
    metadata.domain = std::move(domain);
    metadata.field_access = field_access;
    metadata.write_policy = write_policy;
    metadata.priority = priority;
    metadata.budget_policy = budget_policy;
    return enqueue(OperationKind::UpdateField, std::move(metadata), {}, source);
  }
 
  [[nodiscard]] auto update_field(
      DomainDesc domain, WritePolicy write_policy,
      Priority priority = Priority::GameplayCritical,
      BudgetPolicy budget_policy = BudgetPolicy::MustRun,
      std::source_location source = std::source_location::current())
      -> OpHandle {
    return update_field(std::move(domain), FieldAccessDesc{}, write_policy,
                        priority, budget_policy, source);
  }
 
  [[nodiscard]] auto query_paths(
      PathBatchDesc desc,
      std::source_location source = std::source_location::current())
      -> PathBatchHandle {
    return PathBatchHandle{enqueue(OperationKind::QueryPaths,
                                   std::move(desc.operation), desc.requests,
                                   source)};
  }
 
  [[nodiscard]] auto query_nearest(
      NearestBatchDesc desc,
      std::source_location source = std::source_location::current())
      -> NearestBatchHandle {
    return NearestBatchHandle{enqueue(OperationKind::QueryNearest,
                                      std::move(desc.operation), desc.requests,
                                      source)};
  }
 
  [[nodiscard]] auto build_field_product(
      FieldProductDesc desc,
      std::source_location source = std::source_location::current())
      -> OpHandle {
    return enqueue(OperationKind::BuildFieldProduct, std::move(desc.operation),
                   desc.requests, source);
  }
 
  [[nodiscard]] auto move_entities(
      MoveBatchDesc desc,
      std::source_location source = std::source_location::current())
      -> OpHandle {
    return enqueue(OperationKind::MoveEntities, std::move(desc.operation),
                   desc.requests, source);
  }
 
  [[nodiscard]] auto rebuild_topology(
      TopologyRebuildDesc desc,
      std::source_location source = std::source_location::current())
      -> OpHandle {
    return enqueue(OperationKind::RebuildTopology, std::move(desc.operation),
                   {}, source);
  }
 
  [[nodiscard]] auto ensure_resident(
      ResidencyDesc desc,
      std::source_location source = std::source_location::current())
      -> OpHandle {
    return enqueue(OperationKind::EnsureResident, std::move(desc.operation),
                   desc.requests, source);
  }
 
  [[nodiscard]] auto mark_dirty(
      MarkDirtyDesc desc,
      std::source_location source = std::source_location::current())
      -> OpHandle {
    auto metadata = IntentMetadata{};
    metadata.domain = std::move(desc.domain);
    metadata.field_access.dirty_mask = desc.dirty_mask;
    metadata.invalidations.dirty_mask = desc.dirty_mask;
    return enqueue(OperationKind::MarkDirty, std::move(metadata), {}, source);
  }
 
  [[nodiscard]] auto publish_render_deltas(
      RenderDeltaDesc desc,
      std::source_location source = std::source_location::current())
      -> OpHandle {
    return enqueue(OperationKind::PublishRenderDeltas,
                   std::move(desc.operation), desc.requests, source);
  }
 
  [[nodiscard]] constexpr auto operations() const noexcept
      -> std::span<const QueuedOperation> {
    return {operations_.data(), operations_.size()};
  }
 
  [[nodiscard]] constexpr auto operation(OpHandle handle) const noexcept
      -> const QueuedOperation* {
    if (handle.value >= operations_.size()) {
      return nullptr;
    }
    return &operations_[static_cast<std::size_t>(handle.value)];
  }
 
  [[nodiscard]] constexpr bool empty() const noexcept {
    return operations_.empty();
  }
 
  [[nodiscard]] constexpr auto size() const noexcept -> std::size_t {
    return operations_.size();
  }
 
  // Clears queued operations for per-batch reuse while keeping the enqueue
  // vector's capacity, so warm batch loops re-enqueue without allocating.
  // Previously returned handles are invalidated; handle and id assignment
  // restarts at zero on the next enqueue.
  void clear() noexcept { operations_.clear(); }
 
 private:
  [[nodiscard]] auto enqueue(OperationKind kind, IntentMetadata metadata,
                             IntentPayloadView payload,
                             std::source_location source) -> OpHandle {
    const auto id = OpId{static_cast<std::uint64_t>(operations_.size())};
    const auto handle = OpHandle{id.value};
    auto operation = QueuedOperation{};
    operation.kind = kind;
    operation.handle = handle;
    operation.id = id;
    operation.domain = std::move(metadata.domain);
    operation.field_access = metadata.field_access;
    operation.write_policy = metadata.write_policy;
    operation.priority = metadata.priority;
    operation.budget_policy = metadata.budget_policy;
    operation.source = source;
    operation.payload = payload;
    operation.versions = metadata.versions;
    operation.invalidations = metadata.invalidations;
    operation.backend = metadata.backend;
    operation.exactness = metadata.exactness;
    operations_.push_back(std::move(operation));
    return handle;
  }
 
  std::vector<QueuedOperation> operations_;
};
 
namespace detail {
 
[[nodiscard]] constexpr auto operation_access(
    const QueuedOperation& op) noexcept -> OperationAccess {
  return OperationAccess{
      op.write_policy,
      op.domain.kind(),
      op.domain.mask_bits(),
  };
}
 
[[nodiscard]] constexpr bool is_valid_field_access(
    WritePolicy write_policy, FieldAccessDesc field_access) noexcept {
  if (write_policy == WritePolicy::ReadOnly && field_access.write_mask != 0) {
    return false;
  }
  return true;
}
 
template <typename World>
[[nodiscard]] auto validate_explicit_chunks(const World& world,
                                            std::span<const ChunkKey> chunks,
                                            ChunkKey& invalid_chunk) noexcept
    -> bool {
  for (const auto key : chunks) {
    if (world.try_chunk(key) == nullptr) {
      invalid_chunk = key;
      return false;
    }
  }
  return true;
}
 
// Fills `chunks` in place (clearing it first) so a caller-supplied vector
// keeps its capacity across plans instead of being replaced by a fresh
// allocation per operation.
template <typename World>
[[nodiscard]] auto expand_domain(const World& world, const DomainDesc& domain,
                                 std::vector<ChunkKey>& chunks,
                                 ChunkKey& invalid_chunk) -> bool {
  chunks.clear();
  switch (domain.kind()) {
    case DomainKind::ExplicitChunks:
      if (!validate_explicit_chunks(world, domain.explicit_chunks(),
                                    invalid_chunk)) {
        return false;
      }
      chunks.assign(domain.explicit_chunks().begin(),
                    domain.explicit_chunks().end());
      return true;
    case DomainKind::DirtyChunks:
      world.collect_dirty_chunks(domain.dirty_mask(), chunks);
      return true;
    case DomainKind::ActiveChunks:
      world.collect_active_chunks(domain.active_mask(), chunks);
      return true;
    case DomainKind::ResidentChunks:
      chunks.reserve(static_cast<std::size_t>(World::chunk_count));
      for (std::uint64_t key = 0; key < World::chunk_count; ++key) {
        chunks.push_back(ChunkKey{key});
      }
      return true;
  }
  return false;
}
 
[[nodiscard]] constexpr auto hazard_mask(FieldAccessDesc earlier,
                                         FieldAccessDesc later) noexcept
    -> std::uint32_t {
  return (earlier.write_mask & later.write_mask) |
         (earlier.write_mask & later.read_mask) |
         (earlier.read_mask & later.write_mask);
}
 
[[nodiscard]] constexpr bool chunks_overlap(
    std::span<const ChunkKey> lhs, std::span<const ChunkKey> rhs) noexcept {
  std::size_t lhs_index = 0;
  std::size_t rhs_index = 0;
  while (lhs_index < lhs.size() && rhs_index < rhs.size()) {
    const auto lhs_key = lhs[lhs_index].value;
    const auto rhs_key = rhs[rhs_index].value;
    if (lhs_key == rhs_key) {
      return true;
    }
    if (lhs_key < rhs_key) {
      ++lhs_index;
    } else {
      ++rhs_index;
    }
  }
  return false;
}
 
[[nodiscard]] constexpr auto find_hazard(
    std::span<const PlannedOperation> earlier_ops,
    const PlannedOperation& later) noexcept -> const PlannedOperation* {
  for (const auto& earlier : earlier_ops) {
    if (hazard_mask(earlier.field_access, later.field_access) == 0) {
      continue;
    }
    if (chunks_overlap(earlier.chunks(), later.chunks())) {
      // cppcheck-suppress returnDanglingLifetime
      return &earlier;
    }
  }
  return nullptr;
}
 
// Index-backed equivalent of `find_hazard`. It consults only the operations
// sharing a chunk with `later`, then keeps the earliest of those that also
// hazards on field access -- exactly the operation the linear scan returns,
// since that scan reports its first match in plan order. `conflict_handle`
// and `conflict_id` are observable, so picking any other match would be a
// behaviour change, not just a faster one.
[[nodiscard]] inline auto find_hazard_indexed(
    const ChunkOperationIndex& index, std::span<const std::uint32_t> wide_ops,
    std::span<const PlannedOperation> earlier_ops,
    const PlannedOperation& later) -> const PlannedOperation* {
  // A candidate this wide would walk one chain per chunk. The linear scan
  // it replaces rejects a non-hazarding pair on the mask alone, so for
  // these the scan is strictly cheaper and is what the planner uses.
  if (!operation_is_indexable(later.chunks())) {
    return find_hazard(earlier_ops, later);
  }
 
  auto earliest = ChunkOperationIndex::npos;
  index.for_each_sharing(later.chunks(), [&](std::uint32_t op_index) {
    if (op_index >= earliest) {
      return;
    }
    if (hazard_mask(earlier_ops[op_index].field_access, later.field_access) ==
        0) {
      return;
    }
    earliest = op_index;
  });
  // Wide operations are not in the index, so they are scanned -- mask
  // first, exactly as `find_hazard` orders it. They are in plan order, so
  // the first match is the earliest among them.
  for (const auto op_index : wide_ops) {
    if (op_index >= earliest) {
      break;
    }
    const auto& earlier = earlier_ops[op_index];
    if (hazard_mask(earlier.field_access, later.field_access) == 0) {
      continue;
    }
    if (chunks_overlap(earlier.chunks(), later.chunks())) {
      earliest = op_index;
      break;
    }
  }
  if (earliest == ChunkOperationIndex::npos) {
    return nullptr;
  }
  return &earlier_ops[earliest];
}
 
[[nodiscard]] constexpr bool is_parallel_supported_policy(
    WritePolicy policy) noexcept {
  return policy == WritePolicy::ReadOnly ||
         policy == WritePolicy::UniquePerChunk;
}
 
[[nodiscard]] constexpr bool is_mutating_policy(WritePolicy policy) noexcept {
  return policy != WritePolicy::ReadOnly;
}
 
// Plan length at or above which parallel-phase grouping builds an index
// instead of comparing every candidate against every open-phase member.
inline constexpr std::size_t phase_index_min_operations = 16;
 
// Tail of `parallel_phase_conflict` for callers that already established
// the chunk overlap -- an index lookup does, so re-walking both chunk lists
// would repeat work the lookup just did.
[[nodiscard]] constexpr bool parallel_phase_conflict_given_overlap(
    const PlannedOperation& lhs, const PlannedOperation& rhs) noexcept {
  if (is_mutating_policy(lhs.write_policy) ||
      is_mutating_policy(rhs.write_policy)) {
    return true;
  }
  return hazard_mask(lhs.field_access, rhs.field_access) != 0;
}
 
[[nodiscard]] constexpr bool parallel_phase_conflict(
    const PlannedOperation& lhs, const PlannedOperation& rhs) noexcept {
  if (!chunks_overlap(lhs.chunks(), rhs.chunks())) {
    return false;
  }
  return parallel_phase_conflict_given_overlap(lhs, rhs);
}
 
[[nodiscard]] constexpr auto dirty_axis_end(std::int64_t origin,
                                            std::uint64_t extent) noexcept
    -> std::int64_t {
  // Saturating: an unguarded origin + int64(extent) is undefined for huge
  // caller-supplied extents, so share chunk_meta's guarded helper.
  return detail::box_axis_end(origin, extent);
}
 
[[nodiscard]] constexpr auto dirty_min(std::int64_t lhs,
                                       std::int64_t rhs) noexcept
    -> std::int64_t {
  return lhs < rhs ? lhs : rhs;
}
 
[[nodiscard]] constexpr auto dirty_max(std::int64_t lhs,
                                       std::int64_t rhs) noexcept
    -> std::int64_t {
  return lhs < rhs ? rhs : lhs;
}
 
[[nodiscard]] constexpr auto dirty_union_extent(std::int64_t origin,
                                                std::int64_t end) noexcept
    -> std::uint64_t {
  // end >= origin, but a saturated INT64_MAX end paired with a negative
  // origin spans more than int64 can hold, so the subtraction must happen
  // in unsigned space, mirroring chunk_meta's union.
  return abs_delta(end, origin);
}
 
[[nodiscard]] constexpr auto union_dirty_bounds(Box3 lhs, Box3 rhs) noexcept
    -> Box3 {
  const auto min_x = dirty_min(lhs.origin.x, rhs.origin.x);
  const auto min_y = dirty_min(lhs.origin.y, rhs.origin.y);
  const auto min_z = dirty_min(lhs.origin.z, rhs.origin.z);
  const auto max_x = dirty_max(dirty_axis_end(lhs.origin.x, lhs.extent.x),
                               dirty_axis_end(rhs.origin.x, rhs.extent.x));
  const auto max_y = dirty_max(dirty_axis_end(lhs.origin.y, lhs.extent.y),
                               dirty_axis_end(rhs.origin.y, rhs.extent.y));
  const auto max_z = dirty_max(dirty_axis_end(lhs.origin.z, lhs.extent.z),
                               dirty_axis_end(rhs.origin.z, rhs.extent.z));
 
  return Box3{
      Coord3{min_x, min_y, min_z},
      Extent3{
          dirty_union_extent(min_x, max_x),
          dirty_union_extent(min_y, max_y),
          dirty_union_extent(min_z, max_z),
      },
  };
}
 
}  // namespace detail

template <typename World>
auto plan_operations(const World& world,
                     std::span<const QueuedOperation> operations,
                     ExecutionReport& report) -> const ExecutionReport& {
  // Queued operations require dense storage: ResidentChunks expands every
  // chunk key, and the executor writes through each expanded key. Reject sparse
  // worlds at compile time instead of enumerating non-resident chunks.
  static_assert(std::is_same_v<typename World::residency_type, AlwaysResident>,
                "Queued operations require an AlwaysResidentWorld; use direct "
                "sparse-world operations instead.");
  report.reset();
  report.reserve(operations.size());
 
  for (std::size_t op_index = 0; op_index < operations.size(); ++op_index) {
    const auto& op = operations[op_index];
    const auto canonical_handle =
        OpHandle{static_cast<std::uint64_t>(op_index)};
    const auto canonical_id = OpId{static_cast<std::uint64_t>(op_index)};
    OperationReport op_report{
        canonical_handle,
        canonical_id,
        OperationStatus::Planned,
        OperationFailure::None,
        detail::operation_access(op),
        op.field_access,
        op.versions,
        op.invalidations,
        op.backend,
        op.exactness,
        {},
        {},
        {},
        0,
        false,
        false,
        0,
        op.source,
    };
 
    if (op.handle != canonical_handle || op.id != canonical_id) {
      op_report.status = OperationStatus::InvalidIdentity;
      op_report.failure = op.handle != canonical_handle
                              ? OperationFailure::NonDenseHandle
                              : OperationFailure::NonDenseId;
      TESS_DIAG_TRACE_VALUE(diagnostics::TraceCategory::Planner,
                            "invalid_identity", op_index);
      report.push_report(op_report);
      continue;
    }
 
    if (!is_valid_write_policy(op.write_policy)) {
      op_report.status = OperationStatus::InvalidWritePolicy;
      op_report.failure = OperationFailure::InvalidWritePolicyValue;
      TESS_DIAG_TRACE_VALUE(diagnostics::TraceCategory::Planner,
                            "invalid_write_policy", op_index);
      report.push_report(op_report);
      continue;
    }
 
    if (!detail::is_valid_field_access(op.write_policy, op.field_access)) {
      op_report.status = OperationStatus::InvalidFieldAccess;
      op_report.failure = OperationFailure::ReadOnlyWriteMask;
      TESS_DIAG_TRACE_VALUE(diagnostics::TraceCategory::Planner,
                            "invalid_field_access", op_index);
      report.push_report(op_report);
      continue;
    }
 
    auto planned_chunks = report.acquire_chunks();
    ChunkKey invalid_chunk{};
    if (!detail::expand_domain(world, op.domain, planned_chunks,
                               invalid_chunk)) {
      op_report.status = OperationStatus::InvalidDomain;
      op_report.failure = OperationFailure::ExplicitChunkOutOfRange;
      op_report.detail_chunk = invalid_chunk;
      op_report.has_detail_chunk = true;
      TESS_DIAG_TRACE_VALUE(diagnostics::TraceCategory::Planner,
                            "invalid_domain", op_index);
      report.recycle_chunks(std::move(planned_chunks));
      report.push_report(op_report);
      continue;
    }
 
    auto planned =
        report.template make_planned<World>(op, std::move(planned_chunks));
 
    if (const auto* conflict = detail::find_hazard_indexed(
            report.chunk_index(), report.wide_operations(),
            report.plan().operations(), planned);
        conflict != nullptr) {
      op_report.status = OperationStatus::HazardConflict;
      op_report.failure = OperationFailure::FieldHazardConflict;
      op_report.conflict_handle = conflict->handle;
      op_report.conflict_id = conflict->id;
      op_report.conflict_mask =
          detail::hazard_mask(conflict->field_access, planned.field_access);
      op_report.has_conflict = true;
      op_report.chunk_count = planned.chunks().size();
      TESS_DIAG_TRACE_VALUE(diagnostics::TraceCategory::Planner, "conflict",
                            op_index);
      report.recycle_chunks(std::move(planned));
      report.push_report(op_report);
      continue;
    }
 
    op_report.chunk_count = planned.chunks().size();
    TESS_DIAG_TRACE_VALUE(diagnostics::TraceCategory::Planner, "planned",
                          op_index);
    report.push_planned(std::move(planned));
    report.push_report(op_report);
  }
 
  return report;
}
 
template <typename World>
[[nodiscard]] auto plan_operations(const World& world,
                                   std::span<const QueuedOperation> operations)
    -> ExecutionReport {
  ExecutionReport report;
  plan_operations(world, operations, report);
  return report;
}
 
template <typename World>
auto plan_operations(const World& world, const OperationBatch& ops,
                     ExecutionReport& report) -> const ExecutionReport& {
  return plan_operations(world, ops.operations(), report);
}
 
template <typename World>
[[nodiscard]] auto plan_operations(const World& world,
                                   const OperationBatch& ops)
    -> ExecutionReport {
  return plan_operations(world, ops.operations());
}

[[nodiscard]] constexpr auto planned_chunk_domain(
    const PlannedOperation& operation) noexcept -> ChunkDomain {
  return chunk_domain(operation.chunks());
}

[[nodiscard]] inline auto plan_parallel_execution_phases(
    const ExecutionPlan& plan) -> ExecutionPhasePlan {
  const auto operations = plan.operations();
  auto phases = ExecutionPhasePlan{};
  phases.reserve(operations.size());
  // Same index as hazard detection, rebuilt locally: this entry point takes
  // a plan rather than a report, so there is no long-lived one to reuse.
  // That rebuild costs two allocations, which a short plan never repays --
  // a handful of chunk-overlap checks is cheaper than the table they would
  // avoid -- so small plans keep the all-pairs comparison. No benchmark
  // covers small-plan grouping, so the cutoff is reasoned rather than
  // tuned; both paths are held to the same answer by a differential test.
  const auto indexed = operations.size() >= detail::phase_index_min_operations;
  auto index = detail::ChunkOperationIndex{};
  // Operations too wide to index are compared the way they always were.
  // See `index_max_chunks_per_operation`: a whole-domain operation would
  // otherwise cost one chain per chunk on every later comparison.
  auto wide_operations = std::vector<std::uint32_t>{};
  if (indexed) {
    index.reserve(operations.size());
  }
 
  for (std::size_t i = 0; i < operations.size(); ++i) {
    const auto& operation = operations[i];
    if (!detail::is_parallel_supported_policy(operation.write_policy)) {
      phases.status_ = ExecutionPhaseStatus::UnsupportedWritePolicy;
      phases.failed_operation_index_ = i;
      phases.failed_write_policy_ = operation.write_policy;
      TESS_DIAG_TRACE_VALUE(diagnostics::TraceCategory::Planner,
                            "unsupported_write_policy", i);
      return phases;
    }
 
    // Only members of the open phase matter, and the open phase is always
    // the suffix [first_operation, i), so filtering index hits on
    // `>= phase_first` is the same set the inner loop used to walk.
    auto conflicts = true;
    if (!phases.phases_.empty()) {
      const auto phase_first = phases.phases_.back().first_operation();
      conflicts = false;
      if (!indexed || !detail::operation_is_indexable(operation.chunks())) {
        for (std::size_t j = phase_first; j < i; ++j) {
          if (detail::parallel_phase_conflict(operations[j], operation)) {
            conflicts = true;
            break;
          }
        }
      } else {
        index.for_each_sharing(operation.chunks(), [&](std::uint32_t j) {
          if (conflicts || j < phase_first) {
            return;
          }
          conflicts = detail::parallel_phase_conflict_given_overlap(
              operations[j], operation);
        });
        for (const auto j : wide_operations) {
          if (conflicts) {
            break;
          }
          if (j < phase_first) {
            continue;
          }
          conflicts = detail::parallel_phase_conflict(operations[j], operation);
        }
      }
    }
 
    if (conflicts) {
      TESS_DIAG_TRACE_VALUE(diagnostics::TraceCategory::Planner, "new_phase",
                            i);
      phases.push_phase(plan, i, 1, operation);
    } else {
      TESS_DIAG_TRACE_VALUE(diagnostics::TraceCategory::Planner, "merged", i);
      phases.extend_last_phase(operation);
    }
    if (indexed) {
      if (detail::operation_is_indexable(operation.chunks())) {
        index.insert(operation.chunks(), static_cast<std::uint32_t>(i));
      } else {
        wide_operations.push_back(static_cast<std::uint32_t>(i));
      }
    }
  }
 
  return phases;
}

template <typename World>
// noexcept, and the only overload of this family that is: it merges an
// already-populated accumulator into the world and allocates nothing. The
// rule across the three is "noexcept iff the overload does not allocate",
// not an inconsistency -- marking the allocating ones noexcept would turn
// a std::bad_alloc into std::terminate. An -fno-exceptions consumer can
// therefore read the signature instead of the body.
[[nodiscard]] auto merge_planned_dirty(World& world,
                                       PlannedDirtyAccumulator& dirty) noexcept
    -> PlannedDirtyMergeResult {
  static_assert(std::is_same_v<typename World::residency_type, AlwaysResident>,
                "Queued-op dirty merge requires an AlwaysResidentWorld; sparse "
                "worlds must merge changes through direct operations.");
 
  const auto validation = dirty.validation_status(world);
  if (validation != PlannedDirtyMergeStatus::Merged) {
    return PlannedDirtyMergeResult{
        validation,
        0,
    };
  }
 
  auto& records = dirty.records_;
  std::sort(records.begin(), records.end(),
            [](PlannedDirtyRecord lhs, PlannedDirtyRecord rhs) {
              return lhs.chunk.value < rhs.chunk.value;
            });
 
  // Sorting makes duplicate records adjacent so they can be coalesced without
  // allocating. This noexcept path cannot replace a callback exception while
  // AutoExec unwinds.
  auto merged_count = std::size_t{0};
  for (std::size_t i = 0; i < records.size();) {
    auto chunk = records[i].chunk;
    auto dirty_mask = records[i].dirty_mask;
    auto bounds = records[i].bounds;
    ++i;
 
    while (i < records.size() && records[i].chunk == chunk) {
      dirty_mask |= records[i].dirty_mask;
      bounds = detail::union_dirty_bounds(bounds, records[i].bounds);
      ++i;
    }
 
    world.mark_dirty(chunk, dirty_mask, bounds);
    ++merged_count;
  }
 
  TESS_DIAG_EVENT_VALUE(queued_dirty_merge, merged_count);
  dirty.clear();
  return PlannedDirtyMergeResult{
      PlannedDirtyMergeStatus::Merged,
      merged_count,
  };
}

[[nodiscard]] inline auto collect_planned_dirty(
    PlannedDirtyAccumulator& dirty, PlannedDirtyPartitions& partitions)
    -> PlannedDirtyCollectResult {
  auto* world_stamp = dirty.world_stamp_;
  for (const auto& partition : partitions.partitions_) {
    const auto* partition_stamp = partition.world_stamp_;
    if (partition_stamp == nullptr) {
      continue;
    }
    if (world_stamp == nullptr) {
      world_stamp = partition_stamp;
      continue;
    }
    if (world_stamp->shape_identity != partition_stamp->shape_identity) {
      return PlannedDirtyCollectResult{
          PlannedDirtyCollectStatus::InvalidShape,
          0,
      };
    }
    if (world_stamp->chunk_limit != partition_stamp->chunk_limit) {
      return PlannedDirtyCollectResult{
          PlannedDirtyCollectStatus::InvalidChunk,
          0,
      };
    }
  }
 
  const auto record_limit =
      detail::effective_capacity_limit(dirty.records_.max_size());
  if (dirty.records_.size() > record_limit) {
    return PlannedDirtyCollectResult{
        PlannedDirtyCollectStatus::CapacityExceeded,
        0,
    };
  }
  auto required_capacity = dirty.records_.size();
  auto record_count = std::size_t{0};
  for (const auto& partition : partitions.partitions_) {
    const auto partition_size = partition.records_.size();
    if (partition_size > record_limit - required_capacity) {
      return PlannedDirtyCollectResult{
          PlannedDirtyCollectStatus::CapacityExceeded,
          0,
      };
    }
    required_capacity += partition_size;
    record_count += partition_size;
  }
  // Complete the only potentially allocating step before clearing a source;
  // a failed reserve therefore preserves both destination and partitions.
  dirty.records_.reserve(required_capacity);
 
  for (auto& partition : partitions.partitions_) {
    if (partition.world_stamp_ != nullptr) {
      dirty.world_stamp_ = partition.world_stamp_;
    }
    dirty.records_.insert(dirty.records_.end(), partition.records_.begin(),
                          partition.records_.end());
    partition.clear();
  }
  TESS_DIAG_EVENT_VALUE(queued_dirty_collect, record_count);
  return PlannedDirtyCollectResult{
      PlannedDirtyCollectStatus::Collected,
      record_count,
  };
}
 
template <typename World>
// Deliberately NOT noexcept: collect_planned_dirty reserves the
// destination accumulator, so this can throw std::bad_alloc. See the
// accumulator overload for the rule.
[[nodiscard]] auto merge_planned_dirty(World& world,
                                       PlannedDirtyAccumulator& dirty_scratch,
                                       PlannedDirtyPartitions& partitions)
    -> PlannedDirtyMergeResult {
  for (const auto& partition : partitions.partitions()) {
    const auto validation = partition.validation_status(world);
    if (validation != PlannedDirtyMergeStatus::Merged) {
      return PlannedDirtyMergeResult{validation, 0};
    }
  }
  dirty_scratch.clear();
  const auto collected = collect_planned_dirty(dirty_scratch, partitions);
  if (!collected.ok()) {
    if (collected.status == PlannedDirtyCollectStatus::CapacityExceeded) {
      return PlannedDirtyMergeResult{
          PlannedDirtyMergeStatus::CapacityExceeded,
          0,
      };
    }
    return PlannedDirtyMergeResult{
        collected.status == PlannedDirtyCollectStatus::InvalidShape
            ? PlannedDirtyMergeStatus::InvalidShape
            : PlannedDirtyMergeStatus::InvalidChunk,
        0,
    };
  }
  return merge_planned_dirty(world, dirty_scratch);
}
 
template <typename World>
[[nodiscard]] auto merge_planned_dirty(World& world,
                                       PlannedPhaseExecutionScratch& scratch)
    -> PlannedDirtyMergeResult {
  if (scratch.world_stamp_ == nullptr) {
    return PlannedDirtyMergeResult{
        PlannedDirtyMergeStatus::Merged,
        0,
    };
  }
  const auto validation =
      detail::validate_planned_world_stamp<World>(scratch.world_stamp_);
  if (validation != PlannedExecutionStatus::Executed) {
    return PlannedDirtyMergeResult{
        validation == PlannedExecutionStatus::InvalidShape
            ? PlannedDirtyMergeStatus::InvalidShape
            : PlannedDirtyMergeStatus::InvalidChunk,
        0,
    };
  }
 
  auto& merged = scratch.merged_dirty_;
  const auto record_limit =
      detail::effective_capacity_limit(merged.records_.max_size());
  auto record_count = std::size_t{0};
  for (const auto& partition : scratch.dirty_partitions_) {
    const auto partition_size = partition.records().size();
    if (record_count > record_limit ||
        partition_size > record_limit - record_count) {
      return PlannedDirtyMergeResult{
          PlannedDirtyMergeStatus::CapacityExceeded,
          0,
      };
    }
    record_count += partition_size;
  }
  merged.clear();
  // Reserve before transferring anything, so allocation failure leaves every
  // phase partition available to the caller.
  merged.records_.reserve(record_count);
  merged.world_stamp_ = scratch.world_stamp_;
  for (auto& partition : scratch.dirty_partitions_) {
    const auto records = partition.records();
    merged.records_.insert(merged.records_.end(), records.begin(),
                           records.end());
    partition.clear();
  }
  TESS_DIAG_EVENT_VALUE(queued_dirty_collect, record_count);
  (void)record_count;
  return merge_planned_dirty(world, merged);
}

template <typename World>
auto detail::merge_planned_dirty_after_exception(
    World& world, PlannedPhaseExecutionScratch& scratch) noexcept
    -> PlannedDirtyMergeResult {
  if (scratch.world_stamp_ == nullptr) {
    return PlannedDirtyMergeResult{
        PlannedDirtyMergeStatus::Merged,
        0,
    };
  }
  const auto validation =
      detail::validate_planned_world_stamp<World>(scratch.world_stamp_);
  if (validation != PlannedExecutionStatus::Executed) {
    return PlannedDirtyMergeResult{
        validation == PlannedExecutionStatus::InvalidShape
            ? PlannedDirtyMergeStatus::InvalidShape
            : PlannedDirtyMergeStatus::InvalidChunk,
        0,
    };
  }
 
  // This cold path must preserve the original callback exception. Coalesce
  // with an allocation-free quadratic scan: exceptions are rare, and normal
  // phase sizes should not dictate whether failed work remains observable.
  auto record_count = std::size_t{0};
  for (const auto& partition : scratch.dirty_partitions_) {
    const auto partition_size = partition.records().size();
    if (partition_size >
        std::numeric_limits<std::size_t>::max() - record_count) {
      record_count = std::numeric_limits<std::size_t>::max();
      break;
    }
    record_count += partition_size;
  }
  auto merged_count = std::size_t{0};
  for (std::size_t partition_index = 0;
       partition_index < scratch.dirty_partitions_.size(); ++partition_index) {
    const auto records = scratch.dirty_partitions_[partition_index].records();
    for (std::size_t record_index = 0; record_index < records.size();
         ++record_index) {
      const auto record = records[record_index];
      auto appeared_earlier = false;
      for (std::size_t earlier_partition = 0;
           earlier_partition <= partition_index && !appeared_earlier;
           ++earlier_partition) {
        const auto earlier_records =
            scratch.dirty_partitions_[earlier_partition].records();
        const auto earlier_count = earlier_partition == partition_index
                                       ? record_index
                                       : earlier_records.size();
        for (std::size_t earlier_index = 0; earlier_index < earlier_count;
             ++earlier_index) {
          if (earlier_records[earlier_index].chunk == record.chunk) {
            appeared_earlier = true;
            break;
          }
        }
      }
      if (appeared_earlier) {
        continue;
      }
 
      auto dirty_mask = record.dirty_mask;
      auto bounds = record.bounds;
      for (std::size_t later_partition = partition_index;
           later_partition < scratch.dirty_partitions_.size();
           ++later_partition) {
        const auto later_records =
            scratch.dirty_partitions_[later_partition].records();
        const auto first_later = later_partition == partition_index
                                     ? record_index + 1
                                     : std::size_t{0};
        for (std::size_t later_index = first_later;
             later_index < later_records.size(); ++later_index) {
          const auto later = later_records[later_index];
          if (later.chunk == record.chunk) {
            dirty_mask |= later.dirty_mask;
            bounds = detail::union_dirty_bounds(bounds, later.bounds);
          }
        }
      }
      world.mark_dirty(record.chunk, dirty_mask, bounds);
      ++merged_count;
    }
  }
  for (auto& partition : scratch.dirty_partitions_) {
    partition.clear();
  }
  TESS_DIAG_EVENT_VALUE(queued_dirty_collect, record_count);
  TESS_DIAG_EVENT_VALUE(queued_dirty_merge, merged_count);
  (void)record_count;
  (void)merged_count;
  return PlannedDirtyMergeResult{
      PlannedDirtyMergeStatus::Merged,
      merged_count,
  };
}

template <WritePolicy Policy>
[[nodiscard]] constexpr bool planned_policy_matches(
    const PlannedOperation& operation) noexcept {
  return operation.write_policy == Policy;
}

template <WritePolicy Policy, typename World>
[[nodiscard]] auto validate_planned_operation(
    const World& world, const PlannedOperation& operation) noexcept
    -> PlannedExecutionStatus {
  const auto world_status = operation.world_validation_status(world);
  if (world_status != PlannedExecutionStatus::Executed) {
    return world_status;
  }
  if (!planned_policy_matches<Policy>(operation)) {
    return PlannedExecutionStatus::PolicyMismatch;
  }
  return PlannedExecutionStatus::Executed;
}

template <WritePolicy Policy, typename World>
[[nodiscard]] constexpr auto try_planned_block_ctx(
    World& world, const PlannedOperation& operation) noexcept
    -> std::optional<BlockCtx<World, Policy>> {
  static_assert(
      std::is_same_v<typename World::residency_type, AlwaysResident>,
      "Queued-op execution requires an AlwaysResidentWorld; use direct "
      "sparse-world operations instead.");
  if (validate_planned_operation<Policy>(world, operation) !=
      PlannedExecutionStatus::Executed) {
    return std::nullopt;
  }
  return block_ctx<Policy>(world, planned_chunk_domain(operation));
}

template <WritePolicy Policy, typename World, typename Fn>
[[nodiscard]] auto execute_planned_operation(World& world,
                                             const PlannedOperation& operation,
                                             Fn&& fn)
    -> PlannedExecutionResult {
  const auto validation = validate_planned_operation<Policy>(world, operation);
  if (validation != PlannedExecutionStatus::Executed) {
    return PlannedExecutionResult{
        validation,
        0,
    };
  }
  auto ctx = block_ctx<Policy>(world, planned_chunk_domain(operation));
 
  std::size_t chunk_count = 0;
  auto&& callback = fn;
  ctx.for_each_chunk([&](auto view) {
    if (operation.field_access.dirty_mask) {
      world.mark_dirty(view.key(), operation.field_access.dirty_mask,
                       view.bounds());
    }
    callback(view);
    ++chunk_count;
  });
 
  return PlannedExecutionResult{
      PlannedExecutionStatus::Executed,
      chunk_count,
  };
}

template <WritePolicy Policy, typename World, typename Fn>
[[nodiscard]] auto execute_planned_operation_deferred_dirty(
    World& world, const PlannedOperation& operation,
    PlannedDirtyAccumulator& dirty, Fn&& fn) -> PlannedExecutionResult {
  const auto validation = validate_planned_operation<Policy>(world, operation);
  if (validation != PlannedExecutionStatus::Executed) {
    return PlannedExecutionResult{
        validation,
        0,
    };
  }
  if (operation.field_access.dirty_mask) {
    const auto dirty_validation = dirty.validation_status(world);
    if (dirty_validation != PlannedDirtyMergeStatus::Merged) {
      return PlannedExecutionResult{
          dirty_validation == PlannedDirtyMergeStatus::InvalidShape
              ? PlannedExecutionStatus::InvalidShape
              : PlannedExecutionStatus::InvalidChunk,
          0,
      };
    }
  }
  return detail::execute_validated_planned_operation_deferred_dirty<true,
                                                                    Policy>(
      world, operation, dirty, std::forward<Fn>(fn));
}

template <WritePolicy Policy, typename World, typename Fn>
[[nodiscard]] auto execute_plan(World& world, const ExecutionPlan& plan,
                                Fn&& fn) -> PlannedExecutionResult {
  std::size_t chunk_count = 0;
  auto&& callback = fn;
  for (const auto& operation : plan.operations()) {
    auto result = execute_planned_operation<Policy>(world, operation, callback);
    if (result.status != PlannedExecutionStatus::Executed) {
      return PlannedExecutionResult{
          result.status,
          chunk_count + result.chunk_count,
      };
    }
    chunk_count += result.chunk_count;
  }
  return PlannedExecutionResult{
      PlannedExecutionStatus::Executed,
      chunk_count,
  };
}

template <WritePolicy Policy, typename World, typename Fn>
[[nodiscard]] auto execute_plan_deferred_dirty(World& world,
                                               const ExecutionPlan& plan,
                                               PlannedDirtyAccumulator& dirty,
                                               Fn&& fn)
    -> PlannedExecutionResult {
  std::size_t chunk_count = 0;
  auto&& callback = fn;
  for (const auto& operation : plan.operations()) {
    auto result = execute_planned_operation_deferred_dirty<Policy>(
        world, operation, dirty, callback);
    if (result.status != PlannedExecutionStatus::Executed) {
      return PlannedExecutionResult{
          result.status,
          chunk_count + result.chunk_count,
      };
    }
    chunk_count += result.chunk_count;
  }
  return PlannedExecutionResult{
      PlannedExecutionStatus::Executed,
      chunk_count,
  };
}
 
template <WritePolicy Policy, typename Executor, typename World, typename Fn>
  requires SerialExecutor<Executor>
[[nodiscard]] auto execute_phase_deferred_dirty_with(
    Executor&& executor, World& world, const ExecutionPlan& plan,
    const ExecutionPhase& phase, PlannedDirtyAccumulator& dirty, Fn&& fn)
    -> PlannedExecutionResult {
  const auto operations = plan.operations();
  const auto phase_validation =
      detail::execution_phase_validation_status<Policy>(world, plan, phase);
  if (phase_validation != PlannedExecutionStatus::Executed) {
    detail::record_execution_phase_validation_failure(phase_validation);
    return PlannedExecutionResult{
        phase_validation,
        0,
    };
  }
  const auto dirty_validation = dirty.validation_status(world);
  if (dirty_validation != PlannedDirtyMergeStatus::Merged) {
    TESS_DIAG_EVENT(queued_phase_failure);
    return PlannedExecutionResult{
        dirty_validation == PlannedDirtyMergeStatus::InvalidShape
            ? PlannedExecutionStatus::InvalidShape
            : PlannedExecutionStatus::InvalidChunk,
        0,
    };
  }
 
  TESS_DIAG_EVENT_VALUE(queued_phase_execute, phase.operation_count());
  std::size_t chunk_count = 0;
  auto&& callback = fn;
  auto result = execute_operation_index_range(
      std::forward<Executor>(executor), executor_phase_range(phase),
      [&](std::size_t index) {
        auto operation_result =
            detail::execute_validated_planned_operation_deferred_dirty<true,
                                                                       Policy>(
                world, operations[index], dirty, callback);
        if (operation_result.status == PlannedExecutionStatus::Executed) {
          chunk_count += operation_result.chunk_count;
        }
        return operation_result;
      });
  if (result.status != PlannedExecutionStatus::Executed) {
    TESS_DIAG_EVENT(queued_phase_failure);
    result.chunk_count = chunk_count;
    return result;
  }
 
  return PlannedExecutionResult{
      PlannedExecutionStatus::Executed,
      chunk_count,
  };
}

template <WritePolicy Policy, typename Executor, typename World, typename Fn>
[[nodiscard]] auto execute_phase_partitioned_dirty_with(
    Executor&& executor, World& world, const ExecutionPlan& plan,
    const ExecutionPhase& phase, PlannedPhaseExecutionScratch& scratch, Fn&& fn)
    -> PlannedExecutionResult {
  const auto operations = plan.operations();
  const auto phase_validation =
      detail::execution_phase_validation_status<Policy>(world, plan, phase);
  if (phase_validation != PlannedExecutionStatus::Executed) {
    detail::record_execution_phase_validation_failure(phase_validation);
    return PlannedExecutionResult{
        phase_validation,
        0,
    };
  }
 
  TESS_DIAG_EVENT_VALUE(queued_phase_execute, phase.operation_count());
  TESS_DIAG_EVENT_VALUE(queued_partitioned_phase, phase.operation_count());
  scratch.prepare(world, phase.operation_count());
  auto&& callback = fn;
  constexpr auto no_throw_callback =
      detail::planned_callback_is_nothrow<Policy, World, decltype(callback)>;
  for (std::size_t offset = 0; offset < phase.operation_count(); ++offset) {
    const auto index = phase.first_operation() + offset;
    scratch.dirty_for_operation(offset).reserve(
        !operations[index].field_access.dirty_mask
            ? 0
            : operations[index].chunks().size());
  }
  auto result = execute_operation_index_range(
      std::forward<Executor>(executor), executor_phase_range(phase),
      // Each partition reserved one record per possible chunk visit above;
      // push_back therefore cannot allocate. Clang-tidy 22 does not carry
      // that capacity proof through std::vector.
      // NOLINTNEXTLINE(bugprone-exception-escape)
      [&](std::size_t index) noexcept(no_throw_callback) {
        const auto offset = index - phase.first_operation();
        auto operation_result =
            detail::execute_validated_phase_operation_deferred_dirty<Policy>(
                world, operations[index], scratch.dirty_for_operation(offset),
                callback);
        scratch.record_result(offset, operation_result);
        return operation_result;
      });
 
  std::size_t chunk_count = 0;
  for (const auto operation_result : scratch.results()) {
    if (operation_result.status != PlannedExecutionStatus::Executed) {
      TESS_DIAG_EVENT(queued_phase_failure);
      return PlannedExecutionResult{
          operation_result.status,
          chunk_count,
      };
    }
    chunk_count += operation_result.chunk_count;
  }
 
  if (result.status != PlannedExecutionStatus::Executed) {
    TESS_DIAG_EVENT(queued_phase_failure);
    return PlannedExecutionResult{
        result.status,
        chunk_count,
    };
  }
 
  return PlannedExecutionResult{
      PlannedExecutionStatus::Executed,
      chunk_count,
  };
}

template <WritePolicy Policy, typename World, typename Fn>
[[nodiscard]] auto execute_phase_deferred_dirty(World& world,
                                                const ExecutionPlan& plan,
                                                const ExecutionPhase& phase,
                                                PlannedDirtyAccumulator& dirty,
                                                Fn&& fn)
    -> PlannedExecutionResult {
  const SerialPhaseExecutor executor;
  return execute_phase_deferred_dirty_with<Policy>(executor, world, plan, phase,
                                                   dirty, std::forward<Fn>(fn));
}
 
}  // namespace tess