fix: tolerate null inputs and enforce required policy inputs (#528)

- graph input and output validation treats null optional properties as
  absent (including inside anyOf/oneOf objects) and reports the real
  error when validation still fails
- remove the NULLABILITY_DIVERGENCE diagnostic
- policies enforce required inputs also without goals; a path is missing
  when absent or null unless a step along it is optional, checked after
  reference hydration, and computed entity fields are not inputs

bindings/nodejs/dts-header.d.ts

@@ -69,7 +69,6 @@ export type PolicyDiagnosticCode =
   | 'UNRESOLVED_FUNCTION_TYPE'
   | 'IMPLICIT_ANY'
   | 'UNCHECKED_NODE'
-  | 'NULLABILITY_DIVERGENCE'
   | 'REDUNDANT_NULLISH'
   | 'REPEATED_DERIVATION'
   | 'PREFER_MATCH'

bindings/nodejs/index.d.ts

@@ -69,7 +69,6 @@ export type PolicyDiagnosticCode =
   | 'UNRESOLVED_FUNCTION_TYPE'
   | 'IMPLICIT_ANY'
   | 'UNCHECKED_NODE'
-  | 'NULLABILITY_DIVERGENCE'
   | 'REDUNDANT_NULLISH'
   | 'REPEATED_DERIVATION'
   | 'PREFER_MATCH'

core/engine/src/nodes/context.rs

@@ -5,7 +5,7 @@ use crate::nodes::result::{NodeResponse, NodeResult};
 use crate::nodes::variable_json::{Guards, VariableNode};
 use crate::nodes::NodeError;
 use crate::ZEN_CONFIG;
-use ahash::AHasher;
+use ahash::{AHasher, HashSet};
 use jsonschema::ValidationError;
 use serde::Serialize;
 use serde_json::Value;
@@ -16,6 +16,7 @@ use std::sync::atomic::Ordering;
 use std::sync::Arc;
 use thiserror::Error;
 use zen_expression::Isolate;
+use zen_types::symbol::Symbol;
 use zen_types::variable::{ToVariable, Variable};
 
 #[derive(Clone)]
@@ -110,20 +111,127 @@ where
     }
 
     pub fn validate(&self, schema: &Value, value: &Variable) -> Result<(), NodeError> {
-        let validator_cache = self.extensions.validator_cache();
-        let hash = self.hash_node();
-
-        let validator = validator_cache
-            .get_or_insert(hash, schema)
+        let validator = self
+            .extensions
+            .validator_cache()
+            .get_or_insert(self.hash_node(), schema)
             .node_context(self)?;
+        if validator.is_valid(VariableNode::new(value, &Guards::default())) {
+            return Ok(());
+        }
+
+        let current = value.deep_clone();
+        let mut removed: HashSet<String> = HashSet::default();
+        loop {
+            let guards = Guards::default();
+            let paths: Vec<String> = validator
+                .iter_errors(VariableNode::new(&current, &guards))
+                .map(|error| error.instance_path().to_string())
+                .collect();
+            drop(guards);
+            if paths.is_empty() {
+                return Ok(());
+            }
+            let before = removed.len();
+            for path in &paths {
+                Self::remove_nulls_at(&current, path, &mut removed);
+            }
+            if removed.len() == before {
+                break;
+            }
+        }
 
         let guards = Guards::default();
-        validator
-            .validate(VariableNode::new(value, &guards))
-            .map_err(|err| ValidationErrorJson::from(err))
-            .node_context(self)?;
+        let mut errors: Vec<ValidationErrorJson> = validator
+            .iter_errors(VariableNode::new(value, &guards))
+            .map(ValidationErrorJson::from)
+            .collect();
+        let chosen = errors
+            .iter()
+            .position(|error| !removed.contains(&error.path))
+            .unwrap_or(0);
+        match chosen < errors.len() {
+            true => Err(errors.swap_remove(chosen)).node_context(self),
+            false => Ok(()),
+        }
+    }
 
-        Ok(())
+    fn remove_nulls_at(value: &Variable, pointer: &str, removed: &mut HashSet<String>) {
+        let segments: Vec<String> = pointer
+            .split('/')
+            .skip(1)
+            .map(|s| s.replace("~1", "/").replace("~0", "~"))
+            .collect();
+        let Some(target) = Self::node_at(value, &segments) else {
+            return;
+        };
+        match (&target, segments.split_last()) {
+            (Variable::Null, Some((key, parents))) => {
+                if let Some(Variable::Object(object)) = Self::node_at(value, parents) {
+                    if object
+                        .borrow_mut()
+                        .remove(&Symbol::from(key.as_str()))
+                        .is_some()
+                    {
+                        removed.insert(pointer.to_string());
+                    }
+                }
+            }
+            _ => Self::strip_nulls(&target, pointer, removed),
+        }
+    }
+
+    fn node_at(value: &Variable, segments: &[String]) -> Option<Variable> {
+        segments
+            .iter()
+            .try_fold(value.shallow_clone(), |current, segment| match &current {
+                Variable::Object(o) => o
+                    .borrow()
+                    .get(&Symbol::from(segment.as_str()))
+                    .map(Variable::shallow_clone),
+                Variable::Array(a) => segment
+                    .parse::<usize>()
+                    .ok()
+                    .and_then(|i| a.borrow().get(i).map(Variable::shallow_clone)),
+                _ => None,
+            })
+    }
+
+    fn strip_nulls(value: &Variable, pointer: &str, removed: &mut HashSet<String>) {
+        let children: Vec<(String, Variable)> = match value {
+            Variable::Object(object) => {
+                let entries: Vec<(Symbol, Variable)> = object
+                    .borrow()
+                    .iter()
+                    .map(|(key, field)| (key.clone(), field.shallow_clone()))
+                    .collect();
+                let mut children = Vec::with_capacity(entries.len());
+                for (key, field) in entries {
+                    let child = format!(
+                        "{pointer}/{}",
+                        key.as_str().replace('~', "~0").replace('/', "~1")
+                    );
+                    match field {
+                        Variable::Null => {
+                            object.borrow_mut().remove(&key);
+                            removed.insert(child);
+                        }
+                        field => children.push((child, field)),
+                    }
+                }
+                children
+            }
+            Variable::Array(array) => array
+                .borrow()
+                .iter()
+                .enumerate()
+                .map(|(index, item)| (format!("{pointer}/{index}"), item.shallow_clone()))
+                .collect(),
+            _ => return,
+        };
+        for (child, field) in children {
+            Self::strip_nulls(&field, &child, removed);
+        }
     }
 
     fn hash_node(&self) -> u64 {

core/engine/src/policy/evaluator.rs

@@ -17,7 +17,7 @@ use crate::policy::blocks::{
 use crate::policy::ir::PropertyPath;
 use crate::policy::queries::dependency::{DataModelPaths, EvalGraph, WriteScope};
 use crate::policy::queries::path::PathClassifier;
-use crate::policy::queries::scope::{EntitySources, ReferenceField};
+use crate::policy::queries::scope::{EntityForm, EntitySources, ReferenceField};
 use crate::policy::refs::RefPoolIndex;
 use crate::policy::validator::InputSchema;
 use crate::workspace::db::Db;
@@ -118,8 +118,6 @@ impl EvalArtifact {
 
         self.validate_request(req)?;
 
-        let order_to_run = self.compute_order_to_run(req)?;
-
         let store = req.input.depth_clone(1);
         let ref_targets: HashSet<Arc<str>> = self
             .reference_fields
@@ -129,6 +127,8 @@ impl EvalArtifact {
         let pool_index = RefPoolIndex::from_input(&store, ref_targets);
         store.hydrate_references(&self.reference_fields, &pool_index);
 
+        let order_to_run = self.compute_order_to_run(req, &store)?;
+
         let roots: Vec<Arc<str>> = if req.goals.is_empty() {
             self.eval_graph.terminal_sinks(&self.members)
         } else {
@@ -172,6 +172,7 @@ impl EvalArtifact {
     fn compute_order_to_run(
         &self,
         req: &EvaluateRequest,
+        input: &Variable,
     ) -> Result<Vec<PropertyPath>, EvaluationError> {
         let visible = &self.members;
         let visible_order: Vec<PropertyPath> = self
@@ -185,26 +186,32 @@ impl EvalArtifact {
             .cloned()
             .collect();
 
+        let goals = match req.goals.is_empty() {
+            true => self.eval_graph.terminal_sinks(visible),
+            false => req.goals.clone(),
+        };
+        let entity_form = EntityForm::new(&self.entity_sources);
+        let mut missing: Vec<PropertyPath> = self
+            .eval_graph
+            .reachable_input_paths(&goals, visible)
+            .into_iter()
+            .filter(|p| {
+                !entity_form
+                    .rewrite(p)
+                    .is_some_and(|entity| self.eval_graph.written_at(&entity, visible))
+            })
+            .filter(|p| self.input_missing(input, p))
+            .collect();
+        if !missing.is_empty() {
+            missing.sort();
+            return Err(EvaluationError::MissingRequiredInputs { goals, missing });
+        }
+
         if req.goals.is_empty() {
             return Ok(visible_order);
         }
 
         let reachable = self.eval_graph.reachable_from(&req.goals);
-        let mut missing: Vec<PropertyPath> = self
-            .eval_graph
-            .reachable_input_paths(&req.goals, visible)
-            .into_iter()
-            .filter(|p| {
-                !self.data_model_paths.is_optional(p) && !self.input_satisfied(&req.input, p)
-            })
-            .collect();
-        if !missing.is_empty() {
-            missing.sort();
-            return Err(EvaluationError::MissingRequiredInputs {
-                goals: req.goals.clone(),
-                missing,
-            });
-        }
 
         Ok(visible_order
             .iter()
@@ -213,34 +220,32 @@ impl EvalArtifact {
             .collect())
     }
 
-    fn input_satisfied(&self, input: &Variable, path: &str) -> bool {
-        if Self::input_path_satisfied(input, path) {
-            return true;
+    fn input_missing(&self, input: &Variable, path: &str) -> bool {
+        if !self.path_missing(input, path) {
+            return false;
         }
         let Some((entity, rest)) = path.split_once('.') else {
-            return false;
+            return true;
         };
         match self.entity_sources.get(entity) {
-            Some(src) => {
-                let resolved = format!("{}.{}", src.path, rest);
-                Self::input_path_satisfied(input, &resolved)
-            }
-            None => false,
+            Some(src) => self.path_missing(input, &format!("{}.{}", src.path, rest)),
+            None => true,
         }
     }
 
-    fn input_path_satisfied(input: &Variable, path: &str) -> bool {
+    fn path_missing(&self, input: &Variable, path: &str) -> bool {
+        let optional = self.data_model_paths.optional_steps(path);
         let mut current = input.shallow_clone();
-        for segment in path.split('.') {
-            if current.as_array().is_some() {
-                return true;
+        for (i, segment) in path.split('.').enumerate() {
+            if current.as_object().is_none() {
+                return false;
             }
             match current.dot(segment) {
+                Some(Variable::Null) | None => return !optional[i..].iter().any(|o| *o),
                 Some(v) => current = v,
-                None => return false,
             }
         }
-        true
+        false
     }
 }
 

core/engine/src/policy/queries/dependency.rs

@@ -11,7 +11,7 @@ use crate::policy::blocks::{
     AnalysisContext, AnalysisSummary, Block, InstanceSource, PropertyRead, SharedDeclaredPaths,
     SharedDictionaryTypes, SharedIntelliSense, SharedPoisonedPaths, WriteTarget,
 };
-use crate::policy::ir::{DataModelIr, ParsedPolicy, PropertyPath};
+use crate::policy::ir::{DataModelIr, ParsedPolicy, PropertyPath, PropertyTypeIr};
 use crate::policy::queries::path::{PathClassifier, PathRoot};
 use crate::policy::queries::scope::{EntityForm, EntityGraph, EntitySources, VariableTypeScope};
 use crate::workspace::db::{AnalysisPass, PolicyDerivedCache, Snapshot};
@@ -410,13 +410,22 @@ impl EvalGraph {
     ) -> HashSet<Arc<str>> {
         self.reachable_from(goals)
             .into_iter()
-            .filter(|p| match self.writers.get(p.as_ref()) {
-                None => true,
-                Some(owner) => !visible.contains(&owner.policy_path),
-            })
+            .filter(|p| !self.written_at(p, visible))
             .collect()
     }
 
+    pub(crate) fn written_at(&self, path: &str, visible: &HashSet<Arc<str>>) -> bool {
+        let written = |path: &str| {
+            self.writers
+                .get(path)
+                .is_some_and(|owner| visible.contains(&owner.policy_path))
+        };
+        written(path)
+            || path
+                .match_indices('.')
+                .any(|(index, _)| written(&path[..index]))
+    }
+
     pub fn terminal_sinks(&self, visible: &HashSet<Arc<str>>) -> Vec<Arc<str>> {
         use petgraph::Outgoing;
         let mut sinks: Vec<Arc<str>> = self
@@ -825,12 +834,14 @@ impl PathPrefix {
 pub struct DataModelPaths {
     all: HashSet<PropertyPath>,
     optional: HashSet<PropertyPath>,
+    targets: HashMap<PropertyPath, Arc<str>>,
 }
 
 impl DataModelPaths {
     pub(crate) fn from_models<'a>(models: impl IntoIterator<Item = &'a DataModelIr>) -> Self {
         let mut all = HashSet::default();
         let mut optional = HashSet::default();
+        let mut targets = HashMap::default();
         for dm in models {
             let is_global = dm.scope.is_global();
             for prop in &dm.properties {
@@ -842,10 +853,19 @@ impl DataModelPaths {
                 if prop.optional {
                     optional.insert(path.clone());
                 }
+                if let PropertyTypeIr::Relationship { target }
+                | PropertyTypeIr::Reference { target } = &prop.kind
+                {
+                    targets.insert(path.clone(), target.clone());
+                }
                 all.insert(path);
             }
         }
-        Self { all, optional }
+        Self {
+            all,
+            optional,
+            targets,
+        }
     }
 
     pub fn matches_prefix(&self, write_path: &str) -> Option<&PropertyPath> {
@@ -857,8 +877,24 @@ impl DataModelPaths {
             .find(|p| PathPrefix::extends(p, write_path) || PathPrefix::extends(write_path, p))
     }
 
-    pub fn is_optional(&self, path: &str) -> bool {
-        self.optional.contains(path) || self.optional.iter().any(|p| PathPrefix::extends(p, path))
+    pub(crate) fn optional_steps(&self, path: &str) -> Vec<bool> {
+        let mut owner: Option<Arc<str>> = None;
+        path.split('.')
+            .enumerate()
+            .map(|(i, segment)| {
+                let key = match &owner {
+                    Some(owner) => self.all.get(format!("{owner}.{segment}").as_str()),
+                    None if i == 0 && !self.all.contains(segment) => {
+                        owner = Some(Arc::from(segment));
+                        return false;
+                    }
+                    None if i == 0 => self.all.get(segment),
+                    None => None,
+                };
+                owner = key.and_then(|k| self.targets.get(k).cloned());
+                key.is_some_and(|k| self.optional.contains(k))
+            })
+            .collect()
     }
 }
 

core/engine/src/workspace/graph/analysis.rs

@@ -510,18 +510,6 @@ impl<'a> GraphAnalyzer<'a> {
                             ),
                         ));
                     }
-                    if let Some(schema) = content.schema.as_ref() {
-                        let divergent = super::SchemaType::nullability_divergences(schema);
-                        for path in divergent.iter().take(8) {
-                            self.diagnostics.push(Diagnostic::warning(
-                                DiagnosticCode::NullabilityDivergence,
-                                DiagnosticLocation::block(self.path.clone(), node.id.clone()),
-                                format!(
-                                    "optional property `{path}` reads as nullable, but its schema does not allow null — a payload carrying `{path}: null` fails validation at runtime; add \"null\" to its type if null is a real value, or ignore this if the field is strictly absent-or-present"
-                                ),
-                            ));
-                        }
-                    }
                 }
             }
             DecisionNodeKind::OutputNode { content } => {

core/engine/src/workspace/graph/schema.rs

@@ -61,74 +61,6 @@ impl SchemaType {
         }
     }
 
-    /// Optional properties resolve statically to `T?`, yet the runtime validator
-    /// rejects an explicit `null` unless the type admits it.
-    pub(crate) fn nullability_divergences(schema: &Value) -> Vec<String> {
-        let mut out = Vec::new();
-        Self::collect_divergences(schema, String::new(), &mut out);
-        out
-    }
-
-    fn collect_divergences(schema: &Value, path: String, out: &mut Vec<String>) {
-        let Some(object) = schema.as_object() else {
-            return;
-        };
-        if let Some(items) = object.get("items") {
-            let item_path = if path.is_empty() {
-                "[]".to_string()
-            } else {
-                format!("{path}[]")
-            };
-            Self::collect_divergences(items, item_path, out);
-        }
-        let Some(properties) = object.get("properties").and_then(Value::as_object) else {
-            return;
-        };
-        let required: Vec<&str> = object
-            .get("required")
-            .and_then(Value::as_array)
-            .map(|list| list.iter().filter_map(Value::as_str).collect())
-            .unwrap_or_default();
-        for (name, prop_schema) in properties {
-            let child_path = if path.is_empty() {
-                name.clone()
-            } else {
-                format!("{path}.{name}")
-            };
-            if !required.contains(&name.as_str()) && !Self::admits_null(prop_schema) {
-                out.push(child_path.clone());
-            }
-            Self::collect_divergences(prop_schema, child_path, out);
-        }
-    }
-
-    fn admits_null(schema: &Value) -> bool {
-        let Some(object) = schema.as_object() else {
-            return true;
-        };
-        if object.get("$dictionary").is_some() {
-            return false;
-        }
-        if let Some(cases) = object
-            .get("anyOf")
-            .or_else(|| object.get("oneOf"))
-            .and_then(Value::as_array)
-        {
-            return cases.iter().any(Self::admits_null);
-        }
-        if let Some(values) = object.get("enum").and_then(Value::as_array) {
-            return values.iter().any(Value::is_null);
-        }
-        match object.get("type") {
-            Some(Value::String(kind)) => kind == "null",
-            Some(Value::Array(kinds)) => kinds
-                .iter()
-                .filter_map(Value::as_str)
-                .any(|kind| kind == "null"),
-            _ => true,
-        }
-    }
-
     pub(crate) fn inline_enum_paths(schema: &Value) -> Vec<String> {
         let mut out = Vec::new();
         Self::collect_inline_enums(schema, String::new(), &mut out);

core/engine/src/workspace/types/diagnostic.rs

@@ -187,7 +187,6 @@ pub enum DiagnosticCode {
     UnresolvedFunctionType,
     ImplicitAny,
     UncheckedNode,
-    NullabilityDivergence,
 
     RedundantNullish,
     RepeatedDerivation,

core/engine/tests/workspace_graph.rs

@@ -2992,54 +2992,6 @@ fn decision_boundary_allows_missing_optional_item_field() {
     );
 }
 
-#[test]
-fn optional_property_without_null_type_warns_of_divergence() {
-    let mut ws = Workspace::new();
-    let schema = json!({
-        "type": "object",
-        "properties": {
-            "age": { "type": "number" },
-            "name": { "type": "string" },
-            "alias": { "type": ["string", "null"] },
-            "tags": {
-                "type": "array",
-                "items": {
-                    "type": "object",
-                    "properties": { "label": { "type": "string" }, "weight": { "type": "number" } },
-                    "required": ["label"]
-                }
-            }
-        },
-        "required": ["age", "tags"]
-    });
-    ws.set_document(
-        "g",
-        document(linear_graph(
-            Some(schema),
-            vec![expression_node("calc", &[("x", "age * 2")])],
-        )),
-    );
-    let diagnostics = ws.diagnostics("g");
-    let divergent: Vec<&str> = diagnostics
-        .iter()
-        .filter(|d| d.code == DiagnosticCode::NullabilityDivergence)
-        .map(|d| d.message.as_str())
-        .collect();
-    assert_eq!(divergent.len(), 2, "{divergent:?}");
-    assert!(
-        divergent.iter().any(|m| m.contains("`name`")),
-        "{divergent:?}"
-    );
-    assert!(
-        divergent.iter().any(|m| m.contains("`tags[].weight`")),
-        "{divergent:?}"
-    );
-    assert!(
-        !divergent.iter().any(|m| m.contains("`alias`")),
-        "a type that allows null must not warn: {divergent:?}"
-    );
-}
-
 #[test]
 fn decision_boundary_names_nullability_delta() {
     let mut ws = Workspace::new();