Constraint walker and polarity analysis

- Polarity analysis walker on Constraint.t AST
- Cascade constraint re-checking on DELETE and INSERT
- Focused filtering to narrow re-checks to affected tuples only
- Universal variable substitution for targeted constraint re-evaluation
- Constant discovery for early bail-out (trigger_constants)
- Deferred constraint timing with commit boundary
- attach_constraint and commit API in Manipulation
- extend_tuple namespacing fix (variable.attr convention)
- History capping at 128 entries in Database.update_state
- Benchmark binary (bin/bench.ml)
- Design doc (docs/incremental_constraint_checking.org)

1 files changed, 756 insertions, 81 deletions

diff --git a/test/test_relational_engine.ml b/test/test_relational_engine.ml
index 963337f..081ed39 100644
--- a/test/test_relational_engine.ml
+++ b/test/test_relational_engine.ml
@@ -2,10 +2,6 @@
 
 open Relational_engine
 
-(* ============================================================================
-   Merkle Tree Tests
-   ============================================================================ *)
-
 let%test_unit "merkle: empty tree" =
   let tree = Merkle.empty in
   assert (Merkle.is_empty tree);
@@ -70,10 +66,6 @@ let%test_unit "merkle: same elements produce same root hash" =
     |> Merkle.insert "a" in
   assert (Merkle.root_hash tree1 = Merkle.root_hash tree2)
 
-(* ============================================================================
-   Physical Storage Tests
-   ============================================================================ *)
-
 let%test_unit "storage: create and close" =
   match Management.Physical.Memory.create () with
   | Error _ -> assert false
@@ -145,10 +137,6 @@ let%test_unit "storage: transaction commit" =
         | Ok exists -> assert exists));
     Management.Physical.Memory.close storage
 
-(* ============================================================================
-   Database Tests
-   ============================================================================ *)
-
 let%test_unit "database: create empty" =
   let db = Management.Database.empty ~name:"test_db" in
   assert (db.name = "test_db");
@@ -250,10 +238,6 @@ let%test_unit "database: get relation names" =
   assert (List.mem "users" names);
   assert (List.mem "orders" names)
 
-(* ============================================================================
-   Manipulation Tests (using Memory storage)
-   ============================================================================ *)
-
 (* Helper to create storage for tests *)
 let with_storage f =
   match Management.Physical.Memory.create () with
@@ -556,10 +540,6 @@ let%test_unit "manipulation: get_relation from database" =
         assert (rel.name = "items");
         assert (rel.schema = schema))
 
-(* ============================================================================
-   Schema Persistence Tests
-   ============================================================================ *)
-
 let%test_unit "schema: persisted and loaded correctly" =
   with_storage (fun storage ->
     let db = match Manipulation.Memory.create_database storage ~name:"test_db" with
@@ -585,10 +565,6 @@ let%test_unit "schema: persisted and loaded correctly" =
         assert (List.mem ("name", "string") loaded_rel.schema);
         assert (List.mem ("email", "string") loaded_rel.schema))
 
-(* ============================================================================
-   System Catalog Tests
-   ============================================================================ *)
-
 let%test_unit "catalog: create_database seeds 6 catalog relations" =
   with_storage (fun storage ->
     match Manipulation.Memory.create_database storage ~name:"test" with
@@ -714,10 +690,6 @@ let%test_unit "catalog: register_constraint inserts into sakura:constraint" =
       in
       assert (Manipulation.Memory.tuple_count con_rel = 1))
 
-(* ============================================================================
-   Integration Tests
-   ============================================================================ *)
-
 let%test_unit "integration: full workflow with storage" =
   with_storage (fun storage ->
     (* Create database *)
@@ -853,10 +825,6 @@ let%test_unit "integration: hash bubbles up correctly" =
     | Some rel_hash ->
       assert (Merkle.member rel_hash db.tree))
 
-(* ============================================================================
-   Time-Travel / Branching Tests
-   ============================================================================ *)
-
 let%test_unit "branching: load database from historical hash" =
   with_storage (fun storage ->
     let db = match Manipulation.Memory.create_database storage ~name:"test" with
@@ -1044,10 +1012,6 @@ let%test_unit "branching: full reconstruction from hash" =
           assert (List.mem "Banana" names);
           assert (List.mem "Cherry" names))
 
-(* ============================================================================
-   Algebra Tests
-   ============================================================================ *)
-
 (* Helper: build a materialized tuple with int attributes *)
 let make_int_tuple relation pairs =
   let attributes =
@@ -1266,10 +1230,6 @@ let%test_unit "algebra: take limits output" =
       | Error _ -> assert false
       | Ok rows -> assert (List.length rows = 3))
 
-(* ============================================================================
-   DRL Parser Tests
-   ============================================================================ *)
-
 let%test_unit "drl: parse Base" =
   match Drl.Parser.of_string {|(Base "users")|} with
   | Error _              -> assert false
@@ -1295,10 +1255,6 @@ let%test_unit "drl: parse Select" =
   | Ok (Drl.Ast.Select (Drl.Ast.Const _, Drl.Ast.Base "users")) -> ()
   | Ok _    -> assert false
 
-(* ============================================================================
-   DRL Executor end-to-end Tests
-   ============================================================================ *)
-
 let%test_unit "drl: execute Base" =
   with_storage (fun storage ->
     let db = match Manipulation.Memory.create_database storage ~name:"test_db" with
@@ -1352,10 +1308,6 @@ let%test_unit "drl: execute Select+Const" =
           in
           assert (name_v = "Alice"))
 
-(* ============================================================================
-   Constraint System Tests
-   ============================================================================ *)
-
 (* ---------- Constraint construction ---------- *)
 
 let%test_unit "constraint: vars_in MemberOf" =
@@ -1883,7 +1835,7 @@ let%test_unit "constraint scenario: mutual exclusion subtypes" =
       | Error _ -> assert false
       | Ok x -> x
     in
-    (* Now try to insert same id into manager — should fail *)
+    (* Now try to insert same id into manager, should fail *)
     let manager =
       match Manipulation.Memory.get_relation db ~name:"manager" with
       | None -> assert false
@@ -1970,7 +1922,7 @@ let%test_unit "constraint scenario: foreign key" =
       | Error _ -> assert false
       | Ok x -> x
     in
-    (* Insert item referencing valid order — should succeed *)
+    (* Insert item referencing valid order, should succeed *)
     let items =
       match Manipulation.Memory.get_relation db ~name:"order_items" with
       | None -> assert false
@@ -1994,7 +1946,7 @@ let%test_unit "constraint scenario: foreign key" =
       | Error _ -> assert false
       | Ok x -> x
     in
-    (* Insert item referencing nonexistent order — should fail *)
+    (* Insert item referencing nonexistent order, should fail *)
     let invalid_item : Tuple.materialized =
       {
         relation = "order_items";
@@ -2074,7 +2026,7 @@ let%test_unit "constraint scenario: self-reference neq" =
       | None -> assert false
       | Some r -> r
     in
-    (* Insert (emp_id=1, mgr_id=2) — should succeed *)
+    (* Insert (emp_id=1, mgr_id=2), should succeed *)
     let valid : Tuple.materialized =
       {
         relation = "reports_to";
@@ -2091,7 +2043,7 @@ let%test_unit "constraint scenario: self-reference neq" =
       | Error _ -> assert false
       | Ok x -> x
     in
-    (* Insert (emp_id=3, mgr_id=3) — same person, should fail *)
+    (* Insert (emp_id=3, mgr_id=3): same person, should fail *)
     let invalid : Tuple.materialized =
       {
         relation = "reports_to";
@@ -2173,7 +2125,7 @@ let%test_unit "constraint scenario: open vs closed ticket" =
       | Error _ -> assert false
       | Ok x -> x
     in
-    (* Try to open same ticket — should fail *)
+    (* Try to open same ticket, should fail *)
     let open_ =
       match Manipulation.Memory.get_relation db ~name:"open_ticket" with
       | None -> assert false
@@ -2256,7 +2208,7 @@ let%test_unit "constraint scenario: weak entity dependency" =
       | Error _ -> assert false
       | Ok x -> x
     in
-    (* Insert dependent with valid parent — succeeds *)
+    (* Insert dependent with valid parent, should succeed *)
     let dep =
       match Manipulation.Memory.get_relation db ~name:"dependent" with
       | None -> assert false
@@ -2280,7 +2232,7 @@ let%test_unit "constraint scenario: weak entity dependency" =
       | Error _ -> assert false
       | Ok x -> x
     in
-    (* Insert dependent with nonexistent parent — fails *)
+    (* Insert dependent with nonexistent parent, should fail *)
     let invalid_dep : Tuple.materialized =
       {
         relation = "dependent";
@@ -2298,7 +2250,7 @@ let%test_unit "constraint scenario: weak entity dependency" =
     | Error (Manipulation.ConstraintViolation _) -> ()
     | _ -> assert false)
 
-(* Scenario 6: Algebra propagation — select preserves constraints *)
+(* Scenario 6: Algebra propagation, select preserves constraints *)
 let%test_unit "constraint propagation: select preserves constraints" =
   with_storage (fun storage ->
     let db =
@@ -2336,7 +2288,7 @@ let%test_unit "constraint propagation: select preserves constraints" =
     | Ok result ->
       assert (result.Relation.constraints <> None))
 
-(* Scenario 7: Algebra propagation — project filters constraints *)
+(* Scenario 7: Algebra propagation, project filters constraints *)
 let%test_unit "constraint propagation: project filters constraints" =
   with_storage (fun storage ->
     let db =
@@ -2376,19 +2328,15 @@ let%test_unit "constraint propagation: project filters constraints" =
       | None -> assert false
       | Some r -> r
     in
-    (* Project to only "x" — constraint should survive *)
+    (* Project to only "x", constraint should survive *)
     (match Algebra.Memory.project storage [ "x" ] rel with
      | Error _ -> assert false
      | Ok result -> assert (result.Relation.constraints <> None));
-    (* Project to only "y" — constraint referencing "x" should be dropped *)
+    (* Project to only "y", constraint referencing "x" should be dropped *)
     match Algebra.Memory.project storage [ "y" ] rel with
     | Error _ -> assert false
     | Ok result -> assert (result.Relation.constraints = None))
 
-(* ============================================================================
-   DML Sublanguage Tests
-   ============================================================================ *)
-
 (* Parse tests use canonical flat sexp format: (Constructor (field val) ...) *)
 
 let%test_unit "ddl: parse CreateDatabase" =
@@ -2442,7 +2390,7 @@ let%test_unit "ddl: round-trip RegisterDomain" =
   | Error _ -> assert false
   | Ok parsed -> assert (parsed = src)
 
-(* Executor tests construct AST directly — no dependency on sexp format *)
+(* Executor tests construct AST directly, no dependency on sexp format *)
 
 let%test_unit "ddl: execute CreateDatabase" =
   with_storage (fun storage ->
@@ -2578,10 +2526,6 @@ let%test_unit "dml: insert into nonexistent relation returns error" =
     | Error (Dml.Executor.Memory.RelationNotFound "ghost") -> ()
     | _ -> assert false)
 
-(* ============================================================================
-   DCL Sublanguage Tests
-   ============================================================================ *)
-
 let%test_unit "dcl: round-trip RegisterConstraint MemberOf" =
   let src = Icl.Ast.RegisterConstraint {
     constraint_name = "fk_order";
@@ -2685,10 +2629,6 @@ let%test_unit "dcl: FK constraint enforced on insert" =
     | Error (Dml.Executor.Memory.ManipulationError (Manipulation.ConstraintViolation _)) -> ()
     | _ -> assert false)
 
-(* ============================================================================
-   Branch Tests
-   ============================================================================ *)
-
 module BranchMemory = Management.Branch.Make(Management.Physical.Memory)
 
 let%test_unit "branch: create and get_tip" =
@@ -2760,10 +2700,6 @@ let%test_unit "branch: multiple branches are independent" =
     assert (feat_tip = "hash-feature");
     assert (main_tip <> feat_tip))
 
-(* ============================================================================
-   Diff Tests
-   ============================================================================ *)
-
 let%test_unit "diff: identical databases produce empty diff" =
   with_storage (fun storage ->
     let db = match Manipulation.Memory.create_database storage ~name:"db" with
@@ -2836,10 +2772,6 @@ let%test_unit "diff: modified relation detected with added tuple" =
       assert (removed_tuples = [])
     | _ -> assert false)
 
-(* ============================================================================
-   Merge Tests
-   ============================================================================ *)
-
 module MergeMemory = Management.Merge.Make
   (Management.Physical.Memory)
   (Manipulation.Memory)
@@ -2930,3 +2862,746 @@ let%test_unit "merge: no-op when both sides are identical" =
     | Ok (Management.Merge.Clean merged) ->
       assert (Management.Database.get_relation_names merged =
               Management.Database.get_relation_names base_db))
+
+(* Helper: look up polarity for a relation name in the result list *)
+let find_polarity name pols =
+  List.assoc_opt name pols
+
+let%test_unit "polarity: MemberOf target is Positive" =
+  let c = Constraint.MemberOf {
+    target = "R";
+    binding = Constraint.BindingMap.empty;
+  } in
+  assert (find_polarity "R" (Constraint.polarity_of c) = Some Constraint.Positive)
+
+let%test_unit "polarity: Not MemberOf flips to Negative" =
+  let c = Constraint.Not {
+    body = Constraint.MemberOf { target = "R"; binding = Constraint.BindingMap.empty };
+    universe = "U";
+  } in
+  assert (find_polarity "R" (Constraint.polarity_of c) = Some Constraint.Negative)
+
+let%test_unit "polarity: Exists quantifier is Positive" =
+  let c = Constraint.Exists {
+    variable = "x";
+    quantifier = "Q";
+    body = Constraint.MemberOf { target = "Q"; binding = Constraint.BindingMap.empty };
+  } in
+  assert (find_polarity "Q" (Constraint.polarity_of c) = Some Constraint.Positive)
+
+let%test_unit "polarity: Forall quantifier is Negative" =
+  let c = Constraint.Forall {
+    variable = "x";
+    quantifier = "Q";
+    body = Constraint.MemberOf { target = "T"; binding = Constraint.BindingMap.empty };
+  } in
+  assert (find_polarity "Q" (Constraint.polarity_of c) = Some Constraint.Negative);
+  assert (find_polarity "T" (Constraint.polarity_of c) = Some Constraint.Positive)
+
+let%test_unit "polarity: same relation with both polarities merges to Both" =
+  (* Not(MemberOf R) AND MemberOf R => R appears Negative and Positive => Both *)
+  let c = Constraint.And [
+    Constraint.Not {
+      body = Constraint.MemberOf { target = "R"; binding = Constraint.BindingMap.empty };
+      universe = "U";
+    };
+    Constraint.MemberOf { target = "R"; binding = Constraint.BindingMap.empty };
+  ] in
+  assert (find_polarity "R" (Constraint.polarity_of c) = Some Constraint.Both)
+
+let%test_unit "polarity: Forall body MemberOf keeps Positive" =
+  (* Forall x in Q, MemberOf(T): T still has positive polarity *)
+  let c = Constraint.Forall {
+    variable = "x";
+    quantifier = "Q";
+    body = Constraint.MemberOf { target = "T"; binding = Constraint.BindingMap.empty };
+  } in
+  assert (find_polarity "T" (Constraint.polarity_of c) = Some Constraint.Positive)
+
+let%test_unit "polarity: nested Not double-negation restores Positive" =
+  let c = Constraint.Not {
+    body = Constraint.Not {
+      body = Constraint.MemberOf { target = "R"; binding = Constraint.BindingMap.empty };
+      universe = "U";
+    };
+    universe = "V";
+  } in
+  assert (find_polarity "R" (Constraint.polarity_of c) = Some Constraint.Positive)
+
+let%test_unit "polarity: unrelated relation absent from result" =
+  let c = Constraint.MemberOf {
+    target = "R";
+    binding = Constraint.BindingMap.empty;
+  } in
+  assert (find_polarity "S" (Constraint.polarity_of c) = None)
+
+let abs i = Obj.magic (i : int)  (* lift int to AbstractValue for tests *)
+
+let%test_unit "focused_filter: Var binding maps deleted value to constrained attr" =
+  (* Constraint: MemberOf Dept (dept_id = Var "dept_id")
+     Deleted:    Dept { dept_id = 99 }
+     Expected filter: [("dept_id", 99)]
+
+     This is the core FK pattern. The Var "dept_id" in the binding says:
+     "the constrained tuple's dept_id must equal whatever the Dept tuple's
+     dept_id is." When we delete Dept{dept_id=99}, focused_filter resolves
+     the Var to the deleted value and returns [("dept_id", 99)].
+
+     The cascade checker uses this to scan only Employee tuples where
+     dept_id=99 rather than re-checking every Employee row. *)
+  let binding =
+    Constraint.BindingMap.singleton "dept_id" (Constraint.Var "dept_id")
+  in
+  let c = Constraint.MemberOf { target = "Dept"; binding } in
+  let deleted = [("dept_id", abs 99)] in
+  let filter = Constraint.focused_filter c "Dept" deleted in
+  assert (List.length filter = 1);
+  assert (fst (List.hd filter) = "dept_id");
+  assert (Stdlib.(=) (snd (List.hd filter)) (abs 99))
+
+let%test_unit "focused_filter: Const binding is ignored (no var link)" =
+  (* Constraint: MemberOf Dept (code = Const "eng")
+     Deleted:    Dept { code = "eng" }
+     Expected filter: []
+
+     A Const binding is a fixed literal that restricts which rows the constraint
+     *applies to*: it is not a join condition. There is no Var, so no link
+     exists between the deleted tuple's attributes and the attributes of any
+     constrained relation. focused_filter correctly returns [] here.
+
+     [] does NOT mean "nothing is affected." It means "no tighter filter could
+     be derived": the caller must fall back to re-checking every tuple in the
+     constrained relation. The tempting but wrong reading would be to see
+     that the deleted value "eng" matches Const "eng" and emit [("code","eng")].
+     Doing so would conflate a constraint literal with a join variable. *)
+  let binding =
+    Constraint.BindingMap.singleton "code"
+      (Constraint.Const (Obj.magic ("eng" : string)))
+  in
+  let c = Constraint.MemberOf { target = "Dept"; binding } in
+  let filter = Constraint.focused_filter c "Dept" [("code", Obj.magic ("eng" : string))] in
+  assert (filter = [])
+
+let%test_unit "focused_filter: Exists body MemberOf same relation is followed" =
+  (* Constraint on Employee: Exists d in Dept, MemberOf Dept (dept_id = Var "dept_id")
+     Deleted: Dept { dept_id = 7 }
+     Expected filter on Employee: [("dept_id", 7)]
+
+     Real FK constraints are wrapped in Exists: "there exists a dept d such
+     that this employee's dept_id matches d's dept_id." The dep_rel ("Dept")
+     appears as the Exists quantifier, not directly in a top-level MemberOf.
+
+     focused_filter must recognise this pattern and recurse into the Exists
+     body to find the Var binding. If it stopped at the Exists node and
+     returned [], the cascade checker would lose the narrowing and fall back
+     to scanning every Employee, defeating the optimisation entirely. *)
+  let binding =
+    Constraint.BindingMap.singleton "dept_id" (Constraint.Var "dept_id")
+  in
+  let c = Constraint.Exists {
+    variable = "d";
+    quantifier = "Dept";
+    body = Constraint.MemberOf { target = "Dept"; binding };
+  } in
+  let filter = Constraint.focused_filter c "Dept" [("dept_id", abs 7)] in
+  assert (List.length filter = 1);
+  assert (fst (List.hd filter) = "dept_id")
+
+let%test_unit "focused_filter: unrelated dep_rel yields empty filter" =
+  (* Constraint references relation R. We are deleting from relation S.
+     Expected filter: []
+
+     If the deleted relation does not appear anywhere in the constraint,
+     the deletion cannot possibly affect any constrained tuple since no join
+     variable links them. The cascade checker uses [] as a signal to skip
+     the re-check for this constraint entirely, not to re-check everything. *)
+  let binding =
+    Constraint.BindingMap.singleton "x" (Constraint.Var "x")
+  in
+  let c = Constraint.MemberOf { target = "R"; binding } in
+  let filter = Constraint.focused_filter c "S" [("x", abs 1)] in
+  assert (filter = [])
+
+let%test_unit "trigger_constants: Const value in binding is extracted" =
+  (* Constraint: MemberOf R (status = Const "active")
+     Expected constants: [("status", "active")]
+
+     trigger_constants answers: "for a DELETE from dep_rel to even risk
+     violating this constraint, what attribute values must the deleted tuple
+     have?" A Const binding says the constraint only applies when the dep_rel
+     tuple has that exact value. If the deleted tuple's status ≠ "active",
+     this constraint can never be violated by that deletion, so it is skipped.
+
+     The cascade checker calls trigger_constants before doing any tuple scan
+     and bails out early if the deleted tuple doesn't match. *)
+  let binding =
+    Constraint.BindingMap.singleton "status"
+      (Constraint.Const (Obj.magic ("active" : string)))
+  in
+  let c = Constraint.MemberOf { target = "R"; binding } in
+  let consts = Constraint.trigger_constants c "R" in
+  assert (List.length consts = 1);
+  assert (fst (List.hd consts) = "status")
+
+let%test_unit "trigger_constants: Var binding produces no constant" =
+  (* Constraint: MemberOf R (id = Var "id")
+     Expected constants: []
+
+     A Var binding imposes no fixed-value precondition on the deleted tuple.
+     It is a join variable, not a filter. Any deleted tuple from R could
+     match a constrained tuple via the Var link, so no early-exit is possible.
+     [] means: always proceed to the cascade scan. *)
+  let binding =
+    Constraint.BindingMap.singleton "id" (Constraint.Var "id")
+  in
+  let c = Constraint.MemberOf { target = "R"; binding } in
+  let consts = Constraint.trigger_constants c "R" in
+  assert (consts = [])
+
+let%test_unit "trigger_constants: unrelated dep_rel yields empty" =
+  (* Constraint references relation R. We are deleting from relation S.
+     Expected constants: []
+
+     When dep_rel does not appear in the constraint at all, trigger_constants
+     returns [], meaning no precondition was found so the cascade scan proceeds. However,
+     focused_filter will also return [] for an unrelated dep_rel, causing the
+     cascade checker to skip the re-check for a different reason (no join
+     link). Both [] cases ultimately mean no affected tuples are found. *)
+  let binding =
+    Constraint.BindingMap.singleton "x" (Constraint.Const (abs 1))
+  in
+  let c = Constraint.MemberOf { target = "R"; binding } in
+  let consts = Constraint.trigger_constants c "S" in
+  assert (consts = [])
+
+let%test_unit "substitute_transition: Var \"variable.attr\" is replaced by Const from transition tuple" =
+  (* Constraint: Exists d in Department, MemberOf Target (key = Var "d.dept_id")
+     Transition: Department row { dept_id = 99 }
+     After substitution: Exists d in Department, MemberOf Target (key = Const 99)
+     The namespaced Var "d.dept_id" (quantifier-scoped) is replaced;
+     a base-tuple Var "dept_id" would not be. *)
+  let c =
+    Constraint.Exists {
+      variable = "d"; quantifier = "Department";
+      body = Constraint.MemberOf {
+        target = "Target";
+        binding = Constraint.BindingMap.singleton "key" (Constraint.Var "d.dept_id");
+      };
+    }
+  in
+  let result = Constraint.substitute_transition c "Department" [("dept_id", abs 99)] in
+  (match result with
+   | Constraint.Exists { body = Constraint.MemberOf { binding; _ }; _ } ->
+     (match Constraint.BindingMap.find_opt "key" binding with
+      | Some (Constraint.Const v) -> assert (Stdlib.(=) v (abs 99))
+      | _ -> assert false)
+   | _ -> assert false)
+
+let%test_unit "substitute_transition: base-tuple Var is not substituted" =
+  (* Constraint: Exists d in Department, MemberOf Target (key = Var "dept_id")
+     Var "dept_id" is a base-tuple reference (no "d." prefix), so it must
+     survive substitution unchanged. Only "d.dept_id" would be replaced. *)
+  let c =
+    Constraint.Exists {
+      variable = "d"; quantifier = "Department";
+      body = Constraint.MemberOf {
+        target = "Target";
+        binding = Constraint.BindingMap.singleton "key" (Constraint.Var "dept_id");
+      };
+    }
+  in
+  let result = Constraint.substitute_transition c "Department" [("dept_id", abs 99)] in
+  (match result with
+   | Constraint.Exists { body = Constraint.MemberOf { binding; _ }; _ } ->
+     (match Constraint.BindingMap.find_opt "key" binding with
+      | Some (Constraint.Var "dept_id") -> ()
+      | _ -> assert false)
+   | _ -> assert false)
+
+let%test_unit "substitute_transition: non-matching quantifier relation is unchanged" =
+  (* Constraint: Exists d in Department, MemberOf Target (key = Var "d.dept_id")
+     Transition is for relation "Other", not "Department".
+     The body must survive unmodified. *)
+  let c =
+    Constraint.Exists {
+      variable = "d"; quantifier = "Department";
+      body = Constraint.MemberOf {
+        target = "Target";
+        binding = Constraint.BindingMap.singleton "key" (Constraint.Var "d.dept_id");
+      };
+    }
+  in
+  let result = Constraint.substitute_transition c "Other" [("dept_id", abs 99)] in
+  (match result with
+   | Constraint.Exists { body = Constraint.MemberOf { binding; _ }; _ } ->
+     (match Constraint.BindingMap.find_opt "key" binding with
+      | Some (Constraint.Var "d.dept_id") -> ()
+      | _ -> assert false)
+   | _ -> assert false)
+
+let%test_unit "substitute_transition: substitution applies through And and Not" =
+  (* Constraint:
+       Exists d in Department,
+         And [
+           MemberOf A (x = Var "d.x");
+           Not { MemberOf B (y = Var "d.y"); universe = "B" }
+         ]
+     Transition: { x = 1, y = 2 }
+     After substitution: Var "d.x" -> Const 1, Var "d.y" -> Const 2 *)
+  let c =
+    Constraint.Exists {
+      variable = "d"; quantifier = "Department";
+      body = Constraint.And [
+        Constraint.MemberOf {
+          target = "A";
+          binding = Constraint.BindingMap.singleton "x" (Constraint.Var "d.x");
+        };
+        Constraint.Not {
+          universe = "B";
+          body = Constraint.MemberOf {
+            target = "B";
+            binding = Constraint.BindingMap.singleton "y" (Constraint.Var "d.y");
+          };
+        };
+      ];
+    }
+  in
+  let result = Constraint.substitute_transition c "Department" [("x", abs 1); ("y", abs 2)] in
+  (match result with
+   | Constraint.Exists {
+       body = Constraint.And [
+         Constraint.MemberOf { binding = b1; _ };
+         Constraint.Not { body = Constraint.MemberOf { binding = b2; _ }; _ };
+       ]; _
+     } ->
+     (match Constraint.BindingMap.find_opt "x" b1,
+            Constraint.BindingMap.find_opt "y" b2 with
+      | Some (Constraint.Const v1), Some (Constraint.Const v2) ->
+        assert (Stdlib.(=) v1 (abs 1));
+        assert (Stdlib.(=) v2 (abs 2))
+      | _ -> assert false)
+   | _ -> assert false)
+
+(* Build a minimal db with two relations and an FK constraint:
+   Department { dept_id }  <--  Employee { emp_id, dept_id }
+   Constraint on Employee: Exists d in Department, MemberOf Department (dept_id = Var "dept_id") *)
+let setup_fk_db storage =
+  let db = match Manipulation.Memory.create_database storage ~name:"hr" with
+    | Error _ -> assert false | Ok db -> db
+  in
+  let db = match Manipulation.Memory.create_relation storage db ~name:"Department"
+      ~schema:(Schema.empty |> Schema.add "dept_id" "natural") with
+    | Error _ -> assert false | Ok (db, _) -> db
+  in
+  let db = match Manipulation.Memory.create_relation storage db ~name:"Employee"
+      ~schema:(Schema.empty
+               |> Schema.add "emp_id"  "natural"
+               |> Schema.add "dept_id" "natural") with
+    | Error _ -> assert false | Ok (db, _) -> db
+  in
+  let fk_body =
+    let binding =
+      Constraint.BindingMap.singleton "dept_id" (Constraint.Var "dept_id")
+    in
+    Constraint.Exists {
+      variable = "d";
+      quantifier = "Department";
+      body = Constraint.MemberOf { target = "Department"; binding };
+    }
+  in
+  let db = match Manipulation.Memory.register_constraint storage db
+      ~constraint_name:"fk_dept"
+      ~relation_name:"Employee"
+      ~body:fk_body with
+    | Error _ -> assert false | Ok db -> db
+  in
+  db
+
+let%test_unit "extend_tuple namespacing: insert Employee with non-existent dept_id is rejected" =
+  (* Regression test for the extend_tuple namespacing bug.
+     Without namespacing, extend_tuple would overwrite the Employee's dept_id=99
+     with the Department row's dept_id=1 during Exists iteration.  Var "dept_id"
+     would then resolve to 1, check_membership would succeed, and the insertion
+     would be accepted despite dept_id=99 not existing in Department.
+     With namespacing, Department attributes are stored as d.dept_id=1 and the
+     base Employee attribute dept_id=99 is preserved, so the check correctly fails. *)
+  with_storage (fun storage ->
+    let db = setup_fk_db storage in
+    let db = match Dml.Executor.Memory.execute storage db
+        (Dml.Ast.InsertTuple { relation = "Department";
+                               attributes = [("dept_id", Drl.Ast.Int 1)] }) with
+      | Error _ -> assert false | Ok db -> db
+    in
+    match Dml.Executor.Memory.execute storage db
+        (Dml.Ast.InsertTuple { relation = "Employee";
+                               attributes = [("emp_id", Drl.Ast.Int 1);
+                                             ("dept_id", Drl.Ast.Int 99)] }) with
+    | Ok _ -> assert false  (* dept_id=99 does not exist in Department *)
+    | Error _ -> ())
+
+let%test_unit "cascade: delete referenced row violates FK and is rejected" =
+  with_storage (fun storage ->
+    let db = setup_fk_db storage in
+    (* Insert Dept 1 and Employee referencing it *)
+    let db = match Dml.Executor.Memory.execute storage db
+        (Dml.Ast.InsertTuple { relation = "Department";
+                               attributes = [("dept_id", Drl.Ast.Int 1)] }) with
+      | Error _ -> assert false | Ok db -> db
+    in
+    let db = match Dml.Executor.Memory.execute storage db
+        (Dml.Ast.InsertTuple { relation = "Employee";
+                               attributes = [("emp_id", Drl.Ast.Int 10);
+                                             ("dept_id", Drl.Ast.Int 1)] }) with
+      | Error _ -> assert false | Ok db -> db
+    in
+    let dept_rel = match Manipulation.Memory.get_relation db ~name:"Department" with
+      | None -> assert false | Some r -> r
+    in
+    let dept_hash = Hashing.hash_tuple
+        { Tuple.relation = "Department";
+          attributes = Tuple.AttributeMap.singleton "dept_id"
+              { Attribute.value = Obj.magic (1 : int) } }
+    in
+    (* Deleting Dept 1 while Employee references it must fail *)
+    match Manipulation.Memory.retract_tuple storage db dept_rel ~tuple_hash:dept_hash with
+    | Ok _ -> assert false  (* should have been rejected *)
+    | Error (Manipulation.ConstraintViolation msg) ->
+      assert (String.length msg > 0)
+    | Error _ -> assert false)
+
+let%test_unit "cascade: delete unreferenced row succeeds" =
+  with_storage (fun storage ->
+    let db = setup_fk_db storage in
+    (* Insert two departments; employee only references dept 1 *)
+    let db = match Dml.Executor.Memory.execute storage db
+        (Dml.Ast.InsertTuple { relation = "Department";
+                               attributes = [("dept_id", Drl.Ast.Int 1)] }) with
+      | Error _ -> assert false | Ok db -> db
+    in
+    let db = match Dml.Executor.Memory.execute storage db
+        (Dml.Ast.InsertTuple { relation = "Department";
+                               attributes = [("dept_id", Drl.Ast.Int 2)] }) with
+      | Error _ -> assert false | Ok db -> db
+    in
+    let db = match Dml.Executor.Memory.execute storage db
+        (Dml.Ast.InsertTuple { relation = "Employee";
+                               attributes = [("emp_id", Drl.Ast.Int 10);
+                                             ("dept_id", Drl.Ast.Int 1)] }) with
+      | Error _ -> assert false | Ok db -> db
+    in
+    let dept_rel = match Manipulation.Memory.get_relation db ~name:"Department" with
+      | None -> assert false | Some r -> r
+    in
+    let dept2_hash = Hashing.hash_tuple
+        { Tuple.relation = "Department";
+          attributes = Tuple.AttributeMap.singleton "dept_id"
+              { Attribute.value = Obj.magic (2 : int) } }
+    in
+    (* Deleting unreferenced Dept 2 must succeed *)
+    match Manipulation.Memory.retract_tuple storage db dept_rel ~tuple_hash:dept2_hash with
+    | Error _ -> assert false
+    | Ok (new_db, _) ->
+      let dept = match Manipulation.Memory.get_relation new_db ~name:"Department" with
+        | None -> assert false | Some r -> r
+      in
+      assert (Manipulation.Memory.tuple_count dept = 1))
+
+let%test_unit "cascade: Negative-polarity relation deletion is not checked" =
+  (* Constraint on Employee: NOT (MemberOf Blacklist (emp_id = Var "emp_id"))
+     Blacklist has Negative polarity (INSERT into Blacklist could violate).
+     DELETE from Blacklist should never trigger a cascade check. *)
+  with_storage (fun storage ->
+    let db = match Manipulation.Memory.create_database storage ~name:"hr" with
+      | Error _ -> assert false | Ok db -> db
+    in
+    let db = match Manipulation.Memory.create_relation storage db ~name:"Blacklist"
+        ~schema:(Schema.empty |> Schema.add "emp_id" "natural") with
+      | Error _ -> assert false | Ok (db, _) -> db
+    in
+    let db = match Manipulation.Memory.create_relation storage db ~name:"Employee"
+        ~schema:(Schema.empty |> Schema.add "emp_id" "natural") with
+      | Error _ -> assert false | Ok (db, _) -> db
+    in
+    let not_body =
+      let binding =
+        Constraint.BindingMap.singleton "emp_id" (Constraint.Var "emp_id")
+      in
+      Constraint.Not {
+        body = Constraint.MemberOf { target = "Blacklist"; binding };
+        universe = "Employee";
+      }
+    in
+    let db = match Manipulation.Memory.register_constraint storage db
+        ~constraint_name:"not_blacklisted"
+        ~relation_name:"Employee"
+        ~body:not_body with
+      | Error _ -> assert false | Ok db -> db
+    in
+    (* Insert emp_id=5 into Blacklist, then insert emp_id=99 into Employee (unrelated) *)
+    let db = match Dml.Executor.Memory.execute storage db
+        (Dml.Ast.InsertTuple { relation = "Blacklist";
+                               attributes = [("emp_id", Drl.Ast.Int 5)] }) with
+      | Error _ -> assert false | Ok db -> db
+    in
+    let db = match Dml.Executor.Memory.execute storage db
+        (Dml.Ast.InsertTuple { relation = "Employee";
+                               attributes = [("emp_id", Drl.Ast.Int 99)] }) with
+      | Error _ -> assert false | Ok db -> db
+    in
+    let bl_rel = match Manipulation.Memory.get_relation db ~name:"Blacklist" with
+      | None -> assert false | Some r -> r
+    in
+    let bl_hash = Hashing.hash_tuple
+        { Tuple.relation = "Blacklist";
+          attributes = Tuple.AttributeMap.singleton "emp_id"
+              { Attribute.value = Obj.magic (5 : int) } }
+    in
+    (* Deleting from Blacklist (negative polarity) must not trigger cascade check *)
+    match Manipulation.Memory.retract_tuple storage db bl_rel ~tuple_hash:bl_hash with
+    | Error _ -> assert false
+    | Ok _ -> ())
+
+let%test_unit "cascade: deferred constraint not checked during retract_tuple" =
+  with_storage (fun storage ->
+    let db = match Manipulation.Memory.create_database storage ~name:"hr" with
+      | Error _ -> assert false | Ok db -> db
+    in
+    let db = match Manipulation.Memory.create_relation storage db ~name:"Department"
+        ~schema:(Schema.empty |> Schema.add "dept_id" "natural") with
+      | Error _ -> assert false | Ok (db, _) -> db
+    in
+    let db = match Manipulation.Memory.create_relation storage db ~name:"Employee"
+        ~schema:(Schema.empty
+                 |> Schema.add "emp_id"  "natural"
+                 |> Schema.add "dept_id" "natural") with
+      | Error _ -> assert false | Ok (db, _) -> db
+    in
+    let fk_body =
+      let binding =
+        Constraint.BindingMap.singleton "dept_id" (Constraint.Var "dept_id")
+      in
+      Constraint.Exists {
+        variable = "d";
+        quantifier = "Department";
+        body = Constraint.MemberOf { target = "Department"; binding };
+      }
+    in
+    (* Register as DEFERRED *)
+    let db = match Manipulation.Memory.attach_constraint storage db
+        ~constraint_name:"fk_dept_deferred"
+        ~relation_name:"Employee"
+        ~body:fk_body
+        ~timing:Constraint.Deferred with
+      | Error _ -> assert false | Ok db -> db
+    in
+    let db = match Dml.Executor.Memory.execute storage db
+        (Dml.Ast.InsertTuple { relation = "Department";
+                               attributes = [("dept_id", Drl.Ast.Int 1)] }) with
+      | Error _ -> assert false | Ok db -> db
+    in
+    let db = match Dml.Executor.Memory.execute storage db
+        (Dml.Ast.InsertTuple { relation = "Employee";
+                               attributes = [("emp_id", Drl.Ast.Int 10);
+                                             ("dept_id", Drl.Ast.Int 1)] }) with
+      | Error _ -> assert false | Ok db -> db
+    in
+    let dept_rel = match Manipulation.Memory.get_relation db ~name:"Department" with
+      | None -> assert false | Some r -> r
+    in
+    let dept_hash = Hashing.hash_tuple
+        { Tuple.relation = "Department";
+          attributes = Tuple.AttributeMap.singleton "dept_id"
+              { Attribute.value = Obj.magic (1 : int) } }
+    in
+    (* Deferred: retract_tuple itself should NOT reject the deletion *)
+    match Manipulation.Memory.retract_tuple storage db dept_rel ~tuple_hash:dept_hash with
+    | Error _ -> assert false  (* deferred: must pass here *)
+    | Ok (new_db, _) ->
+      (* But check_deferred_constraints must catch the violation *)
+      match Manipulation.Memory.check_deferred_constraints storage new_db with
+      | Ok () -> assert false  (* should have caught the violation *)
+      | Error (Manipulation.ConstraintViolation _) -> ()
+      | Error _ -> assert false)
+
+(* INSERT cascade tests.
+
+   Schema:
+     Blacklist { emp_id }
+     Employee  { emp_id }
+
+   Constraint on Employee (negative polarity on Blacklist):
+     Not { body = MemberOf Blacklist (emp_id = Var "emp_id"); universe = "Employee" }
+
+   Blacklist has Negative polarity in this constraint, so inserting into Blacklist
+   must trigger a re-check of Employee tuples. *)
+let setup_blacklist_db storage =
+  let db = match Manipulation.Memory.create_database storage ~name:"hr" with
+    | Error _ -> assert false | Ok db -> db
+  in
+  let db = match Manipulation.Memory.create_relation storage db ~name:"Blacklist"
+      ~schema:(Schema.empty |> Schema.add "emp_id" "natural") with
+    | Error _ -> assert false | Ok (db, _) -> db
+  in
+  let db = match Manipulation.Memory.create_relation storage db ~name:"Employee"
+      ~schema:(Schema.empty |> Schema.add "emp_id" "natural") with
+    | Error _ -> assert false | Ok (db, _) -> db
+  in
+  let not_body =
+    Constraint.Not {
+      body = Constraint.MemberOf {
+        target = "Blacklist";
+        binding = Constraint.BindingMap.singleton "emp_id" (Constraint.Var "emp_id");
+      };
+      universe = "Employee";
+    }
+  in
+  match Manipulation.Memory.register_constraint storage db
+      ~constraint_name:"not_blacklisted"
+      ~relation_name:"Employee"
+      ~body:not_body with
+  | Error _ -> assert false | Ok db -> db
+
+let%test_unit "cascade INSERT: inserting into Blacklist violates constraint on existing Employee" =
+  (* Employee emp_id=5 already exists. Inserting Blacklist{emp_id=5} must be
+     rejected because the Employee constraint (NOT member of Blacklist) would
+     be violated for the existing emp_id=5 row. *)
+  with_storage (fun storage ->
+    let db = setup_blacklist_db storage in
+    let db = match Dml.Executor.Memory.execute storage db
+        (Dml.Ast.InsertTuple { relation = "Employee";
+                               attributes = [("emp_id", Drl.Ast.Int 5)] }) with
+      | Error _ -> assert false | Ok db -> db
+    in
+    match Dml.Executor.Memory.execute storage db
+        (Dml.Ast.InsertTuple { relation = "Blacklist";
+                               attributes = [("emp_id", Drl.Ast.Int 5)] }) with
+    | Ok _ -> assert false  (* emp_id=5 exists in Employee, must be rejected *)
+    | Error _ -> ())
+
+let%test_unit "cascade INSERT: inserting into Blacklist with no matching Employee succeeds" =
+  (* Employee emp_id=99 does not exist. Inserting Blacklist{emp_id=5} must
+     succeed because no Employee tuple is affected. *)
+  with_storage (fun storage ->
+    let db = setup_blacklist_db storage in
+    let db = match Dml.Executor.Memory.execute storage db
+        (Dml.Ast.InsertTuple { relation = "Employee";
+                               attributes = [("emp_id", Drl.Ast.Int 99)] }) with
+      | Error _ -> assert false | Ok db -> db
+    in
+    match Dml.Executor.Memory.execute storage db
+        (Dml.Ast.InsertTuple { relation = "Blacklist";
+                               attributes = [("emp_id", Drl.Ast.Int 5)] }) with
+    | Error _ -> assert false  (* emp_id=5 not in Employee, must succeed *)
+    | Ok _ -> ())
+
+(* commit tests.
+
+   Uses the same Department/Employee FK schema with a DEFERRED constraint.
+   Individual mutations succeed even when they temporarily violate the constraint;
+   commit is the boundary where deferred violations are caught. *)
+let setup_deferred_fk_db storage =
+  let db = match Manipulation.Memory.create_database storage ~name:"hr" with
+    | Error _ -> assert false | Ok db -> db
+  in
+  let db = match Manipulation.Memory.create_relation storage db ~name:"Department"
+      ~schema:(Schema.empty |> Schema.add "dept_id" "natural") with
+    | Error _ -> assert false | Ok (db, _) -> db
+  in
+  let db = match Manipulation.Memory.create_relation storage db ~name:"Employee"
+      ~schema:(Schema.empty
+               |> Schema.add "emp_id"  "natural"
+               |> Schema.add "dept_id" "natural") with
+    | Error _ -> assert false | Ok (db, _) -> db
+  in
+  let fk_body =
+    Constraint.Exists {
+      variable = "d"; quantifier = "Department";
+      body = Constraint.MemberOf {
+        target = "Department";
+        binding = Constraint.BindingMap.singleton "dept_id" (Constraint.Var "dept_id");
+      };
+    }
+  in
+  match Manipulation.Memory.attach_constraint storage db
+      ~constraint_name:"fk_dept"
+      ~relation_name:"Employee"
+      ~body:fk_body
+      ~timing:Constraint.Deferred with
+  | Error _ -> assert false | Ok db -> db
+
+let%test_unit "commit: no deferred constraints, returns Ok with db unchanged" =
+  with_storage (fun storage ->
+    let db = setup_deferred_fk_db storage in
+    let db = match Dml.Executor.Memory.execute storage db
+        (Dml.Ast.InsertTuple { relation = "Department";
+                               attributes = [("dept_id", Drl.Ast.Int 1)] }) with
+      | Error _ -> assert false | Ok db -> db
+    in
+    let db = match Dml.Executor.Memory.execute storage db
+        (Dml.Ast.InsertTuple { relation = "Employee";
+                               attributes = [("emp_id", Drl.Ast.Int 10);
+                                             ("dept_id", Drl.Ast.Int 1)] }) with
+      | Error _ -> assert false | Ok db -> db
+    in
+    (* All constraints satisfied, commit must pass *)
+    match Manipulation.Memory.commit storage db with
+    | Error _ -> assert false
+    | Ok _ -> ())
+
+let%test_unit "commit: deferred violation is caught at commit boundary" =
+  with_storage (fun storage ->
+    let db = setup_deferred_fk_db storage in
+    let db = match Dml.Executor.Memory.execute storage db
+        (Dml.Ast.InsertTuple { relation = "Department";
+                               attributes = [("dept_id", Drl.Ast.Int 1)] }) with
+      | Error _ -> assert false | Ok db -> db
+    in
+    let db = match Dml.Executor.Memory.execute storage db
+        (Dml.Ast.InsertTuple { relation = "Employee";
+                               attributes = [("emp_id", Drl.Ast.Int 10);
+                                             ("dept_id", Drl.Ast.Int 1)] }) with
+      | Error _ -> assert false | Ok db -> db
+    in
+    let dept_rel = match Manipulation.Memory.get_relation db ~name:"Department" with
+      | None -> assert false | Some r -> r
+    in
+    let dept_hash = Hashing.hash_tuple
+        { Tuple.relation = "Department";
+          attributes = Tuple.AttributeMap.singleton "dept_id"
+              { Attribute.value = Obj.magic (1 : int) } }
+    in
+    (* Delete the referenced department. Deferred, so retract succeeds *)
+    let db = match Manipulation.Memory.retract_tuple storage db dept_rel
+        ~tuple_hash:dept_hash with
+      | Error _ -> assert false | Ok (db, _) -> db
+    in
+    (* commit must now catch the orphaned Employee *)
+    match Manipulation.Memory.commit storage db with
+    | Ok _ -> assert false
+    | Error (Manipulation.ConstraintViolation _) -> ()
+    | Error _ -> assert false)
+
+let%test_unit "commit: clears deferred list, second commit passes" =
+  (* After a successful commit the deferred list is empty.
+     A follow-up mutation + commit on the returned db must also work correctly
+     rather than re-running stale entries from the previous window. *)
+  with_storage (fun storage ->
+    let db = setup_deferred_fk_db storage in
+    let db = match Dml.Executor.Memory.execute storage db
+        (Dml.Ast.InsertTuple { relation = "Department";
+                               attributes = [("dept_id", Drl.Ast.Int 1)] }) with
+      | Error _ -> assert false | Ok db -> db
+    in
+    let db = match Dml.Executor.Memory.execute storage db
+        (Dml.Ast.InsertTuple { relation = "Employee";
+                               attributes = [("emp_id", Drl.Ast.Int 10);
+                                             ("dept_id", Drl.Ast.Int 1)] }) with
+      | Error _ -> assert false | Ok db -> db
+    in
+    let db = match Manipulation.Memory.commit storage db with
+      | Error _ -> assert false | Ok db -> db
+    in
+    (* Second commit with no new mutations must also pass *)
+    match Manipulation.Memory.commit storage db with
+    | Error _ -> assert false
+    | Ok _ -> ())