diff --git a/CHANGELOG.md b/CHANGELOG.md
index fbb1d49..5c5c7ed 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,6 +6,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- `Engine::add_data` now deep-merges nested data documents instead of only merging top-level keys. Adding `{ "a": { "x": 1 } }` followed by `{ "a": { "y": 2 } }` now yields `{ "a": { "x": 1, "y": 2 } }` (matching OPA's data-document merge). Nested sets under a shared key are unioned. Only genuine leaf conflicts (the same path holding two different values) are reported as errors.
+- A zero-arg function producing two different complete values (e.g. `f() := { "a": 1 }` and `f() := { "b": 2 }`) is now reported as a conflict, matching OPA's complete-rule semantics, instead of silently combining the outputs.
+
+### Security
+
+- `Engine::add_data` now rejects data nested beyond 128 levels instead of risking a stack overflow on adversarially deep input.
+
 ## [0.10.1](https://github.com/microsoft/regorus/compare/regorus-v0.10.0...regorus-v0.10.1) - 2026-05-22
 
 ### Fixed
diff --git a/src/engine.rs b/src/engine.rs
index bb0a70b..b291ef8 100644
--- a/src/engine.rs
+++ b/src/engine.rs
@@ -434,7 +434,13 @@ impl Engine {
 
     /// Add data document.
     ///
-    /// The specified data document is merged into existing data document.
+    /// The specified data document is deep-merged into the existing data document. Nested
+    /// objects are merged recursively (matching OPA's data-document merge), so adding
+    /// `{ "a": { "x": 1 } }` and then `{ "a": { "y": 2 } }` yields `{ "a": { "x": 1, "y": 2 } }`.
+    /// A conflict — the same path holding two different values — is an error.
+    ///
+    /// The merge is atomic: if any conflict is detected (including one deep in a nested
+    /// document), the call fails and the existing data document is left unchanged.
     ///
     /// ```
     /// # use regorus::*;
@@ -453,9 +459,13 @@ impl Engine {
     /// // Merge { "z" : 3 }. Conflict error.
     /// assert!(engine.add_data(Value::from_json_str(r#"{ "z" : 3 }"#)?).is_err());
     ///
+    /// // Nested objects are deep-merged. Merge { "y" : { "a" : 10 } } then { "y" : { "b" : 20 } }.
+    /// assert!(engine.add_data(Value::from_json_str(r#"{ "y" : { "a" : 10 } }"#)?).is_ok());
+    /// assert!(engine.add_data(Value::from_json_str(r#"{ "y" : { "b" : 20 } }"#)?).is_ok());
+    ///
     /// assert_eq!(
     ///   engine.eval_query("data".to_string(), false)?.result[0].expressions[0].value,
-    ///   Value::from_json_str(r#"{ "x": 1, "y": {}, "z": 2}"#)?
+    ///   Value::from_json_str(r#"{ "x": 1, "y": { "a": 10, "b": 20 }, "z": 2}"#)?
     /// );
     /// # Ok(())
     /// # }
@@ -464,8 +474,29 @@ impl Engine {
         if data.as_object().is_err() {
             bail!("data must be object");
         }
-        self.prepared = false;
-        self.interpreter.get_init_data_mut().merge(data)
+
+        // add_data is all-or-nothing; the atomic strategy differs by build because the failure
+        // modes do: a conflict (same path, differing values) is possible everywhere, an
+        // allocator-limit failure mid-merge only under `allocator-memory-limits`.
+        #[cfg(not(feature = "allocator-memory-limits"))]
+        {
+            // Conflict is the only failure mode; `check_mergeable` catches it up front without
+            // 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)
+        }
+        #[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.
+            let mut candidate = self.interpreter.get_init_data().clone();
+            candidate.deep_merge(data)?;
+            *self.interpreter.get_init_data_mut() = candidate;
+            self.prepared = false;
+            Ok(())
+        }
     }
 
     /// Get the data document.
diff --git a/src/interpreter.rs b/src/interpreter.rs
index 2028eb9..f6cff17 100644
--- a/src/interpreter.rs
+++ b/src/interpreter.rs
@@ -60,6 +60,17 @@ enum FunctionModifier {
     Value(Value),
 }
 
+/// How [`Interpreter::update_data`] merges a rule's value into the data document.
+#[derive(Debug, Clone, Copy)]
+enum RuleValueMerge {
+    /// Shallow-merge keeping disjoint keys, so rules sharing a path prefix scaffold into one
+    /// object (`a.foo` + `a.bar` → one `a`) instead of conflicting.
+    Combine,
+    /// Complete-rule semantics: existing value must be absent or exactly equal, else conflict.
+    /// Used for zero-arg function outputs (`f() := …`), which OPA treats like complete rules.
+    Strict,
+}
+
 type RuleValues = BTreeMap<Vec<Value>, (Value, Ref<Expr>)>;
 
 #[derive(Debug)]
@@ -3408,6 +3419,23 @@ impl Interpreter {
         }
     }
 
+    /// Materialize a complete-rule value: the existing value must be absent or *exactly equal*
+    /// to `new`, else it is a conflict.
+    ///
+    /// Unlike the shallow [`Self::merge_rule_value`], differing outputs conflict instead of
+    /// combining — `f() := {"a": 1}` and `f() := {"b": 2}` conflict — matching OPA's semantics
+    /// for zero-arg functions.
+    fn merge_rule_value_strict(span: &Span, value: &mut Value, new: Value) -> Result<()> {
+        if *value == Value::Undefined {
+            *value = new;
+            Ok(())
+        } else if *value == new {
+            Ok(())
+        } else {
+            Err(span.error("rules should not produce multiple outputs."))
+        }
+    }
+
     pub fn get_path_string(refr: &Expr, document: Option<&str>) -> Result<String> {
         let mut comps = vec![];
         let mut expr_opt = Some(refr);
@@ -3663,6 +3691,7 @@ impl Interpreter {
         _refr: &Expr,
         path: &[&str],
         value: Value,
+        merge: RuleValueMerge,
     ) -> Result<()> {
         if value == Value::Undefined {
             return Ok(());
@@ -3670,7 +3699,10 @@ impl Interpreter {
         // Ensure that path is created.
         let vref = Self::make_or_get_value_mut(&mut self.data, path)?;
         if Self::get_value_chained(self.init_data.clone(), path) == Value::Undefined {
-            Self::merge_rule_value(span, vref, value)
+            match merge {
+                RuleValueMerge::Strict => Self::merge_rule_value_strict(span, vref, value),
+                RuleValueMerge::Combine => Self::merge_rule_value(span, vref, value),
+            }
         } else {
             // Retain specified value.
             Ok(())
@@ -3778,7 +3810,13 @@ impl Interpreter {
                         // `a` is created as an empty object.
                         if let Some((_, prefix)) = path.split_last() {
                             if !prefix.is_empty() {
-                                self.update_data(span, refr, prefix, Value::new_object())?;
+                                self.update_data(
+                                    span,
+                                    refr,
+                                    prefix,
+                                    Value::new_object(),
+                                    RuleValueMerge::Combine,
+                                )?;
                             }
                         }
 
@@ -3790,7 +3828,13 @@ impl Interpreter {
                             };
 
                             let value = self.eval_rule_bodies(ctx, span, rule_body)?;
-                            self.update_data(refr.span(), refr, &path[..], value)?;
+                            self.update_data(
+                                refr.span(),
+                                refr,
+                                &path[..],
+                                value,
+                                RuleValueMerge::Strict,
+                            )?;
                         }
                     }
                 }
@@ -4037,6 +4081,7 @@ impl Interpreter {
                         rule_refr,
                         &prefix_path,
                         Value::new_object(),
+                        RuleValueMerge::Combine,
                     )?;
                 }
             }
diff --git a/src/tests/interpreter/mod.rs b/src/tests/interpreter/mod.rs
index 9b88180..981fa10 100644
--- a/src/tests/interpreter/mod.rs
+++ b/src/tests/interpreter/mod.rs
@@ -24,6 +24,7 @@ use anyhow::{bail, Result};
 use core::num::NonZeroU32;
 use core::time::Duration;
 use serde::{Deserialize, Serialize};
+use std::collections::{BTreeMap, BTreeSet};
 use test_generator::test_resources;
 use timer_test_support::{
     apply_engine_timer, configure_time_source, reset_time_source, GlobalTimerGuard,
@@ -818,3 +819,446 @@ fn test_get_data() -> Result<()> {
 
     Ok(())
 }
+
+#[test]
+fn test_add_data_deep_merge() -> Result<()> {
+    let mut engine = Engine::new();
+
+    // Nested objects under a shared top-level key are deep-merged, not replaced.
+    engine.add_data(Value::from_json_str(r#"{ "a" : { "x" : 1 } }"#)?)?;
+    engine.add_data(Value::from_json_str(r#"{ "a" : { "y" : 2 } }"#)?)?;
+
+    assert_eq!(
+        engine.get_data(),
+        Value::from_json_str(r#"{ "a" : { "x" : 1, "y" : 2 } }"#)?
+    );
+
+    Ok(())
+}
+
+#[test]
+fn test_add_data_deep_merge_multi_level() -> Result<()> {
+    let mut engine = Engine::new();
+
+    // Merging recurses through multiple levels of nesting.
+    engine.add_data(Value::from_json_str(
+        r#"{ "a" : { "b" : { "x" : 1 } }, "top" : 0 }"#,
+    )?)?;
+    engine.add_data(Value::from_json_str(
+        r#"{ "a" : { "b" : { "y" : 2 }, "c" : 3 } }"#,
+    )?)?;
+
+    assert_eq!(
+        engine.get_data(),
+        Value::from_json_str(r#"{ "a" : { "b" : { "x" : 1, "y" : 2 }, "c" : 3 }, "top" : 0 }"#)?
+    );
+
+    Ok(())
+}
+
+#[test]
+fn test_add_data_leaf_conflict_errors() -> Result<()> {
+    let mut engine = Engine::new();
+
+    // A genuine leaf conflict (same nested path, different value) is an error.
+    engine.add_data(Value::from_json_str(r#"{ "a" : { "x" : 1 } }"#)?)?;
+    assert!(engine
+        .add_data(Value::from_json_str(r#"{ "a" : { "x" : 2 } }"#)?)
+        .is_err());
+
+    Ok(())
+}
+
+#[test]
+fn test_add_data_object_vs_scalar_conflict_errors() -> Result<()> {
+    let mut engine = Engine::new();
+
+    // An object cannot be merged with a scalar at the same path.
+    engine.add_data(Value::from_json_str(r#"{ "a" : { "x" : 1 } }"#)?)?;
+    assert!(engine
+        .add_data(Value::from_json_str(r#"{ "a" : 5 }"#)?)
+        .is_err());
+
+    Ok(())
+}
+
+#[test]
+fn test_add_data_equal_leaf_is_noop() -> Result<()> {
+    let mut engine = Engine::new();
+
+    // Re-adding identical data (including equal nested leaves) is tolerated as a no-op.
+    engine.add_data(Value::from_json_str(r#"{ "a" : { "x" : 1 } }"#)?)?;
+    engine.add_data(Value::from_json_str(r#"{ "a" : { "x" : 1 }, "b" : 2 }"#)?)?;
+
+    assert_eq!(
+        engine.get_data(),
+        Value::from_json_str(r#"{ "a" : { "x" : 1 }, "b" : 2 }"#)?
+    );
+
+    Ok(())
+}
+
+#[test]
+fn test_add_data_set_union() -> Result<()> {
+    let mut engine = Engine::new();
+
+    // Sets under a shared key are unioned rather than conflicting (consistent with the
+    // rule-evaluation merge, where partial set rules accumulate elements). JSON cannot express
+    // sets, so the data documents are built via the `Value` API.
+    engine.add_data(Value::from(BTreeMap::from([(
+        Value::from("s"),
+        Value::from(BTreeSet::from([Value::from(1_u64), Value::from(2_u64)])),
+    )])))?;
+    engine.add_data(Value::from(BTreeMap::from([(
+        Value::from("s"),
+        Value::from(BTreeSet::from([Value::from(2_u64), Value::from(3_u64)])),
+    )])))?;
+
+    let expected = Value::from(BTreeMap::from([(
+        Value::from("s"),
+        Value::from(BTreeSet::from([
+            Value::from(1_u64),
+            Value::from(2_u64),
+            Value::from(3_u64),
+        ])),
+    )]));
+    assert_eq!(engine.get_data(), expected);
+
+    Ok(())
+}
+
+#[test]
+fn test_add_data_nested_set_union() -> Result<()> {
+    let mut engine = Engine::new();
+
+    // A set nested under an object key exercises the recursive merge: the outer objects are
+    // deep-merged and the inner sets are then unioned.
+    engine.add_data(Value::from(BTreeMap::from([(
+        Value::from("a"),
+        Value::from(BTreeMap::from([(
+            Value::from("s"),
+            Value::from(BTreeSet::from([Value::from(1_u64)])),
+        )])),
+    )])))?;
+    engine.add_data(Value::from(BTreeMap::from([(
+        Value::from("a"),
+        Value::from(BTreeMap::from([(
+            Value::from("s"),
+            Value::from(BTreeSet::from([Value::from(2_u64)])),
+        )])),
+    )])))?;
+
+    let expected = Value::from(BTreeMap::from([(
+        Value::from("a"),
+        Value::from(BTreeMap::from([(
+            Value::from("s"),
+            Value::from(BTreeSet::from([Value::from(1_u64), Value::from(2_u64)])),
+        )])),
+    )]));
+    assert_eq!(engine.get_data(), expected);
+
+    Ok(())
+}
+
+#[test]
+fn test_add_data_equal_set_is_noop() -> Result<()> {
+    let mut engine = Engine::new();
+
+    // Re-adding an identical set is tolerated as a no-op (not a conflict).
+    engine.add_data(Value::from(BTreeMap::from([(
+        Value::from("s"),
+        Value::from(BTreeSet::from([Value::from(1_u64), Value::from(2_u64)])),
+    )])))?;
+    engine.add_data(Value::from(BTreeMap::from([(
+        Value::from("s"),
+        Value::from(BTreeSet::from([Value::from(1_u64), Value::from(2_u64)])),
+    )])))?;
+
+    let expected = Value::from(BTreeMap::from([(
+        Value::from("s"),
+        Value::from(BTreeSet::from([Value::from(1_u64), Value::from(2_u64)])),
+    )]));
+    assert_eq!(engine.get_data(), expected);
+
+    Ok(())
+}
+
+#[test]
+fn test_add_data_failed_merge_is_atomic() -> Result<()> {
+    let mut engine = Engine::new();
+
+    engine.add_data(Value::from_json_str(r#"{ "a" : { "z" : 1 } }"#)?)?;
+
+    // Mixes a new key `m` with a conflicting leaf `z` (1 vs 3). Because `m` sorts
+    // before `z`, a naive in-place merge would insert `m` and only then hit the `z`
+    // conflict. add_data must be all-or-nothing: the whole call fails AND leaves the
+    // existing data untouched — `m` must not leak in.
+    assert!(engine
+        .add_data(Value::from_json_str(r#"{ "a" : { "m" : 2, "z" : 3 } }"#)?)
+        .is_err());
+
+    assert_eq!(
+        engine.get_data(),
+        Value::from_json_str(r#"{ "a" : { "z" : 1 } }"#)?
+    );
+
+    Ok(())
+}
+
+#[test]
+fn test_add_data_failed_set_merge_is_atomic() -> Result<()> {
+    let mut engine = Engine::new();
+
+    // Existing data: a set `s` alongside a scalar `z` under `a`.
+    engine.add_data(Value::from(BTreeMap::from([(
+        Value::from("a"),
+        Value::from(BTreeMap::from([
+            (
+                Value::from("s"),
+                Value::from(BTreeSet::from([Value::from(1_u64), Value::from(2_u64)])),
+            ),
+            (Value::from("z"), Value::from(1_u64)),
+        ])),
+    )])))?;
+
+    // This add would union `s` with {3} but conflicts on `z` (1 vs 2). Since `s`
+    // sorts before `z`, a naive in-place merge would union the set *before* failing
+    // on `z`, leaking {3} into `s`. The atomic add must reject the whole call and
+    // leave `s` as {1, 2}.
+    assert!(engine
+        .add_data(Value::from(BTreeMap::from([(
+            Value::from("a"),
+            Value::from(BTreeMap::from([
+                (
+                    Value::from("s"),
+                    Value::from(BTreeSet::from([Value::from(3_u64)])),
+                ),
+                (Value::from("z"), Value::from(2_u64)),
+            ])),
+        )])))
+        .is_err());
+
+    // `s` must be unchanged ({1, 2}, not {1, 2, 3}) and `z` must still be 1.
+    let expected = Value::from(BTreeMap::from([(
+        Value::from("a"),
+        Value::from(BTreeMap::from([
+            (
+                Value::from("s"),
+                Value::from(BTreeSet::from([Value::from(1_u64), Value::from(2_u64)])),
+            ),
+            (Value::from("z"), Value::from(1_u64)),
+        ])),
+    )]));
+    assert_eq!(engine.get_data(), expected);
+
+    Ok(())
+}
+
+#[test]
+fn test_add_data_failed_array_merge_is_atomic() -> Result<()> {
+    let mut engine = Engine::new();
+
+    engine.add_data(Value::from_json_str(r#"{ "a" : { "arr" : [1, 2] } }"#)?)?;
+
+    // Arrays are atomic leaves (never element-merged), so a differing array at the
+    // same path is a conflict. The new key `aa` sorts before `arr`, so a naive
+    // in-place merge would insert `aa` and only then hit the `arr` conflict. add_data
+    // must reject the whole call and leave the data untouched — `aa` must not leak in.
+    assert!(engine
+        .add_data(Value::from_json_str(
+            r#"{ "a" : { "aa" : 5, "arr" : [3] } }"#
+        )?)
+        .is_err());
+
+    assert_eq!(
+        engine.get_data(),
+        Value::from_json_str(r#"{ "a" : { "arr" : [1, 2] } }"#)?
+    );
+
+    Ok(())
+}
+
+// The `Value::merge` used by `add_data` is shared with the rule-evaluation path
+// (`Interpreter::merge_rule_value`, reached via `with data.* as ...` and rule-value
+// materialization). The tests below pin down that making `merge` recursive changed only the
+// data-document semantics and left rule evaluation — in particular the `with data.* as ...`
+// modifier — behaving exactly as before (an override, never a deep merge).
+
+#[test]
+fn test_with_data_modifier_replaces_nested_object() -> Result<()> {
+    let mut engine = Engine::new();
+
+    // Base data provides a nested object with two keys.
+    engine.add_data(Value::from_json_str(
+        r#"{ "base" : { "foo" : { "a" : 1, "b" : 2 } } }"#,
+    )?)?;
+
+    engine.add_policy(
+        "policy.rego".to_string(),
+        r#"
+package test
+
+result := x if {
+    x := data.base.foo with data.base.foo as {"a": 99}
+}
+"#
+        .to_string(),
+    )?;
+
+    // `with data.base.foo as {"a": 99}` REPLACES the whole subtree for the duration of the
+    // rule; it must NOT deep-merge with the base `{ "a": 1, "b": 2 }`. So `b` is gone.
+    assert_eq!(
+        engine
+            .eval_query("data.test.result".to_string(), false)?
+            .result[0]
+            .expressions[0]
+            .value
+            .clone(),
+        Value::from_json_str(r#"{ "a" : 99 }"#)?
+    );
+
+    Ok(())
+}
+
+#[test]
+fn test_with_data_modifier_replaces_whole_subtree() -> Result<()> {
+    let mut engine = Engine::new();
+
+    engine.add_data(Value::from_json_str(
+        r#"{ "base" : { "foo" : 1, "bar" : 2 } }"#,
+    )?)?;
+
+    engine.add_policy(
+        "policy.rego".to_string(),
+        r#"
+package test
+
+result := x if {
+    x := data.base with data.base as {"only": 3}
+}
+"#
+        .to_string(),
+    )?;
+
+    // `with data.base as {...}` replaces the entire `data.base` object; the original
+    // `foo`/`bar` keys are not merged in.
+    assert_eq!(
+        engine
+            .eval_query("data.test.result".to_string(), false)?
+            .result[0]
+            .expressions[0]
+            .value
+            .clone(),
+        Value::from_json_str(r#"{ "only" : 3 }"#)?
+    );
+
+    Ok(())
+}
+
+#[test]
+fn test_with_data_modifier_nested_replace_preserves_siblings() -> Result<()> {
+    let mut engine = Engine::new();
+
+    // `data.base` has a nested `foo` object AND a sibling `bar`.
+    engine.add_data(Value::from_json_str(
+        r#"{ "base" : { "foo" : { "a" : 1, "b" : 2 }, "bar" : 7 } }"#,
+    )?)?;
+
+    engine.add_policy(
+        "policy.rego".to_string(),
+        r#"
+package test
+
+# `with` targets the nested `data.base.foo`, but the rule observes the PARENT `data.base`.
+result := x if {
+    x := data.base with data.base.foo as {"a": 99}
+}
+"#
+        .to_string(),
+    )?;
+
+    // The nested `foo` is deep-replaced (its `b` is gone — `with` never merges), while the
+    // sibling `bar` under the same parent is preserved.
+    assert_eq!(
+        engine
+            .eval_query("data.test.result".to_string(), false)?
+            .result[0]
+            .expressions[0]
+            .value
+            .clone(),
+        Value::from_json_str(r#"{ "foo" : { "a" : 99 }, "bar" : 7 }"#)?
+    );
+
+    Ok(())
+}
+
+#[test]
+fn test_rule_reads_deep_merged_base_data() -> Result<()> {
+    let mut engine = Engine::new();
+
+    // Two add_data calls deep-merge into a single nested object...
+    engine.add_data(Value::from_json_str(
+        r#"{ "base" : { "foo" : { "a" : 1 } } }"#,
+    )?)?;
+    engine.add_data(Value::from_json_str(
+        r#"{ "base" : { "foo" : { "b" : 2 } } }"#,
+    )?)?;
+
+    engine.add_policy(
+        "policy.rego".to_string(),
+        r#"
+package test
+
+a := data.base.foo.a
+b := data.base.foo.b
+"#
+        .to_string(),
+    )?;
+
+    // ...and both merged leaves are visible to rule evaluation.
+    assert_eq!(
+        engine.eval_query("data.test".to_string(), false)?.result[0].expressions[0]
+            .value
+            .clone(),
+        Value::from_json_str(r#"{ "a" : 1, "b" : 2 }"#)?
+    );
+
+    Ok(())
+}
+
+#[test]
+fn test_rule_values_coexist_with_merged_base_data() -> Result<()> {
+    let mut engine = Engine::new();
+
+    // Deep-merged base data under `base`...
+    engine.add_data(Value::from_json_str(
+        r#"{ "base" : { "foo" : { "a" : 1 } } }"#,
+    )?)?;
+    engine.add_data(Value::from_json_str(
+        r#"{ "base" : { "foo" : { "b" : 2 } } }"#,
+    )?)?;
+
+    engine.add_policy(
+        "policy.rego".to_string(),
+        r#"
+package test
+
+computed := data.base.foo.a + data.base.foo.b
+"#
+        .to_string(),
+    )?;
+
+    let data = engine.eval_query("data".to_string(), false)?.result[0].expressions[0]
+        .value
+        .clone();
+
+    // Base data is preserved and deep-merged...
+    assert_eq!(
+        data["base"],
+        Value::from_json_str(r#"{ "foo" : { "a" : 1, "b" : 2 } }"#)?
+    );
+    // ...and the rule-computed value materializes alongside it without disturbing the merge.
+    assert_eq!(data["test"]["computed"], Value::from(3_u64));
+
+    Ok(())
+}
diff --git a/src/value/mod.rs b/src/value/mod.rs
index ca8607b..2cf85c6 100644
--- a/src/value/mod.rs
+++ b/src/value/mod.rs
@@ -1332,6 +1332,13 @@ impl Value {
     }
 }
 
+/// Depth cap for `deep_merge`/`check_mergeable`, set at serde_json's default recursion limit.
+///
+/// Prevents a stack overflow from adversarially nested data — an uncatchable abort that poisons
+/// every engine in an FFI process. At serde_json's limit it only backstops `Value`s built without
+/// a parse-time cap: the Python/Ruby native bindings, or programmatic construction.
+const MAX_MERGE_DEPTH: usize = 128;
+
 impl Value {
     pub(crate) fn make_or_get_value_mut<'a>(&'a mut self, paths: &[&str]) -> Result<&'a mut Value> {
         if paths.is_empty() {
@@ -1365,6 +1372,11 @@ impl Value {
         }
     }
 
+    /// Shallow-merge `new` into `self` with strict rule-output semantics.
+    ///
+    /// Objects merge one level deep: a key on both sides must hold the *same* value or it is a
+    /// conflict; sets union; equal values are a no-op. Non-recursive by design — data documents
+    /// use [`Value::deep_merge`] instead.
     pub(crate) fn merge(&mut self, mut new: Value) -> Result<()> {
         if self == &new {
             return Ok(());
@@ -1372,24 +1384,26 @@ impl Value {
         match (self, &mut new) {
             (v @ Value::Undefined, _) => *v = new,
             (Value::Set(ref mut set), Value::Set(new)) => {
-                Rc::make_mut(set).append(Rc::make_mut(new));
-                // Enforce allocator limit after merging set entries.
+                // Union without deep-cloning the RHS set (see `deep_merge`).
+                let dst = Rc::make_mut(set);
+                match Rc::try_unwrap(core::mem::take(new)) {
+                    Ok(owned) => dst.extend(owned),
+                    Err(shared) => dst.extend(shared.iter().cloned()),
+                }
                 enforce_limit_anyhow()?;
             }
             (Value::Object(map), Value::Object(new)) => {
                 for (k, v) in new.iter() {
                     match map.get(k) {
-                        Some(pv) if *pv != *v => {
-                            bail!(
-                                "value for key `{}` generated multiple times: `{}` and `{}`",
-                                serde_json::to_string_pretty(&k).map_err(anyhow::Error::msg)?,
-                                serde_json::to_string_pretty(&pv).map_err(anyhow::Error::msg)?,
-                                serde_json::to_string_pretty(&v).map_err(anyhow::Error::msg)?,
-                            )
-                        }
+                        // Same key, different value: the rule produced two outputs for one path.
+                        Some(pv) if *pv != *v => bail!(
+                            "value for key `{}` generated multiple times: `{}` and `{}`",
+                            serde_json::to_string_pretty(&k).map_err(anyhow::Error::msg)?,
+                            serde_json::to_string_pretty(&pv).map_err(anyhow::Error::msg)?,
+                            serde_json::to_string_pretty(&v).map_err(anyhow::Error::msg)?,
+                        ),
                         _ => {
                             Rc::make_mut(map).insert(k.clone(), v.clone());
-                            // Enforce allocator limit after merging object entries.
                             enforce_limit_anyhow()?;
                         }
                     };
@@ -1399,6 +1413,151 @@ impl Value {
         };
         Ok(())
     }
+
+    /// Recursively deep-merge `new` into `self` — the data-document merge behind [`Engine::add_data`].
+    ///
+    /// Objects recurse per-key, sets union, equal values are a no-op, any other differing pair
+    /// conflicts. Set-union is a regorus extension (OPA data is JSON, which has no sets). Distinct
+    /// from the strict, non-recursive [`Value::merge`] used for rule outputs — use deep-merge ONLY
+    /// for data documents.
+    ///
+    /// [`Engine::add_data`]: crate::Engine::add_data
+    pub(crate) fn deep_merge(&mut self, new: Value) -> Result<()> {
+        self.deep_merge_at(new, 0)
+    }
+
+    /// Depth-tracked worker for [`deep_merge`](Value::deep_merge). See [`MAX_MERGE_DEPTH`].
+    fn deep_merge_at(&mut self, mut new: Value, depth: usize) -> Result<()> {
+        if depth >= MAX_MERGE_DEPTH {
+            bail!("data merge exceeds maximum nesting depth of {MAX_MERGE_DEPTH}");
+        }
+        if self == &new {
+            return Ok(());
+        }
+        match (self, &mut new) {
+            (v @ Value::Undefined, _) => *v = new,
+            (Value::Set(ref mut set), Value::Set(new)) => {
+                // Union without deep-cloning the RHS set: move elements if uniquely owned,
+                // else clone only the element handles (`Rc` bumps), never the whole `BTreeSet`.
+                let dst = Rc::make_mut(set);
+                match Rc::try_unwrap(core::mem::take(new)) {
+                    Ok(owned) => dst.extend(owned),
+                    Err(shared) => dst.extend(shared.iter().cloned()),
+                }
+                enforce_limit_anyhow()?;
+            }
+            (Value::Object(map), Value::Object(new)) => {
+                // What each incoming key requires of the target map. Decided from a read-only
+                // probe so a no-op or a conflict never triggers `Rc::make_mut` (and never clones
+                // a shared map); `make_mut` is taken lazily, only when a key actually mutates.
+                enum Step {
+                    Skip,
+                    Insert,
+                    Recurse,
+                    Conflict,
+                }
+                for (k, v) in new.iter() {
+                    let step = match map.get(k) {
+                        None => Step::Insert,
+                        Some(existing) if existing == v => Step::Skip,
+                        Some(existing)
+                            if matches!(
+                                (existing, v),
+                                (Value::Object(_), Value::Object(_))
+                                    | (Value::Set(_), Value::Set(_))
+                            ) =>
+                        {
+                            Step::Recurse
+                        }
+                        Some(_) => Step::Conflict,
+                    };
+                    match step {
+                        Step::Skip => {}
+                        Step::Insert => {
+                            Rc::make_mut(map).insert(k.clone(), v.clone());
+                            enforce_limit_anyhow()?;
+                        }
+                        // Both sides are containers: recurse so nested objects merge rather than
+                        // the subtree being replaced (OPA data-merge semantics).
+                        Step::Recurse => {
+                            let existing = Rc::make_mut(map).get_mut(k).ok_or_else(|| {
+                                anyhow!("internal error: key vanished during merge")
+                            })?;
+                            existing.deep_merge_at(v.clone(), depth.saturating_add(1))?;
+                        }
+                        Step::Conflict => {
+                            let existing = map.get(k).ok_or_else(|| {
+                                anyhow!("internal error: key vanished during merge")
+                            })?;
+                            bail!(
+                                "value for key `{}` generated multiple times: `{}` and `{}`",
+                                serde_json::to_string_pretty(&k).map_err(anyhow::Error::msg)?,
+                                serde_json::to_string_pretty(&existing)
+                                    .map_err(anyhow::Error::msg)?,
+                                serde_json::to_string_pretty(&v).map_err(anyhow::Error::msg)?,
+                            )
+                        }
+                    }
+                }
+            }
+            _ => bail!("error: could not merge value"),
+        };
+        Ok(())
+    }
+
+    /// Read-only check that [`deep_merge`](Value::deep_merge)-ing `other` into `self` would not
+    /// conflict, without mutating or allocating.
+    ///
+    /// Lets [`Engine::add_data`] validate before merging in place. Since a conflict is the only
+    /// way the default-build merge can fail and it depends only on the inputs, a passing scan
+    /// guarantees the in-place `deep_merge` won't fail — avoiding the alternative of cloning the
+    /// whole document into a candidate just to validate. Only overlapping keys are walked, so
+    /// disjoint additions are near-free.
+    ///
+    /// [`Engine::add_data`]: crate::Engine::add_data
+    #[cfg(not(feature = "allocator-memory-limits"))]
+    pub(crate) fn check_mergeable(&self, other: &Value) -> Result<()> {
+        self.check_mergeable_at(other, 0)
+    }
+
+    /// Depth-tracked worker for [`check_mergeable`](Value::check_mergeable). See [`MAX_MERGE_DEPTH`].
+    #[cfg(not(feature = "allocator-memory-limits"))]
+    fn check_mergeable_at(&self, other: &Value, depth: usize) -> Result<()> {
+        if depth >= MAX_MERGE_DEPTH {
+            bail!("data merge exceeds maximum nesting depth of {MAX_MERGE_DEPTH}");
+        }
+        if self == other {
+            return Ok(());
+        }
+        match (self, other) {
+            (Value::Undefined, _) => Ok(()),
+            // Set union never conflicts.
+            (Value::Set(_), Value::Set(_)) => Ok(()),
+            (Value::Object(dst), Value::Object(src)) => {
+                for (k, sv) in src.iter() {
+                    // Only overlapping keys can conflict.
+                    if let Some(dv) = dst.get(k) {
+                        let both_mergeable = matches!(
+                            (dv, sv),
+                            (Value::Object(_), Value::Object(_)) | (Value::Set(_), Value::Set(_))
+                        );
+                        if both_mergeable {
+                            dv.check_mergeable_at(sv, depth.saturating_add(1))?;
+                        } else if dv != sv {
+                            bail!(
+                                "value for key `{}` generated multiple times: `{}` and `{}`",
+                                serde_json::to_string_pretty(&k).map_err(anyhow::Error::msg)?,
+                                serde_json::to_string_pretty(&dv).map_err(anyhow::Error::msg)?,
+                                serde_json::to_string_pretty(&sv).map_err(anyhow::Error::msg)?,
+                            )
+                        }
+                    }
+                }
+                Ok(())
+            }
+            _ => bail!("error: could not merge value"),
+        }
+    }
 }
 
 impl ops::Index<&Value> for Value {
diff --git a/src/value/tests.rs b/src/value/tests.rs
index b62d51c..162d13d 100644
--- a/src/value/tests.rs
+++ b/src/value/tests.rs
@@ -2,6 +2,7 @@
 // Licensed under the MIT License.
 
 #![allow(
+    clippy::panic,
     clippy::expect_used,
     clippy::unwrap_used,
     clippy::indexing_slicing,
@@ -753,3 +754,115 @@ fn set_ord_invariant_to_insertion_order() {
     }
     assert_eq!(a.cmp(&b), core::cmp::Ordering::Equal);
 }
+
+fn top_object_rc(v: &Value) -> crate::Rc<Object> {
+    match v {
+        Value::Object(rc) => crate::Rc::clone(rc),
+        other => panic!("expected object, got {other:?}"),
+    }
+}
+
+/// A no-op deep-merge (every incoming key already present with an equal value) must not clone
+/// the target map. `deep_merge` acquires mutable access lazily, so when nothing changes at a
+/// level the shared `Rc` is left untouched.
+#[test]
+fn deep_merge_noop_subset_does_not_clone_object() {
+    let mut a = Value::from_json_str(r#"{"x": {"deep": 1}, "y": 2}"#).unwrap();
+    // Keep a second reference so the map's refcount > 1: eager `make_mut` would clone here.
+    let shared = a.clone();
+    let before = top_object_rc(&a);
+
+    // Strict subset with identical values: no insert, no recurse, no conflict at any level.
+    a.deep_merge(Value::from_json_str(r#"{"y": 2}"#).unwrap())
+        .unwrap();
+
+    let after = top_object_rc(&a);
+    assert!(
+        crate::Rc::ptr_eq(&before, &after),
+        "no-op merge must not clone the shared object map"
+    );
+    assert_eq!(a, shared, "value must be unchanged by a no-op merge");
+}
+
+/// An equal nested object under a shared key is a no-op too — the equality short-circuit runs
+/// before any mutable access, so the map is not cloned.
+#[test]
+fn deep_merge_equal_nested_object_does_not_clone() {
+    let mut a = Value::from_json_str(r#"{"cfg": {"a": 1, "b": 2}, "n": 5}"#).unwrap();
+    let _shared = a.clone();
+    let before = top_object_rc(&a);
+
+    a.deep_merge(Value::from_json_str(r#"{"cfg": {"a": 1, "b": 2}}"#).unwrap())
+        .unwrap();
+
+    let after = top_object_rc(&a);
+    assert!(
+        crate::Rc::ptr_eq(&before, &after),
+        "merging an equal nested object must not clone the map"
+    );
+}
+
+/// A conflict on the first overlapping key is reported without cloning the target map: the
+/// read-only probe detects the conflict before any mutable access is taken.
+#[test]
+fn deep_merge_conflict_does_not_clone_object() {
+    let mut a = Value::from_json_str(r#"{"x": 1, "y": 2}"#).unwrap();
+    let _shared = a.clone();
+    let before = top_object_rc(&a);
+
+    let err = a
+        .deep_merge(Value::from_json_str(r#"{"x": 999}"#).unwrap())
+        .unwrap_err();
+    assert!(format!("{err}").contains("generated multiple times"));
+
+    let after = top_object_rc(&a);
+    assert!(
+        crate::Rc::ptr_eq(&before, &after),
+        "a conflict must not clone the shared object map"
+    );
+}
+
+/// Nest `depth` objects `{"k": {"k": ... leaf}}` iteratively, so building the value can't itself
+/// overflow and there's no parser to cap depth first.
+fn nest(depth: usize, leaf: Value) -> Value {
+    let mut v = leaf;
+    for _ in 0..depth {
+        let mut m = BTreeMap::new();
+        m.insert(Value::from("k"), v);
+        v = Value::from(m);
+    }
+    v
+}
+
+/// Over-deep data must fail with a clean `Err`, not overflow the stack. A `Value` can be built
+/// without serde_json's parse-time cap (the native bindings), so `deep_merge` must guard itself.
+#[test]
+fn deep_merge_rejects_excessive_depth() {
+    let depth = super::MAX_MERGE_DEPTH + 50;
+    // Shared key `k` on both sides forces full-depth recursion; distinct leaves keep the trees
+    // unequal so the equality short-circuit never fires.
+    let mut a = nest(depth, Value::from_json_str(r#"{"a": 1}"#).unwrap());
+    let b = nest(depth, Value::from_json_str(r#"{"b": 2}"#).unwrap());
+
+    let err = a.deep_merge(b).unwrap_err();
+    assert!(
+        format!("{err}").contains("nesting depth"),
+        "expected a depth-limit error, got: {err}"
+    );
+}
+
+/// The pre-scan carries the same guard, so the default build rejects over-deep input up front
+/// (leaving the live document untouched) instead of overflowing during validation.
+#[cfg(not(feature = "allocator-memory-limits"))]
+#[test]
+fn check_mergeable_rejects_excessive_depth() {
+    let depth = super::MAX_MERGE_DEPTH + 50;
+    let a = nest(depth, Value::from_json_str(r#"{"a": 1}"#).unwrap());
+    let b = nest(depth, Value::from_json_str(r#"{"b": 2}"#).unwrap());
+
+    let err = a.check_mergeable(&b).unwrap_err();
+    assert!(
+        format!("{err}").contains("nesting depth"),
+        "expected a depth-limit error, got: {err}"
+    );
+}
diff --git a/tests/interpreter/cases/rule/multiple_outputs.yaml b/tests/interpreter/cases/rule/multiple_outputs.yaml
new file mode 100644
index 0000000..25b08cf
--- /dev/null
+++ b/tests/interpreter/cases/rule/multiple_outputs.yaml
@@ -0,0 +1,223 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+# How multiple rule definitions writing to the same path combine vs. conflict.
+# Cross-checked against OPA v1.2.0 (`opa eval`); modules use `rego.v1` to match it.
+#
+#   * Zero-arg `f()` materializes as a COMPLETE document: any two differing outputs
+#     conflict, even disjoint objects (OPA does NOT deep-merge).
+#   * Partial-object `p[k]` and ref-head `p.q.r` COMBINE across disjoint keys/paths,
+#     but conflict on the same key/leaf with a different value (no deep-merge).
+#   * Equal re-definitions (same value twice) are tolerated in every family.
+
+cases:
+  # ----------------------------------------------------------------------------
+  # Zero-arg functions: materialize as a complete document.
+  # ----------------------------------------------------------------------------
+  - note: fn_single_output
+    data: {}
+    modules:
+      - |
+        package test
+        import rego.v1
+
+        f() := {"a": 1}
+    query: data.test.f
+    want_result:
+      a: 1
+
+  - note: fn_identical_outputs_tolerated
+    data: {}
+    modules:
+      - |
+        package test
+        import rego.v1
+
+        f() := {"a": 1}
+
+        f() := {"a": 1}
+    query: data.test.f
+    want_result:
+      a: 1
+
+  - note: fn_toplevel_disjoint_conflict
+    data: {}
+    modules:
+      - |
+        package test
+        import rego.v1
+
+        f() := {"a": 1}
+
+        f() := {"b": 2}
+    query: data.test.f
+    error: "rules should not produce multiple outputs"
+
+  - note: fn_nested_object_conflict_no_deep_merge
+    data: {}
+    modules:
+      - |
+        package test
+        import rego.v1
+
+        f() := {"a": {"x": 1}}
+
+        f() := {"a": {"y": 2}}
+    query: data.test.f
+    error: "rules should not produce multiple outputs"
+
+  # ----------------------------------------------------------------------------
+  # Partial-object rules with static keys: combine across disjoint keys.
+  # ----------------------------------------------------------------------------
+  - note: partial_object_disjoint_keys_combine
+    data: {}
+    modules:
+      - |
+        package test
+        import rego.v1
+
+        p["a"] := 1
+
+        p["b"] := 2
+    query: data.test.p
+    want_result:
+      a: 1
+      b: 2
+
+  - note: partial_object_same_key_same_value_tolerated
+    data: {}
+    modules:
+      - |
+        package test
+        import rego.v1
+
+        p["a"] := 1
+
+        p["a"] := 1
+    query: data.test.p
+    want_result:
+      a: 1
+
+  - note: partial_object_same_key_diff_scalar_conflict
+    data: {}
+    modules:
+      - |
+        package test
+        import rego.v1
+
+        p["a"] := 1
+
+        p["a"] := 2
+    query: data.test.p
+    error: "rule conflicts with the following rule"
+
+  - note: partial_object_same_key_object_values_conflict_no_deep_merge
+    data: {}
+    modules:
+      - |
+        package test
+        import rego.v1
+
+        p["a"] := {"x": 1}
+
+        p["a"] := {"y": 2}
+    query: data.test.p
+    error: "rule conflicts with the following rule"
+
+  # ----------------------------------------------------------------------------
+  # Ref-head rules: combine across disjoint sub-paths.
+  # ----------------------------------------------------------------------------
+  - note: refhead_disjoint_subpaths_combine
+    data: {}
+    modules:
+      - |
+        package test
+        import rego.v1
+
+        p.q.r := 1
+
+        p.q.s := 2
+    query: data.test.p
+    want_result:
+      q:
+        r: 1
+        s: 2
+
+  - note: refhead_same_leaf_diff_value_conflict
+    data: {}
+    modules:
+      - |
+        package test
+        import rego.v1
+
+        p.q.r := 1
+
+        p.q.r := 2
+    query: data.test.p
+    error: "rule conflicts with the following rule"
+
+  - note: refhead_same_node_object_values_conflict_no_deep_merge
+    data: {}
+    modules:
+      - |
+        package test
+        import rego.v1
+
+        p.q := {"r": 1}
+
+        p.q := {"s": 2}
+    query: data.test.p
+    error: "rule conflicts with the following rule"
+
+  # ----------------------------------------------------------------------------
+  # Dynamic partial objects (keys computed at eval time).
+  # ----------------------------------------------------------------------------
+  - note: dynamic_partial_disjoint_keys_combine
+    data: {}
+    modules:
+      - |
+        package test
+        import rego.v1
+
+        m := {"a": 1, "b": 2}
+
+        p[k] := v if {
+          some k, v in m
+        }
+    query: data.test.p
+    want_result:
+      a: 1
+      b: 2
+
+  - note: dynamic_partial_same_key_same_value_tolerated
+    data: {}
+    modules:
+      - |
+        package test
+        import rego.v1
+
+        vals := [7, 7]
+
+        p[k] := v if {
+          some v in vals
+          k := "a"
+        }
+    query: data.test.p
+    want_result:
+      a: 7
+
+  - note: dynamic_partial_same_key_diff_value_conflict
+    data: {}
+    modules:
+      - |
+        package test
+        import rego.v1
+
+        vals := [1, 2]
+
+        p[k] := v if {
+          some v in vals
+          k := "a"
+        }
+    query: data.test.p
+    error: "rules must not produce multiple outputs"
diff --git a/tests/memory_limits.rs b/tests/memory_limits.rs
index 0885353..6fa85c1 100644
--- a/tests/memory_limits.rs
+++ b/tests/memory_limits.rs
@@ -185,3 +185,75 @@ fn vm_memory_limit_during_large_allocation() {
         Ok(value) => panic!("expected VM memory limit error, got value {value:?}"),
     }
 }
+
+/// On the `allocator-memory-limits` build, an `add_data` whose merge trips the memory limit
+/// mid-way must leave the data document unchanged — no partial insertions may leak. Atomicity
+/// here relies on the candidate-copy commit (`check_mergeable` models conflicts, not limits).
+#[test]
+fn add_data_memory_limit_partial_merge_is_atomic() {
+    let mut guard = LimitGuard::lock();
+    let mut engine = Engine::new();
+
+    // Seed existing data while the limit is relaxed.
+    engine
+        .add_data(Value::from_json_str(r#"{ "a": { "existing": 1 } }"#).expect("valid JSON"))
+        .expect("seed add_data");
+
+    // Merge `{ "a": { "k0": 0, ... } }` into `a` as pure insertions. The count is sized to
+    // beat the limit check's throttling — a check only fires every MEMORY_CHECK_STRIDE (16)
+    // insertions or per MEMORY_CHECK_DELTA_BYTES (32 KiB), and mimalloc's usage snapshot lags
+    // small allocations — so the trip lands mid-merge rather than after it completes.
+    let elements = 20_000;
+    let mut payload = String::with_capacity(elements * 16);
+    payload.push_str("{\"a\":{");
+    for i in 0..elements {
+        if i > 0 {
+            payload.push(',');
+        }
+        payload.push_str("\"k");
+        payload.push_str(&i.to_string());
+        payload.push_str("\":");
+        payload.push_str(&i.to_string());
+    }
+    payload.push_str("}}");
+    let big = Value::from_json_str(&payload).expect("valid JSON");
+
+    // What the engine must still hold if the add is rejected.
+    let pristine = Value::from_json_str(r#"{ "a": { "existing": 1 } }"#).expect("valid JSON");
+
+    // Budget 0: the merge's insertions trip the limit mid-way.
+    guard.set_with_additional_budget(0);
+
+    let err = engine
+        .add_data(big)
+        .expect_err("expected memory limit error during add_data merge");
+    assert_memory_limit_error(&err);
+
+    // Atomicity: the rejected add must leave data untouched — no `k*` keys leaked.
+    assert_eq!(engine.get_data(), pristine);
+}
+
+/// Companion for the candidate-copy build: a *conflict* must also be atomic (the candidate is
+/// discarded before commit). Default-build conflict atomicity is covered in
+/// `src/tests/interpreter/mod.rs`; this exercises the distinct candidate-copy branch.
+#[test]
+fn add_data_conflict_is_atomic_on_allocator_build() {
+    // Hold the lock (no budget set) so the conflict — not a limit — is the sole failure.
+    let _guard = LimitGuard::lock();
+    let mut engine = Engine::new();
+
+    engine
+        .add_data(Value::from_json_str(r#"{ "a": { "z": 1 } }"#).expect("valid JSON"))
+        .expect("seed add_data");
+
+    // `m` sorts before `z`, so a naive in-place merge inserts `m` then hits the `z` conflict
+    // (1 vs 3). The whole call must be rejected with `m` left out.
+    assert!(engine
+        .add_data(Value::from_json_str(r#"{ "a": { "m": 2, "z": 3 } }"#).expect("valid JSON"))
+        .is_err());
+
+    assert_eq!(
+        engine.get_data(),
+        Value::from_json_str(r#"{ "a": { "z": 1 } }"#).expect("valid JSON")
+    );
+}