feat(rvm): add context/metadata support, new instructions, and serialization v6

Extend the RVM infrastructure with evaluation context, program metadata,
and several new instructions needed by language frontends beyond Rego.

VM & instructions:
- Add LoadContext / LoadMetadata instructions for accessing evaluation
  context and program metadata from within RVM programs
- Add ArrayPushDefined (skip-undefined variant of ArrayPush) for
  wildcard alias collection where absent properties should be excluded
- Add ReturnUndefinedIfNotTrue for early-return semantics without
  triggering VM assertion failures
- Add CoalesceUndefinedToNull for languages where missing fields are
  null rather than undefined
- Support IterationState::Single in loops and comprehensions for
  iterating over scalar values alongside collections

Program metadata:
- Extend ProgramMetadata with language identifier, typed annotations
  (BTreeMap<String, MetadataValue>), and to_value() conversion
- Add MetadataValue enum (String, Bool, Integer, Float, Array, Object)
  with full serde support
- Add has_host_await flag to Program with recompute_host_await_presence()

Serialization:
- Bump binary format version to 6
- Add JSON serialization for ProgramMetadata including annotations

Compiler:
- Track has_host_await during compilation
- Minor import cleanups

Test infrastructure:
- Extend VM test harness with context, metadata_language, and
  metadata_annotations fields
- Add instruction parser support for new instructions
- Add assembly listing support (readable, tabular, compact formats)
- Add 3 new YAML test suites: load_context_metadata,
  coalesce_undefined_to_null, return_undefined_if_not_true

Author: anakrish

src/languages/rego/compiler/core.rs

@@ -259,6 +259,9 @@ impl<'a> Compiler<'a> {
     }
 
     pub fn emit_instruction(&mut self, instruction: Instruction, span: &Span) {
+        if matches!(instruction, Instruction::HostAwait { .. }) {
+            self.program.has_host_await = true;
+        }
         self.program.instructions.push(instruction);
 
         let source_path = span.source.get_path().to_string();

src/languages/rego/compiler/mod.rs

@@ -23,8 +23,8 @@ pub use error::{CompilerError, Result, SpannedCompilerError};
 use crate::ast::ExprRef;
 use crate::lexer::Span;
 use crate::rvm::program::{Program, RuleType, SpanInfo};
-use crate::value::Value;
 use crate::CompiledPolicy;
+use crate::Value;
 use alloc::collections::{BTreeMap, BTreeSet};
 use alloc::string::String;
 use alloc::vec;

src/rvm/instructions/display.rs

@@ -143,6 +143,8 @@ impl core::fmt::Display for Instruction {
             Instruction::LoadBool { dest, value } => format!("LOAD_BOOL R({}) {}", dest, value),
             Instruction::LoadData { dest } => format!("LOAD_DATA R({})", dest),
             Instruction::LoadInput { dest } => format!("LOAD_INPUT R({})", dest),
+            Instruction::LoadContext { dest } => format!("LOAD_CONTEXT R({})", dest),
+            Instruction::LoadMetadata { dest } => format!("LOAD_METADATA R({})", dest),
             Instruction::Move { dest, src } => format!("MOVE R({}) R({})", dest, src),
             Instruction::Add { dest, left, right } => {
                 format!("ADD R({}) R({}) R({})", dest, left, right)
@@ -220,6 +222,9 @@ impl core::fmt::Display for Instruction {
             }
             Instruction::ArrayNew { dest } => format!("ARRAY_NEW R({})", dest),
             Instruction::ArrayPush { arr, value } => format!("ARRAY_PUSH R({}) R({})", arr, value),
+            Instruction::ArrayPushDefined { arr, value } => {
+                format!("ARRAY_PUSH_DEFINED R({}) R({})", arr, value)
+            }
             Instruction::ArrayCreate { params_index } => {
                 format!("ARRAY_CREATE P({})", params_index)
             }
@@ -247,6 +252,12 @@ impl core::fmt::Display for Instruction {
                 };
                 format!("{} R({})", name, register)
             }
+            Instruction::ReturnUndefinedIfNotTrue { condition } => {
+                format!("RETURN_UNDEFINED_IF_NOT_TRUE R({})", condition)
+            }
+            Instruction::CoalesceUndefinedToNull { register } => {
+                format!("COALESCE_UNDEF_TO_NULL R({})", register)
+            }
             Instruction::LoopStart { params_index } => {
                 format!("LOOP_START P({})", params_index)
             }

src/rvm/instructions/mod.rs

@@ -54,6 +54,16 @@ pub enum Instruction {
         dest: u8,
     },
 
+    /// Load evaluation context object into register
+    LoadContext {
+        dest: u8,
+    },
+
+    /// Load program metadata object into register
+    LoadMetadata {
+        dest: u8,
+    },
+
     /// Move value from one register to another
     Move {
         dest: u8,
@@ -206,6 +216,16 @@ pub enum Instruction {
         value: u8,
     },
 
+    /// Push element to array, but skip if the value is undefined.
+    ///
+    /// Used by Azure Policy's `field('alias[*].property')` wildcard collection
+    /// so that absent nested properties are excluded from the collected array
+    /// rather than producing undefined entries.
+    ArrayPushDefined {
+        arr: u8,
+        value: u8,
+    },
+
     /// Create array from registers - returns undefined if any element is undefined
     ArrayCreate {
         /// Index into program's instruction_data.array_create_params table
@@ -254,6 +274,23 @@ pub enum Instruction {
         mode: GuardMode,
     },
 
+    /// Return undefined immediately when condition is false or undefined.
+    ///
+    /// This is used by Azure Policy compilation to model "condition does not match"
+    /// without treating it as a VM assertion failure.
+    ReturnUndefinedIfNotTrue {
+        condition: u8,
+    },
+
+    /// Replace Undefined with Null in a register.
+    ///
+    /// Azure Policy treats missing fields as null rather than undefined.
+    /// This instruction prevents the RVM's undefined-propagation from
+    /// short-circuiting subsequent builtin calls.
+    CoalesceUndefinedToNull {
+        register: u8,
+    },
+
     /// Start a loop over a collection with specified semantics - uses parameter table
     LoopStart {
         /// Index into program's instruction_data.loop_params table

src/rvm/program/core.rs

@@ -70,6 +70,11 @@ pub struct Program {
     /// Flag indicating that VirtualDataDocumentLookup instruction was used and runtime recursion checking is needed
     pub needs_runtime_recursion_check: bool,
 
+    /// Flag indicating whether this program contains any HostAwait instruction.
+    /// Clients can use this to decide whether suspendable execution mode is required.
+    #[serde(default)]
+    pub has_host_await: bool,
+
     /// Flag indicating that recompilation is needed due to partial deserialization failure
     /// This is set to true when the artifact section was successfully read but the extensible
     /// section failed to deserialize (e.g., due to version incompatibility)
@@ -85,7 +90,7 @@ pub struct Program {
 
 impl Program {
     /// Current serialization format version
-    pub const SERIALIZATION_VERSION: u32 = 5;
+    pub const SERIALIZATION_VERSION: u32 = 6;
     /// Magic bytes to identify Regorus program files
     pub const MAGIC: [u8; 4] = *b"REGO";
     /// Maximum instructions supported (matches u16 jump targets)
@@ -122,10 +127,13 @@ impl Program {
                 compiled_at: "unknown".to_string(),
                 source_info: "unknown".to_string(),
                 optimization_level: 0,
+                language: String::new(),
+                annotations: alloc::collections::BTreeMap::new(),
             },
             rule_tree: Value::new_object(),
             resolved_builtins: Vec::new(),
             needs_runtime_recursion_check: false,
+            has_host_await: false,
             needs_recompilation: false,
             rego_v0: false, // Default to Rego v1
         }
@@ -330,10 +338,21 @@ impl Program {
 
     /// Add instruction with optional span
     pub fn add_instruction(&mut self, instruction: Instruction, span: Option<SpanInfo>) {
+        if matches!(instruction, Instruction::HostAwait { .. }) {
+            self.has_host_await = true;
+        }
         self.instructions.push(instruction);
         self.instruction_spans.push(span);
     }
 
+    /// Recompute whether HostAwait is present by scanning instructions.
+    pub fn recompute_host_await_presence(&mut self) {
+        self.has_host_await = self
+            .instructions
+            .iter()
+            .any(|instruction| matches!(instruction, Instruction::HostAwait { .. }));
+    }
+
     /// Add literal value and return its index
     pub fn add_literal(&mut self, value: Value) -> usize {
         for (i, existing) in self.literals.iter().enumerate() {
@@ -408,6 +427,11 @@ impl Program {
     pub const fn is_fully_functional(&self) -> bool {
         !self.needs_recompilation
     }
+
+    /// Check whether HostAwait instructions are present in this program.
+    pub const fn has_host_await(&self) -> bool {
+        self.has_host_await
+    }
 }
 
 impl Default for Program {

src/rvm/program/listing.rs

@@ -133,6 +133,19 @@ pub fn generate_assembly_listing(program: &Program, config: &AssemblyListingConf
                 program.metadata.optimization_level
             ),
         );
+        if !program.metadata.language.is_empty() {
+            push_line(
+                &mut output,
+                format_args!(";   language: {}", program.metadata.language),
+            );
+        }
+        if !program.metadata.annotations.is_empty() {
+            push_line(&mut output, format_args!(";   annotations:"));
+            for (key, value) in &program.metadata.annotations {
+                let json = serde_json::to_string(value).unwrap_or_else(|_| "<invalid>".to_string());
+                push_line(&mut output, format_args!(";     {}: {}", key, json));
+            }
+        }
     }
 
     push_line(&mut output, format_args!(";"));
@@ -294,6 +307,14 @@ fn format_instruction_readable(
             let base = format!("{}LoadInput    r{} ← input", indent, dest);
             align_comment(&base, "Load global input document", config.comment_column)
         }
+        Instruction::LoadContext { dest } => {
+            let base = format!("{}LoadContext  r{} ← context", indent, dest);
+            align_comment(&base, "Load evaluation context", config.comment_column)
+        }
+        Instruction::LoadMetadata { dest } => {
+            let base = format!("{}LoadMetadata r{} ← metadata", indent, dest);
+            align_comment(&base, "Load program metadata", config.comment_column)
+        }
         Instruction::Move { dest, src } => {
             let base = format!("{}Move         r{} ← r{}", indent, dest, src);
             let comment = format!("Copy value from r{} to r{}", src, dest);
@@ -552,6 +573,11 @@ fn format_instruction_readable(
             let comment = format!("Append r{} to array r{}", value, arr);
             align_comment(&base, &comment, config.comment_column)
         }
+        Instruction::ArrayPushDefined { arr, value } => {
+            let base = format!("{}ArrayPushDef r{}.push(r{})", indent, arr, value);
+            let comment = format!("Append r{} to array r{} (skip if undefined)", value, arr);
+            align_comment(&base, &comment, config.comment_column)
+        }
         Instruction::ArrayCreate { params_index } => instruction_data
             .get_array_create_params(params_index)
             .map_or_else(
@@ -645,6 +671,25 @@ fn format_instruction_readable(
             };
             align_comment(&keyword, &comment, config.comment_column)
         }
+        Instruction::ReturnUndefinedIfNotTrue { condition } => {
+            let base = format!(
+                "{}ReturnUndefinedIfNotTrue if r{} != true return undefined",
+                indent, condition
+            );
+            let comment = format!(
+                "Return undefined unless r{} is exactly boolean true",
+                condition
+            );
+            align_comment(&base, &comment, config.comment_column)
+        }
+        Instruction::CoalesceUndefinedToNull { register } => {
+            let base = format!(
+                "{}CoalesceUndefinedToNull r{} = null if undefined",
+                indent, register
+            );
+            let comment = format!("Azure Policy: absent field → null (r{})", register);
+            align_comment(&base, &comment, config.comment_column)
+        }
         Instruction::LoopStart { params_index } => {
             instruction_data.get_loop_params(params_index).map_or_else(
                 || {
@@ -946,6 +991,8 @@ const fn get_instruction_name(instruction: &Instruction) -> &'static str {
         Instruction::LoadBool { .. } => "LOAD_BOOL",
         Instruction::LoadData { .. } => "LOAD_DATA",
         Instruction::LoadInput { .. } => "LOAD_INPUT",
+        Instruction::LoadContext { .. } => "LOAD_CONTEXT",
+        Instruction::LoadMetadata { .. } => "LOAD_METADATA",
         Instruction::Move { .. } => "MOVE",
         Instruction::Add { .. } => "ADD",
         Instruction::Sub { .. } => "SUB",
@@ -971,6 +1018,7 @@ const fn get_instruction_name(instruction: &Instruction) -> &'static str {
         Instruction::IndexLiteral { .. } => "INDEX_LIT",
         Instruction::ArrayNew { .. } => "ARRAY_NEW",
         Instruction::ArrayPush { .. } => "ARRAY_PUSH",
+        Instruction::ArrayPushDefined { .. } => "ARRAY_PUSH_DEF",
         Instruction::ArrayCreate { .. } => "ARRAY_CREATE",
         Instruction::SetNew { .. } => "SET_NEW",
         Instruction::SetAdd { .. } => "SET_ADD",
@@ -983,6 +1031,8 @@ const fn get_instruction_name(instruction: &Instruction) -> &'static str {
             crate::rvm::instructions::GuardMode::Condition => "ASSERT",
             crate::rvm::instructions::GuardMode::NotUndefined => "ASSERT_NOT_UNDEF",
         },
+        Instruction::ReturnUndefinedIfNotTrue { .. } => "RET_UNDEF_IF_NOT_TRUE",
+        Instruction::CoalesceUndefinedToNull { .. } => "COALESCE_NULL",
         Instruction::LoopStart { .. } => "LOOP_START",
         Instruction::LoopNext { .. } => "LOOP_NEXT",
         Instruction::CallRule { .. } => "CALL_RULE",
@@ -1011,6 +1061,12 @@ fn format_operation_compact(
         Instruction::LoadInput { dest } => {
             format!("{}r{} ← input", indent, dest)
         }
+        Instruction::LoadContext { dest } => {
+            format!("{}r{} ← context", indent, dest)
+        }
+        Instruction::LoadMetadata { dest } => {
+            format!("{}r{} ← metadata", indent, dest)
+        }
         Instruction::LoadData { dest } => {
             format!("{}r{} ← data", indent, dest)
         }

src/rvm/program/serialization/binary.rs

@@ -151,13 +151,13 @@ impl Program {
         }
 
         match version {
-            1..=3 => {
+            1..=5 => {
                 let mut program = Program::new();
                 program.needs_recompilation = true;
                 program.rego_v0 = Self::legacy_rego_v0(data, version).unwrap_or(false);
                 Ok(DeserializationResult::Partial(program))
             }
-            4 | 5 => {
+            6 => {
                 if data.len() < 29 {
                     return Err("Data too short for header".to_string());
                 }
@@ -281,15 +281,15 @@ impl Program {
         let version = Self::read_u32(data, 4).ok();
 
         match version {
-            Some(1..=5) => Ok(true),
+            Some(1..=6) => Ok(true),
             _ => Ok(false),
         }
     }
 
     fn legacy_rego_v0(data: &[u8], version: u32) -> Option<bool> {
         match version {
             1 => data.get(16).map(|value| *value != 0),
-            2 | 3 => data.get(24).map(|value| *value != 0),
+            2..=5 => data.get(24).map(|value| *value != 0),
             _ => None,
         }
     }

src/rvm/program/serialization/json.rs

@@ -24,6 +24,7 @@ impl Program {
                 "optimization_level": self.metadata.optimization_level,
                 "rego_v0": self.rego_v0,
                 "needs_runtime_recursion_check": self.needs_runtime_recursion_check,
+                "has_host_await": self.has_host_await,
                 "needs_recompilation": self.needs_recompilation
             },
             "program_structure": {
@@ -84,6 +85,27 @@ impl Program {
             .and_then(|v| v.as_u64())
             .and_then(|v| u8::try_from(v).ok())
             .unwrap_or(0);
+        let language = metadata
+            .get("language")
+            .and_then(|v| v.as_str())
+            .unwrap_or("")
+            .to_string();
+        let annotations: alloc::collections::BTreeMap<String, Value> = metadata
+            .get("annotations")
+            .and_then(|v| match *v {
+                // Deserialize JSON annotations into Value map
+                serde_json::Value::Object(ref map) => {
+                    let mut result = alloc::collections::BTreeMap::new();
+                    for (k, json_val) in map {
+                        if let Ok(val) = serde_json::from_value::<Value>(json_val.clone()) {
+                            result.insert(k.clone(), val);
+                        }
+                    }
+                    Some(result)
+                }
+                _ => None,
+            })
+            .unwrap_or_default();
         let rego_v0 = metadata
             .get("rego_v0")
             .and_then(|v| v.as_bool())
@@ -92,6 +114,10 @@ impl Program {
             .get("needs_runtime_recursion_check")
             .and_then(|v| v.as_bool())
             .unwrap_or(false);
+        let has_host_await = metadata
+            .get("has_host_await")
+            .and_then(|v| v.as_bool())
+            .unwrap_or(false);
         let needs_recompilation = metadata
             .get("needs_recompilation")
             .and_then(|v| v.as_bool())
@@ -183,14 +209,19 @@ impl Program {
                 compiled_at,
                 source_info,
                 optimization_level,
+                language,
+                annotations,
             },
             rule_tree,
             resolved_builtins: Vec::new(),
             needs_runtime_recursion_check,
+            has_host_await,
             needs_recompilation,
             rego_v0,
         };
 
+        program.recompute_host_await_presence();
+
         if !program.builtin_info_table.is_empty() {
             let _ = program.initialize_resolved_builtins();
         }