diff --git a/backend/app/context/history.py b/backend/app/context/history.py
index 18300e6..984f833 100644
--- a/backend/app/context/history.py
+++ b/backend/app/context/history.py
@@ -12,10 +12,10 @@ This module serves those shapes directly from SQL — a tail, a slice, a count 
 so the read is bounded by the context budget instead of by the length of the
 story.
 
-Two rules hold everything together:
+Three rules hold everything together:
 
 * **One definition of "story action".** The cursors in memorybank are
-  *positions* in this filtered, index-ordered list, so SQL and Python must
+  *positions* in this filtered, depth-ordered list, so SQL and Python must
   agree on membership exactly or a cursor silently points at a different
   action. `_STORY_TEXT` and `is_story_text()` are that one definition, written
   twice; keep them in step.
@@ -23,12 +23,22 @@ Two rules hold everything together:
   scripting pipeline hands the whole history to user scripts, as AI Dungeon
   does), every helper here slices that instead of issuing a query, so a
   scripted adventure pays what it always paid and nothing more.
+* **Every read goes through the branch clause** (Phase 14). `adventure.actions`
+  is every branch's actions, not the story being played — so the shortcut above
+  cuts the loaded collection down to the path before slicing it, exactly as the
+  SQL does. This is the line that would silently assemble a prompt out of two
+  different stories, which is why `lineage.Path` owns both halves of it.
+
+Ordering is by `depth` now, not `index`. The two hold the same numbers until
+retry stops mutating rows (SP4), but only one of them is a position along a
+path.
 """
 
 from sqlalchemy import func, inspect as sa_inspect
 from sqlalchemy.orm import Session, defer, object_session
 
 from .. import models
+from . import lineage
 
 # How many of the newest actions to read before checking whether the token
 # budget is covered. When it isn't, the next size is worked out from the
@@ -38,6 +48,12 @@ WINDOW_START = 32
 WINDOW_MARGIN = 0.15  # aim this far past the budget, so one more round is rare
 WINDOW_STEP = 8  # ...and at least this many more actions each round
 
+# Depth is the ordering key; id breaks the tie that a pre-tree row (depth NULL,
+# and so invisible anyway) or a future sibling pair would otherwise leave to
+# the database's mood.
+_OLDEST_FIRST = (models.Action.depth, models.Action.id)
+_NEWEST_FIRST = (models.Action.depth.desc(), models.Action.id.desc())
+
 
 def _sql_stripped(column):
     """`column` with leading/trailing whitespace removed, portably.
@@ -78,33 +94,75 @@ def _loaded_actions(adventure: models.Adventure) -> list[models.Action] | None:
 def _from_memory(
     adventure: models.Adventure, exclude_action_id: int | None
 ) -> list[models.Action] | None:
+    """The story, from the already-loaded collection, or None to go to SQL.
+
+    The collection is the *adventure's* actions — every branch of it. Cutting
+    it down to the path here is the same filter the SQL applies, and skipping
+    it would hand the context builder a prompt assembled from siblings of the
+    story being played. The path needs a session to read the branch row from;
+    without one there is no answer to give, so say so rather than guess.
+    """
     loaded = _loaded_actions(adventure)
     if loaded is None:
         return None
-    return [
+    db = _session(adventure)
+    if db is None:
+        return None
+    path = lineage.path_of(db, adventure)
+    rows = [
         a for a in loaded
-        if is_story_text(a.text) and (exclude_action_id is None or a.id != exclude_action_id)
+        if path.contains(a)
+        and is_story_text(a.text)
+        and (exclude_action_id is None or a.id != exclude_action_id)
     ]
+    # `Adventure.actions` is ordered by `index`; a path is ordered by depth.
+    rows.sort(key=path.sort_key)
+    return rows
 
 
-def _filters(adventure: models.Adventure, exclude_action_id: int | None) -> list:
-    conditions = [models.Action.adventure_id == adventure.id, _STORY_TEXT]
+def _filters(
+    adventure: models.Adventure,
+    path: lineage.Path,
+    exclude_action_id: int | None,
+    entries: int | None = None,
+) -> list:
+    # adventure_id is redundant beside the branch clause — branch ids are
+    # unique, so a branch already names one adventure. It stays because it is
+    # the cheap half of the check that catches a node written onto the wrong
+    # adventure's branch, and because a clause nobody can read is a clause
+    # nobody maintains.
+    conditions = [
+        models.Action.adventure_id == adventure.id,
+        path.clause(models.Action, count=entries),
+        _STORY_TEXT,
+    ]
     if exclude_action_id is not None:
         conditions.append(models.Action.id != exclude_action_id)
     return conditions
 
 
-def _query(db: Session, adventure: models.Adventure, exclude_action_id: int | None):
+def _query(
+    db: Session,
+    adventure: models.Adventure,
+    path: lineage.Path,
+    exclude_action_id: int | None,
+    entries: int | None = None,
+):
     # Reasoning traces are never read from replayed history and can be larger
     # than the narration itself on a reasoning model.
     return (
         db.query(models.Action)
-        .filter(*_filters(adventure, exclude_action_id))
+        .filter(*_filters(adventure, path, exclude_action_id, entries))
         .options(defer(models.Action.reasoning))
     )
 
 
-def _count_query(db: Session, adventure: models.Adventure, exclude_action_id: int | None):
+def _count_query(
+    db: Session,
+    adventure: models.Adventure,
+    path: lineage.Path,
+    exclude_action_id: int | None,
+):
     """A real `SELECT count(...)`.
 
     Deliberately not `_query(...).count()`: that wraps the entity select in a
@@ -114,7 +172,7 @@ def _count_query(db: Session, adventure: models.Adventure, exclude_action_id: in
     that greps the SQL cannot tell the two apart.
     """
     return db.query(func.count(models.Action.id)).filter(
-        *_filters(adventure, exclude_action_id)
+        *_filters(adventure, path, exclude_action_id)
     )
 
 
@@ -122,6 +180,10 @@ def _session(adventure: models.Adventure) -> Session | None:
     return object_session(adventure)
 
 
+def _path(db: Session, adventure: models.Adventure) -> lineage.Path:
+    return lineage.path_of(db, adventure)
+
+
 # ------------------------------------------------------------------ the API
 
 def story_actions(
@@ -143,7 +205,11 @@ def story_actions(
     db = _session(adventure)
     if db is None:
         return []
-    return _query(db, adventure, exclude_action_id).order_by(models.Action.index).all()
+    return (
+        _query(db, adventure, _path(db, adventure), exclude_action_id)
+        .order_by(*_OLDEST_FIRST)
+        .all()
+    )
 
 
 def count(adventure: models.Adventure, exclude_action_id: int | None = None) -> int:
@@ -154,7 +220,10 @@ def count(adventure: models.Adventure, exclude_action_id: int | None = None) ->
     db = _session(adventure)
     if db is None:
         return 0
-    return _count_query(db, adventure, exclude_action_id).scalar() or 0
+    return (
+        _count_query(db, adventure, _path(db, adventure), exclude_action_id).scalar()
+        or 0
+    )
 
 
 def tail_range(
@@ -168,6 +237,15 @@ def tail_range(
     `skip=0` is the newest slice; `skip=32, limit=16` is the 16 actions just
     older than the newest 32. Lets a growing window fetch only the part it
     doesn't already have.
+
+    This is the read the lineage window exists for. The path's ranges are
+    disjoint and descending, so the newest N nodes come from the newest few
+    lineage entries and the rest of the ancestry need not be named at all: a
+    story forked two hundred times reads its tail with as few clauses as one
+    forked never. `prefix_covering` estimates how many entries that takes from
+    depth arithmetic alone; the estimate is only ever short where a middle
+    action was deleted, and then the read widens to the whole lineage and pays
+    one more query.
     """
     if limit <= 0 or skip < 0:
         return []
@@ -178,13 +256,19 @@ def tail_range(
     db = _session(adventure)
     if db is None:
         return []
-    rows = (
-        _query(db, adventure, exclude_action_id)
-        .order_by(models.Action.index.desc())
-        .offset(skip)
-        .limit(limit)
-        .all()
-    )
+    path = _path(db, adventure)
+    entries = path.prefix_covering(skip + limit)
+    while True:
+        rows = (
+            _query(db, adventure, path, exclude_action_id, entries)
+            .order_by(*_NEWEST_FIRST)
+            .offset(skip)
+            .limit(limit)
+            .all()
+        )
+        if len(rows) >= limit or entries >= len(path):
+            break
+        entries = len(path)  # short: widen once, to everything, and re-ask
     rows.reverse()
     return rows
 
@@ -204,8 +288,11 @@ def slice_(
 ) -> list[models.Action]:
     """Story actions at positions [start, start + length), oldest first.
 
-    Positions are into the same filtered, index-ordered list the memory cursors
+    Positions are into the same filtered, depth-ordered list the memory cursors
     count in, which is why the filter has to match Python's exactly.
+
+    Counts from the oldest end, so it names the whole lineage: there is no
+    prefix of the ancestry that holds "the story's first ten actions".
     """
     if length <= 0 or start < 0:
         return []
@@ -216,8 +303,8 @@ def slice_(
     if db is None:
         return []
     return (
-        _query(db, adventure, exclude_action_id)
-        .order_by(models.Action.index)
+        _query(db, adventure, _path(db, adventure), exclude_action_id)
+        .order_by(*_OLDEST_FIRST)
         .offset(start)
         .limit(length)
         .all()
@@ -241,7 +328,7 @@ def position_of_index(adventure: models.Adventure, index: int) -> int:
     if db is None:
         return 0
     return (
-        _count_query(db, adventure, None)
+        _count_query(db, adventure, _path(db, adventure), None)
         .filter(models.Action.index < index)
         .scalar()
         or 0
@@ -249,7 +336,14 @@ def position_of_index(adventure: models.Adventure, index: int) -> int:
 
 
 def max_action_index(adventure: models.Adventure) -> int:
-    """Highest `Action.index` in the adventure, story text or not. -1 if empty."""
+    """Highest `Action.index` in the adventure, story text or not. -1 if empty.
+
+    The one read here that is deliberately *not* path-scoped. `index` is the
+    legacy column, kept unread until SP8 drops it, and its only remaining job
+    is to hand the next row a number nothing else holds — which is a fact about
+    the adventure, not about the story being played. Scoping it to a branch
+    would let two branches issue the same index.
+    """
     loaded = _loaded_actions(adventure)
     if loaded is not None:
         return max((a.index for a in loaded), default=-1)
diff --git a/backend/app/context/lineage.py b/backend/app/context/lineage.py
new file mode 100644
index 0000000..428df99
--- /dev/null
+++ b/backend/app/context/lineage.py
@@ -0,0 +1,191 @@
+"""Phase 14 — which nodes are "this story".
+
+`tree.py` decides where a node is written. This module is the other half: it
+decides which nodes a read can see, and it is the **only** place that knows.
+
+A branch owns the nodes played on it and *borrows* everything before its fork
+point from its ancestors, so "the story on branch C" is not a column you can
+filter on — it is an OR of ranges::
+
+    (branch_id = C)                 -- C's own nodes, to the tip
+    OR (branch_id = B AND depth <= 5)
+    OR (branch_id = A AND depth <= 3)
+
+which is exactly what `branches.lineage` spells out, newest first, computed
+once when the fork happens. Reads never walk parent pointers to rebuild it.
+
+Two properties fall out of the shape, and both are load-bearing:
+
+* **The ranges are disjoint and descending.** A branch's own nodes always sit
+  deeper than its fork point, and each lineage entry is capped at the fork
+  depth of the branch beneath it. So ordering the whole clause by `depth`
+  descending is the same as reading entry 0's nodes, then entry 1's, then
+  entry 2's — which is what lets a tail read use only the newest few entries
+  and stop.
+* **Clause count is bounded by the context window, not by fork count.** A
+  200-fork story whose newest branch is 40 turns long reads with one clause,
+  because the window is covered before the second entry is reached. That is
+  `prefix_covering`, and it is why `history.window_covering` can keep its shape.
+
+Everything here is a read. Nothing in this module creates a branch or writes a
+row: an adventure with no branch has no story, and healing that is the write
+side's job (`tree.place_action`, and the flush guard in `models.py` behind it).
+"""
+
+from sqlalchemy import and_, false, or_
+from sqlalchemy.orm import Session
+
+from .. import models
+
+# The depth of an adventure with no actions. Mirrors tree.NO_DEPTH; kept
+# separately so a read never has to import the write half.
+NO_DEPTH = -1
+
+
+def entries_of(branch: models.Branch) -> list[tuple[int, int | None]]:
+    """`branch.lineage` as (branch_id, max_depth) pairs, newest first.
+
+    An empty lineage reads as "this branch alone, to its tip" rather than as an
+    error. That is what a root branch's lineage means, and it is what a branch
+    row looks like in the moment between being inserted and having its own id
+    to name — so the fallback is the truth, not a guess.
+    """
+    raw = branch.lineage if isinstance(branch.lineage, list) else []
+    entries: list[tuple[int, int | None]] = []
+    for item in raw:
+        # JSON round-trips lists; a hand-written row might hold tuples.
+        if not isinstance(item, (list, tuple)) or not item:
+            continue
+        branch_id = item[0]
+        max_depth = item[1] if len(item) > 1 else None
+        if not isinstance(branch_id, int):
+            continue
+        entries.append((branch_id, max_depth if isinstance(max_depth, int) else None))
+    return entries or [(branch.id, None)]
+
+
+class Path:
+    """One story, as a clause and as a predicate.
+
+    Holds the lineage entries newest first, plus the depth of the tip, which is
+    only used to estimate how much story each entry covers.
+    """
+
+    def __init__(self, entries: list[tuple[int, int | None]], tip: int | None = None):
+        self.entries = entries
+        self.tip = tip
+
+    def __bool__(self) -> bool:
+        return bool(self.entries)
+
+    def __len__(self) -> int:
+        return len(self.entries)
+
+    # ---------------------------------------------------------------- SQL
+
+    def clause(self, model=models.Action, count: int | None = None):
+        """The branch clause, over `model` (`Action` or `Memory`).
+
+        `count` limits it to the newest `count` lineage entries — the windowed
+        read. `None` is the whole lineage, which is what anything counting from
+        the *oldest* end (a slice, a total) has to use.
+
+        An empty path yields `false`, not "no filter": an adventure whose nodes
+        carry no branch has no story, and the loud version of that is an empty
+        page, not every branch at once.
+        """
+        entries = self.entries if count is None else self.entries[:count]
+        if not entries:
+            return false()
+        return or_(*[self._entry_clause(model, b, d) for b, d in entries])
+
+    @staticmethod
+    def _entry_clause(model, branch_id: int, max_depth: int | None):
+        if max_depth is None:
+            return model.branch_id == branch_id
+        return and_(model.branch_id == branch_id, model.depth <= max_depth)
+
+    # ------------------------------------------------------------- Python
+
+    def contains(self, node) -> bool:
+        """The Python half of `clause()` — keep the two in step.
+
+        Used where the rows are already in memory (the scripting pipeline hands
+        user scripts the whole history), so an already-loaded collection can be
+        cut down to the path without a second read.
+        """
+        for branch_id, max_depth in self.entries:
+            if node.branch_id != branch_id:
+                continue
+            if max_depth is None:
+                return True
+            if node.depth is not None and node.depth <= max_depth:
+                return True
+        return False
+
+    def sort_key(self, node) -> tuple[int, int]:
+        """Oldest-first ordering. `depth` is the ordering key; `id` breaks the
+        tie a pre-tree row (depth NULL) or a future sibling pair would leave."""
+        return (node.depth if node.depth is not None else NO_DEPTH, node.id or 0)
+
+    # ------------------------------------------------------------ windowing
+
+    def prefix_covering(self, rows: int) -> int:
+        """How many lineage entries it takes to hold the newest `rows` nodes.
+
+        An estimate from depth arithmetic, not a query: entry *i* covers the
+        depths between its own cap and the cap of the entry below it, and there
+        is at most one node per depth on a path. So the count it returns is
+        never too many, and is too few only where the story has gaps — an
+        action deleted from the middle. The caller widens to the full lineage
+        if the window comes up short, which costs a second query on a story
+        somebody has deleted from, and nothing at all otherwise.
+        """
+        total = len(self.entries)
+        if rows <= 0 or total == 0:
+            return total
+        covered = 0
+        for i, (_, max_depth) in enumerate(self.entries):
+            top = self.tip if max_depth is None else max_depth
+            below = self.entries[i + 1][1] if i + 1 < total else NO_DEPTH
+            if top is None or below is None:
+                # No tip recorded, or a cap missing from a hand-written row:
+                # nothing to estimate from, so read the lot rather than guess
+                # short and hide the older half of the story.
+                return total
+            covered += max(top - below, 0)
+            if covered >= rows:
+                return i + 1
+        return total
+
+
+def branch_of(db: Session, adventure: models.Adventure) -> models.Branch | None:
+    """The branch this adventure is being read at, or None if it has none.
+
+    Deliberately not `tree.head_branch`, which creates one: a GET must not
+    write. An adventure with no branch row also has no nodes carrying a branch,
+    so the two agree — both say "no story here".
+    """
+    if adventure.head_branch_id is not None:
+        branch = db.get(models.Branch, adventure.head_branch_id)
+        if branch is not None:
+            return branch
+        # A head naming a branch that is gone: fall through to the root, the
+        # same recovery `tree.head_branch` makes on the write side.
+    return (
+        db.query(models.Branch)
+        .filter(
+            models.Branch.adventure_id == adventure.id,
+            models.Branch.parent_branch_id.is_(None),
+        )
+        .order_by(models.Branch.id)
+        .first()
+    )
+
+
+def path_of(db: Session, adventure: models.Adventure) -> Path:
+    """The story the adventure's head is currently on."""
+    branch = branch_of(db, adventure)
+    if branch is None:
+        return Path([], adventure.head_depth)
+    return Path(entries_of(branch), adventure.head_depth)
diff --git a/backend/app/limits.py b/backend/app/limits.py
index 0fcf6c6..52d3a27 100644
--- a/backend/app/limits.py
+++ b/backend/app/limits.py
@@ -207,6 +207,12 @@ def check_row_cap(
             "delete some to make room",
         )
     elif kind == "actions":
+        # Every action of the adventure — the whole tree, not the path being
+        # played. That is the number that costs storage, and nothing is ever
+        # auto-pruned, so it is the right one to cap on. It does mean a heavily
+        # branched adventure reaches the cap while its *story* is shorter than
+        # the cap, which is why the message counts "actions in this adventure"
+        # rather than turns.
         count = _count(db, models.Action, models.Action.adventure_id == adventure.id)
         cap, subject, hint = (
             MAX_ACTIONS_PER_ADVENTURE, "actions in this adventure",
diff --git a/backend/app/models.py b/backend/app/models.py
index 5c5df1d..3654729 100644
--- a/backend/app/models.py
+++ b/backend/app/models.py
@@ -2,9 +2,9 @@ from datetime import datetime, timezone
 
 from sqlalchemy import (
     JSON, Boolean, Column, DateTime, Float, ForeignKey, Index, Integer, LargeBinary,
-    String, Table, Text,
+    String, Table, Text, event,
 )
-from sqlalchemy.orm import Mapped, mapped_column, relationship
+from sqlalchemy.orm import Mapped, Session, mapped_column, relationship
 
 from .compression import CompressedJSON
 from .database import Base
@@ -481,3 +481,23 @@ class Settings(Base):
         from . import security  # local import: models is imported before security
 
         return security.decrypt_secret(self.api_key)
+
+
+# Phase 14 — the floor under `tree.place_action`.
+#
+# From SP2 a read selects on (branch_id, depth): a node written without them is
+# a node no page, no context build and no memory pass can see, and it fails by
+# disappearing rather than by raising. The writers all place their nodes
+# explicitly, but "all the writers remember" is a promise that has to hold for
+# every fixture, script and test written from here on, so the session enforces
+# it on the way to the database instead.
+#
+# Registered here rather than in tree.py so that importing the models is enough
+# to arm it — the invariant belongs to the rows, not to the module that usually
+# writes them. The import is deferred into the callback because tree.py imports
+# this module.
+@event.listens_for(Session, "before_flush")
+def _place_new_nodes_on_the_tree(session, flush_context, instances):
+    from . import tree
+
+    tree.place_new_nodes(session)
diff --git a/backend/app/routers/adventures.py b/backend/app/routers/adventures.py
index 30a7274..bc48323 100644
--- a/backend/app/routers/adventures.py
+++ b/backend/app/routers/adventures.py
@@ -12,6 +12,7 @@ from sqlalchemy.orm.attributes import set_committed_value
 from .. import auth, images, limits, memorybank, models, schemas, tree, worldstate
 from ..context import build_context
 from ..context import history as context_history
+from ..context import lineage
 from ..database import get_db
 from ..providers import OpenAICompatibleProvider, PromptParts, ProviderError
 from ..scripting import ScriptPipeline
@@ -56,7 +57,7 @@ ACTION_PAGE = 60
 
 def action_window(
     db: Session,
-    adventure_id: int,
+    adventure: models.Adventure,
     before_id: int | None = None,
     limit: int = ACTION_PAGE,
 ) -> tuple[list[models.Action], int, bool]:
@@ -64,50 +65,56 @@ def action_window(
 
     Returns (actions, total, has_more). `before_id=None` is the newest window.
 
-    Anchored on an action, not on a count, and never on arithmetic over
-    `Action.index`. Two separate reasons, and both bite:
+    Scoped to the story being played — the head branch's lineage — rather than
+    to the adventure, so a sibling branch's turns can never appear in the
+    transcript. `total` counts the same path, because it is what tells the
+    reader there is more above.
+
+    Anchored on an action, not on a count, and never on arithmetic over depth.
+    Two separate reasons, and both bite:
 
     * **Appends.** Counting back from the newest means every older position
       shifts when a turn lands. A reader who scrolls up while a turn is
       generating would be handed a window one row out — re-sending one action
       and silently skipping another. An anchor is fixed: "older than this one"
       means the same thing before and after the story grows.
-    * **The story tree.** Index is a dense 0..n sequence today and branching
-      ends that. Comparing indices to order a branch survives; treating them as
-      positions does not.
+    * **The story tree.** Depth is dense today and branching ends that.
+      Comparing depths to order a path survives; treating them as positions
+      does not.
 
     `has_more` comes from asking for one row past the window rather than from
     counting, so it costs a row and not a scan.
     """
-    total = (
-        db.query(func.count(models.Action.id))
-        .filter(models.Action.adventure_id == adventure_id)
-        .scalar()
+    path = lineage.path_of(db, adventure)
+    on_path = (
+        models.Action.adventure_id == adventure.id,
+        path.clause(models.Action),
     )
+    total = db.query(func.count(models.Action.id)).filter(*on_path).scalar()
     if limit <= 0:
         return [], total, total > 0
 
-    query = (
-        db.query(models.Action)
-        .options(load_only(*ACTION_LIST_COLUMNS))
-        .filter(models.Action.adventure_id == adventure_id)
-    )
+    query = db.query(models.Action).options(load_only(*ACTION_LIST_COLUMNS)).filter(*on_path)
     if before_id is not None:
         anchor = (
-            db.query(models.Action.index)
-            .filter(models.Action.id == before_id,
-                    models.Action.adventure_id == adventure_id)
+            db.query(models.Action.depth)
+            .filter(models.Action.id == before_id, *on_path)
             .scalar()
         )
         if anchor is None:
             # The anchor was deleted (undo, or a turn edited away) while the
-            # reader was scrolling. Nothing older can be identified relative to
-            # a row that no longer exists, so report the end rather than
-            # guessing and handing back a duplicate page.
+            # reader was scrolling, or it belongs to a story this branch is not
+            # on. Nothing older can be identified relative to a row that is not
+            # here, so report the end rather than guessing and handing back a
+            # duplicate page.
             return [], total, False
-        query = query.filter(models.Action.index < anchor)
+        query = query.filter(models.Action.depth < anchor)
 
-    rows = query.order_by(models.Action.index.desc()).limit(limit + 1).all()
+    rows = (
+        query.order_by(models.Action.depth.desc(), models.Action.id.desc())
+        .limit(limit + 1)
+        .all()
+    )
     has_more = len(rows) > limit
     rows = rows[:limit]
     rows.reverse()
@@ -163,13 +170,21 @@ def _snippet(text: str) -> str:
 NARRATION_TYPES = ("ai", "story", "start")
 
 
-def _latest_narration(db: Session, adventure_ids: list[int]) -> dict[int, str]:
+def _latest_narration(db: Session, head_branches: dict[int, int | None]) -> dict[int, str]:
     """Map adventure id -> text of its most recent narrated action.
 
     One window-function query rather than a per-adventure lookup, so the list
     endpoint stays at a fixed number of round trips.
+
+    Scoped by head *branch* rather than by the full lineage, which is the one
+    place in the codebase that is allowed to be: a lineage clause per adventure
+    would put a hundred OR-terms on the index screen's query to pick one row
+    each. The two answers differ only for a branch with no nodes of its own,
+    and a branch is created by playing a turn onto it, so that state does not
+    exist. An adventure with no branch at all has no story to quote.
     """
-    if not adventure_ids:
+    branch_ids = [b for b in head_branches.values() if b is not None]
+    if not branch_ids:
         return {}
     ranked = (
         db.query(
@@ -178,12 +193,13 @@ def _latest_narration(db: Session, adventure_ids: list[int]) -> dict[int, str]:
             func.row_number()
             .over(
                 partition_by=models.Action.adventure_id,
-                order_by=(models.Action.index.desc(), models.Action.id.desc()),
+                order_by=(models.Action.depth.desc(), models.Action.id.desc()),
             )
             .label("rank"),
         )
         .filter(
-            models.Action.adventure_id.in_(adventure_ids),
+            models.Action.adventure_id.in_(list(head_branches)),
+            models.Action.branch_id.in_(branch_ids),
             models.Action.type.in_(NARRATION_TYPES),
         )
         .subquery()
@@ -206,6 +222,7 @@ def list_adventures(db: Session = Depends(get_db), user: models.User = CurrentUs
             models.Adventure.scenario_id,
             models.Adventure.title,
             models.Adventure.updated_at,
+            models.Adventure.head_branch_id,
             func.count(models.Action.id),
             models.Scenario.title,
             models.Scenario.image,
@@ -230,7 +247,7 @@ def list_adventures(db: Session = Depends(get_db), user: models.User = CurrentUs
         .order_by(models.Adventure.updated_at.desc())
         .all()
     )
-    narration = _latest_narration(db, [row[0] for row in rows])
+    narration = _latest_narration(db, {row[0]: row[4] for row in rows})
     return [
         schemas.AdventureListItem(
             id=adv_id,
@@ -245,7 +262,11 @@ def list_adventures(db: Session = Depends(get_db), user: models.User = CurrentUs
             image_url=images.public_url(scenario_id, image or "", scenario_updated),
             icon=icon or "",
         )
-        for (adv_id, scenario_id, title, updated_at, count,
+        # `count` is every action of the adventure, not of the path. Under one
+        # branch they are the same number; once forking ships the index screen
+        # will overstate a story that has siblings hanging off it, and the fix
+        # belongs with SP5, where a fork can first exist.
+        for (adv_id, scenario_id, title, updated_at, _head_branch_id, count,
              scenario_title, image, icon, scenario_updated) in rows
     ]
 
@@ -385,7 +406,7 @@ def get_adventure(
     from GET /{id}/actions as they scroll up.
     """
     adventure = get_adventure_or_404(adventure_id, db, user)
-    actions, total, _ = action_window(db, adventure_id)
+    actions, total, _ = action_window(db, adventure)
     # Hand the response the window as if the relationship had loaded it.
     # `set_committed_value` is the only way to do this safely: assigning
     # `adventure.actions = [...]` marks the collection dirty, and the
@@ -636,13 +657,19 @@ def next_index(adventure: models.Adventure) -> int:
 
 
 def last_action(adventure: models.Adventure, db: Session) -> models.Action | None:
-    """The newest action of any kind, or None. A query rather than
-    `adventure.actions[-1]`, which would load the entire story to look at
-    one row."""
+    """The newest action of any kind on the story being played, or None.
+
+    A query rather than `adventure.actions[-1]`, which would load the entire
+    story to look at one row — and, since that collection is every branch's
+    actions, would sometimes look at the wrong one.
+    """
     return (
         db.query(models.Action)
-        .filter(models.Action.adventure_id == adventure.id)
-        .order_by(models.Action.index.desc())
+        .filter(
+            models.Action.adventure_id == adventure.id,
+            lineage.path_of(db, adventure).clause(models.Action),
+        )
+        .order_by(models.Action.depth.desc(), models.Action.id.desc())
         .first()
     )
 
@@ -1061,8 +1088,11 @@ def undo_turn(
         # consist of rather than the whole story.
         newest = (
             db.query(models.Action)
-            .filter(models.Action.adventure_id == adventure.id)
-            .order_by(models.Action.index.desc())
+            .filter(
+                models.Action.adventure_id == adventure.id,
+                lineage.path_of(db, adventure).clause(models.Action),
+            )
+            .order_by(models.Action.depth.desc(), models.Action.id.desc())
             .limit(2)
             .all()
         )
@@ -1093,7 +1123,7 @@ def undo_turn(
         # transcript with this, and the transcript is a window now. Returning
         # everything here would undo the paging on the one action most likely
         # to be repeated several times in a row.
-        actions, total, has_more = action_window(db, adventure_id)
+        actions, total, has_more = action_window(db, adventure)
         return schemas.ActionPage(
             actions=[schemas.ActionOut.model_validate(a) for a in actions],
             total=total,
@@ -1704,10 +1734,10 @@ def list_actions(
     is "give me what comes before this". Omit it for the newest window. See
     action_window for why this anchors on a row rather than an offset.
     """
-    get_adventure_or_404(adventure_id, db, user)
+    adventure = get_adventure_or_404(adventure_id, db, user)
     limit = max(1, min(limit, ACTION_PAGE * 4))
     actions, total, has_more = action_window(
-        db, adventure_id, before_id=before_id, limit=limit
+        db, adventure, before_id=before_id, limit=limit
     )
     return schemas.ActionPage(
         actions=[schemas.ActionOut.model_validate(a) for a in actions],
diff --git a/backend/app/scripting/pipeline.py b/backend/app/scripting/pipeline.py
index a14555a..60f0a43 100644
--- a/backend/app/scripting/pipeline.py
+++ b/backend/app/scripting/pipeline.py
@@ -4,6 +4,7 @@ state and story-card mutations back to the database after each hook."""
 from sqlalchemy.orm import Session
 
 from .. import models
+from ..context import history as context_history
 from .engine import run_hook
 
 MAX_STORY_CARDS = 5000  # AI Dungeon's per-adventure sanity cap
@@ -23,9 +24,13 @@ class ScriptPipeline:
         return msg if isinstance(msg, str) and msg.strip() else None
 
     def _history(self) -> list[dict]:
+        # The path, not `adventure.actions` — that collection is every branch's
+        # actions, and this is the documented history API a user script reads.
+        # Handing a script the siblings of the turn it is running on would be
+        # the same bug as building a prompt from them, only user-visible.
         return [
             {"text": a.text, "rawText": a.text, "type": a.type}
-            for a in self.adventure.actions
+            for a in context_history.story_actions(self.adventure)
         ]
 
     def _cards(self) -> list[dict]:
@@ -36,7 +41,7 @@ class ScriptPipeline:
 
     def _info(self) -> dict:
         return {
-            "actionCount": len(self.adventure.actions),
+            "actionCount": context_history.count(self.adventure),
             "characterNames": [],
             "memoryLength": len(self.adventure.memory),
             "maxChars": 0,
diff --git a/backend/app/tree.py b/backend/app/tree.py
index db45c94..c30cf33 100644
--- a/backend/app/tree.py
+++ b/backend/app/tree.py
@@ -7,16 +7,22 @@ call sites: a node written without a branch is a node no read can see, and it
 fails by disappearing rather than by raising.
 
 The read half — the lineage clause that turns a branch into "this story" —
-lands beside it in SP2 (`context/lineage.py`). Nothing here is read yet.
+lives beside it in `context/lineage.py`.
 
 Until forking ships there is exactly one branch per adventure and `depth` is
 the number `index` already held, so everything in this module is bookkeeping
 that changes nothing observable. That is the point: by the time a read depends
 on these columns, every row has them — including the rows written between the
 two deploys, which no migration will ever visit.
+
+SP2 added `place_new_nodes`, which the session calls on every flush. Wiring the
+call sites was enough while nothing read the columns; now that reads select on
+them, "every writer remembers" is a promise that has to hold for every fixture,
+script and test ever written too, and its breach is a story quietly missing
+turns. So the invariant is enforced at the flush instead of asked for.
 """
 
-from sqlalchemy import func
+from sqlalchemy import func, insert, update
 from sqlalchemy.orm import Session
 
 from . import models
@@ -45,12 +51,26 @@ def root_branch(db: Session, adventure: models.Adventure) -> models.Branch:
     )
     if branch is not None:
         return branch
-    branch = models.Branch(adventure_id=adventure.id, lineage=[])
-    db.add(branch)
-    # The lineage names the branch's own id, so the row has to exist first.
-    db.flush()
-    branch.lineage = [[branch.id, None]]
-    return branch
+    # Inserted through Core rather than through the unit of work, because this
+    # also runs from `place_new_nodes` inside a flush, and a nested ORM flush
+    # inside a flush raises. Same transaction either way, so it rolls back with
+    # everything else. The lineage names the branch's own id, so it takes a
+    # second statement — once per adventure, ever.
+    new_id = db.execute(
+        insert(models.Branch).values(
+            adventure_id=adventure.id,
+            parent_branch_id=None,
+            fork_depth=None,
+            lineage=[],
+            created_at=models.utcnow(),
+        )
+    ).inserted_primary_key[0]
+    db.execute(
+        update(models.Branch)
+        .where(models.Branch.id == new_id)
+        .values(lineage=[[new_id, None]])
+    )
+    return db.get(models.Branch, new_id)
 
 
 def head_branch(db: Session, adventure: models.Adventure) -> models.Branch:
@@ -81,7 +101,7 @@ def place_action(
     if action.depth is None:
         action.depth = action.index
     adventure.head_branch_id = branch.id
-    if action.depth > adventure.head_depth:
+    if action.depth is not None and action.depth > adventure.head_depth:
         adventure.head_depth = action.depth
     return branch
 
@@ -102,6 +122,34 @@ def place_memory(
     return branch
 
 
+def place_new_nodes(session: Session) -> None:
+    """Place every unplaced node about to be inserted. Runs on every flush.
+
+    The call sites still call `place_action` / `place_memory` themselves, and
+    should: a node placed at the call site is placed *before* the code around
+    it reads the row back, and the explicit call is what makes the ordering
+    visible. This is the floor under them — a fixture built straight through
+    the ORM, a script, a test, or a call site added next year gets a branch
+    without knowing the tree exists.
+
+    Nodes whose adventure has not been inserted yet are left alone: there is no
+    id to hang a branch off, and an Action needs `adventure_id` to be written
+    at all, so the case does not arise from any writer we have.
+    """
+    for obj in list(session.new):
+        if isinstance(obj, models.Action):
+            place = place_action
+        elif isinstance(obj, models.Memory):
+            place = place_memory
+        else:
+            continue
+        if obj.branch_id is not None or obj.adventure_id is None:
+            continue
+        adventure = session.get(models.Adventure, obj.adventure_id)
+        if adventure is not None:
+            place(session, adventure, obj)
+
+
 def refresh_head(db: Session, adventure: models.Adventure) -> None:
     """Re-derive the head depth after nodes were removed (undo, delete).
 
diff --git a/backend/tests/test_branch_clause.py b/backend/tests/test_branch_clause.py
new file mode 100644
index 0000000..7bc48ea
--- /dev/null
+++ b/backend/tests/test_branch_clause.py
@@ -0,0 +1,534 @@
+"""Phase 14 SP2 — a read sees one story, and knows which one.
+
+Nothing in the product forks yet, so these tests build the fork by hand: three
+branch rows and their nodes, written straight to the database, arranged as the
+design doc's own worked example.
+
+    branch C, tip at depth 7, lineage [(C, 7), (B, 5), (A, 3)]
+    → A0 A1 A2 A3 B4 B5 C6 C7
+
+The point of building it by hand is that every read in the app is supposed to
+go through one module, and a forgotten clause does not raise — it quietly shows
+a story assembled out of two different ones. So the fixture deliberately leaves
+nodes lying where a forgotten clause would pick them up: A kept playing past
+the fork (A4, A5), B kept playing past its own (B6), and a second adventure
+holds a whole story of its own. None of them may appear on C.
+
+    python -m pytest tests/test_branch_clause.py -v
+"""
+import os
+import tempfile
+
+_tmp = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
+_tmp.close()
+os.environ["AIDND_DB_PATH"] = _tmp.name
+os.environ.pop("AIDND_DATABASE_URL", None)
+os.environ.pop("DATABASE_URL", None)
+
+import pytest
+from fastapi import Depends
+from fastapi.testclient import TestClient
+from sqlalchemy import event
+
+from app import auth, limits, models, tree
+from app.context import history, lineage
+from app.database import Base, SessionLocal, engine, get_db
+from app.main import app
+from app.scripting import ScriptPipeline
+from tools import dbmeter
+
+
+# --------------------------------------------------------------- the fixture
+
+def make_branch(db, adventure, parent=None, fork_depth=None):
+    """A branch row whose lineage is its parent's, capped, plus itself.
+
+    The same computation SP5 will do at fork time; written out here so the
+    fixture cannot pass by agreeing with a bug in the code under test.
+    """
+    branch = models.Branch(
+        adventure_id=adventure.id,
+        parent_branch_id=parent.id if parent else None,
+        fork_depth=fork_depth,
+        lineage=[],
+    )
+    db.add(branch)
+    db.flush()
+    inherited = []
+    if parent is not None:
+        for ancestor_id, cap in lineage.entries_of(parent):
+            capped = fork_depth if cap is None else min(cap, fork_depth)
+            inherited.append([ancestor_id, capped])
+    branch.lineage = [[branch.id, None]] + inherited
+    db.flush()
+    return branch
+
+
+def add_node(db, adventure, branch, depth, label, index=None):
+    action = models.Action(
+        adventure_id=adventure.id,
+        index=depth if index is None else index,
+        branch_id=branch.id,
+        depth=depth,
+        type="start" if depth == 0 else ("ai" if depth % 2 else "do"),
+        text=f"{label}{depth}",
+    )
+    db.add(action)
+    return action
+
+
+def make_adventure(db, user, title):
+    adventure = models.Adventure(user_id=user.id, title=title, script_state={})
+    db.add(adventure)
+    db.flush()
+    return adventure
+
+
+@pytest.fixture()
+def forked():
+    """The worked example, plus everything a forgotten clause would sweep up."""
+    Base.metadata.create_all(bind=engine)
+    db = SessionLocal()
+    user = models.User(is_guest=False, email="branch@example.com")
+    db.add(user)
+    db.flush()
+    db.add(models.Settings(user_id=user.id, api_key="enc:dummy", model="m"))
+
+    adventure = make_adventure(db, user, "Forked")
+    a = make_branch(db, adventure)
+    b = make_branch(db, adventure, parent=a, fork_depth=3)
+    c = make_branch(db, adventure, parent=b, fork_depth=5)
+
+    for depth in range(4):
+        add_node(db, adventure, a, depth, "A")
+    # A did not stop when B forked off it: these two are siblings of B4/B5.
+    for depth in (4, 5):
+        add_node(db, adventure, a, depth, "A", index=100 + depth)
+    for depth in (4, 5):
+        add_node(db, adventure, b, depth, "B")
+    add_node(db, adventure, b, 6, "B", index=200 + 6)  # B's own continuation
+    for depth in (6, 7):
+        add_node(db, adventure, c, depth, "C")
+
+    # A second adventure, so "does the clause remember its adventure?" has an
+    # answer. A database holding one adventure cannot tell you.
+    other = make_adventure(db, user, "Elsewhere")
+    other_branch = make_branch(db, other)
+    for depth in range(4):
+        add_node(db, other, other_branch, depth, "X")
+
+    adventure.head_branch_id = c.id
+    adventure.head_depth = 7
+    other.head_branch_id = other_branch.id
+    other.head_depth = 3
+    db.commit()
+
+    ids = {"a": a.id, "b": b.id, "c": c.id, "adventure": adventure.id,
+           "other": other.id, "user": user.id}
+    try:
+        yield db, adventure, ids
+    finally:
+        db.close()
+        Base.metadata.drop_all(bind=engine)
+
+
+def switch_to(db, adventure, branch_id, tip):
+    """Move the head, the way SP7's branch picker will."""
+    adventure.head_branch_id = branch_id
+    adventure.head_depth = tip
+    db.commit()
+
+
+def labels(actions):
+    return [a.text for a in actions]
+
+
+# ------------------------------------------------------------- the story read
+
+def test_the_worked_example_reads_back_as_the_design_doc_says(forked):
+    db, adventure, _ = forked
+    assert labels(history.story_actions(adventure)) == [
+        "A0", "A1", "A2", "A3", "B4", "B5", "C6", "C7"
+    ]
+
+
+def test_a_siblings_nodes_are_invisible(forked):
+    db, adventure, _ = forked
+    seen = labels(history.story_actions(adventure))
+    # A4/A5 are A's own continuation past B's fork; B6 is B's past C's.
+    assert "A4" not in seen and "A5" not in seen and "B6" not in seen
+    # And nothing from the adventure next door.
+    assert not [text for text in seen if text.startswith("X")]
+
+
+def test_each_branch_reads_its_own_story(forked):
+    db, adventure, ids = forked
+    switch_to(db, adventure, ids["a"], 5)
+    assert labels(history.story_actions(adventure)) == [
+        "A0", "A1", "A2", "A3", "A4", "A5"
+    ]
+    switch_to(db, adventure, ids["b"], 6)
+    assert labels(history.story_actions(adventure)) == [
+        "A0", "A1", "A2", "A3", "B4", "B5", "B6"
+    ]
+    switch_to(db, adventure, ids["c"], 7)
+    assert labels(history.story_actions(adventure)) == [
+        "A0", "A1", "A2", "A3", "B4", "B5", "C6", "C7"
+    ]
+
+
+def test_the_ancestors_shared_nodes_are_shared_not_copied(forked):
+    db, adventure, ids = forked
+    # A0..A3 appear on all three stories and exist exactly once in the table.
+    rows = (
+        db.query(models.Action)
+        .filter(models.Action.adventure_id == adventure.id,
+                models.Action.text == "A2")
+        .all()
+    )
+    assert len(rows) == 1
+    assert rows[0].branch_id == ids["a"]
+
+
+def test_count_and_tail_and_slice_agree_with_the_path(forked):
+    db, adventure, _ = forked
+    assert history.count(adventure) == 8
+    assert labels(history.tail(adventure, 3)) == ["B5", "C6", "C7"]
+    assert labels(history.tail_range(adventure, 2, 2)) == ["B4", "B5"]
+    assert labels(history.slice_(adventure, 3, 3)) == ["A3", "B4", "B5"]
+
+
+def test_a_window_that_reaches_past_the_fork_still_reads_in_order(forked):
+    db, adventure, _ = forked
+    # 32 is history.WINDOW_START: more than the whole story, so the read has to
+    # widen through all three lineage entries and still come back in order.
+    assert labels(history.window_covering(adventure, 10 ** 6, len)) == [
+        "A0", "A1", "A2", "A3", "B4", "B5", "C6", "C7"
+    ]
+
+
+# ------------------------------------------- the loaded-collection short cut
+
+def test_an_already_loaded_collection_is_cut_down_to_the_path(forked):
+    """`history._from_memory`'s shortcut, which was the highest-risk line here.
+
+    `adventure.actions` is every branch's actions. Slicing it without the path
+    would assemble a prompt out of two different stories, and nothing would
+    raise — so load it deliberately and check the answer is the path anyway.
+    """
+    db, adventure, _ = forked
+    loaded = list(adventure.actions)  # every branch, ordered by index
+    assert len(loaded) == 11  # the path's 8, plus A4, A5 and B6
+    assert labels(history.story_actions(adventure)) == [
+        "A0", "A1", "A2", "A3", "B4", "B5", "C6", "C7"
+    ]
+    assert history.count(adventure) == 8
+    assert labels(history.tail(adventure, 3)) == ["B5", "C6", "C7"]
+
+
+def test_user_scripts_are_handed_the_path(forked):
+    """The same trap, one layer up and user-visible: `pipeline._history()` is
+    the documented scripting history API."""
+    db, adventure, _ = forked
+    list(adventure.actions)  # the pipeline's caller has usually loaded these
+    pipeline = ScriptPipeline(adventure, db)
+    assert [h["text"] for h in pipeline._history()] == [
+        "A0", "A1", "A2", "A3", "B4", "B5", "C6", "C7"
+    ]
+    assert pipeline._info()["actionCount"] == 8
+
+
+# ------------------------------------------------------------ over the wire
+
+@pytest.fixture()
+def client(forked, monkeypatch):
+    db, adventure, ids = forked
+    monkeypatch.setattr(limits, "rate_limit", lambda *a, **k: None)
+    monkeypatch.setattr(limits, "check_row_cap", lambda *a, **k: None)
+
+    def _current_user(session=Depends(get_db)):
+        return session.get(models.User, ids["user"])
+
+    app.dependency_overrides[auth.get_current_user] = _current_user
+    c = TestClient(app)
+    try:
+        yield c, db, adventure, ids
+    finally:
+        app.dependency_overrides.clear()
+
+
+def test_the_page_the_reader_opens_is_the_path(client):
+    c, db, adventure, ids = client
+    r = c.get(f"/api/adventures/{adventure.id}")
+    assert r.status_code == 200, r.text
+    body = r.json()
+    assert [a["text"] for a in body["actions"]] == [
+        "A0", "A1", "A2", "A3", "B4", "B5", "C6", "C7"
+    ]
+    # `action_count` is what tells the reader there is more above, so it counts
+    # the path too — 8, not the 13 rows the adventure holds.
+    assert body["action_count"] == 8
+
+
+def test_paging_up_walks_the_path_across_the_forks(client):
+    c, db, adventure, ids = client
+    first = c.get(f"/api/adventures/{adventure.id}/actions", params={"limit": 3}).json()
+    assert [a["text"] for a in first["actions"]] == ["B5", "C6", "C7"]
+    assert first["has_more"] is True
+    older = c.get(
+        f"/api/adventures/{adventure.id}/actions",
+        params={"limit": 3, "before_id": first["actions"][0]["id"]},
+    ).json()
+    assert [a["text"] for a in older["actions"]] == ["A2", "A3", "B4"]
+    oldest = c.get(
+        f"/api/adventures/{adventure.id}/actions",
+        params={"limit": 3, "before_id": older["actions"][0]["id"]},
+    ).json()
+    assert [a["text"] for a in oldest["actions"]] == ["A0", "A1"]
+    assert oldest["has_more"] is False
+
+
+def test_paging_from_an_anchor_off_the_path_reports_the_end(client):
+    """A stale client holding an id from another branch gets an empty page,
+    not that branch's story."""
+    c, db, adventure, ids = client
+    off_path = (
+        db.query(models.Action)
+        .filter(models.Action.adventure_id == adventure.id,
+                models.Action.text == "A5")
+        .one()
+    )
+    page = c.get(
+        f"/api/adventures/{adventure.id}/actions",
+        params={"before_id": off_path.id},
+    ).json()
+    assert page["actions"] == []
+    assert page["has_more"] is False
+
+
+def test_the_index_screen_quotes_the_branch_being_played(client):
+    c, db, adventure, ids = client
+    listed = {row["id"]: row for row in c.get("/api/adventures").json()}
+    # The newest *narrated* node on C: do/say are the player's voice, and the
+    # fixture alternates types, so C7 is the one that reads as story.
+    assert listed[adventure.id]["snippet"] == "C7"
+    switch_to(db, adventure, ids["a"], 5)
+    listed = {row["id"]: row for row in c.get("/api/adventures").json()}
+    assert listed[adventure.id]["snippet"] == "A5"
+    assert listed[ids["other"]]["snippet"] == "X3"
+
+
+# ------------------------------------------------------- the flush guard
+
+def test_a_node_written_without_a_branch_is_placed_anyway(forked):
+    """SP1 wired the writers; from SP2 an unplaced node is an invisible one.
+
+    This is what lets a fixture, a script or a test built straight through the
+    ORM keep working — and it is why the baseline contract still passes with
+    its actions written directly to the database.
+    """
+    db, adventure, ids = forked
+    written = models.Action(
+        adventure_id=adventure.id, index=99, type="do", text="C8"
+    )
+    db.add(written)
+    db.commit()
+    assert written.branch_id == ids["c"]
+    assert written.depth == 99
+    assert adventure.head_depth == 99
+    assert labels(history.tail(adventure, 2)) == ["C7", "C8"]
+
+
+def test_a_memory_written_without_a_branch_is_placed_anyway(forked):
+    db, adventure, ids = forked
+    memory = models.Memory(
+        adventure_id=adventure.id, text="The cave was cold.", source_start=0, source_end=3
+    )
+    db.add(memory)
+    db.commit()
+    assert memory.branch_id == ids["c"]
+    assert memory.depth == 3
+
+
+def test_an_adventure_with_no_branch_at_all_reads_as_empty(forked):
+    """The loud version of a missing branch: nothing, rather than everything.
+
+    A row with no branch cannot be shown without guessing which story it is
+    on, and a guess here is how a sibling's turns end up in a prompt.
+    """
+    db, adventure, ids = forked
+    stray = make_adventure(db, db.get(models.User, ids["user"]), "Stray")
+    db.query(models.Action).filter(models.Action.text == "A0").update(
+        {"adventure_id": stray.id}, synchronize_session=False
+    )
+    db.commit()
+    db.expire_all()
+    stray = db.get(models.Adventure, stray.id)
+    assert lineage.path_of(db, stray).entries == []
+    assert history.story_actions(stray) == []
+    assert history.count(stray) == 0
+
+
+# --------------------------------------------------- the cost of forking
+
+@pytest.fixture()
+def deeply_forked():
+    """A story forked twenty times, then played forty turns past the last one.
+
+    The shape the design is betting on: reading the tail of this must cost what
+    reading the tail of an unforked story costs, because the window is covered
+    long before the ancestry runs out.
+    """
+    Base.metadata.create_all(bind=engine)
+    db = SessionLocal()
+    user = models.User(is_guest=False, email="deep@example.com")
+    db.add(user)
+    db.flush()
+    adventure = make_adventure(db, user, "Deep")
+    branch = make_branch(db, adventure)
+    depth = 0
+    for _ in range(4):
+        add_node(db, adventure, branch, depth, "n")
+        depth += 1
+    for fork in range(20):
+        branch = make_branch(db, adventure, parent=branch, fork_depth=depth - 1)
+        for _ in range(2):
+            add_node(db, adventure, branch, depth, "n")
+            depth += 1
+    tip_branch = make_branch(db, adventure, parent=branch, fork_depth=depth - 1)
+    for _ in range(40):
+        add_node(db, adventure, tip_branch, depth, "n")
+        depth += 1
+    adventure.head_branch_id = tip_branch.id
+    adventure.head_depth = depth - 1
+    db.commit()
+    try:
+        yield db, adventure
+    finally:
+        db.close()
+        Base.metadata.drop_all(bind=engine)
+
+
+@pytest.fixture()
+def emitted_sql():
+    """Every statement the connection runs, so a clause can be counted."""
+    seen = []
+
+    def on_execute(conn, cursor, statement, parameters, context, executemany):
+        seen.append(statement)
+
+    event.listen(engine, "before_cursor_execute", on_execute)
+    try:
+        yield seen
+    finally:
+        event.remove(engine, "before_cursor_execute", on_execute)
+
+
+def branch_terms(statement: str) -> int:
+    return statement.count("actions.branch_id =")
+
+
+def test_a_tail_read_names_one_branch_however_many_forks_there_were(
+    deeply_forked, emitted_sql
+):
+    db, adventure = deeply_forked
+    assert len(lineage.path_of(db, adventure)) == 22  # the whole ancestry
+    emitted_sql.clear()
+    rows = history.tail(adventure, 32)
+    assert len(rows) == 32
+    selects = [s for s in emitted_sql if "FROM actions" in s and branch_terms(s)]
+    assert selects, "no action read was emitted"
+    # Clause count is bounded by the context window, not by fork count: the
+    # newest branch alone holds forty turns, so one entry covers a window of
+    # thirty-two and the other twenty-one are never named.
+    assert max(branch_terms(s) for s in selects) == 1
+
+
+def test_a_window_reaching_past_the_forks_names_only_what_it_needs(
+    deeply_forked, emitted_sql
+):
+    db, adventure = deeply_forked
+    emitted_sql.clear()
+    rows = history.tail(adventure, 41)  # 40 on the tip branch, one older
+    assert len(rows) == 41
+    selects = [s for s in emitted_sql if "FROM actions" in s and branch_terms(s)]
+    # Two lineage entries reach 41 deep (40 + 2); the other twenty stay
+    # unnamed. Every fork past the window costs the query nothing.
+    assert max(branch_terms(s) for s in selects) == 2
+
+
+def test_the_estimate_is_arithmetic_not_a_query(deeply_forked):
+    db, adventure = deeply_forked
+    path = lineage.path_of(db, adventure)
+    assert path.prefix_covering(1) == 1
+    assert path.prefix_covering(40) == 1
+    assert path.prefix_covering(41) == 2
+    assert path.prefix_covering(43) == 3
+    assert path.prefix_covering(10 ** 6) == len(path)
+
+
+def test_forking_twenty_times_costs_the_same_bytes_as_never_forking(
+    deeply_forked,
+):
+    """The design's bet, in bytes.
+
+    Two stories of the same length, one played straight through and one forked
+    twenty times, read their newest window for the same money — because the
+    window is covered by the newest lineage entry either way, and the ancestry
+    is never named. The forked read pays for one extra row: the branch it read
+    the lineage off.
+    """
+    db, forked_adventure = deeply_forked
+    flat = make_adventure(db, db.get(models.User, forked_adventure.user_id), "Flat")
+    branch = make_branch(db, flat)
+    for depth in range(84):  # the same 84 nodes the forked story is long
+        add_node(db, flat, branch, depth, "n")
+    flat.head_branch_id = branch.id
+    flat.head_depth = 83
+    flat_id, forked_id = flat.id, forked_adventure.id
+    # Commit and let go of the connection: the meter wraps the pool's factory,
+    # so a connection checked out before it attaches is a connection it never
+    # sees. Building the fixture is a write path nobody plays, and is not
+    # charged to either scope.
+    db.commit()
+    db.expire_all()
+
+    meter = dbmeter.Meter()
+    meter.attach(engine)
+    try:
+        with meter.scope("flat"):
+            assert len(history.tail(db.get(models.Adventure, flat_id), 32)) == 32
+        flat_bytes = meter.scopes[-1].total.fetched
+        with meter.scope("forked"):
+            assert len(history.tail(db.get(models.Adventure, forked_id), 32)) == 32
+        forked_bytes = meter.scopes[-1].total.fetched
+    finally:
+        meter.detach()
+
+    assert flat_bytes > 0, "the meter saw nothing; it is measuring the wrong connection"
+
+    assert forked_bytes < flat_bytes * 1.25, (
+        f"reading a 20-fork story cost {forked_bytes:,} B against the "
+        f"{flat_bytes:,} B an unforked one of the same length cost"
+    )
+
+
+def test_a_gap_in_the_story_widens_the_read_rather_than_shortening_it(
+    deeply_forked, emitted_sql
+):
+    """The estimate counts depths, and a deleted action leaves a depth with no
+    row behind it. The read has to notice it came up short and widen."""
+    db, adventure = deeply_forked
+    victim = (
+        db.query(models.Action)
+        .filter(models.Action.branch_id == adventure.head_branch_id)
+        .order_by(models.Action.depth)
+        .first()
+    )
+    db.delete(victim)
+    db.commit()
+    emitted_sql.clear()
+    rows = history.tail(adventure, 40)
+    assert len(rows) == 40  # 39 on the tip branch, one borrowed from above
+    selects = [s for s in emitted_sql if "FROM actions" in s and branch_terms(s)]
+    assert max(branch_terms(s) for s in selects) == len(lineage.path_of(db, adventure))