From c652ce6f1b1cd3591a286c281ad5e8f72d9322ee Mon Sep 17 00:00:00 2001
From: Taylor <taylor@x54.sh>
Date: Sat, 18 Jul 2026 11:14:40 -0400
Subject: [PATCH 1/3] =?UTF-8?q?feat(query):=20scoped=20existence=20tests?=
 =?UTF-8?q?=20=E2=80=94=20inner=20predicates,=20nesting,=20cross-scope=20g?=
 =?UTF-8?q?uard?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The existence test's optional inner lambda is a full ferro predicate
over the child model, resolved through the same validating QueryProxy a
root where() receives: every operator, &/|/~ composition, forward-FK
traversal, and nested existence tests to arbitrary depth. The explicit
lambda scope is what makes multi-condition grouping unambiguous —
exists(lambda l: A & B) (one child row matches both) vs
exists(A-test) & exists(B-test) (some child row each) — the
Django-style implicit-traversal ambiguity ADR-0007 rejects.

- Forward traversal inside the test renders its joins INSIDE the EXISTS
  subquery under unchanged ADR-0006 semantics (INNER at every hop,
  narrowing). The hop facts ride the exists node's own `joins` section
  (QueryJoin entries, always "inner"), serialized only when non-empty —
  bare-test wire bytes are unchanged from the tracer bullet.
- Nesting is recursion, not a second mechanism: a nested test's
  correlation hop resolves against the enclosing SUBQUERY's alias
  through the ExistsScope qualifier, pinned by a Rust render test and
  depth-2 e2e coverage.
- Cross-scope references are rejected at build time pointing at the
  deferred capability (#309): a leaf built from any proxy other than
  the inner lambda's parameter (detected via a compile-side `owner`
  scope tag the proxies now stamp on comparison nodes), a FieldProxy as
  comparison RHS (column-to-column), and a nested test built from
  another scope's proxy. Silent misrendering is the failure mode this
  guard exists to prevent.
- Typed binds inside the subquery resolve against the leaf's OWNING
  table: the SELECT walkers now populate registrations and enum
  catalogs for every table an existence test reaches (correlation hops,
  inner join hops, nested tests).
- Hand-authored golden vectors pin the scoped shape (inner tree +
  inner-traversal joins) and the nested exists-in-exists shape,
  asserted from the Python emitter and the Rust decoder.
- End-to-end on both backends: the full #308 line-aware category filter
  (three matching lines → one root row; child-less roots survive the
  OR; keyset order_by + limit compose), the explicit grouping contrast,
  forward traversal inside the subquery, depth-2 nesting, negated and
  counted scoped tests, and every pinned cross-scope error.

Refs #311, #315
---
 crates/ferro-schema-ir/src/lib.rs             |  75 ++++-
 src/ferro/query/nodes.py                      | 257 ++++++++++++--
 src/operations.rs                             |  49 ++-
 src/query.rs                                  | 226 +++++++++++--
 .../query_account_scoped_exists_v7.json       |  74 ++++
 .../query_owner_nested_exists_v7.json         |  47 +++
 tests/test_ir_vectors_contract.py             |  23 +-
 tests/test_query_exists.py                    | 316 +++++++++++++++++-
 tests/test_query_wire_vectors.py              |  23 ++
 9 files changed, 1022 insertions(+), 68 deletions(-)
 create mode 100644 tests/fixtures/ir_vectors/query_account_scoped_exists_v7.json
 create mode 100644 tests/fixtures/ir_vectors/query_owner_nested_exists_v7.json

diff --git a/crates/ferro-schema-ir/src/lib.rs b/crates/ferro-schema-ir/src/lib.rs
index 3f27977..dbf6d50 100644
--- a/crates/ferro-schema-ir/src/lib.rs
+++ b/crates/ferro-schema-ir/src/lib.rs
@@ -388,6 +388,12 @@ pub enum QueryNode {
         /// nesting is recursion, not a second mechanism.
         #[serde(rename = "where")]
         where_clause: Vec<QueryNode>,
+        /// Forward-traversal joins the inner tree references (#315),
+        /// rendered INSIDE the subquery as inner joins (ADR-0006 traversal
+        /// semantics are unchanged there; `left_join` has no inner-lambda
+        /// spelling). Absent on the wire when empty — pinned bytes.
+        #[serde(default, skip_serializing_if = "Vec::is_empty")]
+        joins: Vec<QueryJoin>,
     },
 }
 
@@ -557,7 +563,9 @@ mod tests {
         let envelope: IrEnvelope<QueryIrPayload> =
             serde_json::from_value(ir.clone()).expect("query exists IR must deserialize");
         match &envelope.payload.where_clause[0] {
-            QueryNode::Exists { hops, where_clause } => {
+            QueryNode::Exists {
+                hops, where_clause, ..
+            } => {
                 assert_eq!(hops.len(), 1);
                 assert_eq!(hops[0].relation, "transactions");
                 assert_eq!(hops[0].from_column, "id");
@@ -597,6 +605,71 @@ mod tests {
         assert_eq!(encoded, ir, "query not-exists round-trip must not drift");
     }
 
+    #[test]
+    fn query_scoped_exists_fixture_roundtrip() {
+        // The scoped existence-test golden vector (#315): the inner tree is
+        // an ordinary condition tree, and a traversed inner leaf's hop facts
+        // ride the exists node's own `joins` section (rendered INSIDE the
+        // subquery). Must survive a deserialize/serialize round-trip.
+        let fixture =
+            include_str!("../../../tests/fixtures/ir_vectors/query_account_scoped_exists_v7.json");
+        let parsed: serde_json::Value =
+            serde_json::from_str(fixture).expect("query scoped-exists fixture must parse");
+        let ir = parsed
+            .get("ir")
+            .cloned()
+            .expect("fixture must contain ir envelope");
+        let envelope: IrEnvelope<QueryIrPayload> =
+            serde_json::from_value(ir.clone()).expect("query scoped-exists IR must deserialize");
+        match &envelope.payload.where_clause[0] {
+            QueryNode::Exists {
+                hops,
+                where_clause,
+                joins,
+            } => {
+                assert_eq!(hops.len(), 1);
+                assert_eq!(where_clause.len(), 1);
+                assert_eq!(joins.len(), 1);
+                assert_eq!(joins[0].join_type, "inner");
+                assert_eq!(joins[0].path[0].relation, "account");
+            }
+            other => panic!("where[0] must be an exists node, got {other:?}"),
+        }
+        let encoded =
+            serde_json::to_value(&envelope).expect("query scoped-exists IR must serialize");
+        assert_eq!(encoded, ir, "query scoped-exists round-trip must not drift");
+    }
+
+    #[test]
+    fn query_nested_exists_fixture_roundtrip() {
+        // Nested exists-in-exists (#315): depth is recursion, not a second
+        // mechanism, and the bare inner node omits `joins` entirely (absent,
+        // not empty — pinned wire bytes via skip_serializing_if).
+        let fixture =
+            include_str!("../../../tests/fixtures/ir_vectors/query_owner_nested_exists_v7.json");
+        let parsed: serde_json::Value =
+            serde_json::from_str(fixture).expect("query nested-exists fixture must parse");
+        let ir = parsed
+            .get("ir")
+            .cloned()
+            .expect("fixture must contain ir envelope");
+        let envelope: IrEnvelope<QueryIrPayload> =
+            serde_json::from_value(ir.clone()).expect("query nested-exists IR must deserialize");
+        match &envelope.payload.where_clause[0] {
+            QueryNode::Exists { where_clause, .. } => match &where_clause[0] {
+                QueryNode::Exists { hops, joins, .. } => {
+                    assert_eq!(hops[0].relation, "transactions");
+                    assert!(joins.is_empty(), "bare nested test carries no joins");
+                }
+                other => panic!("inner where[0] must be an exists node, got {other:?}"),
+            },
+            other => panic!("where[0] must be an exists node, got {other:?}"),
+        }
+        let encoded =
+            serde_json::to_value(&envelope).expect("query nested-exists IR must serialize");
+        assert_eq!(encoded, ir, "query nested-exists round-trip must not drift");
+    }
+
     #[test]
     fn query_traversal_fixture_roundtrip() {
         // Multi-hop `joins` section + path-carrying leaves must survive a
diff --git a/src/ferro/query/nodes.py b/src/ferro/query/nodes.py
index 132467c..33fc8b1 100644
--- a/src/ferro/query/nodes.py
+++ b/src/ferro/query/nodes.py
@@ -76,6 +76,7 @@ def __init__(
         path: tuple[str, ...] = (),
         child: "QueryNode | None" = None,
         exists: "ExistsTest | None" = None,
+        owner: type | None = None,
     ):
         """Initialize a query expression node
 
@@ -95,6 +96,11 @@ def __init__(
             exists: The correlation hops and inner condition tree of an
                 existence test (built by ``t.rel.exists()``, ADR-0007);
                 ``None`` for every other node kind.
+            owner: The model class whose scope built this leaf (the proxy's
+                owner), when known. A compile-side fact, never serialized —
+                the cross-scope guard on scoped existence tests (#315) reads
+                it to catch a leaf smuggled in from another lambda's
+                parameter.
         """
         self.column = column
         self.operator = operator
@@ -105,6 +111,7 @@ def __init__(
         self.path = path
         self.child = child
         self.exists = exists
+        self.owner = owner
 
     def __or__(self, other: "QueryNode") -> "QueryNode":
         """Combine two nodes with logical OR
@@ -169,11 +176,19 @@ def to_ir_dict(self) -> dict[str, Any]:
         if self.child is not None:
             return {"node_kind": "not", "child": self.child.to_ir_dict()}
         if self.exists is not None:
-            return {
+            serialized_exists: dict[str, Any] = {
                 "node_kind": "exists",
                 "hops": [hop.to_ir_dict() for hop in self.exists.hops],
                 "where": [node.to_ir_dict() for node in self.exists.where],
             }
+            # The inner-traversal joins key is absent, not empty, on a
+            # traversal-free test (#315) — pinned wire bytes, mirroring the
+            # Rust skip_serializing_if.
+            if self.exists.joins:
+                serialized_exists["joins"] = [
+                    join.to_ir_dict() for join in self.exists.joins
+                ]
+            return serialized_exists
         if not self.is_compound:
             serialized = _serialize_query_value(self.value)
             return {
@@ -250,10 +265,18 @@ class ExistsTest:
     bare test; nesting and negation come free from :class:`QueryNode`
     recursion. There is no negation flag: NOT EXISTS is ``~`` (a ``not``
     node) over the exists node, like every other predicate (ADR-0008).
+
+    ``joins`` carries the inner tree's forward-traversal hop facts (#315,
+    ``QueryJoin`` entries, always ``"inner"`` — ADR-0006 semantics inside the
+    subquery), serialized only when non-empty. ``owner`` is the model whose
+    proxy built this test — a compile-side scope tag for the cross-scope
+    guard, never serialized.
     """
 
     hops: tuple[Any, ...]
     where: tuple["QueryNode", ...]
+    joins: tuple[Any, ...] = ()
+    owner: type | None = None
 
 
 def _query_value_kind(value: Any) -> str:
@@ -399,29 +422,29 @@ def __eq__(  # type: ignore[override]  # ty: ignore[invalid-method-override]
         self, other: "TField | FieldProxy[TField]"
     ) -> QueryNode:
         """Build an equality comparison node"""
-        return QueryNode(self.column, "==", other, path=self.path)
+        return QueryNode(self.column, "==", other, path=self.path, owner=self._owner)
 
     def __ne__(  # type: ignore[override]  # ty: ignore[invalid-method-override]
         self, other: "TField | FieldProxy[TField]"
     ) -> QueryNode:
         """Build an inequality comparison node"""
-        return QueryNode(self.column, "!=", other, path=self.path)
+        return QueryNode(self.column, "!=", other, path=self.path, owner=self._owner)
 
     def __lt__(self, other: "TField | FieldProxy[TField]") -> QueryNode:
         """Build a less-than comparison node"""
-        return QueryNode(self.column, "<", other, path=self.path)
+        return QueryNode(self.column, "<", other, path=self.path, owner=self._owner)
 
     def __le__(self, other: "TField | FieldProxy[TField]") -> QueryNode:
         """Build a less-than-or-equal comparison node"""
-        return QueryNode(self.column, "<=", other, path=self.path)
+        return QueryNode(self.column, "<=", other, path=self.path, owner=self._owner)
 
     def __gt__(self, other: "TField | FieldProxy[TField]") -> QueryNode:
         """Build a greater-than comparison node"""
-        return QueryNode(self.column, ">", other, path=self.path)
+        return QueryNode(self.column, ">", other, path=self.path, owner=self._owner)
 
     def __ge__(self, other: "TField | FieldProxy[TField]") -> QueryNode:
         """Build a greater-than-or-equal comparison node"""
-        return QueryNode(self.column, ">=", other, path=self.path)
+        return QueryNode(self.column, ">=", other, path=self.path, owner=self._owner)
 
     def in_(
         self, other: "list[TField] | tuple[TField, ...] | set[TField]"
@@ -446,7 +469,9 @@ def in_(
             raise TypeError(
                 f"The 'in_' operator expects a list, tuple, or set, got {type(other).__name__}"
             )
-        return QueryNode(self.column, "IN", list(other), path=self.path)
+        return QueryNode(
+            self.column, "IN", list(other), path=self.path, owner=self._owner
+        )
 
     def like(self: "FieldProxy[str]", pattern: str) -> QueryNode:
         """Build a ``LIKE`` comparison node
@@ -466,7 +491,9 @@ def like(self: "FieldProxy[str]", pattern: str) -> QueryNode:
             >>> email_filter.operator
             'LIKE'
         """
-        return QueryNode(self.column, "LIKE", pattern, path=self.path)
+        return QueryNode(
+            self.column, "LIKE", pattern, path=self.path, owner=self._owner
+        )
 
     def __lshift__(
         self, other: "list[TField] | tuple[TField, ...] | set[TField]"
@@ -722,20 +749,24 @@ def _relation_name(self) -> str:
         """The last-hop relation field name (the one being compared)."""
         return self._path[-1]
 
-    def _last_hop_shadow_column(self) -> str:
-        """Shadow FK column of the LAST hop, resolved along the spec chain.
+    def _last_hop_shadow_column(self) -> tuple[str, type]:
+        """Shadow FK column of the LAST hop and the model declaring it.
 
-        For ``t.account`` this is ``account_id`` on the root table; for
-        ``t.account.owner`` it is ``owner_id`` on the hop-1 (account) table.
+        For ``t.account`` this is ``account_id`` on the root table (owner =
+        root model); for ``t.account.owner`` it is ``owner_id`` on the hop-1
+        (account) table. The owner rides the comparison node as a
+        compile-side scope fact (#315 cross-scope guard).
         """
         current = self._root_model
+        declaring = current
         spec = None
         for name in self._path:
             specs = getattr(current, "__ferro_relation_specs__", None) or {}
             spec = specs[name]
+            declaring = current
             current = spec.target
         assert spec is not None  # a RelationProxy always has ≥ 1 hop
-        return spec.shadow_column
+        return spec.shadow_column, declaring
 
     def _instance_comparison(self, other: object, operator: str) -> QueryNode:
         """Desugar ``== instance`` / ``== None`` to a shadow-FK leaf (#273).
@@ -751,11 +782,15 @@ def _instance_comparison(self, other: object, operator: str) -> QueryNode:
                 the relation-vs-scalar guardrail, suggesting a column compare.
         """
         relation = self._relation_name()
-        shadow = self._last_hop_shadow_column()
+        shadow, shadow_owner = self._last_hop_shadow_column()
         prefix_path = self._path[:-1]
         if other is None:
             return QueryNode(
-                column=shadow, operator=operator, value=None, path=prefix_path
+                column=shadow,
+                operator=operator,
+                value=None,
+                path=prefix_path,
+                owner=shadow_owner,
             )
         if isinstance(other, self._target):
             # Reuse the Task 3 PK resolver (loud on zero/multiple PKs); local
@@ -771,7 +806,11 @@ def _instance_comparison(self, other: object, operator: str) -> QueryNode:
                     "save it first"
                 )
             return QueryNode(
-                column=shadow, operator=operator, value=pk_value, path=prefix_path
+                column=shadow,
+                operator=operator,
+                value=pk_value,
+                path=prefix_path,
+                owner=shadow_owner,
             )
         raise TypeError(
             f"cannot compare relation {relation!r} to {other!r}: expected a "
@@ -826,6 +865,150 @@ def __repr__(self) -> str:
         return f"RelationProxy(path={joined!r}, target={self._target.__name__!r})"
 
 
+def _resolve_scoped_predicate(
+    inner: "Predicate[Any]", model_cls: type, relation: str
+) -> QueryNode:
+    """Evaluate an existence test's inner lambda over the related model (#315).
+
+    The inner predicate is ordinary ferro — the same validating
+    :class:`QueryProxy` a root ``where()`` receives, constructed for the
+    related model — so every operator, combinator, traversal, and nested
+    existence test works unchanged. The rejection shapes mirror
+    ``where()``'s own.
+
+    Raises:
+        TypeError: If ``inner`` is not callable, returns a bare relation or
+            reverse relation, an aggregate, or any other non-predicate value.
+    """
+    if not callable(inner):
+        raise TypeError(
+            f"t.{relation}.exists(...) expected a predicate callable over "
+            f"{model_cls.__name__} (e.g. `lambda l: l.amount < 0`), got "
+            f"{type(inner).__name__}"
+        )
+    result = inner(QueryProxy(model_cls))
+    if isinstance(result, RelationProxy):
+        bare = result._path[-1]
+        raise TypeError(
+            f"t.{relation}.exists(...) inner predicate returned the bare "
+            f"relation {bare!r}; compare a column (e.g. l.{bare}.<column> == "
+            "...) or use == None / == an instance."
+        )
+    if isinstance(result, ReverseRelationProxy):
+        bare = result._name
+        raise TypeError(
+            f"t.{relation}.exists(...) inner predicate returned the bare "
+            f"reverse relation {bare!r}; test membership with "
+            f"l.{bare}.exists()."
+        )
+    if isinstance(result, AggregateExpr):
+        raise TypeError(
+            f"t.{relation}.exists(...) cannot filter on the aggregate "
+            f"{result._dotted()}: an existence test answers membership, not "
+            "aggregation."
+        )
+    if not isinstance(result, QueryNode):
+        raise TypeError(
+            f"t.{relation}.exists(...) inner callable must return a "
+            f"predicate (QueryNode), got {type(result).__name__}"
+        )
+    return result
+
+
+def _validate_inner_scope(node: QueryNode, model_cls: type, relation: str) -> None:
+    """Reject cross-scope references in an inner condition tree (#315).
+
+    The inner lambda may reference only its own parameter's scope. A leaf
+    built from any other proxy (the outer lambda's parameter), a
+    ``FieldProxy`` as a comparison right-hand side (column-to-column), and a
+    nested existence test built from another scope's proxy are all
+    build-time errors pointing at the deferred capability (#309) — silent
+    misrendering is the failure mode this guard exists to prevent.
+
+    Detection reads compile-side facts the proxies stamp on their nodes: a
+    leaf's ``owner`` (the model whose proxy built it) checked against the
+    model its ``path`` resolves to FROM THE INNER SCOPE, and an exists
+    node's ``owner`` (the scope whose proxy built the test).
+    """
+    if node.child is not None:
+        _validate_inner_scope(node.child, model_cls, relation)
+        return
+    if node.exists is not None:
+        if node.exists.owner is not None and node.exists.owner is not model_cls:
+            raise TypeError(
+                f"t.{relation}.exists(...) inner predicate contains an "
+                "existence test built from another scope's parameter "
+                f"(over {node.exists.owner.__name__}, not "
+                f"{model_cls.__name__}). Cross-scope references inside an "
+                "existence test are not supported yet (#309) — the inner "
+                "lambda may reference only its own parameter."
+            )
+        # Its own inner tree was validated against its own scope when built.
+        return
+    if node.is_compound:
+        if node.left is not None:
+            _validate_inner_scope(node.left, model_cls, relation)
+        if node.right is not None:
+            _validate_inner_scope(node.right, model_cls, relation)
+        return
+    if isinstance(
+        node.value,
+        (FieldProxy, AggregateExpr, QueryProxy, RelationProxy, ReverseRelationProxy),
+    ):
+        raise TypeError(
+            f"t.{relation}.exists(...) inner predicate compares "
+            f"{node.column!r} against another column reference; "
+            "column-to-column comparison is cross-scope correlation, not "
+            "supported yet (#309) — compare against a value."
+        )
+    expected = model_cls
+    for hop_name in node.path:
+        specs = getattr(expected, "__ferro_relation_specs__", None) or {}
+        spec = specs.get(hop_name)
+        if spec is None:
+            raise TypeError(
+                f"t.{relation}.exists(...) inner predicate leaf "
+                f"{node.column!r} traverses {hop_name!r}, which is not a "
+                f"relation of {expected.__name__} — the leaf was built from "
+                "another scope's parameter. Cross-scope references are not "
+                "supported yet (#309)."
+            )
+        expected = spec.target
+    if node.owner is not None and node.owner is not expected:
+        raise TypeError(
+            f"t.{relation}.exists(...) inner predicate leaf {node.column!r} "
+            f"belongs to {node.owner.__name__}, not the inner scope "
+            f"({expected.__name__}) — it was built from another lambda's "
+            "parameter. Cross-scope correlation is not supported yet "
+            "(#309); the inner lambda may reference only its own parameter."
+        )
+
+
+def _collect_inner_traversal_paths(
+    node: QueryNode, paths: dict[tuple[str, ...], None]
+) -> None:
+    """Collect an inner tree's forward-traversal paths in first-use order.
+
+    The same walk as the builder's ``_register_join_paths`` but scoped to
+    one existence test: full paths only (the Rust render dedups shared
+    prefixes), and nested exists nodes are skipped — their traversal facts
+    ride their own ``joins`` section.
+    """
+    if node.child is not None:
+        _collect_inner_traversal_paths(node.child, paths)
+        return
+    if node.exists is not None:
+        return
+    if node.is_compound:
+        if node.left is not None:
+            _collect_inner_traversal_paths(node.left, paths)
+        if node.right is not None:
+            _collect_inner_traversal_paths(node.right, paths)
+        return
+    if node.path:
+        paths.setdefault(tuple(node.path))
+
+
 class ReverseRelationProxy:
     """Predicate proxy for a reverse (BackRef) relation (#314, ADR-0007).
 
@@ -845,25 +1028,38 @@ def __init__(self, root_model: type, name: str, spec: Any) -> None:
         self._name = name
         self._spec = spec
 
-    def exists(self) -> QueryNode:
+    def exists(self, inner: "Predicate[Any] | None" = None) -> QueryNode:
         """Build the existence test: a correlated EXISTS over the child rows.
 
         Always a correlated EXISTS at every cardinality (a one-to-one BackRef
         renders identically to a to-many one); the result stays root-shaped,
         so the node composes with any other predicate, ordering, and paging.
 
+        Args:
+            inner: Optional scoping predicate — a full ferro predicate over
+                the related model (#315): every operator, ``&``/``|``/``~``,
+                forward traversal (joins rendered INSIDE the subquery,
+                ADR-0006 unchanged), and nested existence tests. It may
+                reference only its own parameter's scope; cross-scope
+                references raise at build time (#309).
+
         Returns:
             An exists :class:`QueryNode` carrying the one-hop correlation
-            path (child table, child FK column against the root PK).
+            path (child table, child FK column against the root PK) and,
+            when scoped, the inner condition tree plus its forward-traversal
+            join facts.
 
         Raises:
             ValueError: If the root model declares no primary-key column —
                 the EXISTS correlates child FK to root PK, so a PK-less root
                 is a loud error, never a guess.
+            TypeError: If ``inner`` is not a predicate callable, does not
+                return a predicate, or references a scope other than its own
+                parameter (cross-scope, #309).
         """
         # Late import: wire.py imports this module (nodes owns the predicate
         # shape, wire owns the hop-fact shape).
-        from .wire import QueryJoinHop
+        from .wire import QueryJoin, QueryJoinHop, resolve_join_hops
 
         root_pk = getattr(self._root_model, "__ferro_pk__", None)
         if root_pk is None:
@@ -880,7 +1076,26 @@ def exists(self) -> QueryNode:
             to_column=self._spec.child_fk_column,
             target=self._spec.target,
         )
-        return QueryNode(exists=ExistsTest(hops=(hop,), where=()))
+        where: tuple[QueryNode, ...] = ()
+        joins: tuple[Any, ...] = ()
+        if inner is not None:
+            node = _resolve_scoped_predicate(inner, self._spec.target, self._name)
+            _validate_inner_scope(node, self._spec.target, self._name)
+            paths: dict[tuple[str, ...], None] = {}
+            _collect_inner_traversal_paths(node, paths)
+            joins = tuple(
+                QueryJoin(
+                    join_type="inner",
+                    path=resolve_join_hops(self._spec.target, path),
+                )
+                for path in paths
+            )
+            where = (node,)
+        return QueryNode(
+            exists=ExistsTest(
+                hops=(hop,), where=where, joins=joins, owner=self._root_model
+            )
+        )
 
     def _reject_operator(self, symbol: str) -> NoReturn:
         raise TypeError(
diff --git a/src/operations.rs b/src/operations.rs
index 5cc2fe2..effaf1a 100644
--- a/src/operations.rs
+++ b/src/operations.rs
@@ -372,8 +372,20 @@ async fn populate_hop_bind_context(
     exec: Executor<'_>,
     backend: Dialect,
 ) -> PyResult<()> {
-    for edge in &join_plan.renders {
-        let table = &edge.to_table;
+    // Every table an existence test reaches (#314/#315) — correlation hops
+    // plus inner-traversal join hops, at any nesting depth — binds inner
+    // leaves against its own model, so it needs the same registration + enum
+    // catalog resolution as a rendered join edge. Collected up front (the
+    // walk borrows the plan immutably; the inserts below mutate it).
+    let mut tables: Vec<String> = join_plan
+        .renders
+        .iter()
+        .map(|edge| edge.to_table.clone())
+        .collect();
+    for node in &plan.where_clause {
+        collect_exists_tables(node, &mut tables);
+    }
+    for table in &tables {
         if !plan.hop_registrations.contains_key(table) {
             let registration = crate::state::registration_for_table(table)?;
             plan.hop_registrations.insert(table.clone(), registration);
@@ -387,6 +399,39 @@ async fn populate_hop_bind_context(
     Ok(())
 }
 
+/// Collect every table an existence test in `node`'s subtree touches:
+/// correlation `hops`, inner-traversal `joins` hops, and nested tests
+/// recursively (#315). Leaves contribute nothing — their tables are the
+/// scope tables these very entries establish.
+fn collect_exists_tables(node: &ferro_schema_ir::QueryNode, tables: &mut Vec<String>) {
+    use ferro_schema_ir::QueryNode;
+    match node {
+        QueryNode::Leaf { .. } => {}
+        QueryNode::Compound { left, right, .. } => {
+            collect_exists_tables(left, tables);
+            collect_exists_tables(right, tables);
+        }
+        QueryNode::Not { child } => collect_exists_tables(child, tables),
+        QueryNode::Exists {
+            hops,
+            where_clause,
+            joins,
+        } => {
+            for hop in hops {
+                tables.push(hop.to_table.clone());
+            }
+            for join in joins {
+                for hop in &join.path {
+                    tables.push(hop.to_table.clone());
+                }
+            }
+            for inner in where_clause {
+                collect_exists_tables(inner, tables);
+            }
+        }
+    }
+}
+
 /// Reject relation traversal on a mutating operation (#270 renders joins in SELECT
 /// only). The Python builder never emits joins on a mutating payload, so a `joins`
 /// list or a path-carrying WHERE leaf here is misuse — fail loud.
diff --git a/src/query.rs b/src/query.rs
index 379e0f4..1d9c44e 100644
--- a/src/query.rs
+++ b/src/query.rs
@@ -63,11 +63,17 @@ enum ColumnQualifier<'a> {
         root_table: &'a str,
         join_plan: &'a JoinPlan,
     },
-    /// Inside an EXISTS subquery (#314, ADR-0007): an empty-path leaf
+    /// Inside an EXISTS subquery (#314/#315, ADR-0007): an empty-path leaf
     /// qualifies by the subquery scope's `alias` and binds against `table`'s
-    /// model. The variant owns its strings — subquery aliases are minted
-    /// during the render walk, not borrowed from the plan.
-    ExistsScope { alias: String, table: String },
+    /// model; a path-carrying leaf resolves through the subquery's own
+    /// `join_plan` (the exists node's `joins` section, rendered inside the
+    /// subquery). The variant owns its data — subquery aliases are minted
+    /// during the render walk, not borrowed from the statement plan.
+    ExistsScope {
+        alias: String,
+        table: String,
+        join_plan: JoinPlan,
+    },
 }
 
 /// Qualify a column reference (a WHERE leaf or an `ORDER BY` term) by its
@@ -286,15 +292,22 @@ fn qualify_leaf_column(
             })?;
             Ok(Expr::col((Alias::new(alias.as_str()), Alias::new(column))))
         }
-        ColumnQualifier::ExistsScope { alias, .. } => {
-            if !path.is_empty() {
-                return Err(format!(
-                    "WHERE column {column:?} carries relation path {path:?} \
-                     inside an existence test; forward traversal inside the \
-                     subquery is not rendered yet (#315)"
-                ));
+        ColumnQualifier::ExistsScope {
+            alias, join_plan, ..
+        } => {
+            if path.is_empty() {
+                return Ok(Expr::col((Alias::new(alias.as_str()), Alias::new(column))));
             }
-            Ok(Expr::col((Alias::new(alias.as_str()), Alias::new(column))))
+            let hop_alias = join_plan.prefix_alias.get(path).ok_or_else(|| {
+                format!(
+                    "column {column:?} carries relation path {path:?} with no \
+                     matching join entry inside the existence test"
+                )
+            })?;
+            Ok(Expr::col((
+                Alias::new(hop_alias.as_str()),
+                Alias::new(column),
+            )))
         }
     }
 }
@@ -599,9 +612,13 @@ impl QueryPlan {
                     .node_to_condition_for_backend(child, backend, qualifier, exists_counter)?
                     .not())
             }
-            QueryNode::Exists { hops, where_clause } => {
-                // Existence test (#314, ADR-0007): one render loop for every
-                // hop count. The first hop's table is the subquery FROM,
+            QueryNode::Exists {
+                hops,
+                where_clause,
+                joins,
+            } => {
+                // Existence test (#314/#315, ADR-0007): one render loop for
+                // every hop count. The first hop's table is the subquery FROM,
                 // correlated to the enclosing scope's alias; remaining hops
                 // (M2M target) render as inner joins inside the subquery; the
                 // inner tree recurses through this same builder scoped to the
@@ -644,9 +661,51 @@ impl QueryPlan {
                     last_alias = hop_alias;
                     last_table = hop.to_table.clone();
                 }
+                // Inner forward-traversal joins (#315): rendered INSIDE the
+                // subquery, rooted at the inner scope, always INNER
+                // (ADR-0006 traversal semantics; there is no inner-lambda
+                // left_join spelling). Shared prefixes across entries dedup
+                // to one edge; aliases come from the same statement-wide
+                // counter, so they can never collide with any other scope.
+                let mut inner_plan = JoinPlan::default();
+                for join in joins {
+                    if join.join_type != "inner" {
+                        return Err(format!(
+                            "unsupported join_type {:?} inside an existence test; \
+                             traversal inside the subquery is always \"inner\"",
+                            join.join_type
+                        ));
+                    }
+                    let mut prefix: Vec<String> = Vec::new();
+                    let mut prev_alias = last_alias.clone();
+                    for hop in &join.path {
+                        prefix.push(hop.relation.clone());
+                        if let Some(existing) = inner_plan.prefix_alias.get(&prefix) {
+                            prev_alias = existing.clone();
+                            continue;
+                        }
+                        *exists_counter += 1;
+                        let hop_alias = format!("x{}_{}", exists_counter, hop.relation);
+                        subquery.join_as(
+                            JoinType::InnerJoin,
+                            Alias::new(&hop.to_table),
+                            Alias::new(&hop_alias),
+                            Expr::col((Alias::new(&prev_alias), Alias::new(&hop.from_column)))
+                                .equals((Alias::new(&hop_alias), Alias::new(&hop.to_column))),
+                        );
+                        inner_plan
+                            .prefix_alias
+                            .insert(prefix.clone(), hop_alias.clone());
+                        inner_plan
+                            .prefix_table
+                            .insert(prefix.clone(), hop.to_table.clone());
+                        prev_alias = hop_alias;
+                    }
+                }
                 let inner_scope = ColumnQualifier::ExistsScope {
                     alias: last_alias,
                     table: last_table,
+                    join_plan: inner_plan,
                 };
                 for node in where_clause {
                     inner = inner.add(self.node_to_condition_for_backend(
@@ -791,23 +850,35 @@ impl QueryPlan {
         qualifier: &ColumnQualifier<'_>,
         path: &[String],
     ) -> Result<(Option<&crate::codec_plan::ModelCodecPlan>, &HashMap<String, String>), String> {
-        // Inside an EXISTS subquery (#314) an empty-path leaf belongs to the
-        // subquery scope's table, NEVER the root model — resolve before the
-        // root early-return below. Registration is optional here (mirroring
-        // the root's own `Option`): an inner leaf on an unregistered/
-        // unprobed table degrades to generic binds, exactly like an
-        // unregistered root model. The walkers populate both maps for scoped
-        // tests (#315).
-        if let ColumnQualifier::ExistsScope { table, .. } = qualifier {
-            if !path.is_empty() {
-                return Err(format!(
-                    "relation path {path:?} inside an existence test is not \
-                     rendered yet (#315); cannot resolve typed binds"
-                ));
-            }
+        // Inside an EXISTS subquery (#314/#315) a leaf belongs to the
+        // subquery scope — the scope table for an empty path, the inner
+        // join plan's hop table for a traversed one — NEVER the root model;
+        // resolve before the root early-return below. Registration is
+        // optional here (mirroring the root's own `Option`): an inner leaf
+        // on an unregistered/unprobed table degrades to generic binds,
+        // exactly like an unregistered root model. The walkers populate
+        // both maps for every table an existence test reaches.
+        if let ColumnQualifier::ExistsScope {
+            table, join_plan, ..
+        } = qualifier
+        {
+            let leaf_table = if path.is_empty() {
+                table
+            } else {
+                join_plan.prefix_table.get(path).ok_or_else(|| {
+                    format!(
+                        "relation path {path:?} has no matching join entry \
+                         inside the existence test for typed binds"
+                    )
+                })?
+            };
             return Ok((
-                self.hop_registrations.get(table).map(|m| &m.codec_plan),
-                self.hop_enum_udt.get(table).unwrap_or(empty_enum_udt()),
+                self.hop_registrations
+                    .get(leaf_table)
+                    .map(|m| &m.codec_plan),
+                self.hop_enum_udt
+                    .get(leaf_table)
+                    .unwrap_or(empty_enum_udt()),
             ));
         }
         if path.is_empty() {
@@ -1104,6 +1175,99 @@ mod tests {
         assert!(sql.contains(" or "), "OR composition preserved: {sql}");
     }
 
+    #[test]
+    fn scoped_exists_renders_inner_conditions_and_traversal_joins() {
+        // Scoped existence test (#315): the inner tree's empty-path leaf
+        // qualifies by the subquery scope's alias, and a traversed inner
+        // leaf resolves through the exists node's own `joins` section —
+        // rendered as an INNER join inside the subquery (ADR-0006).
+        let payload: ferro_schema_ir::QueryIrPayload = serde_json::from_value(json!({
+            "model_name": "Acct",
+            "where": [
+                {"node_kind": "exists",
+                 "hops": [{"relation": "transactions", "from_column": "id",
+                           "to_table": "transaction", "to_column": "account_id"}],
+                 "where": [
+                    {"node_kind": "compound", "operator": "AND",
+                     "left": {"node_kind": "leaf", "column": "amount", "operator": ">=",
+                              "value": {"kind": "int", "value": 100}, "path": []},
+                     "right": {"node_kind": "leaf", "column": "name", "operator": "==",
+                               "value": {"kind": "string", "value": "checking"},
+                               "path": ["account"]}}
+                 ],
+                 "joins": [
+                    {"join_type": "inner",
+                     "path": [{"relation": "account", "from_column": "account_id",
+                               "to_table": "account", "to_column": "id"}]}
+                 ]}
+            ],
+            "order_by": [],
+            "limit": null, "offset": null, "m2m": null, "materialization": {"kind": "root_instances"}, "joins": []
+        }))
+        .expect("payload deserializes");
+        let plan = QueryPlan::from_ir_payload(payload).expect("plan builds");
+        let mut select = Query::select();
+        select.from(Alias::new("acct")).cond_where(
+            plan.to_condition_for_backend(Dialect::Sqlite, Some("acct"))
+                .expect("valid test query"),
+        );
+        let sql = select.to_string(SqliteQueryBuilder).to_lowercase();
+        assert!(
+            sql.contains(
+                "inner join \"account\" as \"x2_account\" on \
+                 \"x1_transactions\".\"account_id\" = \"x2_account\".\"id\""
+            ),
+            "inner traversal join must render INSIDE the subquery: {sql}"
+        );
+        assert!(
+            sql.contains("\"x1_transactions\".\"amount\" >= 100"),
+            "empty-path inner leaf qualifies by the subquery scope alias: {sql}"
+        );
+        assert!(
+            sql.contains("\"x2_account\".\"name\" = 'checking'"),
+            "traversed inner leaf qualifies by its inner join alias: {sql}"
+        );
+    }
+
+    #[test]
+    fn nested_exists_correlates_to_the_inner_scope() {
+        // Depth-2 nesting (#315): the inner exists node's correlation hop
+        // resolves against the ENCLOSING SUBQUERY's alias, not the root —
+        // recursion through the ExistsScope qualifier, no second mechanism.
+        let payload: ferro_schema_ir::QueryIrPayload = serde_json::from_value(json!({
+            "model_name": "Owner",
+            "where": [
+                {"node_kind": "exists",
+                 "hops": [{"relation": "accounts", "from_column": "id",
+                           "to_table": "account", "to_column": "owner_id"}],
+                 "where": [
+                    {"node_kind": "exists",
+                     "hops": [{"relation": "transactions", "from_column": "id",
+                               "to_table": "transaction", "to_column": "account_id"}],
+                     "where": []}
+                 ]}
+            ],
+            "order_by": [],
+            "limit": null, "offset": null, "m2m": null, "materialization": {"kind": "root_instances"}, "joins": []
+        }))
+        .expect("payload deserializes");
+        let plan = QueryPlan::from_ir_payload(payload).expect("plan builds");
+        let mut select = Query::select();
+        select.from(Alias::new("owner")).cond_where(
+            plan.to_condition_for_backend(Dialect::Postgres, Some("owner"))
+                .expect("valid test query"),
+        );
+        let sql = select.to_string(PostgresQueryBuilder).to_lowercase();
+        assert!(
+            sql.contains("\"x1_accounts\".\"owner_id\" = \"owner\".\"id\""),
+            "outer test correlates to the root alias: {sql}"
+        );
+        assert!(
+            sql.contains("\"x2_transactions\".\"account_id\" = \"x1_accounts\".\"id\""),
+            "nested test correlates to the enclosing SUBQUERY alias: {sql}"
+        );
+    }
+
     #[test]
     fn exists_requires_a_qualified_enclosing_scope() {
         // Correlation needs an alias to correlate against; the unqualified
diff --git a/tests/fixtures/ir_vectors/query_account_scoped_exists_v7.json b/tests/fixtures/ir_vectors/query_account_scoped_exists_v7.json
new file mode 100644
index 0000000..acec1b4
--- /dev/null
+++ b/tests/fixtures/ir_vectors/query_account_scoped_exists_v7.json
@@ -0,0 +1,74 @@
+{
+  "vector_name": "query_account_scoped_exists_v7",
+  "domain": "query",
+  "expect_valid": true,
+  "ir": {
+    "ir_kind": "query",
+    "ir_version": 7,
+    "payload": {
+      "model_name": "Account",
+      "where": [
+        {
+          "node_kind": "exists",
+          "hops": [
+            {
+              "relation": "transactions",
+              "from_column": "id",
+              "to_table": "transaction",
+              "to_column": "account_id"
+            }
+          ],
+          "where": [
+            {
+              "node_kind": "compound",
+              "operator": "AND",
+              "left": {
+                "node_kind": "leaf",
+                "column": "amount",
+                "operator": ">=",
+                "value": {
+                  "kind": "int",
+                  "value": 100
+                },
+                "path": []
+              },
+              "right": {
+                "node_kind": "leaf",
+                "column": "name",
+                "operator": "==",
+                "value": {
+                  "kind": "string",
+                  "value": "checking"
+                },
+                "path": [
+                  "account"
+                ]
+              }
+            }
+          ],
+          "joins": [
+            {
+              "join_type": "inner",
+              "path": [
+                {
+                  "relation": "account",
+                  "from_column": "account_id",
+                  "to_table": "account",
+                  "to_column": "id"
+                }
+              ]
+            }
+          ]
+        }
+      ],
+      "order_by": [],
+      "limit": null,
+      "offset": null,
+      "m2m": null,
+      "joins": [],
+      "materialization": {
+        "kind": "root_instances"
+      }
+    }
+  }
+}
diff --git a/tests/fixtures/ir_vectors/query_owner_nested_exists_v7.json b/tests/fixtures/ir_vectors/query_owner_nested_exists_v7.json
new file mode 100644
index 0000000..cfb1b00
--- /dev/null
+++ b/tests/fixtures/ir_vectors/query_owner_nested_exists_v7.json
@@ -0,0 +1,47 @@
+{
+  "vector_name": "query_owner_nested_exists_v7",
+  "domain": "query",
+  "expect_valid": true,
+  "ir": {
+    "ir_kind": "query",
+    "ir_version": 7,
+    "payload": {
+      "model_name": "Owner",
+      "where": [
+        {
+          "node_kind": "exists",
+          "hops": [
+            {
+              "relation": "accounts",
+              "from_column": "id",
+              "to_table": "account",
+              "to_column": "owner_id"
+            }
+          ],
+          "where": [
+            {
+              "node_kind": "exists",
+              "hops": [
+                {
+                  "relation": "transactions",
+                  "from_column": "id",
+                  "to_table": "transaction",
+                  "to_column": "account_id"
+                }
+              ],
+              "where": []
+            }
+          ]
+        }
+      ],
+      "order_by": [],
+      "limit": null,
+      "offset": null,
+      "m2m": null,
+      "joins": [],
+      "materialization": {
+        "kind": "root_instances"
+      }
+    }
+  }
+}
diff --git a/tests/test_ir_vectors_contract.py b/tests/test_ir_vectors_contract.py
index a74bfb8..04f4c4a 100644
--- a/tests/test_ir_vectors_contract.py
+++ b/tests/test_ir_vectors_contract.py
@@ -56,7 +56,10 @@ def _validate_query_node(node: dict[str, Any], label: str) -> None:
         # v7 (ADR-0007): an existence test carries a correlation hop path (1
         # hop for a reverse FK, 2 for M2M — the `joins`-section hop shape) and
         # an ordinary inner condition tree (empty = bare test). No negation
-        # flag: NOT EXISTS is the `not` node over this one.
+        # flag: NOT EXISTS is the `not` node over this one. When the inner
+        # tree traverses forward FKs (#315), the hop facts ride the node's
+        # own `joins` section — present only when non-empty (absent on a
+        # traversal-free test, pinned wire bytes).
         _require_keys(node, {"hops", "where"}, label)
         hops = node["hops"]
         assert isinstance(hops, list) and hops, (
@@ -69,6 +72,24 @@ def _validate_query_node(node: dict[str, Any], label: str) -> None:
         for i, child in enumerate(inner):
             assert isinstance(child, dict), f"{label}.where[{i}] must be object"
             _validate_query_node(child, f"{label}.where[{i}]")
+        if "joins" in node:
+            joins = node["joins"]
+            assert isinstance(joins, list) and joins, (
+                f"{label}.joins must be a non-empty list when present"
+            )
+            for j, join in enumerate(joins):
+                join_label = f"{label}.joins[{j}]"
+                assert isinstance(join, dict), f"{join_label} must be object"
+                _require_keys(join, {"join_type", "path"}, join_label)
+                assert join["join_type"] == "inner", (
+                    f"{join_label}.join_type must be \"inner\" (ADR-0006: "
+                    "traversal inside a subquery narrows)"
+                )
+                assert isinstance(join["path"], list) and join["path"], (
+                    f"{join_label}.path must be a non-empty list"
+                )
+                for h, hop in enumerate(join["path"]):
+                    _validate_hop(hop, f"{join_label}.path[{h}]")
         return
 
     _require_keys(node, {"operator"}, label)
diff --git a/tests/test_query_exists.py b/tests/test_query_exists.py
index ca9c03a..f2a939b 100644
--- a/tests/test_query_exists.py
+++ b/tests/test_query_exists.py
@@ -1,14 +1,17 @@
-"""End-to-end behavior of existence tests on reverse relations (#314, ADR-0007).
+"""End-to-end behavior of existence tests on reverse relations (ADR-0007).
 
 A reverse (BackRef) relation appears in a predicate in exactly one form — the
-existence test ``t.rel.exists()`` — rendered as a correlated EXISTS at every
-cardinality (one-to-one BackRefs included), negated with ``~``. These tests
-assert result sets only, on both database backends via the backend matrix;
-the wire shape is pinned separately by the ``exists`` golden vectors.
-
-The model graph mirrors the #307 workload: a transaction with two one-to-one
-BackRefs into a transfer link row (membership via either FK column) plus a
-to-many BackRef onto split lines.
+existence test ``t.rel.exists(...)`` — rendered as a correlated EXISTS at
+every cardinality (one-to-one BackRefs included), negated with ``~``, and
+optionally scoped by a full ferro predicate over the related model (#315).
+These tests assert result sets only, on both database backends via the
+backend matrix; the wire shape is pinned separately by the ``exists`` golden
+vectors.
+
+The model graph mirrors the #307/#308 workloads: a transaction with two
+one-to-one BackRefs into a transfer link row (membership via either FK
+column), a to-many BackRef onto split lines, and a category the lines and
+transactions both point at (the line-aware category filter).
 """
 
 import pytest
@@ -19,9 +22,19 @@
 pytestmark = pytest.mark.backend_matrix
 
 
+class ExCat(Model):
+    id: Annotated[int | None, FerroField(primary_key=True)] = None
+    name: str = ""
+    txns: Relation[list["ExTxn"]] = BackRef()
+    lines: Relation[list["ExLine"]] = BackRef()
+
+
 class ExTxn(Model):
     id: Annotated[int | None, FerroField(primary_key=True)] = None
     amount: int = 0
+    category: Annotated[
+        ExCat | None, ForeignKey(related_name="txns", on_delete="SET NULL")
+    ] = None
     transfer_out: "ExTransfer" = BackRef()
     transfer_in: "ExTransfer" = BackRef()
     lines: Relation[list["ExLine"]] = BackRef()
@@ -40,7 +53,11 @@ class ExTransfer(Model):
 class ExLine(Model):
     id: Annotated[int | None, FerroField(primary_key=True)] = None
     txn: Annotated[ExTxn, ForeignKey(related_name="lines", on_delete="CASCADE")]
-    category: str = ""
+    category: Annotated[
+        ExCat | None, ForeignKey(related_name="lines", on_delete="SET NULL")
+    ] = None
+    amount: int = 0
+    memo: str = ""
 
 
 async def _seed_transfers() -> dict[str, ExTxn]:
@@ -108,7 +125,7 @@ async def test_to_many_exists_returns_each_root_once(db_url):
     async with engines.session():
         split = await ExTxn.create(amount=-70)
         for _ in range(3):
-            await ExLine.create(txn=split, category="groceries")
+            await ExLine.create(txn=split)
         await ExTxn.create(amount=-10)
 
         results = await ExTxn.where(lambda t: t.lines.exists()).all()
@@ -123,7 +140,7 @@ async def test_exists_composes_with_root_predicates_order_and_limit(db_url):
     async with engines.session():
         for amount in (-40, -50, -60):
             txn = await ExTxn.create(amount=amount)
-            await ExLine.create(txn=txn, category="travel")
+            await ExLine.create(txn=txn)
         await ExTxn.create(amount=-80)  # child-less
 
         results = (
@@ -216,3 +233,278 @@ async def test_exists_rejected_on_mutating_verbs(db_url):
             await ExTxn.where(lambda t: t.lines.exists()).update(amount=0)
         with pytest.raises(ValueError, match="delete"):
             await ExTxn.where(lambda t: t.lines.exists()).delete()
+
+
+# ---------------------------------------------------------------------------
+# Scoped inner predicates (#315): the optional inner lambda is a full ferro
+# predicate over the child model — every operator, forward traversal (joins
+# INSIDE the subquery, ADR-0006 unchanged), nesting — with cross-scope
+# references rejected at build time (deferred to #309).
+# ---------------------------------------------------------------------------
+
+
+async def _seed_categories() -> dict[str, object]:
+    """The #308 fixture: a split transaction whose three lines carry the
+    category (its own category vacated), a plain transaction categorized at
+    the root, and a transaction matching neither."""
+    groceries = await ExCat.create(name="Groceries")
+    travel = await ExCat.create(name="Travel")
+    split = await ExTxn.create(amount=-7000)  # category vacated while split
+    for amount in (-3000, -2000, -2000):
+        await ExLine.create(txn=split, category=groceries, amount=amount)
+    plain = await ExTxn.create(amount=-1000, category=groceries)
+    other = await ExTxn.create(amount=-500, category=travel)
+    return {
+        "groceries": groceries,
+        "travel": travel,
+        "split": split,
+        "plain": plain,
+        "other": other,
+    }
+
+
+@pytest.mark.asyncio
+async def test_scoped_exists_filters_by_child_predicate(db_url):
+    """``t.lines.exists(lambda line: ...)`` keeps exactly the roots with a
+    matching child row."""
+    await connect(db_url, auto_migrate=True)
+    async with engines.session():
+        seeded = await _seed_categories()
+
+        rows = await _amounts(
+            ExTxn.where(
+                lambda t: t.lines.exists(
+                    lambda line, cat=seeded["groceries"]: line.category_id == cat.id
+                )
+            )
+        )
+        assert rows == [-7000]
+
+
+@pytest.mark.asyncio
+async def test_308_line_aware_category_filter(db_url):
+    """The full #308 demo: root-or-line category membership. A three-line
+    split matches exactly once, the child-less root survives through the OR's
+    root branch, and keyset ``order_by`` + ``limit`` compose unchanged."""
+    await connect(db_url, auto_migrate=True)
+    async with engines.session():
+        seeded = await _seed_categories()
+        ids = [seeded["groceries"].id]
+
+        query = ExTxn.where(
+            lambda t: (
+                t.category_id.in_(ids)
+                | t.lines.exists(lambda line: line.category_id.in_(ids))
+            )
+        )
+        rows = await _amounts(query)
+        assert rows == [-7000, -1000]
+
+        paged = (
+            await ExTxn.where(
+                lambda t: (
+                    t.category_id.in_(ids)
+                    | t.lines.exists(lambda line: line.category_id.in_(ids))
+                )
+            )
+            .order_by("amount", "desc")
+            .limit(1)
+            .all()
+        )
+        assert [r.amount for r in paged] == [-1000]
+
+
+@pytest.mark.asyncio
+async def test_inner_lambda_supports_full_operator_set(db_url):
+    """Every operator and ``&``/``|``/``~`` composition works over the child
+    model — the inner predicate is ordinary ferro, not a sub-language."""
+    await connect(db_url, auto_migrate=True)
+    async with engines.session():
+        txn = await ExTxn.create(amount=-70)
+        await ExLine.create(txn=txn, amount=-30, memo="grocery run")
+        await ExLine.create(txn=txn, amount=-40, memo="fuel")
+        bare = await ExTxn.create(amount=-10)
+        await ExLine.create(txn=bare, amount=5, memo="refund")
+
+        assert await _amounts(
+            ExTxn.where(lambda t: t.lines.exists(lambda line: line.memo.like("%fuel%")))
+        ) == [-70]
+        assert await _amounts(
+            ExTxn.where(
+                lambda t: t.lines.exists(
+                    lambda line: (line.amount <= -30) & ~line.memo.like("%grocery%")
+                )
+            )
+        ) == [-70]
+        assert await _amounts(
+            ExTxn.where(
+                lambda t: t.lines.exists(
+                    lambda line: (line.amount > 0) | (line.amount < -35)
+                )
+            )
+        ) == [-70, -10]
+
+
+@pytest.mark.asyncio
+async def test_forward_traversal_inside_subquery(db_url):
+    """A forward-FK traversal inside the inner lambda renders its join INSIDE
+    the EXISTS subquery under unchanged ADR-0006 semantics (INNER,
+    narrowing)."""
+    await connect(db_url, auto_migrate=True)
+    async with engines.session():
+        await _seed_categories()
+
+        rows = await _amounts(
+            ExTxn.where(
+                lambda t: t.lines.exists(lambda line: line.category.name == "Groceries")
+            )
+        )
+        assert rows == [-7000]
+
+        # INNER semantics: a line with no category can never match a
+        # traversed inner predicate.
+        uncategorized = await ExTxn.create(amount=-42)
+        await ExLine.create(txn=uncategorized, amount=-42)
+        rows = await _amounts(
+            ExTxn.where(
+                lambda t: t.lines.exists(lambda line: line.category.name != "nope")
+            )
+        )
+        assert rows == [-7000]
+
+
+@pytest.mark.asyncio
+async def test_nested_exists_depth_two(db_url):
+    """Existence tests nest: categories with a transaction that has a
+    negative line — the exists node's inner tree is an ordinary condition
+    tree, so recursion is free."""
+    await connect(db_url, auto_migrate=True)
+    async with engines.session():
+        cat = await ExCat.create(name="Active")
+        idle = await ExCat.create(name="Idle")
+        txn = await ExTxn.create(amount=-100, category=cat)
+        await ExLine.create(txn=txn, amount=-60)
+        pos = await ExTxn.create(amount=200, category=idle)
+        await ExLine.create(txn=pos, amount=200)
+
+        results = await ExCat.where(
+            lambda c: c.txns.exists(
+                lambda t: t.lines.exists(lambda line: line.amount < 0)
+            )
+        ).all()
+        assert [r.name for r in results] == ["Active"]
+
+
+@pytest.mark.asyncio
+async def test_explicit_grouping_contrast(db_url):
+    """``exists(lambda line: A & B)`` (one child row matches both) and
+    ``exists(A-test) & exists(B-test)`` (some child row matches each) are
+    different, correct row sets — the ambiguity ADR-0007 rejects is
+    unspellable by construction."""
+    await connect(db_url, auto_migrate=True)
+    async with engines.session():
+        both_in_one = await ExTxn.create(amount=-10)
+        await ExLine.create(txn=both_in_one, amount=-50, memo="fuel")
+        spread = await ExTxn.create(amount=-20)
+        await ExLine.create(txn=spread, amount=-50, memo="snacks")
+        await ExLine.create(txn=spread, amount=-5, memo="fuel")
+
+        one_row_matches_both = await _amounts(
+            ExTxn.where(
+                lambda t: t.lines.exists(
+                    lambda line: (line.amount <= -50) & line.memo.like("%fuel%")
+                )
+            )
+        )
+        assert one_row_matches_both == [-10]
+
+        some_row_matches_each = await _amounts(
+            ExTxn.where(
+                lambda t: (
+                    t.lines.exists(lambda line: line.amount <= -50)
+                    & t.lines.exists(lambda line: line.memo.like("%fuel%"))
+                )
+            )
+        )
+        assert some_row_matches_each == [-20, -10]
+
+
+@pytest.mark.asyncio
+async def test_negated_scoped_exists(db_url):
+    """``~t.lines.exists(lambda line: ...)`` renders NOT EXISTS over the scoped
+    subquery."""
+    await connect(db_url, auto_migrate=True)
+    async with engines.session():
+        seeded = await _seed_categories()
+        ids = [seeded["groceries"].id]
+
+        rows = await _amounts(
+            ExTxn.where(
+                lambda t: ~t.lines.exists(lambda line: line.category_id.in_(ids))
+            )
+        )
+        assert rows == [-1000, -500]
+
+
+@pytest.mark.asyncio
+async def test_scoped_exists_count(db_url):
+    await connect(db_url, auto_migrate=True)
+    async with engines.session():
+        seeded = await _seed_categories()
+        ids = [seeded["groceries"].id]
+
+        n = await ExTxn.where(
+            lambda t: (
+                t.category_id.in_(ids)
+                | t.lines.exists(lambda line: line.category_id.in_(ids))
+            )
+        ).count()
+        assert n == 2
+
+
+# ---------------------------------------------------------------------------
+# Cross-scope guard (#315): the inner lambda may reference only its own
+# parameter's scope; everything else fails at build time pointing at the
+# deferred capability (#309). Silent misrendering is the failure mode this
+# guard exists to prevent.
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_cross_scope_outer_column_rejected(db_url):
+    """An inner-tree leaf built from the OUTER lambda's parameter is a
+    build-time error, not a silently re-scoped column."""
+    await connect(db_url, auto_migrate=True)
+    with pytest.raises(TypeError, match="#309"):
+        ExTxn.where(lambda t: t.lines.exists(lambda line: t.amount > 5))
+
+
+@pytest.mark.asyncio
+async def test_cross_scope_field_proxy_rhs_rejected(db_url):
+    """A FieldProxy as a comparison right-hand side (column-to-column) is a
+    build-time error pointing at #309 — whichever scope it came from."""
+    await connect(db_url, auto_migrate=True)
+    with pytest.raises(TypeError, match="#309"):
+        ExTxn.where(
+            lambda t: t.lines.exists(lambda line: line.category_id == t.category_id)
+        )
+    with pytest.raises(TypeError, match="#309"):
+        ExTxn.where(lambda t: t.lines.exists(lambda line: line.amount == line.amount))
+
+
+@pytest.mark.asyncio
+async def test_cross_scope_nested_exists_rejected(db_url):
+    """A nested existence test built from the OUTER proxy inside the inner
+    lambda is cross-scope too."""
+    await connect(db_url, auto_migrate=True)
+    with pytest.raises(TypeError, match="#309"):
+        ExTxn.where(lambda t: t.lines.exists(lambda line: t.transfer_out.exists()))
+
+
+@pytest.mark.asyncio
+async def test_inner_lambda_must_return_a_predicate(db_url):
+    """A non-predicate inner lambda fails with the same pointed shape as
+    where() itself."""
+    await connect(db_url, auto_migrate=True)
+    with pytest.raises(TypeError, match="predicate"):
+        ExTxn.where(lambda t: t.lines.exists(lambda line: line.amount))
diff --git a/tests/test_query_wire_vectors.py b/tests/test_query_wire_vectors.py
index 25485a2..d7b4822 100644
--- a/tests/test_query_wire_vectors.py
+++ b/tests/test_query_wire_vectors.py
@@ -225,12 +225,35 @@ def _q_not_exists(m: dict[str, type]) -> Any:
     return m["Owner"].where(lambda o: ~o.accounts.exists())
 
 
+def _q_scoped_exists(m: dict[str, type]) -> Any:
+    # Scoped existence test (#315): the inner lambda is a full ferro
+    # predicate over the child model. The traversed inner leaf
+    # (`t.account.name`) puts its hop facts on the exists node's own `joins`
+    # section — rendered INSIDE the subquery, never on the root query.
+    return m["Account"].where(
+        lambda a: a.transactions.exists(
+            lambda t: (t.amount >= 100) & (t.account.name == "checking")
+        )
+    )
+
+
+def _q_nested_exists(m: dict[str, type]) -> Any:
+    # Nested exists-in-exists (#315): the inner tree is an ordinary condition
+    # tree, so depth comes from recursion, not a second mechanism. The bare
+    # inner node carries no `joins` key at all (absent, not empty).
+    return m["Owner"].where(
+        lambda o: o.accounts.exists(lambda a: a.transactions.exists())
+    )
+
+
 CASES: list[tuple[str, Callable[[dict[str, type]], Any], str]] = [
     ("query_user_compound_v7", _q_user_compound, "User"),
     ("query_user_not_leaf_v7", _q_not_leaf, "User"),
     ("query_user_not_compound_v7", _q_not_compound, "User"),
     ("query_account_exists_v7", _q_exists_bare, "Account"),
     ("query_owner_not_exists_v7", _q_not_exists, "Owner"),
+    ("query_account_scoped_exists_v7", _q_scoped_exists, "Account"),
+    ("query_owner_nested_exists_v7", _q_nested_exists, "Owner"),
     ("query_transaction_traversal_v7", _q_traversal, "Transaction"),
     ("query_transaction_left_join_v7", _q_left_join, "Transaction"),
     ("query_transaction_include_v7", _q_include, "Transaction"),

From ee216a33d6d9b5cb3617b3a7c89f73c87fd31500 Mon Sep 17 00:00:00 2001
From: Taylor <taylor@x54.sh>
Date: Sat, 18 Jul 2026 11:21:39 -0400
Subject: [PATCH 2/3] =?UTF-8?q?feat(query):=20M2M=20existence=20tests=20?=
 =?UTF-8?q?=E2=80=94=20the=20two-hop=20correlation=20path?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

t.tags.exists(...) works on many-to-many relations, bare, scoped, and
~-negated, spelled and behaving exactly like the reverse-FK form
(ADR-0007: one verb at every cardinality and relation kind). The same
exists node carries a two-hop correlation path — join table first,
correlated to the enclosing scope, then the target — and the existing
Rust render loop covers it unchanged: no new render mechanism, proving
the design's claim that M2M is test surface, not a second mechanism.

- build_reverse_specs maps M2M descriptors to join-table specs; each
  side's descriptor already orients source_col/target_col for that
  side, so both the declaring and the related_name side spell
  identically.
- The proxy builds hops from the spec: join_table.source_col against
  the root PK, then target on join_table.target_col = target.pk. The
  inner lambda resolves over the M2M target with full predicate power
  (operators, combinators, negation, cross-scope guard) — the #315
  machinery, untouched.
- Hand-authored golden vector pins the 2-hop shape, asserted from the
  Python emitter and the Rust decoder; a Rust render test pins
  join-table-FROM + inner-join-target + last-hop-alias qualification.
- e2e on both backends: bare (each root exactly once regardless of how
  many join rows match), scoped, negated, both sides of the relation,
  root-predicate/count composition, and the proxy's single-verb error
  surfaces.

Refs #311, #316
---
 crates/ferro-schema-ir/src/lib.rs             |  33 ++++
 src/ferro/columns.py                          |  66 +++++---
 src/ferro/query/nodes.py                      |  63 ++++++--
 src/query.rs                                  |  53 +++++++
 .../ir_vectors/query_user_m2m_exists_v7.json  |  51 ++++++
 tests/test_query_exists.py                    | 148 +++++++++++++++++-
 tests/test_query_wire_vectors.py              |  18 ++-
 7 files changed, 394 insertions(+), 38 deletions(-)
 create mode 100644 tests/fixtures/ir_vectors/query_user_m2m_exists_v7.json

diff --git a/crates/ferro-schema-ir/src/lib.rs b/crates/ferro-schema-ir/src/lib.rs
index dbf6d50..679f9e4 100644
--- a/crates/ferro-schema-ir/src/lib.rs
+++ b/crates/ferro-schema-ir/src/lib.rs
@@ -670,6 +670,39 @@ mod tests {
         assert_eq!(encoded, ir, "query nested-exists round-trip must not drift");
     }
 
+    #[test]
+    fn query_m2m_exists_fixture_roundtrip() {
+        // The two-hop M2M existence-test golden vector (#316): the SAME
+        // exists node, a two-hop correlation path — join table first, then
+        // the target — with the scoped inner tree over the target model.
+        // Must survive a deserialize/serialize round-trip without drift.
+        let fixture =
+            include_str!("../../../tests/fixtures/ir_vectors/query_user_m2m_exists_v7.json");
+        let parsed: serde_json::Value =
+            serde_json::from_str(fixture).expect("query m2m-exists fixture must parse");
+        let ir = parsed
+            .get("ir")
+            .cloned()
+            .expect("fixture must contain ir envelope");
+        let envelope: IrEnvelope<QueryIrPayload> =
+            serde_json::from_value(ir.clone()).expect("query m2m-exists IR must deserialize");
+        match &envelope.payload.where_clause[0] {
+            QueryNode::Exists {
+                hops, where_clause, ..
+            } => {
+                assert_eq!(hops.len(), 2, "M2M correlates through two hops");
+                assert_eq!(hops[0].to_table, "tag_users");
+                assert_eq!(hops[0].to_column, "user_id");
+                assert_eq!(hops[1].from_column, "tag_id");
+                assert_eq!(hops[1].to_table, "tag");
+                assert_eq!(where_clause.len(), 1);
+            }
+            other => panic!("where[0] must be an exists node, got {other:?}"),
+        }
+        let encoded = serde_json::to_value(&envelope).expect("query m2m-exists IR must serialize");
+        assert_eq!(encoded, ir, "query m2m-exists round-trip must not drift");
+    }
+
     #[test]
     fn query_traversal_fixture_roundtrip() {
         // Multi-hop `joins` section + path-carrying leaves must survive a
diff --git a/src/ferro/columns.py b/src/ferro/columns.py
index aeb8af5..faf5c3b 100644
--- a/src/ferro/columns.py
+++ b/src/ferro/columns.py
@@ -89,25 +89,38 @@ class RelationSpec:
 
 @dataclass(frozen=True, slots=True)
 class ReverseSpec:
-    """One reverse (BackRef) relation's existence-test facts (#314, ADR-0007).
+    """One reverse (BackRef/M2M) relation's existence-test facts (ADR-0007).
 
     Built beside :class:`RelationSpec` at the same compile choke point and
-    exposed as ``cls.__ferro_reverse_specs__``. A reverse relation supports
-    exactly one predicate form — the existence test ``t.rel.exists()`` — so
-    the spec carries only what a correlated EXISTS needs: the child model and
-    the child-side FK column the subquery correlates on. Reverse relations are
-    *tested*, never *traversed*; traversal facts stay on :class:`RelationSpec`.
+    exposed as ``cls.__ferro_reverse_specs__``. A reverse or M2M relation
+    supports exactly one predicate form — the existence test
+    ``t.rel.exists()`` — so the spec carries only what a correlated EXISTS
+    needs: the related model plus the correlation columns (the child-side FK
+    for a reverse FK; the join-table triple for M2M, #316). Reverse relations
+    are *tested*, never *traversed*; traversal facts stay on
+    :class:`RelationSpec`.
     """
 
     field_name: str
-    #: The child model — the side declaring the ``ForeignKey``.
+    #: The related model — the FK-declaring child (reverse FK) or the M2M
+    #: target: the model the inner lambda's parameter resolves against.
     target: type
-    #: Shadow FK column on the child table the EXISTS correlates against.
-    child_fk_column: str
+    #: Shadow FK column on the child table the EXISTS correlates against
+    #: (reverse FK only; ``None`` for M2M).
+    child_fk_column: str | None
     #: True when the child FK is ``unique=True`` (one-to-one BackRef). The
     #: rendering is identical at every cardinality (always correlated EXISTS);
     #: carried as a compile-side fact, not a render switch.
-    is_one_to_one: bool
+    is_one_to_one: bool = False
+    #: True for an M2M edge — the existence test correlates through the join
+    #: table (two hops) instead of a child FK (one hop).
+    is_m2m: bool = False
+    #: M2M join-table triple (``None`` for reverse FK): the association
+    #: table, its column referencing THIS side, and its column referencing
+    #: the target side — exactly as the descriptor orients them per side.
+    join_table: str | None = None
+    source_col: str | None = None
+    target_col: str | None = None
 
 
 @dataclass(frozen=True, slots=True)
@@ -495,15 +508,17 @@ def build_reverse_specs(model_cls: type[Any]) -> dict[str, ReverseSpec]:
 
     Reads the :class:`~ferro.relations.descriptors.RelationshipDescriptor`
     instances ``resolve_relationships`` installs on the class — the facts a
-    correlated EXISTS needs (child model, child FK column, one-to-one flag)
-    already live there, so this derives, never re-computes. At provisional
-    class-body time no descriptors exist yet and the map is empty; the
-    resolved second pass recompiles with descriptors installed, so the lookup
-    is complete once resolution finishes (same lifecycle as
-    :func:`build_relation_specs`).
-
-    M2M descriptors are skipped for now: the existence test's M2M form (a
-    two-hop correlation through the join table) lands in #316.
+    correlated EXISTS needs (related model, child FK column or join-table
+    triple, one-to-one flag) already live there, so this derives, never
+    re-computes. At provisional class-body time no descriptors exist yet and
+    the map is empty; the resolved second pass recompiles with descriptors
+    installed, so the lookup is complete once resolution finishes (same
+    lifecycle as :func:`build_relation_specs`).
+
+    M2M descriptors (#316) map to a join-table spec; each side's descriptor
+    already orients ``source_col``/``target_col`` for that side, so the spec
+    is a straight copy of resolution's facts on both the declaring and the
+    ``related_name`` side.
 
     Callers store the result on ``cls.__ferro_reverse_specs__`` and must
     replace, never mutate — same convention as ``__ferro_columns__``.
@@ -518,11 +533,20 @@ def build_reverse_specs(model_cls: type[Any]) -> dict[str, ReverseSpec]:
         for name, attr in vars(klass).items():
             if not isinstance(attr, RelationshipDescriptor) or name in specs:
                 continue
-            if attr.is_m2m:
-                continue
             target = REGISTRY.resolve_reference(attr.target_model_name, default=None)
             if target is None:
                 continue
+            if attr.is_m2m:
+                specs[name] = ReverseSpec(
+                    field_name=name,
+                    target=target,
+                    child_fk_column=None,
+                    is_m2m=True,
+                    join_table=attr.join_table,
+                    source_col=attr.source_col,
+                    target_col=attr.target_col,
+                )
+                continue
             specs[name] = ReverseSpec(
                 field_name=name,
                 target=target,
diff --git a/src/ferro/query/nodes.py b/src/ferro/query/nodes.py
index 33fc8b1..75f3e3d 100644
--- a/src/ferro/query/nodes.py
+++ b/src/ferro/query/nodes.py
@@ -1010,15 +1010,15 @@ def _collect_inner_traversal_paths(
 
 
 class ReverseRelationProxy:
-    """Predicate proxy for a reverse (BackRef) relation (#314, ADR-0007).
+    """Predicate proxy for a reverse (BackRef) or M2M relation (ADR-0007).
 
     Returned by attribute access on a :class:`QueryProxy` when the accessed
-    name is a resolved reverse relation. It exposes exactly one verb — the
-    existence test :meth:`exists` — because reverse relations are *tested*,
-    never *traversed*: column access, comparisons (including ``!= None`` /
-    ``== None``), and ``in_`` raise at build time with the supported spelling
-    in the message. Negation is uniform ``~`` over the returned node
-    (ADR-0008), so NOT EXISTS is ``~t.rel.exists()``.
+    name is a resolved reverse or many-to-many relation. It exposes exactly
+    one verb — the existence test :meth:`exists` — because reverse relations
+    are *tested*, never *traversed*: column access, comparisons (including
+    ``!= None`` / ``== None``), and ``in_`` raise at build time with the
+    supported spelling in the message. Negation is uniform ``~`` over the
+    returned node (ADR-0008), so NOT EXISTS is ``~t.rel.exists()``.
     """
 
     __slots__ = ("_root_model", "_name", "_spec")
@@ -1069,13 +1069,46 @@ def exists(self, inner: "Predicate[Any] | None" = None) -> QueryNode:
                 "column: the existence test correlates the child's FK "
                 "against the root primary key."
             )
-        hop = QueryJoinHop(
-            relation=self._name,
-            from_column=root_pk,
-            to_table=self._spec.target.__ferro_table__,
-            to_column=self._spec.child_fk_column,
-            target=self._spec.target,
-        )
+        if self._spec.is_m2m:
+            # Two hops through the join table (#316), same node and render
+            # loop as the one-hop form: the join table correlates to the
+            # enclosing scope, the target joins on inside the subquery. Both
+            # hops carry this relation's name — one relation, one alias
+            # family.
+            target_pk = getattr(self._spec.target, "__ferro_pk__", None)
+            if target_pk is None:
+                raise ValueError(
+                    f"t.{self._name}.exists() requires "
+                    f"{self._spec.target.__name__} to declare a primary-key "
+                    "column: the M2M existence test joins the join table to "
+                    "the target primary key."
+                )
+            hops = (
+                QueryJoinHop(
+                    relation=self._name,
+                    from_column=root_pk,
+                    to_table=self._spec.join_table,
+                    to_column=self._spec.source_col,
+                    target=self._spec.target,
+                ),
+                QueryJoinHop(
+                    relation=self._name,
+                    from_column=self._spec.target_col,
+                    to_table=self._spec.target.__ferro_table__,
+                    to_column=target_pk,
+                    target=self._spec.target,
+                ),
+            )
+        else:
+            hops = (
+                QueryJoinHop(
+                    relation=self._name,
+                    from_column=root_pk,
+                    to_table=self._spec.target.__ferro_table__,
+                    to_column=self._spec.child_fk_column,
+                    target=self._spec.target,
+                ),
+            )
         where: tuple[QueryNode, ...] = ()
         joins: tuple[Any, ...] = ()
         if inner is not None:
@@ -1093,7 +1126,7 @@ def exists(self, inner: "Predicate[Any] | None" = None) -> QueryNode:
             where = (node,)
         return QueryNode(
             exists=ExistsTest(
-                hops=(hop,), where=where, joins=joins, owner=self._root_model
+                hops=hops, where=where, joins=joins, owner=self._root_model
             )
         )
 
diff --git a/src/query.rs b/src/query.rs
index 1d9c44e..5b11ea6 100644
--- a/src/query.rs
+++ b/src/query.rs
@@ -1268,6 +1268,59 @@ mod tests {
         );
     }
 
+    #[test]
+    fn two_hop_exists_renders_join_table_then_target() {
+        // M2M existence test (#316): the SAME render loop — first hop's
+        // table is the subquery FROM correlated to the enclosing scope,
+        // the second hop renders as an inner join inside the subquery, and
+        // the inner tree qualifies by the LAST hop's alias (the target).
+        let payload: ferro_schema_ir::QueryIrPayload = serde_json::from_value(json!({
+            "model_name": "User",
+            "where": [
+                {"node_kind": "exists",
+                 "hops": [
+                    {"relation": "tags", "from_column": "id",
+                     "to_table": "tag_users", "to_column": "user_id"},
+                    {"relation": "tags", "from_column": "tag_id",
+                     "to_table": "tag", "to_column": "id"}
+                 ],
+                 "where": [
+                    {"node_kind": "leaf", "column": "name", "operator": "==",
+                     "value": {"kind": "string", "value": "admin"}, "path": []}
+                 ]}
+            ],
+            "order_by": [],
+            "limit": null, "offset": null, "m2m": null, "materialization": {"kind": "root_instances"}, "joins": []
+        }))
+        .expect("payload deserializes");
+        let plan = QueryPlan::from_ir_payload(payload).expect("plan builds");
+        let mut select = Query::select();
+        select.from(Alias::new("user")).cond_where(
+            plan.to_condition_for_backend(Dialect::Postgres, Some("user"))
+                .expect("valid test query"),
+        );
+        let sql = select.to_string(PostgresQueryBuilder).to_lowercase();
+        assert!(
+            sql.contains("exists(select 1 from \"tag_users\" as \"x1_tags\""),
+            "join table is the subquery FROM: {sql}"
+        );
+        assert!(
+            sql.contains(
+                "inner join \"tag\" as \"x2_tags\" on \
+                 \"x1_tags\".\"tag_id\" = \"x2_tags\".\"id\""
+            ),
+            "target joins on inside the subquery: {sql}"
+        );
+        assert!(
+            sql.contains("\"x1_tags\".\"user_id\" = \"user\".\"id\""),
+            "join table correlates to the enclosing alias: {sql}"
+        );
+        assert!(
+            sql.contains("\"x2_tags\".\"name\" = 'admin'"),
+            "inner tree qualifies by the LAST hop's alias: {sql}"
+        );
+    }
+
     #[test]
     fn exists_requires_a_qualified_enclosing_scope() {
         // Correlation needs an alias to correlate against; the unqualified
diff --git a/tests/fixtures/ir_vectors/query_user_m2m_exists_v7.json b/tests/fixtures/ir_vectors/query_user_m2m_exists_v7.json
new file mode 100644
index 0000000..b0b6668
--- /dev/null
+++ b/tests/fixtures/ir_vectors/query_user_m2m_exists_v7.json
@@ -0,0 +1,51 @@
+{
+  "vector_name": "query_user_m2m_exists_v7",
+  "domain": "query",
+  "expect_valid": true,
+  "ir": {
+    "ir_kind": "query",
+    "ir_version": 7,
+    "payload": {
+      "model_name": "User",
+      "where": [
+        {
+          "node_kind": "exists",
+          "hops": [
+            {
+              "relation": "tags",
+              "from_column": "id",
+              "to_table": "tag_users",
+              "to_column": "user_id"
+            },
+            {
+              "relation": "tags",
+              "from_column": "tag_id",
+              "to_table": "tag",
+              "to_column": "id"
+            }
+          ],
+          "where": [
+            {
+              "node_kind": "leaf",
+              "column": "name",
+              "operator": "==",
+              "value": {
+                "kind": "string",
+                "value": "admin"
+              },
+              "path": []
+            }
+          ]
+        }
+      ],
+      "order_by": [],
+      "limit": null,
+      "offset": null,
+      "m2m": null,
+      "joins": [],
+      "materialization": {
+        "kind": "root_instances"
+      }
+    }
+  }
+}
diff --git a/tests/test_query_exists.py b/tests/test_query_exists.py
index f2a939b..bdfdf3a 100644
--- a/tests/test_query_exists.py
+++ b/tests/test_query_exists.py
@@ -17,7 +17,16 @@
 import pytest
 from typing import Annotated
 
-from ferro import BackRef, FerroField, ForeignKey, Model, Relation, connect, engines
+from ferro import (
+    BackRef,
+    FerroField,
+    ForeignKey,
+    ManyToMany,
+    Model,
+    Relation,
+    connect,
+    engines,
+)
 
 pytestmark = pytest.mark.backend_matrix
 
@@ -508,3 +517,140 @@ async def test_inner_lambda_must_return_a_predicate(db_url):
     await connect(db_url, auto_migrate=True)
     with pytest.raises(TypeError, match="predicate"):
         ExTxn.where(lambda t: t.lines.exists(lambda line: line.amount))
+
+
+# ---------------------------------------------------------------------------
+# Many-to-many (#316): the same verb, the same node, a two-hop correlation
+# path — join table first (correlated to the enclosing scope), then the
+# target. M2M is test surface, not a second mechanism.
+# ---------------------------------------------------------------------------
+
+
+class ExTag(Model):
+    id: Annotated[int | None, FerroField(primary_key=True)] = None
+    name: str = ""
+    users: Relation[list["ExUser"]] = ManyToMany(related_name="tags")
+
+
+class ExUser(Model):
+    id: Annotated[int | None, FerroField(primary_key=True)] = None
+    username: str = ""
+    tags: Relation[list["ExTag"]] = BackRef()
+
+
+async def _seed_tags() -> dict[str, object]:
+    admin = await ExTag.create(name="admin")
+    beta = await ExTag.create(name="beta")
+    alice = await ExUser.create(username="alice")
+    bob = await ExUser.create(username="bob")
+    await ExUser.create(username="carol")  # tag-less
+    await admin.users.add(alice)
+    await beta.users.add(alice)
+    await beta.users.add(bob)
+    return {"admin": admin, "beta": beta, "alice": alice, "bob": bob}
+
+
+async def _usernames(query) -> list[str]:
+    return sorted(r.username for r in await query.all())
+
+
+@pytest.mark.asyncio
+async def test_m2m_bare_exists(db_url):
+    """Bare ``.exists()`` on an M2M relation: any linked row, each root
+    exactly once no matter how many join-table rows match."""
+    await connect(db_url, auto_migrate=True)
+    async with engines.session():
+        await _seed_tags()
+
+        rows = await _usernames(ExUser.where(lambda u: u.tags.exists()))
+        assert rows == ["alice", "bob"]
+
+
+@pytest.mark.asyncio
+async def test_m2m_scoped_exists(db_url):
+    """The #316 demo: ``u.tags.exists(lambda tag: tag.name == "admin")``."""
+    await connect(db_url, auto_migrate=True)
+    async with engines.session():
+        await _seed_tags()
+
+        rows = await _usernames(
+            ExUser.where(lambda u: u.tags.exists(lambda tag: tag.name == "admin"))
+        )
+        assert rows == ["alice"]
+
+
+@pytest.mark.asyncio
+async def test_m2m_negated_exists(db_url):
+    """``~u.tags.exists(...)`` renders NOT EXISTS over the two-hop path."""
+    await connect(db_url, auto_migrate=True)
+    async with engines.session():
+        await _seed_tags()
+
+        assert await _usernames(ExUser.where(lambda u: ~u.tags.exists())) == ["carol"]
+        assert await _usernames(
+            ExUser.where(lambda u: ~u.tags.exists(lambda tag: tag.name == "admin"))
+        ) == ["bob", "carol"]
+
+
+@pytest.mark.asyncio
+async def test_m2m_exists_from_the_declaring_side(db_url):
+    """The declaring side spells identically: tags with at least one user."""
+    await connect(db_url, auto_migrate=True)
+    async with engines.session():
+        await _seed_tags()
+        await ExTag.create(name="unused")
+
+        results = await ExTag.where(lambda t: t.users.exists()).all()
+        assert sorted(r.name for r in results) == ["admin", "beta"]
+
+
+@pytest.mark.asyncio
+async def test_m2m_inner_lambda_full_predicate_power(db_url):
+    """Operators, combinators, and ``~`` inside the M2M inner lambda, exactly
+    like the reverse-FK slice."""
+    await connect(db_url, auto_migrate=True)
+    async with engines.session():
+        await _seed_tags()
+
+        rows = await _usernames(
+            ExUser.where(
+                lambda u: u.tags.exists(
+                    lambda tag: (
+                        tag.name.in_(["admin", "beta"]) & ~tag.name.like("%adm%")
+                    )
+                )
+            )
+        )
+        assert rows == ["alice", "bob"]
+
+
+@pytest.mark.asyncio
+async def test_m2m_exists_composes_with_root_predicates(db_url):
+    await connect(db_url, auto_migrate=True)
+    async with engines.session():
+        await _seed_tags()
+
+        rows = await _usernames(
+            ExUser.where(
+                lambda u: (
+                    u.tags.exists(lambda tag: tag.name == "beta")
+                    & (u.username != "bob")
+                )
+            )
+        )
+        assert rows == ["alice"]
+
+        n = await ExUser.where(lambda u: u.tags.exists()).count()
+        assert n == 2
+
+
+@pytest.mark.asyncio
+async def test_m2m_proxy_rejects_everything_but_exists(db_url):
+    """The M2M reverse proxy has the same single verb as the reverse-FK one."""
+    await connect(db_url, auto_migrate=True)
+    with pytest.raises(AttributeError, match=r"\.exists\(\)"):
+        ExUser.where(lambda u: u.tags.name == "admin")
+    with pytest.raises(TypeError, match=r"\.exists\(\)"):
+        ExUser.where(lambda u: u.tags != None)  # noqa: E711
+    with pytest.raises(TypeError, match=r"\.exists\(\)"):
+        ExUser.where(lambda u: u.tags.in_([1]))
diff --git a/tests/test_query_wire_vectors.py b/tests/test_query_wire_vectors.py
index d7b4822..0ccfb9a 100644
--- a/tests/test_query_wire_vectors.py
+++ b/tests/test_query_wire_vectors.py
@@ -26,7 +26,7 @@
 
 import pytest
 
-from ferro import BackRef, FerroField, ForeignKey, Model, Relation
+from ferro import BackRef, FerroField, ForeignKey, ManyToMany, Model, Relation
 from ferro.query.wire import compile_query
 from ferro.relations import resolve_relationships
 
@@ -69,6 +69,12 @@ class User(Model):
         active: bool = True
         email: str = ""
         role: str = ""
+        tags: Relation[list["Tag"]] = BackRef()
+
+    class Tag(Model):
+        id: Annotated[int | None, FerroField(primary_key=True)] = None
+        name: str = ""
+        users: Relation[list["User"]] = ManyToMany(related_name="tags")
 
     resolve_relationships()
     return {
@@ -76,6 +82,7 @@ class User(Model):
         "Account": Account,
         "Transaction": Transaction,
         "User": User,
+        "Tag": Tag,
     }
 
 
@@ -246,6 +253,14 @@ def _q_nested_exists(m: dict[str, type]) -> Any:
     )
 
 
+def _q_m2m_exists(m: dict[str, type]) -> Any:
+    # M2M existence test (#316): the same exists node carries a TWO-hop
+    # correlation path — join table first (correlated to the enclosing
+    # scope), then the target — both hops named for the one relation they
+    # belong to. The scoped inner tree resolves over the target model.
+    return m["User"].where(lambda u: u.tags.exists(lambda tag: tag.name == "admin"))
+
+
 CASES: list[tuple[str, Callable[[dict[str, type]], Any], str]] = [
     ("query_user_compound_v7", _q_user_compound, "User"),
     ("query_user_not_leaf_v7", _q_not_leaf, "User"),
@@ -254,6 +269,7 @@ def _q_nested_exists(m: dict[str, type]) -> Any:
     ("query_owner_not_exists_v7", _q_not_exists, "Owner"),
     ("query_account_scoped_exists_v7", _q_scoped_exists, "Account"),
     ("query_owner_nested_exists_v7", _q_nested_exists, "Owner"),
+    ("query_user_m2m_exists_v7", _q_m2m_exists, "User"),
     ("query_transaction_traversal_v7", _q_traversal, "Transaction"),
     ("query_transaction_left_join_v7", _q_left_join, "Transaction"),
     ("query_transaction_include_v7", _q_include, "Transaction"),

From e3e07daea5c1ed29b8f5d7051bb08c458b3e26c9 Mon Sep 17 00:00:00 2001
From: Taylor <taylor@x54.sh>
Date: Sat, 18 Jul 2026 11:30:05 -0400
Subject: [PATCH 3/3] =?UTF-8?q?feat(query):=20existence-test=20error=20sur?=
 =?UTF-8?q?faces=20and=20docs=20=E2=80=94=20the=20feature's=20edges?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Every remaining surface where a reverse or M2M relation can be named
now answers with the supported spelling, and the docs teach existence
tests end to end.

Error surfaces beyond the proxy's own (those shipped in the tracer
bullet):

- join()/left_join() on a reverse/M2M edge stays rejected — the pinned
  "a join never multiplies root rows" property is preserved by
  rejection, not implicit DISTINCT — with the error naming .exists()
  (previously an opaque "must return a relation path" TypeError).
- in_() with a query RHS stays a TypeError whose message names the
  existence test when the RHS is a query (the #307 repro's second
  guess); a non-query RHS keeps the plain message.
- include() rejection verified unchanged for M2M edges too —
  population remains a separate future mechanism.

Docs:

- New guide section "Existence Tests on Reverse & Many-to-Many
  Relations": real models (Assignment + Annotated tabs), lambda-style
  predicates throughout, rendered SQL for the #307/#308 shapes, the
  explicit grouping contrast, negation, traversal-inside-the-test,
  nesting, M2M, the loud dead ends, and the exists-test vs
  query-terminal disambiguation. Backed by two new runnable example
  scripts exercised by the docs harness.
- Operators table gains the .exists() row; the forward
  existence/absence section and the reverse-relation filtering section
  cross-link the reverse direction.
- Reference (api/queries.md) states the one-verb rule and its
  rejections.
- Roadmap and "Not Yet Supported" now distinguish reverse predicates
  (shipped — the existence test) from reverse population (still
  future; the include() rejection stays), and track cross-scope
  correlation as #309.

Refs #311, #317
---
 docs/examples/existence_tests.py           | 195 +++++++++++++++++++++
 docs/examples/existence_tests_annotated.py | 101 +++++++++++
 docs/pages/api/queries.md                  |   2 +
 docs/pages/guide/queries.md                | 126 ++++++++++++-
 docs/pages/roadmap.md                      |   3 +-
 src/ferro/query/builder.py                 |  12 ++
 src/ferro/query/nodes.py                   |  17 ++
 tests/test_query_exists.py                 |  51 ++++++
 8 files changed, 504 insertions(+), 3 deletions(-)
 create mode 100644 docs/examples/existence_tests.py
 create mode 100644 docs/examples/existence_tests_annotated.py

diff --git a/docs/examples/existence_tests.py b/docs/examples/existence_tests.py
new file mode 100644
index 0000000..b5f4ed9
--- /dev/null
+++ b/docs/examples/existence_tests.py
@@ -0,0 +1,195 @@
+"""Runnable companion to the Existence Tests guide (docs/pages/guide/queries.md).
+
+Seeds the transfer/split-line domain the guide walks through (one-to-one and
+to-many BackRefs plus a category both layers point at) and a tagged-user M2M
+pair, then runs every existence test the guide shows and asserts the exact
+rows that come back.
+"""
+
+import asyncio
+from typing import Annotated
+
+from ferro import BackRef, Field, ForeignKey, ManyToMany, Model, Relation, connect, engines
+
+
+# --8<-- [start:schema]
+class Category(Model):
+    id: int | None = Field(default=None, primary_key=True)
+    name: str
+    transactions: Relation[list["Transaction"]] = BackRef()
+    lines: Relation[list["SplitLine"]] = BackRef()
+
+
+class Transaction(Model):
+    id: int | None = Field(default=None, primary_key=True)
+    amount: int
+    category: Annotated[
+        Category | None, ForeignKey(related_name="transactions", on_delete="SET NULL")
+    ] = None
+    # One-to-one BackRefs: a transfer references a transaction via a unique FK
+    transfer_out: "Transfer" = BackRef()
+    transfer_in: "Transfer" = BackRef()
+    # To-many BackRef: a split transaction carries its lines
+    lines: Relation[list["SplitLine"]] = BackRef()
+
+
+class Transfer(Model):
+    id: int | None = Field(default=None, primary_key=True)
+    outflow_transaction: Annotated[
+        Transaction | None, ForeignKey(related_name="transfer_out", unique=True)
+    ] = None
+    inflow_transaction: Annotated[
+        Transaction | None, ForeignKey(related_name="transfer_in", unique=True)
+    ] = None
+
+
+class SplitLine(Model):
+    id: int | None = Field(default=None, primary_key=True)
+    txn: Annotated[Transaction, ForeignKey(related_name="lines", on_delete="CASCADE")]
+    category: Annotated[
+        Category | None, ForeignKey(related_name="lines", on_delete="SET NULL")
+    ] = None
+    amount: int = 0
+# --8<-- [end:schema]
+
+
+# --8<-- [start:m2m-schema]
+class Tag(Model):
+    id: int | None = Field(default=None, primary_key=True)
+    name: str
+    users: Relation[list["User"]] = ManyToMany(related_name="tags")
+
+
+class User(Model):
+    id: int | None = Field(default=None, primary_key=True)
+    username: str
+    tags: Relation[list["Tag"]] = BackRef()
+# --8<-- [end:m2m-schema]
+
+
+async def main() -> None:
+    await connect("sqlite::memory:", auto_migrate=True)
+
+    async with engines.session():
+        outflow = await Transaction.create(amount=-100)
+        inflow = await Transaction.create(amount=100)
+        plain = await Transaction.create(amount=-20)
+        assert plain.amount == -20
+        await Transfer.create(outflow_transaction=outflow)
+        await Transfer.create(inflow_transaction=inflow)
+
+        # --8<-- [start:bare]
+        # "Is this transaction part of any transfer?" — membership via either
+        # FK column, one EXISTS per side, each matching row exactly once.
+        in_transfer = await Transaction.where(
+            lambda t: t.transfer_out.exists() | t.transfer_in.exists()
+        ).all()
+
+        # The negated branch: ~ renders NOT EXISTS
+        not_in_transfer = await Transaction.where(
+            lambda t: ~t.transfer_out.exists() & ~t.transfer_in.exists()
+        ).all()
+        # --8<-- [end:bare]
+        assert {t.amount for t in in_transfer} == {-100, 100}
+        assert {t.amount for t in not_in_transfer} == {-20}
+
+        groceries = await Category.create(name="Groceries")
+        split = await Transaction.create(amount=-70)  # category vacated while split
+        await SplitLine.create(txn=split, category=groceries, amount=-30)
+        await SplitLine.create(txn=split, category=groceries, amount=-40)
+        await Transaction.create(amount=-10, category=groceries)
+        ids = [groceries.id]
+
+        # --8<-- [start:scoped]
+        # The inner lambda is a full ferro predicate over the related model:
+        # "transactions carrying the category at the root OR on any line".
+        # A line-less transaction survives through the OR's root branch, and
+        # a transaction with several matching lines comes back exactly once.
+        matching = await Transaction.where(
+            lambda t: t.category_id.in_(ids)
+            | t.lines.exists(lambda line: line.category_id.in_(ids))
+        ).all()
+        # --8<-- [end:scoped]
+        assert {t.amount for t in matching} == {-70, -10}
+
+        # --8<-- [start:composes]
+        # Root-shaped results: existence tests compose with every other
+        # predicate, ordering, and paging — nothing about the query changes.
+        page = (
+            await Transaction.where(
+                lambda t: t.category_id.in_(ids)
+                | t.lines.exists(lambda line: line.category_id.in_(ids))
+            )
+            .order_by("amount", "desc")
+            .limit(1)
+            .all()
+        )
+        # --8<-- [end:composes]
+        assert [t.amount for t in page] == [-10]
+
+        fuel_run = await Transaction.create(amount=-55)
+        await SplitLine.create(txn=fuel_run, category=groceries, amount=-55)
+        spread = await Transaction.create(amount=-60)
+        await SplitLine.create(txn=spread, category=groceries, amount=-5)
+        await SplitLine.create(txn=spread, amount=-55)
+
+        # --8<-- [start:grouping]
+        # Grouping is YOUR choice, spelled explicitly — the two shapes below
+        # are different questions with different answers.
+
+        # One line matches BOTH conditions:
+        one_line_both = await Transaction.where(
+            lambda t: t.lines.exists(
+                lambda line: line.category_id.in_(ids) & (line.amount <= -50)
+            )
+        ).all()
+
+        # SOME line matches each condition (possibly different lines):
+        some_line_each = await Transaction.where(
+            lambda t: t.lines.exists(lambda line: line.category_id.in_(ids))
+            & t.lines.exists(lambda line: line.amount <= -50)
+        ).all()
+        # --8<-- [end:grouping]
+        assert {t.amount for t in one_line_both} == {-55}
+        assert {t.amount for t in some_line_each} == {-55, -60}
+
+        # --8<-- [start:traversal-inside]
+        # Forward traversal works inside the test (joins render INSIDE the
+        # EXISTS subquery, ADR-0006 semantics unchanged), and tests nest.
+        by_name = await Transaction.where(
+            lambda t: t.lines.exists(lambda line: line.category.name == "Groceries")
+        ).all()
+
+        active_categories = await Category.where(
+            lambda c: c.lines.exists(lambda line: line.txn.amount < -50)
+        ).all()
+        # --8<-- [end:traversal-inside]
+        assert {t.amount for t in by_name} == {-70, -55, -60}
+        assert {c.name for c in active_categories} == {"Groceries"}
+
+        admin = await Tag.create(name="admin")
+        beta = await Tag.create(name="beta")
+        alice = await User.create(username="alice")
+        bob = await User.create(username="bob")
+        await User.create(username="carol")
+        await admin.users.add(alice)
+        await beta.users.add(alice, bob)
+
+        # --8<-- [start:m2m]
+        # Many-to-many spells identically — the test correlates through the
+        # join table, and the inner lambda scopes over the target model.
+        admins = await User.where(
+            lambda u: u.tags.exists(lambda tag: tag.name == "admin")
+        ).all()
+        tagged = await User.where(lambda u: u.tags.exists()).all()
+        untagged = await User.where(lambda u: ~u.tags.exists()).all()
+        # --8<-- [end:m2m]
+        assert {u.username for u in admins} == {"alice"}
+        assert {u.username for u in tagged} == {"alice", "bob"}
+        assert {u.username for u in untagged} == {"carol"}
+
+    print("existence_tests example ran successfully")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
diff --git a/docs/examples/existence_tests_annotated.py b/docs/examples/existence_tests_annotated.py
new file mode 100644
index 0000000..f274c40
--- /dev/null
+++ b/docs/examples/existence_tests_annotated.py
@@ -0,0 +1,101 @@
+"""Annotated-style companion to existence_tests.py (AGENTS.md I-7)."""
+
+import asyncio
+from typing import Annotated
+
+from ferro import BackRef, Field, ForeignKey, ManyToMany, Model, Relation, connect, engines
+
+
+# --8<-- [start:schema]
+class Category(Model):
+    id: Annotated[int | None, Field(default=None, primary_key=True)]
+    name: str
+    transactions: Relation[list["Transaction"]] = BackRef()
+    lines: Relation[list["SplitLine"]] = BackRef()
+
+
+class Transaction(Model):
+    id: Annotated[int | None, Field(default=None, primary_key=True)]
+    amount: int
+    category: Annotated[
+        Category | None, ForeignKey(related_name="transactions", on_delete="SET NULL")
+    ] = None
+    # One-to-one BackRefs: a transfer references a transaction via a unique FK
+    transfer_out: "Transfer" = BackRef()
+    transfer_in: "Transfer" = BackRef()
+    # To-many BackRef: a split transaction carries its lines
+    lines: Relation[list["SplitLine"]] = BackRef()
+
+
+class Transfer(Model):
+    id: Annotated[int | None, Field(default=None, primary_key=True)]
+    outflow_transaction: Annotated[
+        Transaction | None, ForeignKey(related_name="transfer_out", unique=True)
+    ] = None
+    inflow_transaction: Annotated[
+        Transaction | None, ForeignKey(related_name="transfer_in", unique=True)
+    ] = None
+
+
+class SplitLine(Model):
+    id: Annotated[int | None, Field(default=None, primary_key=True)]
+    txn: Annotated[Transaction, ForeignKey(related_name="lines", on_delete="CASCADE")]
+    category: Annotated[
+        Category | None, ForeignKey(related_name="lines", on_delete="SET NULL")
+    ] = None
+    amount: int = 0
+# --8<-- [end:schema]
+
+
+# --8<-- [start:m2m-schema]
+class Tag(Model):
+    id: Annotated[int | None, Field(default=None, primary_key=True)]
+    name: str
+    users: Relation[list["User"]] = ManyToMany(related_name="tags")
+
+
+class User(Model):
+    id: Annotated[int | None, Field(default=None, primary_key=True)]
+    username: str
+    tags: Relation[list["Tag"]] = BackRef()
+# --8<-- [end:m2m-schema]
+
+
+async def main() -> None:
+    await connect("sqlite::memory:", auto_migrate=True)
+
+    async with engines.session():
+        outflow = await Transaction.create(amount=-100)
+        await Transaction.create(amount=-20)
+        await Transfer.create(outflow_transaction=outflow)
+
+        in_transfer = await Transaction.where(
+            lambda t: t.transfer_out.exists() | t.transfer_in.exists()
+        ).all()
+        assert {t.amount for t in in_transfer} == {-100}
+
+        groceries = await Category.create(name="Groceries")
+        split = await Transaction.create(amount=-70)
+        await SplitLine.create(txn=split, category=groceries, amount=-70)
+
+        line_aware = await Transaction.where(
+            lambda t: t.category_id.in_([groceries.id])
+            | t.lines.exists(lambda line: line.category_id.in_([groceries.id]))
+        ).all()
+        assert {t.amount for t in line_aware} == {-70}
+
+        admin = await Tag.create(name="admin")
+        alice = await User.create(username="alice")
+        await User.create(username="bob")
+        await admin.users.add(alice)
+
+        admins = await User.where(
+            lambda u: u.tags.exists(lambda tag: tag.name == "admin")
+        ).all()
+        assert {u.username for u in admins} == {"alice"}
+
+    print("existence_tests_annotated example ran successfully")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
diff --git a/docs/pages/api/queries.md b/docs/pages/api/queries.md
index 615d893..ca749b0 100644
--- a/docs/pages/api/queries.md
+++ b/docs/pages/api/queries.md
@@ -6,6 +6,8 @@ Prefix `~` negates **any** predicate — leaf comparison or `&`/`|` compound —
 
 `where()` and `order_by()` lambdas may **traverse** a forward-FK relation (`lambda t: t.account.ledger_id == 1`): each hop renders one INNER join, deduplicated by relation path (ADR-0006). `join()` forces a join on a relation path (a bare `join()` is an existence filter on a nullable relation), and `left_join()` marks the whole path LEFT to keep relation-less rows. See the [Querying Across Relationships](../guide/queries.md#querying-across-relationships) guide for worked examples.
 
+A reverse (`BackRef`) or many-to-many relation in a predicate supports exactly one verb — the **existence test** `t.rel.exists(inner_lambda=None)` (ADR-0007). It renders as a correlated `EXISTS` at every cardinality (never a join, so the result stays root-shaped and each matching root returns once), negates with `~`, and the optional inner lambda is a full ferro predicate over the related model (operators, `&`/`|`/`~`, forward traversal rendered inside the subquery, nested tests). Everything else on a reverse edge — column access, comparisons (including `!= None`), `in_` (including a query RHS), `join()`/`left_join()` — raises at build time naming `.exists()`; an inner lambda referencing any scope but its own parameter is likewise rejected ([#309](https://github.com/syn54x/ferro-orm/issues/309)). See [Existence Tests](../guide/queries.md#existence-tests-on-reverse-many-to-many-relations) for worked examples.
+
 ## `include()` and populated relations
 
 `include(lambda t: t.account)` delivers each result with the relation **populated** (ADR-0008): access becomes a plain attribute holding the complete related instance — no await, no query — while unpopulated relations keep the awaitable contract. Include is the third orthogonal query axis (joins decide membership, projection decides shape, include decides attached data): it never changes which rows come back, `.all()` still returns `list[Model]`, and `count()`/`exists()` are unaffected. Paths populate whole (`include(lambda t: t.account.owner)` populates both hops); includes are cumulative, order-free, and idempotent; populated instances run the full session identity-map protocol, and a refresh keeps a population only while the row's FK still points at it.
diff --git a/docs/pages/guide/queries.md b/docs/pages/guide/queries.md
index dfa4115..029c259 100644
--- a/docs/pages/guide/queries.md
+++ b/docs/pages/guide/queries.md
@@ -101,6 +101,7 @@ Lambda predicates keep the call site fully type-checked: the proxy's attributes
 | `== None` | `IS NULL` | `lambda user: user.deleted_at == None` |
 | `!= None` | `IS NOT NULL` | `lambda user: user.deleted_at != None` |
 | `~` | `NOT` | `lambda user: ~user.role.in_(["admin", "moderator"])` |
+| `.exists(...)` | `EXISTS (SELECT 1 …)` | `lambda user: user.posts.exists(lambda post: post.published == True)` — reverse/M2M relations only; see [Existence Tests](#existence-tests-on-reverse-many-to-many-relations) |
 
 ```python
 --8<-- "docs/examples/predicates.py:operators"
@@ -329,6 +330,8 @@ For the opposite question — "has *no* related row" — compare the relation to
 --8<-- "docs/examples/traversal.py:is-null"
 ```
 
+Both spellings here are the **forward** direction — the FK column lives on the queried table. Asking the same question in the *reverse* direction ("has at least one / no related child row") is an [existence test](#existence-tests-on-reverse-many-to-many-relations): `t.lines.exists()` / `~t.lines.exists()`.
+
 ### Keeping rows that have no relation
 
 When you want the relation-less rows *kept* rather than filtered out, opt into a `left_join`. It marks **every edge of its path** LEFT (the whole-path rule), so a left-marked two-hop path retains rows missing the relation at either hop:
@@ -412,6 +415,8 @@ latest = await author.posts.order_by(lambda post: post.created_at, "desc").limit
 n = await author.posts.count()
 ```
 
+That is a query *from* an instance. To filter the **root query** on membership in a reverse relation — "authors who have at least one published post" — use an [existence test](#existence-tests-on-reverse-many-to-many-relations): `Author.where(lambda a: a.posts.exists(lambda post: post.published == True))`.
+
 ### Results are plain root instances
 
 A traversed query is **shape-preserving**: filtering `Transaction` through `transaction.account.ledger_id` still returns `Transaction` instances, no matter how deep the predicate reaches. Traversal does *not* pre-load the related rows onto the results — `await transaction.account` still issues its own query, exactly as it does without any traversal. Attaching related data is a separate, explicit request: [`include()`](#populating-relations-with-include).
@@ -438,6 +443,122 @@ Do the two-step: fetch the primary keys with the joined query, then mutate by th
 
 See [Relationships](relationships.md) for the schema-declaration side of foreign keys and reverse relations.
 
+## Existence Tests on Reverse & Many-to-Many Relations
+
+Traversal reaches *forward* along a foreign key. The reverse question — "which transactions have at least one split line?", "which transactions appear in any transfer?" — is a different shape: the related rows live on the **other** table, keyed back at you. In a predicate, a reverse (`BackRef`) or many-to-many relation supports exactly one verb for that question, the **existence test**:
+
+```python
+matches = await Transaction.where(lambda t: t.lines.exists()).all()
+```
+
+`.exists()` renders as a correlated `EXISTS` subquery — never a join — so the result stays **root-shaped**: each matching row comes back exactly once (a transaction with three lines is one result, no `DISTINCT` bookkeeping), rows are never multiplied, and the test composes with every other predicate, ordering, and paging. One verb covers every cardinality: a one-to-one `BackRef`, a to-many `BackRef`, and an M2M edge all spell the same, so a schema cardinality change never breaks a call site.
+
+The examples below use this schema — a transfer links two transactions through unique FKs (one-to-one BackRefs), split lines hang off a transaction (to-many), and a category is referenced by both layers:
+
+=== "Assignment"
+
+    ```python
+    --8<-- "docs/examples/existence_tests.py:schema"
+    ```
+
+=== "Annotated"
+
+    ```python
+    --8<-- "docs/examples/existence_tests_annotated.py:schema"
+    ```
+
+### Bare tests and negation
+
+A bare `.exists()` asks "is any related row there?". Negation is the uniform `~` — NOT EXISTS is not a separate spelling:
+
+```python
+--8<-- "docs/examples/existence_tests.py:bare"
+```
+
+The first query ships to the database as:
+
+```sql
+SELECT ... FROM transaction t
+WHERE EXISTS (SELECT 1 FROM transfer WHERE transfer.outflow_transaction_id = t.id)
+   OR EXISTS (SELECT 1 FROM transfer WHERE transfer.inflow_transaction_id = t.id)
+```
+
+### Scoping with an inner predicate
+
+Pass a lambda to filter *which* related rows count. The inner lambda is a **full ferro predicate over the related model** — every operator, `&`/`|`/`~`, forward traversal, even nested existence tests — not a sub-language:
+
+```python
+--8<-- "docs/examples/existence_tests.py:scoped"
+```
+
+Rendered SQL — the root branch keeps line-less transactions, the `EXISTS` branch finds categories on any line:
+
+```sql
+SELECT ... FROM transaction t
+WHERE t.category_id IN (...)
+   OR EXISTS (SELECT 1 FROM split_line l
+              WHERE l.txn_id = t.id AND l.category_id IN (...))
+```
+
+Because the result is root-shaped, keyset ordering and paging compose unchanged:
+
+```python
+--8<-- "docs/examples/existence_tests.py:composes"
+```
+
+### Grouping is explicit
+
+With conditions on the *same* relation, there are two different questions: does **one** related row match all conditions, or does **some** related row match each? The lambda scope makes the choice visible — one test with a compound inner predicate, or two tests combined outside:
+
+```python
+--8<-- "docs/examples/existence_tests.py:grouping"
+```
+
+This is why reverse relations have an explicit combinator rather than implicit path traversal (`t.lines.category_id == x` raises): with implicit traversal, nothing on the page says which of those two questions `(t.lines.a == 1) & (t.lines.b == 2)` asks (ADR-0007).
+
+### Traversal inside the test, and nesting
+
+Forward-FK traversal works inside the inner lambda — its joins render *inside* the `EXISTS` subquery with the same [INNER semantics](#every-traversed-hop-is-an-inner-join) as everywhere else — and existence tests nest to any depth:
+
+```python
+--8<-- "docs/examples/existence_tests.py:traversal-inside"
+```
+
+### Many-to-many
+
+An M2M relation spells identically, from either side — the test correlates through the association table and the inner lambda scopes over the target model:
+
+=== "Assignment"
+
+    ```python
+    --8<-- "docs/examples/existence_tests.py:m2m-schema"
+    ```
+
+=== "Annotated"
+
+    ```python
+    --8<-- "docs/examples/existence_tests_annotated.py:m2m-schema"
+    ```
+
+```python
+--8<-- "docs/examples/existence_tests.py:m2m"
+```
+
+### One verb, loud dead ends
+
+Reverse relations are **tested, never traversed** (ADR-0007). Every other way of naming a reverse or M2M relation in a query fails at build time with the supported spelling in the message:
+
+- **Column access** (`t.lines.category_id`) raises `AttributeError` — scope the columns with the inner lambda instead: `t.lines.exists(lambda line: line.category_id == ...)`.
+- **Comparisons**, including the tempting `t.transfer_out != None`, raise `TypeError` — a reverse relation has no root-side column to be `NULL`; the spelling is `t.transfer_out.exists()` (and `~t.transfer_out.exists()` for absence).
+- **`in_()` with a query** (`t.id.in_(subquery)`) raises `TypeError` — the workloads an `IN (subquery)` serves are existence-test workloads, without hand-correlating on id columns.
+- **`join()` / `left_join()` on a reverse edge** raise `TypeError` — a join there would multiply root rows; membership is the existence test's job.
+- **The inner lambda sees only its own parameter.** Referencing the outer lambda's parameter (or comparing column-to-column) is a build-time error — cross-scope correlation is a tracked future capability ([#309](https://github.com/syn54x/ferro-orm/issues/309)), never a silent misrender.
+
+Existence tests answer **membership** only. *Populating* a reverse collection onto results (the data axis) is a separate future mechanism — see [Not Yet Supported](#not-yet-supported); until it lands, fetch collections through the relation itself (`await txn.lines.all()`).
+
+!!! note "Two `exists`, two levels"
+    `t.lines.exists(...)` inside a predicate is the existence *test* on a relation. `await query.exists()` is the query *terminal* asking whether the whole query matches any row. Same word, deliberately — both ask "is there at least one?" — at different levels.
+
 ## Populating Relations with include()
 
 Every relation access is a query. A list view that renders 100 transactions with their account labels awaits `transaction.account` 100 times — 101 statements for one screen. `include()` cures that N+1: ask the query to bring the related rows along, and each result's relation arrives **populated**:
@@ -536,7 +657,7 @@ Every fetch refreshes instances the session already holds. A refresh keeps each
 
 Misuse raises at build time, before any SQL:
 
-- **Forward foreign keys only.** Including a `BackRef` or `ManyToMany` relation raises `TypeError`: reverse and M2M population will be a separate mechanism (a batched second query stitched onto the results), not `include()`. Until it lands, fetch collections through the relation itself (`await author.posts.all()`).
+- **Forward foreign keys only.** Including a `BackRef` or `ManyToMany` relation raises `TypeError`: reverse and M2M population will be a separate mechanism (a batched second query stitched onto the results), not `include()`. Until it lands, fetch collections through the relation itself (`await author.posts.all()`). *Filtering* on reverse membership is a different axis and already works — the [existence test](#existence-tests-on-reverse-many-to-many-relations).
 - **Lambda selectors only.** `include("account")` raises pointing at the lambda form — strings never traverse. Selecting a *column* (`include(lambda t: t.account.label)`) raises too: every populated hop is a complete row, so there is nothing to select per column.
 - **No include × projection.** A query carries exactly one materialization plan — populated instances or projected records, never both — and record results are flat, permanently. Either order raises `ValueError` pointing at [traversed projection](#reaching-across-a-relation), the record-shaped way across a relation.
 - **No mutations.** `update()`/`delete()` on an included query raise — a mutation returns no instances to populate.
@@ -675,7 +796,8 @@ Aggregation builds directly on this machinery — `select(lambda t: {"total": t.
     The following query features are **not yet implemented** — see the [Roadmap](../roadmap.md):
 
     - `having()` — post-aggregation filtering; `where()` rejects aggregate predicates pointing at it ([#291](https://github.com/syn54x/ferro-orm/issues/291))
-    - Reverse (`BackRef`) and many-to-many population — [`include()`](#populating-relations-with-include) covers forward FKs; collection population is a separate future mechanism
+    - Reverse (`BackRef`) and many-to-many **population** — [`include()`](#populating-relations-with-include) covers forward FKs; collection population is a separate future mechanism. (*Filtering* on reverse/M2M membership is supported — that's the [existence test](#existence-tests-on-reverse-many-to-many-relations).)
+    - Cross-scope correlation inside an existence test — comparing an inner-lambda column to an outer column ([#309](https://github.com/syn54x/ferro-orm/issues/309)); rejected loudly at build time today
     - Case-insensitive `ilike()`
 
 ## See Also
diff --git a/docs/pages/roadmap.md b/docs/pages/roadmap.md
index 45db79d..1a6d9de 100644
--- a/docs/pages/roadmap.md
+++ b/docs/pages/roadmap.md
@@ -6,7 +6,8 @@ Ferro is pre-1.0 and under active development. The items below are known gaps we
 
 - **`having()`** — post-aggregation filtering for [grouped queries](guide/aggregations.md); `where()` rejects aggregate predicates pointing at it ([#291](https://github.com/syn54x/ferro-orm/issues/291)). Workaround: filter groups in Python after `all()`.
 - **Typed record fields** — a projected `Row`'s fields type as `Any` on access today; inferring per-field types from the selector (so `row.total` checks as `int | None`) is [#290](https://github.com/syn54x/ferro-orm/issues/290).
-- **Reverse and many-to-many population** — [`include()`](guide/queries.md#populating-relations-with-include) populates forward-FK paths in one statement; populating `BackRef` collections and M2M sets is a separate future mechanism (a batched second query stitched onto the results). Today each awaited collection is its own query.
+- **Reverse and many-to-many population** — [`include()`](guide/queries.md#populating-relations-with-include) populates forward-FK paths in one statement; populating `BackRef` collections and M2M sets is a separate future mechanism (a batched second query stitched onto the results). Today each awaited collection is its own query. (*Filtering* on reverse/M2M membership already works — the [existence test](guide/queries.md#existence-tests-on-reverse-many-to-many-relations), `t.lines.exists(...)`.)
+- **Cross-scope correlation in existence tests** — comparing an inner-lambda column against the outer scope (`t.lines.exists(lambda line: line.category_id == t.category_id)`) is rejected at build time today; correlated column-to-column comparison is [#309](https://github.com/syn54x/ferro-orm/issues/309).
 - **`ilike()`** — case-insensitive pattern matching. Workaround: `like()` with normalized case.
 - **Atomic update expressions** — database-side expressions in batch updates, e.g. `update(view_count=Post.view_count + 1)`, avoiding the read-modify-write race. Workaround today: load, mutate, `save()` (or raw SQL).
 
diff --git a/src/ferro/query/builder.py b/src/ferro/query/builder.py
index 47a64e5..a7c1aa1 100644
--- a/src/ferro/query/builder.py
+++ b/src/ferro/query/builder.py
@@ -158,6 +158,18 @@ def _resolve_join_selector(
             "(e.g. `lambda t: t.account.name`); a join selector names a relation "
             "path (e.g. `lambda t: t.account`)."
         )
+    if isinstance(result, ReverseRelationProxy):
+        # Joining a reverse/M2M edge would multiply root rows — the pinned
+        # "a join never multiplies root rows" property is preserved by
+        # rejection (ADR-0007); membership is the existence test's job.
+        relation = result._name
+        raise TypeError(
+            f"join()/left_join() cannot join the reverse relation "
+            f"{relation!r}: a join on a reverse or many-to-many edge would "
+            "multiply root rows. Test membership with a predicate instead — "
+            f"where(lambda t: t.{relation}.exists(...)), negated with ~ — "
+            "reverse relations are tested, not traversed (ADR-0007)."
+        )
     if not isinstance(result, RelationProxy):
         raise TypeError(
             "join()/left_join() selector must return a relation path "
diff --git a/src/ferro/query/nodes.py b/src/ferro/query/nodes.py
index 75f3e3d..3808f5f 100644
--- a/src/ferro/query/nodes.py
+++ b/src/ferro/query/nodes.py
@@ -466,6 +466,23 @@ def in_(
             'IN'
         """
         if not isinstance(other, (list, tuple, set)):
+            # Late import: builder imports this module at load time.
+            # ProjectedQuery and Relation subclass Query, so one check
+            # covers every query shape.
+            from .builder import Query
+
+            if isinstance(other, Query):
+                # in_(subquery) is declined, not deferred (ADR-0007): the
+                # workloads it serves are existence-test workloads, and
+                # hand-correlating on id columns leaks the join column into
+                # every call site.
+                raise TypeError(
+                    f"The 'in_' operator expects a list, tuple, or set, got "
+                    f"{type(other).__name__}. To filter on membership in "
+                    "related rows, use the existence test on the relation — "
+                    "where(lambda t: t.<relation>.exists(lambda r: ...)) — "
+                    "instead of an in_(subquery) (ADR-0007)."
+                )
             raise TypeError(
                 f"The 'in_' operator expects a list, tuple, or set, got {type(other).__name__}"
             )
diff --git a/tests/test_query_exists.py b/tests/test_query_exists.py
index bdfdf3a..b67c6a7 100644
--- a/tests/test_query_exists.py
+++ b/tests/test_query_exists.py
@@ -644,6 +644,57 @@ async def test_m2m_exists_composes_with_root_predicates(db_url):
         assert n == 2
 
 
+# ---------------------------------------------------------------------------
+# Remaining error surfaces (#317): every place a reverse or M2M relation can
+# be named now answers with the supported spelling.
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_left_join_on_reverse_edge_names_exists(db_url):
+    """``left_join()`` on a reverse edge stays rejected — the pinned "a join
+    never multiplies root rows" property is preserved by rejection — and the
+    error now names ``.exists()``."""
+    await connect(db_url, auto_migrate=True)
+    with pytest.raises(TypeError, match=r"\.exists\("):
+        ExTxn.select().left_join(lambda t: t.transfer_out)
+    with pytest.raises(TypeError, match=r"\.exists\("):
+        ExTxn.select().join(lambda t: t.lines)
+
+
+@pytest.mark.asyncio
+async def test_left_join_on_m2m_edge_names_exists(db_url):
+    await connect(db_url, auto_migrate=True)
+    with pytest.raises(TypeError, match=r"\.exists\("):
+        ExUser.select().left_join(lambda u: u.tags)
+
+
+@pytest.mark.asyncio
+async def test_in_with_query_rhs_names_exists(db_url):
+    """``in_()`` with a query RHS stays a TypeError, and the message names the
+    existence test when the RHS is a query (the #307 repro's second guess)."""
+    await connect(db_url, auto_migrate=True)
+    sub = ExLine.select(lambda line: line.txn_id)
+    with pytest.raises(TypeError, match=r"\.exists\("):
+        ExTxn.where(lambda t: t.id.in_(sub))
+    with pytest.raises(TypeError, match=r"\.exists\("):
+        ExTxn.where(lambda t: t.id.in_(ExLine.select()))
+    # A non-query, non-collection RHS keeps the plain message — no exists
+    # hint where none applies.
+    with pytest.raises(TypeError, match="expects a list, tuple, or set") as exc_info:
+        ExTxn.where(lambda t: t.id.in_(42))
+    assert ".exists(" not in str(exc_info.value)
+
+
+@pytest.mark.asyncio
+async def test_include_on_m2m_edge_unchanged(db_url):
+    """``include()`` population rejection is unchanged for M2M too — reverse
+    population stays a separate future mechanism."""
+    await connect(db_url, auto_migrate=True)
+    with pytest.raises(TypeError, match="reverse .BackRef. or many-to-many"):
+        ExUser.select().include(lambda u: u.tags)
+
+
 @pytest.mark.asyncio
 async def test_m2m_proxy_rejects_everything_but_exists(db_url):
     """The M2M reverse proxy has the same single verb as the reverse-FK one."""