@@ -154,6 +154,20 @@ def stroke_rule(light_hex, dark_hex, opacity=None):
     return "<style>\n@media (prefers-color-scheme: dark) {\n" + body + "\n}\n</style>"
 
 
+def _seed_restriction(table_expr):
+    """
+    Return a traversal seed's restriction and the attributes it names.
+
+    A ``dj.Top`` restriction is an ORDER BY ... LIMIT, not part of the WHERE
+    clause, so a seed restricted by ``Top`` is replaced by the primary keys
+    it selects. Otherwise the traversal would start from every row that
+    matches the seed's other conditions.
+    """
+    if table_expr._top is not None:
+        return AndList([table_expr.keys()]), set(table_expr.primary_key)
+    return AndList(table_expr.restriction), set(table_expr.restriction_attributes)
+
+
 class Diagram(nx.MultiDiGraph):  # noqa: C901
     """
     Schema diagram as a directed acyclic graph (DAG).
@@ -210,9 +224,12 @@ def __init__(self, source, context=None) -> None:
             self._expanded_nodes = set(source._expanded_nodes)
             self.context = source.context
             self._connection = source._connection
-            self._cascade_restrictions = copy.deepcopy(source._cascade_restrictions)
-            self._restrict_conditions = copy.deepcopy(source._restrict_conditions)
-            self._restriction_attrs = copy.deepcopy(source._restriction_attrs)
+            # Copy each node's restriction list, not its items: projected
+            # restrictions are query expressions bound to a live connection,
+            # which cannot be deep-copied.
+            self._cascade_restrictions = {k: copy.copy(v) for k, v in source._cascade_restrictions.items()}
+            self._restrict_conditions = {k: copy.copy(v) for k, v in source._restrict_conditions.items()}
+            self._restriction_attrs = {k: set(v) for k, v in source._restriction_attrs.items()}
             super().__init__(source)
             return
 
@@ -471,9 +488,9 @@ def cascade(cls, table_expr, part_integrity="enforce"):
         result._expanded_nodes = set(descendants)
 
         # Seed restriction
-        restriction = AndList(table_expr.restriction)
+        restriction, restriction_attrs = _seed_restriction(table_expr)
         result._cascade_restrictions[node] = [restriction] if restriction else []
-        result._restriction_attrs[node] = set(table_expr.restriction_attributes)
+        result._restriction_attrs[node] = restriction_attrs
 
         # Propagate downstream
         result._propagate_restrictions(node, mode="cascade", part_integrity=part_integrity)
@@ -557,9 +574,9 @@ def trace(cls, table_expr):
         result._expanded_nodes = set(ancestors)
 
         # Seed restriction
-        restriction = AndList(table_expr.restriction)
+        restriction, restriction_attrs = _seed_restriction(table_expr)
         result._cascade_restrictions[node] = [restriction] if restriction else []
-        result._restriction_attrs[node] = set(table_expr.restriction_attributes)
+        result._restriction_attrs[node] = restriction_attrs
 
         # Propagate upstream
         result._propagate_restrictions_upstream(node)
@@ -883,16 +900,22 @@ def _apply_propagation_rule(
 
         Rules (from table.py restriction propagation):
 
-        1. Non-aliased AND parent restriction attrs ⊆ child PK:
+        1. Non-aliased AND parent restriction attrs ⊆ FK attrs ⊆ child PK:
            Copy parent restriction directly.
         2. Aliased FK (attr_map renames columns):
            ``parent.proj(**{fk: pk for fk, pk in attr_map.items()})``
-        3. Non-aliased AND parent restriction attrs ⊄ child PK:
-           ``parent.proj()``
+        3. Otherwise (non-aliased): ``parent.proj()``
+
+        Copying is exact only when the restriction names attributes that this
+        foreign key carries and the foreign key lies in the child's primary
+        key. A restriction on an attribute that the child inherits through
+        another foreign key would select child rows by that attribute instead
+        of by the parent rows they reference. A secondary foreign key may be
+        NULL, and then the child row references no parent row at all.
         """
         child_pk = self.nodes[child_node].get("primary_key", set())
 
-        if not aliased and parent_attrs and parent_attrs <= child_pk:
+        if not aliased and parent_attrs and parent_attrs <= set(attr_map) <= child_pk:
             # Rule 1: copy parent restriction directly
             parent_restr = restrictions.get(
                 parent_ft.full_table_name,
@@ -912,7 +935,7 @@ def _apply_propagation_rule(
                 restrictions.setdefault(child_node, AndList()).append(child_item)
             child_attrs = set(attr_map.keys())
         else:
-            # Rule 3: non-aliased, restriction attrs ⊄ child PK — project
+            # Rule 3: non-aliased, not a pure copy — project
             child_item = parent_ft.proj()
             if mode == "cascade":
                 restrictions.setdefault(child_node, []).append(child_item)
@@ -922,7 +945,9 @@ def _apply_propagation_rule(
 
         self._restriction_attrs.setdefault(child_node, set()).update(child_attrs)
 
-    def _apply_propagation_rule_upward(self, child_ft, child_attrs, parent_node, attr_map, aliased, mode, restrictions):
+    def _apply_propagation_rule_upward(
+        self, child_ft, child_attrs, parent_node, attr_map, aliased, mode, restrictions, copy_allowed=True
+    ):
         """
         Apply the symmetric (upward) propagation rule to a parent←child edge.
 
@@ -947,10 +972,15 @@ def _apply_propagation_rule_upward(self, child_ft, child_attrs, parent_node, att
            ``child.proj(*attr_map.keys())`` — project child onto its FK columns
            (which, being non-aliased, share names with parent's PK columns) so
            the subsequent restriction on the parent joins on the right columns.
+
+        ``copy_allowed=False`` disables Rule 1, so the parent is restricted to
+        the rows that the restricted child rows actually reference. Copying
+        the child's condition would also select parent rows that no child row
+        references.
         """
         parent_pk = self.nodes[parent_node].get("primary_key", set())
 
-        if not aliased and child_attrs and child_attrs <= parent_pk:
+        if copy_allowed and not aliased and child_attrs and child_attrs <= parent_pk:
             # Backward Rule 1: copy child restriction directly
             child_restr = restrictions.get(
                 child_ft.full_table_name,
@@ -1044,6 +1074,17 @@ def _propagate_part_to_master(self, part_node, master_name, mode, restrictions):
         if len(real_path) < 2 or real_path[-1] != part_node or real_path[0] != master_name:
             return False
 
+        # The walk restricts the intermediate parts only to carry the
+        # restriction up to the master. Those restrictions reference the
+        # descendant part, which is deleted first, so they are restored once
+        # the master is materialized. The master then cascades back down to
+        # all of its parts.
+        intermediates = real_path[1:-1]
+        saved = {
+            node: (copy.copy(restrictions[node]) if node in restrictions else None, self._restriction_attrs.get(node))
+            for node in intermediates
+        }
+
         # Walk real_path in reverse (child → parent direction). For each
         # adjacent (parent, child) pair, look up the FK edge props.
         any_propagated = False
@@ -1059,6 +1100,8 @@ def _propagate_part_to_master(self, part_node, master_name, mode, restrictions):
             child_ft = self._restricted_table(child)
             child_attrs = self._restriction_attrs.get(child, set())
 
+            # The lift is existential: a master joins the cascade only if one
+            # of its part rows did, so the part's condition is never copied.
             self._apply_propagation_rule_upward(
                 child_ft,
                 child_attrs,
@@ -1067,6 +1110,7 @@ def _propagate_part_to_master(self, part_node, master_name, mode, restrictions):
                 aliased,
                 mode,
                 restrictions,
+                copy_allowed=False,
             )
             any_propagated = True
 
@@ -1094,6 +1138,16 @@ def _propagate_part_to_master(self, part_node, master_name, mode, restrictions):
                     restrictions[master_name] = [False]
                 self._restriction_attrs.setdefault(master_name, set())
 
+        for node, (restriction, attrs) in saved.items():
+            if restriction is None:
+                restrictions.pop(node, None)
+            else:
+                restrictions[node] = restriction
+            if attrs is None:
+                self._restriction_attrs.pop(node, None)
+            else:
+                self._restriction_attrs[node] = attrs
+
         return any_propagated
 
     def _edge_props(self, parent, child):
 
@@ -1,4 +1,5 @@
 import collections
+import copy
 import csv
 import inspect
 import itertools
@@ -1083,6 +1084,12 @@ def delete_quick(self, get_count=False):
         Deletes the table without cascading and without user prompt.
         If this table has populated dependent tables, this will fail.
         """
+        if self._top is not None:
+            # dj.Top is an ORDER BY ... LIMIT, not part of the WHERE clause:
+            # delete exactly the rows it selects.
+            target = copy.copy(self)
+            target._top = None
+            return (target & self.keys()).delete_quick(get_count=get_count)
         query = "DELETE FROM " + self.full_table_name + self.where_clause()
         cursor = self.connection.query(query)
         # Use cursor.rowcount (DB-API 2.0 standard, works for both MySQL and PostgreSQL)
 
@@ -0,0 +1,194 @@
+"""
+Regression tests for cascade restrictions that reached rows they should not.
+
+A restriction may be copied verbatim across a foreign key only when it names
+attributes the foreign key carries and the foreign key lies in the child's
+primary key. The part-to-master lift never copies: a master joins the cascade
+only if one of its part rows did. A ``Top`` seed restricts the seed to the keys
+it selects.
+"""
+
+import pytest
+
+import datajoint as dj
+
+
+@pytest.fixture(scope="function")
+def schema_by_backend(connection_by_backend, db_creds_by_backend, request):
+    """Create a fresh schema per test (mirrors test_cascade_integrity.py)."""
+    backend = db_creds_by_backend["backend"]
+    import time
+
+    test_id = str(int(time.time() * 1000))[-8:]
+    schema_name = f"djtest_overdelete_{backend}_{test_id}"[:64]
+
+    if connection_by_backend.is_connected:
+        try:
+            connection_by_backend.query(
+                f"DROP DATABASE IF EXISTS {connection_by_backend.adapter.quote_identifier(schema_name)}"
+            )
+        except Exception:
+            pass
+
+    schema = dj.Schema(schema_name, connection=connection_by_backend)
+    yield schema
+
+    if connection_by_backend.is_connected:
+        try:
+            connection_by_backend.query(
+                f"DROP DATABASE IF EXISTS {connection_by_backend.adapter.quote_identifier(schema_name)}"
+            )
+        except Exception:
+            pass
+
+
+def test_top_seed_deletes_only_the_top_rows(schema_by_backend):
+    """``(T & dj.Top(n)).delete()`` and ``delete_quick()`` delete n rows, not
+    every row of T."""
+
+    @schema_by_backend
+    class Item(dj.Manual):
+        definition = """
+        item_id : int32
+        """
+
+    @schema_by_backend
+    class Child(dj.Manual):
+        definition = """
+        -> Item
+        child_id : int32
+        """
+
+    Item.insert([(i,) for i in range(5)])
+    Child.insert([(i, 0) for i in range(5)])
+
+    seed = Item & dj.Top(2, order_by="item_id")
+    counts = dj.Diagram.cascade(seed).counts()
+    assert counts[Item.full_table_name] == 2
+    assert counts[Child.full_table_name] == 2
+
+    seed.delete(prompt=False)
+    assert sorted(Item.to_arrays("item_id")) == [2, 3, 4]
+    assert sorted(Child.to_arrays("item_id")) == [2, 3, 4]
+
+    (Child & dj.Top(1, order_by="item_id")).delete_quick()
+    assert sorted(Child.to_arrays("item_id")) == [3, 4]
+
+
+def test_nullable_fk_sharing_a_primary_key_column(schema_by_backend):
+    """A child whose nullable foreign key shares a primary-key column with the
+    parent keeps the rows whose foreign key is NULL: they reference nothing."""
+
+    @schema_by_backend
+    class Subject(dj.Manual):
+        definition = """
+        subject_id : int32
+        """
+
+    @schema_by_backend
+    class Surgery(dj.Manual):
+        definition = """
+        -> Subject
+        surgery_id : int32
+        """
+
+    @schema_by_backend
+    class Session(dj.Manual):
+        definition = """
+        -> Subject
+        session_id : int32
+        ---
+        -> [nullable] Surgery
+        """
+
+    Subject.insert([(1,)])
+    Surgery.insert([(1, 1)])
+    Session.insert(
+        [
+            {"subject_id": 1, "session_id": 1, "surgery_id": 1},
+            {"subject_id": 1, "session_id": 2, "surgery_id": None},
+        ]
+    )
+
+    seed = Surgery & {"subject_id": 1}
+    assert dj.Diagram.cascade(seed).counts()[Session.full_table_name] == 1
+
+    seed.delete(prompt=False)
+    assert len(Surgery()) == 0
+    assert Session.to_dicts() == [{"subject_id": 1, "session_id": 2, "surgery_id": None}]
+
+
+def test_diamond_child_is_reached_through_its_secondary_fk(schema_by_backend):
+    """D references B directly (primary) and C (secondary). Deleting B=1 must
+    delete the D rows that reference the deleted C rows, whatever D's own b_id."""
+
+    @schema_by_backend
+    class B(dj.Manual):
+        definition = """
+        b_id : int32
+        """
+
+    @schema_by_backend
+    class C(dj.Manual):
+        definition = """
+        c_id : int32
+        ---
+        -> B
+        """
+
+    @schema_by_backend
+    class D(dj.Manual):
+        definition = """
+        -> B
+        d_id : int32
+        ---
+        -> C
+        """
+
+    B.insert([(1,), (2,)])
+    C.insert([{"c_id": 10, "b_id": 1}])
+    D.insert([{"b_id": 2, "d_id": 1, "c_id": 10}])
+
+    seed = B & {"b_id": 1}
+    assert dj.Diagram.cascade(seed).counts()[D.full_table_name] == 1
+
+    seed.delete(prompt=False)
+    assert B.to_arrays("b_id").tolist() == [2]
+    assert len(C()) == 0
+    assert len(D()) == 0
+
+
+def test_part_cascade_lifts_only_masters_that_have_a_matching_part(schema_by_backend):
+    """With part_integrity="cascade", a part seed lifts to the masters of the
+    matching part rows. A master that has no matching part row is untouched."""
+
+    @schema_by_backend
+    class Master(dj.Manual):
+        definition = """
+        master_id : int32
+        """
+
+        class P(dj.Part):
+            definition = """
+            -> master
+            p_id : int32
+            """
+
+    @schema_by_backend
+    class Downstream(dj.Manual):
+        definition = """
+        -> Master
+        d_id : int32
+        """
+
+    Master.insert([(1,), (2,)])
+    Master.P.insert([(1, 1)])  # master 2 has no parts
+    Downstream.insert([(1, 1), (2, 1)])
+
+    seed = Master.P & [{"master_id": 1}, {"master_id": 2}]
+    assert dj.Diagram.cascade(seed, part_integrity="cascade").counts()[Master.full_table_name] == 1
+
+    seed.delete(prompt=False, part_integrity="cascade")
+    assert Master.to_arrays("master_id").tolist() == [2]
+    assert len(Master.P()) == 0
+    assert Downstream.to_arrays("master_id").tolist() == [2]