rvm: add debug-mode invariant assertions

Encode VM stack/context/register lifecycle invariants as
debug_assert!s. Zero cost in release; surfaces violations during
debug-mode tests and CI.

Invariants covered:
- reset_execution_state postcondition: all stacks empty, registers
  resized to base and Undefined, rule_cache reset, pc/executed counters
  zeroed, builtins_cache cleared, execution_state Ready.
- Per-opcode invariant check (assert_vm_invariants) invoked at the top
  of run_stackless_loop and jump_to iterations: state is Ready/Running,
  registers non-empty, rule_cache sized to program, execution stack
  bounded.
- resume() precondition: execution_state is Suspended.
- execute_suspendable_entry precondition: clean state (callers reset
  immediately before).
- ComprehensionEnd dispatch: comprehension_stack non-empty.
- Rule finalize / call_rule completion: call_rule_stack non-empty and
  popped frame's rule_index matches.
- IterationState::advance: Single iterator not advanced past consumption,
  Array index not at usize::MAX before saturating_add.

All gated by #[cfg(debug_assertions)] (directly or via debug_assert!)
so release builds are unaffected.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: anakrish

src/rvm/vm/comprehension.rs

@@ -528,7 +528,6 @@ impl RegoVM {
             Ok(false)
         }
     }
-
     pub(super) fn handle_comprehension_condition_failure_suspendable(&mut self) -> Result<bool> {
         if let Some(mut frame) = self.execution_stack.pop() {
             let handled = if let &mut FrameKind::Comprehension {
@@ -606,6 +605,13 @@ impl RegoVM {
     }
 
     fn execute_comprehension_end_run_to_completion(&mut self) -> Result<()> {
+        // A ComprehensionEnd instruction can only be reached from inside an
+        // active comprehension; the stack must be non-empty here.
+        debug_assert!(
+            !self.comprehension_stack.is_empty(),
+            "ComprehensionEnd reached with empty comprehension_stack at pc {}",
+            self.pc
+        );
         self.comprehension_stack.pop().map_or_else(
             || {
                 Err(VmError::InvalidIteration {

src/rvm/vm/context.rs

@@ -54,6 +54,10 @@ impl IterationState {
     pub(super) const fn advance(&mut self) {
         match *self {
             Self::Array { ref mut index, .. } => {
+                debug_assert!(
+                    *index < usize::MAX,
+                    "IterationState::Array index overflow on advance"
+                );
                 *index = index.saturating_add(1);
             }
             Self::Object {
@@ -69,6 +73,10 @@ impl IterationState {
             Self::Single {
                 ref mut consumed, ..
             } => {
+                debug_assert!(
+                    !*consumed,
+                    "IterationState::Single advanced after consumption"
+                );
                 *consumed = true;
             }
         }

src/rvm/vm/execution.rs

@@ -117,6 +117,7 @@ impl RegoVM {
         let target = self.convert_pc(target, "jump target")?;
         self.pc = target;
         while self.pc < program.instructions.len() {
+            self.assert_vm_invariants();
             self.memory_check()?;
             if self.executed_instructions >= self.max_instructions {
                 return Err(VmError::InstructionLimitExceeded {
@@ -189,6 +190,9 @@ impl RegoVM {
     }
 
     fn execute_suspendable_entry(&mut self, entry_point_pc: usize) -> Result<Value> {
+        // Precondition: callers (execute_entry_point_by_{index,name}) reset the
+        // VM before invoking this method, so the VM must be in a clean state.
+        self.debug_assert_state_is_clean();
         self.execution_state = ExecutionState::Running;
         self.reset_execution_timer_state();
         match self.run_stackless_from(entry_point_pc) {
@@ -201,6 +205,12 @@ impl RegoVM {
     }
 
     pub fn resume(&mut self, resume_value: Option<Value>) -> Result<Value> {
+        // Precondition: resume only makes sense from a Suspended state.
+        debug_assert!(
+            matches!(self.execution_state, ExecutionState::Suspended { .. }),
+            "resume precondition: execution_state must be Suspended, was {:?}",
+            self.execution_state
+        );
         let (reason, mut last_result) = match self.execution_state.clone() {
             ExecutionState::Suspended {
                 reason,
@@ -289,6 +299,7 @@ impl RegoVM {
 
     fn run_stackless_loop(&mut self, program: &Program, last_result: &mut Value) -> Result<()> {
         while !self.execution_stack.is_empty() {
+            self.assert_vm_invariants();
             self.memory_check()?;
             self.frame_pc_overridden = false;
             let should_finalize_rule = self.execution_stack.last().is_some_and(|frame| {

src/rvm/vm/rules.rs

@@ -277,6 +277,11 @@ impl RegoVM {
             .call_rule_stack
             .pop()
             .ok_or(VmError::CallRuleStackUnderflow { pc: self.pc })?;
+        debug_assert_eq!(
+            call_context.rule_index, rule_index,
+            "call_rule_stack pop mismatch: expected rule_index {} got {}",
+            rule_index, call_context.rule_index,
+        );
         self.pc = call_context.return_pc;
 
         let result_from_rule = if !rule_failed_due_to_inconsistency {
@@ -831,6 +836,13 @@ impl RegoVM {
 
         self.registers = parent_registers;
 
+        // Finalizing a rule frame must always have a matching CallRuleContext
+        // pushed by execute_call_rule_suspendable; underflow is a state-leak bug.
+        debug_assert!(
+            !self.call_rule_stack.is_empty(),
+            "rule finalize: call_rule_stack must be non-empty at pc {}",
+            self.pc
+        );
         if self.call_rule_stack.pop().is_none() {
             return Err(VmError::CallRuleStackUnderflow { pc: self.pc });
         }

src/rvm/vm/state.rs

@@ -34,6 +34,125 @@ impl RegoVM {
 
         // Builtin cache entries only live for a single execution
         self.builtins_cache.clear();
+
+        // Postcondition: every stack/cache that `reset_execution_state` touches
+        // must be in its documented "clean" shape. This catches accidental
+        // omissions in future edits to this function.
+        self.debug_assert_state_is_clean();
+    }
+
+    /// Debug-only postcondition for `reset_execution_state`.
+    ///
+    /// Asserts the invariants every caller of `reset_execution_state` relies on
+    /// before starting a fresh execution. The body is fully gated by
+    /// `#[cfg(debug_assertions)]` so this is a zero-cost no-op in release.
+    #[inline]
+    pub(super) fn debug_assert_state_is_clean(&self) {
+        #[cfg(debug_assertions)]
+        {
+            debug_assert!(
+                self.execution_stack.is_empty(),
+                "reset_execution_state postcondition: execution_stack must be empty"
+            );
+            debug_assert!(
+                self.loop_stack.is_empty(),
+                "reset_execution_state postcondition: loop_stack must be empty"
+            );
+            debug_assert!(
+                self.comprehension_stack.is_empty(),
+                "reset_execution_state postcondition: comprehension_stack must be empty"
+            );
+            debug_assert!(
+                self.call_rule_stack.is_empty(),
+                "reset_execution_state postcondition: call_rule_stack must be empty"
+            );
+            debug_assert!(
+                self.register_stack.is_empty(),
+                "reset_execution_state postcondition: register_stack must be empty"
+            );
+            debug_assert!(
+                self.builtins_cache.is_empty(),
+                "reset_execution_state postcondition: builtins_cache must be empty"
+            );
+            debug_assert_eq!(
+                self.registers.len(),
+                self.base_register_count,
+                "reset_execution_state postcondition: registers must be sized to base_register_count"
+            );
+            debug_assert!(
+                self.registers.iter().all(|v| matches!(v, Value::Undefined)),
+                "reset_execution_state postcondition: all registers must be Undefined"
+            );
+            debug_assert_eq!(
+                self.rule_cache.len(),
+                self.program.rule_infos.len(),
+                "reset_execution_state postcondition: rule_cache size must match program rule_infos"
+            );
+            debug_assert!(
+                self.rule_cache.iter().all(|entry| !entry.0),
+                "reset_execution_state postcondition: rule_cache entries must be uncomputed"
+            );
+            debug_assert_eq!(
+                self.pc, 0,
+                "reset_execution_state postcondition: pc must be 0"
+            );
+            debug_assert_eq!(
+                self.executed_instructions, 0,
+                "reset_execution_state postcondition: executed_instructions must be 0"
+            );
+            debug_assert!(
+                matches!(self.execution_state, ExecutionState::Ready),
+                "reset_execution_state postcondition: execution_state must be Ready"
+            );
+        }
+    }
+
+    /// Per-opcode VM invariants checked from the inner dispatch loop.
+    ///
+    /// These hold every time control re-enters the dispatch loop with another
+    /// instruction to execute. Fully `#[cfg(debug_assertions)]`-gated so the
+    /// method body compiles out in release.
+    #[inline]
+    pub(super) fn assert_vm_invariants(&self) {
+        #[cfg(debug_assertions)]
+        {
+            // The dispatch loop only runs while execution is live. Once the VM
+            // has transitioned to a terminal state (Suspended/Completed/Error)
+            // the loop must have exited. Note `Ready` is also valid here because
+            // some entry points (e.g. `execute_entry_point_by_index` in
+            // RunToCompletion mode) drive `jump_to` without flipping the state.
+            debug_assert!(
+                matches!(
+                    self.execution_state,
+                    ExecutionState::Ready | ExecutionState::Running
+                ),
+                "vm invariant: execution_state must be Ready or Running inside the dispatch loop, was {:?}",
+                self.execution_state
+            );
+
+            // Registers must always be non-empty when the dispatch loop is
+            // running — every instruction operates on registers.
+            debug_assert!(
+                !self.registers.is_empty(),
+                "vm invariant: register window must be non-empty inside the dispatch loop"
+            );
+
+            // Rule cache is sized once at reset and must not change shape mid-run.
+            debug_assert_eq!(
+                self.rule_cache.len(),
+                self.program.rule_infos.len(),
+                "vm invariant: rule_cache size must equal program.rule_infos size"
+            );
+
+            // Bound the execution stack. There is no hard MAX_FRAMES today;
+            // 4096 is a generous safety net well above any legitimate test or
+            // policy nesting depth and catches runaway frame-leaks early.
+            debug_assert!(
+                self.execution_stack.len() <= 4096,
+                "vm invariant: execution_stack depth ({}) exceeds sanity bound",
+                self.execution_stack.len(),
+            );
+        }
     }
 
     /// Return all active objects to their respective pools for reuse