diff --git a/src/datajoint/condition.py b/src/datajoint/condition.py
index 650015539..161ed4407 100644
--- a/src/datajoint/condition.py
+++ b/src/datajoint/condition.py
@@ -167,15 +167,13 @@ def merge(self, other: "Top") -> "Top":
         Top
             New Top with merged limit/offset and preserved ordering.
         """
-        # Compute effective limit (minimum of defined limits)
-        if self.limit is None and other.limit is None:
-            new_limit = None
-        elif self.limit is None:
+        # other skips other.offset rows of self's window and takes up to other.limit of the
+        # rest, so the merged window never extends past self's limit
+        if self.limit is None:
             new_limit = other.limit
-        elif other.limit is None:
-            new_limit = self.limit
         else:
-            new_limit = min(self.limit, other.limit)
+            remaining = max(self.limit - other.offset, 0)
+            new_limit = remaining if other.limit is None else min(remaining, other.limit)
 
         return Top(
             limit=new_limit,
@@ -379,6 +377,15 @@ def combine_conditions(negate, conditions):
         negate = not negate
         condition = condition.restriction
 
+    # restrict() applies dj.Top itself; here it was negated or nested in a list or AndList
+    if isinstance(condition, Top):
+        raise DataJointError(
+            "dj.Top must be applied on its own, as `expression & dj.Top(...)`. It cannot be negated "
+            "(with - or dj.Not) or combined with other restrictions in a list, an AndList, or the "
+            "restrictions passed to populate() or progress(). For the rows after the first n, "
+            "use dj.Top(None, order_by=..., offset=n)."
+        )
+
     # restrict by string
     if isinstance(condition, str):
         columns.update(extract_column_names(condition))
@@ -442,7 +449,9 @@ def combine_conditions(negate, conditions):
         # Match on all non-hidden namesakes (hidden attributes excluded)
         common_attributes = [q for q in condition.heading.names if q in query_expression.heading.names]
         columns.update(common_attributes)
-        if isinstance(condition, Aggregation):
+        # A dj.Top limit stays inside a subquery: it applies to the expression's rows before
+        # they are projected to the common attributes, and MySQL rejects LIMIT inside IN (...).
+        if isinstance(condition, Aggregation) or condition._top is not None:
             condition = condition.make_subquery()
         return (
             # without common attributes, any non-empty set matches everything
diff --git a/src/datajoint/diagram.py b/src/datajoint/diagram.py
index d3d7c40bc..598d36d51 100644
--- a/src/datajoint/diagram.py
+++ b/src/datajoint/diagram.py
@@ -799,8 +799,9 @@ def restrict(self, table_expr):
         if node not in result.nodes():
             raise DataJointError(f"Table {node} is not in the diagram.")
         # Seed restriction (AND accumulation)
-        result._restrict_conditions.setdefault(node, AndList()).extend(table_expr.restriction)
-        result._restriction_attrs.setdefault(node, set()).update(table_expr.restriction_attributes)
+        restriction, restriction_attrs = _seed_restriction(table_expr)
+        result._restrict_conditions.setdefault(node, AndList()).extend(restriction)
+        result._restriction_attrs.setdefault(node, set()).update(restriction_attrs)
         # Propagate downstream
         result._propagate_restrictions(node, mode="restrict")
         return result
diff --git a/src/datajoint/expression.py b/src/datajoint/expression.py
index 1b5f5ac9e..25755e287 100644
--- a/src/datajoint/expression.py
+++ b/src/datajoint/expression.py
@@ -400,18 +400,21 @@ def join(self, other, semantic_check=True, left=False, allow_nullable_pk=False):
         need_subquery1 = need_subquery2 = bool(
             (set(self.original_heading.names) & set(other.original_heading.names)) - join_attributes
         )
-        # need subquery if any of the join attributes are derived
+        # need subquery if any of the join attributes are derived, or if an operand is limited
+        # by dj.Top: its ORDER BY/LIMIT must apply to the operand before the join
         need_subquery1 = (
             need_subquery1
             or isinstance(self, Aggregation)
             or any(n in self.heading.new_attributes for n in join_attributes)
             or isinstance(self, Union)
+            or self._top is not None
         )
         need_subquery2 = (
             need_subquery2
             or isinstance(other, Aggregation)
             or any(n in other.heading.new_attributes for n in join_attributes)
             or isinstance(self, Union)
+            or other._top is not None
         )
         # With USING clause (instead of NATURAL JOIN), we need subqueries when
         # joining with multi-table expressions to ensure correct column matching
@@ -614,6 +617,22 @@ def proj(self, *attributes, **named_attributes):
         if not need_subquery and self.restriction:
             # need a subquery if the restriction applies to attributes that have been renamed
             need_subquery = any(name in self.restriction_attributes for name in self.heading.new_attributes)
+        if not need_subquery and self._top is not None:
+            # need a subquery if dj.Top's ORDER BY names an attribute that the projection redefines,
+            # or a computed or renamed attribute that it drops. A dropped table column needs none.
+            projected = self.heading.select(
+                attributes, rename_map=dict(**rename_map, **replicate_map), compute_map=compute_map
+            )
+            ordering = ", ".join(_flatten_attribute_list(self.primary_key, self._top.order_by or ["KEY"]))
+            need_subquery = any(
+                (
+                    projected[name].attribute_expression != self.heading[name].attribute_expression
+                    if name in projected.names
+                    else self.heading[name].attribute_expression is not None
+                )
+                for name in extract_column_names(ordering)
+                if name in self.heading.names
+            )
 
         result = self.make_subquery() if need_subquery else copy.copy(self)
         result._original_heading = result.original_heading
@@ -1186,6 +1205,8 @@ def __bool__(self):
         bool
             True if the result is not empty.
         """
+        if self._top is not None:
+            return bool(self.make_subquery())  # EXISTS must see the ORDER BY/LIMIT
         return bool(
             self.connection.query(
                 "SELECT EXISTS(SELECT 1 FROM {from_}{where})".format(from_=self.from_clause(), where=self.where_clause())
@@ -1434,15 +1455,17 @@ def make_sql(self):
         if not arg1.heading.secondary_attributes and not arg2.heading.secondary_attributes:
             # no secondary attributes: use UNION DISTINCT
             fields = arg1.primary_key
-            alias_name = f"_u{next(self.__count)}{self.sorting_clauses()}"
-            alias_quoted = self.connection.adapter.quote_identifier(alias_name)
+            alias_quoted = self.connection.adapter.quote_identifier(f"_u{next(self.__count)}")
             sql1 = arg1.make_sql() if isinstance(arg1, Union) else arg1.make_sql(fields)
             sql2 = arg2.make_sql() if isinstance(arg2, Union) else arg2.make_sql(fields)
-            return f"SELECT * FROM (({sql1}) UNION ({sql2})) as {alias_quoted}"
+            return f"SELECT * FROM (({sql1}) UNION ({sql2})) as {alias_quoted}{self.sorting_clauses()}"
         # with secondary attributes, use union of left join with anti-restriction
         fields = self.heading.names
         sql1 = arg1.join(arg2, left=True).make_sql(fields)
         sql2 = (arg2 - arg1).proj(..., **{k: "NULL" for k in arg1.heading.secondary_attributes}).make_sql(fields)
+        if self._top is not None:
+            alias_quoted = self.connection.adapter.quote_identifier(f"_u{next(self.__count)}")
+            return f"SELECT * FROM (({sql1}) UNION ({sql2})) as {alias_quoted}{self.sorting_clauses()}"
         return "({sql1})  UNION ({sql2})".format(sql1=sql1, sql2=sql2)
 
     def from_clause(self):
@@ -1509,6 +1532,8 @@ def __and__(self, other):
             other = other()  # instantiate if a class
         if not isinstance(other, QueryExpression):
             raise DataJointError("Set U can only be restricted with a QueryExpression.")
+        if other._top is not None:
+            other = other.make_subquery()  # apply the limit before DISTINCT
         result = copy.copy(other)
         result._distinct = True
         result._heading = result.heading.set_primary_key(self.primary_key)
@@ -1563,6 +1588,9 @@ def aggr(self, group, **named_attributes):
         if missing:
             raise DataJointError(f"Attributes {missing} not found in the group expression.")
 
+        if group._top is not None:
+            group = group.make_subquery()  # apply the limit before GROUP BY
+
         # Create Aggregation directly without join - just group by U's primary key
         result = Aggregation()
         result._connection = group.connection
diff --git a/src/datajoint/heading.py b/src/datajoint/heading.py
index 9addae474..ce2d07611 100644
--- a/src/datajoint/heading.py
+++ b/src/datajoint/heading.py
@@ -775,5 +775,6 @@ def make_subquery_heading(self):
         """
         return Heading(
             (dict(v.todict(), attribute_expression=None) for v in self.attributes.values()),
+            table_info=self.table_info,  # keep the adapter, so a later rename quotes for the backend
             lineage_available=self._lineage_available,
         )
diff --git a/tests/integration/test_relational_operand.py b/tests/integration/test_relational_operand.py
index eea53288e..afdd12328 100644
--- a/tests/integration/test_relational_operand.py
+++ b/tests/integration/test_relational_operand.py
@@ -608,12 +608,14 @@ def test_top_errors(self, schema_simp_pop):
             L() & dj.Top(order_by=1)
         with pytest.raises(TypeError) as err5:
             L() & dj.Top(offset="1")
-        assert "datajoint.errors.DataJointError: Invalid restriction type Top(limit=1, order_by=['KEY'], offset=0)" == str(
-            err1.exconly()
-        )
-        assert "datajoint.errors.DataJointError: Invalid restriction type Top(limit=1, order_by=['KEY'], offset=0)" == str(
-            err2.exconly()
+        nested_top = (
+            "datajoint.errors.DataJointError: dj.Top must be applied on its own, as `expression & dj.Top(...)`. "
+            "It cannot be negated (with - or dj.Not) or combined with other restrictions in a list, an AndList, "
+            "or the restrictions passed to populate() or progress(). For the rows after the first n, "
+            "use dj.Top(None, order_by=..., offset=n)."
         )
+        assert nested_top == str(err1.exconly())
+        assert nested_top == str(err2.exconly())
         assert "TypeError: Top limit must be an integer" == str(err3.exconly())
         assert "TypeError: Top order_by attributes must all be strings" == str(err4.exconly())
         assert "TypeError: The offset argument must be an integer" == str(err5.exconly())
diff --git a/tests/integration/test_top_in_queries.py b/tests/integration/test_top_in_queries.py
new file mode 100644
index 000000000..52f0ff334
--- /dev/null
+++ b/tests/integration/test_top_in_queries.py
@@ -0,0 +1,391 @@
+"""
+dj.Top is honored when the limited expression is an operand of another query (#1656).
+
+An expression restricted by dj.Top carries its ORDER BY/LIMIT in ``_top``. Joins (on either
+side), aggr, dj.U.aggr, dj.U restriction, union, and restriction by a limited expression must
+apply that limit to the operand itself, which takes a subquery. dj.Top cannot be negated; the
+rows after the first n are ``dj.Top(None, order_by=..., offset=n)``.
+"""
+
+import time
+
+import pytest
+
+import datajoint as dj
+from datajoint.errors import DataJointError
+
+
+@pytest.fixture(scope="function")
+def schema_by_backend(connection_by_backend, db_creds_by_backend):
+    """Create a fresh schema per test, parameterized across backends."""
+    backend = db_creds_by_backend["backend"]
+    test_id = str(int(time.time() * 1000))[-8:]
+    schema_name = f"djtest_top_{backend}_{test_id}"[:64]
+    drop = f"DROP DATABASE IF EXISTS {connection_by_backend.adapter.quote_identifier(schema_name)}"
+    if connection_by_backend.is_connected:
+        try:
+            connection_by_backend.query(drop)
+        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(drop)
+        except Exception:
+            pass
+
+
+@pytest.fixture(scope="function")
+def tables(schema_by_backend):
+    schema = schema_by_backend
+
+    @schema
+    class Parent(dj.Manual):
+        definition = """
+        parent_id : int32
+        ---
+        label : varchar(8)
+        """
+
+    @schema
+    class Child(dj.Manual):
+        definition = """
+        -> Parent
+        child_id : int32
+        ---
+        color : varchar(8)
+        """
+
+    @schema
+    class ParentInfo(dj.Manual):
+        definition = """
+        -> Parent
+        ---
+        info : varchar(8)
+        """
+
+    Parent.insert(
+        [
+            {"parent_id": 1, "label": "a"},
+            {"parent_id": 2, "label": "b"},
+            {"parent_id": 3, "label": "a"},
+        ]
+    )
+    Child.insert(
+        [
+            {"parent_id": 1, "child_id": 1, "color": "red"},
+            {"parent_id": 1, "child_id": 2, "color": "blue"},
+            {"parent_id": 2, "child_id": 1, "color": "blue"},
+        ]
+    )
+    ParentInfo.insert(
+        [
+            {"parent_id": 1, "info": "x"},
+            {"parent_id": 2, "info": "y"},
+            {"parent_id": 3, "info": "z"},
+        ]
+    )
+    return Parent, Child, ParentInfo
+
+
+# Limits used below, with the rows they select on their own.
+def top_child():  # child (1, 2)
+    return dj.Top(1, order_by="child_id DESC, parent_id")
+
+
+def top2_child():  # children (1, 2), (1, 1)
+    return dj.Top(2, order_by="child_id DESC, parent_id")
+
+
+def rest_child():  # children (1, 1), (2, 1): all but the first
+    return dj.Top(None, order_by="child_id DESC, parent_id", offset=1)
+
+
+def top_parent():  # parent 1
+    return dj.Top(1, order_by="parent_id")
+
+
+def rest_parent():  # parents 2, 3
+    return dj.Top(None, order_by="parent_id", offset=1)
+
+
+def rows(query, *attributes):
+    result = [tuple(row[a] for a in attributes) for row in query.to_dicts()]
+    assert len(result) == len(set(result)), "duplicate rows"
+    return set(result)
+
+
+def counts(query, key, value="n"):
+    result = {row[key]: row[value] for row in query.to_dicts()}
+    return result
+
+
+def test_limits_on_their_own(tables):
+    Parent, Child, _ = tables
+    assert rows(Child & top_child(), "parent_id", "child_id") == {(1, 2)}
+    assert rows(Child & top2_child(), "parent_id", "child_id") == {(1, 2), (1, 1)}
+    assert rows(Child & rest_child(), "parent_id", "child_id") == {(1, 1), (2, 1)}
+    assert rows(Parent & top_parent(), "parent_id") == {(1,)}
+    assert rows(Parent & rest_parent(), "parent_id") == {(2,), (3,)}
+
+
+# --- joins: a limited operand on either side keeps its limit ---
+
+
+@pytest.mark.parametrize(
+    "make, expected",
+    [
+        (lambda P, C: P * (C & top_child()), {(1, 2)}),
+        (lambda P, C: (P & top_parent()) * C, {(1, 1), (1, 2)}),
+        (lambda P, C: (P & top_parent()) * (C & top_child()), {(1, 2)}),
+        (lambda P, C: P * ((C & "color='blue'") & dj.Top(1, order_by="parent_id DESC")), {(2, 1)}),
+        (lambda P, C: P * (C & rest_child()), {(1, 1), (2, 1)}),
+        (lambda P, C: (P & rest_parent()) * C, {(2, 1)}),
+        (lambda P, C: P * (C & top_child()).proj(), {(1, 2)}),
+    ],
+    ids=["right", "left", "both", "with-condition", "right-offset", "left-offset", "projected"],
+)
+def test_join_keeps_limit(tables, make, expected):
+    Parent, Child, _ = tables
+    assert rows(make(Parent, Child), "parent_id", "child_id") == expected
+
+
+def test_join_without_limit_has_no_subquery(tables):
+    Parent, Child, _ = tables
+    query = Parent * (Child & "color='blue'")
+    assert rows(query, "parent_id", "child_id") == {(1, 2), (2, 1)}
+    assert "(SELECT" not in query.from_clause()
+
+
+def test_extend_with_limited_left_operand(tables):
+    Parent, Child, _ = tables
+    query = (Child & top_child()).extend(Parent)
+    assert rows(query, "parent_id", "child_id", "label") == {(1, 2, "a")}
+
+
+# --- aggregation ---
+
+
+def test_aggr_with_limited_grouping_expression(tables):
+    Parent, Child, _ = tables
+    assert counts((Parent & top_parent()).aggr(Child, n="count(child_id)"), "parent_id") == {1: 2}
+    assert counts((Parent & rest_parent()).aggr(Child, n="count(child_id)"), "parent_id") == {2: 1, 3: 0}
+
+
+@pytest.mark.parametrize(
+    "limit, expected",
+    [(top_child, {1: 1}), (rest_child, {1: 1, 2: 1})],
+    ids=["top", "offset"],
+)
+def test_inner_aggr_with_limited_group(tables, limit, expected):
+    Parent, Child, _ = tables
+    query = Parent.aggr(Child & limit(), n="count(child_id)", exclude_nonmatching=True)
+    assert counts(query, "parent_id") == expected
+
+
+def test_left_join_with_limited_right_operand(tables):
+    Parent, Child, _ = tables
+    query = Child.extend(Parent & top_parent())
+    assert rows(query, "parent_id", "child_id", "label") == {(1, 1, "a"), (1, 2, "a"), (2, 1, None)}
+    query = Parent.aggr(Child & rest_child(), n="count(child_id)")
+    assert counts(query, "parent_id") == {1: 1, 2: 1, 3: 0}
+
+
+@pytest.mark.parametrize(
+    "limit, expected",
+    [(top_child, {"blue": 1}), (rest_child, {"red": 1, "blue": 1})],
+    ids=["top", "offset"],
+)
+def test_universal_aggr_with_limited_group(tables, limit, expected):
+    _, Child, _ = tables
+    assert counts(dj.U("color").aggr(Child & limit(), n="count(*)"), "color") == expected
+
+
+def test_universal_aggr_without_limit_has_no_subquery(tables):
+    _, Child, _ = tables
+    query = dj.U("color").aggr(Child, n="count(*)")
+    assert counts(query, "color") == {"red": 1, "blue": 2}
+    assert "(SELECT" not in query.from_clause()
+
+
+# --- restriction by a limited expression (MySQL rejects LIMIT inside IN) ---
+
+
+@pytest.mark.parametrize(
+    "limit, matching, rest",
+    [
+        (top_child, {(1,)}, {(2,), (3,)}),
+        (top2_child, {(1,)}, {(2,), (3,)}),
+        (rest_child, {(1,), (2,)}, {(3,)}),
+    ],
+    ids=["top", "top2", "offset"],
+)
+def test_restriction_by_limited_expression(tables, limit, matching, rest):
+    Parent, Child, _ = tables
+    assert rows(Parent & (Child & limit()), "parent_id") == matching
+    assert rows(Parent - (Child & limit()), "parent_id") == rest
+
+
+def test_restriction_by_limit_on_computed_attribute(tables):
+    """The limit applies before the projection to the common attributes."""
+    Parent, Child, _ = tables
+    # score: (1, 1) -> 11, (1, 2) -> 21, (2, 1) -> 12
+    limited = Child.proj(score="child_id * 10 + parent_id") & dj.Top(1, order_by="score DESC")
+    assert rows(limited, "parent_id", "child_id", "score") == {(1, 2, 21)}
+    assert rows(Parent & limited, "parent_id") == {(1,)}
+    assert rows(Parent - limited, "parent_id") == {(2,), (3,)}
+
+
+def test_restriction_by_limited_union(tables):
+    Parent, Child, ParentInfo = tables
+    first = (Parent.proj() + ParentInfo.proj()) & dj.Top(1, order_by="parent_id")
+    assert rows(Child & first, "parent_id", "child_id") == {(1, 1), (1, 2)}
+    assert rows(Child - first, "parent_id", "child_id") == {(2, 1)}
+
+
+def test_restriction_without_limit_has_no_derived_table(tables):
+    Parent, Child, _ = tables
+    query = Parent & (Child & "color='blue'")
+    assert rows(query, "parent_id") == {(1,), (2,)}
+    assert query.where_clause().count("SELECT") == 1
+
+
+@pytest.mark.parametrize(
+    "limit",
+    [top2_child, lambda: dj.Top(2, order_by="parent_id")],
+    ids=["order-by-dropped-attribute", "order-by-kept-attribute"],
+)
+def test_universal_restriction_by_limited_expression(tables, limit):
+    """The limit applies to Child's rows before dj.U takes the distinct parent_id values."""
+    _, Child, _ = tables
+    assert rows(dj.U("parent_id") & (Child & limit()), "parent_id") == {(1,)}
+
+
+def test_universal_restriction_without_limit_has_no_subquery(tables):
+    _, Child, _ = tables
+    query = dj.U("parent_id") & Child
+    assert rows(query, "parent_id") == {(1,), (2,)}
+    assert "(SELECT" not in query.from_clause()
+
+
+# --- projection of a limited expression ---
+
+
+def test_keys_of_limit_on_aggregate(tables):
+    """The ORDER BY names a computed attribute that keys() drops."""
+    Parent, Child, _ = tables
+    limited = Parent.aggr(Child, n="count(child_id)") & dj.Top(1, order_by="n DESC")
+    assert limited.keys() == [{"parent_id": 1}]
+    assert rows(Child * limited.proj(), "parent_id", "child_id") == {(1, 1), (1, 2)}
+
+
+def test_projection_drops_computed_ordering_attribute(tables):
+    Parent, _, _ = tables
+    limited = Parent.proj(neg="-parent_id") & dj.Top(1, order_by="neg")  # parent 3
+    assert rows(limited.proj(), "parent_id") == {(3,)}
+    assert len(limited.proj()) == 1
+
+
+@pytest.mark.parametrize(
+    "make, attributes, expected",
+    [
+        (
+            lambda P, C: (P & dj.Top(1, order_by="label DESC, parent_id")).proj(label="-parent_id"),
+            ("parent_id", "label"),
+            {(2, -2)},
+        ),
+        (
+            lambda P, C: (C & top_child()).proj(child_id="parent_id", parent_id="child_id"),
+            ("parent_id", "child_id"),
+            {(2, 1)},
+        ),
+    ],
+    ids=["recomputed", "swapped"],
+)
+def test_projection_redefines_ordering_attribute(tables, make, attributes, expected):
+    """A new attribute that takes an ORDER BY name must not change which rows the limit selects."""
+    Parent, Child, _ = tables
+    assert rows(make(Parent, Child), *attributes) == expected
+
+
+def test_projection_dropping_table_column_keeps_order_in_place(tables):
+    """ORDER BY on a table column that the projection drops needs no subquery."""
+    _, Child, _ = tables
+    query = (Child & dj.Top(1, order_by="color DESC")).proj()
+    assert rows(query, "parent_id", "child_id") == {(1, 1)}
+    assert "(SELECT" not in query.from_clause()
+
+
+# --- merged limits, emptiness, diagrams ---
+
+
+@pytest.mark.parametrize(
+    "second",
+    [
+        lambda: dj.Top(None, order_by="parent_id", offset=1),
+        lambda: dj.Top(5, order_by=None, offset=1),
+    ],
+    ids=["offset", "offset-inherit"],
+)
+def test_second_limit_stays_within_first(tables, second):
+    Parent, _, _ = tables
+    first_two = Parent & dj.Top(2, order_by="parent_id")  # parents 1, 2
+    assert rows(first_two & second(), "parent_id") == {(2,)}
+    assert [row["parent_id"] for row in first_two.to_dicts(limit=5, offset=1)] == [2]
+
+
+def test_bool_of_limited_expression(tables):
+    Parent, _, _ = tables
+    assert not (Parent & dj.Top(None, order_by="parent_id", offset=3))
+    assert not (Parent & dj.Top(0))
+    assert Parent & rest_parent()
+
+
+def test_diagram_restrict_by_limited_expression(tables, schema_by_backend):
+    Parent, Child, ParentInfo = tables
+    restricted = dj.Diagram(schema_by_backend).restrict(Parent & top_parent())
+    expected = {Parent.full_table_name: 1, Child.full_table_name: 2, ParentInfo.full_table_name: 1}
+    assert restricted.counts() == expected
+
+
+# --- union ---
+
+
+def test_union_restricted_by_top(tables):
+    Parent, _, ParentInfo = tables
+    keys_only = (Parent.proj() + ParentInfo.proj()) & dj.Top(1, order_by="parent_id DESC")
+    assert rows(keys_only, "parent_id") == {(3,)}
+    with_attributes = (Parent + ParentInfo) & dj.Top(1, order_by="parent_id DESC")
+    assert rows(with_attributes, "parent_id", "label", "info") == {(3, "a", "z")}
+
+
+def test_union_of_limited_operands(tables):
+    Parent, _, ParentInfo = tables
+    last_info = ParentInfo & dj.Top(1, order_by="parent_id DESC")
+    keys_only = (Parent & top_parent()).proj() + last_info.proj()
+    assert rows(keys_only, "parent_id") == {(1,), (3,)}
+    with_attributes = (Parent & top_parent()) + last_info
+    assert rows(with_attributes, "parent_id", "label", "info") == {(1, "a", None), (3, None, "z")}
+
+
+# --- dj.Top cannot be negated; the error names the offset form ---
+
+
+@pytest.mark.parametrize(
+    "make",
+    [
+        lambda P: P - top_parent(),
+        lambda P: P & dj.Not(top_parent()),
+        lambda P: P & [top_parent()],
+        lambda P: P - [top_parent()],
+        lambda P: P & dj.AndList([top_parent(), "parent_id > 1"]),
+    ],
+    ids=["minus", "not", "list", "minus-list", "andlist"],
+)
+def test_negated_or_nested_top_is_rejected(tables, make):
+    Parent, _, _ = tables
+    with pytest.raises(DataJointError, match="offset=n"):
+        make(Parent)
diff --git a/tests/unit/test_condition.py b/tests/unit/test_condition.py
index 3200e34c4..4342f3518 100644
--- a/tests/unit/test_condition.py
+++ b/tests/unit/test_condition.py
@@ -53,6 +53,21 @@ def test_merge_offsets_add(self):
         merged = top1.merge(top2)
         assert merged.offset == 7
 
+    @pytest.mark.parametrize(
+        "first, second, limit, offset",
+        [
+            (Top(limit=2, order_by="id"), Top(limit=None, order_by=None, offset=1), 1, 1),
+            (Top(limit=2, order_by="id"), Top(limit=5, order_by=None, offset=1), 1, 1),
+            (Top(limit=2, order_by="id", offset=3), Top(limit=1, order_by=None, offset=1), 1, 4),
+            (Top(limit=2, order_by="id"), Top(limit=1, order_by=None, offset=5), 0, 5),
+            (Top(limit=None, order_by="id"), Top(limit=None, order_by=None, offset=2), 999999999999, 2),
+        ],
+    )
+    def test_merge_offset_stays_within_first_limit(self, first, second, limit, offset):
+        """The second Top skips rows of the first Top's window and never reaches past its limit."""
+        merged = first.merge(second)
+        assert (merged.limit, merged.offset) == (limit, offset)
+
     def test_merge_preserves_existing_order(self):
         """Merged Top preserves first Top's ordering."""
         top1 = Top(limit=10, order_by=["col1 desc", "col2 asc"])