feat(query): after(position) for non-null root order keys

docs/examples/predicates.py

@@ -118,6 +118,26 @@ async def main() -> None:
         assert oldest_first[0].name == "carol"
         assert len(second_page) == 2
 
+        # --8<-- [start:after-paging]
+        page = (
+            await User.select()
+            .order_by(lambda user: user.age)
+            .order_by(lambda user: user.id)
+            .limit(2)
+            .all()
+        )
+        next_page = (
+            await User.select()
+            .order_by(lambda user: user.age)
+            .order_by(lambda user: user.id)
+            .after(page[-1])
+            .limit(2)
+            .all()
+        )
+        # --8<-- [end:after-paging]
+        assert [user.name for user in page] == ["dave", "bob"]
+        assert [user.name for user in next_page] == ["alice", "carol"]
+
         t0 = datetime(2026, 1, 1, tzinfo=UTC)
         t1 = datetime(2026, 2, 1, tzinfo=UTC)
         t2 = datetime(2026, 3, 1, tzinfo=UTC)

docs/pages/guide/queries.md

@@ -196,7 +196,15 @@ Omitting `nulls=` on a nullable sort key means `NULLS LAST` on every backend. Pa
 --8<-- "docs/examples/predicates.py:ordering-slicing"
 ```
 
-Chain `.order_by()` multiple times for multi-column sorts. For robust pagination patterns, see [Pagination](../howto/pagination.md).
+Chain `.order_by()` multiple times for multi-column sorts.
+
+To page forward from a known row, pass that row's place in the declared order to `.after()`. The bound is exclusive, the order keys must include the primary key, and every key must be a non-nullable root column:
+
+```python
+--8<-- "docs/examples/predicates.py:after-paging"
+```
+
+`after(row)` is the same as `after(position_of(row))`. `after()` cannot be combined with `offset()` — a query has one start. For robust pagination patterns, see [Pagination](../howto/pagination.md).
 
 ## Executing Queries
 

docs/solutions/patterns/position-paging-after.md

@@ -0,0 +1,23 @@
+---
+title: Exclusive stepwise compare is the after() expansion
+type: pattern
+tags: [query, paging, ir]
+related_files:
+  - src/query.rs
+  - src/ferro/query/builder.py
+  - src/ferro/query/wire.py
+related_issues: [393, 394, 395]
+captured: 2026-08-30
+---
+
+## Problem
+
+`after(position)` must render as an exclusive keyset bound — `(a > :a) OR (a = :a AND b > :b)` with DESC flipping `>` to `<` — without turning paging into a `where()` predicate and without copying that tree into every SELECT walker. #394 will add NULL-bucket expansion to the same bound.
+
+## Takeaway
+
+One function owns the compare tree: `exclusive_stepwise_compare` in `src/query.rs`. The SELECT walker qualifies columns, binds typed values, and ANDs the result onto WHERE. Do not sprinkle inequalities in `operations.rs`. #394 extends this function; it does not add a second expander.
+
+Python validates the wedge at `after()` / `position_of()` (root columns, PK included, non-nullable) and `compile_query` is the only assembler that puts `after` on the fetch payload as typed `kind`/`value` nodes. Count omits the key; mutations reject it.
+
+Datetime slots go through `_serialize_query_value` → pydantic JSON mode (`…Z` for UTC), the same bytes `save()` writes. `datetime.isoformat()` emits `…+00:00`; on SQLite that is a different TEXT value, so the prefix-equality arm of the stepwise compare never matches.

src/ferro/query/builder.py

@@ -555,6 +555,7 @@ def __init__(
         self.order_by_clause: list[OrderByEntry] = []
         self._limit: int | None = None
         self._offset: int | None = None
+        self._after: tuple[Any, ...] | None = None
         self._m2m_context: M2mContext | None = None
         # Relation paths that must render a join, insertion-ordered (full path
         # tuple -> registered join_type). Populated by where()/order_by()
@@ -1029,15 +1030,128 @@ def offset(self, value: int) -> Self:
         Returns:
             A new ``Query`` with the clause added; ``self`` is unchanged.
 
+        Raises:
+            ValueError: If ``after()`` is already set — a query has one start.
+
         Examples:
             >>> query = User.select().offset(20)
             >>> query._offset
             20
         """
+        if self._after is not None:
+            raise ValueError(
+                "after() cannot be combined with offset(): a query has one "
+                "start (an offset or a position bound, never both)."
+            )
         new = self._clone()
         new._offset = value
         return new
 
+    def _assert_after_order_keys(self) -> None:
+        """Require root, non-null order keys that include the primary key (#393)."""
+        pk = getattr(self.model_cls, "__ferro_pk__", None)
+        if pk is None:
+            raise ValueError(
+                f"{self.model_cls.__name__} has no primary-key column, and "
+                "after()/position_of() require one."
+            )
+        if not self.order_by_clause:
+            raise ValueError(
+                "after() requires order_by() keys that include the model's "
+                "primary key"
+            )
+        specs = getattr(self.model_cls, "__ferro_columns__", {})
+        pk_seen = False
+        for entry in self.order_by_clause:
+            if entry.path:
+                dotted = ".".join((*entry.path, entry.column))
+                raise ValueError(
+                    "after() requires root-column order keys; traversed "
+                    f"order key {dotted!r} is not supported yet"
+                )
+            spec = specs.get(entry.column)
+            if spec is not None and spec.nullable:
+                raise ValueError(
+                    f"after() does not support nullable order key "
+                    f"{entry.column!r}"
+                )
+            if entry.column == pk:
+                pk_seen = True
+        if not pk_seen:
+            raise ValueError(
+                "after() requires the model's primary key in the order keys; "
+                "Ferro will not append it silently"
+            )
+
+    def position_of(self, row: T) -> tuple[Any, ...]:
+        """Read this query's order-key tuple off a model instance.
+
+        Args:
+            row: A hydrated instance of the queried model.
+
+        Returns:
+            The ordered tuple of order-key values, matching ``order_by``
+            declaration order.
+
+        Raises:
+            TypeError: If ``row`` is not an instance of the queried model.
+            ValueError: If the order keys are not a legal ``after()`` set
+                (no primary key, a nullable key, or a traversed key).
+        """
+        if not isinstance(row, self.model_cls):
+            raise TypeError(
+                "position_of() expected an instance of "
+                f"{self.model_cls.__name__}, got {type(row).__name__}"
+            )
+        self._assert_after_order_keys()
+        return tuple(getattr(row, entry.column) for entry in self.order_by_clause)
+
+    def after(self, position: tuple[Any, ...] | T) -> Self:
+        """Start the page after an exclusive position in the declared order.
+
+        ``position`` is the ordered tuple of this query's order-key values, or
+        a model instance (sugar for ``after(position_of(row))``). Order keys
+        must be root columns, include the primary key, and be non-nullable.
+
+        Args:
+            position: A tuple of order-key values, or a model instance.
+
+        Returns:
+            A new ``Query`` with the bound set; ``self`` is unchanged.
+
+        Raises:
+            TypeError: If ``position`` is neither a tuple nor a model instance.
+            ValueError: If the order keys are illegal for ``after()``, the
+                tuple has the wrong arity, a slot is ``None``, or ``offset()``
+                is already set.
+        """
+        if self._offset is not None:
+            raise ValueError(
+                "after() cannot be combined with offset(): a query has one "
+                "start (an offset or a position bound, never both)."
+            )
+        if isinstance(position, tuple):
+            self._assert_after_order_keys()
+            expected = len(self.order_by_clause)
+            if len(position) != expected:
+                raise ValueError(
+                    f"after() expected {expected} position values "
+                    f"(one per order_by key), got {len(position)}"
+                )
+            bound = position
+        elif isinstance(position, self.model_cls):
+            bound = self.position_of(position)
+        else:
+            raise TypeError(
+                "after() expected a position tuple or a model instance, "
+                f"got {type(position).__name__}"
+            )
+        if any(value is None for value in bound):
+            raise ValueError("after() does not support None in a position slot")
+        new = self._clone()
+        new._after = bound
+        return new
+
     async def all(self) -> list[T]:
         """Return all model instances that match the current query
 
@@ -1362,7 +1476,7 @@ def _append_order_by(
         direction: str,
         path: tuple[str, ...],
         *,
-        nulls: str | None = None,
+        nulls: str,
     ) -> Self:
         """Clone with one resolved ORDER BY entry appended (#295)."""
         new = self._clone()

src/ferro/query/nodes.py

@@ -1,12 +1,16 @@
 """Define query AST nodes and field proxies for fluent filtering"""
 
 import difflib
+import json
 import uuid
 from collections.abc import Callable, Mapping
 from dataclasses import dataclass
+from datetime import datetime
 from decimal import Decimal
 from typing import TYPE_CHECKING, Any, Generic, NoReturn, TypeAlias, TypeVar, get_origin
 
+from pydantic_core import to_json
+
 TField = TypeVar("TField")
 TModel = TypeVar("TModel")
 
@@ -254,7 +258,15 @@ def __repr__(self):
 
 
 def _serialize_query_value(value: Any) -> Any:
-    """Normalize Python values into JSON-friendly query payloads."""
+    """Normalize Python values into JSON-friendly query payloads.
+
+    Datetimes use pydantic JSON mode (the same canonical form as
+    ``save_bind_payload``): UTC is ``...Z``, not ``datetime.isoformat()``'s
+    ``...+00:00``. SQLite stores INSERT text in that form; ``after()`` prefix
+    equality is a TEXT compare there and must match the stored bytes.
+    """
+    if isinstance(value, datetime):
+        return json.loads(to_json(value))
     if hasattr(value, "isoformat"):
         return value.isoformat()
     if isinstance(value, (Decimal, uuid.UUID)):