feat(value): Frozen object storage (boxed-slice) for compact parsed objects

src/builtins/azure_policy/template_functions_collection.rs

@@ -10,7 +10,6 @@ use crate::builtins;
 use crate::lexer::Span;
 use crate::value::Object;
 use crate::value::Value;
-use crate::Rc;
 
 use alloc::vec::Vec;
 use anyhow::Result;
@@ -79,7 +78,9 @@ fn fn_intersection(
                 };
                 result.retain(|k, v| other.get(k).is_some_and(|ov| *ov == *v));
             }
-            Ok(Value::Object(Rc::new(result)))
+            let value = result.into_value();
+            crate::utils::limits::enforce_memory_limit().map_err(anyhow::Error::new)?;
+            Ok(value)
         }
         _ => Ok(Value::Undefined),
     }
@@ -123,14 +124,19 @@ fn fn_union(_span: &Span, _params: &[Ref<Expr>], args: &[Value], _strict: bool)
                     #[allow(clippy::needless_borrowed_reference)]
                     let merged = match (result.get(k), v) {
                         (Some(&Value::Object(ref prev)), &Value::Object(ref next)) => {
-                            merge_objects(prev, next)
+                            merge_objects(prev, next)?
                         }
                         _ => v.clone(),
                     };
                     result.insert(k.clone(), merged);
+                    // Throttled check bounds peak allocation across nested merges.
+                    crate::utils::limits::check_memory_limit_if_needed()
+                        .map_err(anyhow::Error::new)?;
                 }
             }
-            Ok(Value::Object(Rc::new(result)))
+            let value = result.into_value();
+            crate::utils::limits::enforce_memory_limit().map_err(anyhow::Error::new)?;
+            Ok(value)
         }
         _ => Ok(Value::Undefined),
     }
@@ -274,27 +280,35 @@ fn fn_create_object(
         #[allow(clippy::pattern_type_mismatch)]
         if let [key, value] = pair {
             map.insert(key.clone(), value.clone());
+            // Throttled check bounds peak allocation while building the object.
+            crate::utils::limits::check_memory_limit_if_needed().map_err(anyhow::Error::new)?;
         }
     }
 
-    Ok(Value::Object(Rc::new(map)))
+    let value = map.into_value();
+    crate::utils::limits::enforce_memory_limit().map_err(anyhow::Error::new)?;
+    Ok(value)
 }
 
 // ── Helpers ───────────────────────────────────────────────────────────
 
 /// Recursively merge two objects.  Nested objects are merged; everything
 /// else (including arrays) uses the value from `incoming`.
-fn merge_objects(base: &Object, overlay: &Object) -> Value {
+fn merge_objects(base: &Object, overlay: &Object) -> Result<Value> {
     let mut result = base.clone();
     for (k, v) in overlay.iter() {
         #[allow(clippy::needless_borrowed_reference)]
         let merged = match (result.get(k), v) {
-            (Some(&Value::Object(ref prev)), &Value::Object(ref next)) => merge_objects(prev, next),
+            (Some(&Value::Object(ref prev)), &Value::Object(ref next)) => {
+                merge_objects(prev, next)?
+            }
             _ => v.clone(),
         };
         result.insert(k.clone(), merged);
+        // Throttled check bounds peak allocation across recursive merges.
+        crate::utils::limits::check_memory_limit_if_needed().map_err(anyhow::Error::new)?;
     }
-    Value::Object(Rc::new(result))
+    Ok(result.into_value())
 }
 
 fn extract_usize(v: &Value) -> Option<usize> {

src/builtins/azure_policy/template_functions_misc.rs

@@ -10,7 +10,6 @@ use crate::builtins;
 use crate::lexer::Span;
 use crate::value::Object;
 use crate::value::Value;
-use crate::Rc;
 
 use alloc::string::{String, ToString as _};
 use alloc::vec::Vec;
@@ -88,8 +87,11 @@ fn fn_items(_span: &Span, _params: &[Ref<Expr>], args: &[Value], _strict: bool)
         let mut entry = Object::new();
         entry.insert(Value::from("key"), k.clone());
         entry.insert(Value::from("value"), v.clone());
-        result.push(Value::Object(Rc::new(entry)));
+        result.push(entry.into_value());
+        // Throttled check bounds peak allocation while building large results.
+        crate::utils::limits::check_memory_limit_if_needed().map_err(anyhow::Error::new)?;
     }
+    crate::utils::limits::enforce_memory_limit().map_err(anyhow::Error::new)?;
     Ok(Value::from_array(result))
 }
 

src/builtins/objects.rs

@@ -399,7 +399,8 @@ fn object_union(span: &Span, params: &[Ref<Expr>], args: &[Value], _strict: bool
     let _ = ensure_object(name, &params[0], args[0].clone())?;
     let _ = ensure_object(name, &params[1], args[1].clone())?;
 
-    union(&args[0], &args[1])
+    let result = union(&args[0], &args[1])?;
+    Ok(result)
 }
 
 fn object_union_n(

src/engine.rs

@@ -398,6 +398,13 @@ impl Engine {
     /// # }
     /// ```
     pub fn set_input(&mut self, input: Value) {
+        // Store the document as-is -- this is a pure setter. Inputs produced by
+        // `set_input_json`/`from_json_str` already carry compact frozen storage; a
+        // programmatically-built `Value` is kept in whatever representation the caller
+        // supplied. Not freezing here avoids a full recursive traversal and the
+        // unnecessary materialization (BTree -> boxed slice) of a value we were
+        // just handed, and keeps the API free of any infallible allocation under
+        // `allocator-memory-limits`.
         self.interpreter.set_input(input);
     }
 
@@ -484,13 +491,15 @@ impl Engine {
             // allocating, so validate then deep-merge in place (zero-copy fast path).
             self.interpreter.get_init_data().check_mergeable(&data)?;
             self.prepared = false;
-            self.interpreter.get_init_data_mut().deep_merge(data)
+            self.interpreter.get_init_data_mut().deep_merge(data)?;
+            Ok(())
         }
         #[cfg(feature = "allocator-memory-limits")]
         {
             // A limit failure can strike mid-merge and can't be predicted, so merge into a
-            // candidate and commit only on success. `Value` is copy-on-write, so only touched
-            // subtrees are cloned.
+            // candidate and commit only on success; `deep_merge` limit-checks each insertion,
+            // so the committed document is already bounded. `Value` is copy-on-write, so only
+            // touched subtrees are cloned.
             let mut candidate = self.interpreter.get_init_data().clone();
             candidate.deep_merge(data)?;
             *self.interpreter.get_init_data_mut() = candidate;

src/rvm/program/serialization/value.rs

@@ -265,8 +265,15 @@ impl<'de> Visitor<'de> for BinaryValueVisitor {
                 let mut map = Object::new();
                 for (key, value) in entries {
                     map.insert(key.into_value(), value.into_value());
+                    crate::utils::limits::check_memory_limit_if_needed()
+                        .map_err(|err| de::Error::custom(format!("{err}")))?;
                 }
-                Ok(BinaryValue(Value::Object(crate::Rc::new(map))))
+                let object = map.into_value();
+                // Mirror the JSON object path: re-check after `into_value` compacts
+                // storage so binary program loading honors allocator memory limits.
+                crate::utils::limits::check_memory_limit_if_needed()
+                    .map_err(|err| de::Error::custom(format!("{err}")))?;
+                Ok(BinaryValue(object))
             }
             (BinaryVariant::Undefined, variant) => {
                 variant.unit_variant()?;

src/rvm/vm/comprehension.rs

@@ -4,7 +4,6 @@
 use crate::rvm::instructions::{ComprehensionBeginParams, ComprehensionMode};
 use crate::value::Object;
 use crate::value::Value;
-use crate::Rc;
 use alloc::format;
 use alloc::sync::Arc;
 use alloc::vec::Vec;
@@ -34,7 +33,7 @@ impl RegoVM {
         let initial_result = match params.mode {
             ComprehensionMode::Set => Value::new_set(),
             ComprehensionMode::Array => Value::new_array(),
-            ComprehensionMode::Object => Value::Object(Rc::new(Object::new())),
+            ComprehensionMode::Object => Object::new().into_value(),
         };
         self.set_register(params.result_reg, initial_result.clone())?;
 
@@ -118,7 +117,7 @@ impl RegoVM {
         let initial_result = match params.mode {
             ComprehensionMode::Set => Value::new_set(),
             ComprehensionMode::Array => Value::new_array(),
-            ComprehensionMode::Object => Value::Object(Rc::new(Object::new())),
+            ComprehensionMode::Object => Object::new().into_value(),
         };
         self.set_register(params.result_reg, initial_result.clone())?;
 
@@ -519,15 +518,15 @@ impl RegoVM {
         // `ComprehensionEnd` is reached from a loaded program; an empty stack
         // here means malformed user-supplied bytecode, which must still surface
         // as a typed error rather than a panic — including in debug builds.
-        self.comprehension_stack.pop().map_or_else(
-            || {
-                Err(VmError::InvalidIteration {
-                    value: Value::String(Arc::from("No active comprehension context")),
-                    pc: self.pc,
-                })
-            },
-            |_context| Ok(()),
-        )
+        // The accumulated result already resides in its result register and needs
+        // no post-processing, so popping the context to unwind the stack suffices.
+        self.comprehension_stack
+            .pop()
+            .ok_or_else(|| VmError::InvalidIteration {
+                value: Value::String(Arc::from("No active comprehension context")),
+                pc: self.pc,
+            })?;
+        Ok(())
     }
 
     fn execute_comprehension_end_suspendable(&mut self) -> Result<()> {
@@ -558,6 +557,8 @@ impl RegoVM {
                     return_pc: _,
                     context,
                 } => {
+                    // The accumulated result already resides in `result_reg`; no
+                    // post-processing is required before resuming.
                     let raw_target = context.resume_pc;
                     let resume_pc = if raw_target <= self.pc {
                         self.pc.saturating_add(1)

src/value/interning.rs

@@ -0,0 +1,101 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+//! Thread-local interning of object **keys** during JSON deserialization.
+//!
+//! [`Value::from_json_str`](crate::Value::from_json_str) installs a thread-local
+//! [`InternTable`] for the duration of the parse (see [`InternGuard`]). Object
+//! keys parsed while a table is installed are deduplicated: a homogeneous array
+//! of `N` objects sharing `K` keys allocates `K` key strings instead of
+//! `N * K`, and every later occurrence becomes a reference-count clone. Only
+//! keys are interned -- string *values* are left untouched, since their
+//! cardinality is unbounded and interning them can cost more than it saves.
+//!
+//! The table lives only for the parse and is torn down when the guard drops, so
+//! there is no unbounded, thread-lifetime cache. This module is std-only: it
+//! relies on `thread_local!`.
+
+use core::cell::RefCell;
+use std::collections::HashSet;
+
+/// A content-addressed interning table scoped to a single deserialization call.
+///
+/// Stores each distinct key's own `Rc<str>` (the allocation serde already made),
+/// so interning a repeated key returns a reference-count clone of that existing
+/// allocation and never copies the key bytes a second time.
+struct InternTable {
+    set: HashSet<crate::Rc<str>>,
+}
+
+impl InternTable {
+    fn new() -> Self {
+        Self {
+            set: HashSet::new(),
+        }
+    }
+
+    /// Return an interned handle equal to `rc`. The first time a given key is
+    /// seen its own allocation is retained and returned; later occurrences reuse
+    /// that allocation and the caller's `rc` is dropped.
+    fn intern(&mut self, rc: crate::Rc<str>) -> crate::Rc<str> {
+        if let Some(existing) = self.set.get(rc.as_ref()) {
+            return crate::Rc::clone(existing);
+        }
+        self.set.insert(crate::Rc::clone(&rc));
+        rc
+    }
+}
+
+std::thread_local! {
+    static TABLE: RefCell<Option<InternTable>> = const { RefCell::new(None) };
+}
+
+/// RAII guard that installs a thread-local [`InternTable`] for the duration of a
+/// deserialization call. Re-entrant: if a table is already installed (e.g. a
+/// nested parse), this guard reuses it and does not tear it down on drop.
+pub struct InternGuard {
+    installed_by_us: bool,
+}
+
+impl InternGuard {
+    /// Install a fresh intern table for the current thread, unless one is
+    /// already installed (in which case the existing table is reused).
+    pub fn install() -> Self {
+        let installed_by_us = TABLE.with(|t| {
+            let mut slot = t.borrow_mut();
+            if slot.is_none() {
+                *slot = Some(InternTable::new());
+                true
+            } else {
+                false
+            }
+        });
+        Self { installed_by_us }
+    }
+}
+
+impl Drop for InternGuard {
+    fn drop(&mut self) {
+        if self.installed_by_us {
+            TABLE.with(|t| {
+                *t.borrow_mut() = None;
+            });
+        }
+    }
+}
+
+/// Intern `rc` using the current thread's active [`InternTable`], if one is
+/// installed. Outside a parse (no guard active) the original `rc` is returned
+/// unchanged, so callers keep their allocation with no interning overhead.
+pub fn intern_key(rc: crate::Rc<str>) -> crate::Rc<str> {
+    TABLE.with(|t| match t.borrow_mut().as_mut() {
+        Some(table) => table.intern(rc),
+        None => rc,
+    })
+}
+
+/// Test-only probe: whether an intern table is currently installed on this thread.
+#[cfg(test)]
+pub fn is_installed() -> bool {
+    TABLE.with(|t| t.borrow().is_some())
+}

src/value/mod.rs

@@ -15,6 +15,9 @@ mod array;
 mod object;
 mod set;
 
+#[cfg(feature = "std")]
+mod interning;
+
 #[cfg(test)]
 mod tests;
 
@@ -237,21 +240,42 @@ impl<'de> Visitor<'de> for ValueVisitor {
                 }
             }
             let mut map = BTreeMap::new();
-            map.insert(key, value);
+            map.insert(intern_object_key(key), value);
             // Enforce allocator limit while expanding a deserialized object.
             enforce_limit_for::<V::Error>()?;
             while let Some((key, value)) = visitor.next_entry()? {
-                map.insert(key, value);
+                map.insert(intern_object_key(key), value);
                 // Enforce allocator limit while expanding a deserialized object.
                 enforce_limit_for::<V::Error>()?;
             }
-            Ok(Value::from(map))
+            let value = Value::from(map);
+            // The freeze into compact object storage allocates after the final map insert.
+            // Re-check fallible deserialization paths so allocator-limit builds observe it.
+            enforce_limit_for::<V::Error>()?;
+            Ok(value)
         } else {
             Ok(Value::new_object())
         }
     }
 }
 
+/// Key-only interning hook for the JSON deserialization hot path (see
+/// [`crate::value::interning`]). No-op when no interning session is installed
+/// or when built without the `std` feature.
+#[cfg(feature = "std")]
+fn intern_object_key(key: Value) -> Value {
+    match key {
+        Value::String(rc) => Value::String(interning::intern_key(rc)),
+        other => other,
+    }
+}
+
+/// No-op key hook for builds without the `std` feature (no interning table).
+#[cfg(not(feature = "std"))]
+const fn intern_object_key(key: Value) -> Value {
+    key
+}
+
 #[doc(hidden)]
 impl<'de> Deserialize<'de> for Value {
     fn deserialize<D>(deserializer: D) -> Result<Value, D::Error>
@@ -358,6 +382,10 @@ impl Value {
     /// # }
     /// ```
     pub fn from_json_str(json: &str) -> Result<Value> {
+        // Intern object keys for the duration of the parse: a homogeneous array
+        // of N objects sharing K keys allocates K key strings instead of N * K.
+        #[cfg(feature = "std")]
+        let _guard = interning::InternGuard::install();
         match serde_json::from_str::<Value>(json) {
             Ok(value) => Ok(value),
             Err(err) => {
@@ -808,7 +836,7 @@ impl From<BTreeMap<Value, Value>> for Value {
     /// # Ok(())
     /// # }
     fn from(s: BTreeMap<Value, Value>) -> Self {
-        Value::Object(Rc::new(Object::from(s)))
+        Object::from(s).into_value()
     }
 }