diff --git a/Cargo.lock b/Cargo.lock
index 76c410c1..45747b7f 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -37,6 +37,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5a15f179cd60c4584b8a8c596927aadc462e27f2ca70c04e0071964a73ba7a75"
 dependencies = [
  "cfg-if",
+ "const-random",
  "getrandom 0.3.4",
  "once_cell",
  "serde",
@@ -154,6 +155,54 @@ version = "0.7.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7c02d123df017efcdfbd739ef81735b36c5ba83ec3c59c80a9d7ecc718f92e50"
 
+[[package]]
+name = "arrow-array"
+version = "56.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b02ccba2e977a3aabb4384036109ca32f552399a2bc0588f925f91ed073ce70c"
+dependencies = [
+ "ahash 0.8.12",
+ "arrow-buffer",
+ "arrow-data",
+ "arrow-schema",
+ "chrono",
+ "half",
+ "hashbrown 0.16.1",
+ "num",
+]
+
+[[package]]
+name = "arrow-buffer"
+version = "56.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a90f8bece6a9ee316a699fbbfde368a206676a1206ce89b50f07937648e76c3c"
+dependencies = [
+ "bytes",
+ "half",
+ "num",
+]
+
+[[package]]
+name = "arrow-data"
+version = "56.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "78468c813909465dd0f858950c8a0614eb63608134acf95c602ec21381258b28"
+dependencies = [
+ "arrow-buffer",
+ "arrow-schema",
+ "half",
+ "num",
+]
+
+[[package]]
+name = "arrow-schema"
+version = "56.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5a0d5eb3fe25337ff83e8333a08379bdd1540b0961b1c888f6e505d971c198e1"
+dependencies = [
+ "bitflags 2.13.0",
+]
+
 [[package]]
 name = "ascii"
 version = "1.1.0"
@@ -1536,6 +1585,7 @@ checksum = "6ea2d84b969582b4b1864a92dc5d27cd2b77b622a8d79306834f1be5ba20d84b"
 dependencies = [
  "cfg-if",
  "crunchy",
+ "num-traits",
  "zerocopy",
 ]
 
@@ -6131,6 +6181,8 @@ name = "zen-python"
 version = "2.0.2"
 dependencies = [
  "anyhow",
+ "arrow-array",
+ "arrow-schema",
  "either",
  "json_dotpath",
  "pyo3",
diff --git a/bindings/nodejs/Cargo.toml b/bindings/nodejs/Cargo.toml
index fee8c62d..6d98aaa3 100644
--- a/bindings/nodejs/Cargo.toml
+++ b/bindings/nodejs/Cargo.toml
@@ -24,5 +24,9 @@ json_dotpath = { workspace = true }
 [target.'cfg(target_family = "wasm")'.dependencies]
 iana-time-zone = { version = "0.1", features = ["fallback"] }
 
+[features]
+default = []
+data = ["zen-engine/data"]
+
 [build-dependencies]
 napi-build = "2.3"
\ No newline at end of file
diff --git a/bindings/nodejs/index.d.ts b/bindings/nodejs/index.d.ts
index cbe7fd83..3ca05eaf 100644
--- a/bindings/nodejs/index.d.ts
+++ b/bindings/nodejs/index.d.ts
@@ -493,6 +493,7 @@ export declare class ZenEngine {
   safeEvaluate(key: string, context: any, opts?: ZenEvaluateOptions | undefined | null): Promise<{ success: true, data: ZenEngineResponse } | { success: false; error: any; }>
   safeGetDecision(key: string): Promise<{ success: true, data: ZenDecision } | { success: false; error: any; }>
   evaluateBatch(requests: Array<EvaluateBatchRequest>, opts?: ZenEvaluateOptions | undefined | null): Promise<Array<{ success: true; data: ZenEngineResponse } | { success: false; error: any }>>
+  evaluateBatchRaw(requests: Array<EvaluateBatchRawRequest>, opts?: ZenEvaluateOptions | undefined | null): Promise<Array<{ success: true; data: Buffer } | { success: false; error: any }>>
   reload(): Promise<void>
   compileFailures(): Array<{ key: string; kind: string; diagnostics?: Array<{ code: string; message: string; severity: string }>; error?: string }>
   dispose(): void
@@ -506,6 +507,28 @@ export declare class ZenEngineHandlerRequest {
   getFieldRaw(path: string): unknown
 }
 
+export declare class ZenImpactAnalysis {
+  /**
+   * Candidate and baseline are full engines — they may use different
+   * loaders, documents and configuration.
+   */
+  constructor(candidate: ZenEngine, baseline: ZenEngine)
+  /**
+   * Merge shard aggregate states, finalize, and render the report template
+   * — the closing call of a declarative impact run.
+   */
+  static finish(aggregate: Buffer, states: Array<Buffer>, report?: Buffer | undefined | null): string
+  /**
+   * Both arms AND declarative aggregation in one native pass — per-record
+   * data never crosses the boundary; the response is the tiny additive
+   * aggregate state plus the impact summary: `{state, summary}`.
+   * Synchronous because the browser hosts run this inside a worker over the
+   * wasi build, where the async napi machinery is unreliable; evaluation
+   * futures resolve immediately off a current-thread runtime.
+   */
+  runAggregateSync(candidateKey: string, baselineKey: string, inputs: Array<Buffer>, aggregate: Buffer, startIndex?: number | undefined | null, opts?: ZenEvaluateOptions | undefined | null): string
+}
+
 export interface DecisionNode {
   id: string
   name: string
@@ -513,6 +536,11 @@ export interface DecisionNode {
   config: any
 }
 
+export interface EvaluateBatchRawRequest {
+  key: string
+  context: Buffer
+}
+
 export interface EvaluateBatchRequest {
   key: string
   context: any
@@ -520,12 +548,27 @@ export interface EvaluateBatchRequest {
 
 export declare function evaluateExpression(expression: string, context?: any | undefined | null): Promise<any>
 
+/**
+ * Filters raw JSON context buffers through ONE compiled expression — the
+ * expression compiles once and rows never exist as JS objects, so the
+ * per-row boundary conversion that dominates `evaluateExpressionSync`
+ * disappears. Rows that fail to parse or evaluate simply don't match.
+ */
+export declare function evaluateExpressionMany(expression: string, contexts: Array<Buffer>): Promise<Uint32Array>
+
 export declare function evaluateExpressionSync(expression: string, context?: any | undefined | null): any
 
 export declare function evaluateUnaryExpression(expression: string, context: any): Promise<boolean>
 
 export declare function evaluateUnaryExpressionSync(expression: string, context: any): boolean
 
+/**
+ * Root-level context keys the expression can read — `null` when it addresses
+ * the whole context or cannot be analyzed. Lets callers prune columns before
+ * data is ever composed or serialized.
+ */
+export declare function expressionRootReferences(expression: string): Array<string> | null
+
 export declare function nlEncodeString(value: string): string | null
 
 export declare function nlTokenizeBatch(requests: NlTokenizeRequest[], rootType: PolicyVariableType, strict?: boolean): NlResult[]
diff --git a/bindings/nodejs/index.js b/bindings/nodejs/index.js
index 937748aa..99c5a532 100644
--- a/bindings/nodejs/index.js
+++ b/bindings/nodejs/index.js
@@ -77,8 +77,8 @@ function requireNative() {
       try {
         const binding = require('@gorules/zen-engine-android-arm64')
         const bindingPackageVersion = require('@gorules/zen-engine-android-arm64/package.json').version
-        if (bindingPackageVersion !== '1.0.0-beta.7' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
-          throw new Error(`Native binding package version mismatch, expected 1.0.0-beta.7 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
+        if (bindingPackageVersion !== '2.0.2' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
+          throw new Error(`Native binding package version mismatch, expected 2.0.2 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
         }
         return binding
       } catch (e) {
@@ -93,8 +93,8 @@ function requireNative() {
       try {
         const binding = require('@gorules/zen-engine-android-arm-eabi')
         const bindingPackageVersion = require('@gorules/zen-engine-android-arm-eabi/package.json').version
-        if (bindingPackageVersion !== '1.0.0-beta.7' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
-          throw new Error(`Native binding package version mismatch, expected 1.0.0-beta.7 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
+        if (bindingPackageVersion !== '2.0.2' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
+          throw new Error(`Native binding package version mismatch, expected 2.0.2 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
         }
         return binding
       } catch (e) {
@@ -114,8 +114,8 @@ function requireNative() {
       try {
         const binding = require('@gorules/zen-engine-win32-x64-gnu')
         const bindingPackageVersion = require('@gorules/zen-engine-win32-x64-gnu/package.json').version
-        if (bindingPackageVersion !== '1.0.0-beta.7' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
-          throw new Error(`Native binding package version mismatch, expected 1.0.0-beta.7 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
+        if (bindingPackageVersion !== '2.0.2' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
+          throw new Error(`Native binding package version mismatch, expected 2.0.2 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
         }
         return binding
       } catch (e) {
@@ -130,8 +130,8 @@ function requireNative() {
       try {
         const binding = require('@gorules/zen-engine-win32-x64-msvc')
         const bindingPackageVersion = require('@gorules/zen-engine-win32-x64-msvc/package.json').version
-        if (bindingPackageVersion !== '1.0.0-beta.7' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
-          throw new Error(`Native binding package version mismatch, expected 1.0.0-beta.7 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
+        if (bindingPackageVersion !== '2.0.2' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
+          throw new Error(`Native binding package version mismatch, expected 2.0.2 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
         }
         return binding
       } catch (e) {
@@ -147,8 +147,8 @@ function requireNative() {
       try {
         const binding = require('@gorules/zen-engine-win32-ia32-msvc')
         const bindingPackageVersion = require('@gorules/zen-engine-win32-ia32-msvc/package.json').version
-        if (bindingPackageVersion !== '1.0.0-beta.7' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
-          throw new Error(`Native binding package version mismatch, expected 1.0.0-beta.7 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
+        if (bindingPackageVersion !== '2.0.2' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
+          throw new Error(`Native binding package version mismatch, expected 2.0.2 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
         }
         return binding
       } catch (e) {
@@ -163,8 +163,8 @@ function requireNative() {
       try {
         const binding = require('@gorules/zen-engine-win32-arm64-msvc')
         const bindingPackageVersion = require('@gorules/zen-engine-win32-arm64-msvc/package.json').version
-        if (bindingPackageVersion !== '1.0.0-beta.7' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
-          throw new Error(`Native binding package version mismatch, expected 1.0.0-beta.7 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
+        if (bindingPackageVersion !== '2.0.2' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
+          throw new Error(`Native binding package version mismatch, expected 2.0.2 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
         }
         return binding
       } catch (e) {
@@ -182,8 +182,8 @@ function requireNative() {
     try {
       const binding = require('@gorules/zen-engine-darwin-universal')
       const bindingPackageVersion = require('@gorules/zen-engine-darwin-universal/package.json').version
-      if (bindingPackageVersion !== '1.0.0-beta.7' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
-        throw new Error(`Native binding package version mismatch, expected 1.0.0-beta.7 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
+      if (bindingPackageVersion !== '2.0.2' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
+        throw new Error(`Native binding package version mismatch, expected 2.0.2 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
       }
       return binding
     } catch (e) {
@@ -198,8 +198,8 @@ function requireNative() {
       try {
         const binding = require('@gorules/zen-engine-darwin-x64')
         const bindingPackageVersion = require('@gorules/zen-engine-darwin-x64/package.json').version
-        if (bindingPackageVersion !== '1.0.0-beta.7' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
-          throw new Error(`Native binding package version mismatch, expected 1.0.0-beta.7 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
+        if (bindingPackageVersion !== '2.0.2' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
+          throw new Error(`Native binding package version mismatch, expected 2.0.2 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
         }
         return binding
       } catch (e) {
@@ -214,8 +214,8 @@ function requireNative() {
       try {
         const binding = require('@gorules/zen-engine-darwin-arm64')
         const bindingPackageVersion = require('@gorules/zen-engine-darwin-arm64/package.json').version
-        if (bindingPackageVersion !== '1.0.0-beta.7' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
-          throw new Error(`Native binding package version mismatch, expected 1.0.0-beta.7 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
+        if (bindingPackageVersion !== '2.0.2' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
+          throw new Error(`Native binding package version mismatch, expected 2.0.2 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
         }
         return binding
       } catch (e) {
@@ -234,8 +234,8 @@ function requireNative() {
       try {
         const binding = require('@gorules/zen-engine-freebsd-x64')
         const bindingPackageVersion = require('@gorules/zen-engine-freebsd-x64/package.json').version
-        if (bindingPackageVersion !== '1.0.0-beta.7' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
-          throw new Error(`Native binding package version mismatch, expected 1.0.0-beta.7 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
+        if (bindingPackageVersion !== '2.0.2' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
+          throw new Error(`Native binding package version mismatch, expected 2.0.2 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
         }
         return binding
       } catch (e) {
@@ -250,8 +250,8 @@ function requireNative() {
       try {
         const binding = require('@gorules/zen-engine-freebsd-arm64')
         const bindingPackageVersion = require('@gorules/zen-engine-freebsd-arm64/package.json').version
-        if (bindingPackageVersion !== '1.0.0-beta.7' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
-          throw new Error(`Native binding package version mismatch, expected 1.0.0-beta.7 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
+        if (bindingPackageVersion !== '2.0.2' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
+          throw new Error(`Native binding package version mismatch, expected 2.0.2 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
         }
         return binding
       } catch (e) {
@@ -271,8 +271,8 @@ function requireNative() {
         try {
           const binding = require('@gorules/zen-engine-linux-x64-musl')
           const bindingPackageVersion = require('@gorules/zen-engine-linux-x64-musl/package.json').version
-          if (bindingPackageVersion !== '1.0.0-beta.7' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
-            throw new Error(`Native binding package version mismatch, expected 1.0.0-beta.7 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
+          if (bindingPackageVersion !== '2.0.2' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
+            throw new Error(`Native binding package version mismatch, expected 2.0.2 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
           }
           return binding
         } catch (e) {
@@ -287,8 +287,8 @@ function requireNative() {
         try {
           const binding = require('@gorules/zen-engine-linux-x64-gnu')
           const bindingPackageVersion = require('@gorules/zen-engine-linux-x64-gnu/package.json').version
-          if (bindingPackageVersion !== '1.0.0-beta.7' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
-            throw new Error(`Native binding package version mismatch, expected 1.0.0-beta.7 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
+          if (bindingPackageVersion !== '2.0.2' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
+            throw new Error(`Native binding package version mismatch, expected 2.0.2 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
           }
           return binding
         } catch (e) {
@@ -305,8 +305,8 @@ function requireNative() {
         try {
           const binding = require('@gorules/zen-engine-linux-arm64-musl')
           const bindingPackageVersion = require('@gorules/zen-engine-linux-arm64-musl/package.json').version
-          if (bindingPackageVersion !== '1.0.0-beta.7' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
-            throw new Error(`Native binding package version mismatch, expected 1.0.0-beta.7 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
+          if (bindingPackageVersion !== '2.0.2' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
+            throw new Error(`Native binding package version mismatch, expected 2.0.2 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
           }
           return binding
         } catch (e) {
@@ -321,8 +321,8 @@ function requireNative() {
         try {
           const binding = require('@gorules/zen-engine-linux-arm64-gnu')
           const bindingPackageVersion = require('@gorules/zen-engine-linux-arm64-gnu/package.json').version
-          if (bindingPackageVersion !== '1.0.0-beta.7' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
-            throw new Error(`Native binding package version mismatch, expected 1.0.0-beta.7 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
+          if (bindingPackageVersion !== '2.0.2' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
+            throw new Error(`Native binding package version mismatch, expected 2.0.2 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
           }
           return binding
         } catch (e) {
@@ -339,8 +339,8 @@ function requireNative() {
         try {
           const binding = require('@gorules/zen-engine-linux-arm-musleabihf')
           const bindingPackageVersion = require('@gorules/zen-engine-linux-arm-musleabihf/package.json').version
-          if (bindingPackageVersion !== '1.0.0-beta.7' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
-            throw new Error(`Native binding package version mismatch, expected 1.0.0-beta.7 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
+          if (bindingPackageVersion !== '2.0.2' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
+            throw new Error(`Native binding package version mismatch, expected 2.0.2 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
           }
           return binding
         } catch (e) {
@@ -355,8 +355,8 @@ function requireNative() {
         try {
           const binding = require('@gorules/zen-engine-linux-arm-gnueabihf')
           const bindingPackageVersion = require('@gorules/zen-engine-linux-arm-gnueabihf/package.json').version
-          if (bindingPackageVersion !== '1.0.0-beta.7' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
-            throw new Error(`Native binding package version mismatch, expected 1.0.0-beta.7 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
+          if (bindingPackageVersion !== '2.0.2' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
+            throw new Error(`Native binding package version mismatch, expected 2.0.2 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
           }
           return binding
         } catch (e) {
@@ -373,8 +373,8 @@ function requireNative() {
         try {
           const binding = require('@gorules/zen-engine-linux-loong64-musl')
           const bindingPackageVersion = require('@gorules/zen-engine-linux-loong64-musl/package.json').version
-          if (bindingPackageVersion !== '1.0.0-beta.7' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
-            throw new Error(`Native binding package version mismatch, expected 1.0.0-beta.7 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
+          if (bindingPackageVersion !== '2.0.2' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
+            throw new Error(`Native binding package version mismatch, expected 2.0.2 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
           }
           return binding
         } catch (e) {
@@ -389,8 +389,8 @@ function requireNative() {
         try {
           const binding = require('@gorules/zen-engine-linux-loong64-gnu')
           const bindingPackageVersion = require('@gorules/zen-engine-linux-loong64-gnu/package.json').version
-          if (bindingPackageVersion !== '1.0.0-beta.7' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
-            throw new Error(`Native binding package version mismatch, expected 1.0.0-beta.7 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
+          if (bindingPackageVersion !== '2.0.2' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
+            throw new Error(`Native binding package version mismatch, expected 2.0.2 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
           }
           return binding
         } catch (e) {
@@ -407,8 +407,8 @@ function requireNative() {
         try {
           const binding = require('@gorules/zen-engine-linux-riscv64-musl')
           const bindingPackageVersion = require('@gorules/zen-engine-linux-riscv64-musl/package.json').version
-          if (bindingPackageVersion !== '1.0.0-beta.7' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
-            throw new Error(`Native binding package version mismatch, expected 1.0.0-beta.7 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
+          if (bindingPackageVersion !== '2.0.2' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
+            throw new Error(`Native binding package version mismatch, expected 2.0.2 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
           }
           return binding
         } catch (e) {
@@ -423,8 +423,8 @@ function requireNative() {
         try {
           const binding = require('@gorules/zen-engine-linux-riscv64-gnu')
           const bindingPackageVersion = require('@gorules/zen-engine-linux-riscv64-gnu/package.json').version
-          if (bindingPackageVersion !== '1.0.0-beta.7' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
-            throw new Error(`Native binding package version mismatch, expected 1.0.0-beta.7 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
+          if (bindingPackageVersion !== '2.0.2' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
+            throw new Error(`Native binding package version mismatch, expected 2.0.2 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
           }
           return binding
         } catch (e) {
@@ -440,8 +440,8 @@ function requireNative() {
       try {
         const binding = require('@gorules/zen-engine-linux-ppc64-gnu')
         const bindingPackageVersion = require('@gorules/zen-engine-linux-ppc64-gnu/package.json').version
-        if (bindingPackageVersion !== '1.0.0-beta.7' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
-          throw new Error(`Native binding package version mismatch, expected 1.0.0-beta.7 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
+        if (bindingPackageVersion !== '2.0.2' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
+          throw new Error(`Native binding package version mismatch, expected 2.0.2 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
         }
         return binding
       } catch (e) {
@@ -456,8 +456,8 @@ function requireNative() {
       try {
         const binding = require('@gorules/zen-engine-linux-s390x-gnu')
         const bindingPackageVersion = require('@gorules/zen-engine-linux-s390x-gnu/package.json').version
-        if (bindingPackageVersion !== '1.0.0-beta.7' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
-          throw new Error(`Native binding package version mismatch, expected 1.0.0-beta.7 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
+        if (bindingPackageVersion !== '2.0.2' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
+          throw new Error(`Native binding package version mismatch, expected 2.0.2 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
         }
         return binding
       } catch (e) {
@@ -476,8 +476,8 @@ function requireNative() {
       try {
         const binding = require('@gorules/zen-engine-openharmony-arm64')
         const bindingPackageVersion = require('@gorules/zen-engine-openharmony-arm64/package.json').version
-        if (bindingPackageVersion !== '1.0.0-beta.7' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
-          throw new Error(`Native binding package version mismatch, expected 1.0.0-beta.7 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
+        if (bindingPackageVersion !== '2.0.2' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
+          throw new Error(`Native binding package version mismatch, expected 2.0.2 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
         }
         return binding
       } catch (e) {
@@ -492,8 +492,8 @@ function requireNative() {
       try {
         const binding = require('@gorules/zen-engine-openharmony-x64')
         const bindingPackageVersion = require('@gorules/zen-engine-openharmony-x64/package.json').version
-        if (bindingPackageVersion !== '1.0.0-beta.7' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
-          throw new Error(`Native binding package version mismatch, expected 1.0.0-beta.7 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
+        if (bindingPackageVersion !== '2.0.2' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
+          throw new Error(`Native binding package version mismatch, expected 2.0.2 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
         }
         return binding
       } catch (e) {
@@ -508,8 +508,8 @@ function requireNative() {
       try {
         const binding = require('@gorules/zen-engine-openharmony-arm')
         const bindingPackageVersion = require('@gorules/zen-engine-openharmony-arm/package.json').version
-        if (bindingPackageVersion !== '1.0.0-beta.7' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
-          throw new Error(`Native binding package version mismatch, expected 1.0.0-beta.7 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
+        if (bindingPackageVersion !== '2.0.2' && process.env.NAPI_RS_ENFORCE_VERSION_CHECK && process.env.NAPI_RS_ENFORCE_VERSION_CHECK !== '0') {
+          throw new Error(`Native binding package version mismatch, expected 2.0.2 but got ${bindingPackageVersion}. You can reinstall dependencies to fix this issue.`)
         }
         return binding
       } catch (e) {
@@ -577,10 +577,13 @@ module.exports.ZenDecision = nativeBinding.ZenDecision
 module.exports.ZenDecisionContent = nativeBinding.ZenDecisionContent
 module.exports.ZenEngine = nativeBinding.ZenEngine
 module.exports.ZenEngineHandlerRequest = nativeBinding.ZenEngineHandlerRequest
+module.exports.ZenImpactAnalysis = nativeBinding.ZenImpactAnalysis
 module.exports.evaluateExpression = nativeBinding.evaluateExpression
+module.exports.evaluateExpressionMany = nativeBinding.evaluateExpressionMany
 module.exports.evaluateExpressionSync = nativeBinding.evaluateExpressionSync
 module.exports.evaluateUnaryExpression = nativeBinding.evaluateUnaryExpression
 module.exports.evaluateUnaryExpressionSync = nativeBinding.evaluateUnaryExpressionSync
+module.exports.expressionRootReferences = nativeBinding.expressionRootReferences
 module.exports.nlEncodeString = nativeBinding.nlEncodeString
 module.exports.nlTokenizeBatch = nativeBinding.nlTokenizeBatch
 module.exports.overrideConfig = nativeBinding.overrideConfig
diff --git a/bindings/nodejs/src/engine.rs b/bindings/nodejs/src/engine.rs
index 7848937b..198c11f3 100644
--- a/bindings/nodejs/src/engine.rs
+++ b/bindings/nodejs/src/engine.rs
@@ -113,6 +113,29 @@ pub struct EvaluateBatchRequest {
     pub context: Value,
 }
 
+#[napi(object)]
+pub struct EvaluateBatchRawRequest {
+    pub key: String,
+    pub context: Buffer,
+}
+
+pub struct EvaluateBatchRawResult {
+    pub success: bool,
+    pub data: Option<Buffer>,
+    pub error: Option<Value>,
+}
+
+impl ToNapiValue for EvaluateBatchRawResult {
+    unsafe fn to_napi_value(env: napi_env, val: Self) -> napi::Result<napi_value> {
+        let env_wrapper = &Env::from(env);
+        let mut obj = Object::new(env_wrapper)?;
+        obj.set("success", val.success)?;
+        obj.set("data", val.data)?;
+        obj.set("error", val.error)?;
+        Object::to_napi_value(env, obj)
+    }
+}
+
 pub struct EvaluateBatchResult {
     pub success: bool,
     pub data: Option<NodeEvalResponse>,
@@ -398,6 +421,74 @@ impl ZenEngine {
         Ok(out)
     }
 
+    #[napi(
+        ts_return_type = "Promise<Array<{ success: true; data: Buffer } | { success: false; error: any }>>"
+    )]
+    pub async fn evaluate_batch_raw(
+        &self,
+        requests: Vec<EvaluateBatchRawRequest>,
+        opts: Option<ZenEvaluateOptions>,
+    ) -> napi::Result<Vec<EvaluateBatchRawResult>> {
+        let options: EvaluationSerializedOptions = opts.unwrap_or_default().into();
+        let mode = options.trace;
+        let max_depth = options.max_depth;
+
+        let mut handles = Vec::with_capacity(requests.len());
+        for req in requests {
+            let engine = self.graph.clone();
+            let EvaluateBatchRawRequest { key, context } = req;
+            let bytes = context.to_vec();
+            handles.push(spawn_worker(move || async move {
+                let eval_opts = EvaluationOptions {
+                    trace: mode != EvaluationTraceKind::None,
+                    max_depth,
+                };
+
+                let context: zen_engine::Variable = serde_json::from_slice(&bytes).map_err(
+                    |e| serde_json::json!({ "type": "ContextError", "source": e.to_string() }),
+                )?;
+                let response = engine
+                    .evaluate_with_opts(key, context, eval_opts)
+                    .await
+                    .map_err(|e| {
+                        e.serialize_with_mode(serde_json::value::Serializer, mode)
+                            .unwrap_or_default()
+                    })?;
+                let serialized = response
+                    .serialize_with_mode(serde_json::value::Serializer, mode)
+                    .map_err(|e| {
+                        serde_json::json!({ "type": "SerializeError", "source": e.to_string() })
+                    })?;
+                serde_json::to_vec(&serialized).map_err(
+                    |e| serde_json::json!({ "type": "SerializeError", "source": e.to_string() }),
+                )
+            }));
+        }
+
+        let mut out = Vec::with_capacity(handles.len());
+        for handle in handles {
+            out.push(match handle.await {
+                Ok(Ok(data)) => EvaluateBatchRawResult {
+                    success: true,
+                    data: Some(data.into()),
+                    error: None,
+                },
+                Ok(Err(error)) => EvaluateBatchRawResult {
+                    success: false,
+                    data: None,
+                    error: Some(error),
+                },
+                Err(_) => EvaluateBatchRawResult {
+                    success: false,
+                    data: None,
+                    error: Some(Value::String("evaluation worker panicked".into())),
+                },
+            });
+        }
+
+        Ok(out)
+    }
+
     #[napi]
     pub async fn reload(&self) -> napi::Result<()> {
         let graph = self.graph.clone();
@@ -430,3 +521,114 @@ impl ZenEngine {
         }
     }
 }
+
+#[cfg(feature = "data")]
+#[napi]
+pub struct ZenImpactAnalysis {
+    pub(crate) inner: Arc<zen_engine::data::impact::ImpactAnalysis>,
+}
+
+#[cfg(feature = "data")]
+#[napi]
+impl ZenImpactAnalysis {
+    /// Candidate and baseline are full engines — they may use different
+    /// loaders, documents and configuration.
+    #[napi(constructor)]
+    pub fn new(candidate: &ZenEngine, baseline: &ZenEngine) -> Self {
+        Self {
+            inner: Arc::new(zen_engine::data::impact::ImpactAnalysis::new(
+                candidate.graph.clone(),
+                baseline.graph.clone(),
+            )),
+        }
+    }
+
+    /// Merge shard aggregate states, finalize, and render the report template
+    /// — the closing call of a declarative impact run.
+    #[napi]
+    pub fn finish(
+        aggregate: Buffer,
+        states: Vec<Buffer>,
+        report: Option<Buffer>,
+    ) -> napi::Result<String> {
+        let spec: zen_engine::data::aggregate::AggregateSpec =
+            serde_json::from_slice(aggregate.as_ref())
+                .map_err(|e| anyhow!("aggregate is not a valid spec: {e}"))?;
+        let mut aggregator = zen_engine::data::aggregate::Aggregator::compile(&spec)
+            .map_err(|e| anyhow!(e.to_string()))?;
+        for state in &states {
+            let value: serde_json::Value = serde_json::from_slice(state.as_ref())
+                .map_err(|e| anyhow!("state is not JSON: {e}"))?;
+            aggregator
+                .merge_value(value)
+                .map_err(|e| anyhow!(e.to_string()))?;
+        }
+        let rendered = report
+            .map(|template| -> napi::Result<serde_json::Value> {
+                let template: serde_json::Value = serde_json::from_slice(template.as_ref())
+                    .map_err(|e| anyhow!("report template is not JSON: {e}"))?;
+                Ok(aggregator.render(&template))
+            })
+            .transpose()?;
+
+        serde_json::to_string(&serde_json::json!({
+            "aggregate": aggregator.finalize(),
+            "report": rendered,
+        }))
+        .map_err(|e| anyhow!(e).into())
+    }
+
+    /// Both arms AND declarative aggregation in one native pass — per-record
+    /// data never crosses the boundary; the response is the tiny additive
+    /// aggregate state plus the impact summary: `{state, summary}`.
+    /// Synchronous because the browser hosts run this inside a worker over the
+    /// wasi build, where the async napi machinery is unreliable; evaluation
+    /// futures resolve immediately off a current-thread runtime.
+    #[napi]
+    pub fn run_aggregate_sync(
+        &self,
+        candidate_key: String,
+        baseline_key: String,
+        inputs: Vec<Buffer>,
+        aggregate: Buffer,
+        start_index: Option<f64>,
+        opts: Option<ZenEvaluateOptions>,
+    ) -> napi::Result<String> {
+        let analysis = self.inner.clone();
+        let spec: zen_engine::data::aggregate::AggregateSpec =
+            serde_json::from_slice(aggregate.as_ref())
+                .map_err(|e| anyhow!("aggregate is not a valid spec: {e}"))?;
+        let options: EvaluationSerializedOptions = opts.unwrap_or_default().into();
+        let eval_options = EvaluationOptions {
+            trace: false,
+            max_depth: options.max_depth,
+        };
+        let start = start_index.unwrap_or(0.0) as u64;
+
+        let runtime = napi::tokio::runtime::Builder::new_current_thread()
+            .enable_time()
+            .build()
+            .map_err(|e| anyhow!(e))?;
+
+        let mut aggregator = zen_engine::data::aggregate::Aggregator::compile(&spec)
+            .map_err(|e| anyhow!(e.to_string()))?;
+        let comparison = runtime
+            .block_on(analysis.compare(&candidate_key, &baseline_key))
+            .map_err(|e| anyhow!(e.to_string()))?;
+
+        let mut summary = zen_engine::data::impact::ImpactSummary::default();
+        for (offset, payload) in inputs.iter().enumerate() {
+            let input: zen_engine::Variable = serde_json::from_slice(payload.as_ref())
+                .map_err(|e| anyhow!("input is not JSON: {e}"))?;
+            let row = runtime.block_on(comparison.run_one_carrying(input.clone(), eval_options));
+            summary.record(&row);
+            aggregator.record_at(row.into_aggregate_context(input), start + offset as u64);
+        }
+
+        serde_json::to_string(&serde_json::json!({
+            "state": aggregator.state(),
+            "summary": summary,
+        }))
+        .map_err(|e| anyhow!(e).into())
+    }
+}
diff --git a/bindings/nodejs/src/expression.rs b/bindings/nodejs/src/expression.rs
index 0767a9b2..6505c3e7 100644
--- a/bindings/nodejs/src/expression.rs
+++ b/bindings/nodejs/src/expression.rs
@@ -1,7 +1,10 @@
 use napi::anyhow::anyhow;
+use napi::bindgen_prelude::{Buffer, Uint32Array};
 use napi_derive::napi;
 use serde_json::Value;
 
+use crate::mt::spawn_worker;
+
 #[napi]
 pub fn evaluate_expression_sync(expression: String, context: Option<Value>) -> napi::Result<Value> {
     let ctx = zen_expression::Variable::try_from_value(context.unwrap_or(Value::Null))
@@ -60,6 +63,107 @@ pub async fn render_template(template: String, context: Value) -> napi::Result<V
         .map_err(|_| anyhow!("Hook timed out"))?
 }
 
+struct ProjectedContext<'a> {
+    roots: &'a std::collections::HashSet<String>,
+}
+
+impl<'de> serde::de::Visitor<'de> for ProjectedContext<'_> {
+    type Value = zen_expression::Variable;
+
+    fn expecting(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result {
+        formatter.write_str("a JSON object")
+    }
+
+    fn visit_map<M>(self, mut access: M) -> Result<Self::Value, M::Error>
+    where
+        M: serde::de::MapAccess<'de>,
+    {
+        let mut object = zen_expression::variable::VariableMap::new();
+        while let Some(key) = access.next_key::<std::borrow::Cow<'de, str>>()? {
+            if self.roots.contains(key.as_ref()) {
+                object.insert(
+                    key.as_ref().into(),
+                    access.next_value::<zen_expression::Variable>()?,
+                );
+            } else {
+                access.next_value::<serde::de::IgnoredAny>()?;
+            }
+        }
+        Ok(zen_expression::Variable::from_object(object))
+    }
+}
+
+impl<'de> serde::de::DeserializeSeed<'de> for ProjectedContext<'_> {
+    type Value = zen_expression::Variable;
+
+    fn deserialize<D>(self, deserializer: D) -> Result<Self::Value, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        deserializer.deserialize_map(self)
+    }
+}
+
+fn parse_context(
+    payload: &[u8],
+    roots: Option<&std::collections::HashSet<String>>,
+) -> Option<zen_expression::Variable> {
+    if let Some(roots) = roots {
+        let mut deserializer = serde_json::Deserializer::from_slice(payload);
+        let seed = ProjectedContext { roots };
+        if let Ok(context) = serde::de::DeserializeSeed::deserialize(seed, &mut deserializer) {
+            return Some(context);
+        }
+    }
+    serde_json::from_slice(payload).ok()
+}
+
+fn matching_indices(expression: &str, contexts: &[Vec<u8>]) -> napi::Result<Vec<u32>> {
+    let compiled = zen_expression::compile_expression(expression)
+        .map_err(|e| anyhow!(serde_json::to_string(&e).unwrap_or_else(|_| e.to_string())))?;
+    let roots = zen_expression::expression_root_references(expression)
+        .map(|roots| roots.into_iter().collect::<std::collections::HashSet<_>>());
+
+    let mut vm = zen_expression::vm::VM::new();
+    let mut matches = Vec::new();
+    for (index, payload) in contexts.iter().enumerate() {
+        let Some(context) = parse_context(payload, roots.as_ref()) else {
+            continue;
+        };
+        if matches!(
+            compiled.evaluate_with(context, &mut vm),
+            Ok(zen_expression::Variable::Bool(true))
+        ) {
+            matches.push(index as u32);
+        }
+    }
+    Ok(matches)
+}
+
+/// Root-level context keys the expression can read — `null` when it addresses
+/// the whole context or cannot be analyzed. Lets callers prune columns before
+/// data is ever composed or serialized.
+#[napi]
+pub fn expression_root_references(expression: String) -> Option<Vec<String>> {
+    zen_expression::expression_root_references(&expression)
+}
+
+/// Filters raw JSON context buffers through ONE compiled expression — the
+/// expression compiles once and rows never exist as JS objects, so the
+/// per-row boundary conversion that dominates `evaluateExpressionSync`
+/// disappears. Rows that fail to parse or evaluate simply don't match.
+#[napi]
+pub async fn evaluate_expression_many(
+    expression: String,
+    contexts: Vec<Buffer>,
+) -> napi::Result<Uint32Array> {
+    let payloads: Vec<Vec<u8>> = contexts.iter().map(|buffer| buffer.to_vec()).collect();
+    let matches = spawn_worker(move || async move { matching_indices(&expression, &payloads) })
+        .await
+        .map_err(|_| anyhow!("Hook timed out"))??;
+    Ok(Uint32Array::new(matches))
+}
+
 #[napi]
 pub fn nl_encode_string(value: String) -> Option<String> {
     zen_expression::nl::encode_string(&value)
diff --git a/bindings/nodejs/zen-engine.wasi-browser.js b/bindings/nodejs/zen-engine.wasi-browser.js
index a23ebd00..2c3ccf9b 100644
--- a/bindings/nodejs/zen-engine.wasi-browser.js
+++ b/bindings/nodejs/zen-engine.wasi-browser.js
@@ -61,10 +61,13 @@ export const ZenDecision = __napiModule.exports.ZenDecision
 export const ZenDecisionContent = __napiModule.exports.ZenDecisionContent
 export const ZenEngine = __napiModule.exports.ZenEngine
 export const ZenEngineHandlerRequest = __napiModule.exports.ZenEngineHandlerRequest
+export const ZenImpactAnalysis = __napiModule.exports.ZenImpactAnalysis
 export const evaluateExpression = __napiModule.exports.evaluateExpression
+export const evaluateExpressionMany = __napiModule.exports.evaluateExpressionMany
 export const evaluateExpressionSync = __napiModule.exports.evaluateExpressionSync
 export const evaluateUnaryExpression = __napiModule.exports.evaluateUnaryExpression
 export const evaluateUnaryExpressionSync = __napiModule.exports.evaluateUnaryExpressionSync
+export const expressionRootReferences = __napiModule.exports.expressionRootReferences
 export const nlEncodeString = __napiModule.exports.nlEncodeString
 export const nlTokenizeBatch = __napiModule.exports.nlTokenizeBatch
 export const overrideConfig = __napiModule.exports.overrideConfig
diff --git a/bindings/nodejs/zen-engine.wasi.cjs b/bindings/nodejs/zen-engine.wasi.cjs
index 33201a8c..8bf1bc11 100644
--- a/bindings/nodejs/zen-engine.wasi.cjs
+++ b/bindings/nodejs/zen-engine.wasi.cjs
@@ -113,10 +113,13 @@ module.exports.ZenDecision = __napiModule.exports.ZenDecision
 module.exports.ZenDecisionContent = __napiModule.exports.ZenDecisionContent
 module.exports.ZenEngine = __napiModule.exports.ZenEngine
 module.exports.ZenEngineHandlerRequest = __napiModule.exports.ZenEngineHandlerRequest
+module.exports.ZenImpactAnalysis = __napiModule.exports.ZenImpactAnalysis
 module.exports.evaluateExpression = __napiModule.exports.evaluateExpression
+module.exports.evaluateExpressionMany = __napiModule.exports.evaluateExpressionMany
 module.exports.evaluateExpressionSync = __napiModule.exports.evaluateExpressionSync
 module.exports.evaluateUnaryExpression = __napiModule.exports.evaluateUnaryExpression
 module.exports.evaluateUnaryExpressionSync = __napiModule.exports.evaluateUnaryExpressionSync
+module.exports.expressionRootReferences = __napiModule.exports.expressionRootReferences
 module.exports.nlEncodeString = __napiModule.exports.nlEncodeString
 module.exports.nlTokenizeBatch = __napiModule.exports.nlTokenizeBatch
 module.exports.overrideConfig = __napiModule.exports.overrideConfig
diff --git a/bindings/python/Cargo.toml b/bindings/python/Cargo.toml
index a289e098..69ab85e1 100644
--- a/bindings/python/Cargo.toml
+++ b/bindings/python/Cargo.toml
@@ -23,3 +23,10 @@ tokio-util = { version = "0.7", features = ["rt"] }
 zen-engine = { path = "../../core/engine", features = ["arbitrary_precision"] }
 zen-expression = { path = "../../core/expression", features = ["arbitrary_precision"] }
 zen-tmpl = { path = "../../core/template" }
+arrow-array = { version = "56", optional = true, features = ["ffi"] }
+arrow-schema = { version = "56", optional = true }
+
+[features]
+default = []
+data = ["zen-engine/data"]
+arrow = ["data", "dep:arrow-array", "dep:arrow-schema"]
diff --git a/bindings/python/src/columnar.rs b/bindings/python/src/columnar.rs
new file mode 100644
index 00000000..4099c200
--- /dev/null
+++ b/bindings/python/src/columnar.rs
@@ -0,0 +1,158 @@
+use std::sync::Arc;
+
+use anyhow::anyhow;
+use arrow_array::cast::AsArray;
+use arrow_array::types::{Float32Type, Float64Type, Int16Type, Int32Type, Int64Type, Int8Type};
+use arrow_array::{Array, RecordBatch};
+use arrow_schema::DataType;
+use rust_decimal::Decimal;
+use zen_engine::Variable;
+use zen_expression::variable::VariableMap;
+
+use crate::data::insert_path;
+
+pub fn inputs_from_batch(batch: &RecordBatch) -> Result<Vec<Variable>, anyhow::Error> {
+    let rows = batch.num_rows();
+    let schema = batch.schema();
+
+    let mut columns: Vec<(Vec<String>, Vec<Option<Variable>>)> = Vec::new();
+    for (index, field) in schema.fields().iter().enumerate() {
+        let name = field.name();
+        let segments = name.split('.').map(str::to_string).collect();
+        columns.push((segments, column_values(batch.column(index), name)?));
+    }
+
+    Ok((0..rows)
+        .map(|row| {
+            let mut map = VariableMap::default();
+            for (segments, values) in &columns {
+                if let Some(value) = &values[row] {
+                    insert_path(&mut map, segments, value.clone());
+                }
+            }
+            Variable::from_object(map)
+        })
+        .collect())
+}
+
+fn column_values(
+    column: &Arc<dyn Array>,
+    name: &str,
+) -> Result<Vec<Option<Variable>>, anyhow::Error> {
+    let rows = column.len();
+    let values = match column.data_type() {
+        DataType::Null => vec![None; rows],
+        DataType::Boolean => {
+            let array = column.as_boolean();
+            (0..rows)
+                .map(|row| (!array.is_null(row)).then(|| Variable::Bool(array.value(row))))
+                .collect()
+        }
+        DataType::Int8 => int_values(column.as_primitive::<Int8Type>(), |v| v as i64),
+        DataType::Int16 => int_values(column.as_primitive::<Int16Type>(), |v| v as i64),
+        DataType::Int32 => int_values(column.as_primitive::<Int32Type>(), |v| v as i64),
+        DataType::Int64 => int_values(column.as_primitive::<Int64Type>(), |v| v),
+        DataType::Float32 => float_values(column.as_primitive::<Float32Type>(), |v| v as f64),
+        DataType::Float64 => float_values(column.as_primitive::<Float64Type>(), |v| v),
+        DataType::Utf8 => {
+            let array = column.as_string::<i32>();
+            (0..rows)
+                .map(|row| {
+                    (!array.is_null(row)).then(|| Variable::String(array.value(row).into()))
+                })
+                .collect()
+        }
+        DataType::LargeUtf8 => {
+            let array = column.as_string::<i64>();
+            (0..rows)
+                .map(|row| {
+                    (!array.is_null(row)).then(|| Variable::String(array.value(row).into()))
+                })
+                .collect()
+        }
+        DataType::Dictionary(_, value_type) if **value_type == DataType::Utf8 => {
+            let array = column.as_any_dictionary();
+            let dictionary = array
+                .values()
+                .as_string_opt::<i32>()
+                .ok_or_else(|| anyhow!("column '{name}': unsupported dictionary values"))?;
+            let interned: Vec<Variable> = (0..dictionary.len())
+                .map(|index| Variable::String(dictionary.value(index).into()))
+                .collect();
+            let keys = array.normalized_keys();
+            (0..rows)
+                .map(|row| (!array.is_null(row)).then(|| interned[keys[row]].clone()))
+                .collect()
+        }
+        DataType::Struct(fields) => {
+            let array = column.as_struct();
+            let mut children: Vec<(String, Vec<Option<Variable>>)> =
+                Vec::with_capacity(fields.len());
+            for (field, child) in fields.iter().zip(array.columns()) {
+                children.push((field.name().clone(), column_values(child, field.name())?));
+            }
+            (0..rows)
+                .map(|row| {
+                    if array.is_null(row) {
+                        return None;
+                    }
+                    let mut map = VariableMap::with_capacity(children.len());
+                    for (key, values) in &children {
+                        if let Some(value) = &values[row] {
+                            map.insert(key.as_str().into(), value.clone());
+                        }
+                    }
+                    Some(Variable::from_object(map))
+                })
+                .collect()
+        }
+        DataType::List(_) => {
+            let array = column.as_list::<i32>();
+            let values = column_values(array.values(), name)?;
+            let offsets = array.offsets();
+            (0..rows)
+                .map(|row| {
+                    if array.is_null(row) {
+                        return None;
+                    }
+                    let start = offsets[row] as usize;
+                    let end = offsets[row + 1] as usize;
+                    let items: Vec<Variable> = values[start..end]
+                        .iter()
+                        .map(|value| value.clone().unwrap_or(Variable::Null))
+                        .collect();
+                    Some(Variable::from_array(items))
+                })
+                .collect()
+        }
+        other => return Err(anyhow!("column '{name}': unsupported arrow type {other}")),
+    };
+    Ok(values)
+}
+
+fn int_values<T: arrow_array::ArrowPrimitiveType>(
+    array: &arrow_array::PrimitiveArray<T>,
+    to_i64: impl Fn(T::Native) -> i64,
+) -> Vec<Option<Variable>> {
+    (0..array.len())
+        .map(|row| {
+            (!array.is_null(row)).then(|| Variable::Number(Decimal::from(to_i64(array.value(row)))))
+        })
+        .collect()
+}
+
+fn float_values<T: arrow_array::ArrowPrimitiveType>(
+    array: &arrow_array::PrimitiveArray<T>,
+    to_f64: impl Fn(T::Native) -> f64,
+) -> Vec<Option<Variable>> {
+    (0..array.len())
+        .map(|row| {
+            if array.is_null(row) {
+                return None;
+            }
+            Decimal::try_from(to_f64(array.value(row)))
+                .ok()
+                .map(Variable::Number)
+        })
+        .collect()
+}
diff --git a/bindings/python/src/data.rs b/bindings/python/src/data.rs
new file mode 100644
index 00000000..3a0fbd5e
--- /dev/null
+++ b/bindings/python/src/data.rs
@@ -0,0 +1,185 @@
+use std::sync::Arc;
+
+use anyhow::anyhow;
+use pyo3::types::PyBytes;
+use pyo3::{pyclass, pymethods, Py, PyResult, Python};
+use zen_engine::data::impact::ImpactAnalysis;
+use zen_engine::EvaluationOptions;
+
+use crate::engine::PyZenEngine;
+use crate::mt::{block_on, worker_pool};
+use zen_engine::Variable;
+use zen_expression::variable::VariableMap;
+
+/// Dotted top-level keys compose into nested objects — `customer.firstName`
+/// becomes `{customer: {firstName: …}}`, matching how zen expressions resolve
+/// member access. Spark's `df.toJSON()` emits flat dotted keys for dotted
+/// column names; without this they would silently never match a rule.
+pub(crate) fn normalize_dotted(variable: Variable) -> Variable {
+    let Variable::Object(object) = &variable else {
+        return variable;
+    };
+    if !object.borrow().keys().any(|key| key.contains('.')) {
+        return variable;
+    }
+
+    let mut out = VariableMap::new();
+    for (key, value) in object.borrow().iter() {
+        let segments: Vec<String> = key.as_str().split('.').map(str::to_string).collect();
+        insert_path(&mut out, &segments, value.clone());
+    }
+    Variable::from_object(out)
+}
+
+pub(crate) fn insert_path(map: &mut VariableMap, segments: &[String], value: Variable) {
+    let (first, rest) = match segments.split_first() {
+        Some(parts) => parts,
+        None => return,
+    };
+    if rest.is_empty() {
+        map.insert(first.as_str().into(), value);
+        return;
+    }
+
+    let key = first.as_str().into();
+    if !matches!(map.get(&key), Some(Variable::Object(_))) {
+        map.insert(key.clone(), Variable::from_object(VariableMap::new()));
+    }
+    if let Some(Variable::Object(child)) = map.get(&key) {
+        let child = child.clone();
+        insert_path(&mut child.borrow_mut(), rest, value);
+    }
+}
+
+#[pyclass]
+#[pyo3(name = "ZenImpactAnalysis")]
+pub struct PyZenImpactAnalysis {
+    inner: Arc<ImpactAnalysis>,
+}
+
+#[pymethods]
+impl PyZenImpactAnalysis {
+    #[new]
+    pub fn new(candidate: &PyZenEngine, baseline: &PyZenEngine) -> Self {
+        Self {
+            inner: Arc::new(ImpactAnalysis::new(
+                candidate.engine.clone(),
+                baseline.engine.clone(),
+            )),
+        }
+    }
+
+    /// One entry point for both feeds: a list of JSON documents (bytes or
+    /// str), or any object implementing the Arrow PyCapsule protocol
+    /// (`__arrow_c_stream__` — a pyarrow RecordBatch, Table, polars frame…).
+    /// Inputs are parsed once and shared by both arms; dotted top-level keys
+    /// and dotted column names compose into nested objects. Returns the
+    /// serialized batch — `{rows, summary}`, with `before`/`after` present
+    /// only for changed or failed records and summary counts that merge
+    /// additively across batches. The GIL is released for the batch.
+    #[pyo3(signature = (candidate_key, baseline_key, data, max_depth=None))]
+    pub fn run_batch(
+        &self,
+        py: Python,
+        candidate_key: String,
+        baseline_key: String,
+        data: pyo3::Bound<'_, pyo3::PyAny>,
+        max_depth: Option<u8>,
+    ) -> PyResult<Py<PyBytes>> {
+        use pyo3::prelude::PyAnyMethods;
+
+        enum Feed {
+            Lines(Vec<Vec<u8>>),
+            #[cfg(feature = "arrow")]
+            Batches(Vec<arrow_array::RecordBatch>),
+        }
+
+        let feed = if let Ok(list) = data.downcast::<pyo3::types::PyList>() {
+            use pyo3::prelude::PyListMethods;
+            let mut lines = Vec::with_capacity(list.len());
+            for item in list.iter() {
+                let payload = item
+                    .extract::<Vec<u8>>()
+                    .or_else(|_| item.extract::<String>().map(String::into_bytes))
+                    .map_err(|_| {
+                        pyo3::exceptions::PyTypeError::new_err(
+                            "list items must be JSON documents as bytes or str",
+                        )
+                    })?;
+                lines.push(payload);
+            }
+            Feed::Lines(lines)
+        } else if data.hasattr("__arrow_c_stream__")? {
+            #[cfg(not(feature = "arrow"))]
+            {
+                return Err(pyo3::exceptions::PyTypeError::new_err(
+                    "Arrow input requires a build with the 'arrow' feature",
+                ));
+            }
+            #[cfg(feature = "arrow")]
+            {
+                use pyo3::types::PyCapsuleMethods;
+                let capsule_any = data.call_method0("__arrow_c_stream__")?;
+                let capsule = capsule_any.downcast::<pyo3::types::PyCapsule>()?;
+                let pointer =
+                    capsule.pointer() as *mut arrow_array::ffi_stream::FFI_ArrowArrayStream;
+                let reader =
+                    unsafe { arrow_array::ffi_stream::ArrowArrayStreamReader::from_raw(pointer) }
+                        .map_err(|e| anyhow!("arrow stream: {e}"))?;
+                let mut batches = Vec::new();
+                for batch in reader {
+                    batches.push(batch.map_err(|e| anyhow!("arrow batch: {e}"))?);
+                }
+                Feed::Batches(batches)
+            }
+        } else {
+            return Err(pyo3::exceptions::PyTypeError::new_err(
+                "expected a list of JSON documents or an Arrow stream object",
+            ));
+        };
+
+        let analysis = self.inner.clone();
+        let options = EvaluationOptions {
+            trace: false,
+            max_depth: max_depth.unwrap_or(10),
+        };
+
+        let body = py.allow_threads(move || {
+            block_on(
+                worker_pool().spawn_pinned(move || async move {
+                    let variables = match feed {
+                        Feed::Lines(lines) => {
+                            let mut variables = Vec::with_capacity(lines.len());
+                            for payload in &lines {
+                                let parsed =
+                                    serde_json::from_slice::<zen_engine::Variable>(payload)
+                                        .map_err(|e| anyhow!("input is not JSON: {e}"))?;
+                                variables.push(normalize_dotted(parsed));
+                            }
+                            variables
+                        }
+                        #[cfg(feature = "arrow")]
+                        Feed::Batches(batches) => {
+                            let mut variables = Vec::new();
+                            for batch in &batches {
+                                variables.extend(crate::columnar::inputs_from_batch(batch)?);
+                            }
+                            variables
+                        }
+                    };
+
+                    let comparison = analysis
+                        .compare(&candidate_key, &baseline_key)
+                        .await
+                        .map_err(|e| anyhow!(e.to_string()))?;
+                    let batch = comparison.run_batch(variables, options).await;
+
+                    serde_json::to_vec(&batch).map_err(|e| anyhow!(e))
+                }),
+            )
+            .map_err(|_| anyhow!("evaluation worker panicked"))?
+        })?;
+
+        Ok(PyBytes::new(py, &body).into())
+    }
+}
diff --git a/bindings/python/src/engine.rs b/bindings/python/src/engine.rs
index 081eb2d9..229a4844 100644
--- a/bindings/python/src/engine.rs
+++ b/bindings/python/src/engine.rs
@@ -22,7 +22,7 @@ use zen_engine::{DecisionEngine, EvaluationOptions};
 #[pyclass]
 #[pyo3(name = "ZenEngine")]
 pub struct PyZenEngine {
-    engine: Arc<DecisionEngine>,
+    pub(crate) engine: Arc<DecisionEngine>,
 }
 
 #[derive(Serialize, Deserialize)]
diff --git a/bindings/python/src/lib.rs b/bindings/python/src/lib.rs
index 35d17853..e43f9fea 100644
--- a/bindings/python/src/lib.rs
+++ b/bindings/python/src/lib.rs
@@ -9,6 +9,10 @@ use pyo3::prelude::PyModuleMethods;
 use pyo3::types::PyModule;
 use pyo3::{pymodule, wrap_pyfunction, Bound, PyResult, Python};
 
+#[cfg(feature = "arrow")]
+mod columnar;
+#[cfg(feature = "data")]
+mod data;
 mod content;
 mod convert;
 mod custom_node;
@@ -27,6 +31,8 @@ fn zen(_py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> {
     m.add_class::<PyZenDecision>()?;
     m.add_class::<PyExpression>()?;
     m.add_class::<PyZenDecisionContent>()?;
+    #[cfg(feature = "data")]
+    m.add_class::<data::PyZenImpactAnalysis>()?;
     m.add_function(wrap_pyfunction!(evaluate_expression, m)?)?;
     m.add_function(wrap_pyfunction!(evaluate_unary_expression, m)?)?;
     m.add_function(wrap_pyfunction!(render_template, m)?)?;
diff --git a/core/engine/Cargo.toml b/core/engine/Cargo.toml
index d376012b..6f8a289c 100644
--- a/core/engine/Cargo.toml
+++ b/core/engine/Cargo.toml
@@ -68,4 +68,5 @@ name = "engine"
 [features]
 default = []
 bindgen = ["rquickjs/bindgen"]
-arbitrary_precision = ["serde_json/arbitrary_precision", "zen-expression/arbitrary_precision", "zen-types/arbitrary_precision", "zen-tmpl/arbitrary_precision"]
\ No newline at end of file
+data = []
+arbitrary_precision = ["serde_json/arbitrary_precision", "zen-expression/arbitrary_precision", "zen-types/arbitrary_precision", "zen-tmpl/arbitrary_precision"]
diff --git a/core/engine/src/data/aggregate.rs b/core/engine/src/data/aggregate.rs
new file mode 100644
index 00000000..23f3d8f1
--- /dev/null
+++ b/core/engine/src/data/aggregate.rs
@@ -0,0 +1,742 @@
+use std::collections::BTreeMap;
+
+use rust_decimal::Decimal;
+use serde::Deserialize;
+use serde_json::Value;
+use zen_expression::expression::Standard;
+use zen_expression::vm::VM;
+use zen_expression::{compile_expression, Expression, Variable};
+
+/// Declarative, mergeable aggregation over impact records — the SQL
+/// `GROUP BY` + `FILTER (WHERE …)` vocabulary compiled to ZEN expressions.
+/// Every metric is distributive or algebraic, so shard states merge exactly;
+/// holistic metrics (quantiles, distinct) are deliberately out of scope.
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct AggregateSpec {
+    #[serde(default)]
+    pub facts: BTreeMap<String, String>,
+    pub metrics: BTreeMap<String, MetricSpec>,
+}
+
+/// One raw metric declaration — classified by which field is present.
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
+pub struct MetricSpec {
+    /// `true` counts every record; an expression counts strict-`true` results.
+    #[serde(default)]
+    pub count: Option<Value>,
+    #[serde(default)]
+    pub sum: Option<String>,
+    #[serde(default)]
+    pub min: Option<String>,
+    #[serde(default)]
+    pub max: Option<String>,
+    /// Group key expression; `metrics` nest one level (no groups in groups).
+    #[serde(default)]
+    pub by: Option<String>,
+    #[serde(default)]
+    pub metrics: Option<BTreeMap<String, MetricSpec>>,
+    /// Bounded exemplar collection: keeps the first N matching records
+    /// (by record index, exact across shard merges).
+    #[serde(default)]
+    pub sample: Option<usize>,
+    /// Sample payload expression (an object literal, usually).
+    #[serde(default)]
+    pub value: Option<String>,
+    /// Ranked samples: keep the N records with the LARGEST value of this
+    /// expression (numeric), instead of the first N. Mergeable exactly.
+    #[serde(default)]
+    pub rank_by: Option<String>,
+    /// SQL FILTER clause: the metric sees only records where this is `true`.
+    #[serde(default)]
+    pub when: Option<String>,
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum AggregateError {
+    #[error("metric '{name}': {message}")]
+    Spec { name: String, message: String },
+    #[error("expression '{source_text}' failed to compile: {message}")]
+    Compile {
+        source_text: String,
+        message: String,
+    },
+    #[error("state does not match the spec: {0}")]
+    State(String),
+}
+
+enum CompiledKind {
+    Count(Option<Expression<Standard>>),
+    Sum(Expression<Standard>),
+    Min(Expression<Standard>),
+    Max(Expression<Standard>),
+    Group {
+        by: Expression<Standard>,
+        metrics: Vec<CompiledMetric>,
+    },
+    Sample {
+        limit: usize,
+        value: Expression<Standard>,
+        rank_by: Option<Expression<Standard>>,
+    },
+}
+
+struct CompiledMetric {
+    name: String,
+    when: Option<Expression<Standard>>,
+    kind: CompiledKind,
+}
+
+pub struct Aggregator {
+    facts: Vec<(String, Expression<Standard>)>,
+    metrics: Vec<CompiledMetric>,
+    state: Vec<MetricState>,
+    index: u64,
+    /// One VM reused across every expression evaluation — a fresh VM per call
+    /// costs more than the evaluations themselves at ~19 expressions/record.
+    vm: VM,
+}
+
+#[derive(Debug, Clone)]
+enum MetricState {
+    Count(u64),
+    Sum(Decimal),
+    Min(Option<Decimal>),
+    Max(Option<Decimal>),
+    Group(BTreeMap<String, Vec<MetricState>>),
+    /// (order key, payload): record index for first-N samples, negated rank
+    /// for ranked ones — ascending sort works for both.
+    Sample(Vec<(Decimal, Variable)>),
+}
+
+fn compile(source: &str) -> Result<Expression<Standard>, AggregateError> {
+    compile_expression(source).map_err(|e| AggregateError::Compile {
+        source_text: source.to_string(),
+        message: e.to_string(),
+    })
+}
+
+fn compile_metric(
+    name: &str,
+    spec: &MetricSpec,
+    nested: bool,
+) -> Result<CompiledMetric, AggregateError> {
+    let err = |message: &str| AggregateError::Spec {
+        name: name.to_string(),
+        message: message.to_string(),
+    };
+
+    let when = spec.when.as_deref().map(compile).transpose()?;
+    let kind = if let Some(count) = &spec.count {
+        match count {
+            Value::Bool(true) => CompiledKind::Count(None),
+            Value::String(expression) => CompiledKind::Count(Some(compile(expression)?)),
+            _ => return Err(err("count must be true or an expression")),
+        }
+    } else if let Some(sum) = &spec.sum {
+        CompiledKind::Sum(compile(sum)?)
+    } else if let Some(min) = &spec.min {
+        CompiledKind::Min(compile(min)?)
+    } else if let Some(max) = &spec.max {
+        CompiledKind::Max(compile(max)?)
+    } else if let Some(by) = &spec.by {
+        if nested {
+            return Err(err("groups cannot nest"));
+        }
+        let Some(metric_specs) = &spec.metrics else {
+            return Err(err("a group needs `metrics`"));
+        };
+        let mut metrics = Vec::with_capacity(metric_specs.len());
+        for (child_name, child) in metric_specs {
+            metrics.push(compile_metric(child_name, child, true)?);
+        }
+        CompiledKind::Group {
+            by: compile(by)?,
+            metrics,
+        }
+    } else if let Some(limit) = spec.sample {
+        let Some(value) = &spec.value else {
+            return Err(err("a sample needs `value`"));
+        };
+        CompiledKind::Sample {
+            limit,
+            value: compile(value)?,
+            rank_by: spec.rank_by.as_deref().map(compile).transpose()?,
+        }
+    } else {
+        return Err(err("declare one of count/sum/min/max/by/sample"));
+    };
+
+    Ok(CompiledMetric {
+        name: name.to_string(),
+        when,
+        kind,
+    })
+}
+
+fn with_facts(
+    facts: &[(String, Expression<Standard>)],
+    context: Variable,
+    vm: &mut VM,
+) -> Variable {
+    if facts.is_empty() {
+        return context;
+    }
+    let augmented = context.depth_clone(1);
+    if let Variable::Object(object) = &augmented {
+        let mut computed = Vec::with_capacity(facts.len());
+        for (name, expression) in facts {
+            if let Ok(value) = eval(expression, &augmented, vm) {
+                computed.push((name.as_str(), value));
+            }
+        }
+        let mut borrowed = object.borrow_mut();
+        for (name, value) in computed {
+            borrowed.insert(name.into(), value);
+        }
+    }
+    augmented
+}
+
+fn initial_state(metric: &CompiledMetric) -> MetricState {
+    match &metric.kind {
+        CompiledKind::Count(_) => MetricState::Count(0),
+        CompiledKind::Sum(_) => MetricState::Sum(Decimal::ZERO),
+        CompiledKind::Min(_) => MetricState::Min(None),
+        CompiledKind::Max(_) => MetricState::Max(None),
+        CompiledKind::Group { .. } => MetricState::Group(BTreeMap::new()),
+        CompiledKind::Sample { .. } => MetricState::Sample(Vec::new()),
+    }
+}
+
+fn is_true(value: Result<Variable, zen_expression::IsolateError>) -> bool {
+    matches!(value, Ok(Variable::Bool(true)))
+}
+
+fn eval(
+    expression: &Expression<Standard>,
+    context: &Variable,
+    vm: &mut VM,
+) -> Result<Variable, zen_expression::IsolateError> {
+    expression.evaluate_with(context.clone(), vm)
+}
+
+fn as_decimal(value: Result<Variable, zen_expression::IsolateError>) -> Option<Decimal> {
+    match value {
+        Ok(Variable::Number(number)) => Some(number),
+        _ => None,
+    }
+}
+
+fn group_key(value: Result<Variable, zen_expression::IsolateError>) -> Option<String> {
+    match value {
+        Ok(Variable::String(text)) => Some(text.to_string()),
+        Ok(Variable::Number(number)) => Some(number.to_string()),
+        Ok(Variable::Bool(flag)) => Some(flag.to_string()),
+        _ => None,
+    }
+}
+
+fn apply(
+    metric: &CompiledMetric,
+    state: &mut MetricState,
+    context: &Variable,
+    index: u64,
+    vm: &mut VM,
+) {
+    if let Some(when) = &metric.when {
+        if !is_true(eval(when, context, vm)) {
+            return;
+        }
+    }
+    match (&metric.kind, state) {
+        (CompiledKind::Count(condition), MetricState::Count(count)) => {
+            let matched = match condition {
+                None => true,
+                Some(expression) => is_true(eval(expression, context, vm)),
+            };
+            if matched {
+                *count += 1;
+            }
+        }
+        (CompiledKind::Sum(expression), MetricState::Sum(total)) => {
+            if let Some(number) = as_decimal(eval(expression, context, vm)) {
+                *total += number;
+            }
+        }
+        (CompiledKind::Min(expression), MetricState::Min(slot)) => {
+            if let Some(number) = as_decimal(eval(expression, context, vm)) {
+                *slot = Some(slot.map_or(number, |current| current.min(number)));
+            }
+        }
+        (CompiledKind::Max(expression), MetricState::Max(slot)) => {
+            if let Some(number) = as_decimal(eval(expression, context, vm)) {
+                *slot = Some(slot.map_or(number, |current| current.max(number)));
+            }
+        }
+        (CompiledKind::Group { by, metrics }, MetricState::Group(groups)) => {
+            let Some(key) = group_key(eval(by, context, vm)) else {
+                return;
+            };
+            let states = groups
+                .entry(key)
+                .or_insert_with(|| metrics.iter().map(initial_state).collect());
+            for (child, child_state) in metrics.iter().zip(states.iter_mut()) {
+                apply(child, child_state, context, index, vm);
+            }
+        }
+        (
+            CompiledKind::Sample {
+                limit,
+                value,
+                rank_by,
+            },
+            MetricState::Sample(items),
+        ) => {
+            let key = match rank_by {
+                None => Decimal::from(index),
+                Some(expression) => match as_decimal(eval(expression, context, vm)) {
+                    Some(rank) => -rank,
+                    None => return,
+                },
+            };
+            if items.len() >= *limit {
+                let Some((worst, _)) = items.last() else {
+                    return;
+                };
+                if key >= *worst {
+                    return;
+                }
+            }
+            if let Ok(payload) = eval(value, context, vm) {
+                items.push((key, payload));
+                items.sort_by(|a, b| a.0.cmp(&b.0));
+                items.truncate(*limit);
+            }
+        }
+        _ => {}
+    }
+}
+
+fn merge_state(metric: &CompiledMetric, into: &mut MetricState, from: MetricState) {
+    match (into, from) {
+        (MetricState::Count(a), MetricState::Count(b)) => *a += b,
+        (MetricState::Sum(a), MetricState::Sum(b)) => *a += b,
+        (MetricState::Min(a), MetricState::Min(b)) => {
+            *a = match (*a, b) {
+                (Some(x), Some(y)) => Some(x.min(y)),
+                (x, y) => x.or(y),
+            }
+        }
+        (MetricState::Max(a), MetricState::Max(b)) => {
+            *a = match (*a, b) {
+                (Some(x), Some(y)) => Some(x.max(y)),
+                (x, y) => x.or(y),
+            }
+        }
+        (MetricState::Group(a), MetricState::Group(b)) => {
+            let CompiledKind::Group { metrics, .. } = &metric.kind else {
+                return;
+            };
+            for (key, states) in b {
+                match a.remove(&key) {
+                    None => {
+                        a.insert(key, states);
+                    }
+                    Some(mut existing) => {
+                        for ((child, into_state), from_state) in
+                            metrics.iter().zip(existing.iter_mut()).zip(states)
+                        {
+                            merge_state(child, into_state, from_state);
+                        }
+                        a.insert(key, existing);
+                    }
+                }
+            }
+        }
+        (MetricState::Sample(a), MetricState::Sample(b)) => {
+            let CompiledKind::Sample { limit, .. } = &metric.kind else {
+                return;
+            };
+            a.extend(b);
+            a.sort_by(|x, y| x.0.cmp(&y.0));
+            a.truncate(*limit);
+        }
+        _ => {}
+    }
+}
+
+fn state_to_value(state: &MetricState) -> Value {
+    match state {
+        MetricState::Count(count) => Value::from(*count),
+        MetricState::Sum(total) => decimal_value(*total),
+        MetricState::Min(slot) | MetricState::Max(slot) => {
+            slot.map(decimal_value).unwrap_or(Value::Null)
+        }
+        MetricState::Group(groups) => Value::Object(
+            groups
+                .iter()
+                .map(|(key, states)| {
+                    (key.clone(), Value::Array(states.iter().map(state_to_value).collect()))
+                })
+                .collect(),
+        ),
+        MetricState::Sample(items) => Value::Array(
+            items
+                .iter()
+                .map(|(key, value)| serde_json::json!({ "i": key.to_string(), "v": value.to_value() }))
+                .collect(),
+        ),
+    }
+}
+
+fn decimal_value(number: Decimal) -> Value {
+    serde_json::from_str::<serde_json::Number>(&number.normalize().to_string())
+        .map(Value::Number)
+        .unwrap_or(Value::Null)
+}
+
+fn state_from_value(metric: &CompiledMetric, value: Value) -> Result<MetricState, AggregateError> {
+    let bad = |what: &str| AggregateError::State(format!("metric '{}': {what}", metric.name));
+    let number = |value: Value, what: &str| -> Result<Decimal, AggregateError> {
+        value
+            .as_str()
+            .map(str::to_string)
+            .or_else(|| value.as_f64().map(|f| f.to_string()))
+            .or_else(|| Some(value.to_string()))
+            .and_then(|text| text.parse::<Decimal>().ok())
+            .ok_or_else(|| bad(what))
+    };
+    Ok(match &metric.kind {
+        CompiledKind::Count(_) => MetricState::Count(value.as_u64().ok_or_else(|| bad("count"))?),
+        CompiledKind::Sum(_) => MetricState::Sum(number(value, "sum")?),
+        CompiledKind::Min(_) => MetricState::Min(match value {
+            Value::Null => None,
+            other => Some(number(other, "min")?),
+        }),
+        CompiledKind::Max(_) => MetricState::Max(match value {
+            Value::Null => None,
+            other => Some(number(other, "max")?),
+        }),
+        CompiledKind::Group { metrics, .. } => {
+            let Value::Object(map) = value else {
+                return Err(bad("group"));
+            };
+            let mut groups = BTreeMap::new();
+            for (key, entry) in map {
+                let Value::Array(items) = entry else {
+                    return Err(bad("group entry"));
+                };
+                if items.len() != metrics.len() {
+                    return Err(bad("group arity"));
+                }
+                let states = metrics
+                    .iter()
+                    .zip(items)
+                    .map(|(child, item)| state_from_value(child, item))
+                    .collect::<Result<Vec<_>, _>>()?;
+                groups.insert(key, states);
+            }
+            MetricState::Group(groups)
+        }
+        CompiledKind::Sample { .. } => {
+            let Value::Array(items) = value else {
+                return Err(bad("sample"));
+            };
+            let mut collected = Vec::with_capacity(items.len());
+            for item in items {
+                let key = item
+                    .get("i")
+                    .and_then(|raw| match raw {
+                        Value::String(text) => text.parse::<Decimal>().ok(),
+                        Value::Number(number) => number.to_string().parse::<Decimal>().ok(),
+                        _ => None,
+                    })
+                    .ok_or_else(|| bad("sample key"))?;
+                let payload = item.get("v").cloned().unwrap_or(Value::Null);
+                collected.push((key, Variable::from(&payload)));
+            }
+            MetricState::Sample(collected)
+        }
+    })
+}
+
+impl Aggregator {
+    pub fn compile(spec: &AggregateSpec) -> Result<Self, AggregateError> {
+        let mut facts = Vec::with_capacity(spec.facts.len());
+        for (name, source) in &spec.facts {
+            facts.push((name.clone(), compile(source)?));
+        }
+        let mut metrics = Vec::with_capacity(spec.metrics.len());
+        for (name, metric) in &spec.metrics {
+            metrics.push(compile_metric(name, metric, false)?);
+        }
+        let state = metrics.iter().map(initial_state).collect();
+        Ok(Self {
+            facts,
+            metrics,
+            state,
+            index: 0,
+            vm: VM::new(),
+        })
+    }
+
+    /// Records one impact context — `{input, before, after, errors?}` — after
+    /// augmenting it with the computed facts at top level (facts win on name
+    /// clashes, mirroring how the canonical filter context shadows).
+    pub fn record(&mut self, context: Variable) {
+        let index = self.index;
+        self.record_at(context, index);
+    }
+
+    /// Like [`Self::record`] with an explicit GLOBAL record index — sharded
+    /// hosts pass the dataset position so sample ordering merges exactly.
+    pub fn record_at(&mut self, context: Variable, index: u64) {
+        let context = with_facts(&self.facts, context, &mut self.vm);
+        self.index += 1;
+        for (metric, state) in self.metrics.iter().zip(self.state.iter_mut()) {
+            apply(metric, state, &context, index, &mut self.vm);
+        }
+    }
+
+    /// Opaque, additive state — ship it across any boundary and merge exactly.
+    pub fn state(&self) -> Value {
+        serde_json::json!({
+            "index": self.index,
+            "metrics": self
+                .metrics
+                .iter()
+                .zip(self.state.iter())
+                .map(|(metric, state)| (metric.name.clone(), state_to_value(state)))
+                .collect::<serde_json::Map<String, Value>>(),
+        })
+    }
+
+    pub fn merge_value(&mut self, state: Value) -> Result<(), AggregateError> {
+        let index = state
+            .get("index")
+            .and_then(Value::as_u64)
+            .ok_or_else(|| AggregateError::State("missing index".to_string()))?;
+        let Some(Value::Object(map)) = state.get("metrics").cloned().into() else {
+            return Err(AggregateError::State("missing metrics".to_string()));
+        };
+        self.index += index;
+        for (metric, into) in self.metrics.iter().zip(self.state.iter_mut()) {
+            let Some(entry) = map.get(&metric.name).cloned() else {
+                return Err(AggregateError::State(format!(
+                    "missing metric '{}'",
+                    metric.name
+                )));
+            };
+            let from = state_from_value(metric, entry)?;
+            merge_state(metric, into, from);
+        }
+        Ok(())
+    }
+
+    /// The aggregate results as a report-template context: scalars for
+    /// counts/sums, `[{key, …metrics}]` arrays for groups (key-ordered),
+    /// index-ordered payload arrays for samples.
+    pub fn finalize(&self) -> Value {
+        let mut out = serde_json::Map::new();
+        out.insert("recordCount".to_string(), Value::from(self.index));
+        for (metric, state) in self.metrics.iter().zip(self.state.iter()) {
+            out.insert(metric.name.clone(), finalize_state(metric, state));
+        }
+        Value::Object(out)
+    }
+
+    /// Renders a report template against the finalized aggregate: any object
+    /// of the exact shape `{"$": "expression"}` evaluates in place; every
+    /// other node passes through untouched. Iteration, arithmetic and string
+    /// assembly all live in the expressions themselves (`map(bands, {...})`).
+    pub fn render(&self, template: &Value) -> Value {
+        render_node(template, &Variable::from(&self.finalize()))
+    }
+}
+
+fn finalize_state(metric: &CompiledMetric, state: &MetricState) -> Value {
+    match state {
+        MetricState::Count(count) => Value::from(*count),
+        MetricState::Sum(total) => decimal_value(*total),
+        MetricState::Min(slot) | MetricState::Max(slot) => {
+            slot.map(decimal_value).unwrap_or(Value::Null)
+        }
+        MetricState::Group(groups) => {
+            let CompiledKind::Group { metrics, .. } = &metric.kind else {
+                return Value::Null;
+            };
+            Value::Array(
+                groups
+                    .iter()
+                    .map(|(key, states)| {
+                        let mut object = serde_json::Map::new();
+                        object.insert("key".to_string(), Value::String(key.clone()));
+                        for (child, child_state) in metrics.iter().zip(states.iter()) {
+                            object.insert(child.name.clone(), finalize_state(child, child_state));
+                        }
+                        Value::Object(object)
+                    })
+                    .collect(),
+            )
+        }
+        MetricState::Sample(items) => {
+            Value::Array(items.iter().map(|(_, value)| value.to_value()).collect())
+        }
+    }
+}
+
+fn render_node(template: &Value, context: &Variable) -> Value {
+    match template {
+        Value::Object(map) => {
+            if map.len() == 1 {
+                if let Some(Value::String(expression)) = map.get("$") {
+                    return zen_expression::evaluate_expression(expression, context.clone())
+                        .map(|value| value.to_value())
+                        .unwrap_or(Value::Null);
+                }
+            }
+            Value::Object(
+                map.iter()
+                    .map(|(key, value)| (key.clone(), render_node(value, context)))
+                    .collect(),
+            )
+        }
+        Value::Array(items) => Value::Array(
+            items
+                .iter()
+                .map(|item| render_node(item, context))
+                .collect(),
+        ),
+        other => other.clone(),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn perf_drilldown() {
+        let parsed: AggregateSpec = serde_json::from_value(serde_json::json!({
+            "facts": {
+                "band": "input.applicant.creditScore >= 760 ? '760+' : input.applicant.creditScore >= 700 ? '700-759' : input.applicant.creditScore >= 660 ? '660-699' : input.applicant.creditScore >= 620 ? '620-659' : 'under 620'",
+                "approvedBoth": "before.approved == true and after.approved == true",
+            },
+            "metrics": {
+                "total": { "count": true },
+                "failed": { "count": "errors != null" },
+                "approvedBefore": { "count": "before.approved == true" },
+                "approvedAfter": { "count": "after.approved == true" },
+                "rateSumBefore": { "sum": "before.interestRate", "when": "approvedBoth" },
+                "rateSumAfter": { "sum": "after.interestRate", "when": "approvedBoth" },
+                "rateCount": { "count": "approvedBoth" },
+                "limitBefore": { "sum": "before.creditLimit" },
+                "limitAfter": { "sum": "after.creditLimit" },
+                "newlyApproved": { "count": "before.approved != true and after.approved == true" },
+                "newlyDeclined": { "count": "before.approved == true and after.approved != true" },
+                "bands": { "by": "band", "metrics": {
+                    "total": { "count": true },
+                    "approvedBefore": { "count": "before.approved == true" },
+                    "approvedAfter": { "count": "after.approved == true" },
+                    "limitBefore": { "sum": "before.creditLimit" },
+                    "limitAfter": { "sum": "after.creditLimit" },
+                } },
+                "movers": { "sample": 8, "when": "before.approved != after.approved", "value": "{id: input.evaluationId}" },
+            },
+        })).unwrap();
+        let mut aggregator = Aggregator::compile(&parsed).unwrap();
+        let contexts: Vec<Variable> = (0..20_000u64).map(|i| Variable::from(&serde_json::json!({
+            "input": { "evaluationId": format!("e{i}"), "applicant": { "creditScore": 500 + (i % 400) } },
+            "before": { "approved": i % 4 == 0, "interestRate": 7.49, "creditLimit": if i % 4 == 0 { 5000 } else { 0 } },
+            "after": { "approved": i % 4 == 0, "interestRate": 7.49, "creditLimit": if i % 4 == 0 { 5000 } else { 0 } },
+        }))).collect();
+        let started = std::time::Instant::now();
+        for context in &contexts {
+            aggregator.record(context.clone());
+        }
+        let elapsed = started.elapsed();
+        eprintln!(
+            "record(): {:.2} us/rec over {} records",
+            elapsed.as_micros() as f64 / contexts.len() as f64,
+            contexts.len()
+        );
+    }
+
+    fn spec() -> AggregateSpec {
+        serde_json::from_value(serde_json::json!({
+            "facts": {
+                "band": "input.score >= 700 ? 'high' : 'low'",
+                "flip": "before.approved != after.approved",
+            },
+            "metrics": {
+                "total": { "count": true },
+                "approvedAfter": { "count": "after.approved == true" },
+                "limitAfter": { "sum": "after.limit" },
+                "bestRate": { "min": "after.rate", "when": "after.approved == true" },
+                "bands": {
+                    "by": "band",
+                    "metrics": {
+                        "total": { "count": true },
+                        "approvedAfter": { "count": "after.approved == true" },
+                    },
+                },
+                "movers": {
+                    "sample": 2,
+                    "when": "flip",
+                    "value": "{id: input.id, after: after.approved}",
+                },
+            },
+        }))
+        .unwrap()
+    }
+
+    fn record(id: u64, score: i64, before: bool, after: bool, limit: i64, rate: f64) -> Variable {
+        Variable::from(&serde_json::json!({
+            "input": { "id": id, "score": score },
+            "before": { "approved": before },
+            "after": { "approved": after, "limit": limit, "rate": rate },
+        }))
+    }
+
+    #[test]
+    fn aggregates_and_merges() {
+        let parsed = spec();
+        let mut a = Aggregator::compile(&parsed).unwrap();
+        let mut b = Aggregator::compile(&parsed).unwrap();
+
+        a.record(record(1, 720, true, true, 5000, 7.5));
+        a.record(record(2, 650, true, false, 0, 9.0));
+        b.record_at(record(3, 710, false, true, 3000, 6.5), 2);
+        b.record_at(record(4, 600, false, false, 0, 0.0), 3);
+
+        let mut merged = Aggregator::compile(&parsed).unwrap();
+        merged.merge_value(a.state()).unwrap();
+        merged.merge_value(b.state()).unwrap();
+        let out = merged.finalize();
+
+        assert_eq!(out["recordCount"], 4);
+        assert_eq!(out["total"], Value::from(4u64));
+        assert_eq!(out["approvedAfter"], Value::from(2u64));
+        assert_eq!(out["limitAfter"].to_string(), "8000");
+        assert_eq!(out["bestRate"].to_string(), "6.5");
+        let bands = out["bands"].as_array().unwrap();
+        assert_eq!(bands.len(), 2);
+        assert_eq!(bands[0]["key"], "high");
+        assert_eq!(bands[0]["total"], Value::from(2u64));
+        assert_eq!(bands[0]["approvedAfter"], Value::from(2u64));
+        let movers = out["movers"].as_array().unwrap();
+        assert_eq!(movers.len(), 2);
+        assert_eq!(movers[0]["id"], Value::from(2u64));
+
+        let report = merged.render(&serde_json::json!({
+            "summary": { "$": "approvedAfter == 2 ? 'ok' : 'bad'" },
+            "series": { "$": "map(bands, {label: #.key, value: #.approvedAfter / #.total})" },
+            "static": "unchanged",
+        }));
+        assert_eq!(report["summary"], "ok");
+        assert_eq!(report["static"], "unchanged");
+        assert_eq!(report["series"].as_array().unwrap().len(), 2);
+    }
+}
diff --git a/core/engine/src/data/impact.rs b/core/engine/src/data/impact.rs
new file mode 100644
index 00000000..ad8ffa0d
--- /dev/null
+++ b/core/engine/src/data/impact.rs
@@ -0,0 +1,273 @@
+use std::sync::Arc;
+
+use serde::{Deserialize, Serialize};
+use zen_expression::variable::VariableMap;
+use zen_expression::Variable;
+
+use crate::decision::Decision;
+use crate::engine::{DecisionEngine, EvaluationOptions};
+use crate::error::ContentKindError;
+use crate::loader::LoaderError;
+use crate::EvaluationError;
+
+/// Compares evaluation outcomes between two engines — candidate vs baseline.
+/// The engines may differ in loaders, documents and configuration; the
+/// comparison itself is engine-agnostic: evaluate both arms per input, compare
+/// the result variables structurally, and materialise outputs only when they
+/// differ (or fail).
+pub struct ImpactAnalysis {
+    candidate: Arc<DecisionEngine>,
+    baseline: Arc<DecisionEngine>,
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum ImpactError {
+    #[error("failed to load '{key}': {source}")]
+    Load { key: String, source: LoaderError },
+    #[error("'{key}' is not an evaluable graph: {source}")]
+    ContentKind {
+        key: String,
+        source: ContentKindError,
+    },
+}
+
+#[derive(Debug, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ImpactRow {
+    pub changed: bool,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub before: Option<Variable>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub after: Option<Variable>,
+    /// Carrying mode only: the single shared output of an UNCHANGED row —
+    /// consumers that aggregate over every record read it as both arms while
+    /// the payload stays at one copy instead of two.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub result: Option<Variable>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub before_error: Option<serde_json::Value>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub after_error: Option<serde_json::Value>,
+}
+
+impl ImpactRow {
+    pub fn failed(&self) -> bool {
+        self.before_error.is_some() || self.after_error.is_some()
+    }
+
+    /// The aggregation context for this row — `{input, changed, before, after,
+    /// errors?}`. Unchanged carrying rows expose the single shared result as
+    /// both arms; `changed` is the engine's exact structural comparison (ZEN
+    /// `!=` does not deep-compare objects, so specs must not recompute it).
+    pub fn into_aggregate_context(self, input: Variable) -> Variable {
+        let mut context = VariableMap::new();
+        context.insert("input".into(), input);
+        context.insert("changed".into(), Variable::Bool(self.changed));
+        if self.failed() {
+            let mut errors = VariableMap::new();
+            if let Some(error) = &self.before_error {
+                errors.insert("before".into(), Variable::from(error));
+            }
+            if let Some(error) = &self.after_error {
+                errors.insert("after".into(), Variable::from(error));
+            }
+            context.insert("errors".into(), Variable::from_object(errors));
+        }
+        let (before, after) = match self.result {
+            Some(result) => (Some(result.clone()), Some(result)),
+            None => (self.before, self.after),
+        };
+        context.insert("before".into(), before.unwrap_or(Variable::Null));
+        context.insert("after".into(), after.unwrap_or(Variable::Null));
+        Variable::from_object(context)
+    }
+}
+
+/// Headline counts over impact rows. `changed` counts rows where both arms
+/// succeeded with differing results; rows with any failure count under
+/// `failed` instead, so `total = unchanged + changed + failed`. Summaries
+/// from separate batches merge additively, which is what makes incremental
+/// runs exact rather than approximate.
+#[derive(Debug, Default, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ImpactSummary {
+    pub total: u64,
+    pub changed: u64,
+    pub failed: u64,
+    pub before_errors: u64,
+    pub after_errors: u64,
+}
+
+impl ImpactSummary {
+    pub fn record(&mut self, row: &ImpactRow) {
+        self.total += 1;
+        if row.failed() {
+            self.failed += 1;
+            self.before_errors += u64::from(row.before_error.is_some());
+            self.after_errors += u64::from(row.after_error.is_some());
+        } else if row.changed {
+            self.changed += 1;
+        }
+    }
+
+    pub fn merge(&mut self, other: &ImpactSummary) {
+        self.total += other.total;
+        self.changed += other.changed;
+        self.failed += other.failed;
+        self.before_errors += other.before_errors;
+        self.after_errors += other.after_errors;
+    }
+}
+
+#[derive(Debug, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ImpactBatch {
+    pub rows: Vec<ImpactRow>,
+    pub summary: ImpactSummary,
+}
+
+impl ImpactAnalysis {
+    pub fn new(candidate: Arc<DecisionEngine>, baseline: Arc<DecisionEngine>) -> Self {
+        Self {
+            candidate,
+            baseline,
+        }
+    }
+
+    /// Resolves both decision handles once — every evaluation in the
+    /// comparison reuses them, so decision content is parsed per comparison,
+    /// never per record.
+    pub async fn compare(
+        &self,
+        candidate_key: &str,
+        baseline_key: &str,
+    ) -> Result<ImpactComparison, ImpactError> {
+        let candidate = resolve(&self.candidate, candidate_key).await?;
+        let baseline = resolve(&self.baseline, baseline_key).await?;
+        Ok(ImpactComparison {
+            candidate,
+            baseline,
+        })
+    }
+}
+
+async fn resolve(engine: &DecisionEngine, key: &str) -> Result<Decision, ImpactError> {
+    engine
+        .get_decision(key)
+        .await
+        .map_err(|source| ImpactError::Load {
+            key: key.to_string(),
+            source,
+        })?
+        .map_err(|source| ImpactError::ContentKind {
+            key: key.to_string(),
+            source,
+        })
+}
+
+pub struct ImpactComparison {
+    candidate: Decision,
+    baseline: Decision,
+}
+
+impl ImpactComparison {
+    /// Each input evaluates through both arms off the same variable (an Rc
+    /// bump, not a copy). Per-arm failures land in the row — a bad record
+    /// never aborts the batch.
+    pub async fn run_batch(
+        &self,
+        inputs: Vec<Variable>,
+        options: EvaluationOptions,
+    ) -> ImpactBatch {
+        self.run_batch_with(inputs, options, false).await
+    }
+
+    /// Like [`Self::run_batch`], but unchanged rows carry their single shared
+    /// output in `result` — for consumers that aggregate over every record and
+    /// would otherwise need both full outputs per row.
+    pub async fn run_batch_carrying(
+        &self,
+        inputs: Vec<Variable>,
+        options: EvaluationOptions,
+    ) -> ImpactBatch {
+        self.run_batch_with(inputs, options, true).await
+    }
+
+    async fn run_batch_with(
+        &self,
+        inputs: Vec<Variable>,
+        options: EvaluationOptions,
+        carry_unchanged: bool,
+    ) -> ImpactBatch {
+        let mut rows = Vec::with_capacity(inputs.len());
+        let mut summary = ImpactSummary::default();
+        for input in inputs {
+            let row = self.run_one_with(input, options, carry_unchanged).await;
+            summary.record(&row);
+            rows.push(row);
+        }
+        ImpactBatch { rows, summary }
+    }
+
+    pub async fn run_one_carrying(&self, input: Variable, options: EvaluationOptions) -> ImpactRow {
+        self.run_one_with(input, options, true).await
+    }
+
+    async fn run_one_with(
+        &self,
+        input: Variable,
+        options: EvaluationOptions,
+        carry_unchanged: bool,
+    ) -> ImpactRow {
+        let after = self
+            .candidate
+            .evaluate_with_opts(input.clone(), options)
+            .await;
+        let before = self.baseline.evaluate_with_opts(input, options).await;
+
+        match (before, after) {
+            (Ok(before), Ok(after)) => {
+                let changed = before.result != after.result;
+                ImpactRow {
+                    changed,
+                    before: changed.then_some(before.result),
+                    after: changed.then(|| after.result.clone()),
+                    result: (!changed && carry_unchanged).then_some(after.result),
+                    before_error: None,
+                    after_error: None,
+                }
+            }
+            (before, after) => {
+                let (before_value, before_error) = split(before);
+                let (after_value, after_error) = split(after);
+                ImpactRow {
+                    changed: true,
+                    before: before_value,
+                    after: after_value,
+                    result: None,
+                    before_error,
+                    after_error,
+                }
+            }
+        }
+    }
+}
+
+fn split(
+    outcome: Result<crate::DecisionGraphResponse, Box<EvaluationError>>,
+) -> (Option<Variable>, Option<serde_json::Value>) {
+    match outcome {
+        Ok(response) => (Some(response.result), None),
+        Err(error) => (
+            None,
+            Some(
+                error
+                    .serialize_with_mode(
+                        serde_json::value::Serializer,
+                        crate::engine::EvaluationTraceKind::None,
+                    )
+                    .unwrap_or_default(),
+            ),
+        ),
+    }
+}
diff --git a/core/engine/src/data/mod.rs b/core/engine/src/data/mod.rs
new file mode 100644
index 00000000..c2372c42
--- /dev/null
+++ b/core/engine/src/data/mod.rs
@@ -0,0 +1,2 @@
+pub mod aggregate;
+pub mod impact;
diff --git a/core/engine/src/lib.rs b/core/engine/src/lib.rs
index e5f255d8..bbc549c7 100644
--- a/core/engine/src/lib.rs
+++ b/core/engine/src/lib.rs
@@ -123,6 +123,8 @@
 #![allow(clippy::module_inception)]
 
 mod config;
+#[cfg(feature = "data")]
+pub mod data;
 mod decision;
 mod decision_graph;
 mod engine;
diff --git a/core/expression/src/exports.rs b/core/expression/src/exports.rs
index 88be01c9..881b1d3a 100644
--- a/core/expression/src/exports.rs
+++ b/core/expression/src/exports.rs
@@ -1,4 +1,10 @@
+use std::collections::BTreeSet;
+use std::rc::Rc;
+
 use crate::expression::{Standard, Unary};
+use crate::intellisense::dependency::{DependencyResolutionWalker, ReadDependency};
+use crate::lexer::Lexer;
+use crate::parser::Parser;
 use crate::variable::Variable;
 use crate::{Expression, Isolate, IsolateError};
 
@@ -24,6 +30,70 @@ pub fn evaluate_unary_expression(
     Isolate::with_environment(context).run_unary(expression)
 }
 
+/// Root-level context keys a standard expression can read, in stable order —
+/// `None` when the expression addresses the whole context (`$root` and
+/// friends) or cannot be analyzed, meaning no projection is safe. Callers use
+/// this to avoid materialising data the expression provably never touches;
+/// the set may over-include (harmless), never under-include.
+pub fn expression_root_references(expression: &str) -> Option<Vec<String>> {
+    let arena = bumpalo::Bump::new();
+    let tokens = Lexer::new().tokenize(&arena, expression).ok()?;
+    let parser = Parser::try_new(&tokens, &arena).ok()?;
+    let result = parser.standard().with_metadata().parse();
+    if !result.is_complete || result.root.has_error() {
+        return None;
+    }
+    let metadata = result.metadata.unwrap_or_default();
+    let dependencies = DependencyResolutionWalker::walk(result.root, &metadata);
+
+    let mut roots = BTreeSet::new();
+    if !collect_read_roots(&dependencies.reads, &mut roots) {
+        return None;
+    }
+    for reference in &dependencies.references {
+        if let Some(root) = reference.path.first() {
+            if !insert_root(root, &mut roots) {
+                return None;
+            }
+        }
+    }
+    Some(roots.into_iter().collect())
+}
+
+fn collect_read_roots(reads: &[ReadDependency], roots: &mut BTreeSet<String>) -> bool {
+    for read in reads {
+        match read {
+            ReadDependency::Direct { path, .. } | ReadDependency::Unresolved { path, .. } => {
+                let Some(root) = path.first() else {
+                    return false;
+                };
+                if !insert_root(root, roots) {
+                    return false;
+                }
+            }
+            ReadDependency::Iteration {
+                collection, reads, ..
+            } => {
+                let Some(root) = collection.first() else {
+                    return false;
+                };
+                if !insert_root(root, roots) || !collect_read_roots(reads, roots) {
+                    return false;
+                }
+            }
+        }
+    }
+    true
+}
+
+fn insert_root(root: &Rc<str>, roots: &mut BTreeSet<String>) -> bool {
+    if matches!(root.as_ref(), "$root" | "$" | "$nodes") {
+        return false;
+    }
+    roots.insert(root.to_string());
+    true
+}
+
 /// Compiles a standard expression
 pub fn compile_expression(expression: &str) -> Result<Expression<Standard>, IsolateError> {
     Isolate::new().compile_standard(expression)
diff --git a/core/expression/src/lib.rs b/core/expression/src/lib.rs
index 6b7d9448..8b52f51f 100644
--- a/core/expression/src/lib.rs
+++ b/core/expression/src/lib.rs
@@ -74,6 +74,7 @@ pub mod vm;
 
 pub use exports::{
     compile_expression, compile_unary_expression, evaluate_expression, evaluate_unary_expression,
+    expression_root_references,
 };
 pub use expression::{Expression, ExpressionKind, OpcodeCache};
 pub use isolate::{Isolate, IsolateError};