diff --git a/crates/ferro-schema-ir/src/lib.rs b/crates/ferro-schema-ir/src/lib.rs
index 3f27977..679f9e4 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,104 @@ 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_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/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/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/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 132467c..3808f5f 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]"
@@ -443,10 +466,29 @@ 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__}"
             )
-        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 +508,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 +766,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 +799,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 +823,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,16 +882,160 @@ 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).
+    """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")
@@ -845,25 +1045,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:
@@ -873,14 +1086,66 @@ def exists(self) -> 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:
+            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=hops, where=where, joins=joins, owner=self._root_model
+            )
         )
-        return QueryNode(exists=ExistsTest(hops=(hop,), where=()))
 
     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..5b11ea6 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,152 @@ 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 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_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/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_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..b67c6a7 100644
--- a/tests/test_query_exists.py
+++ b/tests/test_query_exists.py
@@ -1,27 +1,49 @@
-"""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
 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
 
 
+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 +62,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 +134,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 +149,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 +242,466 @@ 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))
+
+
+# ---------------------------------------------------------------------------
+# 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
+
+
+# ---------------------------------------------------------------------------
+# 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."""
+    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 25485a2..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,
     }
 
 
@@ -225,12 +232,44 @@ 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())
+    )
+
+
+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"),
     ("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_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"),